LLM网关实践：统一接入GPT、Claude与DeepSeek的完整方案

把GPT、Claude和DeepSeek接进同一个LLM网关后，我最直观的感受是：代码里的if-else分支终于能砍掉了。之前每个模型一套SDK、一套鉴权、一套出参格式，业务方每接一个新模型就要重写一遍调用逻辑。这次通过统一接入层，把三家厂商的API收敛成一套OpenAI兼容格式，后面接模型只需要在配置文件里加一行路由记录，业务侧几乎不用改代码。这篇文章就把这次网关实践的完整过程拆开讲讲，包括方案选型、核心代码、路由切换以及我在上线前后踩过的坑，适合正在做多模型接入、想给团队搭一个统一入口的开发者参考。

1. 为什么需要一个LLM网关：从三把key说起

1.1 直接对接多家模型API时，最痛的是什么

如果你只在代码里接过一家模型厂商的API，可能觉得这事没什么大不了——申请一个API Key，装个官方SDK，按照文档拼请求体，解析返回结果，完事。但一旦同时接入GPT、Claude和DeepSeek，事情就完全变味了。

首先是鉴权方式不一样。OpenAI用的是Authorization: Bearer sk-xxx，Anthropic用的是x-api-key和anthropic-version两个Header，DeepSeek虽然兼容OpenAI格式，但它的服务地址、模型名称、限流策略又是另一套。业务代码里每接一个模型，就要多写一段“如何拼鉴权头”的逻辑，这些代码高度重复却没法抽象成统一函数——因为每家格式就是不一样。

其次是请求体和响应体差异。OpenAI的下一代版本已经改成结构化消息格式，响应里是choices[0].message.content；Anthropic的Messages接口响应是content数组，里面还可能分text块和thinking块；DeepSeek的响应结构和OpenAI接近，但它的reasoner模型会在输出里多一个思考内容字段。如果你在业务代码里直接解析这些结构，每个模型都要写一种解析器，而且一旦厂商更新字段，你的解析逻辑就得跟着改。

然后是流式输出。LLM应用基本都逃不开流式，但三家厂商的SSE事件格式完全不同。OpenAI是choices[0].delta.content，Anthropic是content_block_delta，DeepSeek是choices[0].delta.content但事件命名略有区别。很多团队在集成多个模型时，花时间最多的不是模型效果调优，而是流式协议转换。

最后是成本与用量统计。每个厂商的计费单位、价格、token统计口径都有差异，有的按输入输出分开计费，有的按字符计费，有的缓存命中和未命中价格差一倍。如果直接对接三家，月底对账能把人搞疯。

1.2 网关给出的答案：把“多样”变成“统一”

LLM网关做的事情，本质上就是一句话：业务方只面对一个API入口，所有厂商差异都在网关内部消化。

你可以把它理解成一个总电闸——房间里有很多电器，各自插头标准不同，但总闸只有一个；也可以把它理解成快递分拣中心，前端统一收件，后端按地址分拣到不同运输线路。业务方不再关心请求最终发给谁、用什么鉴权方式、响应怎么解析，这些全部由网关这一层负责。

具体来说，一个合格的LLM网关至少要管好几件事：统一API格式、密钥池管理、请求路由、故障切换、流式协议转换、用量计量与成本统计。把这些从业务代码里剥离出来后，业务方的接入成本能下降一个量级，我们团队后面接新模型时，几乎只需要填一张路由配置表。

这里想多说一句：很多人以为网关就是转发请求，其实真正的工程量全在适配器上。你要把不同厂商的协议差异抹平，同时保持请求语义一致。同一个messages数组，发给GPT时是一套，发给Claude时要转成对应结构，字段怎么映射、空值怎么处理、特殊指令怎么兼容，这些是网关最有价值的部分，也是最容易被低估的部分。

2. 方案选型与核心设计决策

2.1 为什么统一出口选OpenAI兼容格式

网关内部可以百花齐放，但对外暴露的统一接口格式，我强烈建议直接用OpenAI兼容格式。这不是因为OpenAI的技术最先进，而是因为它已经成为事实上的行业标准。

现在市面上绝大多数LLM工具链都默认支持OpenAI格式：LangChain、LlamaIndex、Vercel AI SDK，甚至编码工具Codex、Cline、Continue，都允许你通过修改base_url指向任意兼容服务。这意味着格式选OpenAI兼容，几乎所有生态工具都可以直接接入，不需要额外开发。

