Claude API 完全实战——从第一个请求到生产级集成

第13章：Claude API 完全实战——从第一个请求到生产级集成

本章学习目标：5分钟内跑通第一个 API 请求；掌握 Messages API 每个参数的实际影响；用 Python 和 TypeScript 实现流式输出；理解 Tool Use 工具调用的完整循环；用 Prompt Caching 把重复请求成本降低 90%；写出带指数退避的生产级错误处理。

快速入门：5分钟跑通第一个请求

pip install anthropic
export ANTHROPIC_API_KEY="sk-ant-api03-..."
# 或者放在项目 .env 文件，确保 .env 已加入 .gitignore

import anthropic

client = anthropic.Anthropic()  # 自动读取 ANTHROPIC_API_KEY 环境变量

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system="你是一个代码审查专家，只用中文回复，回复简洁专业。",
    messages=[
        {"role": "user", "content": "帮我审查这段代码：\n```python\ndef divide(a, b):\n    return a / b\n```"}
    ]
)

print(message.content[0].text)
print(f"消耗 tokens: 输入 {message.usage.input_tokens}, 输出 {message.usage.output_tokens}")

运行后你会看到 Claude 指出 b=0 时会抛出 ZeroDivisionError，并给出修复建议。这就是一次完整的 API 调用。

Messages API 参数完全解析

每个参数都有实际影响，不设置不等于"无所谓"：

temperature 的实际影响（重要）：

• 代码生成：用 0，要确定性结果，同样的 prompt 每次生成相同代码
• 分析/解释：用 0.3，稳定但允许一点措辞变化
• 创意写作：用 0.7–1.0，每次生成不同的内容
• 永远不要在代码生成任务上用高 temperature，输出会不稳定甚至错误

流式输出（Streaming）

对于需要生成长文本的场景，流式输出让用户看到实时生成过程，感知速度从"等待5秒"变成"立刻看到内容"：

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "写一个完整的 FastAPI CRUD 接口"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)  # 实时打印，不换行

# 流结束后获取完整用量信息
final = stream.get_final_message()
print(f"\n输入 {final.usage.input_tokens} tokens，输出 {final.usage.output_tokens} tokens")

TypeScript / Next.js Route Handler 的流式实现：

// src/app/api/generate/route.ts
import Anthropic from "@anthropic-ai/sdk";
import { NextRequest } from "next/server";

const client = new Anthropic();

export async function POST(req: NextRequest) {
  const { message } = await req.json();

  const stream = await client.messages.stream({
    model: "claude-sonnet-4-6",
    max_tokens: 2048,
    messages: [{ role: "user", content: message }],
  });

  const readableStream = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        if (
          chunk.type === "content_block_delta" &&
          chunk.delta.type === "text_delta"
        ) {
          controller.enqueue(new TextEncoder().encode(chunk.delta.text));
        }
      }
      controller.close();
    },
  });

  return new Response(readableStream, {
    headers: { "Content-Type": "text/plain; charset=utf-8" },
  });
}

Tool Use（函数调用）完整示例

Tool Use 让 Claude 能调用你定义的函数，是构建 Agent 的基础。工作流程：你定义工具描述 → Claude 决定调用哪个工具 → 你执行并返回结果 → Claude 基于结果给出最终回复。

import anthropic
import json

client = anthropic.Anthropic()

# 定义工具——本质是 JSON Schema 描述函数签名
tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名称"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    },
    {
        "name": "query_database",
        "description": "查询数据库中的订单信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"},
                "status": {"type": "string", "enum": ["pending", "completed", "cancelled"]}
            }
        }
    }
]

def handle_tool_call(tool_name: str, tool_input: dict) -> dict:
    """实际执行工具的函数——这里调用真实的 API 或数据库"""
    if tool_name == "get_weather":
        return {"temperature": 25, "condition": "晴天", "city": tool_input["city"]}
    elif tool_name == "query_database":
        return {"order_id": tool_input["order_id"], "status": "completed", "amount": 199.00}
    return {}

def chat_with_tools(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "tool_use":
            # Claude 要调用工具，处理所有 tool_use 块
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = handle_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False)
                    })

            # 把工具结果发回给 Claude，继续对话
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            # stop_reason == "end_turn"，Claude 给出最终回复
            return response.content[0].text

