在最近几周的 AI 原生项目里我明显感受到团队结构和系统架构都在经历一次“重排”。传统后端团队开始增加提示词工程、模型评测、RAG 数据标注等角色与此同时接入大模型的服务也不再像以前那样直接调用 SDK而是会在客户端和模型之间插入一层 LLM 网关用来统一处理路由、鉴权、限流、缓存、重试和降级。尤其是当线上模型服务出现超时或 5xx 时网关的故障取舍决定了最终用户看到的是错误提示还是一个可接受的兜底结果。这篇文章就把这两个话题一起梳理清楚AI 原生开发为什么会导致团队重排LLM 网关在架构中的定位以及故障场景下 Fail-Open、Fail-Closed、重试、熔断、降级这些策略应该如何选择。最后会带大家用一个最小可运行的示例把多上游故障转移的 LLM 网关跑起来。1. AI原生开发为什么会让团队“重排”1.1 什么是 AI 原生开发AI 原生开发不只是“在系统里调用一个大模型接口”。它的核心是指整个产品从需求定义、数据采集、交互设计到系统架构、运维监控都以大模型能力作为基础组件来设计和运作。传统应用开发里流程通常是需求评审、数据库设计、接口设计、开发、测试、上线。功能是确定性的输入输出可以明确枚举。而 AI 原生应用最大的不同在于核心逻辑不是一串确定的 if-else而是模型的概率输出。同一个 Prompt用户换一个说法返回结果可能就不同模型版本升级效果可能变好也可能变差上游大模型服务故障整个业务也会跟着抖动。这种不确定性直接影响团队分工。以前团队可以明确分成前端、后端、测试、运维前端管页面后端管业务逻辑测试管用例运维管机器。到了 AI 原生开发阶段模型效果、数据质量、幻觉控制、成本控制、故障降级这些新问题无法被传统岗位自然覆盖团队结构就需要重排。1.2 团队角色重排的三个方向我观察到的团队重排主要集中在三个方向。第一个方向新增“模型效果”相关角色。比如提示词工程师、模型评测工程师、数据构建工程师。提示词工程师负责把业务需求转成高质量的 Prompt评测工程师负责建设评测集和回归指标数据工程师负责清洗、标注 RAG 所需的业务文档。这些岗位在传统研发团队里不存在需要单独建角色。第二个方向后端工程师的职责边界扩大。后端同学不再只写 CRUD而是要理解 Agent 编排、工具调用、上下文管理、向量检索、LLM 网关策略。很多团队甚至要求后端能读懂模型返回的 token 使用量能定位成本上涨的根因。第三个方向运维和 SRE 角色开始接触模型服务。传统运维关注 CPU、内存、磁盘而 AI 原生开发还要关注模型 Provider 的可用性、API 限流、超时重试、熔断降级。LLM 网关的故障演练逐渐成为运维团队新的必修课。1.3 重排之后的协作变化团队重排后最直观的变化是评审方式变了。以前评审代码现在还要评 Prompt 设计和评测指标以前上线前做功能回归现在还要做模型效果回归。举个例子一个客服问答功能后端通过 LLM 网关调用大模型。如果模型版本升级后某个专业领域问题的回答质量明显下降靠传统测试用例很难发现。团队必须建立评测集让评测工程师在版本发布前跑一遍确认效果达标。另一个变化是故障责任边界。模型回答质量差算算法问题还是后端问题上游模型服务超时算运维问题还是研发问题如果团队在重排时没有理清边界线上故障时就会出现互相推诿。所以我会建议团队在成立第一天就明确模型效果由模型组负责网关可用性由平台组负责业务接入由业务组负责每层都有清晰的 SLO。2. LLM 网关概念与定位2.1 为什么需要一个 LLM 网关如果你只是做技术 Demo直接调用大模型 SDK 没有问题。但到了生产环境客户端直连模型 Provider 会暴露出一堆问题。首先是密钥安全。模型 API Key 放在前端等于把账号直接交给用户别人可以拿你的 Key 去刷量。正确做法是把 Key 放在服务端由网关统一管理。然后是模型切换成本。今天用 A 模型明天觉得 B 模型效果好如果代码里到处写死了 A 的 SDK替换工作量和风险都很大。通过网关屏蔽底层实现业务只面向一个兼容接口切换模型时只需要改配置。此外还有成本控制、限流、缓存、可观测性、故障转移这些问题。生产级系统不可能让每个业务团队各自实现一套最好是有一个统一的中间层来承载这些横切能力。这个中间层就是 LLM 网关。2.2 LLM 网关的核心能力LLM 网关本质上是客户端和大模型服务之间的一个控制面主要解决“接入、治理、观测、成本”四类问题。第一接入能力。网关向上游提供统一的 OpenAI 兼容接口底层可以对接商业大模型 API、开源模型私有化部署、本地 Ollama 服务等多种 Provider。业务方只需要通过 HTTP 调用网关不需要关心上游是哪家模型。第二治理能力。包括 API Key 管理、用户认证、限流配额、模型路由、内容审核、缓存等。网关可以按用户、按部门、按接口维度配置不同的限额防止某个业务把整个模型的预算跑光。第三观测能力。网关需要记录每次调用的模型名称、耗时、token 使用量、费用、错误状态方便业务方定位问题。LLM 网关的日志和指标是 AI 原生应用排障时最重要的数据来源。第四成本控制。大模型按 token 计费成本不透明。网关可以统计每个部门、每个应用的 token 消耗并设置预算上限和告警避免月底账单爆掉。2.3 和普通 API 网关的核心区别有人会问Kong、APISIX、Spring Cloud Gateway 这些已经很成熟了为什么还要单独做一个 LLM 网关普通 API 网关关注的是 HTTP 请求路由、服务发现、负载均衡、认证鉴权、限流这些通用能力。它不理解大模型调用的语义比如 Prompt、上下文长度、token 用量、模型选择、幻觉风险。LLM 网关则要额外处理几个事情按上下文长度合理分配模型对模型返回内容做校验区分哪些错误可以重试、哪些错误不能重试根据成本预算限制请求在模型故障时切换到备选模型。理想状态下普通 API 网关负责南北向流量接入LLM 网关负责面向模型 Provider 的策略治理两者可以共存。很多团队也会直接把 LLM 网关能力集成到自研的 API 网关里具体看基础设施现状。3. 故障取舍LLM 网关的三种关键策略3.1 LLM 调用失败的常见形态做 LLM 网关第一件事是理解模型调用会以什么方式失败。从网络层看有连接超时、读取超时、DNS 解析失败、TLS 握手失败。大模型推理通常比较慢如果网关设置的超时时间太短很多正常请求会被误判为失败如果太长又可能拖垮下游线程池。从 HTTP 状态码看上游可能返回 401 鉴权失败、403 权限不足、429 触发限流、5xx 服务端过载或停机。429 和 5xx 通常代表服务暂时不可用可以重试401、403 通常是配置问题重试意义不大。从业务语义看模型可能返回空内容、HTML 代码、非法 JSON、重复文本甚至在关键问题上给出错误答案。这类“业务质量失败”最隐蔽因为 HTTP 状态码是 200但结果不可用。这类失败目前无法单靠网关解决最好的方式是增加内容校验层和人工评测流程。3.2 Fail-Closed 与 Fail-Open 的选择LLM 网关遇到上游故障时通常有两种极端策略Fail-Closed 和 Fail-Open。Fail-Closed直译是“关闭失败”。意思是一旦模型服务不可用网关直接拒绝请求返回错误提示。这种策略适合安全要求高、错误结果代价大的场景。比如金融风控、医疗诊断、法律建议、内容审核。在这些场景里返回一个“我不知道”或一个保守结果也比给用户一个可能错误的模型输出更安全。Fail-Open直译是“开放失败”。意思是模型服务不可用时网关自动降级比如切换到备用模型、本地小模型或者返回一个预设的兜底文案。这种策略适合体验要求高、错误容忍度大的场景。比如智能客服首响、闲聊机器人、商品文案生成。用户没得到高质量回答但至少得到一个友好反馈不至于直接体验中断。实际生产环境很少是纯二选一更常见的是做一条降级链主模型 → 备用模型 → 本地小模型 → 静态回复 → 报错。每一级都有自己的超时、限流和熔断设置。3.3 重试、熔断与限流重试、熔断、限流是 LLM 网关故障处理的三件套。重试要谨慎。模型调用不是所有请求都幂等有些调用会产生费用重复发送意味着重复计费。所以我建议遵循几个原则只在超时、5xx、429 时重试重试次数限制在 1 到 3 次每次重试之间使用指数退避重试时优先切换到备用 Provider而不是死磕同一个 Provider。熔断可以防止对故障上游持续打流量。比如某个模型 Provider 连续返回 5 次 5xx熔断器打开网关在冷却期内直接绕过该 Provider不再发真实请求。冷却期结束后进入半开状态放一个探测请求成功则关闭熔断失败则继续打开。这个机制在高并发场景下非常重要否则一个小故障就可能被无限放大形成雪崩。限流则是保护模型 Provider 和自身成本的关键。网关必须支持按用户、按应用、按模型维度限流并且要关注 token 级别的限额而不仅仅是对请求次数的限额。比如每秒 100 次请求看起来不多但每个请求上下文都很长模型 Provider 可能按 token 计费穿透网关后把上游打爆。4. 实战从零实现一个最小可用 LLM 网关4.1 环境准备我们先准备一个最小可运行的演示环境。操作系统Windows / macOS / Linux 均可Python3.10 及以上依赖库fastapi、uvicorn、httpx安装命令pip install fastapi uvicorn[standard] httpx版本不需要刻意固定只要保证三者能共同工作即可。本文示例重点是演示 LLM 网关的故障转移思路实际生产环境请根据项目依赖重新评估版本。为了不依赖真实商业大模型 API我们会先写一个模拟上游服务用本地随机 503 模拟模型服务故障。这样可以在完全离线的情况下验证网关的降级逻辑。4.2 项目结构最终项目结构如下llm-gateway-demo/ ├── mock_upstream.py # 模拟上游 LLM 服务 ├── main.py # 最小 LLM 网关 └── requirements.txt # 依赖列表其中mock_upstream.py可以启动两个实例分别模拟“主模型”和“备用模型”。main.py是网关主体负责路由、超时、重试、熔断和故障转移。4.3 模拟上游服务先来看模拟上游服务。它实现了一个 OpenAI 兼容的/v1/chat/completions接口并通过FAIL_RATE环境变量控制返回 503 的概率。# mock_upstream.py 模拟一个 OpenAI 兼容的上游 LLM 服务。 通过 FAIL_RATE 环境变量控制故障概率。 import asyncio import os import random import sys import uvicorn from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel app FastAPI() FAIL_RATE float(os.getenv(FAIL_RATE, 0.4)) class ChatRequest(BaseModel): model: str messages: list max_tokens: int 512 temperature: float 0.7 app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest): # 模拟模型推理耗时 await asyncio.sleep(random.uniform(0.1, 0.4)) # 按概率返回 503模拟上游过载 if random.random() FAIL_RATE: return JSONResponse( status_code503, content{error: {message: upstream overloaded}}, ) content fmock response from {req.model} return JSONResponse( content{ id: mock-id, object: chat.completion, choices: [ { index: 0, message: {role: assistant, content: content}, finish_reason: stop, } ], usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30, }, } ) if __name__ __main__: port int(sys.argv[1]) if len(sys.argv) 1 else 9001 uvicorn.run(app, host127.0.0.1, portport)这个文件的核心点有两个第一FAIL_RATE环境变量。我们可以把主模型实例设为FAIL_RATE1.0表示 100% 故障备用模型设为FAIL_RATE0.0表示 100% 正常。这样就能清晰观察网关是否会把请求自动切到备用模型。第二返回体结构模仿 OpenAI 的 Chat Completions 格式。网关不需要关心上游真实实现只要返回结构兼容就能完成集成。4.4 核心网关代码接下来是网关主体main.py。这个文件比较长我会拆成几个部分讲解。首先是依赖、数据模型和错误类型定义# main.py 一个最小可运行的 LLM 网关示例。 能力 - 多上游路由 - 超时处理 - 重试与故障转移 - 熔断器 - 基础可观测性响应头 import asyncio import os import time from contextlib import asynccontextmanager from typing import List, Optional import httpx from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): model: Optional[str] None messages: List[ChatMessage] max_tokens: int Field(default512, ge1, le4096) temperature: float Field(default0.7, ge0.0, le2.0) class UpstreamError(Exception): 上游服务不可用可以触发重试或故障转移。 class ClientError(Exception): 客户端或上游业务错误不应重试。 def __init__(self, message: str, status_code: int 400): super().__init__(message) self.message message self.status_code status_code这里要注意错误类型的区分。UpstreamError代表上游服务暂时不可用可以走重试和降级分支ClientError代表参数、鉴权、权限等业务错误直接返回给调用方避免无意义重试。然后是上游配置和熔断器# 上游配置 # 生产环境建议放到配置中心这里仅用于演示。 PROVIDERS [ { name: primary, base_url: os.getenv(PRIMARY_LLM_BASE_URL, http://127.0.0.1:9001/v1), api_key: os.getenv(PRIMARY_LLM_API_KEY, ), timeout: 5.0, }, { name: fallback, base_url: os.getenv(FALLBACK_LLM_BASE_URL, http://127.0.0.1:9002/v1), api_key: os.getenv(FALLBACK_LLM_API_KEY, ), timeout: 10.0, }, ] RETRY_BACKOFF 0.5 class CircuitBreaker: 熔断器状态 closed正常放行连续失败达到阈值后进入 open open 拒绝请求冷却期结束后进入 half_open half_open放行一个探测请求成功则 closed失败则重新 open def __init__(self, name: str, failure_threshold: int 5, cooldown: float 15.0): self.name name self.failure_threshold failure_threshold self.cooldown cooldown self.failure_count 0 self.state closed self.opened_at 0.0 def allow_request(self) - bool: if self.state closed: return True if self.state open: if time.time() - self.opened_at self.cooldown: self.state half_open return True return False # half_open 只放行一个探测请求 return True def record_success(self): self.failure_count 0 self.state closed def record_failure(self): if self.state half_open: self.open_circuit() return self.failure_count 1 if self.failure_count self.failure_threshold: self.open_circuit() def open_circuit(self): self.state open self.opened_at time.time() def info(self): return { name: self.name, state: self.state, failure_count: self.failure_count, opened_at: self.opened_at, } breakers {p[name]: CircuitBreaker(p[name]) for p in PROVIDERS}这里的熔断器实现比较简化但逻辑完整。failure_threshold是连续失败多少次后打开熔断cooldown是冷却期秒数。网关在处理请求时每个 Provider 对应一个独立的熔断器避免一个 Provider 故障影响所有流量。然后是 FastAPI 应用和调用逻辑asynccontextmanager async def lifespan(app: FastAPI): app.state.client httpx.AsyncClient() yield await app.state.client.ac