API 并发调用最佳实践

import asyncio
import aiohttp
from typing import List
import time

class ConcurrentAPIClient:
    def __init__(self, api_key: str, max_concurrent: int = 10):
        self.api_key = api_key
        self.base_url = "https://api.n1n.ai/v1"
        self.semaphore = asyncio.Semaphore(max_concurrent)
        
    async def make_request(self, prompt: str):
        async with self.semaphore:  # 限制并发数
            async with aiohttp.ClientSession() as session:
                headers = {
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                }
                
                payload = {
                    "model": "gpt-4o-mini",
                    "messages": [{"role": "user", "content": prompt}],
                    "max_tokens": 100
                }
                
                async with session.post(
                    f"{self.base_url}/chat/completions",
                    headers=headers,
                    json=payload
                ) as response:
                    if response.status == 200:
                        result = await response.json()
                        return result['choices'][0]['message']['content']
                    else:
                        raise Exception(f"API error: {response.status}")
    
    async def batch_process(self, prompts: List[str]):
        tasks = [self.make_request(p) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 使用示例
async def main():
    client = ConcurrentAPIClient("your-api-key", max_concurrent=10)
    prompts = [f"翻译：Hello {i}" for i in range(50)]
    
    start = time.time()
    results = await client.batch_process(prompts)
    print(f"处理 {len(prompts)} 个请求耗时: {time.time() - start:.2f}秒")

asyncio.run(main())

const axios = require('axios');
const pLimit = require('p-limit');

class ConcurrentAPIClient {
    constructor(apiKey, maxConcurrent = 10) {
        this.apiKey = apiKey;
        this.baseURL = 'https://api.n1n.ai/v1';
        this.limit = pLimit(maxConcurrent);
    }

    async makeRequest(prompt) {
        return this.limit(async () => {
            const response = await axios.post(
                `${this.baseURL}/chat/completions`,
                {
                    model: 'gpt-4o-mini',
                    messages: [{ role: 'user', content: prompt }],
                    max_tokens: 100
                },
                {
                    headers: {
                        'Authorization': `Bearer ${this.apiKey}`,
                        'Content-Type': 'application/json'
                    }
                }
            );
            return response.data.choices[0].message.content;
        });
    }

    async batchProcess(prompts) {
        const start = Date.now();
        const promises = prompts.map(p => this.makeRequest(p));
        const results = await Promise.all(promises);
        
        console.log(`处理 ${prompts.length} 个请求耗时: ${(Date.now() - start) / 1000}秒`);
        return results;
    }
}

// 使用示例
async function main() {
    const client = new ConcurrentAPIClient('your-api-key', 10);
    const prompts = Array.from({ length: 50 }, (_, i) => `翻译: Hello ${i}`);
    await client.batchProcess(prompts);
}

class RateLimiter:
    """智能限流器"""
    def __init__(self):
        self.limits = {
            "gpt-4o": {"rpm": 500, "tpm": 30000},
            "gpt-4o-mini": {"rpm": 5000, "tpm": 200000}
        }
        self.request_times = []
        self.token_counts = []
    
    async def wait_if_needed(self, model: str, tokens: int):
        """必要时等待以遵守限流"""
        current_time = time.time()
        
        # 清理60秒前的记录
        self.request_times = [t for t in self.request_times if t > current_time - 60]
        self.token_counts = [(t, c) for t, c in self.token_counts if t > current_time - 60]
        
        # 检查 RPM
        if len(self.request_times) >= self.limits[model]["rpm"]:
            wait_time = 60 - (current_time - self.request_times[0])
            if wait_time > 0:
                await asyncio.sleep(wait_time)
        
        # 检查 TPM
        total_tokens = sum(c for _, c in self.token_counts) + tokens
        if total_tokens > self.limits[model]["tpm"]:
            await asyncio.sleep(5)  # 等待token窗口滑动
        
        # 记录请求
        self.request_times.append(current_time)
        self.token_counts.append((current_time, tokens))

import backoff

class RobustAPIClient:
    @backoff.on_exception(
        backoff.expo,
        (aiohttp.ClientError, asyncio.TimeoutError),
        max_tries=3,
        max_time=30
    )
    async def make_request_with_retry(self, prompt: str):
        """带指数退避重试的请求"""
        try:
            response = await self._make_request(prompt)
            return {"success": True, "data": response}
        except aiohttp.ClientResponseError as e:
            if e.status == 429:  # Rate limit
                retry_after = int(e.headers.get('Retry-After', 5))
                await asyncio.sleep(retry_after)
                raise  # 触发重试
            elif e.status >= 500:  # 服务器错误
                raise  # 触发重试
            else:
                return {"success": False, "error": str(e)}

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "successful": 0,
            "failed": 0,
            "total_tokens": 0,
            "response_times": []
        }
    
    def record_success(self, response_time: float, tokens: int):
        self.metrics["total_requests"] += 1
        self.metrics["successful"] += 1
        self.metrics["total_tokens"] += tokens
        self.metrics["response_times"].append(response_time)
    
    def get_stats(self):
        avg_time = sum(self.metrics["response_times"]) / len(self.metrics["response_times"])
        return {
            "成功率": f"{self.metrics['successful'] / self.metrics['total_requests'] * 100:.1f}%",
            "平均响应时间": f"{avg_time:.2f}秒",
            "吞吐量": f"{self.metrics['total_requests'] / sum(self.metrics['response_times']):.1f} req/s"
        }