result = chat_with_tools("北京今天天气怎么样？另外帮我查一下订单 ORD-12345 的状态")
print(result)

注意 Claude 可能同时调用多个工具： 一次响应中可能有多个 tool_use 块。上面的代码用循环处理所有块，这是正确做法。如果只处理第一个，在多工具场景下会出错。

Prompt Caching（成本优化关键）

使用场景： 每次请求都包含相同的大段内容——系统提示、参考文档、代码库上下文。不加 cache_control 的情况下，这些内容每次都按正常价格计费。加了之后，缓存命中时价格降低约 90%。

import anthropic

client = anthropic.Anthropic()

# 假设有一个 5000 token 的大型代码库上下文
long_codebase_context = "...（项目所有文件的内容，几千行）..."

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是代码审查专家，审查时只指出真实的 bug，不要泛泛而谈。",
            "cache_control": {"type": "ephemeral"}  # 标记为可缓存
        },
        {
            "type": "text",
            "text": long_codebase_context,
            "cache_control": {"type": "ephemeral"}  # 这段也缓存
        }
    ],
    messages=[{"role": "user", "content": "审查最新提交的 auth.py 文件"}]
)

# 查看缓存命中情况
usage = response.usage
print(f"缓存创建 tokens: {usage.cache_creation_input_tokens}")  # 首次：正常价格
print(f"缓存读取 tokens: {usage.cache_read_input_tokens}")      # 命中：约降 90%

实际节省计算： 如果 system prompt 有 5000 tokens，每次请求节省约 $0.075（Sonnet 价格 $3/MTok × 0.9 折扣）。一天 100 次请求节省 $7.5，一个月节省 $225。缓存有效期约 5 分钟，高频调用场景下收益显著。

生产级错误处理与重试

直接调用 API 而不做重试，在生产环境会遇到不必要的失败。Rate Limit 错误是最常见的——用指数退避重试几乎总能解决。

import anthropic
import time
from anthropic import APIStatusError, APIConnectionError, RateLimitError

client = anthropic.Anthropic()

def call_claude_with_retry(messages: list, max_retries: int = 3) -> str:
    """带指数退避重试的生产级 API 调用"""
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=messages
            )
            return response.content[0].text

        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 指数退避：1s, 2s, 4s
                print(f"Rate limit，等待 {wait_time}s 后重试（第 {attempt + 1} 次）")
                time.sleep(wait_time)
            else:
                raise  # 重试耗尽，向上抛出

        except APIConnectionError:
            # 网络问题，短暂重试
            if attempt < max_retries - 1:
                time.sleep(1)
            else:
                raise

        except APIStatusError as e:
            if e.status_code >= 500:
                # 5xx 服务端错误，可重试
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)
                    continue
            # 4xx 客户端错误（prompt 太长、参数错误等），不重试，直接报错
            raise

    raise RuntimeError(f"重试 {max_retries} 次后仍失败")

模型选择与成本参考（2025年）

实际选型原则： 绝大多数场景用 Sonnet 就够了。只有当 Sonnet 明确给出错误答案时才升级到 Opus。Haiku 用于高频调用（每秒数十次）或对成本极敏感的批量任务——同等质量的简单任务，Haiku 比 Sonnet 便宜约 15 倍。

本章要点

1. max_tokens 和 model 是必填参数，temperature 默认 1.0 对代码任务不友好，代码生成设为 0。
2. 流式输出的关键是监听 content_block_delta 事件中的 text_delta，Python 用 .stream() 上下文管理器，TypeScript 用 for await 循环。
3. Tool Use 必须处理循环：Claude 可能多轮调用工具，直到 stop_reason 为 end_turn 才是最终回复；一次响应中可能有多个 tool_use 块。
4. Prompt Caching 只对重复的大段内容有效：cache_control 加在每次请求都相同的部分，变化的用户消息不加；缓存约 5 分钟有效期，低频场景收益有限。
5. 重试逻辑区分错误类型：RateLimitError 和 5xx 用指数退避重试，4xx 客户端错误不重试（重试也不会成功）。