Python SDK 使用指南
ClawdRouter 兼容 OpenAI Python SDK 和 Anthropic Python SDK,你可以根据需要选择使用。
OpenAI SDK
安装
pip install openai
初始化客户端
from openai import OpenAI
client = OpenAI(
api_key="YOUR_API_KEY",
base_url="https://api.clawdrouter.com/v1",
)
环境变量
建议通过环境变量管理 API Key,避免在代码中硬编码:
export OPENAI_API_KEY="YOUR_API_KEY"
export OPENAI_BASE_URL="https://api.clawdrouter.com/v1"
# 会自动读取环境变量
client = OpenAI()
基础对话
completion = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个有用的助手"},
{"role": "user", "content": "用简单的话解释什么是机器学习"},
],
)
print(completion.choices[0].message.content)
流式输出
stream = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "写一篇关于人工智能的短文"},
],
stream=True,
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content is not None:
print(content, end="", flush=True)
print() # 换行
多轮对话
messages = [
{"role": "system", "content": "你是一个有用的助手"},
]
def chat(user_input):
messages.append({"role": "user", "content": user_input})
completion = client.chat.completions.create(
model="gpt-4o",
messages=messages,
)
reply = completion.choices[0].message.content
messages.append({"role": "assistant", "content": reply})
return reply
# 多轮对话
print(chat("什么是量子计算?"))
print(chat("它和传统计算有什么区别?"))
print(chat("目前有哪些实际应用?"))
异步调用
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key="YOUR_API_KEY",
base_url="https://api.clawdrouter.com/v1",
)
async def main():
completion = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "你好"},
],
)
print(completion.choices[0].message.content)
asyncio.run(main())
工具调用 (Function Calling)
import json
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如 北京",
},
},
"required": ["city"],
},
},
}
]
completion = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "北京今天天气怎么样?"},
],
tools=tools,
tool_choice="auto",
)
message = completion.choices[0].message
if message.tool_calls:
tool_call = message.tool_calls[0]
print(f"调用工具: {tool_call.function.name}")
print(f"参数: {tool_call.function.arguments}")
错误处理
from openai import (
APIError,
AuthenticationError,
RateLimitError,
)
try:
completion = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "你好"}],
)
except AuthenticationError:
print("API Key 无效,请检查配置")
except RateLimitError:
print("请求频率超限,请稍后重试")
except APIError as e:
print(f"API 错误: {e.message}")
Anthropic SDK
如果你需要使用 Anthropic 原生协议 (/v1/messages) 调用 Claude 模型,可以使用 Anthropic Python SDK。
安装
pip install anthropic
初始化客户端
import anthropic
client = anthropic.Anthropic(
api_key="YOUR_API_KEY",
base_url="https://api.clawdrouter.com",
)
环境变量
export ANTHROPIC_API_KEY="YOUR_API_KEY"
export ANTHROPIC_BASE_URL="https://api.clawdrouter.com"
client = anthropic.Anthropic()
基础对话
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "用简单的话解释什么是机器学习"},
],
)
print(message.content[0].text)
带系统提示词
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system="你是一个专业的技术文档撰写专家。",
messages=[
{"role": "user", "content": "帮我写一段 API 错误处理的最佳实践"},
],
)
print(message.content[0].text)
流式输出
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "写一篇关于人工智能的短文"},
],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
print()
异步调用
import asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic(
api_key="YOUR_API_KEY",
base_url="https://api.clawdrouter.com",
)
async def main():
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "你好"},
],
)
print(message.content[0].text)
asyncio.run(main())
错误处理
from anthropic import (
APIError,
AuthenticationError,
RateLimitError,
)
try:
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "你好"}],
)
except AuthenticationError:
print("API Key 无效,请检查配置")
except RateLimitError:
print("请求频率超限,请稍后重试")
except APIError as e:
print(f"API 错误: {e.message}")