# LiteLLM实战：5分钟搞定Python项目中的多模型切换（附避坑指南） 如果你正在开发一个基于大语言模型的应用，大概率会遇到这样的场景：项目初期用OpenAI的GPT-4跑得挺好，但上线后成本飙升，想切换到更经济的Claude或国产模型试试；或者某个关键功能需要特定领域的微调模型，而另一个功能则需要更强的推理能力。这时候，你发现每个模型的API格式、参数命名、错误处理都各不相同，改代码改到头大。 这就是为什么我们需要LiteLLM——一个能让你用同一套代码调用上百种不同大语言模型的Python库。想象一下，你只需要改变一个字符串参数，就能在GPT-4、Claude、Gemini、Llama甚至本地运行的Ollama模型之间无缝切换，而业务逻辑代码完全不用动。这不仅仅是方便，更是构建健壮、可扩展AI应用的基础设施。 我最近在一个中型企业项目中深度使用了LiteLLM，负责将原本只支持OpenAI的智能客服系统改造为多模型架构。过程中踩了不少坑，也积累了一些实战经验。这篇文章就是为你准备的快速上手指南，我会用最直白的方式告诉你如何用LiteLLM解决实际开发中的多模型切换痛点，重点放在那些文档里不会写的“坑”和解决方案上。 ## 1. 为什么你的项目需要LiteLLM：不仅仅是统一API 在深入代码之前，我们先搞清楚LiteLLM到底解决了什么问题。很多人以为它只是个API格式转换器，但实际上它的价值远不止于此。 ### 1.1 真实场景下的多模型需求 让我分享一个实际案例。我们团队开发了一个智能文档分析系统，最初只集成了GPT-4。随着用户量增长，出现了几个明显问题： - **成本失控**：GPT-4处理大量文档时，月度账单轻松突破五位数 - **单点故障**：OpenAI服务偶尔不稳定，导致整个系统瘫痪 - **功能局限**：某些专业领域任务，开源微调模型表现更好 - **合规要求**：部分客户数据不能出境，必须使用国内或本地模型 如果没有LiteLLM，我们需要为每个模型编写独立的调用逻辑，管理不同的API密钥，实现各自的错误处理。代码会迅速膨胀，维护成本呈指数级增长。 ### 1.2 LiteLLM的核心价值矩阵 为了更直观地理解LiteLLM的价值，我整理了一个功能对比表格： | 功能维度 | 原生多模型开发 | 使用LiteLLM | 优势提升 | |---------|---------------|------------|---------| | **API调用统一性** | 每个模型一套代码 | 统一`completion()`接口 | 代码量减少80%+ | | **错误处理** | 分别处理各提供商异常 | 统一OpenAI格式异常 | 调试时间减少70% | | **流式响应** | 各模型实现方式不同 | 统一`stream=True`参数 | 用户体验一致性 | | **成本跟踪** | 手动计算或依赖各平台报表 | 内置成本计算和预算控制 | 财务透明度提升 | | **模型切换** | 重构代码和测试 | 修改一个字符串参数 | 切换时间从天到分钟 | > **注意**：表格中的“优势提升”数据基于我们团队的实际项目经验，你的具体收益可能因项目复杂度而异，但方向是一致的。 ### 1.3 谁应该使用LiteLLM？ 根据我的经验，以下几类开发者会从LiteLLM中获得最大收益： 1. **中小型AI应用团队**：资源有限，需要快速试错不同模型找到最优解 2. **企业级AI系统开发者**：需要构建高可用、多后备的生成式AI服务 3. **独立开发者/创业者**：希望用最小成本验证AI产品想法，不被单一模型绑定 4. **研究者和数据科学家**：需要对比不同模型在特定任务上的表现 如果你属于以上任何一类，那么继续往下看，我会带你快速上手。 ## 2. 5分钟快速集成：从零到第一个多模型调用 理论讲得差不多了，现在让我们动手写代码。我保证，5分钟后你就能在自己的项目里调用多个大模型。 ### 2.1 环境准备与安装 首先，确保你的Python环境是3.8或更高版本。我推荐使用虚拟环境来管理依赖，避免版本冲突。 ```bash # 创建并激活虚拟环境（可选但推荐） python -m venv litellm-env source litellm-env/bin/activate # Linux/macOS # litellm-env\Scripts\activate # Windows # 安装LiteLLM pip install litellm ``` 就这么简单。LiteLLM的依赖很轻量，安装过程通常几秒钟就完成。 ### 2.2 你的第一个多模型调用 现在创建一个Python文件，比如`first_litellm.py`，写入以下代码： ```python import litellm import os # 设置环境变量（实际项目中建议使用.env文件或密钥管理服务） os.environ["OPENAI_API_KEY"] = "your-openai-key" os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key" def test_multiple_models(): """测试用同一套代码调用不同模型""" # 统一的提示词 messages = [ {"role": "system", "content": "你是一个乐于助人的助手，回答要简洁明了。"}, {"role": "user", "content": "用一句话解释量子计算的基本概念"} ] models_to_test = [ "gpt-3.5-turbo", # OpenAI "claude-3-haiku-20240307", # Anthropic Claude # "gemini/gemini-1.5-pro", # Google Gemini（需要额外配置） ] for model_name in models_to_test: try: print(f"\n{'='*50}") print(f"正在调用模型: {model_name}") print(f"{'='*50}") # 核心：统一的completion调用 response = litellm.completion( model=model_name, messages=messages, max_tokens=100, temperature=0.7 ) # 统一的响应处理 answer = response.choices[0].message.content print(f"回答: {answer}") # 额外信息：成本和使用统计 print(f"使用token数: {response.usage.total_tokens}") print(f"预估成本: ${response._response_cost if hasattr(response, '_response_cost') else 'N/A'}") except Exception as e: print(f"调用模型 {model_name} 时出错: {str(e)}") # 这里可以添加重试逻辑或切换到备用模型 if __name__ == "__main__": test_multiple_models() ``` 运行这个脚本前，你需要替换`your-openai-key`和`your-anthropic-key`为真实的API密钥。如果你暂时没有这些密钥，可以用Ollama本地模型测试： ```python # 使用本地Ollama模型（无需API密钥） response = litellm.completion( model="ollama/llama3", # 需要先安装并运行Ollama messages=messages, api_base="http://localhost:11434" # Ollama默认地址 ) ``` ### 2.3 理解LiteLLM的模型命名规则 LiteLLM使用统一的模型命名约定，这是实现多模型切换的关键。基本格式是： ``` [provider]/[model_name] ``` 常见提供商的示例： | 提供商 | 模型字符串示例 | 说明 | |--------|---------------|------| | OpenAI | `gpt-4o` | 直接使用模型名 | | Anthropic | `claude-3-5-sonnet-20241022` | Claude模型全名 | | Google | `gemini/gemini-1.5-pro` | 需要gemini前缀 | | 本地Ollama | `ollama/llama3` | ollama前缀+模型名 | | Azure OpenAI | `azure/gpt-4` | azure前缀+部署名 | | HuggingFace | `huggingface/mistralai/Mistral-7B-Instruct-v0.2` | 完整HuggingFace路径 | > **提示**：你可以在LiteLLM文档中找到完整的支持模型列表，但实践中最常用的是前几种。对于不熟悉的模型，先在小规模测试中验证其表现。 ## 3. 实战配置：管理多模型、密钥与错误处理 基础调用会了，现在进入实战环节。在实际项目中，你需要更健壮的配置和管理策略。 ### 3.1 安全的密钥管理方案 硬编码API密钥是安全大忌。以下是几种推荐的做法： **方案一：环境变量（适合开发环境）** ```bash # .env文件（不要提交到版本控制） OPENAI_API_KEY=sk-... ANTHROPIC_API_KEY=sk-ant-... AZURE_API_KEY=... AZURE_API_BASE=https://... ``` ```python # config.py import os from dotenv import load_dotenv load_dotenv() # 加载.env文件 # 获取密钥 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY") ``` **方案二：配置文件（适合复杂项目）** ```yaml # config/models.yaml models: openai: api_key: ${OPENAI_API_KEY} default_model: "gpt-4o" fallback_model: "gpt-3.5-turbo" anthropic: api_key: ${ANTHROPIC_API_KEY} default_model: "claude-3-5-sonnet" azure: api_key: ${AZURE_API_KEY} api_base: ${AZURE_API_BASE} api_version: "2023-12-01-preview" deployments: gpt4: "gpt-4-deployment" gpt35: "gpt-35-turbo-deployment" ``` **方案三：密钥管理服务（生产环境必备）** ```python # 使用AWS Secrets Manager、Azure Key Vault或Hashicorp Vault import boto3 from botocore.exceptions import ClientError def get_secret(secret_name): """从AWS Secrets Manager获取密钥""" client = boto3.client('secretsmanager') try: response = client.get_secret_value(SecretId=secret_name) return response['SecretString'] except ClientError as e: print(f"获取密钥失败: {e}") return None # 使用 openai_key = json.loads(get_secret("prod/openai-api-key"))["api_key"] ``` ### 3.2 配置模型路由与回退策略 这是LiteLLM最强大的功能之一。你可以配置一个模型列表，并定义当主模型失败时如何回退到备用模型。 ```python import litellm from litellm import Router # 定义模型列表，包含不同提供商和配置 model_list = [ { "model_name": "primary-gpt4", # 自定义名称 "litellm_params": { "model": "gpt-4o", "api_key": os.getenv("OPENAI_API_KEY"), "rpm": 10, # 每分钟请求数限制 } }, { "model_name": "backup-claude", "litellm_params": { "model": "claude-3-5-sonnet-20241022", "api_key": os.getenv("ANTHROPIC_API_KEY"), "rpm": 5, } }, { "model_name": "local-llama", "litellm_params": { "model": "ollama/llama3", "api_base": "http://localhost:11434", "timeout": 30, # 本地模型可能需要更长时间 } } ] # 创建路由器 router = Router( model_list=model_list, routing_strategy="usage-based", # 基于使用情况的智能路由 # routing_strategy="simple-shuffle", # 简单随机 # routing_strategy="latency-based", # 基于延迟 timeout=15, num_retries=2 ) # 使用路由器进行调用 async def smart_completion(messages, **kwargs): """智能路由的completion函数""" try: response = await router.acompletion( model="primary-gpt4", # 指定首选模型 messages=messages, **kwargs ) return response except Exception as e: print(f"主模型失败，尝试备用模型: {e}") # 手动切换到备用模型 for model_config in model_list[1:]: # 跳过第一个（主模型） try: response = litellm.acompletion( model=model_config["litellm_params"]["model"], messages=messages, api_key=model_config["litellm_params"].get("api_key"), api_base=model_config["litellm_params"].get("api_base"), **kwargs ) print(f"成功切换到模型: {model_config['model_name']}") return response except Exception as fallback_error: print(f"备用模型 {model_config['model_name']} 也失败: {fallback_error}") continue raise Exception("所有模型都失败了") ``` ### 3.3 错误处理与重试机制 大模型API调用失败是常态，不是异常。完善的错误处理是生产级应用的基础。 ```python import asyncio import litellm from litellm.exceptions import ( APIConnectionError, RateLimitError, ServiceUnavailableError, ContentPolicyViolationError ) class RobustLLMClient: def __init__(self, max_retries=3, backoff_factor=2): self.max_retries = max_retries self.backoff_factor = backoff_factor async def completion_with_retry(self, model, messages, **kwargs): """带指数退避重试的completion函数""" for attempt in range(self.max_retries): try: response = await litellm.acompletion( model=model, messages=messages, **kwargs ) return response except RateLimitError as e: wait_time = self.backoff_factor ** attempt print(f"速率限制，等待 {wait_time} 秒后重试 (尝试 {attempt+1}/{self.max_retries})") await asyncio.sleep(wait_time) except APIConnectionError as e: if attempt == self.max_retries - 1: raise wait_time = self.backoff_factor ** attempt print(f"连接错误，等待 {wait_time} 秒后重试") await asyncio.sleep(wait_time) except ServiceUnavailableError as e: if attempt == self.max_retries - 1: # 服务不可用，切换到备用模型 return await self.fallback_completion(messages, **kwargs) wait_time = self.backoff_factor ** attempt * 5 # 服务不可用等待更久 print(f"服务不可用，等待 {wait_time} 秒后重试") await asyncio.sleep(wait_time) except ContentPolicyViolationError as e: # 内容策略违规，不能重试，需要修改提示词 print(f"内容策略违规: {e}") raise except Exception as e: print(f"未知错误 (尝试 {attempt+1}/{self.max_retries}): {e}") if attempt == self.max_retries - 1: raise await asyncio.sleep(self.backoff_factor ** attempt) async def fallback_completion(self, messages, **kwargs): """回退到备用模型的逻辑""" fallback_models = ["claude-3-haiku", "gpt-3.5-turbo", "ollama/llama3"] for fallback_model in fallback_models: try: print(f"尝试回退模型: {fallback_model}") response = await litellm.acompletion( model=fallback_model, messages=messages, **kwargs ) return response except Exception as e: print(f"回退模型 {fallback_model} 失败: {e}") continue raise Exception("所有回退模型都失败了") # 使用示例 async def main(): client = RobustLLMClient(max_retries=3) messages = [{"role": "user", "content": "解释一下机器学习中的过拟合现象"}] try: response = await client.completion_with_retry( model="gpt-4o", messages=messages, temperature=0.7, max_tokens=500 ) print(f"成功获取响应: {response.choices[0].message.content[:100]}...") except Exception as e: print(f"最终失败: {e}") # 运行 if __name__ == "__main__": asyncio.run(main()) ``` ## 4. 高级特性：流式响应、成本跟踪与性能优化 掌握了基础用法后，让我们看看LiteLLM的一些高级功能，这些功能能让你的应用更专业、更高效。 ### 4.1 实现流式响应提升用户体验 对于需要长时间生成内容的场景，流式响应能显著提升用户体验。LiteLLM让这变得非常简单。 ```python import litellm import asyncio from typing import AsyncGenerator class StreamingLLMClient: def __init__(self, model="gpt-4o"): self.model = model async def stream_completion(self, messages, **kwargs) -> AsyncGenerator[str, None]: """流式生成响应""" try: response = await litellm.acompletion( model=self.model, messages=messages, stream=True, # 关键参数 **kwargs ) full_response = "" async for chunk in response: if chunk.choices[0].delta.content is not None: content = chunk.choices[0].delta.content full_response += content yield content # 实时返回每个片段 # 流式结束后，可以记录完整响应 print(f"\n完整响应长度: {len(full_response)} 字符") except Exception as e: yield f"[错误: {str(e)}]" def stream_completion_sync(self, messages, **kwargs): """同步版本的流式响应（用于非异步环境）""" response = litellm.completion( model=self.model, messages=messages, stream=True, **kwargs ) full_response = "" for chunk in response: if chunk.choices[0].delta.content is not None: content = chunk.choices[0].delta.content full_response += content yield content print(f"\n完整响应: {full_response}") # 使用示例 - 异步版本 async def demo_async_stream(): client = StreamingLLMClient(model="gpt-3.5-turbo") messages = [ {"role": "system", "content": "你是一个技术文档作家，用清晰的语言解释复杂概念。"}, {"role": "user", "content": "详细解释RESTful API设计的最佳实践，至少列出10条。"} ] print("开始流式响应:") print("-" * 50) async for chunk in client.stream_completion(messages, max_tokens=1000): print(chunk, end="", flush=True) # 实时打印 print("\n" + "-" * 50) print("流式响应结束") # 使用示例 - 同步版本 def demo_sync_stream(): client = StreamingLLMClient(model="claude-3-haiku") messages = [ {"role": "user", "content": "写一个关于人工智能的短故事"} ] print("故事生成中:") print("-" * 30) for chunk in client.stream_completion_sync(messages, max_tokens=300): print(chunk, end="", flush=True) print("\n" + "-" * 30) # 运行演示 if __name__ == "__main__": # 选择运行异步或同步版本 import sys if len(sys.argv) > 1 and sys.argv[1] == "async": asyncio.run(demo_async_stream()) else: demo_sync_stream() ``` ### 4.2 成本跟踪与预算控制 对于商业应用，成本控制至关重要。LiteLLM提供了内置的成本跟踪功能。 ```python import litellm import json from datetime import datetime, timedelta from typing import Dict, List class CostTracker: def __init__(self, budget_limit=100.0): # 默认预算100美元 self.budget_limit = budget_limit self.daily_costs: Dict[str, float] = {} self.model_usage: Dict[str, Dict] = {} # 设置成本回调 litellm.success_callback = [self.track_cost_callback] litellm.failure_callback = [self.track_failure_callback] def track_cost_callback(self, kwargs, completion_response, start_time, end_time): """成功调用的成本跟踪回调""" model = kwargs.get("model", "unknown") response_cost = completion_response._response_cost if hasattr(completion_response, '_response_cost') else 0 # 更新日成本 today = datetime.now().strftime("%Y-%m-%d") self.daily_costs[today] = self.daily_costs.get(today, 0) + response_cost # 更新模型使用统计 if model not in self.model_usage: self.model_usage[model] = { "total_cost": 0, "call_count": 0, "total_tokens": 0 } self.model_usage[model]["total_cost"] += response_cost self.model_usage[model]["call_count"] += 1 self.model_usage[model]["total_tokens"] += getattr(completion_response.usage, 'total_tokens', 0) # 检查预算 if self.daily_costs[today] > self.budget_limit: print(f"⚠️ 警告: 今日成本已超过预算限制 (${self.budget_limit})") # 这里可以触发警报或自动切换到更便宜的模型 def track_failure_callback(self, kwargs, exception, start_time, end_time): """失败调用的跟踪（可能仍有成本）""" model = kwargs.get("model", "unknown") print(f"模型 {model} 调用失败: {exception}") def get_daily_report(self, date=None) -> Dict: """获取指定日期的成本报告""" if date is None: date = datetime.now().strftime("%Y-%m-%d") return { "date": date, "total_cost": self.daily_costs.get(date, 0), "budget_limit": self.budget_limit, "remaining_budget": self.budget_limit - self.daily_costs.get(date, 0), "model_breakdown": { model: stats for model, stats in self.model_usage.items() # 这里可以按日期过滤，简化示例显示全部 } } def get_cost_optimization_suggestions(self) -> List[str]: """基于使用情况提供成本优化建议""" suggestions = [] total_cost = sum(self.daily_costs.values()) if total_cost == 0: return ["暂无使用数据"] # 分析模型使用情况 for model, stats in self.model_usage.items(): avg_cost_per_call = stats["total_cost"] / stats["call_count"] if stats["call_count"] > 0 else 0 avg_tokens_per_call = stats["total_tokens"] / stats["call_count"] if stats["call_count"] > 0 else 0 # 根据模型提供具体建议 if "gpt-4" in model and stats["call_count"] > 10: suggestions.append( f"考虑将部分 {model} 调用降级到 gpt-3.5-turbo，预计可节省 {stats['total_cost'] * 0.7:.2f} 美元" ) if avg_tokens_per_call > 2000: suggestions.append( f"模型 {model} 的平均响应过长 ({avg_tokens_per_call:.0f} tokens)，考虑设置 max_tokens 限制" ) # 总体建议 if total_cost > self.budget_limit * 0.8: # 使用超过预算80% suggestions.append( f"当前使用率已达预算的 {(total_cost/self.budget_limit)*100:.1f}%，建议增加预算或优化使用策略" ) return suggestions if suggestions else ["当前使用模式较为经济"] # 使用示例 def demo_cost_tracking(): tracker = CostTracker(budget_limit=50.0) # 50美元日预算 # 模拟多次调用 models_to_test = ["gpt-3.5-turbo", "gpt-4o", "claude-3-haiku"] for i, model in enumerate(models_to_test): try: print(f"\n测试调用 {i+1}: {model}") response = litellm.completion( model=model, messages=[{"role": "user", "content": "解释一下深度学习"}], max_tokens=100 * (i+1) # 逐渐增加token数 ) print(f"响应: {response.choices[0].message.content[:50]}...") except Exception as e: print(f"调用失败: {e}") # 生成报告 print("\n" + "="*50) print("成本跟踪报告") print("="*50) report = tracker.get_daily_report() print(f"日期: {report['date']}") print(f"总成本: ${report['total_cost']:.4f}") print(f"预算限制: ${report['budget_limit']:.2f}") print(f"剩余预算: ${report['remaining_budget']:.2f}") print("\n模型使用详情:") for model, stats in report["model_breakdown"].items(): print(f" {model}:") print(f" 调用次数: {stats['call_count']}") print(f" 总成本: ${stats['total_cost']:.4f}") print(f" 总tokens: {stats['total_tokens']}") print("\n优化建议:") for suggestion in tracker.get_cost_optimization_suggestions(): print(f" • {suggestion}") if __name__ == "__main__": demo_cost_tracking() ``` ### 4.3 性能优化与缓存策略 对于高并发应用，性能优化是关键。LiteLLM支持多种缓存策略。 ```python import litellm import hashlib import json from functools import lru_cache from datetime import datetime, timedelta class OptimizedLLMClient: def __init__(self, use_cache=True, cache_ttl=3600): self.use_cache = use_cache self.cache_ttl = cache_ttl # 缓存过期时间（秒） self.response_cache = {} # 简单内存缓存，生产环境可用Redis # 性能优化配置 litellm.drop_params = True # 自动移除模型不支持的参数 litellm.set_verbose = False # 生产环境关闭详细日志 def _generate_cache_key(self, model, messages, **kwargs) -> str: """生成缓存键""" # 排除流式参数和随机性参数 cacheable_kwargs = {k: v for k, v in kwargs.items() if k not in ['stream', 'temperature', 'top_p', 'n']} cache_data = { "model": model, "messages": messages, "kwargs": cacheable_kwargs } # 使用SHA256生成唯一键 cache_str = json.dumps(cache_data, sort_keys=True) return hashlib.sha256(cache_str.encode()).hexdigest() @lru_cache(maxsize=100) def cached_completion(self, cache_key: str, model: str, messages, **kwargs): """带LRU缓存的内存缓存""" print(f"缓存未命中，实际调用模型: {model}") return litellm.completion(model=model, messages=messages, **kwargs) def completion_with_cache(self, model, messages, **kwargs): """带缓存的completion调用""" # 流式响应不缓存 if kwargs.get('stream', False): return litellm.completion(model=model, messages=messages, **kwargs) # 高随机性请求不缓存 if kwargs.get('temperature', 0) > 0.9 or kwargs.get('top_p', 1) < 0.1: return litellm.completion(model=model, messages=messages, **kwargs) if not self.use_cache: return litellm.completion(model=model, messages=messages, **kwargs) cache_key = self._generate_cache_key(model, messages, **kwargs) # 检查内存缓存 if cache_key in self.response_cache: cache_entry = self.response_cache[cache_key] cache_time = cache_entry['timestamp'] # 检查是否过期 if datetime.now() - cache_time < timedelta(seconds=self.cache_ttl): print(f"缓存命中: {cache_key[:16]}...") return cache_entry['response'] else: # 缓存过期，移除 del self.response_cache[cache_key] # 实际调用 response = litellm.completion(model=model, messages=messages, **kwargs) # 存储到缓存 self.response_cache[cache_key] = { 'response': response, 'timestamp': datetime.now(), 'model': model, 'message_count': len(messages) } # 清理过期缓存（简单示例，生产环境需要更高效的清理策略） if len(self.response_cache) > 1000: # 限制缓存大小 self._clean_expired_cache() return response def _clean_expired_cache(self): """清理过期缓存""" now = datetime.now() expired_keys = [] for key, entry in self.response_cache.items(): if now - entry['timestamp'] > timedelta(seconds=self.cache_ttl): expired_keys.append(key) for key in expired_keys: del self.response_cache[key] print(f"清理了 {len(expired_keys)} 个过期缓存项") def batch_completion(self, requests, max_concurrent=5): """批量处理多个completion请求""" import asyncio async def process_batch(): semaphore = asyncio.Semaphore(max_concurrent) async def process_one(request): async with semaphore: model = request.get('model', 'gpt-3.5-turbo') messages = request['messages'] kwargs = request.get('kwargs', {}) try: response = await litellm.acompletion( model=model, messages=messages, **kwargs ) return {'success': True, 'response': response} except Exception as e: return {'success': False, 'error': str(e)} # 并发处理所有请求 tasks = [process_one(req) for req in requests] return await asyncio.gather(*tasks) return asyncio.run(process_batch()) # 性能测试示例 def performance_demo(): client = OptimizedLLMClient(use_cache=True, cache_ttl=300) # 5分钟缓存 # 测试数据 test_messages = [ {"role": "user", "content": "Python中如何读取CSV文件？"} ] import time # 第一次调用（缓存未命中） print("第一次调用（缓存未命中）:") start = time.time() response1 = client.completion_with_cache( model="gpt-3.5-turbo", messages=test_messages, max_tokens=100 ) time1 = time.time() - start print(f"时间: {time1:.2f}秒") print(f"响应: {response1.choices[0].message.content[:50]}...\n") # 第二次调用（缓存命中） print("第二次调用（缓存命中）:") start = time.time() response2 = client.completion_with_cache( model="gpt-3.5-turbo", messages=test_messages, max_tokens=100 ) time2 = time.time() - start print(f"时间: {time2:.2f}秒") print(f"缓存加速: {time1/time2:.1f}倍\n") # 批量处理演示 print("批量处理演示:") batch_requests = [ { 'model': 'gpt-3.5-turbo', 'messages': [{'role': 'user', 'content': f'问题 {i}: 解释概念{i}'}], 'kwargs': {'max_tokens': 50} } for i in range(5) ] start = time.time() batch_results = client.batch_completion(batch_requests, max_concurrent=3) batch_time = time.time() - start print(f"批量处理5个请求用时: {batch_time:.2f}秒") print(f"平均每个请求: {batch_time/5:.2f}秒") success_count = sum(1 for r in batch_results if r['success']) print(f"成功: {success_count}/{len(batch_requests)}") if __name__ == "__main__": performance_demo() ``` ## 5. 避坑指南：我踩过的坑和解决方案 在真实项目中使用LiteLLM时，我遇到了一些预料之外的问题。这里分享出来，希望能帮你避开这些坑。 ### 5.1 常见问题与解决方案 **问题1：模型响应格式不一致** 虽然LiteLLM承诺统一响应格式，但不同模型在细节上仍有差异。 ```python def safe_extract_response(response): """安全地从不同模型响应中提取内容""" # 方法1: 标准OpenAI格式 if hasattr(response, 'choices') and len(response.choices) > 0: choice = response.choices[0] if hasattr(choice, 'message') and hasattr(choice.message, 'content'): return choice.message.content # 方法2: Anthropic格式（通过LiteLLM转换后） if hasattr(response, 'content') and isinstance(response.content, list): for item in response.content: if hasattr(item, 'text'): return item.text # 方法3: 原始响应文本 if isinstance(response, str): return response # 方法4: 尝试JSON解析 try: import json if isinstance(response, dict): response_dict = response else: response_dict = json.loads(str(response)) # 尝试常见路径 paths_to_try = [ ['choices', 0, 'message', 'content'], ['content', 0, 'text'], ['text'], ['response'], ['answer'] ] for path in paths_to_try: current = response_dict try: for key in path: if isinstance(key, int) and isinstance(current, list): current = current[key] elif isinstance(current, dict): current = current[key] else: break else: if isinstance(current, str): return current except (KeyError, IndexError, TypeError): continue except: pass # 最后手段 return str(response) ``` **问题2：速率限制处理不当** 每个模型提供商都有不同的速率限制策略，需要分别处理。 ```python class RateLimitManager: def __init__(self): self.provider_limits = { 'openai': { 'rpm': 60, # 每分钟请求数 'tpm': 60000, # 每分钟tokens数 'last_request': None, 'request_count': 0, 'token_count': 0 }, 'anthropic': { 'rpm': 30, 'tpm': 40000, 'last_request': None, 'request_count': 0, 'token_count': 0 }, 'azure': { 'rpm': 120, # Azure通常有更高限制 'tpm': 120000, 'last_request': None, 'request_count': 0, 'token_count': 0 } } def get_provider_from_model(self, model_name): """从模型名推断提供商""" if model_name.startswith('gpt-'): return 'openai' elif model_name.startswith('claude-'): return 'anthropic' elif model_name.startswith('azure/'): return 'azure' elif model_name.startswith('gemini'): return 'google' else: return 'unknown' async def wait_if_needed(self, model_name, estimated_tokens=100): """如果需要，等待直到可以安全发送请求""" provider = self.get_provider_from_model(model_name) if provider not in self.provider_limits: return # 未知提供商，不进行限制 limits = self.provider_limits[provider] now = datetime.now() # 重置每分钟计数 if limits['last_request'] and (now - limits['last_request']).seconds >= 60: limits['request_count'] = 0 limits['token_count'] = 0 # 检查请求数限制 if limits['request_count'] >= limits['rpm']: wait_time = 60 - (now - limits['last_request']).seconds if wait_time > 0: print(f"达到 {provider} RPM 限制，等待 {wait_time} 秒") await asyncio.sleep(wait_time) # 检查token数限制 if limits['token_count'] + estimated_tokens > limits['tpm']: wait_time = 60 - (now - limits['last_request']).seconds if wait_time > 0: print(f"达到 {provider} TPM 限制，等待 {wait_time} 秒") await asyncio.sleep(wait_time) # 更新计数 limits['request_count'] += 1 limits['token_count'] += estimated_tokens limits['last_request'] = now ``` **问题3：长上下文处理** 不同模型对上下文长度的支持不同，需要智能截断。 ```python def smart_truncate_messages(messages, model_name, max_tokens=4000): """根据模型智能截断消息""" # 不同模型的上下文窗口大小 context_windows = { 'gpt-3.5-turbo': 4096, 'gpt-4o': 128000, 'claude-3-5-sonnet': 200000, 'claude-3-haiku': 200000, 'gemini-1.5-pro': 1000000, 'llama3': 8192, } # 获取模型的上下文窗口，默认4000 window_size = context_windows.get(model_name.split('/')[-1], 4000) # 保留10%的余量给响应 available_tokens = int(window_size * 0.9) if max_tokens > available_tokens: max_tokens = available_tokens # 简单估算token数（实际应该使用tiktoken等库） def estimate_tokens(text): # 粗略估算：英文约4字符1个token，中文约2字符1个token import re chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text)) other_chars = len(text) - chinese_chars return int(chinese_chars / 2 + other_chars / 4) total_tokens = 0 truncated_messages = [] # 从最新消息开始添加（保持对话连贯性） for message in reversed(messages): message_tokens = estimate_tokens(message.get('content', '')) + 10 # 加上角色等开销 if total_tokens + message_tokens > available_tokens - max_tokens: # 如果添加这条消息会超出限制，停止添加 break truncated_messages.insert(0, message) # 保持原始顺序 total_tokens += message_tokens # 如果还是太长，截断最后一条消息的内容 if truncated_messages and total_tokens > available_tokens - max_tokens: last_message = truncated_messages[-1] content = last_message.get('content', '') # 保留开头部分（通常是最重要的） max_chars = int((available_tokens - max_tokens - (total_tokens - estimate_tokens(content))) * 4) if max_chars < 100: # 至少保留100字符 max_chars = 100 truncated_content = content[:max_chars] + "... [内容已截断]" truncated_messages[-1] = {**last_message, 'content': truncated_content} return truncated_messages ``` ### 5.2 调试技巧与工具 当LiteLLM调用出现问题时，以下调试技巧很有用： ```python # 1. 启用详细日志 litellm.set_verbose = True # 2. 自定义日志回调 def debug_callback(kwargs, response, start_time, end_time): print(f"\n=== 调试信息 ===") print(f"模型: {kwargs.get('model')}") print(f"消息数: {len(kwargs.get('messages', []))}") print(f"耗时: {(end_time - start_time).total_seconds():.2f}秒") if hasattr(response, 'usage'): print(f"使用token: {response.usage.total_tokens}") if hasattr(response, '_response_cost'): print(f"成本: ${response._response_cost}") # 记录原始请求和响应（敏感信息需脱敏） import json debug_info = { 'timestamp': start_time.isoformat(), 'model': kwargs.get('model'), 'request': { 'messages': kwargs.get('messages'), 'params': {k: v for k, v in kwargs.items() if k not in ['messages', 'api_key']} }, 'response': { 'content': response.choices[0].message.content[:200] if hasattr(response, 'choices') else str(response)[:200], 'usage': getattr(response, 'usage', {}), 'cost': getattr(response, '_response_cost', None) } } # 保存到文件（生产环境应使用日志系统） with open('litellm_debug.log', 'a') as f: f.write(json.dumps(debug_info, ensure_ascii=False) + '\n') litellm.success_callback.append(debug_callback) # 3. 健康检查函数 async def health_check(models_to_check=None): """检查所有配置的模型是否可用""" if models_to_check is None: models_to_check = [ 'gpt-3.5-turbo', 'claude-3-haiku', 'ollama/llama3' ] results = {} test_message = [{"role": "user", "content": "回复'OK'表示你正常工作。"}] for model in models_to_check: try: start = datetime.now() response = await litellm.acompletion( model=model, messages=test_message, max_tokens=10, timeout=10 ) elapsed = (datetime.now() - start).total_seconds() if 'OK' in response.choices[0].message.content: results[model] = { 'status': 'healthy', 'response_time': elapsed, 'response': response.choices[0].message.content } else: results[model] = { 'status': 'unexpected_response', 'response_time': elapsed, 'response': response.choices[0].message.content } except Exception as e: results[model] = { 'status': 'error', 'error': str(e), 'response_time': None } # 生成健康报告 print("\n" + "="*50) print("模型健康检查报告") print("="*50) healthy_count = sum(1 for r in results.values() if r['status'] == 'healthy') total_count = len(results) print(f"总体健康度: {healthy_count}/{total_count} ({healthy_count/total_count*100:.1f}%)") for model, result in results.items(): status_icon = "✅" if result['status'] == 'healthy' else "❌" print(f"{status_icon} {model}: {result['status']}", end="") if result['status'] == 'healthy': print(f" (响应时间: {result['response_time']:.2f}秒)") elif 'error' in result: print(f" (错误: {result['error'][:50]}...)") else: print(f" (响应: {result['response'][:30]}...)") return results ``` ### 5.3 生产环境部署建议 基于实际项目经验，以下是我总结的生产环境部署建议： 1. **使用LiteLLM代理模式**：对于多实例部署，使用LiteLLM代理作为统一的API网关 2. **实现分级降级策略**：定义明确的模型降级路径（如GPT-4 → GPT-3.5 → Claude Haiku → 本地模型） 3. **设置预算告警**：当成本达到预算的50%、80%、90%时发送告警 4. **监控关键指标**： - 各模型成功率、响应时间、错误率 - 成本随时间变化趋势 - Token使用效率（有效输出/总token） 5. **定期模型评估**：每月评估各模型在关键任务上的表现，调整路由策略 ```yaml # 生产环境配置示例 (docker-compose.yml) version: '3.8' services: litellm-proxy: image: ghcr.io/berriai/litellm:main-latest ports: - "4000:4000" volumes: - ./config.yaml:/app/config.yaml - ./logs:/app/logs environment: - LITELLM_MASTER_KEY=${LITELLM_MASTER_KEY} - DATABASE_URL=postgresql://user:pass@db:5432/litellm - REDIS_URL=redis://redis:6379/0 depends_on: - db - redis command: > --config /app/config.yaml --port 4000 --num_workers 4 --detailed_debug --log_file /app/logs/litellm.log db: image: postgres:15 environment: - POSTGRES_USER=litellm - POSTGRES_PASSWORD=${DB_PASSWORD} - POSTGRES_DB=litellm volumes: - postgres_data:/var/lib/postgresql/data redis: image: redis:7-alpine volumes: - redis_data:/data volumes: postgres_data: redis_data: ``` 通过以上配置，你可以获得一个高可用的LiteLLM代理服务，支持负载均衡、故障转移、详细日志和持久化存储。