另外DeepSeek等厂商主动兼容OpenAI格式这件事也验证了这个判断——当厂商都向一个格式看齐时，选择它就是选择最低的适配成本。我们的网关虽然内部要处理Anthropic的特殊格式，但对外只暴露一套/v1/chat/completions接口，业务方只认这一套契约，这样才能做到“接新模型不改业务代码”。

2.2 三种网关实现路径对比

做统一接入层，摆在面前的无非三条路：写一个公共封装包、自建网关服务、接现成的第三方网关。我分别说下我的实测感受。

第一条路，在业务代码里封装一个common包，内部做厂商分支。这种方案对于只有一个服务的团队来说最快，但如果业务服务很多，每个服务都要部署这份代码，后续升级适配器就要全量发布，很容易出现版本漂移。而且它没法做集中式密钥管理和流量治理——key分散在各服务，想轮换或统计成本都得逐个改。适合做临时过渡，不适合长期使用。

第二条路，自建网关服务。这是我这篇文章采用的方案，用FastAPI加几个适配器就能搭出一个轻量的统一入口。它的优势是路由策略、成本统计、密钥管理都掌握在自己手里，数据链路可控，适合对安全要求高的公司内部场景。缺点是前期开发和维护需要投入，尤其是流式协议转换和故障切换逻辑要写得足够健壮。

第三条路，直接用现成的第三方网关。比如OpenRouter这类聚合平台，一个Key就能调用几百个模型，确实省事。但把公司内部数据发给第三方网关，对很多团队来说存在合规风险；而且第三方网关的计费和限流策略不可控，高峰期可能被限制并发。我的建议是：个人项目和个人调试可以用这种现成网关，但给公司业务做基础设施，还是自建更稳妥。

三条路的对比如下，方便你根据自己的场景做判断：

2.3 自建网关的模块拆分

确定自建之后，我把网关拆成五个核心模块，每个模块关注一件事，避免代码混成一团。

上游适配器是第一个模块，也是最核心的模块。它负责把统一请求格式翻译成各厂商的协议。OpenAI和DeepSeek可以直接透传，Anthropic需要做专门的格式转换。每个适配器还要负责处理本厂商特有的字段，比如Anthropic要求必须传max_tokens，OpenAI不要求但你传了也没事。

第二个模块是路由层，负责根据请求里的模型名找到对应的上游配置。模型名和供应商的映射关系放在配置文件里，业务方传什么模型名，就路由到什么供应商。

第三个模块是鉴权层，负责校验调用方身份。内部网关面对的调用方是公司各业务服务，不要直接用厂商的Key做校验，而应该给每个业务方发放独立的内部AppKey，这样即使某个业务服务的Key泄露，也只能影响它自己，不会把厂商主Key暴露出去。

第四个模块是审计与计量模块，负责记录每次调用的模型、输入输出token数、估算成本。日志要结构化存储，方便按业务方维度做成本分摊。

第五个模块是熔断与故障切换模块，负责处理上游限流、超时、5xx错误。它的核心逻辑是：当某个模型不可用时，自动把请求切换到同级别的备用模型上，同时把失败事件记录下来方便事后分析。

3. 网关核心代码实现

3.1 项目骨架与路由配置

我用FastAPI + httpx来写网关，原因很简单：FastAPI的异步特性适合转发场景，httpx支持HTTP/2和流式请求，代码写起来清爽。项目结构上不搞复杂框架，按“入口-路由-适配器”三层组织。

先看路由配置文件，这是整个网关的“通讯录”：

# router_config.yaml models: - name: gpt-4o provider: openai base_url: https://api.openai.com/v1 api_key_env: OPENAI_API_KEY - name: claude-sonnet provider: anthropic base_url: https://api.anthropic.com api_key_env: ANTHROPIC_API_KEY version: "2023-06-01" - name: deepseek-chat provider: deepseek base_url: https://api.deepseek.com api_key_env: DEEPSEEK_API_KEY

配置里写着模型名、供应商、上游地址、Key的环境变量名。每次接新模型，只要照着这个格式加一行，网关就能自动处理。为什么用环境变量存Key而不是直接写在配置文件里？因为配置文件会进代码仓库，一旦仓库泄露，Key就全暴露了；环境变量属于部署层配置，可以单独管理，也能做到不同环境用不同Key。这个习惯大家从第一天就要养好。

3.2 统一入口：OpenAI兼容的Chat Completions接口

