用 Claude 与 OpenAI API 构建应用

今天运行的每一个 AI 编码工具、聊天机器人、文档分析器和自主 agent,都建立在同一基础之上:一个 HTTP API,接收一组 message、返回一个 completion。Claude 的 Messages API 与 OpenAI 的 Chat Completions API 在结构上几乎相同,精通其一就给了你一个对两者都通用的心智模型。但会基础调用只是起点,而非全部。生产应用需要处理 streaming 以获得感知上的响应速度、prompt caching 以控制成本、structured output 以可靠解析、指数退避以应对瞬时错误,以及严密的密钥管理以让 secret 保持 secret。本文全部覆盖。

我们会从基础 API 结构搭起,再逐层叠加每个生产关切。读完你将对“交付一个可靠、低成本、低延迟的 LLM 驱动 feature”所需的一切有完整图景——而不只是在终端里拿回一个响应。

Messages API:结构与角色

每次对 Messages API 的调用都是一组 message 对象,每个带一个 role 和 content。三种角色语义各异:

• system——在任何 user 回合之前发送一次。设定模型的人格、任务约束、输出格式,以及任何持久指令。模型把它视为比 user 消息更高的权威。这里放“You are a senior Python engineer. Always return valid JSON. Never include markdown code fences.”
• user——来自人类(或扮演人类的应用)的消息。对话的每一回合加一条 user 消息。
• assistant——模型之前的响应。构建多轮对话时,你把先前的 assistant 消息纳入列表,好让模型维持上下文。单轮调用则省略它们。

Anthropic 的 API 把 system 消息作为顶层参数处理,而非列表里的一条消息,但概念角色完全相同。OpenAI 把 system 作为 messages 数组的第一个对象。两者实践中行为一致。

import anthropic

client = anthropic.Anthropic()  # 从 env 读取 ANTHROPIC_API_KEY

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system="You are a concise technical writer. Reply in plain text, no markdown.",
    messages=[
        {"role": "user", "content": "Explain what a context window is in one paragraph."},
    ],
)

print(response.content[0].text)
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Output tokens: {response.usage.output_tokens}")

关键参数及其作用

temperature vs. top_p:多数应用用 temperature。想要确定、事实性输出(解析、代码生成、JSON 抽取)时设低(0–0.2)。想要创造或变化时设高(0.6–0.9)。避免同时改两者——它们的相互作用很难推理。

Streaming:首 token 赢得用户注意力

没有 streaming,你的 app 要等整段响应才显示任何东西——对长响应这可能是 5–20 秒的空白屏。有了 streaming,token 在模型生成时陆续到达,用户在发出请求后几百毫秒内就看到文字出现。总延迟往往相同,但感知延迟低得多。对任何面向用户的 feature,streaming 几乎总是正确的默认。

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a merge sort in Python."}],
) as stream:
    for text in stream.text_stream():
        print(text, end="", flush=True)  # 每个 chunk 随生成到达

final = stream.get_final_message()
print(f"\nTotal tokens: {final.usage.input_tokens + final.usage.output_tokens}")

在 web 应用里,你通常用 Server-Sent Events(SSE)或 WebSocket 把流从服务器代理到浏览器。浏览器随每个 chunk 到达即渲染。多数 LLM API SDK 在服务端替你处理 SSE 分帧——你只管迭代这个流。

JavaScript 里的 streaming

import OpenAI from "openai";

const client = new OpenAI();  // 从 env 读取 OPENAI_API_KEY

const stream = await client.chat.completions.create({
  model: "gpt-4o",
  stream: true,
  messages: [{ role: "user", content: "Explain async/await in JavaScript." }],
});

for await (const chunk of stream) {
  const delta = chunk.choices[0]?.delta?.content ?? "";
  process.stdout.write(delta);  // 生产中经 SSE 流到浏览器
}

Prompt Caching:最划算的单项成本优化

LLM 推理之所以贵,是因为模型每次调用都必须从头处理每个输入 token。但许多生产工作负载有一个大而稳定的前缀:一段长 system prompt、一份正被分析的文档,或一整个被逐查询处理的代码库。prompt caching 让 API 把这个稳定前缀预处理一次,并为后续相同前缀的调用复用 KV cache——大幅削减成本与延迟。

缓存经济学如何运作

这笔账很有说服力:若你的 system prompt 是 10,000 token、每天 100 次调用,朴素算就是 1,000,000 输入 token。有了缓存,就是 10,000 缓存写入 token(首次调用或 TTL 刷新时)加上其余的 90,000 缓存读取 token——这部分 prompt 的输入 token 成本大约降 90%。

用 Claude 实现 prompt caching

import anthropic

client = anthropic.Anthropic()

