手写 LLM API 网关：Anthropic 与 OpenAI 协议转换的完整实现
英文标题：手写一个 LLM API 网关：Anthropic 与 OpenAI 协议转换的完整实现
本文介绍一个 LLM API 网关项目的核心实现，目标是让 Claude Code、Cursor 等 AI 编程工具通过统一接口访问不同厂商的模型。文章重点拆解 Anthropic Messages 协议与 OpenAI Chat Completions 协议在 system prompt、tool use 及流式 SSE 响应方面的转换差异。模型路由采用“精确映射、厂商前缀推断、默认兜底”三级 fallback，以兼顾白名单控制与新模型名的自动适配。对于开发者而言，协议适配、流式 tool call 增量拼接和流结束后的 Token 计费，是构建多模型 API 网关时最关键的工程问题。
3.2 KB
589 B
1.6 KB
中文
最近在做一个 LLM API 网关项目，核心需求是让 Claude Code、Cursor 这类 AI 编程工具能够通过统一的接口访问不同厂商的模型。这篇文章分享协议转换和模型路由两个核心模块的实现思路。
一、为什么需要协议转换
目前主流的大模型 API 主要有两套协议：
OpenAI 协议（Chat Completions）
{
"model": "gpt-4o",
"messages": [
{ "role": "user", "content": "Hello" }
],
"stream": true
}
Anthropic 协议（Messages）
{
"model": "claude-sonnet-4-20250514",
"max_tokens": 4096,
"messages": [
{ "role": "user", "content": "Hello" }
]
}
看起来差不多，但在 tool use、streaming、system prompt 的处理上差异很大。如果只支持 OpenAI 格式，Claude Code 发出的 Anthropic 原生请求就会直接 404。
二、协议转换的核心差异点
2.1 System Prompt
OpenAI 的 system 消息直接放在 messages 数组中，Anthropic 则使用独立的 system 字段：
function anthropicToOpenAI(anthropicReq) {
const messages = [];
if (anthropicReq.system) {
const systemText = Array.isArray(anthropicReq.system)
? anthropicReq.system.map(s => s.text).join('\n')
: anthropicReq.system;
messages.push({ role: 'system', content: systemText });
}
for (const msg of anthropicReq.messages) {
messages.push(convertMessage(msg));
}
return {
model: anthropicReq.model,
messages,
stream: anthropicReq.stream,
max_tokens: anthropicReq.max_tokens,
temperature: anthropicReq.temperature,
};
}
2.2 Tool Use 的 Content Block 结构
这是最复杂的部分。Anthropic 的 tool_use 和 tool_result 是结构化的 content block：
{
"role": "assistant",
"content": [
{ "type": "text", "text": "Let me check..." },
{ "type": "tool_use", "id": "tool_001", "name": "get_weather", "input": {...} }
]
}
而 OpenAI 的 tool calls 是消息级别的独立字段：
{
"role": "assistant",
"content": "Let me check...",
"tool_calls": [
{ "id": "tool_001", "function": { "name": "get_weather", "arguments": "{...}" } }
]
}
转换逻辑：
function convertAssistantMessage(msg) {
const textBlocks = msg.content.filter(c => c.type === 'text');
const toolBlocks = msg.content.filter(c => c.type === 'tool_use');
const result = {
role: 'assistant',
content: textBlocks.map(t => t.text).join('\n') || null
};
if (toolBlocks.length > 0) {
result.tool_calls = toolBlocks.map(t => ({
id: t.id,
type: 'function',
function: {
name: t.name,
arguments: JSON.stringify(t.input)
}
}));
}
return result;
}
2.3 Stream 模式下的增量转换
流式场景更复杂。Anthropic 的 SSE 事件结构：
event: content_block_start
data: {"type":"content_block_start","content_block":{"type":"text","text":""}}
event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Hello"}}
event: content_block_stop
data: {"type":"content_block_stop"}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}
需要实时转换为 OpenAI 的 SSE chunk 格式：
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"index":0}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{},"finish_reason":"stop","index":0}]}
这里需要一个 stream transform 来完成增量映射，关键是处理 tool_use 的增量——Anthropic 会分步骤发送 tool_use（content_block_start → delta → stop），需要在收到完整的 input_json_delta 后，才能拼接出 OpenAI 的 function.arguments。
三、模型路由：三级 Fallback
另一块核心是模型路由——如何将用户发来的模型名匹配到正确的上游。
function resolveModel(modelName) {
const exact = db.prepare(
'SELECT * FROM model_mappings WHERE model_name = ? AND enabled = 1'
).get(modelName);
if (exact) return exact;
const provider = inferProvider(modelName);
if (provider && hasActiveKeys(provider)) {
return buildVirtualMapping(modelName, provider);
}
return fallbackToFirstEnabled();
}
function inferProvider(modelName) {
if (/^claude-/.test(modelName)) return 'anthropic';
if (/^gpt-|^o1|^o3/.test(modelName)) return 'openai';
if (/^deepseek-/.test(modelName)) return 'siliconflow';
if (/^gemini-/.test(modelName)) return 'google';
return null;
}
这样做的好处是：用户使用 Claude Code 发出 claude-sonnet-4-20250514 请求时，即使后台没有为这一具体版本创建映射，前缀匹配 claude- 也能将请求正确路由到 Anthropic。
四、流式场景下的 Token 计费
流式响应的计费比较棘手——总 Token 数只能从上游 API 最后一条 SSE 事件中取得。不能一边流式输出一边扣费，因为此时还不知道最终需要扣除多少。
做法是在上游流结束后，读取 usage 信息，并一次性写入 credit_transactions 表，由事务包裹：
async function handleStreamBilling(ctx, usage) {
const totalTokens = usage.input_tokens + usage.output_tokens;
const cost = calculateCost(ctx.mapping, totalTokens);
db.transaction(() => {
db.prepare('UPDATE users SET credits = credits - ? WHERE id = ?').run(cost, ctx.userId);
db.prepare(
'INSERT INTO credit_transactions (user_id, amount, type, tokens, model) VALUES (?, ?, ?, ?, ?)'
).run(ctx.userId, -cost, 'consume', totalTokens, ctx.model);
})();
}
五、一些踩坑记录
-
SQLite 并发：使用 WAL 模式加
busy_timeout，足以应对中转站级别的并发，不需要 Redis。单表主键查询耗时低于 1ms。 -
GFW 阻断：大陆 ECS 到 Cloudflare 边缘在某天突然被全量阻断（TCP + QUIC），网站出现 Error 1033。云厂商的自定义镜像跨地域复制救了急——从大陆创建镜像，复制到香港后再创建实例，
cloudflared在香港启动后即可连通。 -
SSH 反向隧道保活：支付回调跨云不通时，使用由 systemd 管理的 SSH
-R隧道解决。ServerAliveInterval=60用于防止断开，Restart=always保证进程崩溃后自动恢复。
总结
做一个能够同时对接多个 AI 工具的 API 网关，核心在于协议适配与路由策略的健壮性。Anthropic 协议虽然和 OpenAI 看起来相似，但在 tool use 和 stream 的实现上有大量细节需要处理。模型路由采用三级 fallback，可以让系统保持足够灵活：既支持白名单的精确控制，也能自动适配新的模型名称。
本文是个人项目实践总结，欢迎讨论技术细节。
English / 原文
最近在做一个 LLM API 网关项目，核心需求是让 Claude Code、Cursor 这类 AI 编程工具能够通过统一的接口访问不同厂商的模型。这篇文章分享协议转换和模型路由两个核心模块的实现思路。
一、为什么需要协议转换
目前主流的大模型 API 主要有两套协议：
OpenAI 协议（Chat Completions）
{
"model": "gpt-4o",
"messages": [
{ "role": "user", "content": "Hello" }
],
"stream": true
}
Anthropic 协议（Messages）
{
"model": "claude-sonnet-4-20250514",
"max_tokens": 4096,
"messages": [
{ "role": "user", "content": "Hello" }
]
}
看起来差不多，但在 tool use、streaming、system prompt 的处理上差异很大。如果只支持 OpenAI 格式，Claude Code 发的 Anthropic 原生请求就会直接 404。
二、协议转换的核心差异点
2.1 System Prompt
OpenAI 的 system 消息直接在 messages 数组里，Anthropic 有独立的 system 字段：
function anthropicToOpenAI(anthropicReq) {
const messages = [];
if (anthropicReq.system) {
const systemText = Array.isArray(anthropicReq.system)
? anthropicReq.system.map(s => s.text).join('\n')
: anthropicReq.system;
messages.push({ role: 'system', content: systemText });
}
for (const msg of anthropicReq.messages) {
messages.push(convertMessage(msg));
}
return {
model: anthropicReq.model,
messages,
stream: anthropicReq.stream,
max_tokens: anthropicReq.max_tokens,
temperature: anthropicReq.temperature,
};
}
2.2 Tool Use 的 Content Block 结构
这是最复杂的部分。Anthropic 的 tool_use 和 tool_result 是结构化的 content block：
{
"role": "assistant",
"content": [
{ "type": "text", "text": "Let me check..." },
{ "type": "tool_use", "id": "tool_001", "name": "get_weather", "input": {...} }
]
}
而 OpenAI 的 tool calls 是消息级别的独立字段：
{
"role": "assistant",
"content": "Let me check...",
"tool_calls": [
{ "id": "tool_001", "function": { "name": "get_weather", "arguments": "{...}" } }
]
}
转换逻辑：
function convertAssistantMessage(msg) {
const textBlocks = msg.content.filter(c => c.type === 'text');
const toolBlocks = msg.content.filter(c => c.type === 'tool_use');
const result = {
role: 'assistant',
content: textBlocks.map(t => t.text).join('\n') || null
};
if (toolBlocks.length > 0) {
result.tool_calls = toolBlocks.map(t => ({
id: t.id,
type: 'function',
function: {
name: t.name,
arguments: JSON.stringify(t.input)
}
}));
}
return result;
}
2.3 Stream 模式下的增量转换
流式场景更复杂。Anthropic 的 SSE 事件结构：
event: content_block_start
data: {"type":"content_block_start","content_block":{"type":"text","text":""}}
event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Hello"}}
event: content_block_stop
data: {"type":"content_block_stop"}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}
需要实时转换为 OpenAI 的 SSE chunk 格式：
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"index":0}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{},"finish_reason":"stop","index":0}]}
这里需要一个 stream transform 来做增量映射，关键是要处理 tool_use 的增量——Anthropic 对 tool_use 是分步发送的（content_block_start → delta → stop），需要在收到完整的 input_json_delta 后才能拼接出 OpenAI 的 function.arguments。
三、模型路由：三级 Fallback
另一块核心是模型路由——用户发来的模型名怎么匹配到正确的上游。
function resolveModel(modelName) {
const exact = db.prepare(
'SELECT * FROM model_mappings WHERE model_name = ? AND enabled = 1'
).get(modelName);
if (exact) return exact;
const provider = inferProvider(modelName);
if (provider && hasActiveKeys(provider)) {
return buildVirtualMapping(modelName, provider);
}
return fallbackToFirstEnabled();
}
function inferProvider(modelName) {
if (/^claude-/.test(modelName)) return 'anthropic';
if (/^gpt-|^o1|^o3/.test(modelName)) return 'openai';
if (/^deepseek-/.test(modelName)) return 'siliconflow';
if (/^gemini-/.test(modelName)) return 'google';
return null;
}
这样做的好处：用户用 Claude Code 发 claude-sonnet-4-20250514，即使后台没有为这个具体版本建映射，前缀匹配 claude- 也能正确路由到 Anthropic。
四、流式场景下的 Token 计费
流式响应的计费比较棘手——总 token 数只能从上游 API 的最后一条 SSE 事件里拿到。不能一边流一边扣，那样不知道要扣多少。
做法是当上游流结束后，取 usage 信息，一次性写入 credit_transactions 表，事务包裹：
async function handleStreamBilling(ctx, usage) {
const totalTokens = usage.input_tokens + usage.output_tokens;
const cost = calculateCost(ctx.mapping, totalTokens);
db.transaction(() => {
db.prepare('UPDATE users SET credits = credits - ? WHERE id = ?').run(cost, ctx.userId);
db.prepare(
'INSERT INTO credit_transactions (user_id, amount, type, tokens, model) VALUES (?, ?, ?, ?, ?)'
).run(ctx.userId, -cost, 'consume', totalTokens, ctx.model);
})();
}
五、一些踩坑记录
-
SQLite 并发 ：用 WAL 模式 +
busy_timeout足够应对中转站级别的并发，不需要 Redis。单表主键查询 <1ms。 -
GFW 阻断 ：大陆 ECS 到 Cloudflare 边缘在某天突然全阻断（TCP + QUIC），网站 Error 1033。云厂商的自定义镜像跨地域复制救了一命——从大陆建镜像 → 复制到香港 → 创建实例，cloudflared 在香港启动即通。
-
SSH 反向隧道保活 ：支付回调跨云不通时，用 systemd 管理的 SSH -R 隧道解决。
ServerAliveInterval=60防止断开，Restart=always保证崩溃自愈。
总结
做一个能同时对接多个 AI 工具的 API 网关，核心在于协议适配和路由策略的健壮性。Anthropic 协议虽然和 OpenAI 看起来相似，但在 tool use 和 stream 的实现上有大量细节需要处理。模型路由用三级 fallback 可以让系统足够灵活，既支持白名单精确控制，又能自动适配新的模型名。
本文是个人项目实践总结，欢迎讨论技术细节。