网关对外暴露的接口只有一个：POST /v1/chat/completions。调用方把请求发进来，网关根据model字段找到上游配置，然后交给对应的适配器处理。

from fastapi import FastAPI, Request, HTTPException import httpx, asyncio app = FastAPI() @app.post("/v1/chat/completions") async def chat_completions(request: Request): body = await request.json() model_name = body.get("model") target = get_route_config(model_name) if not target: raise HTTPException(status_code=404, detail="model not found") adapter = get_adapter(target["provider"]) return await adapter.forward(body, target)

这个入口本身不关心发给GPT还是DeepSeek，它只做三件事：解析请求体、查路由配置、交给适配器。HTTP状态码也要做统一处理：模型不存在返回404，上游限流返回429，网关内部错误返回500，调用方只需要按这些状态码写重试逻辑。

如果你的内部系统已经有用户体系，可以在这一层做租户识别。从请求头里取X-Tenant-Id，后面所有计量和限流都基于这个标识。网关前的网关——也就是公司的统一认证层——也可以继续留着，不需要冲突。

3.3 上游适配器：把不同协议翻译成统一格式

适配器是网关的翻译官。OpenAI和DeepSeek走的是同一套逻辑，Anthropic需要单独写转换逻辑。我先把OpenAI适配器写出来：

class OpenAIAdapter: async def forward(self, body, target): headers = { "Authorization": f"Bearer {os.environ[target['api_key_env']]}", "Content-Type": "application/json" } async with httpx.AsyncClient() as client: resp = await client.post( f"{target['base_url']}/chat/completions", json=body, headers=headers, timeout=60 ) return Response(resp.content, status_code=resp.status_code, media_type="application/json")

注意这里把整个body直接透传上去了，没有做任何字段改动。因为调用方发的就是OpenAI格式，OpenAI和DeepSeek都能直接认。这看起来简单，但有一个隐藏优点：上游新增的参数（比如DeepSeek的thinking配置）不需要网关先感知，调用方直接传就能生效。网关越不聪明，业务方越灵活。

Anthropic适配器就没这么简单了。我们需要把OpenAI格式的messages转换成Anthropic的格式：

class AnthropicAdapter: async def forward(self, body, target): anthropic_body = { "model": body["model"], "max_tokens": body.get("max_tokens", 4096), "messages": [ {"role": msg["role"], "content": msg["content"]} for msg in body["messages"] if msg["role"] != "system" ] } if system_messages: anthropic_body["system"] = system_messages headers = { "x-api-key": os.environ[target["api_key_env"]], "anthropic-version": target["version"], "Content-Type": "application/json" } resp = await client.post(f"{target['base_url']}/v1/messages", ...) return convert_response(resp.json())

这里有两个容易踩坑的点。第一个是Anthropic要求max_tokens必传，OpenAI请求里没有这个字段的话，网关必须补一个默认值。第二个是Anthropic不支持OpenAI的system消息，系统指令要抽出来单独放到system字段里。如果直接透传，Claude会报错或者把系统消息也当成普通对话内容，导致行为异常。这个转换逻辑虽然不多，但漏掉任何一点，接Claude的时候一定会出问题。

3.4 流式响应统一处理

流式是LLM网关里最让人头疼的部分。非流式的转发只需要一个return resp，流式则要把上游推送的SSE事件逐个解析、转换、再推给下游。

我先把OpenAI和DeepSeek这类OpenAI兼容格式的流式处理写出来：

async def stream_openai(generator, target): async for chunk in generator: yield f"data: {chunk.decode('utf-8')}\n\n" yield "data: [DONE]\n\n"

下面这个例子完整展示了Anthropic流式适配的核心逻辑——上游的流格式和OpenAI不同，需要逐块解析，在content_block_delta里取内容，再拼成OpenAI的delta.content结构：

async def stream_anthropic(generator): async for line in generator: raw = line.decode("utf-8").strip() if not raw.startswith("data: "): continue data = raw[6:] try: event = json.loads(data) except json.JSONDecodeError: continue if event.get("type") == "content_block_delta": delta = event.get("delta", {}) text = delta.get("text", "") if text: payload = { "choices": [{ "delta": {"content": text}, "index": 0 }] } yield f"data: {json.dumps(payload)}\n\n" yield "data: [DONE]\n\n"

