LLM API 网关开发者指南：OpenRouter、Portkey、Helicone 与国内中转怎么选

import requests
import json
import time

class LLMGateway:
    def __init__(self, api_key, base_url="https://api.laozhang.ai" ):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
    
    def chat_completion(self, model, messages, **kwargs):
        """
        统一的聊天接口，支持所有模型
        """
        payload = {
            "model": model,
            "messages": messages,
            **kwargs
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/v1/chat/completions",
                headers=self.headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"API调用失败: {e}")
            return None
    
    def smart_routing(self, messages, max_cost=0.1):
        """
        智能路由：根据任务复杂度选择最合适的模型
        """
        # 简单任务判断
        total_tokens = sum(len(m['content'].split()) for m in messages)
        
        if total_tokens < 100:
            # 简单任务使用便宜的模型
            model = "gpt-6-luna"
        elif total_tokens < 500:
            # 中等任务使用平衡的模型
            model = "gpt-6.1-sol"
        else:
            # 复杂任务使用高性能模型
            model = "gpt-6-astra"
        
        print(f"智能路由选择模型: {model}")
        return self.chat_completion(model, messages)
    
    def parallel_inference(self, messages, models=["gpt-6.1-sol", "grok-4.7"]):
        """
        并行推理：同时调用多个模型，选择最快的响应
        """
        import concurrent.futures
        
        def call_model(model):
            start_time = time.time()
            result = self.chat_completion(model, messages)
            elapsed = time.time() - start_time
            return model, result, elapsed
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=len(models)) as executor:
            futures = [executor.submit(call_model, model) for model in models]
            
            # 获取第一个完成的结果
            for future in concurrent.futures.as_completed(futures):
                model, result, elapsed = future.result()
                print(f"模型 {model} 响应时间: {elapsed:.2f}秒")
                
                # 取消其他未完成的任务
                for f in futures:
                    f.cancel()
                
                return result
    
    def image_generation(self, prompt, model="gpt-image-2", quality="medium"):
        """
        图像生成示例（gpt-image-1 将于 2026-10-23 停用，示例改用 gpt-image-2）
        """
        payload = {
            "model": model,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": prompt
                        }
                    ]
                }
            ],
            "n": 1,  # 生成数量
            "quality": quality  # low/medium/high/auto
        }
        
        return self.chat_completion(model, payload['messages'], n=payload['n'])

# 使用示例
if __name__ == "__main__":
    # 初始化网关（使用您的API密钥）
    gateway = LLMGateway(api_key="your-api-key-here")
    
    # 示例1：基础对话
    messages = [
        {"role": "system", "content": "你是一个专业的技术顾问"},
        {"role": "user", "content": "解释什么是LLM API网关"}
    ]
    
    response = gateway.chat_completion("gpt-6-luna", messages)
    if response:
        print("基础对话响应:", response['choices'][0]['message']['content'])
    
    # 示例2：智能路由
    complex_messages = [
        {"role": "user", "content": "请详细分析transformer架构的自注意力机制，包括数学推导"}
    ]
    
    smart_response = gateway.smart_routing(complex_messages)
    
    # 示例3：并行推理提高可用性
    parallel_response = gateway.parallel_inference(messages)
    
    # 示例4：图像生成（laozhang.ai 上 gpt-image-2 截至 2026-10-08 为 $0.03/次）
    image_response = gateway.image_generation(
        prompt="画一个现代化的API网关架构图，包含负载均衡和智能路由【16:9】",
        quality="high"
    )

import hashlib
import redis
import json

class CachedLLMGateway(LLMGateway):
    def __init__(self, api_key, redis_client=None):
        super().__init__(api_key)
        self.cache = redis_client or redis.Redis()
        self.cache_ttl = 3600  # 1小时缓存
    
    def _get_cache_key(self, model, messages):
        content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
        return f"llm_cache:{hashlib.md5(content.encode()).hexdigest()}"
    
    def chat_completion_with_cache(self, model, messages, use_cache=True):
        if not use_cache:
            return self.chat_completion(model, messages)
        
        cache_key = self._get_cache_key(model, messages)
        
        # 尝试从缓存获取
        cached = self.cache.get(cache_key)
        if cached:
            print("缓存命中！")
            return json.loads(cached)
        
        # 缓存未命中，调用API
        response = self.chat_completion(model, messages)
        if response:
            self.cache.setex(cache_key, self.cache_ttl, json.dumps(response))
        
        return response

def batch_process_with_rate_limit(gateway, tasks, max_rpm=60):
    """
    批量处理任务，遵守速率限制
    """
    import time
    from collections import deque
    
    results = []
    request_times = deque()
    
    for task in tasks:
        # 速率限制检查
        current_time = time.time()
        request_times = deque([t for t in request_times if current_time - t < 60])
        
        if len(request_times) >= max_rpm:
            sleep_time = 60 - (current_time - request_times[0]) + 0.1
            print(f"达到速率限制，等待 {sleep_time:.1f} 秒")
            time.sleep(sleep_time)
        
        # 执行请求
        result = gateway.chat_completion(task['model'], task['messages'])
        results.append(result)
        request_times.append(time.time())
        
        # 可选：显示进度
        print(f"进度: {len(results)}/{len(tasks)}")
    
    return results