用网关调 Claude API:可靠性、价格、零留存转发的工程实践

网关在你的应用和 Anthropic 之间只是薄薄一层。做对了,你拿到更好的成本结构、更多上游选择,以及生产环境的运维余地,而请求和响应一个字节都没动过。做错了,牺牲的是模型表现的正确性。这篇文章讲怎么做对。

工程实践 · Anthropic Messages API · Streaming / Tool Use / Prompt Cache

直连 Claude API 留下了哪些没解决的问题

生产代码里直接打 api.anthropic.com 没问题,很多团队在第一天这样做就够了。但等到服务真正承载流量,有四个空洞会变得难受。

可靠性。 单一上游就是单一故障域。当上游出现区域事故、对吵闹账号做限流、或者推送一个细微改变行为的模型版本时,你的应用直接暴露在风险里,中间没有任何缓冲。没地方做熔断、没地方做带预算的重试、没地方在异常时切换到对等模型,除非你在调用点重写一大块代码。

价格。 一手 list price 对所有人都一样:Opus 输入 $5、输出 $25(每百万 token);Sonnet 是 $3 和 $15;Haiku 是 $1 和 $5。中小账号没有任何议价空间。网关把需求聚合起来,以更低的每 token 价格转售。BUZZ 价格明显低于上游,但保留了一致的结构(输入、输出、Cache 倍率),所以你的成本模型不需要重写。

多模型覆盖。 你的应用大概率不止用 Claude。Gemini 跑长上下文便宜草稿,GPT 做某个特定 eval,Grok 处理某个细分场景。每一个直连都意味着多一个凭证、多一个账单关系、多一个 SDK、多一种错误结构。一个网关、一个 key、一套错误词汇能把这些都收敛掉。

审计和可观测性。 精确到 token 的计费埋点、按功能维度归属用量、按团队做速率限制、一处统一吊销 key —— 这些不放在网关里就只能自己造。

这些空洞不是空想。当一次发布刚好赶上上游事故、当财务问"为什么两个产品共享一行账单"、当安保审计要追溯过去一个季度哪些 key 调用了哪些模型,缺口就会变成没人排期的工程债。网关就是把这些活提前干完的地方。

一个干净的网关该做什么、不该做什么

做转发层最大的诱惑是"帮帮用户"。注入一段安全前置语,看着 system prompt 太长就裁掉,负载高时把 Opus 流量悄悄路由到 Sonnet,缓存响应来省 token。每一个这样的选择都打破了应用以为自己跟模型签下的契约。

正确的网关应该是无聊的。接请求,转字节,返字节。要紧的就这四条性质:

• 透明转发。 请求体,包括 system、messages、tools、tool_choice、temperature、top_p、top_k、metadata,以及任何 cache control 标记,逐字节透传到上游。响应也是逐字节回来。不重写 prompt、不悄悄换模型、不注入指令。
• 零数据留存。 请求体和响应体不写盘、不进数据库、不进日志。只持久化计费元数据:模型名、输入和输出 token 数、时间戳,以及用于归属的 user_id。如果监管或客户问"你们留了什么",答案就是一份很短的清单。
• 不缓冲流。 Anthropic 的 Server-Sent Events 按 chunk 透传,第一个 token 上游一发出来就到客户端。在 benchmark 上你看不出区别,但所有用聊天 UI 的用户都能感觉到那种缓冲到完整响应再 flush 的网关。
• 稳定的错误语义。 上游的 4xx 和 5xx 原状态、原响应体返回。网关只对自身层面的失败(认证、限流、计费)报错,并用易于区分的错误码标记出来。

BUZZ 把这四条都做到了。一句话总结契约:调用 BUZZ 和直连 Anthropic 之间唯一的差别,应该只有 URL、key 和账单。

评估任何网关(包括我们自己)时该警惕的反模式

评估任何 AI 转发层时,问运营方三个问题,看他有没有在迂回:

1. 请求体在任何情况下被记录吗,哪怕是临时、加密、所谓"防滥用"目的?"是的,但是……"就等于是。
2. 请求会因为任何理由被路由到与所指定模型不同的模型吗?容量、延迟、成本都不行?静默替换会摧毁评估结果。
3. 请求体会被转换吗?剥字段、归一化大小写、追加 system 指令都算。除了 TLS 终止之外的任何动作都是转换。

代码迁移:从直连 Anthropic 到走网关

迁移路径故意做得很轻。Anthropic Python SDK 接受 base_url 参数,对大多数代码库来说,差异就这一行。

直连版本(迁移前)

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarize the CAP theorem in one paragraph."}],
)
print(resp.content[0].text)

走 BUZZ(迁移后)

import anthropic

client = anthropic.Anthropic(
    api_key="buzz-...",
    base_url="https://buzzai.cc",
)

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Summarize the CAP theorem in one paragraph."}],
)
print(resp.content[0].text)

变了两行:key 的值,以及 base_url。模型 ID 还是 Anthropic 那串字符串,因为网关是逐字节透传的。

Streaming

Streaming 用 SDK 自带的 helper。网关不缓冲 SSE chunk,首 token 的感知延迟就是上游延迟加一个固定的小网络跳。

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "Write a haiku about idempotency."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()
    print()
    print("usage:", final.usage)

Tool Use

Tool 定义和 tool_choice 原样透传。响应结构,包括 tool_use block 和后续 tool_result 的回环,跟直连 Anthropic 完全一致。

tools = [{
    "name": "get_weather",
    "description": "Get current weather for a city.",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string"},
            "unit": {"type": "string", "enum": ["c", "f"]},
        },
        "required": ["city"],
    },
}]

resp = client.messages.create(
    model="claude-opus-4-8",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "auto"},
    messages=[{"role": "user", "content": "What's the weather in Reykjavik in celsius?"}],
)

for block in resp.content:
    if block.type == "tool_use":
        print("call:", block.name, block.input)

Prompt Cache

Anthropic Prompt Cache 完整保留。把一段长且稳定的前缀(system prompt、文档、tool catalog)用 cache_control 标记,网关把指令原样转发。第一次请求写入缓存,TTL 内的后续请求按基础输入费率的一小部分读取。

system = [
    {
        "type": "text",
        "text": LONG_STABLE_RULES_DOCUMENT,
        "cache_control": {"type": "ephemeral", "ttl": "1h"},
    }
]

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    system=system,
    messages=[{"role": "user", "content": "Apply the rules to this transcript: ..."}],
)

print(resp.usage)  # cache_creation_input_tokens / cache_read_input_tokens

计费按 Anthropic 官方公布的模型走:5 分钟 cache 写入是基础输入费率的 1.25 倍,1 小时是 2 倍,读取是 0.1 倍。这些倍率叠加在网关已经打折的基础费率上,所以缓存场景下省下来的额是复利的。

OpenAI SDK 兼容

如果你的代码已经统一用 OpenAI 的 Python 或 TypeScript SDK,把 base_url 指向 https://buzzai.cc/v1,就能用 OpenAI 兼容接口调 Claude。这对于希望跨模型家族用同一个 client library 的团队很方便,不过要拿到 Tool Use 和 Streaming 事件的全部保真度,原生 Anthropic SDK 还是更直接。

from openai import OpenAI

client = OpenAI(
    api_key="buzz-...",
    base_url="https://buzzai.cc/v1",
)

resp = client.chat.completions.create(
    model="claude-opus-4-8",
    messages=[{"role": "user", "content": "Hello, Claude."}],
)
print(resp.choices[0].message.content)

Claude Code 一行接入

对于交互式编码场景,Claude Code 一行命令就能接到网关上:

curl -fsSL https://buzzai.cc/sh/claudecode.sh | bash

安装脚本会把 CLI 的鉴权切到 BUZZ key,流量也走网关。已有项目什么都不用改。如果你想在 pipe 之前先看一眼,脚本源码在 https://buzzai.cc/sh/claudecode.sh。

价格模型

