各厂商官方 SDK 的安装、初始化和核心用法对比。
为什么使用 SDK?
直接调用 HTTP API 虽然可行,但 SDK 提供了更多便利:
- 类型安全:完整的类型提示,IDE 自动补全
- 自动重试:内置指数退避重试机制
- 错误处理:结构化的异常类型
- 流式支持:简化 SSE 解析逻辑
- 认证管理:自动处理 Header 和 Token
SDK vs 直接 HTTP 调用对比:
| 方面 | 直接 HTTP | 使用 SDK |
|---|---|---|
| 代码量 | 多(需要处理请求、响应、错误) | 少(封装好了) |
| 类型提示 | 无 | 完整 |
| 错误处理 | 需要自己解析 | 结构化异常 |
| 流式处理 | 需要手动解析 SSE | 自动处理 |
| 重试逻辑 | 需要自己实现 | 内置 |
| 维护成本 | 高(API 变化需要手动适配) | 低(SDK 更新即可) |
什么时候直接用 HTTP?
- 语言没有官方 SDK
- 需要极致的控制和定制
- 学习和理解 API 原理
- 特殊的网络环境或代理需求
SDK 概览
graph LR
A[你的应用] --> B{选择 SDK}
B --> C[OpenAI SDK]
B --> D[Anthropic SDK]
B --> E[Google GenAI SDK]
B --> F[LiteLLM 统一SDK]
C --> G[GPT 系列]
D --> H[Claude 系列]
E --> I[Gemini 系列]
F --> G & H & ISDK 选择建议:
| 场景 | 推荐 SDK |
|---|---|
| 只用 OpenAI | OpenAI SDK |
| 只用 Claude | Anthropic SDK |
| 只用 Gemini | Google GenAI SDK |
| 多模型切换 | LiteLLM |
| 需要统一接口 | LiteLLM |
安装
# OpenAI
pip install openai
# Anthropic
pip install anthropic
# Google
pip install google-generativeai
# 统一 SDK(推荐多模型场景)
pip install litellm
版本管理建议:
建议在 requirements.txt 或 pyproject.toml 中锁定主版本号,避免意外的破坏性更新:
openai>=1.0.0,<2.0.0
anthropic>=0.20.0,<1.0.0
google-generativeai>=0.5.0
初始化对比
| 厂商 | 环境变量 | 初始化代码 |
|---|---|---|
| OpenAI | OPENAI_API_KEY | client = OpenAI() |
| Anthropic | ANTHROPIC_API_KEY | client = Anthropic() |
GOOGLE_API_KEY | genai.configure(api_key=key) |
环境变量配置:
推荐使用环境变量管理 API Key,避免硬编码:
# .env 文件
OPENAI_API_KEY=sk-xxx
ANTHROPIC_API_KEY=sk-ant-xxx
GOOGLE_API_KEY=AIza-xxx
# 使用 python-dotenv 加载
from dotenv import load_dotenv
load_dotenv()
# SDK 会自动读取环境变量
client = OpenAI() # 自动使用 OPENAI_API_KEY
显式传递 API Key:
# 如果不想用环境变量,可以显式传递
client = OpenAI(api_key="sk-xxx")
client = Anthropic(api_key="sk-ant-xxx")
基础调用
三家 SDK 的调用方式各有特点。OpenAI 和 Anthropic 采用客户端实例模式,Google 则使用模块级配置。
OpenAI
OpenAI SDK 设计最为简洁,client.chat.completions.create() 是最常用的方法。返回的 response 对象包含 choices 数组,通常取第一个结果。
from openai import OpenAI
client = OpenAI()
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hi"}]
)
print(resp.choices[0].message.content)
OpenAI SDK 特点:
- 方法链式调用:
client.chat.completions.create() - 返回 Pydantic 模型,支持
.model_dump()转字典 - 自动从环境变量读取 API Key
- 支持同步和异步两种客户端
Anthropic
Claude SDK 的特点是 max_tokens 为必需参数,这是与 OpenAI 最大的区别。响应结构也不同,内容在 content[0].text 中。
from anthropic import Anthropic
client = Anthropic()
resp = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024, # 必需
messages=[{"role": "user", "content": "Hi"}]
)
print(resp.content[0].text)
Anthropic SDK 特点:
max_tokens是必需参数(OpenAI 可选)system提示词是独立参数,不在 messages 中- 响应的
content是数组,支持多种内容类型 - 内置 Prompt Caching 支持
Google SDK 采用不同的设计理念,先创建模型实例,再调用生成方法。这种方式在多轮对话时更方便,可以直接使用 model.start_chat()。
import google.generativeai as genai
genai.configure(api_key="YOUR_KEY")
model = genai.GenerativeModel("gemini-1.5-flash")
resp = model.generate_content("Hi")
print(resp.text)
Google SDK 特点:
- 模块级配置,而非客户端实例
- 先创建模型对象,再调用方法
- 内置多轮对话支持(
start_chat()) - 支持多模态输入(图像、视频、音频)
流式输出
流式输出是提升用户体验的关键技术。用户无需等待完整响应,可以实时看到生成内容,特别适合聊天场景。
三家 SDK 都支持流式输出,但实现方式略有不同:
- OpenAI:在原方法上添加
stream=True参数 - Anthropic:使用专门的
stream()上下文管理器 - Google:同样使用
stream=True参数
sequenceDiagram
participant App
participant SDK
participant API
App->>SDK: stream=True
SDK->>API: POST /messages
loop 逐块返回
API-->>SDK: chunk
SDK-->>App: yield chunk
end流式输出的优势:
| 指标 | 非流式 | 流式 |
|---|---|---|
| 首字节时间 | 数秒~数十秒 | 通常 < 1 秒 |
| 用户体验 | 长时间空白等待 | 实时看到内容 |
| 超时风险 | 高(长回复可能超时) | 低(持续有数据) |
| 可中断性 | 无法中断 | 可随时停止 |
代码对比
| 厂商 | 流式参数 | 迭代方式 |
|---|---|---|
| OpenAI | stream=True | for chunk in resp |
| Anthropic | client.messages.stream() | with stream: for text in stream.text_stream |
stream=True | for chunk in resp |
异步调用
在高并发场景下,异步调用可以显著提升性能。当你需要同时发起多个 API 请求时,异步方式可以避免阻塞,充分利用等待时间。
OpenAI 和 Anthropic 都提供了专门的异步客户端类,使用方式与同步版本几乎一致,只需将客户端换成 Async 版本,并在调用时使用 await。
# OpenAI
from openai import AsyncOpenAI
client = AsyncOpenAI()
resp = await client.chat.completions.create(...)
# Anthropic
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
resp = await client.messages.create(...)
LiteLLM 统一调用
如果你的应用需要支持多个模型厂商,LiteLLM 是最佳选择。它提供了统一的 OpenAI 格式接口,内部自动处理不同厂商的格式转换。
主要优势:
- 统一接口:一套代码调用所有模型
- 自动转换:请求/响应格式自动适配
- 容灾切换:支持 fallback 到备用模型
- 成本追踪:内置 Token 计数和成本计算
from litellm import completion
# 同一接口,不同模型
resp = completion(model="gpt-4o", messages=[...])
resp = completion(model="claude-3-5-sonnet-20241022", messages=[...])
resp = completion(model="gemini/gemini-1.5-flash", messages=[...])
SDK 功能对比
| 功能 | OpenAI | Anthropic | |
|---|---|---|---|
| 同步调用 | ✅ | ✅ | ✅ |
| 异步调用 | ✅ | ✅ | ✅ |
| 流式输出 | ✅ | ✅ | ✅ |
| 自动重试 | ✅ | ✅ | ❌ |
| 类型提示 | ✅ | ✅ | 部分 |
| 代理设置 | ✅ | ✅ | ✅ |
代理配置
# OpenAI
client = OpenAI(base_url="https://your-proxy.com/v1")
# Anthropic
client = Anthropic(base_url="https://your-proxy.com")
# 环境变量方式
export OPENAI_BASE_URL="https://your-proxy.com/v1"
超时与连接配置
网络不稳定时,合理的超时配置可以避免请求长时间挂起。
超时设置
# OpenAI - 默认超时 10 分钟
client = OpenAI(timeout=60.0) # 60 秒
# Anthropic
client = Anthropic(timeout=60.0)
# 分别设置连接和读取超时
from openai import Timeout
client = OpenAI(timeout=Timeout(connect=5.0, read=60.0))
连接池配置
高并发场景下,配置连接池可以复用 TCP 连接,减少握手开销。
import httpx
# 自定义 HTTP 客户端
http_client = httpx.Client(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
client = OpenAI(http_client=http_client)
Node.js 示例
OpenAI
import OpenAI from 'openai';
const client = new OpenAI();
const resp = await client.chat.completions.create({
model: 'gpt-4o',
messages: [{ role: 'user', content: 'Hi' }]
});
console.log(resp.choices[0].message.content);
Anthropic
import Anthropic from '@anthropic-ai/sdk';
const client = new Anthropic();
const resp = await client.messages.create({
model: 'claude-sonnet-4-20250514',
max_tokens: 1024,
messages: [{ role: 'user', content: 'Hi' }]
});
console.log(resp.content[0].text);
流式输出
// OpenAI 流式
const stream = await client.chat.completions.create({
model: 'gpt-4o',
messages: [{ role: 'user', content: 'Hi' }],
stream: true
});
for await (const chunk of stream) {
process.stdout.write(chunk.choices[0]?.delta?.content || '');
}
错误处理
SDK 提供了结构化的异常类型,便于针对性处理不同错误。
OpenAI 错误处理
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
client = OpenAI()
try:
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hi"}]
)
except RateLimitError as e:
print(f"速率限制: {e}")
# 等待后重试
except APIConnectionError as e:
print(f"连接错误: {e}")
# 检查网络
except APIError as e:
print(f"API 错误: {e.status_code} - {e.message}")
Anthropic 错误处理
from anthropic import Anthropic, APIError, RateLimitError
client = Anthropic()
try:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
messages=[{"role": "user", "content": "Hi"}]
)
except RateLimitError as e:
print(f"速率限制: {e}")
except APIError as e:
print(f"API 错误: {e}")
错误类型对照
| 错误类型 | OpenAI | Anthropic | 说明 |
|---|---|---|---|
| 认证错误 | AuthenticationError | AuthenticationError | API Key 无效 |
| 速率限制 | RateLimitError | RateLimitError | 请求过于频繁 |
| 连接错误 | APIConnectionError | APIConnectionError | 网络问题 |
| 超时 | APITimeoutError | APITimeoutError | 请求超时 |
| 服务器错误 | InternalServerError | InternalServerError | 服务端问题 |
最佳实践
1. 使用环境变量
# ✅ 推荐:使用环境变量
client = OpenAI() # 自动读取 OPENAI_API_KEY
# ❌ 避免:硬编码 API Key
client = OpenAI(api_key="sk-xxx") # 不要这样做
2. 配置合理的超时
# 根据场景设置超时
client = OpenAI(
timeout=60.0, # 普通请求
# 或者分别设置
# timeout=Timeout(connect=5.0, read=120.0) # 长文本生成
)
3. 实现重试逻辑
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=60)
)
def call_api(messages):
return client.chat.completions.create(
model="gpt-4o",
messages=messages
)
4. 使用连接池
import httpx
# 高并发场景配置连接池
http_client = httpx.Client(
limits=httpx.Limits(
max_connections=100,
max_keepalive_connections=20
),
timeout=httpx.Timeout(connect=5.0, read=60.0)
)
client = OpenAI(http_client=http_client)
5. 日志记录
import logging
# 开启 SDK 日志
logging.basicConfig(level=logging.DEBUG)
# 或者只开启 HTTP 日志
import httpx
httpx_logger = logging.getLogger("httpx")
httpx_logger.setLevel(logging.DEBUG)
常见问题
Q: SDK 版本不兼容怎么办?
OpenAI SDK 在 1.0 版本有重大变化:
# 旧版本 (< 1.0)
import openai
openai.api_key = "sk-xxx"
response = openai.ChatCompletion.create(...)
# 新版本 (>= 1.0)
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(...)
Q: 如何切换不同的 API 端点?
# 使用代理或自定义端点
client = OpenAI(
base_url="https://your-proxy.com/v1",
api_key="your-key"
)
# 使用 Azure OpenAI
from openai import AzureOpenAI
client = AzureOpenAI(
azure_endpoint="https://your-resource.openai.azure.com",
api_key="your-key",
api_version="2024-02-01"
)
Q: 如何在 Jupyter Notebook 中使用异步?
import nest_asyncio
nest_asyncio.apply()
from openai import AsyncOpenAI
client = AsyncOpenAI()
# 现在可以在 Notebook 中使用 await
response = await client.chat.completions.create(...)