# 一段跨多次调用保持稳定的长 system prompt
LONG_SYSTEM_PROMPT = """
You are a senior Python engineer reviewing code for a fintech company.
[... 5,000 tokens of detailed coding standards, style rules, security
 requirements, and example patterns ...]
"""

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": LONG_SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},  # 标记为可缓存
        }
    ],
    messages=[
        {"role": "user", "content": "Review this function: def process(x): return x*2"}
    ],
)

usage = response.usage
print(f"Cache write tokens: {usage.cache_creation_input_tokens}")
print(f"Cache read tokens:  {usage.cache_read_input_tokens}")
print(f"Regular input tokens: {usage.input_tokens}")

Claude 上缓存 TTL 默认 5 分钟。只要调用在 TTL 内到达,后续请求就命中缓存。对高流量应用这基本上是常态;对低流量应用,你可能需要实现一个 keep-alive ping,在真实调用之间保持缓存温热。

缓存何时帮助最大

• 文档 Q&A:把整份文档嵌进 prompt 一次;每个用户问题都是对文档前缀的缓存读取。
• 代码评审流水线:把你整份编码规范文档放进 system prompt;每个 PR 都对照同一份被缓存的规范评审。
• 带长 system prompt 的多轮聊天:system prompt 跨所有回合稳定;缓存它,只为每条新 user 消息付全价。
• 批处理:若你用同一模板处理 10,000 份文档,模板前缀在首次请求后被缓存。

token 成本与成本估算

每次 API 调用按消耗的 token 计费:输入 token(你的 prompt + 对话历史)和输出 token(模型的响应)。在多数模型上,输出 token 比输入 token 贵 3–5 倍,反映自回归生成所需的额外算力。理解这个比例对成本感知的应用设计至关重要。

动手前估算成本

一个粗略经验法则:1 token ≈ 0.75 个英文单词。一份 1,000 词的文档大约 1,333 token。按应用类型的常见输入/输出比:

给产品定价前,在真实输入上实测你的实际 token 用量。API 响应里的 usage 对象总含精确 token 计数——把它们打进日志。你的估算往往会偏差 2–3 倍,而这对单位经济性很重要。

降本杠杆

• prompt caching——对稳定前缀工作负载的最大单项杠杆。
• 简单任务用更小模型——Haiku 或 GPT-4o-mini 比 Opus 或 GPT-4o 便宜 10–20 倍;许多分类与抽取任务不需要最大的模型。
• 限制 max_tokens——为你的任务设一个贴切的紧上限。从文档抽取单个字段的函数不需要 4096 输出 token。
• 截断对话历史——多轮应用会累积历史;裁剪或摘要旧回合,避免反复为陈旧上下文付费。
• Batch API——Anthropic 和 OpenAI 都提供异步 batch 端点,价格约为同步的 50%,适合非实时工作负载。

延迟优化

延迟有两个不同成分:首 token 时间(TTFT)——用户多久才看到东西——和总完成时间——完整响应多久才可用。streaming 解决 TTFT。总时间取决于模型大小、prompt 长度、输出长度。

降低首 token 时间

• 流式响应。TTFT 降到模型开始生成第一个 token 的时间——主流 API 通常 200–800ms——与总响应长度无关。
• 用 prompt caching。缓存命中跳过输入处理,对大的被缓存前缀把 TTFT 降低 3–5 倍。
• 用地理上更近的 API 端点。到 API 服务器的往返延迟直接加到 TTFT 上。Anthropic 和 OpenAI 都提供区域端点。
• 减小输入体量。更小的 prompt 处理起步更快。若你在加载一整份大文档,考虑改用检索(只取相关片段)。

降低总完成时间

• 任务允许时用更小、更快的模型。Haiku 生成 token 的速度大约是 Opus 的 3–4 倍。
• 设一个紧的 max_tokens。模型一直生成到撞上 max_tokens 或 stop sequence。不必要的高上限不会让模型生成更多——但也不会让它更早停下。让 max_tokens 匹配现实的输出长度。
• 并行化独立调用。若你需要两个互不依赖的 LLM 输出,用 asyncio.gather 或 Promise.all 并发发起两个 API 调用。

错误处理与重试逻辑

LLM API 是网络服务,偶尔会返回错误。最常见的两类是瞬时错误(限流、临时过载)和永久错误(无效输入、认证失败)。对它们一视同仁——要么总重试要么从不重试——都是错的。设计良好的客户端会区分两者。

HTTP 状态码及对策

import time, random, anthropic
from anthropic import RateLimitError, APIStatusError

def call_with_retry(client, max_retries=5, **kwargs):
    for attempt in range(max_retries):
        try:
            return client.messages.create(**kwargs)

        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)  # 指数 + jitter
            print(f"rate limit hit, retry {attempt+1} in {wait:.1f}s")
            time.sleep(wait)

        except APIStatusError as e:
            if e.status_code in (500, 529) and attempt < max_retries - 1:
                wait = (2 ** attempt) + random.uniform(0, 1)
                time.sleep(wait)
            else:
                raise  # 4xx 错误:别重试

