生产级 API 代码片段
经过实战检验的代码骨架,用于在 API 上构建应用。请根据你的技术栈进行调整和加固。
一个有韧性的调用封装(重试 + 退避)
- Python
- TypeScript
import time, random, anthropic
client = anthropic.Anthropic()
def ask(messages, model="claude-sonnet-4-6", max_tokens=1024, system=None):
for attempt in range(5):
try:
return client.messages.create(
model=model, max_tokens=max_tokens,
system=system or anthropic.NOT_GIVEN, messages=messages,
)
except (anthropic.RateLimitError, anthropic.APIStatusError) as e:
if attempt == 4:
raise
time.sleep(min(2 ** attempt + random.random(), 30))
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
export async function ask(messages, { model = "claude-sonnet-4-6", maxTokens = 1024, system } = {}) {
for (let attempt = 0; attempt < 5; attempt++) {
try {
return await client.messages.create({ model, max_tokens: maxTokens, system, messages });
} catch (e: any) {
if (attempt === 4 || ![429, 500, 529].includes(e?.status)) throw e;
await new Promise((r) => setTimeout(r, Math.min(2 ** attempt * 1000 + Math.random() * 1000, 30000)));
}
}
}
(SDK 默认也会对瞬时错误进行重试 —— 在叠加自己的重试逻辑之前,先了解你所用客户端的行为。参见错误与速率限制。)
流式聊天
with client.messages.stream(model="claude-sonnet-4-6", max_tokens=1024,
messages=[{"role": "user", "content": "Hello"}]) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
工具使用循环(骨架)
messages = [{"role": "user", "content": "What's the weather in Rome?"}]
while True:
resp = client.messages.create(model="claude-sonnet-4-6", max_tokens=1024,
tools=TOOLS, messages=messages)
if resp.stop_reason != "tool_use":
break
messages.append({"role": "assistant", "content": resp.content})
results = [run_tool(b) for b in resp.content if b.type == "tool_use"]
messages.append({"role": "user", "content": results}) # tool_result blocks
:::warning 加固 为迭代次数和成本设上限,校验工具输入,并将密钥保存在环境变量中。对于自主运行的场景,请阅读保护智能体安全。 :::