LiteLLM 提供商切换

什么是 LiteLLM？

LiteLLM 是一个统一 100+ LLM 提供商 API 的框架。它通过提供对原始 LLM API 的高层抽象，简化了 AI 应用的构建。

最适合：多提供商

安装

bash
pip install litellm
或者使用 uv:
uv add litellm

核心概念

LiteLLM 围绕几个关键思想构建：

快速开始

python
最小工作示例
import os
os.environ["OPENAI_API_KEY"] = "sk-..."
导入 LiteLLM
（请参阅特定框架的文档以获取准确的导入方式）
统一 100+ LLM 提供商 API 的基本使用模式
def create_pipeline():
    """创建一个用于多提供商的 LiteLLM 管道。"""
    # 1. 初始化框架
    # 2. 配置你的 LLM（GPT-4o、Claude 等）
    # 3. 定义管道逻辑
    # 4. 返回配置好的管道
    pass
pipeline = create_pipeline()
result = pipeline.run("你的输入")
print(result)

真实示例：多提供商

python
from openai import OpenAI
import json
class LiteLLMPipeline:
    """
    用于多提供商的 LiteLLM 实现。
    
    架构：
    - 输入验证
    - LiteLLM 处理
    - 输出结构化
    """
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.system_prompt = f"""你是一个专门研究 {specialty} 的 AI 助手。
        利用你的专业知识提供准确、有用的回答。
        回答要简洁且结构化。"""
    
    def process(self, user_input: str, context: dict = None) -> dict:
        """通过 LiteLLM 管道处理输入。"""
        
        # 构建上下文感知的提示
        context_str = json.dumps(context, indent=2) if context else "None"
        
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": f"上下文：\n{context_str}\n\n请求：\n{user_input}"}
        ]
        
        # 执行 LLM 调用
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.2,
            max_tokens=2000
        )
        
        content = response.choices[0].message.content
        
        return {
            "result": content,
            "model": self.model,
            "framework": "LiteLLM",
            "tokens_used": response.usage.total_tokens
        }
    
    def batch_process(self, inputs: list[str]) -> list[dict]:
        """高效处理多个输入。"""
        return [self.process(inp) for inp in inputs]
使用
pipeline = LiteLLMPipeline()
result = pipeline.process("用代码示例解释多提供商")
print(result["result"])
print(f"使用的 tokens：{result['tokens_used']}")

高级模式

流式响应

python
def stream_response(self, user_input: str):
    """流式传输 tokens 以实现实时输出。"""
    stream = self.client.chat.completions.create(
        model=self.model,
        messages=[{"role": "user", "content": user_input}],
        stream=True
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

错误处理与重试

python
import time
from openai import RateLimitError, APIError
def process_with_retry(self, input_text: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            return self.process(input_text)
        except RateLimitError:
            wait_time = 2 ** attempt
            print(f"触发速率限制，等待 {wait_time} 秒...")
            time.sleep(wait_time)
        except APIError as e:
            if attempt == max_retries - 1:
                raise
            print(f"API 错误：{e}，正在重试...")
    raise Exception("超过最大重试次数")

测试

python
import pytest
@pytest.fixture
def pipeline():
    return LiteLLMPipeline(model="gpt-4o-mini")
def test_basic_processing(pipeline):
    result = pipeline.process("什么是多提供商？")
    assert "result" in result
    assert len(result["result"]) > 10
def test_batch_processing(pipeline):
    inputs = ["问题 1", "问题 2", "问题 3"]
    results = pipeline.batch_process(inputs)
    assert len(results) == len(inputs)

生产部署

python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="LiteLLM API")
pipeline = LiteLLMPipeline()
class ProcessRequest(BaseModel):
    input: str
    context: dict = {}
@app.get("/health")
async def health():
    return {"status": "ok", "framework": "LiteLLM"}

最佳实践

资源

LiteLLM 官方文档

包含示例的 GitHub 仓库

社区 Discord/Slack 支持

包含真实世界模式的食谱