加 jitter(那个 random.uniform(0, 1))是为避免惊群问题:若 100 个客户端同时撞限流、又都在精确的 2 秒、4 秒、8 秒重试——它们会在每个重试点一起锤 API。jitter 把重试摊到一个时间窗里,显著降低重试引发的负载尖峰。

Structured Output:让响应可解析

当你的应用以编程方式解析模型响应——抽取字段、填库、触发下游逻辑——你需要可预测格式的输出。自由文本不是。一个在你需要 {"answer": 42} 时返回 "The answer is 42." 的模型会弄坏你的解析器。structured output 通过约束模型能生成什么来解决这点。

三种方法,按可靠性排

1. 纯 prompt 的 JSON:在 system prompt 里指示模型“返回有效 JSON”并给一个例子。在强模型上 80–95% 的时候有效;其余情况产出带尾随文本、缺引号或转义非法的 JSON。没有校验 + 重试循环的话不可用于生产。
2. JSON 模式:一个参数(OpenAI 上 response_format: {type: "json_object"}),约束模型始终产出有效 JSON。解决语法错误;但不保证 schema(键、类型、嵌套)符合你的预期。
3. schema 约束生成(把 tool use 当输出):把期望输出经由 tool 定义为一个 JSON Schema;模型被迫发出一个符合该 schema 的 tool call。这是最可靠的方法,也是我们对生产的推荐。

import json, anthropic

client = anthropic.Anthropic()

# 把期望的输出形状定义为一个 tool
extract_tool = {
    "name": "extract_bug_report",
    "description": "Extract structured data from a bug report.",
    "input_schema": {
        "type": "object",
        "properties": {
            "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
            "affected_component": {"type": "string"},
            "reproduction_steps": {"type": "array", "items": {"type": "string"}},
            "is_regression": {"type": "boolean"},
        },
        "required": ["severity", "affected_component", "is_regression"],
    },
}

response = client.messages.create(
    model="claude-haiku-4-5",  # 抽取任务 → 用更便宜的模型
    max_tokens=512,
    tools=[extract_tool],
    tool_choice={"type": "tool", "name": "extract_bug_report"},  # 强制使用 tool
    messages=[{"role": "user", "content": BUG_REPORT_TEXT}],
)

# 模型必须发出符合 schema 的有效 tool call
tool_call = response.content[0]
extracted = tool_call.input  # 已是解析好的 dict,无需 json.loads()
print(extracted["severity"], extracted["affected_component"])

把 tool_choice 设为强制使用特定 tool,你就保证了模型的整个输出是符合你 schema 的有效 JSON 对象。无需正则,无需对 json.loads 加 try/except,无需为格式错乱的输出做重试循环。

API 密钥管理与安全

API key 是 bearer token:谁拿到它就能发起记到你账上的 API 调用、读你发送的任何 prompt、并可能访问对话历史。粗心对待它,是同时收获意外账单和数据泄露的最快方式。

铁律

• 绝不把 key 嵌进客户端代码。JavaScript bundle 可被检视。移动 app 可被逆向。任何下载你 app 的人都能提取硬编码的 key。永远经你的服务器代理。
• 绝不把 key 提交进版本控制。哪怕在私有 repo,git 历史是永久的,而 repo 会被分享。给 .env 文件用 .gitignore,并跑一个 pre-commit hook 阻止误提交 secret。
• 从环境变量读取 key。所有 LLM SDK 在你不显式传入时会自动从环境读取 ANTHROPIC_API_KEY / OPENAI_API_KEY。这是最低限度的正确做法。
• 生产用 secrets manager。AWS Secrets Manager、GCP Secret Manager、HashiCorp Vault、Doppler 都提供集中、可审计、可轮换的 secret 存储。你的应用在启动时取 key,而非把它存进配置文件里的 env var。
• 定期轮换 key,怀疑泄露时立即轮换。两个 API 都允许同时存在多个 key,所以轮换无中断:创建新 key → 更新所有服务 → 吊销旧 key。

服务端代理模式

对浏览器应用,正确架构是:你的后端持有 API key,你的前端把请求发给你自己的后端,你的后端再转发给 LLM API。这个模式还让你能加上按用户的限流、内容过滤、日志、成本归因——全都在代理层,不必碰客户端代码。

选对模型

Anthropic 和 OpenAI 都提供分层模型阵容:一个大而强的模型(Opus、GPT-4o)、一个快而均衡的模型(Sonnet、GPT-4o-mini)、一个便宜又快的模型(Haiku、折扣档的 GPT-4o-mini)。“总用最好的模型”这种直觉在经济上是错的。任务—模型匹配是一个真实的工程决策。

选择的正确方式是实测:在 50–100 个真实样例上跑两个模型,在你的具体任务上比质量,算出成本差,再决定质量差是否值这个价差。别假设——去测。