全部笔记All notes

大模型 SDK 使用指南

阅读 7m 59s7m 59s read

各厂商官方 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 & I

SDK 选择建议:

场景推荐 SDK
只用 OpenAIOpenAI SDK
只用 ClaudeAnthropic SDK
只用 GeminiGoogle 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

初始化对比

厂商环境变量初始化代码
OpenAIOPENAI_API_KEYclient = OpenAI()
AnthropicANTHROPIC_API_KEYclient = Anthropic()
GoogleGOOGLE_API_KEYgenai.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

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 秒
用户体验长时间空白等待实时看到内容
超时风险高(长回复可能超时)低(持续有数据)
可中断性无法中断可随时停止

代码对比

厂商流式参数迭代方式
OpenAIstream=Truefor chunk in resp
Anthropicclient.messages.stream()with stream: for text in stream.text_stream
Googlestream=Truefor 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 功能对比

功能OpenAIAnthropicGoogle
同步调用✅✅✅
异步调用✅✅✅
流式输出✅✅✅
自动重试✅✅❌
类型提示✅✅部分
代理设置✅✅✅

代理配置

# 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}")

错误类型对照

错误类型OpenAIAnthropic说明
认证错误AuthenticationErrorAuthenticationErrorAPI Key 无效
速率限制RateLimitErrorRateLimitError请求过于频繁
连接错误APIConnectionErrorAPIConnectionError网络问题
超时APITimeoutErrorAPITimeoutError请求超时
服务器错误InternalServerErrorInternalServerError服务端问题

最佳实践

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(...)

相关文档