这个转换过程看起来不复杂，但它揭示了流式适配器的核心原则：无论上游发什么格式，你要输出的永远是OpenAI SSE格式。上游的事件类型、增量字段、结束标记，都在适配器里完成翻译，下游永远只认一种格式。这样做的价值在后端是巨大的，因为业务方的前端代码只需处理一种流式协议，而常见工具库对OpenAI流式的支持是最成熟的。

3.5 用量记录与成本估算

网关做完了转发，真正的价值还要体现在成本数据上。每转发一次请求，都要记录：调用方、模型名、输入token数、输出token数、上下游状态码、耗时。

def record_usage(tenant, model, usage, duration): cost = estimate_cost(model, usage) log = { "tenant": tenant, "model": model, "prompt_tokens": usage.get("prompt_tokens"), "completion_tokens": usage.get("completion_tokens"), "cost_usd": cost, "duration_ms": duration } # 写入结构化日志或消息队列

成本估算公式不复杂：模型单价（每百万token价格）× 实际token数。但如果要做得精细，就要处理缓存命中、输入输出分开计价、以及厂商不同计费口径的差异。我是把单价表放在配置里，按月更新，月底按租户聚合就能出成本报表。这个功能越早做越好，等费用账单出来再补，对账成本就高了。

4. 路由与故障切换实战

4.1 请求级路由策略：不只是“固定发到某一家”

路由是网关里最有策略价值的部分。最基础的路由就是“查表”：请求说用gpt-4o，我就发到OpenAI；说用deepseek-chat，就发到DeepSeek。但实际运营中，你往往会希望同一个业务请求能按策略自动选模型。

以我的实践为例，我实现了三种路由模式：

• 手动指定：调用方在请求里写死模型名，网关严格按名字路由，不做任何干预。
• 质量优先：当某个模型不可用时，自动按优先级列表切换到同级模型，比如gpt-4o失败后用claude-sonnet顶上。
• 成本优先：同样的请求，在效果可接受的前提下，优先选便宜的模型，比如非核心场景默认走deepseek-chat。

核心路由逻辑可以抽象成这样一个判断：

def select_target(request_model, strategy): candidates = get_candidates(request_model) if strategy == "cost_first": candidates.sort(key=lambda c: c["price_per_1k"]) for c in candidates: if is_healthy(c): return c return None # 触发降级或报错

在实际业务里，我会建议把路由策略做成可配置的，而不是在代码里写死。因为业务方对模型效果、成本、延迟的偏好随时会变。今天觉得DeepSeek效果够了，明天上线高难度任务又切回GPT，如果每次都在代码里改，这个网关就变成运维噩梦了。

4.2 健康检查与自动切换机制

网关的自动切换要解决的问题是：某个上游模型突然不可用，怎么保证业务不中断。在我看来，判断一个模型是否“健康”，不能只靠一次请求的成败，而要看连续错误率和错误类型。

我记录每个上游的滑动窗口错误率：最近100次请求中，5xx响应和超时请求的比例超过阈值（比如20%），就标记为不健康，后续请求不再路由到该模型。但如果只是403、404这类明确表示“请求参数错误”的错误，不应该触发切换——这种错误切换模型也没用，换了也是同样报错。

触发切换后，网关要返回一个明确的提示给调用方，比如在响应头里加一个X-Failover-From字段，标明原始请求应该发到哪个模型但实际发到了哪个模型。这个信息对调试非常有用，否则你会看到日志里模型名和数据对不上，排查要花半天。

切换逻辑还需要配合重试机制。如果是429限流错误，先做指数退避重试（第一次等1秒，第二次等2秒，最多重试3次），重试仍失败才切换模型。如果是超时错误，建议直接切换，因为超时的上游往往积压了大量请求，重试只会让情况更糟。

4.3 Key池管理与多租户限流

当请求量涨上去之后，单把Key会成为瓶颈。不管是OpenAI还是DeepSeek，对单个Key都有并发限制，超过就会返回429。怎么解决？不把压力全压在一把Key上，而是做一个Key池，网关在转发时轮流使用池中的多把Key。

def pick_key(provider): pool = KEY_POOLS[provider] return random.choice(pool)

这里的核心是随机或轮询时要做加权处理，不要做成严格轮流，否则某个Key恰好遇到慢请求时，下一个请求又排到它，会让它雪上加霜。我用的是带权重的随机算法：每个Key根据自己的历史成功率分配权重，成功率高的Key被选中的概率更高。实现不复杂，但效果明显，实测下来429的数量能下降一半以上。

