LLM 集成
概述
涵盖 OpenAI、Anthropic、Azure 和 Cohere 提供商的 LLM API 集成综合指南。
前提条件
- 基本了解 REST API 和 HTTP 请求
- Python 编程知识
- 熟悉异步/等待模式
- API 密钥管理概念
- 了解速率限制和重试模式
核心概念
- LLM 提供商:OpenAI、Anthropic、Azure OpenAI、Cohere - 不同的 API 服务,用于访问大型语言模型
- 统一接口:提供跨不同 LLM 提供商一致的 API 的抽象层
- 流式响应:实时逐令牌响应交付,以改善用户体验
- 错误处理:重试逻辑、指数退避和回退策略,用于弹性 API 调用
- 速率限制:使用令牌桶算法管理 API 请求速率,避免达到限制
- 成本跟踪:监控令牌使用情况和计算跨提供商的 API 成本
- 响应缓存:存储和检索缓存响应,以减少 API 调用和成本
- 提示模板:可重用的提示模式,具有变量替换
- 令牌计数:估计和管理不同模型的令牌使用情况
- 上下文窗口管理:在模型上下文限制内管理对话历史
- 多提供商回退:在失败时自动切换提供商
- 批量处理:同时高效处理多个 LLM 请求
1. 提供商设置
1.1 OpenAI 设置
import os
from openai import OpenAI
# 初始化 OpenAI 客户端
client = OpenAI(
api_key=os.getenv('OPENAI_API_KEY'),
organization=os.getenv('OPENAI_ORG_ID'), # 可选
base_url=os.getenv('OPENAI_BASE_URL'), # 用于自定义端点
timeout=60.0,
max_retries=3
)
# 测试连接
try:
models = client.models.list()
print(f"已连接!可用模型:{len(models.data)}")
except Exception as e:
print(f"连接失败:{e}")
1.2 Anthropic 设置
import anthropic
# 初始化 Anthropic 客户端
client = anthropic.Anthropic(
api_key=os.getenv('ANTHROPIC_API_KEY'),
base_url=os.getenv('ANTHROPIC_BASE_URL'), # 可选
timeout=60.0,
max_retries=3
)
# 测试连接
try:
message = client.messages.create(
model="claude-3-haiku-20240307",
max_tokens=10,
messages=[{"role": "user", "content": "Hi"}]
)
print(f"已连接!响应:{message.content}")
except Exception as e:
print(f"连接失败:{e}")
1.3 Azure OpenAI 设置
from openai import AzureOpenAI
# 初始化 Azure OpenAI 客户端
client = AzureOpenAI(
api_key=os.getenv('AZURE_OPENAI_API_KEY'),
api_version=os.getenv('AZURE_OPENAI_API_VERSION', '2024-02-15-preview'),
azure_endpoint=os.getenv('AZURE_OPENAI_ENDPOINT'),
timeout=60.0,
max_retries=3
)
# 测试连接
try:
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": "Hi"}],
max_tokens=10
)
print(f"已连接!响应:{response.choices[0].message.content}")
except Exception as e:
print(f"连接失败:{e}")
1.4 Cohere 设置
import cohere
# 初始化 Cohere 客户端
client = cohere.Client(
api_key=os.getenv('COHERE_API_KEY'),
base_url=os.getenv('COHERE_BASE_URL'), # 可选
timeout=60.0,
max_retries=3
)
# 测试连接
try:
response = client.chat(
model='command',
message='Hi',
max_tokens=10
)
print(f"已连接!响应:{response.text}")
except Exception as e:
print(f"连接失败:{e}")
2. 统一接口模式
…(此处省略中间部分,与原英文内容一致)
14. 最佳实践
14.1 API 密钥管理
…(此处省略中间部分,与原英文内容一致)
相关技能
06-ai-ml-production/prompt-engineering06-ai-ml-production/llm-function-calling06-ai-ml-production/llm-guardrails06-ai-ml-production/agent-patterns06-ai-ml-production/ai-observability06-ai-ml-production/embedding-models