多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

System One 判断下沉:开源 AI Agent 扛住并发的工程实践

System One 判断下沉:开源 AI Agent 扛住并发的工程实践 Jev 在开源社区爆火之后我看到很多讨论都集中在“轻量、可本地部署”这几件事上于是我做了一件很有意思的事把“System One 判断下沉”的思想直接搬进了自己维护的开源 AI Agent 项目里目标只有一个——让智能体在并发压力下依然稳定而不是每个请求都去烧大模型的 token也不是动辄好几秒才返回。所谓“System One 判断下沉”说白了就是给 Agent 的判断分个级高频、简单、明确的请求尽量在规则、缓存、轻量模型这一层就地解决只有真正复杂、模糊、需要多步推理的请求才交给大模型。这个思路其实不新但真正落到 Agent 工程里之后我发现收益远超预期。这篇内容会把整套架构怎么从零搭起来、哪些环节容易踩坑以及我是怎么靠它扛住并发压力完整交代一遍。如果你正在做 AI Agent、智能客服或聊天助手或者整天被“AI Agent 怎么扛并发”这个问题折磨那这篇可能正好适合你。下文涉及到的模块划分和参数调节都是我自己实际项目里的做法也参考了社区同类开源项目的常见实践。每个项目场景不一样建议当模板看按需替换。1. 为什么“判断下沉”能救 AI Agent1.1 大模型什么都管才是并发扛不住的根因我最早那版 Agent 非常“朴素”用户发来一句话我就把这句话连同历史记录一起丢给大模型让它自己去理解意图、决定要不要调工具、最后生成回复。这个流程在小流量下完全没问题一两个并发时体验相当好但流量一起来问题立刻全部暴露。问题出在“什么都要大模型做”。假设一个 Agent 同时有 100 个用户在线每人每 5 秒发一条消息其中大部分其实是高频问题比如“帮我查天气”“今天有什么新闻”“设个提醒”。这些请求根本没有复杂的推理需求但全都挤进了大模型的输入窗口。大模型每来一个请求都要重新解析、重新生成计算资源和 token 消耗呈线性上涨接口延迟很快拉满随后就是超时、报错。更要紧的是费用一天几十万次调用单看一次不贵合起来就是一笔不小的数字。打个比方这就相当于公司所有层级的人都跑来找 CEO问要不要开空调、要不要换打印机墨盒。CEO 的业务判断能力很强但流程全堵在他一个人身上公司肯定瘫痪。一个健康的组织需要前台、审批流和分诊机制把大多数问题挡在低层。Agent 也是一样的道理需要一层“快速判断”先接住大部分请求而不是所有流量都直通大模型。这也就是我在项目里引入 System One 的原因——它不是一个炫技的设计而是一个务实的分层结构。1.2 把双系统理论翻译成 Agent 架构“System One / System Two”原本来自心理学里的双系统理论System One 负责快速、直觉式的判断System Two 负责慢速、复杂的推理。大家应该都有体验看到前面有块石头你不会先做一段逻辑推导再绕开而是本能地就拐弯了这是 System One 在起作用。放到 Agent 工程里我是这么翻译的System One规则匹配、关键词、缓存、Embedding 距离计算、本地轻量模型。特点是快、便宜、稳定但泛化能力有限。System Two云端大模型或者配合 ReAct、Function Calling 的多步推理。理解能力强但慢、贵、并发能力天然不足。“判断下沉”这四个字的重点是把大量原本交给 System Two 的简单判断提前压缩到 System One 去完成。它不等于“让大模型变聪明”而是“让大模型少干活”。这不是牺牲体验换性能而是多数高频请求在快速路径上返回的结果本来就和大模型差不多用户却能体验到几十毫秒的响应。我在实际项目里做下沉之后效果非常直观System One 命中率到 80% 以后大模型调用量直接砍到原来的五分之一绝大多数请求秒回。剩下的 20% 复杂请求再交给大模型慢慢处理体验反而比以前更好因为资源全留给了真正需要它的场景。1.3 Jev 在 System One 里的真实工位Jev 爆火那阵子我的第一反应是“又一个能在本地跑的开源模型”。社区里关于它的讨论大多集中在嵌入式、轻量、聊天助手本地部署这些方向上感觉大家都很关心小资源能不能跑起来。我看了它的定位之后觉得它很适合当一个 Agent 的 System One 快思考模型——不是让它替代大模型而是让它在本地做意图识别和短回复生成。我在 System One 模块里封装了一个 Jev 本地推理客户端专门处理一件事判断“用户这句话是不是意图明确的常见请求”如果是就把它落到对应的模板或动作上。这里先说清楚我不打算展开 Jev 的模型结构和训练细节因为没必要它在这个体系里最重要的一点是能在本地跑、没有网络延迟、单次推理成本接近零。当然这套设计不绑定某个模型。你用任何能本地跑的轻量模型来做意图识别和分类效果都是等价的。我选 Jev 的理由很简单现成、够轻、社区讨论活跃非常适合开源 Agent 项目“到手就能跑”的调性。2. 双层判断架构实战先过 System One再上大模型2.1 核心链路一次请求的完整路由我把完整链路拆成了两个大层所有请求进来以后按以下方向走客户端 → 鉴权/参数校验 → System One 路由 →命中→ 本地动作/模板回复/缓存直接返回└→低置信或未命中→ System Two 大模型 → 返回结果并回填缓存System One 路由是整个设计的核心。它不负责生成复杂内容只负责一个判断这个请求要不要进 System Two。每进来一个请求路由会按顺序跑几层拦截器任何一层用足够高的置信度命中链路立刻结束。只有所有层都觉得“搞不定”时才把请求交给大模型。这里有一个很多人容易忽略的设计点System One 和 System Two 是独立扩展的。System One 是本地快速路径可以随意加并发System Two 走大模型用单独的并发池控制。两边的失败模式也不一样System One 失败可以立即降级System Two 失败则需要超时重试或熔断。把两者混在一个循环里管理是很多 Agent 项目并发一高就崩的重要原因。2.2 四类拦截器把高频请求挡在模型外面我在 System One 路由里放了四层拦截器按从快到慢的顺序排列第一层是硬规则匹配。高频且固定写法的请求比如“/start”“设置提醒”“查北京天气”这类用正则或字符串匹配就能解决。这一层不算智能但速度无敌几毫秒结束。规则需要人工维护但覆盖的是最容易被识别的那批请求维护成本完全可控。第二层是缓存复用。用户问过的问题如果之前大模型已经生成过回答就直接返回缓存。重点在于缓存键不能只放问题原文必须加用户维度和归一化处理否则会出现“A 用户查到的信息被 B 用户拿到”的串号事故。缓存命中时完全不需要额外计算是所有分支里响应最快的。第三层是 Embedding 意图识别。我会给每个意图准备几条种子问题用向量检索算相似度。比如“今天深圳天气怎么样”“上海下雨吗”都会归到 query_weather 这个意图上。这一步能覆盖自然表达的变体但它需要一个阈值兜底我的经验是相似度低于 0.75 就不要信。第四层才是本地 Jev 模型。它负责处理上面三层拿不准、但确实属于常见意图的请求。因为 Jev 这类模型能读懂更自然的表述很多规则和向量检索漏掉的请求它都能兜住而且它在本地跑单次推理耗时通常能压在几十到一百多毫秒。这四层下来绝大多数高频请求就都走了短路径。放在代码里每层返回结果时我都会标记来源这样线上就能很清楚看到流量到底集中在哪一层、哪些地方判断不准。判断下沉不是搭好就完事它是一个需要持续调优的过程而这四个拦截器就是整个调优的数据来源。2.3 什么时候必须放弃 System OneSystem One 很好用但它不是万能的。我还在项目里保留了一张“是否下沉”的决策表凡是涉及高风险操作的场景一律不走快速路径。场景是否走 System One原因高频标准问法是意图固定规则和缓存能覆盖需要知识库检索的事实问答部分检索结果可拼模板但生成部分仍需模型多轮对话中的重度上下文依赖否上下文一变固定模板容易答错代码生成、长文档总结否需要真正理解语义不能拍脑袋删除数据、转账、外发信息否强制复核宁可多走一次大模型也不能误判最后一行特别重要。判断下沉的目标是省钱省时间但在高风险操作上我宁愿让用户二次确认或者强制走 System Two 复核一遍。下沉带来的收益在安全事故面前完全不值一提。3. 把 System One 搬进开源 Agent我的实现记录3.1 项目目录与依赖宁可拆碎不要堆大我开源的这个 Agent 项目目录结构大致长这样agent_app/ ├── system_one/ │ ├── router.py # System One 路由主逻辑 │ ├── rules.py # 规则与正则匹配 │ ├── cache.py # 缓存封装 │ ├── vector.py # Embedding 意图召回 │ └── jev_client.py # 本地 Jev 轻模型客户端 ├── system_two/ │ ├── llm_client.py # 大模型调用封装 │ └── fallback.py # 超时与熔断处理 ├── api/ │ ├── main.py # FastAPI 入口 │ └── schemas.py # 请求/响应结构 └── config.yaml # 阈值、并发、缓存参数依赖方面基础的是 FastAPI、Redis缓存和限流、一个向量计算库再加上本地 Jev 的推理依赖。量不大时用 numpy 算 Embedding 距离就够了没必要上一套重型向量数据库。我没有把代码都塞进一个巨大文件而是按 System One / System Two 拆成小模块开源项目最怕别人到手跑不起来拆分之后读者可以只摘自己需要的那一块。3.2 System One 路由核心代码与解释System One 路由的核心逻辑其实不长核心大概长这样# system_one/router.py class SystemOneRouter: def __init__(self, rules, cache, vector, jev_client): self.rules rules self.cache cache self.vector vector self.jev_client jev_client async def decide(self, query: str, user_id: str): # Layer 1: 硬规则 hit self.rules.match(query) if hit: return {intent: hit.intent, params: hit.params, source: rule} # Layer 2: 缓存复用 cache_key fagent:{user_id}:{normalize(query)} cached await self.cache.get(cache_key) if cached: return {intent: cached[intent], params: cached[params], source: cache} # Layer 3: 向量召回 intent, score self.vector.search(query) if intent and score 0.75: return {intent: intent, params: extract_params(query), source: vector} # Layer 4: 本地模型兜底 if self.jev_client.available(): result await self.jev_client.classify(query) if result.confidence 0.8: return {intent: result.intent, params: result.params, source: jev} return {intent: None, params: None, source: None}有几个细节必须说明白。第一decide 方法本身是 async 的因为向量检索和本地模型推理都可能阻塞事件循环写成同步会拖死整个 FastAPI 服务。第二我给每一层判断加了 asyncio.wait_for 的超时控制单层超时直接跳过确保 System One 无论出什么问题都不会无限拖住请求。第三本地 Jev 模型这里我封装成返回 intent 和 confidence 的接口如果你用的模型不支持直接输出结构化标签可以用文本生成加解析或者换成任何轻量分类器完全等价。3.3 System Two 降级链路超时、重试、熔断System Two 的封装没有多花哨就是三件套超时、重试、熔断。参数我放在 config.yaml 里方便不同部署环境调整。system_two: llm_timeout: 20s max_retries: 2 circuit_breaker: true fallback_message: 当前请求人数较多请稍后再试熔断逻辑是如果最近一分钟内大模型超时率超过 30%直接把所有请求降级到 System One 的兜底话术不再调用大模型。这样在极端流量下整个 Agent 不会完全不可用最坏情况是回答质量下降而不是全站崩溃。我见过很多 Agent 项目一上线就挂绝大多数原因不是代码写得差而是完全没有降级概念把大模型当成了唯一可用资源。实际上大模型接口的可用性受太多因素影响没有熔断和降级的 Agent就是一颗随时会炸的雷。System One 和 System Two 的衔接还有一个原则只要置信度低于阈值就必须走 System Two不允许“勉强猜一个”。勉强猜错比直接告诉用户“没听懂”更糟糕因为前者会让用户觉得这个 Agent 很不可靠。3.4 如何平滑接入 LangGraph / LangChain很多人的 Agent 项目已经在用 LangGraph 或 LangChain这时候不需要推翻重来。System One 可以作为一个前置节点直接插进图里。我在 LangGraph 里的做法是入口节点不是 LLM而是 SystemOneRouter如果判断命中就直接走到对应的工具节点根本不会初始化 LLM 节点。只有 System One 未命中的情况下才进入 llm_planner 节点。这样做相当于在原本“所有请求都进模型”的流水线前面装了一道闸门。改动量不大但省下来的 token 和并发资源非常明显。这也是为什么我一直强调判断下沉是一种工程思想不是某个特定框架的插件。你用它去驾驭任何编排框架效果都一样。4. 并发与成本实战扛住流量还得省 token4.1 用一张表算清楚命中率决定成本和延迟我用一个简单的模型算过账假设每天有 10 万次请求大模型单次调用平均需要 5 秒、成本约 0.02 元System One 单次判断平均 50 毫秒成本几乎可以忽略因为它是本地模型加缓存。如果完全不做判断下沉10 万次全部进大模型一天成本约 2000 元平均延迟 5 秒。如果 System One 命中率达到 80%大模型只剩 2 万次成本约 400 元同时 80% 的请求延迟直接降到 50 毫秒。如果命中率能上到 95%成本只剩约 100 元95% 的请求都是毫秒级返回。这个账谁算完谁都会心动。命中率进大模型请求数/天单日调用成本平均响应体验0%100,000约 2000 元全部 5 秒级60%40,000约 800 元60% 毫秒级40% 秒级80%20,000约 400 元80% 毫秒级95%5,000约 100 元95% 毫秒级当然这个表用的是假设单价不同模型价格差异很大但趋势不会变越下沉越便宜越快。真正能在开源项目里吃到多少红利要看你的场景里高频问题占比有多高。如果业务本身全是开放性问题那下沉空间有限如果是客服、助手、企业内部工具这类偏固定的场景效果会非常显著。4.2 并发上不去先调这四处单靠判断下沉还不够工程上还需要配合几个手段才能真正扛住并发。第一接口整体走异步。FastAPI 本身是异步框架但你的 System One 和缓存客户端不能因为同步调用把事件循环卡死。我用的是 httpx 的 AsyncClient 和 Redis 的异步客户端任何一个阻塞调用放到事件循环里都会拖慢所有请求。第二连接池和连接复用要做对。不要每个请求都新建一次 Redis 连接或模型推理连接。我本地压测时发现连接频繁重建会让性能差出好几倍。连接池的大小要压测后定开得太大浪费内存太小又不够用。第三给本地 Jev 模型加一个并发队列。限制同时推理的数量否则模型推理本身就会把 CPU 打满反过来拖慢整条链路。我自己是把最大并发限制在 24超出的请求排队等待配合向量检索先顶住大部分流量。第四处理热 key 击穿。某些高频问题会瞬间集中打过来即便有缓存如果所有请求都在缓存失效瞬间去后端重新计算也会把小服务打爆。我用的是“单飞 互斥锁”同一个缓存键同时只允许一个请求去后端计算结果其他请求等结果返回后直接读缓存。实现不算复杂但对突发流量帮助特别大。4.3 8 核 16G 机器上的真实压测观察我在一台日常跑开源项目的 8 核 16G 机器上做过对比测试模拟 200 个并发用户每人持续 5 分钟、每分钟发两条消息。没有开 System One 的时候大模型接口平均延迟从正常的几秒一路飙升大量请求超时整站吞吐量掉到很低的水平。开了 System One 之后同样的测试流量80% 的请求只用几十毫秒返回整体稳定性和吞吐量明显高出一大截。这里我不贴精确数字因为结果很依赖硬件和模型版本怕有人拿我的数字套自己的机器导致误判。但趋势肯定是有效的System One 路由本身的计算很轻再加上缓存让 Agent 的大部分流量在进入大模型之前就被消化掉。你用的是普通开发机还是 GPU 服务器只影响压力上限不影响“下沉有效”这个结论。如果非要说一个参考值我个人的经验是加了 System One 之后同一台机器能承担的并发用户数至少翻倍延迟的中位数则会从秒级降到毫秒级。5. 常见问题与排查技巧实录5.1 命中率太高回复变机械了怎么办这问题很容易出现规则和模板调得过猛很多语义接近但微妙的请求被固定话术接住用户会觉得对面是个机器人。我踩过这个坑后来把阈值往回压了一点同时给同一个意图准备多套模板轮换着返回。另外如果用户连续两次对同一回答点“不喜欢”我就强制把这条链路标成 System Two人工参与纠偏。判断下沉的目标是高频率问题响应更准不是让智能体失去应变能力。你要在“快速正确”和“灵活自然”之间找一个平衡点这个平衡点几乎不能一次找对必须看线上数据反复调。5.2 缓存张冠李戴差点把用户数据发给别人缓存是最容易埋坑的地方。我早期做过一版全局缓存结果不同用户问“帮我设置提醒”后者直接拿到了前面那个人的时间和事项这属于典型的设计失误。缓存键必须带用户维度和上下文维度不能只放问题原文。更隐蔽的问题是动态信息。回复内容里如果包含时间、价格、库存这类变量就不能长时间原样缓存。我的做法是先把动态字段替换成占位符再缓存模板真正返回时再填充当前值或者直接给动态类回复设置很短的 TTL。宁可让缓存频繁失效也不能让用户拿到过期数据。5.3 本地模型把 CPU 打满我用队列解决了本地模型跑在 Agent 里听上去很爽但并发一旦上来推理的 CPU 占用会直接拉满。我最初把 Jev 模型当成主力判断器结果一台小机器很快就撑不住了。后来我把模型调用改成队列消费限制最大并发数 24超出部分排队等待。同时把最核心的意图识别先交给 Embedding 向量检索只有向量检索置信度不够时才调用模型。这样模型就从“主力”退成了“兜底”机器压力小了很多。如果还卡就换更小版本的量化模型或者在离线阶段把高频意图整理成静态分类表运行时直接查表把模型彻底从热路径上去掉。5.4 接口变慢按这个顺序排查接手任何 Agent 项目如果发现接口变慢按这个顺序排查最有效先看日志里请求走了 System One 还是 System Two有没有不小心绕过 System One 的分支再看链路里缓存的命中率和大模型调用次数最后看 CPU 和内存指标。我遇到过一种很诡异的情况缓存明明命中了但请求还是慢。查了很久才发现是向量检索里有一段 O(N^2) 的代码请求量一大每层判断都要多花几十毫秒。判断下沉的核心链路本来就短但你写得不仔细短路路径也能变慢路径。这里我分享一个小技巧在 System One 每次命中时往响应 header 里打一个 X-Cache-Source 字段值分别是 rule、cache、vector、jev、llm。这样你在压测和线上巡检时用浏览器开发者工具就能直观看到流量分布不需要额外接监控系统就能发现问题。这个技巧成本几乎为零但排查问题的时候非常好用。我这个项目做下来最大的体会是System One 判断下沉不是“为了不用大模型而不用”而是把每一个 token 和每一次推理都花在刀刃上。大模型是 Agent 的大脑但如果连“今天天气怎么样”都要让大脑做完整推理那它根本没精力处理真正困难的问题。开源 Agent 尤其如此大家资源都有限并发扛不住、成本炸了项目自然就没人用了。这套思想几乎不挑框架、不挑模型值得每个做 Agent 的人亲自试一遍。最后再分享一个小经验先别急着上完整版把你场景里最高频的 20 个问题用规则和缓存接住观察延迟和成本的变化你会自愿回来把整套判断下沉机制补齐的。
返回列表