多租户限流也不能漏掉。网关如果同时服务多个业务方，必须按租户做独立限流，否则某个业务方的突发流量会拖垮整个网关。我是按租户和模型维度各做一层限流：每个租户每秒最多N个请求，每个模型每秒最多M个请求，超过直接返回429，并附上重试时间建议。这个用简单的令牌桶算法就能实现，不需要引入复杂的流控组件。

5. 常见问题排查与避坑记录

5.1 401 unauthorized：incorrect api key provided

这个错误排查下来，90%的原因都不是Key本身错了，而是转发过程中Header丢了。

我在网关里遇到了一个典型问题：转发请求时，代码写的是headers={"Authorization": ...}，而调用方传的Key通过api_key字段放在请求体里。结果网关去环境变量里取Key，环境变量没配好，转发时Authorization头就是空的，上游直接返回401。

排查方法很简单：在网关的日志里把每次请求的Header打出来，看对应的Key前缀是否是sk-开头，是否和配置里的Key一致。另一个常见坑是环境变量里有换行符或引号，例如Key从某个管理平台复制出来带了前后空格或\n，在代码里strip()一下就解决了。

5.2 400 context length exceeded

上下文长度超限是每个做过LLM接入的人都绕不开的坎。大模型的输入token数受上下文窗口限制，请求超长时，厂商会返回类似maximum context length is 1048576 tokens的错误。

这个问题的根因通常是：业务方把系统提示词、历史对话、工具文档一股脑全塞进messages里，没做长度控制。网关在这一层能做两件事：

第一是在转发前计算请求的近似token数（可用tiktoken估算），超过模型上限时直接返回400，附带消耗最大的消息位置，方便业务方定位。第二是实现截断策略，比如只保留最近的N条消息，或按用户维度的滑动窗口，超出部分丢弃。截断策略要谨慎，不能盲目丢系统指令，否则模型行为会异常。我的经验是系统指令永远保留，历史对话按“最近一轮优先”截断。

5.3 流式响应出现空文本或截断

流式接入最容易出问题的点是格式转换。Anthropic流式的content_block_start事件不带文本，如果你把这段当文本直接输出，前端会收到一个空的delta；而message_stop事件不代表文本结束，真正的结束信号是content_block_stop。

我在第一次接Claude流式时就是踩了这个坑，导致前端一直显示“正在输入……”但永远没有结束事件。正确的做法是只处理content_block_delta里带text的事件，忽略其他事件类型，等收到message_stop再输出data: [DONE]。

另外，很多团队在网关里做流式缓冲——等上游全部接受完再一次性发给下游。这个对长回答来说体验会非常差。一定要用StreamingResponse边收边转边发，不要缓冲。

5.4 DeepSeek接入的几个容易被忽略的细节

DeepSeek虽然是OpenAI兼容格式，但实测下来有三个细节值得重点注意。

第一个是base_url要设置为https://api.deepseek.com，很多照着OpenAI的习惯会在后面加上/v1，实际上DeepSeek文档里推荐直接写在根路径，加了/v1在某些SDK版本里会多出一层路径导致404。这个不一致几乎每个新接入的同事都会踩一次。

第二个是DeepSeek有两个模型名：deepseek-chat和deepseek-reasoner。前者是普通对话模型，后者是推理模型，会输出思考过程。你想用推理能力时却不能简单地把历史消息全量塞给它，因为推理模型的上下文开销大，超限概率更高。

第三个是DeepSeek的限流策略相对激进，单Key并发一高就429。如果你要跑批量任务，建议在网关里加一层简单的队列和令牌桶，或者直接用Key池轮转。另外DeepSeek的上下文缓存会在命中时显著降低成本，网关要留意厂商返回的缓存命中字段，别把缓存命中也算成全额成本。

5.5 常见错误速查表

写在最后的一点经验

这套网关从设计到上线，我最大的体会是：别急着写代码，先把三家厂商的接口差异列成一张表，把请求体、鉴权头、流式事件、错误码、usage字段全部对照写清楚。这张表就是网关的需求文档，也是后面排查问题的索引。另一个心得是日志一定要带请求ID和租户ID，否则多模型切换后，你根本没法把一个用户请求的完整链路串起来。网关本身不复杂，复杂的是把它做成一个能持续演进、不会成为新瓶颈的基础设施。接入DeepSeek只是开始，后面有新模型上线，你只需要改配置、加适配器，剩下的交给时间检验。