大模型调用层实战：构建高可用LLM网关的设计与踩坑

做AI应用开发这一年多我踩得最深的坑不在模型效果上而是“把大模型调用直接写死在业务代码里”。一开始项目只有一个模型供应商代码里到处是SDK调用看着没啥问题。等第二个供应商接进来立刻乱了请求参数、返回结构、鉴权方式全不一样更别提哪家一限流整个业务跟着抖。后来我专门抽了一层LLM网关和大模型调用层把所有供应商适配、高可用策略、成本控制全部收拢到这一层才算是把这块地耕顺了。这篇就把我在实际项目里的做法和踩过的坑整理出来。适合正在做AI应用开发、准备接入多个大模型供应商或者被线上调用稳定性折腾得睡不好的同学。1. 为什么需要单独搞一层“大模型调用层”1.1 直接调SDK写业务代码前期爽后面痛很多AI应用一开始都是从“调一个供应商”起步的。业务代码里直接new一个OpenAI客户端或者用某个国产模型的Python SDK写到哪调到哪。看起来简单直接第一版可能两三天就上线了。但这时候你欠下的技术债会在第二个供应商进来的时候一次性偿还。举个例子。业务方今天说“我们想同时对比几家模型的效果”你要接Claude、文心、通义、智谱这些。每家SDK的接口不一样有的用messages数组有的用prompt字段有的叫max_tokens有的叫max_new_tokens有的返回在choices[0].message.content里有的返回在output.text里。如果你在每一个业务接口里都对不同供应商做适配那这个适配代码会散落在几十个文件中后面改一个超时参数都要全局搜索还容易漏。更麻烦的是异常处理。一个供应商超时了要不要重试重试会不会重复扣费另一个返回400是不是因为prompt格式有问题这些问题散落到业务代码里之后排查一次线上事故能把人逼疯。我见过最夸张的情况是某业务模块里写满了“if model xxx”这样的分支后来要下掉一个渠道团队讨论了一周都不敢动怕误伤。1.2 网关层解决什么问题什么时候需要上把大模型调用层独立出来本质上就是做一套“LLM网关”它挡在业务和多个模型供应商之间对外提供一套稳定的接口对内做统一的模型路由和生命周期管理。这套网关通常要解决四个问题。第一是协议统一把不同厂商的请求参数、返回结构、鉴权方式全部归一化第二是高可用包括超时、重试、熔断、限流不能让单个供应商故障拖垮整个业务第三是成本与配额管理按业务线、按API Key统计调用量和token消耗防止预算失控第四是可观测性每个请求走了哪个模型、耗时多少、用了多少token、出错在哪个环节都要有日志和指标。那是不是所有项目一上来就要做这套东西我的建议是未必。如果你只是在本地跑个Demo或者业务规模很小只有一个供应商临时调一下也无妨。但出现下面几个信号就说明你该上了开始接入第二个供应商做模型对比或灾备业务方经常提出“这个接口换个模型试试”某次供应商服务抖动导致线上大面积超时月底算账时发现根本说不清每位客户消耗了多少token。这些信号出现任何一个花一周时间把LLM网关补齐都远比后续反复救火划算。2. 多供应商适配核心是把“差异”挡在网关里2.1 统一请求模型和响应模型先定义家园做多供应商适配第一件事不是写代码而是定义一套“网关自己的”请求和响应数据模型。它不跟任何一家供应商绑定是所有适配逻辑的共同语言。我们当时用的是一套精简的聊天补全模型。请求侧包含model、messages、temperature、max_tokens、stream、stop等字段响应侧则是id、model、choices、usage、created等字段。这套模型只保留业务真正关心的信息不追求覆盖所有厂商的所有参数。为什么必须统一因为你不能让上层业务感知“这个请求要发给OpenAI所以要用OpenAI的字段格式”。业务只负责把意图发给网关至于网关内部怎么把temperature翻译成各家参数那是网关的事。我用一个表格说明当时踩过的典型差异差异点OpenAI风格Anthropic风格部分国产模型风格对话消息字段messages数组system独立字段messagesmessages/prompt都有system提示词messages中rolesystemsystem参数messages中rolesystem最大生成长度max_tokensmax_tokens新版max_new_tokens随机性控制temperaturetemperaturetop_p/temperature流式返回text/event-stream不同类型eventtext/event-stream部分响应内容choices[0].deltacontent_block_deltachoices[0].delta这些差异看起来小但如果不做抽象业务层写一套兼容所有厂商的解析代码会非常痛苦。我在设计时把各家返回先通过适配器转换成统一的ChatResponse再提交给上层的业务方。就算某家模型改了返回层级也只需要改适配器业务代码完全不用动。2.2 供应商插件化与动态路由统一模型有了之后下一步就是供应商插件化。在我看来每个供应商就是一个Provider它要实现几个固定方法chat_completion、chat_completion_stream、embedding如果网关还要做向量化、check_health。上层路由只需要依赖这个接口不需要关心Provider内部调的是HTTP API还是SDK。插件化之后动态路由就顺理成章了。所谓动态就是供应商列表、模型映射、权重、优先级都不需要改代码通过配置文件或配置中心动态调整。路由这件事我建议从最简单的开始不要一上来就整什么机器学习路由。我们最早期就两个规则按指定模型路由、按主备降级路由。比如请求里带modelturbo网关认为这个逻辑名对应OpenAI的gpt-4o-mini和国产某厂的qwen-turbo两个真实模型平时默认走OpenAI如果OpenAI健康状态异常自动切到国产模型。权重路由也可以作为补充比如某段时间内60%流量走供应商A40%走供应商B用于灰度对比模型效果。但权重路由要小心会话类应用同一个用户的上下文可能分散到不同模型导致体验不一致。所以对于聊天场景我更推荐“按用户或会话绑定供应商”即在网关里维护一个会话级别的路由亲和性同一个session_id始终走同一个供应商。下面这个简化代码片段展示了路由的核心结构# router.py import random from typing import List, Optional class RouteDecision: def __init__(self, provider_name: str, model_name: str): self.provider_name provider_name self.model_name model_name class GatewayRouter: def __init__(self, provider_registry, health_status): self.provider_registry provider_registry self.health_status health_status async def route(self, request): # 请求指定了逻辑模型名例如turbo candidates self.provider_registry.get_models(request.model) if not candidates: raise ValueError(fmodel {request.model} not configured) # 如果指定了要走的供应商优先用 if request.preferred_provider: for cand in candidates: if cand.provider request.preferred_provider: return RouteDecision(cand.provider, cand.model_name) # 根据权重轮询或随机 healthy_candidates [] for cand in candidates: if self.health_status.is_healthy(cand.provider): healthy_candidates.append(cand) if not healthy_candidates: raise RuntimeError(no healthy provider) weights [c.weight for c in healthy_candidates] chosen random.choices(healthy_candidates, weightsweights)[0] return RouteDecision(chosen.provider, chosen.model_name)这里没有做太复杂的东西但足以支撑一个稳定的路由循环。真正的复杂度其实不在路由算法而在健康状态管理和错误处理。2.3 模型能力差异与降级策略多供应商适配还有一个容易翻车的点不同模型的能力差异很大。有的模型支持function calling有的还不支持有的支持JSON模式有的只支持字符串输出有的流式稳定有的断流概率高。所以网关里一定要有一份“能力矩阵”登记每个真实模型的能力标签。比如function_call、json_mode、vision、streaming、long_context等。请求进来时网关根据业务声明的能力要求筛选符合条件的供应商而不是什么请求都往第一个可用模型上丢。降级策略也要基于能力矩阵设计。一个典型场景主供应商因为限流或者故障不可用网关自动切到备供应商。这个“备”不能只看健康状态还要看能力和输入格式。比如主供应商支持function calling备供应商不支持你直接把原始请求丢过去对方大概率会报错或者忽略函数定义。正确的做法是提供一个降级处理器在切换时把function calling相关的参数剥掉并提示应用层“当前模型不支持工具调用”。我在实际项目中总结了一条原则降级宁可慢不能错。永远不要在降级路径上访问一个未知字段也永远不要在降级时丢掉必要的上下文。更安全的做法是在降级前用配置中心把模型映射和参数映射临时调整好再切换流量。人工可控的降级往往比“全自动聪明降级”更可靠。3. 高可用大模型调用层的设计要点3.1 超时、重试与熔断三个老伙计换个新环境LLM调用和普通HTTP调用有个很大区别它真的慢。一个复杂推理请求可能要几十秒流式更是能持续一分钟以上。所以“超时”这件事不能照搬普通接口的标准。我一般把超时分成三层。第一层是连接超时通常设置3到5秒第二层是首包超时也就是发完请求后等第一个token返回的时间这个根据模型不同可能从5秒到30秒不等第三层是空闲超时针对流式响应两个token之间超过15秒没有数据就认为连接已经死了。对于非流式请求还需要设置一个总超时比如60秒或120秒。重试就要更克制。生成型接口不天然幂等同一个prompt请求两次可能生成内容不同还会产生两次费用。所以我的重试原则是只有网络层错误、连接重置、429限流以及5xx服务端错误才重试业务参数错误400、401一律不重试。重试次数控制在2次以内并且使用指数退避加随机抖动。重试请求要打上retry_count标签方便在日志里识别。熔断是防止雪崩的关键。如果某个供应商连续报错再多的重试都是浪费。我在网关里为每个Provider单独维护一个熔断器状态有CLOSED、OPEN和HALF_OPEN。CLOSED代表正常错误率超过阈值就变成OPENOPEN状态下直接快速失败不再真实请求过一段时间放少量探测请求成功则恢复为CLOSED失败则回到OPEN。有一个容易被忽略的点熔断的指标不能只看HTTP状态码还要把超时和空响应算进去。有些供应商响应200但request_id异常或者返回的token数量为0这种也应当算作失败。3.2 限流、配额与健康检查的联动LLM网关的限流分为入口和出口两级。入口限流是保护业务后端按API Key或用户维度限制每秒请求数出口限流是尊重供应商配额防止单个渠道被打爆。入口限流我们用的是令牌桶算法分布式环境用Redis存储令牌保证多个网关实例共享限流状态。出口限流更要精细。每家供应商都有每分钟请求数RPM和每分钟token数TPM限制。网关要动态统计最近一分钟的请求量和token消耗在接近阈值时主动排队或降级而不是傻乎乎地硬冲。否则你会在供应商侧收到一堆429然后被对方限得更狠。健康检查和限流通常是联动的。我每隔30秒会调用一次供应商的轻量级接口做健康探测同时把真实请求的错误率、平均延迟也纳入健康判断。如果某个供应商连续5个请求失败就把它的状态置为“不健康”路由自动跳过。当它恢复稳定后再逐步把流量调回去。这个机制还要考虑“冷却”。不要一恢复就立刻全量放流量我一般会让恢复后的供应商承担20%流量观察几分钟没有异常再增加到50%、100%。这本质上是一种手动/半自动的灰度回归。3.3 缓存、成本与数据安全一个都不能少LLM调用成本不低。同样是聊天对话每天上万次调用几个月下来账单可能让人肉疼。所以网关层一定要做成本优化第一招就是缓存。缓存适合那些“相同或高度相似的请求”。比如智能客服里的常见问题用户问法接近完全可以让网关缓存住第一次的回复后续相同意图直接命中。最简单的缓存key是请求参数的哈希值精细一点的是把消息内容做向量化用余弦相似度判断是否命中缓存。但向量缓存要慎重相似不等于完全一致如果业务对答案准确性要求极高建议还是只做精确缓存。成本优化的第二招是模型选择。同一个任务用旗舰模型和轻量模型可能效果差距不大成本却相差10倍。网关可以根据业务优先级设置“成本路由”高价值用户走贵模型普通用户走便宜模型高峰期自动降级到性价比更高的模型。数据安全同样要在网关层做。日志里绝不能明文记录完整的prompt和response必须做脱敏或者截断。密钥管理要单独拎出来存放在专门的密钥管理服务中不要在配置文件里写死密钥更不要把密钥打进容器镜像。另外不同供应商的数据驻留和隐私协议不同接入前要让安全团队评估是否符合自己公司的合规要求。4. 实操从零构建一个轻量LLM网关4.1 技术选型FastAPI httpx Redis够了自研LLM网关不一定要搞很重的框架。我选的组合是FastAPI作为HTTP服务框架httpx作为异步HTTP客户端Redis承担限流和分布式锁Prometheus Grafana做监控指标展示。如果公司已经有完善的监控体系也可以直接上报到现有平台。为什么不用某些大而全的网关中间件因为LLM调用层和传统流量网关有一个关键区别它需要理解模型语义要做流式转发、token统计、模型降级。自研虽然要写一些代码但胜在灵活能按照自己的业务需求调整路由策略。当然如果你们团队时间紧、供应商也不多也可以先基于开源方案改哪怕后面再自研也能少走很多弯路。目录结构我会按“领域”划分而不是按“技术层”划分llm-gateway/ main.py core/ config.py schemas.py router.py circuit_breaker.py providers/ base.py openai_provider.py anthropic_provider.py custom_provider.py middleware/ auth.py rate_limit.py logging.py storage/ redis_client.py这样做的目的是让“供应商”这一维度非常清晰。新增一个供应商只需要在providers目录里加一个文件然后在配置里登记不用动业务代码。4.2 统一数据模型与配置文件我用Pydantic定义请求和响应模型。这里要特别注意流式和非流式需要复用同一套消息格式。# core/schemas.py from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str Field(..., description逻辑模型名) messages: List[ChatMessage] temperature: float 0.7 max_tokens: int 1024 stream: bool False stop: Optional[List[str]] None preferred_provider: Optional[str] None user_id: Optional[str] None request_id: Optional[str] None capabilities: Optional[List[str]] Field( default_factorylambda: [chat] ) class ChatChoice(BaseModel): index: int message: Optional[ChatMessage] None delta: Optional[ChatMessage] None finish_reason: str class ChatResponse(BaseModel): id: str model: str provider: str choices: List[ChatChoice] usage: Dict[str, Any] Field(default_factorydict) created: int 0配置文件我用YAML。里面维护两个重要映射一个是逻辑模型到供应商真实模型的映射另一个是供应商实例的连接参数。# config/providers.yaml providers: - name: openai base_url: https://api.openai.com api_key_env: OPENAI_API_KEY timeout: 60 health_check_path: /models weights: turbo: model_name: gpt-4o-mini weight: 80 pro: model_name: gpt-4o weight: 50 - name: qwen base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key_env: DASHSCOPE_API_KEY timeout: 30 weights: turbo: model_name: qwen-turbo weight: 20 models: turbo: fallback_order: [openai, qwen] capabilities: [chat, streaming] pro: fallback_order: [openai, qwen] capabilities: [chat, streaming, function_call]这里有个小细节不同供应商的base_url、鉴权方式差异大所以我让Provider实现类去解析这些配置而不是在配置里强行统一。4.3 转发路由与流式处理的实现下面这个代码是一个关键片段展示了Provider基类和OpenAI风格Provider的实现。# providers/base.py from abc import ABC, abstractmethod from typing import AsyncIterator, Dict, Any class BaseProvider(ABC): def __init__(self, name: str, config: Dict[str, Any]): self.name name self.config config abstractmethod async def chat_completion(self, request, params) - Dict[str, Any]: pass abstractmethod async def chat_completion_stream(self, request, params) - AsyncIterator[Dict[str, Any]]: pass async def check_health(self) - bool: return TrueOpenAI兼容接口的Provider实现起来最省事因为国内很多模型也提供OpenAI兼容协议。但即使如此我还是建议在Provider内做一层响应转换不直接依赖它“恰好兼容”。# providers/openai_provider.py import httpx from .base import BaseProvider from core.schemas import ChatResponse, ChatChoice from typing import AsyncIterator, Dict class OpenAIProvider(BaseProvider): def __init__(self, name, config): super().__init__(name, config) self.api_key os.getenv(config[api_key_env]) self.client httpx.AsyncClient( base_urlconfig[base_url], timeoutconfig.get(timeout, 60), headers{Authorization: fBearer {self.api_key}}, ) async def chat_completion(self, request, params): payload { model: params[model_name], messages: [m.dict() for m in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: False, } resp await self.client.post(/chat/completions, jsonpayload) resp.raise_for_status() data resp.json() # 这里只提取网关需要的字段 return { id: data.get(id), provider: self.name, model: data.get(model), choices: [ { index: item.get(index, 0), message: item.get(message, {}), finish_reason: item.get(finish_reason, ), } for item in data.get(choices, []) ], usage: data.get(usage, {}), created: data.get(created, 0), }流式处理是LLM网关最容易翻车的点。我在实现时选用了Server-Sent EventsSSE也就是text/event-stream。网关要做的不是等完整响应再返回而是边收供应商的token边转发给客户端。async def chat_completion_stream(self, request, params): payload { model: params[model_name], messages: [m.dict() for m in request.messages], temperature: request.temperature, max_tokens: request.max_tokens, stream: True, } async with self.client.stream(POST, /chat/completions, jsonpayload) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line.startswith(data:): continue data line[5:].strip() if data [DONE]: break yield data # 这里可以直接向上游转发流式转发要注意网关在转发时不能随便吞掉事件。有些模型的流式事件里携带usage信息通常放在最后一个chunk里网关要尽量原样透传同时自己也解析一份用于指标统计。如果中途断了网关要记得给客户端发送一个自定义的error事件而不是默默断开。4.4 熔断、重试与监控的落地编码熔断器我不推荐引入很重的中间件自己实现一个简洁版就够了。下面是一个滑动窗口计数的基础版本只保留最近10秒的错误率。# core/circuit_breaker.py import time from collections import deque class CircuitBreaker: def __init__(self, name, failure_threshold0.5, window_seconds10, open_time30): self.name name self.failure_threshold failure_threshold self.window deque() self.window_seconds window_seconds self.state CLOSED self.open_since 0 self.open_time open_time def record(self, success: bool): now time.time() self.window.append((now, success)) while self.window and now - self.window[0][0] self.window_seconds: self.window.popleft() if self.state CLOSED and len(self.window) 10: rate sum(1 for _, s in self.window if not s) / len(self.window) if rate self.failure_threshold: self.state OPEN self.open_since now def allow_request(self) - bool: if self.state OPEN: if time.time() - self.open_since self.open_time: self.state HALF_OPEN return True return False return True def on_success(self): self.record(True) if self.state HALF_OPEN: self.state CLOSED self.window.clear() def on_failure(self): if self.state HALF_OPEN: self.state OPEN self.open_since time.time() else: self.record(False)熔断器需要和重试逻辑搭配。我的处理流程是先查熔断器是否允许请求发送请求根据结果更新熔断器如果允许重试且是对应错误类型重试前先退避一段时间。请求要携带request_id贯穿整个链路方便在日志里串联。监控方面我每个请求会在结束时记录这些指标provider_name、model_name、succeed/failed、http_status、latency_ms、total_tokens、prompt_tokens、completion_tokens、retry_count、circuit_breaker_state。这些指标打进Prometheus配合Grafana做面板。告警规则至少要有三条单供应商错误率5分钟超过20%P95延迟超过10秒日调用成本超过设定阈值。5. 常见问题与故障排查实录5.1 路由结果和预期不一致先查配置和健康状态有一次我们线上流量突然全部切到了备用供应商业务反馈响应变慢了但主供应商看起来没挂。查日志才发现我们在配置中心改权重的时候不小心把主供应商的权重改成了0。看起来权重是0路由就把所有流量分给了备用。这个问题排查了半天最后是直接看RouteDecision的日志信息才定位到。所以我强烈建议网关里每个请求都要打印路由决策详情至少包括逻辑模型名、候选供应商列表、每个候选的健康状态、最终选择的供应商和原因。不要只打印一个最终选择否则出了问题你都不知道它是怎么选出来的。另外要检查健康检查本身是否误判。有些供应商的/models接口很轻但推理接口可能不稳定。如果你只依赖/models做健康检查很可能接口健康、推理超时。5.2 流式响应中途断开重试要非常小心流式断连在所有AI应用里都让人头疼。我们遇到过一种典型场景客户端从网关读取SSE流读到一半网关和供应商的连接断了客户端卡住直到超时才报错。排查时发现是网关到供应商之间没有设置空闲超时恰好某个模型生成长文本时两个token之间的间隔超过了默认超时连接被系统回收。后来我们把空闲超时加大到30秒并增加了首包超时问题缓解很多。但要提醒一句流式请求一旦已经向客户端推送了部分内容就尽量不要做自动重试。因为客户端可能已经渲染了部分文本再从头来一次会造成内容重复甚至前后不一致。正确做法是上游断了就立刻给客户端发一个错误事件并把这个请求标记为失败让应用层决定是提示用户重试还是继续等待。5.3 高并发下供应商被限流别想着硬扛某次活动流量上来后大量请求打到主供应商触发了对端的TPM限制返回了一堆429。当时网关的自动降级没有配置好导致所有请求都在等主供应商的429重试反而加剧了服务端压力。后来我们调整了策略当检测到429比例升高时直接把部分流量降级到备用供应商同时在网关出口限流中降低该供应商的每秒并发数。这里有个经验429响应里通常带Retry-After头网关一定要读取并遵守。不能对每个429都立刻重试。更好的做法是把这些请求放入一个带延迟的队列一点点放出去而不是在同一秒内全部冲回去。5.4 生产环境排障速查表我整理了一个简单的速查表线上出问题时可以照着看现象可能原因排查动作大量请求超时供应商单点故障/网络抖动看该供应商错误率、延迟指标摘除不健康节点切换备用后效果变差备用模型能力不足/参数映射不对检查能力矩阵和降级处理器必要时手动调回路由分配不均权重配置错误/健康状态异常查看配置中心快照和路由日志429突然增多出口限流未生效/配额不足检查令牌桶队列长度优化降级优先级token成本飙升缓存命中率低/重试过多统计缓存命中率限制单次请求重试次数流式响应断续空闲超时过短/上游不稳定调整超时参数开启断线自动补测这张表不是万能的但大多数LLM网关的线上故障都能归到这六类里。5.5 上线前一定要做故障注入测试最后说一个我的习惯。每次新增供应商或者改路由策略我都会在测试环境做几轮故障注入测试直接关掉主供应商的服务看网关能否在30秒内自动切换到备用供应商人为制造超时看超时设置和重试次数是否符合预期把某个备用供应商的API Key改成错的看健康检查能否及时摘除它用大流量压测看出口限流会不会触发雪崩。这些测试不需要很复杂的工具用脚本模拟失败请求就行。但一定要做。因为LLM网关的高可用不是写在代码里就算数而是要真的能扛住一次突如其来的故障才算数。我个人的体会是LLM网关最难的并不是某个技术点有多高深而是你愿不愿意把“供应商差异”“失败场景”“成本指标”这些脏活累活提前想清楚。把这一层做扎实之后业务换模型就像切换数据库连接池一样自然供应商出故障也不再是半夜夺命call的理由。希望这篇实战记录能给你一些可复用的思路少走点弯路。