价格结构和 Anthropic 保持一致,这样你的成本模型不用换形状。每个模型有一个基础输入费率和基础输出费率,都按每百万 token 计价。Prompt Cache 在输入费率上叠加倍率。网关的核心数字是:基础费率明显低于一手价格,具体折扣随上游报价和容量浮动。

作为参照,Anthropic 公布的 list price 是:

BUZZ 的价格明显低于上面的数字。因为折扣随市场变动,正式数字以实时价格页为准,而不是这篇文章:https://buzzai.cc/api/pricing。当前可路由的完整模型列表在 https://buzzai.cc/models,包括完整版本号(claude-opus-4-5-20251101、claude-sonnet-4-5-20250929、claude-haiku-4-5-20251001),便于团队 pin 到具体快照。

Cache 倍率和上游一致:5 分钟写入是基础输入费率的 1.25 倍,1 小时写入是 2 倍,读取是 0.1 倍。基础数小了,同样的倍率乘出来的绝对值也小,这才是预算上真正吃到的部分。

运维要点

速率限制和 backpressure

网关按 key 强制请求和 token 配额,独立于上游限制。当某个 key 接近配额时,网关返回 429 并带 Retry-After header。上游的 429 和 529(overloaded)按原状态原语义透传,SDK 重试逻辑还能像 Anthropic 文档里那样工作。

重试

Anthropic SDK 已经对幂等失败做指数退避重试。网关在服务端不做二次重试,因为这会在事故期成倍放大负载。如果你想要跨模型家族的 failover(比如 Anthropic 过载时切到 Gemini),用同一个 BUZZ key 在应用里显式实现。

错误处理

网关保留 Anthropic 错误信封。Anthropic 返回的 400 还是 400,error.type 和 error.message 都不变。网关只对自己层面的问题报错,这些错误的 error.type 用 gateway_ 前缀标记,下游代码可以干净地分支处理。

可观测性

每请求的计费元数据通过控制台和导出端点开放。元数据故意做得很窄:模型、输入与输出 token 数(包含 cache 写入和读取计数)、时间戳、归属的 user_id。没有请求体、没有 prompt 前缀、没有响应片段。如果你出于自身审计需要记录请求内容,在请求离开你的进程之前在自己的应用里做。

Key 卫生

Key 按项目隔离,一键吊销。因为同一把 key 跨 Claude、GPT、Gemini、Grok 都能用,泄露后旋转一个凭证就同时旋转了对所有模型的访问,这通常正是你想要的。

跨团队的成本归属

真实部署里反复出现的一种模式:同一个产品入口(聊天助手、code reviewer、内部 Q&A 工具)会因为不同任务调用多个模型。草稿走 Haiku,硬推理走 Opus,长上下文检索走 Sonnet。没有网关时,账单按 provider 一行平摊,只能从日志里反推。有了网关,每次请求带一组小标签(key、user_id、model),控制台按功能维度汇总开支,而且不存任何 prompt 内容。这就够每个季度回答"哪些功能拉高了成本线,单位经济模型还成立吗"。

版本纪律

pin 到快照 ID(比如 claude-opus-4-5-20251101)是评估流水线和回归测试套件里的有意识选择。网关原样透传快照 ID,你的 eval harness 会一直产出相同的分数,直到你主动迁移。对于交互式产品入口,通常用 family alias(claude-opus-4-8)更合适,因为它能自动吃到上游迭代。两者混用,CI 用快照、生产用 alias,是常见且合理的形态。

FAQ

结语

Claude API 网关正确的心智模型是"管道",不是"中间件"。它该接你的请求,完整送达,把上游响应完整还回来,按经过线缆的 token 计费。其他的一切 —— 转换、留存、悄悄换路 —— 都是你没要的功能,以及不想要的正确性风险。

BUZZ 就是围着这份契约设计的。同一套 SDK、同一组模型名、同一种响应结构,更低的每 token 成本,一把 key 跨多个模型家族,而且不留下你的数据副本。从 https://buzzai.cc 开始,实时价格在 https://buzzai.cc/api/pricing,可路由模型清单在 https://buzzai.cc/models。