1 框架介绍
1.1 核心理念
01.统一抽象层
a.多模型统一接口
a.功能说明
LiteLLM提供统一的OpenAI格式接口调用100+种LLM。无需修改代码即可切换不同模型提供商。支持OpenAI、Anthropic、Cohere、Azure等主流服务。统一接口简化多模型管理和切换。
b.代码示例
---
# 1. 基础调用示例
from litellm import completion
import os
# OpenAI
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
# Anthropic Claude
response = completion(
model="claude-3-sonnet-20240229",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
# Azure OpenAI
response = completion(
model="azure/gpt-35-turbo",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
# Cohere
response = completion(
model="command-nightly",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
print("✓ 统一接口调用完成")
# 2. 模型标识符格式
# 格式: provider/model-name
models = [
"gpt-3.5-turbo", # OpenAI (默认)
"claude-3-opus-20240229", # Anthropic
"azure/gpt-4", # Azure OpenAI
"bedrock/anthropic.claude-v2", # AWS Bedrock
"vertex_ai/gemini-pro", # Google Vertex AI
"palm/chat-bison", # Google PaLM
"replicate/llama-2-70b", # Replicate
"huggingface/mistralai/Mistral-7B-v0.1", # HuggingFace
"ollama/qwen:7b", # Ollama本地模型
]
for model in models:
print(f"支持的模型: {model}")
# 3. 动态模型切换
def chat_with_model(model_name, message):
"""统一接口调用不同模型"""
try:
response = completion(
model=model_name,
messages=[{"role": "user", "content": message}]
)
return response.choices[0].message.content
except Exception as e:
return f"调用失败: {str(e)}"
# 使用不同模型
result1 = chat_with_model("gpt-3.5-turbo", "介绍Python")
result2 = chat_with_model("claude-3-sonnet-20240229", "介绍Python")
result3 = chat_with_model("ollama/qwen:7b", "介绍Python")
print(f"GPT-3.5: {result1}")
print(f"Claude: {result2}")
print(f"Qwen: {result3}")
# 4. 配置管理
import litellm
# 设置API密钥
litellm.openai_key = os.getenv("OPENAI_API_KEY")
litellm.anthropic_key = os.getenv("ANTHROPIC_API_KEY")
litellm.cohere_key = os.getenv("COHERE_API_KEY")
# 设置超时
litellm.request_timeout = 60
# 设置重试
litellm.num_retries = 3
# 启用调试日志
litellm.set_verbose = True
# 5. 响应格式统一
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
# 统一的响应结构
print(f"模型: {response.model}")
print(f"内容: {response.choices[0].message.content}")
print(f"角色: {response.choices[0].message.role}")
print(f"完成原因: {response.choices[0].finish_reason}")
print(f"输入tokens: {response.usage.prompt_tokens}")
print(f"输出tokens: {response.usage.completion_tokens}")
print(f"总tokens: {response.usage.total_tokens}")
# 6. 异常处理统一
from litellm import AuthenticationError, RateLimitError, ServiceUnavailableError
def safe_completion(model, messages):
"""带异常处理的统一调用"""
try:
return completion(model=model, messages=messages)
except AuthenticationError:
return "认证失败，请检查API密钥"
except RateLimitError:
return "达到速率限制，请稍后重试"
except ServiceUnavailableError:
return "服务不可用"
except Exception as e:
return f"未知错误: {str(e)}"
result = safe_completion(
"gpt-3.5-turbo",
[{"role": "user", "content": "你好"}]
)
print(result)
print("✓ 核心理念示例完成")
---
b.成本优化
a.功能说明
自动选择性价比最优的模型。实现请求路由和负载均衡。支持预算控制和成本追踪。成本优化降低AI应用开支。
b.代码示例
---
# 1. 成本追踪
import litellm
from litellm import completion_cost
# 启用成本追踪
litellm.success_callback = ["langfuse"]
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "介绍AI"}]
)
# 计算成本
cost = completion_cost(
completion_response=response
)
print(f"本次调用成本: ${cost:.6f}")
print(f"输入tokens: {response.usage.prompt_tokens}")
print(f"输出tokens: {response.usage.completion_tokens}")
# 2. 预算控制
from litellm import BudgetManager
# 创建预算管理器
budget_manager = BudgetManager(
project_name="my_project",
client_type="local"
)
# 设置预算上限
budget_manager.create_budget(
budget_id="daily_budget",
max_budget=10.0, # 每日最多$10
duration="1d"
)
# 检查预算
def chat_with_budget(message):
if budget_manager.get_current_cost("daily_budget") >= 10.0:
return "今日预算已用完"
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": message}]
)
# 记录成本
cost = completion_cost(response)
budget_manager.update_cost("daily_budget", cost)
return response.choices[0].message.content
result = chat_with_budget("你好")
print(result)
# 3. 模型降级策略
def cost_effective_completion(message, budget_limit=0.001):
"""根据预算选择模型"""
# 模型优先级（从便宜到贵）
models = [
"ollama/qwen:7b", # 本地免费
"gpt-3.5-turbo", # 便宜
"claude-3-haiku", # 中等
"gpt-4", # 昂贵
]
for model in models:
try:
response = completion(
model=model,
messages=[{"role": "user", "content": message}]
)
cost = completion_cost(response)
if cost <= budget_limit:
return {
"content": response.choices[0].message.content,
"model": model,
"cost": cost
}
except Exception as e:
continue
return {"error": "没有符合预算的可用模型"}
result = cost_effective_completion("介绍Python")
print(f"使用模型: {result.get('model')}")
print(f"成本: ${result.get('cost', 0):.6f}")
print(f"响应: {result.get('content')}")
print("✓ 成本优化完成")
---
1.2 统一接口
01.OpenAI兼容
a.标准格式
a.功能说明
所有模型使用OpenAI标准格式调用。支持messages、temperature、max_tokens等参数。兼容OpenAI SDK和工具。标准化降低学习成本。
b.代码示例
---
# 1. 标准OpenAI格式
from litellm import completion
# 使用OpenAI标准格式
response = completion(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "你是一个AI助手"},
{"role": "user", "content": "介绍Python"}
],
temperature=0.7,
max_tokens=1000,
top_p=1.0,
frequency_penalty=0.0,
presence_penalty=0.0
)
print(response.choices[0].message.content)
# 2. 其他模型使用相同格式
# Claude使用OpenAI格式
response = completion(
model="claude-3-sonnet-20240229",
messages=[
{"role": "system", "content": "你是一个AI助手"},
{"role": "user", "content": "介绍Python"}
],
temperature=0.7,
max_tokens=1000
)
# Ollama使用OpenAI格式
response = completion(
model="ollama/qwen:7b",
messages=[
{"role": "system", "content": "你是一个AI助手"},
{"role": "user", "content": "介绍Python"}
],
temperature=0.7
)
print("✓ 统一接口完成")
---
1.3 使用场景
01.应用场景
a.多模型对比
a.功能说明
快速对比不同模型的输出质量。评估模型性能和成本。选择最适合的模型。多模型对比优化AI应用。
b.代码示例
---
# 1. 模型A/B测试
from litellm import completion
import time
def compare_models(prompt, models):
"""对比多个模型的响应"""
results = []
for model in models:
start = time.time()
try:
response = completion(
model=model,
messages=[{"role": "user", "content": prompt}]
)
duration = time.time() - start
results.append({
"model": model,
"response": response.choices[0].message.content,
"tokens": response.usage.total_tokens,
"duration": duration
})
except Exception as e:
results.append({
"model": model,
"error": str(e)
})
return results
# 对比三个模型
models = ["gpt-3.5-turbo", "claude-3-haiku", "ollama/qwen:7b"]
results = compare_models("介绍AI的未来发展", models)
for r in results:
print(f"\n模型: {r['model']}")
if 'error' in r:
print(f"错误: {r['error']}")
else:
print(f"响应: {r['response'][:100]}...")
print(f"Tokens: {r['tokens']}, 耗时: {r['duration']:.2f}s")
print("✓ 使用场景完成")
---
2 快速开始
2.1 安装配置
01.安装部署
a.pip安装
a.功能说明
使用pip快速安装LiteLLM。支持Python 3.8+环境。安装简单无复杂依赖。快速开始AI开发。
b.代码示例
---
# 1. 基础安装
# 安装LiteLLM
pip install litellm
# 验证安装
python -c "import litellm; print(litellm.__version__)"
# 2. 完整安装（包含所有providers）
pip install 'litellm[proxy]'
# 3. 环境变量配置
# .env文件
OPENAI_API_KEY=your_openai_key
ANTHROPIC_API_KEY=your_anthropic_key
COHERE_API_KEY=your_cohere_key
# 加载环境变量
from dotenv import load_dotenv
load_dotenv()
import os
import litellm
litellm.openai_key = os.getenv("OPENAI_API_KEY")
litellm.anthropic_key = os.getenv("ANTHROPIC_API_KEY")
print("✓ 安装配置完成")
---
2.2 基础调用
01.简单调用
a.completion函数
a.功能说明
使用completion函数调用任意LLM。支持同步和异步调用。返回标准化响应。简单易用的API。
b.代码示例
---
# 1. 同步调用
from litellm import completion
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
# 2. 异步调用
from litellm import acompletion
import asyncio
async def async_chat():
response = await acompletion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
return response.choices[0].message.content
result = asyncio.run(async_chat())
print(result)
# 3. 多轮对话
messages = [
{"role": "system", "content": "你是AI助手"},
{"role": "user", "content": "我叫张三"},
{"role": "assistant", "content": "你好张三！"},
{"role": "user", "content": "我叫什么名字？"}
]
response = completion(
model="gpt-3.5-turbo",
messages=messages
)
print(response.choices[0].message.content)
print("✓ 基础调用完成")
---
2.3 流式输出
01.流式响应
a.streaming参数
a.功能说明
启用streaming实现实时输出。逐token返回生成内容。提升用户体验。流式输出适合长文本生成。
b.代码示例
---
# 1. 流式调用
from litellm import completion
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "讲个故事"}],
stream=True
)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end='', flush=True)
print("\n✓ 流式输出完成")
---
3 模型支持
3.1 主流LLM
01.OpenAI模型
a.GPT系列
a.功能说明
支持GPT-3.5和GPT-4全系列模型。配置简单直接调用。性能稳定响应快速。OpenAI是主流选择。
b.代码示例
---
# OpenAI模型调用
from litellm import completion
# GPT-3.5
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "介绍AI"}]
)
# GPT-4
response = completion(
model="gpt-4",
messages=[{"role": "user", "content": "介绍AI"}]
)
# Claude
response = completion(
model="claude-3-sonnet-20240229",
messages=[{"role": "user", "content": "介绍AI"}]
)
print("✓ 主流LLM完成")
---
3.2 国产大模型
01.国产模型支持
a.通义千问
a.功能说明
支持阿里通义千问系列模型。适配国内网络环境。满足数据合规要求。国产模型是信创首选。
b.代码示例
---
# 国产大模型调用
from litellm import completion
# 通义千问（通过DashScope）
response = completion(
model="dashscope/qwen-turbo",
messages=[{"role": "user", "content": "你好"}],
api_key=os.getenv("DASHSCOPE_API_KEY")
)
# 文心一言
response = completion(
model="wenxin/ernie-bot",
messages=[{"role": "user", "content": "你好"}]
)
# 智谱ChatGLM
response = completion(
model="zhipu/chatglm3-6b",
messages=[{"role": "user", "content": "你好"}]
)
print("✓ 国产大模型完成")
---
3.3 本地模型
01.Ollama集成
a.本地部署
a.功能说明
通过Ollama调用本地模型。无需API Key离线运行。数据完全本地化。适合隐私敏感场景。
b.代码示例
---
# 本地模型调用
from litellm import completion
# Ollama本地模型
response = completion(
model="ollama/qwen:7b",
messages=[{"role": "user", "content": "你好"}],
api_base="http://localhost:11434"
)
print(response.choices[0].message.content)
# 批量调用本地模型
models = ["ollama/qwen:7b", "ollama/llama2", "ollama/mistral"]
for model in models:
try:
response = completion(
model=model,
messages=[{"role": "user", "content": "你好"}]
)
print(f"{model}: {response.choices[0].message.content}")
except Exception as e:
print(f"{model}: 调用失败")
print("✓ 本地模型完成")
---
4 高级特性
4.1 负载均衡
01.自动路由
a.Router配置
a.功能说明
Router实现多模型负载均衡。自动分发请求到可用模型。支持权重和优先级配置。负载均衡提升可用性。
b.代码示例
---
# 负载均衡配置
from litellm import Router
# 创建路由器
router = Router(
model_list=[
{
"model_name": "gpt-3.5",
"litellm_params": {
"model": "gpt-3.5-turbo",
"api_key": os.getenv("OPENAI_API_KEY")
}
},
{
"model_name": "claude",
"litellm_params": {
"model": "claude-3-sonnet-20240229",
"api_key": os.getenv("ANTHROPIC_API_KEY")
}
},
{
"model_name": "local",
"litellm_params": {
"model": "ollama/qwen:7b"
}
}
]
)
# 使用路由器
response = router.completion(
model="gpt-3.5",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
print("✓ 负载均衡完成")
---
4.2 失败重试
01.重试机制
a.自动重试
a.功能说明
自动重试失败的请求。配置重试次数和延迟。支持fallback备用模型。重试机制提升稳定性。
b.代码示例
---
# 重试配置
from litellm import Router
router = Router(
model_list=[...],
num_retries=3,
timeout=60,
fallbacks=[
{"gpt-4": ["gpt-3.5-turbo"]},
{"claude-3-opus": ["claude-3-sonnet"]}
]
)
# 自动重试和fallback
response = router.completion(
model="gpt-4",
messages=[{"role": "user", "content": "你好"}]
)
print("✓ 失败重试完成")
---
4.3 缓存机制
01.响应缓存
a.Redis缓存
a.功能说明
使用Redis缓存相同请求的响应。减少API调用节省成本。支持TTL过期时间。缓存提升性能降低延迟。
b.代码示例
---
# Redis缓存
from litellm import completion
from litellm.caching import Cache
# 启用Redis缓存
litellm.cache = Cache(
type="redis",
host="localhost",
port=6379
)
# 第一次调用（无缓存）
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "什么是AI？"}],
caching=True
)
# 第二次调用（命中缓存）
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "什么是AI？"}],
caching=True
)
print("✓ 缓存机制完成")
---
4.4 成本追踪
01.成本监控
a.实时统计
a.功能说明
实时追踪API调用成本。按模型和用户统计费用。生成成本报表。成本追踪优化预算管理。
b.代码示例
---
# 成本追踪
from litellm import completion, completion_cost
import litellm
# 成本回调
costs = []
def track_cost(kwargs, completion_response, start_time, end_time):
cost = completion_cost(completion_response)
costs.append({
"model": kwargs["model"],
"cost": cost,
"tokens": completion_response.usage.total_tokens
})
litellm.success_callback = [track_cost]
# 调用模型
response = completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "你好"}]
)
# 查看成本
print(f"总成本: ${sum(c['cost'] for c in costs):.6f}")
print("✓ 成本追踪完成")
---
5 Proxy代理服务
5.1 代理部署
01.Proxy服务
a.启动代理
a.功能说明
LiteLLM Proxy提供统一API网关。管理多个模型和API密钥。实现访问控制和监控。Proxy简化企业级部署。
b.代码示例
---
# 1. 配置文件 config.yaml
model_list:
- model_name: gpt-3.5
litellm_params:
model: gpt-3.5-turbo
api_key: os.environ/OPENAI_API_KEY
- model_name: claude
litellm_params:
model: claude-3-sonnet-20240229
api_key: os.environ/ANTHROPIC_API_KEY
- model_name: local
litellm_params:
model: ollama/qwen:7b
api_base: http://localhost:11434
# 2. 启动Proxy服务
# 命令行启动
litellm --config config.yaml --port 8000
# 或使用Docker
docker run -p 8000:8000 -v $(pwd)/config.yaml:/app/config.yaml ghcr.io/berriai/litellm:latest
# 3. 客户端调用
import openai
client = openai.OpenAI(
api_key="sk-1234", # LiteLLM Proxy密钥
base_url="http://localhost:8000"
)
response = client.chat.completions.create(
model="gpt-3.5",
messages=[{"role": "user", "content": "你好"}]
)
print(response.choices[0].message.content)
print("✓ 代理部署完成")
---
5.2 API密钥管理
01.密钥配置
a.多密钥管理
a.功能说明
为不同用户和团队配置独立API密钥。实现密钥轮换和过期管理。追踪密钥使用情况。密钥管理保障安全。
b.代码示例
---
# config.yaml密钥配置
general_settings:
master_key: sk-1234 # 主密钥
database_url: postgresql://user:pass@localhost/litellm
litellm_settings:
drop_params: true
set_verbose: false
router_settings:
enable_pre_call_checks: true
# 虚拟密钥配置
keys:
- key: sk-team-a
models: ["gpt-3.5", "claude"]
max_budget: 100.0
budget_duration: "1mo"
- key: sk-team-b
models: ["local"]
max_budget: 0
print("✓ API密钥管理完成")
---
5.3 访问控制
01.权限管理
a.RBAC控制
a.功能说明
基于角色的访问控制。限制用户可用的模型和功能。设置请求速率限制。访问控制保障合规。
b.代码示例
---
# 访问控制配置
keys:
- key: sk-admin
permissions:
- "create_key"
- "delete_key"
- "view_spend"
models: ["*"]
- key: sk-developer
permissions:
- "chat_completion"
models: ["gpt-3.5", "local"]
tpm_limit: 10000
rpm_limit: 100
- key: sk-readonly
permissions:
- "view_spend"
models: []
print("✓ 访问控制完成")
---
5.4 监控告警
01.系统监控
a.指标收集
a.功能说明
收集请求数、延迟、成本等指标。集成Prometheus和Grafana。配置告警规则。监控告警保障稳定。
b.代码示例
---
# Prometheus监控配置
general_settings:
callbacks: ["prometheus"]
prometheus_settings:
port: 9090
# Grafana仪表盘查询
# 请求总数
sum(litellm_requests_total)
# 平均延迟
avg(litellm_request_duration_seconds)
# 成本统计
sum(litellm_spend_total) by (model)
# 告警规则
# 高延迟告警
avg(litellm_request_duration_seconds) > 5
# 高成本告警
sum(litellm_spend_total) > 100
print("✓ 监控告警完成")
---
6 集成应用
6.1 LangChain集成
01.LangChain适配
a.作为LLM Provider
a.功能说明
LiteLLM作为LangChain的LLM提供者。统一管理多个模型。简化模型切换。集成提升开发效率。
b.代码示例
---
# LangChain集成
from langchain.chat_models import ChatLiteLLM
from langchain.schema import HumanMessage
# 使用LiteLLM作为LangChain模型
chat = ChatLiteLLM(model="gpt-3.5-turbo")
messages = [
HumanMessage(content="介绍Python编程语言")
]
response = chat(messages)
print(response.content)
# 切换模型无需修改代码
chat = ChatLiteLLM(model="claude-3-sonnet-20240229")
response = chat(messages)
print(response.content)
# 本地模型
chat = ChatLiteLLM(model="ollama/qwen:7b")
response = chat(messages)
print(response.content)
print("✓ LangChain集成完成")
---
6.2 自定义集成
01.SDK集成
a.Python SDK
a.功能说明
在Python项目中集成LiteLLM。封装为统一服务层。支持依赖注入和配置管理。SDK集成简化开发。
b.代码示例
---
# 自定义LLM服务类
from litellm import completion
from typing import List, Dict
class LLMService:
"""统一LLM服务"""
def __init__(self, default_model: str = "gpt-3.5-turbo"):
self.default_model = default_model
def chat(self, message: str, model: str = None) -> str:
"""单轮对话"""
model = model or self.default_model
response = completion(
model=model,
messages=[{"role": "user", "content": message}]
)
return response.choices[0].message.content
def chat_with_history(
self,
messages: List[Dict],
model: str = None
) -> str:
"""多轮对话"""
model = model or self.default_model
response = completion(
model=model,
messages=messages
)
return response.choices[0].message.content
# 使用服务
llm = LLMService()
result = llm.chat("你好")
print(result)
result = llm.chat("你好", model="claude-3-sonnet-20240229")
print(result)
print("✓ 自定义集成完成")
---
6.3 企业级部署
01.生产环境
a.高可用部署
a.功能说明
Kubernetes集群部署LiteLLM Proxy。配置负载均衡和自动扩缩容。实现多区域容灾。企业级部署保障稳定。
b.代码示例
---
# Kubernetes部署配置
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: litellm-proxy
spec:
replicas: 3
selector:
matchLabels:
app: litellm
template:
metadata:
labels:
app: litellm
spec:
containers:
- name: litellm
image: ghcr.io/berriai/litellm:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: litellm-secrets
key: openai-key
volumeMounts:
- name: config
mountPath: /app/config.yaml
subPath: config.yaml
volumes:
- name: config
configMap:
name: litellm-config
---
# service.yaml
apiVersion: v1
kind: Service
metadata:
name: litellm-service
spec:
selector:
app: litellm
ports:
- port: 80
targetPort: 8000
type: LoadBalancer
print("✓ 企业级部署完成")
---
7 最佳实践
7.1 模型选型
01.选择策略
a.场景匹配
a.功能说明
根据应用场景选择合适模型。平衡性能、成本和延迟。考虑数据合规要求。科学选型优化体验。
b.代码示例
---
# 模型选型矩阵
model_matrix = {
"简单问答": {
"推荐": ["gpt-3.5-turbo", "claude-3-haiku"],
"成本": "低",
"延迟": "快"
},
"复杂推理": {
"推荐": ["gpt-4", "claude-3-opus"],
"成本": "高",
"延迟": "慢"
},
"代码生成": {
"推荐": ["gpt-4", "claude-3-sonnet"],
"成本": "中",
"延迟": "中"
},
"本地部署": {
"推荐": ["ollama/qwen:7b", "ollama/llama2"],
"成本": "无",
"延迟": "取决于硬件"
},
"信创环境": {
"推荐": ["dashscope/qwen-turbo", "ollama/qwen:7b"],
"成本": "低",
"延迟": "快"
}
}
# 根据场景选择模型
def select_model(scenario: str) -> str:
if scenario in model_matrix:
return model_matrix[scenario]["推荐"][0]
return "gpt-3.5-turbo"
model = select_model("简单问答")
print(f"推荐模型: {model}")
print("✓ 模型选型完成")
---
7.2 性能优化
01.优化技巧
a.并发处理
a.功能说明
使用异步并发提升吞吐量。配置连接池复用连接。启用缓存减少重复调用。性能优化提升响应速度。
b.代码示例
---
# 并发调用优化
from litellm import acompletion
import asyncio
async def batch_process(messages: list):
"""批量并发处理"""
tasks = [
acompletion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": msg}]
)
for msg in messages
]
results = await asyncio.gather(*tasks)
return [r.choices[0].message.content for r in results]
# 使用
messages = ["问题1", "问题2", "问题3", "问题4", "问题5"]
results = asyncio.run(batch_process(messages))
for i, result in enumerate(results, 1):
print(f"结果{i}: {result}")
print("✓ 性能优化完成")
---
7.3 成本控制
01.成本管理
a.预算控制
a.功能说明
设置团队和项目预算上限。监控实时成本消耗。自动限流和告警。成本控制避免超支。
b.代码示例
---
# 成本控制策略
from litellm import completion, completion_cost
class CostController:
"""成本控制器"""
def __init__(self, daily_budget: float):
self.daily_budget = daily_budget
self.current_cost = 0.0
def check_budget(self) -> bool:
"""检查预算"""
return self.current_cost < self.daily_budget
def chat(self, message: str, model: str = "gpt-3.5-turbo"):
"""带预算控制的对话"""
if not self.check_budget():
return "今日预算已用完，请明天再试"
response = completion(
model=model,
messages=[{"role": "user", "content": message}]
)
# 更新成本
cost = completion_cost(response)
self.current_cost += cost
return {
"response": response.choices[0].message.content,
"cost": cost,
"remaining": self.daily_budget - self.current_cost
}
# 使用
controller = CostController(daily_budget=1.0)
result = controller.chat("你好")
print(f"响应: {result['response']}")
print(f"本次成本: ${result['cost']:.6f}")
print(f"剩余预算: ${result['remaining']:.6f}")
print("✓ 成本控制完成")
---