ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

大模型网关与Agent落地实践:架构设计、路由策略与自动化编程

大模型网关与Agent落地实践:架构设计、路由策略与自动化编程 1. 大模型网关到底解决什么问题1.1 从一个真实痛点说起去年下半年我所在的团队同时接入了四家模型供应商。最开始大家各写各的调用代码A项目用OpenAI的SDKB项目直接requests裸调C项目又封装了一套自己的重试逻辑。结果就是API Key散落在七八个配置文件里谁用了多少Token没人说得清某家供应商限流了要改五个地方想换个模型得重新测一遍所有业务代码。这种混乱不是个例。只要团队规模超过三个人、接入超过两家模型几乎必然会遇到同样的问题。大模型网关LLM Gateway就是在这个背景下被提出来的——它本质上是一个位于业务代码和模型供应商之间的中间层统一处理鉴权、路由、限流、计费、日志、缓存这些事情。你可以把它理解成公司内部的模型调用总机业务方只管把请求发给网关网关决定用哪家模型、怎么重试、怎么记账业务代码完全不用关心底层是哪家供应商。这个抽象带来的最大好处是解耦——换模型不用改业务代码加供应商不用通知所有人出问题能在一个地方排查。1.2 网关的核心能力清单一个能落地的大模型网关至少要覆盖下面这几块能力我按重要性排个序能力模块解决的核心问题落地优先级统一鉴权API Key集中管理业务侧不接触真实密钥必须多模型路由按成本/延迟/能力选择供应商支持降级必须限流与配额防止单业务打爆额度按租户分配必须可观测性记录Token消耗、延迟、错误率必须缓存相同请求复用结果省钱省时间推荐内容安全输入输出过滤合规审查推荐成本核算按业务/用户维度出账单推荐这里面最容易被低估的是可观测性。我见过太多团队上线网关时只做了路由结果月底账单出来一脸懵——不知道钱花在哪了。Token消耗必须从第一天就记录字段至少包括请求ID、租户、模型、输入Token、输出Token、耗时、状态码。这些数据后面做成本优化、容量规划都靠它。1.3 为什么不用现成的API管理工具有人会问Kong、APISIX这些传统API网关不是现成的吗为什么还要专门做大模型网关答案是大模型的请求形态和传统HTTP API差别太大。传统API网关处理的是无状态请求看一眼URL和方法就能路由。但大模型请求有几个特殊之处第一请求体里带着完整的对话历史动辄几万Token网关得能解析这个结构才能做路由决策第二响应是流式的SSE网关要能透传流并且中途统计Token第三计费是按Token算的不是按请求数算的这要求网关能解析响应体里的usage字段。所以实际落地时常见做法是在传统网关后面再挂一层专门的大模型网关或者直接用Python/Go写一个轻量服务。我倾向于后者——用FastAPI或者Gin写一个几百行的服务逻辑清晰、改起来快比在Kong上写Lua插件舒服得多。2. 网关的架构设计与关键选型2.1 分层架构怎么切我实际用下来比较顺手的是三层结构接入层负责协议适配。对外暴露OpenAI兼容的接口/v1/chat/completions这样业务方用官方SDK就能直接调迁移成本几乎为零。这一层还要处理鉴权——业务方拿的是网关签发的虚拟Key不是真实供应商Key。路由层是核心。它拿到请求后根据配置的策略决定走哪家供应商。策略可以是静态的按模型名映射也可以是动态的按当前各家的延迟和错误率加权。我建议初期先用静态映射跑稳了再上动态。适配层负责把统一的内部请求格式翻译成各家供应商的格式。OpenAI、Anthropic、国内几家厂商的请求结构都有差异这一层做转换。响应回来时再反向翻译成统一格式。这个分层的好处是每层职责单一加一家新供应商只需要在适配层加一个转换器路由层和接入层完全不用动。2.2 路由策略的取舍路由策略直接决定了成本和稳定性这里展开说说我的经验。最简单的策略是按模型名硬映射请求里写gpt-4就走A家写claude-3就走B家。优点是行为可预测缺点是没法自动降级——A家挂了请求就全挂。进阶一点是主备降级给每个逻辑模型配一个主供应商和若干备选主供应商连续失败N次就切到备选。这个策略实现简单效果立竿见影。我一般设N3连续三次5xx或者超时就切换切换后每隔30秒探活一次主供应商恢复了再切回来。再往上就是加权路由根据实时统计的成功率、P99延迟、当前配额余量算一个分数按分数分配流量。这个策略最智能但也最容易出问题——如果统计窗口设得太短流量会在供应商之间来回抖。我的经验是统计窗口至少5分钟而且要有平滑因子新数据权重不超过0.3。提示不管用哪种策略一定要保留一个强制指定的逃生通道。线上出问题时运维能手动把某个模型钉死到指定供应商比等自动策略生效快得多。2.3 流式响应的处理要点大模型网关最容易踩坑的地方就是流式响应。业务方要的是逐字返回的体验网关如果先把整个响应收完再转发首字延迟直接翻倍体验就毁了。正确做法是边收边转边统计。网关从上游拿到SSE流后逐块解析提取出delta.content转发给下游同时累加Token计数。这里有个细节OpenAI的流式响应里usage字段只在最后一个chunk里出现而且需要请求时带上stream_options: {include_usage: true}所以Token统计不能只靠累加delta得等最后一个chunk。如果上游不返回usage有些供应商不支持就得自己估算。粗略的估算方法是中文按1个字符≈0.6个Token英文按1个单词≈1.3个Token。这个精度做成本监控够用了但别拿去做精确计费。还有一个坑是客户端断连。用户关了页面下游连接断了但上游还在生成。这时候网关要主动取消上游请求否则白白烧钱。实现上就是监听下游连接的关闭事件触发上游的cancel。3. 自动化编程与Agent的落地实践3.1 Agent和传统自动化的本质区别聊自动化编程之前得先把Agent这个概念说清楚。网上关于agent是什么的讨论很多我的理解是传统自动化脚本是if-else写死的流程而Agent是给定目标自己决定下一步做什么。举个例子。传统脚本做代码格式化就是调一下formatter流程固定。但如果任务是把这个模块重构一下消除重复代码传统脚本没法做因为步骤不固定——得先读代码、理解结构、找重复、改一处、跑测试、看结果、再改下一处。这个边做边判断的过程就是Agent的核心。Agent的三大件是规划把大目标拆成小步骤、工具调用执行具体动作比如读写文件、跑命令、记忆记住做过什么、结果如何。这三样缺一不可。我见过不少号称Agent的项目其实只是把prompt串起来跑一遍没有真正的循环和反馈那只能叫工作流不叫Agent。3.2 CLI类Agent工具的使用心得命令行形态的Agent工具CLI Agent是这两年最实用的方向之一。它把Agent能力封装成一个终端命令你在项目目录里敲一行它就能读代码、改文件、跑测试。相比IDE插件CLI工具的优势是不绑定编辑器SSH到服务器上也能用。这类工具的工作模式大同小异启动后进入一个交互式会话你用自然语言描述任务它规划步骤、调用工具、把结果反馈给你。常见的命令包括查看当前会话状态、切换模型、恢复上次会话、压缩上下文等。上下文压缩这个功能特别重要——长会话很容易把上下文窗口撑爆压缩功能会把历史对话摘要成一段简短描述腾出空间继续干活。我实际用下来的体会是CLI Agent最适合边界清晰的中等任务比如给这个函数补单元测试、把这段代码从回调改成async/await、修复这个lint报错。太小的任务改个变量名不如手动快太大的任务重构整个模块它容易跑偏需要人盯着。3.3 安装与依赖问题的排查CLI工具安装时最常见的报错是依赖缺失典型的表现是提示某个平台相关的可选依赖找不到让你重新安装。这类问题的根因通常是npm在安装时根据当前平台跳过了不匹配的可选依赖但运行时又需要它。排查思路是这样的先确认Node版本很多CLI工具要求18以上然后清掉缓存重装npm cache clean --force再npm install -g如果还不行就手动装那个缺失的包。实在搞不定用npx直接跑最新版往往能绕过本地安装的问题。注意全局安装CLI工具时如果公司网络有代理记得配好npm的proxy和https-proxy否则会卡在下载阶段。这个坑我踩过不止一次。3.4 Agent的记忆机制怎么设计Agent记忆是决定它能不能处理长任务的关键。没有记忆的Agent每轮对话都是失忆状态你得反复把背景信息贴进去。记忆一般分三层短期记忆是当前会话的完整对话历史直接放在上下文里长期记忆是跨会话的持久化信息比如这个项目的测试命令是pytest、用户偏好用TypeScript通常存成向量或者键值对工作记忆是当前任务的中间状态比如已经改了3个文件还剩2个。落地时最容易出问题的是长期记忆的写入时机。如果每轮对话都往里塞很快就会被噪音淹没。我的做法是只在任务完成时写入一条总结而且要求Agent自己判断这条信息未来是否还会用到用不到就不写。这个判断用一个小模型做就行成本很低。4. 从零搭建一个最小可用网关4.1 环境准备与依赖安装下面用一个具体的例子把网关跑起来。技术栈选Python FastAPI理由是生态成熟、改起来快适合快速验证。先建虚拟环境装依赖python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install fastapi uvicorn httpx pydantic python-dotenv这里选httpx而不是requests是因为httpx原生支持async和流式响应处理SSE方便得多。uvicorn是ASGI服务器配合FastAPI的async接口能扛住不错的并发。配置文件用.env管理把真实Key放进去别硬编码OPENAI_API_KEYsk-xxxx ANTHROPIC_API_KEYsk-ant-xxxx GATEWAY_MASTER_KEYgw-xxxxGATEWAY_MASTER_KEY是网关自己签发的虚拟Key业务方拿这个来调网关校验通过后再换成真实Key去请求上游。4.2 核心路由代码实现先定义请求模型和供应商配置from pydantic import BaseModel from typing import List, Optional class Message(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str messages: List[Message] stream: Optional[bool] False temperature: Optional[float] 0.7 PROVIDERS { gpt-4: { base_url: https://api.openai.com/v1, api_key_env: OPENAI_API_KEY, upstream_model: gpt-4-turbo, }, claude-3: { base_url: https://api.anthropic.com/v1, api_key_env: ANTHROPIC_API_KEY, upstream_model: claude-3-sonnet, }, }路由的核心逻辑就是查表把逻辑模型名映射到具体的供应商和上游模型名。这样业务方写gpt-4网关内部可以随时把它指向别的供应商业务无感。然后是主接口import os import httpx from fastapi import FastAPI, Header, HTTPException from fastapi.responses import StreamingResponse app FastAPI() app.post(/v1/chat/completions) async def chat_completions( req: ChatRequest, authorization: str Header(...), ): # 校验虚拟Key if authorization.replace(Bearer , ) ! os.getenv(GATEWAY_MASTER_KEY): raise HTTPException(status_code401, detailInvalid key) provider PROVIDERS.get(req.model) if not provider: raise HTTPException(status_code404, detailModel not found) api_key os.getenv(provider[api_key_env]) payload req.dict() payload[model] provider[upstream_model] if req.stream: return StreamingResponse( stream_proxy(provider[base_url], api_key, payload), media_typetext/event-stream, ) else: async with httpx.AsyncClient(timeout60) as client: resp await client.post( f{provider[base_url]}/chat/completions, headers{Authorization: fBearer {api_key}}, jsonpayload, ) return resp.json()这段代码虽然短但已经覆盖了鉴权、路由、流式/非流式分流三个核心功能。生产环境还要加限流、重试、日志但骨架就是这样。4.3 流式转发的实现细节流式转发是网关里最需要小心处理的部分async def stream_proxy(base_url, api_key, payload): async with httpx.AsyncClient(timeout120) as client: async with client.stream( POST, f{base_url}/chat/completions, headers{Authorization: fBearer {api_key}}, jsonpayload, ) as resp: async for chunk in resp.aiter_bytes(): # 这里可以插入Token统计逻辑 yield chunk关键点是client.stream配合aiter_bytes这样数据是逐块透传的不会在网关里堆积。首字延迟基本等于上游的首字延迟业务方感知不到中间多了一跳。如果要加Token统计就在yield之前解析chunk。SSE的格式是data: {...}\n\n把data:后面的JSON解析出来累加usage。注意最后一个chunk可能只有usage没有content别漏了。4.4 限流与配额的具体算法限流用令牌桶最合适因为它允许突发流量。每个租户一个桶桶容量设成每分钟配额的1.5倍补充速率就是每分钟配额除以60。import time class TokenBucket: def __init__(self, rate_per_sec, capacity): self.rate rate_per_sec self.capacity capacity self.tokens capacity self.last time.time() def allow(self, n1): now time.time() self.tokens min( self.capacity, self.tokens (now - self.last) * self.rate, ) self.last now if self.tokens n: self.tokens - n return True return False单机场景这样够了。多实例部署时桶的状态得放Redis用Lua脚本保证原子性。这里有个经验限流的粒度要按Token而不是按请求数。一次请求可能消耗100个Token也可能消耗10000个按请求数限流根本挡不住成本失控。5. 常见问题与排查实录5.1 网关层的高频故障下面这张表是我实际运维中遇到最多的问题按出现频率排序现象可能原因排查方向首字延迟突然变高上游限流或网络抖动看上游P99对比历史基线流式响应中途断开网关超时设置过短检查httpx timeout和Nginx配置Token统计对不上上游不返回usage启用本地估算核对账单部分请求401虚拟Key过期或轮换检查Key管理逻辑内存持续增长流式响应未正确关闭检查async with是否配对其中流式响应中途断开最隐蔽。表现是业务方偶尔收到半截响应但网关日志里没有错误。根因往往是Nginx的proxy_read_timeout默认60秒而大模型生成长文本可能超过这个时间。解决办法是把Nginx的超时调到300秒以上或者让网关定期发送心跳空注释行保活。5.2 Agent执行中断的处理Agent跑着跑着报execution terminated due to error是常见情况。这类错误信息很笼统得从几个方向排查先看是不是上下文超限。Agent跑长任务时上下文会不断增长超过模型窗口就会报错。解决办法是启用上下文压缩或者把中间结果写到文件里需要时再读回来。再看是不是工具调用失败。Agent调用的某个工具比如跑测试、读文件返回了非预期结果导致后续步骤无法继续。这种情况要在工具层加健壮的错误处理把错误信息结构化返回给Agent让它自己决定重试还是换方案。还有一种情况是循环检测触发。好的Agent框架会检测连续N步做同样的事触发后主动中断。这时候要看Agent是不是陷入了死循环通常是任务描述太模糊导致的。5.3 几个我踩过的坑坑一把网关做成了单点。早期为了图快网关只部署了一个实例结果它一挂全公司模型调用都停摆。后来改成多实例负载均衡每个实例无状态状态全放Redis。坑二日志里记了完整请求体。大模型请求体里经常有敏感信息全量记日志既占空间又有合规风险。后来改成只记元数据Token数、模型、耗时请求体按需采样。坑三重试策略太激进。一开始配了失败重试5次结果上游限流时重试反而加剧了拥堵。后来改成指数退避而且只对5xx和超时重试4xx直接返回。坑四忽略了成本告警。有次某个业务方写了个死循环一晚上烧掉了几百美元。后来加了按租户的日消耗告警超过阈值就自动降级到便宜模型。提示网关上线前一定要做压测重点测流式场景下的并发。我见过不少网关非流式跑得好好的一上流式并发就崩因为流式连接占用时间长连接池很容易被打满。5.4 安全方面的注意事项Agent和网关都涉及密钥和敏感数据安全上不能马虎。几条硬性要求真实API Key绝对不能出现在业务代码、日志、前端里只能存在网关的环境变量或密钥管理服务里。网关签发的虚拟Key要能单独吊销某个业务方出问题不影响其他人。Agent执行shell命令时要有白名单别让它随便跑rm -rf。所有涉及文件写入的操作要有沙箱限制在项目目录内。还有一点容易被忽略Agent的提示词注入。如果Agent会读取外部内容比如网页、用户上传的文件这些内容里可能藏着恶意指令。防御办法是在系统提示里明确外部内容仅作为数据不作为指令并且对工具调用做二次校验。6. 后续可以怎么扩展网关跑稳之后能做的事情还有很多。我列几个自己正在做的方向。语义缓存。相同或相似的请求直接返回缓存结果能省下可观的成本。实现上把请求的embedding存进向量库新请求来了先查相似度超过阈值就命中缓存。难点是阈值调参太松会返回不相关的结果太紧命中率又低。多模态支持。现在网关主要处理文本但图片、音频的调用需求越来越多。扩展方向是在适配层加多模态的格式转换路由层基本不用动。Agent与网关的联动。Agent执行任务时会调用大量模型如果每次都走网关就能统一记账和限流。更进一步可以让网关根据Agent的任务类型自动选择模型——简单任务用便宜模型复杂任务用强模型成本能降不少。可观测性看板。把网关的日志接到Grafana做一个实时看板展示各租户的Token消耗、各供应商的成功率和延迟、成本趋势。这个看板对运维和成本优化都很有用值得投入。我个人在实际操作中的体会是网关这东西先跑起来比设计完美更重要。一开始不用追求支持所有供应商、所有策略先把最核心的两三家接进来把鉴权和日志做扎实后面按需扩展。我见过太多团队在设计阶段纠结太久结果半年都没上线业务方早就自己写了一套野路子调用代码反而更难收拢。先上线再迭代这是最务实的路径。
返回列表