1. 为什么企业需要一个LLM网关2023年的时候我们团队第一次把大模型接入生产系统做法特别朴素业务代码里直接写个HTTP请求把API Key塞在环境变量里调就完了。那时候一天也就几千次调用模型只有一个供应商也只有一家根本谈不上什么“网关”的概念。但到了2024年下半年情况完全变了——我们同时对接了四家模型供应商线上跑着七个不同参数量的模型日均调用量突破百万级财务每个月拿着账单来问“为什么这个月比上个月贵了40%”运维半夜被叫起来处理某家供应商的限流告警安全团队要求所有请求必须做内容审计和脱敏。这时候你再看那堆散落在各个业务代码里的API调用就知道什么叫技术债了。LLM网关本质上是在业务应用和大模型服务之间加了一层中间层所有请求先经过网关再由网关决定用哪个模型、怎么计费、要不要拦截、如何重试。听起来跟传统的API网关很像但LLM场景有几个特殊之处第一计费维度复杂不是简单的请求次数而是输入token和输出token分开计价不同模型价格差几十倍第二响应是流式的token一个一个吐出来网关得在流式传输的同时完成计量和审计第三模型输出不确定同样的输入可能得到完全不同的结果这对缓存和重试策略提出了新要求。这篇文章想聊的是一个企业级LLM网关从最简陋的“能转发请求”到“能计价、能路由、能治理”的完整演进路径。我会把它拆成五个阶段每个阶段解决什么问题、引入什么组件、踩过什么坑都尽量讲清楚。如果你正在负责公司的大模型基础设施或者正在评估要不要自建网关这篇文章应该能帮你少走一些弯路。2. 第一级演进从裸调API到统一转发层2.1 最原始的状态API Key满天飞我们最开始的状态是这样的三个业务团队各自申请了模型供应商的账号每个团队有自己的API Key代码里直接调供应商的SDK。这种做法在早期没问题但很快就暴露出几个致命缺陷。密钥管理失控。API Key散落在各个项目的配置文件、环境变量、甚至硬编码在代码里。有一次一个同事不小心把带Key的代码推到了公开仓库虽然及时发现删掉了但那种后怕的感觉至今记得。后来我们统计了一下光是有效的API Key就有十几个分布在四个不同的供应商平台上没有任何集中管理。无法统一观测。A团队说他们调用了10万次B团队说他们只用了5万次但供应商账单显示总共20万次。到底谁用了多少哪个模型消耗最多哪些请求失败了完全是一笔糊涂账。每个团队自己记自己的日志格式不统一也没有集中的监控面板。切换成本极高。有一次某家供应商突发故障我们想紧急切到备用模型结果发现业务代码里到处写死了模型名称和API地址改起来要动五六个仓库发版至少半天。等改完故障早就恢复了。2.2 统一转发层的最小可行方案痛定思痛之后我们决定先做一个最简版的统一转发层。目标很明确所有业务不再直接调供应商API而是调我们自己的网关地址网关再转发到真正的模型服务。这个阶段的技术选型其实不复杂。我们用的是Python的FastAPI框架因为它原生支持异步和流式响应写起来也快。核心逻辑就是接收请求根据请求里的模型名称找到对应的供应商配置替换API Key转发请求然后把响应流式返回给调用方。# 简化的转发核心逻辑 from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx app FastAPI() PROVIDER_CONFIG { gpt-4: { base_url: https://api.provider-a.com/v1, api_key: sk-xxx, provider: a }, claude-3: { base_url: https://api.provider-b.com/v1, api_key: sk-yyy, provider: b } } app.post(/v1/chat/completions) async def chat_completions(request: Request): body await request.json() model body.get(model) config PROVIDER_CONFIG.get(model) if not config: return {error: fModel {model} not supported} headers { Authorization: fBearer {config[api_key]}, Content-Type: application/json } async def stream_response(): async with httpx.AsyncClient(timeout120) as client: async with client.stream( POST, f{config[base_url]}/chat/completions, jsonbody, headersheaders ) as response: async for chunk in response.aiter_bytes(): yield chunk return StreamingResponse(stream_response(), media_typetext/event-stream)这段代码看起来简单但有几个关键点值得展开说。流式响应的处理。LLM的响应是Server-Sent Events格式一个token一个token地推过来。网关必须用流式转发不能等完整响应再返回否则用户体验会非常差。httpx的stream方法配合FastAPI的StreamingResponse是当时我们找到的最优解。注意timeout要设得足够长因为有些复杂请求模型要思考几十秒才吐出第一个token。API Key的集中管理。所有Key存在网关的环境变量或配置中心里业务方完全感知不到。这解决了密钥散落的问题但也带来了新的挑战网关本身成了单点如果网关挂了所有模型调用都断了。所以这个阶段我们就开始考虑高可用部署至少跑两个实例加一个负载均衡。模型名称的映射。业务方传过来的模型名称是统一的逻辑名称网关负责映射到具体的供应商模型ID。比如业务方说要用“fast”模型网关根据配置决定是走GPT-3.5还是Claude Haiku。这为后面的路由策略打下了基础。2.3 这个阶段踩过的坑第一个坑是超时设置。最开始我们给httpx设了30秒超时结果发现很多长文本生成请求根本跑不完。后来改成连接超时10秒、读取超时300秒才基本稳定。但读取超时太长也有问题如果供应商那边卡住了网关会一直挂着连接不释放。最终的方案是加了一个心跳检测如果超过60秒没有收到任何token就主动断开并返回错误。第二个坑是并发控制。早期没有做任何限流某个业务方写了个死循环疯狂调接口直接把网关打挂了。后来加了基于Redis的令牌桶限流每个业务方有独立的配额互不影响。第三个坑是日志脱敏。最开始我们把完整的请求和响应都记到日志里方便排查问题。后来安全团队审查时发现很多请求里包含用户隐私数据日志里明文存储是合规风险。之后我们加了脱敏规则对特定字段做掩码处理同时把日志保留周期从永久改成30天。3. 第二级演进计价与配额体系3.1 为什么计价这么难统一转发层跑通之后下一个问题马上就来了钱。财务拿着供应商账单来找我们说这个月比上个月多了三万块问能不能拆清楚每个业务线花了多少。我们当时只有请求日志没有token计数根本算不出来。LLM的计价跟传统API完全不是一个逻辑。传统API按调用次数计费一次请求就是一次。但LLM是按token计费而且输入和输出分开算。更麻烦的是不同模型的token计价差异巨大同样是100万token便宜的模型可能只要几块钱贵的要几百块。再加上有些供应商有阶梯定价、缓存命中折扣、批量调用优惠计费规则复杂得像电信套餐。3.2 Token计数的实现方案要计价首先得准确计数。我们试过几种方案最后落地的是在网关层做token计数。具体来说对于非流式请求等完整响应返回后从响应的usage字段里直接读取prompt_tokens和completion_tokens。这是最准确的因为供应商自己算的。但问题是很多供应商在流式响应里不返回usage或者只在最后一个chunk里返回。所以我们还需要一个本地的tokenizer作为兜底。import tiktoken def count_tokens(text: str, model: str) - int: 本地token计数作为供应商usage的兜底 try: encoding tiktoken.encoding_for_model(model) except KeyError: encoding tiktoken.get_encoding(cl100k_base) return len(encoding.encode(text)) def extract_usage_from_stream(chunks: list) - dict: 从流式响应的chunks里提取usage信息 usage {prompt_tokens: 0, completion_tokens: 0} for chunk in chunks: if usage in chunk and chunk[usage]: usage chunk[usage] break return usage这里有个细节不同供应商的tokenizer不一样同样的文本在不同模型下token数可能差20%以上。所以本地计数只能作为参考最终计费还是以供应商返回的usage为准。但如果供应商不返回usage我们就只能用本地计数这时候要在账单里标注“估算值”。3.3 配额与预算控制计数解决了“用了多少”的问题接下来要解决“谁能用多少”的问题。我们设计了一套基于配额Quota和预算Budget的双层控制体系。配额是硬限制比如A团队每月最多调用100万次超过就直接拒绝。预算是软限制比如A团队每月最多花5000块达到80%时发告警达到100%时降级到便宜模型或者直接拒绝。这套体系的实现依赖Redis做实时累加。每次请求完成后网关把消耗的token数和预估费用写入Redis按团队、按模型、按天/月维度分别累加。查询配额时直接从Redis读取当前值跟阈值比较。import redis from datetime import datetime r redis.Redis(hostlocalhost, port6379, db0) def check_quota(team_id: str, model: str, estimated_tokens: int) - bool: 检查团队配额是否充足 month_key fquota:{team_id}:{datetime.now().strftime(%Y%m)} used int(r.get(month_key) or 0) limit get_team_limit(team_id) if used estimated_tokens limit: return False return True def record_usage(team_id: str, model: str, prompt_tokens: int, completion_tokens: int): 记录实际用量 month_key fquota:{team_id}:{datetime.now().strftime(%Y%m)} total_tokens prompt_tokens completion_tokens r.incrby(month_key, total_tokens) # 按模型维度记录 model_key fusage:{team_id}:{model}:{datetime.now().strftime(%Y%m%d)} r.incrby(model_key, total_tokens) # 记录费用 cost calculate_cost(model, prompt_tokens, completion_tokens) cost_key fcost:{team_id}:{datetime.now().strftime(%Y%m)} r.incrbyfloat(cost_key, cost)这里有个坑预估值和实际值可能差很多。我们在请求前用本地tokenizer估算token数但实际输出长度完全不可控。有时候预估100个token实际输出了2000个。所以配额检查只能做粗粒度控制不能保证绝对精确。我们的做法是留20%的buffer比如配额是100万实际到80万就开始告警到95万就限流。3.4 计价体系的账单生成每个月1号我们会跑一个离线任务从Redis和日志里汇总上个月的用量生成每个团队的账单。账单里包含总调用次数、总token数输入/输出分开、按模型拆分的明细、总费用、环比变化。这个账单最开始是给财务看的后来业务团队也开始关注因为他们要控制自己的成本。再后来我们把它做成了一个自助查询的Dashboard团队负责人可以实时看到自己用了多少、还剩多少、哪个模型最费钱。实操心得计价体系一定要从第一天就做不要觉得“现在用量小以后再说”。等用量大了再补历史数据缺失根本算不清账。而且计价逻辑一旦上线就不要轻易改否则前后数据对不上财务会疯掉。4. 第三级演进智能路由与降级策略4.1 路由要解决什么问题有了计价体系之后我们开始思考一个更本质的问题每次请求真的需要用到最贵的模型吗实际情况是我们线上70%的请求都是简单任务——分类、抽取、格式转换、简单问答。这些任务用GPT-3.5或者更小的模型完全能搞定效果跟GPT-4差别不大但成本只有十分之一。剩下30%的复杂任务——长文推理、代码生成、多步规划——才真正需要大模型。所以路由的核心目标就是在保证效果的前提下把请求发给最便宜且够用的模型。4.2 路由策略的几种实现方式我们先后试过三种路由策略各有优劣。第一种是基于规则的路由。业务方在请求头里指定一个“任务类型”网关根据任务类型映射到模型。比如task_typeclassification走小模型task_typereasoning走大模型。这种方式简单直接但依赖业务方自觉标注实际执行中很多人图省事全标成reasoning路由就失效了。第二种是基于请求特征的路由。网关分析请求内容比如prompt长度、是否包含代码块、是否有多轮对话历史自动决定路由。比如prompt超过2000token的走大模型包含代码的走代码专用模型。这种方式不需要业务方配合但规则维护成本高而且准确率有限。第三种是基于模型置信度的路由。先用小模型跑一遍如果小模型的输出置信度低比如logprob低于阈值再升级到大模型。这种方式效果最好但延迟翻倍因为要跑两次。我们只在少数对成本极度敏感、对延迟不敏感的场景用了这种策略。最终我们落地的是规则路由为主、特征路由为辅的混合方案。业务方可以显式指定模型也可以指定任务类型让网关自动选。如果什么都不指定网关根据prompt特征做默认路由。def route_request(request: dict) - str: 路由决策函数 # 1. 业务方显式指定了模型直接使用 if request.get(model) and request[model] ! auto: return request[model] # 2. 根据任务类型路由 task_type request.get(task_type) if task_type: return TASK_MODEL_MAP.get(task_type, DEFAULT_MODEL) # 3. 根据prompt特征路由 prompt extract_prompt(request) prompt_tokens count_tokens(prompt, gpt-3.5-turbo) if prompt_tokens 2000: return gpt-4 # 长文本走大模型 elif contains_code(prompt): return code-model # 代码走专用模型 else: return gpt-3.5-turbo # 默认走小模型4.3 降级与重试策略路由解决的是“选哪个模型”降级解决的是“选的那个挂了怎么办”。我们的降级策略分三层。第一层是同供应商重试如果某个请求失败超时、限流、5xx错误先在同一供应商内重试最多2次每次间隔指数退避。第二层是跨供应商降级如果同供应商重试都失败了切换到备用供应商的同等模型。第三层是跨模型降级如果备用供应商也挂了降级到更小但更稳定的模型保证服务不中断。async def call_with_fallback(request: dict) - dict: 带降级的模型调用 primary_model route_request(request) fallback_chain get_fallback_chain(primary_model) last_error None for model in fallback_chain: for attempt in range(3): try: response await call_model(model, request) return response except RateLimitError: await asyncio.sleep(2 ** attempt) last_error rate_limit except TimeoutError: last_error timeout break # 超时直接换模型不重试 except ServerError: await asyncio.sleep(1) last_error server_error raise AllModelsFailedError(fAll models failed: {last_error})这里有个关键决策超时要不要重试。我们最开始是重试的后来发现超时往往意味着模型正在处理复杂请求重试只会加重负担而且用户已经等了很久了再重试体验更差。所以后来改成超时直接降级到更快的模型不重试。注意事项降级链不要设太长一般3层就够了。层数太多会导致故障时延迟急剧上升而且排查问题困难。另外降级后的模型输出质量可能下降要在响应头里标注实际使用的模型让业务方知道发生了降级。5. 第四级演进可观测性与治理能力5.1 可观测性的三个支柱网关跑稳定之后我们开始补可观测性。LLM网关的可观测性跟传统服务不太一样我把它拆成三个维度指标Metrics、日志Logs、追踪Traces。指标层面我们重点关注这几个QPS、P50/P95/P99延迟、错误率按错误类型分、token消耗速率、各模型调用占比、降级触发次数。这些指标用Prometheus采集Grafana展示。延迟指标特别重要因为LLM的延迟波动很大P99可能是P50的十倍以上只看平均值会严重误判。日志层面每次请求记录请求ID、团队ID、模型、prompt token数、completion token数、延迟、状态码、是否降级、实际使用的模型。日志用ELK收集支持按各种维度检索。这里要注意脱敏prompt和completion内容不能明文存储。追踪层面我们在网关里集成了OpenTelemetry每次请求生成一个tracespan包括网关接收、路由决策、供应商调用、流式传输、计量记录。这样当一个请求变慢时能快速定位是网关本身慢、还是供应商慢、还是网络慢。5.2 内容安全与合规治理LLM网关还有一个传统网关没有的职责内容安全。用户输入可能包含敏感信息模型输出可能产生不当内容这些都需要在网关层拦截。我们做了三层防护。第一层是输入过滤用正则和关键词库拦截明显的敏感词、身份证号、手机号等PII信息。第二层是输出审核对模型返回的内容做实时检测发现违规内容立即截断并返回安全提示。第三层是审计日志所有请求的元数据不包含内容保留180天满足合规审计要求。import re PII_PATTERNS { phone: r1[3-9]\d{9}, id_card: r\d{17}[\dXx], email: r[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,} } def mask_pii(text: str) - str: 对PII信息做掩码处理 for name, pattern in PII_PATTERNS.items(): text re.sub(pattern, f[{name}_masked], text) return text def check_content_safety(text: str) - tuple[bool, str]: 内容安全检查返回(是否安全, 原因) # 敏感词检测 for word in SENSITIVE_WORDS: if word in text: return False, fcontains_sensitive_word # 长度检查 if len(text) MAX_INPUT_LENGTH: return False, input_too_long return True, ok这里有个平衡问题安全策略太严会误伤正常请求太松又起不到防护作用。我们的做法是分级处理高风险内容直接拒绝中风险内容脱敏后放行低风险内容记录但不拦截。分级规则每季度review一次根据实际拦截情况调整。5.3 多租户隔离与权限管理企业级网关必须支持多租户。我们的租户模型是团队Team→ 应用App→ 密钥Key三级结构。每个团队有独立的配额和预算每个应用有独立的权限能用哪些模型每个密钥可以单独吊销。权限管理用的是RBAC模型。角色分为管理员、开发者、只读用户。管理员可以管理所有团队和配额开发者只能管理自己团队的密钥和查看自己的用量只读用户只能看Dashboard不能做任何修改。from enum import Enum class Role(Enum): ADMIN admin DEVELOPER developer VIEWER viewer class Permission(Enum): MANAGE_TEAM manage_team MANAGE_KEY manage_key VIEW_USAGE view_usage CALL_MODEL call_model ROLE_PERMISSIONS { Role.ADMIN: [Permission.MANAGE_TEAM, Permission.MANAGE_KEY, Permission.VIEW_USAGE, Permission.CALL_MODEL], Role.DEVELOPER: [Permission.MANAGE_KEY, Permission.VIEW_USAGE, Permission.CALL_MODEL], Role.VIEWER: [Permission.VIEW_USAGE] } def check_permission(role: Role, permission: Permission) - bool: return permission in ROLE_PERMISSIONS.get(role, [])多租户隔离还有一个容易忽略的点资源隔离。如果A团队的请求量突然暴增不能影响B团队的正常使用。我们在网关层做了基于团队的并发限制和队列隔离每个团队有独立的连接池和队列互不干扰。6. 第五级演进从网关到LLM治理平台6.1 网关的边界在哪里做到第四级的时候我们的网关已经能转发、能计价、能路由、能治理了。但新的需求还在不断涌来业务方想要A/B测试不同模型的效果想要缓存重复请求的结果想要对模型输出做自动评估想要管理prompt模板想要做模型微调的数据回流。这些问题如果都塞进网关网关会变得无比臃肿。所以我们开始思考网关的边界到底在哪里我们的结论是网关只负责请求路径上的实时决策——转发、路由、限流、计费、安全拦截。至于A/B测试、缓存、评估、prompt管理、数据回流这些属于离线或准实时的治理能力应该放在网关之外的治理平台里。6.2 治理平台的模块划分按照这个思路我们把LLM治理平台拆成了几个独立模块网关作为数据入口和策略执行点其他模块各司其职。模型注册中心管理所有可用模型的元数据模型名称、供应商、价格、上下文长度、能力标签是否支持函数调用、是否支持视觉、SLA等级。网关的路由决策依赖这些元数据。Prompt管理中心统一管理prompt模板支持版本控制、变量替换、A/B测试。业务方不再在代码里硬编码prompt而是引用模板ID。这样修改prompt不需要发版而且可以做效果对比。评估与实验平台负责模型效果的离线评估和在线A/B测试。离线评估用标注数据集跑分在线实验按流量比例分流对比不同模型的输出质量和成本。缓存层对重复请求做结果缓存。LLM请求的缓存命中率其实不低尤其是那些模板化的分类、抽取任务。我们用的是语义缓存不仅缓存完全相同的请求还缓存语义相似的请求。缓存命中时直接返回结果成本为零。数据回流管道收集高质量的请求-响应对用于后续的模型微调。不是所有数据都值得回流我们只回流那些经过人工确认或用户反馈为正例的数据。6.3 五级演进的整体架构回头看这五级演进其实是一个从“能用”到“好用”再到“管好”的过程。演进阶段核心能力解决的核心问题关键组件第一级统一转发API Key散落、无法观测反向代理、流式转发第二级计价配额成本不清、无法控制Token计数、Redis配额第三级智能路由成本高、可用性差路由引擎、降级链第四级可观测治理黑盒运行、合规风险监控、日志、安全过滤第五级治理平台能力碎片化、无法闭环模型注册、Prompt管理、评估、缓存每一级都不是必须的小团队可能停在第二级就够了大企业可能需要走到第五级。关键是根据实际痛点决定演进节奏不要为了架构而架构。6.4 自建还是采购最后聊一个实际问题LLM网关到底该自建还是买现成的我们的经验是如果只是统一转发和简单计价用开源方案或云厂商自带的网关就够了。但如果涉及到复杂的路由策略、多租户隔离、深度定制化的安全合规自建几乎是唯一选择。因为每个企业的组织架构、成本核算方式、安全要求都不一样通用产品很难完全匹配。自建的成本主要在人力。我们前后投入了大概两个工程师大半年的时间才把五级演进全部跑通。如果团队没有足够的工程能力建议先从第一级开始用最小成本验证价值再决定要不要继续投入。踩坑记录我们最开始想一步到位直接设计了一个包含所有功能的架构结果开发了三个月还没上线业务方等不及又回去裸调API了。后来改成小步快跑每两周发一个版本反而推进得更快。所以千万别想着一次做完演进式建设才是正道。7. 几个容易被忽略的实操细节7.1 流式响应的计量时机流式响应有个很坑的地方什么时候记录用量。如果等流结束再记录万一客户端中途断开了这次调用就漏记了。如果一开始就记录又不知道实际输出多少token。我们的做法是分段记录每收到一个chunk就累加token数定期比如每10个chunk写入Redis。如果流中断了已经记录的部分不会丢。流正常结束时再做一次最终校准以供应商返回的usage为准。7.2 模型名称的版本管理供应商的模型名称会变。今天叫gpt-4明天可能叫gpt-4-0613后天又出了gpt-4-turbo。如果业务方直接写死模型名称供应商一升级就全挂了。我们的方案是逻辑模型名 版本映射。业务方只用逻辑名比如chat-standard、chat-premium。网关维护一张映射表把逻辑名映射到具体的供应商模型ID。供应商升级时只需要改映射表业务方无感知。7.3 错误码的统一不同供应商的错误码完全不一样。A家的429表示限流B家的429可能表示余额不足。如果直接把供应商错误码透传给业务方业务方要适配多家苦不堪言。我们在网关层做了错误码归一化定义了一套统一的错误码体系RATE_LIMITED、QUOTA_EXCEEDED、MODEL_UNAVAILABLE、CONTENT_BLOCKED、TIMEOUT、INTERNAL_ERROR。网关把供应商错误码映射到统一错误码业务方只需要处理这一套。7.4 灰度发布的流量切分每次网关本身升级都担心影响线上流量。我们的做法是基于请求头的灰度内部测试流量带特定header路由到新版本网关外部流量默认走稳定版本。新版本跑一周没问题后再逐步把外部流量切过去。这个策略帮我们避免了好几次事故。有一次新版本网关的内存泄漏只影响了内部测试流量外部用户完全无感知。8. 写在最后LLM网关这个东西说复杂也复杂说简单也简单。核心就一句话在业务和大模型之间做一层可控的中间层。但“可控”两个字展开来就是计价、路由、治理、可观测性、安全合规这一大堆事情。我的建议是不要被这些名词吓到从最痛的那个点开始做。如果现在最痛的是API Key管理混乱那就先做统一转发。如果最痛的是成本失控那就先做计价。如果最痛的是模型经常挂那就先做降级。每解决一个痛点网关就演进一级团队的能力也跟着成长一级。我们团队从最开始三个人兼职维护到现在有专门的LLM基础设施小组走了差不多一年半。中间踩过的坑、熬过的夜、背过的故障现在回头看都是值得的。因为这套东西一旦跑起来整个公司的AI应用迭代速度会快很多——业务方不用再关心模型怎么调、Key怎么管、成本怎么算只管把业务逻辑写好就行。这大概就是基础设施的价值让上层的人感觉不到它的存在但它一直在那里撑着。
