1. 从零理解多引擎同步优化 Agent 智能系统1.1 这套系统到底在解决什么问题先说说我为什么会对“多引擎同步优化 Agent”这个方向这么感兴趣。去年我接手了一个内部知识助手的项目单 Agent 跑起来挺顺可一旦把任务拆成“检索、推理、校验、执行”四段让不同模型分别负责问题就来了检索引擎返回的结果和推理引擎的上下文对不上校验环节拿到的又是过期状态整个链路像四个人各说各话。这就是单引擎 Agent 的天花板——它只能在一个模型、一套上下文里打转遇到需要多能力协作的场景就露怯。多引擎同步优化 Agent 智能系统本质上是把“一个大脑干所有事”改成“多个专职引擎协同干一件事”并且让这些引擎在同一个时间线上保持状态同步。这里的“引擎”可以理解为不同的大模型、不同的工具执行器、不同的检索后端甚至是不同语言写的推理服务。同步优化则是指当某个引擎的输出发生变化时其他引擎能及时感知并调整自己的策略而不是各跑各的。它适合谁如果你已经写过单 Agent 的 demo想往生产级多智能体协同走这篇就是给你准备的。如果你还在纠结“agent 是什么”也没关系我会把每个概念拆开讲保证你能跟上。整套东西落地之后你能拿它做智能客服编排、自动化数据分析流水线、多角色内容生产甚至复杂任务的分解执行。1.2 为什么是“多引擎”而不是“多模型”很多人一上来就把多引擎等同于多模型其实不是。多模型只是换了几个不同的 LLM而多引擎强调的是能力维度的拆分。我举个实际例子一个电商售后 Agent需要理解用户情绪情感引擎、查询订单数据库引擎、生成回复生成引擎、判断是否升级人工决策引擎。这四个引擎里只有生成引擎是纯 LLM其他三个可能是规则引擎、SQL 执行器、分类模型。这么拆的好处很直接每个引擎可以独立优化、独立扩容、独立替换。情感引擎换成更准的模型不影响订单查询数据库引擎加个缓存生成引擎完全无感。这就是“同步优化”的价值——优化是局部的收益是全局的。反过来如果你把所有逻辑塞进一个超大 prompt 里让一个模型干调试的时候你根本不知道是哪段逻辑出了问题改一处又怕影响另一处。我踩过这个坑一个 3000 字的 prompt 改了三天最后发现是中间一句指令和结尾的格式要求冲突了。1.3 整体架构的选型思路架构上我推荐“中心编排 引擎注册 共享状态”的三层结构。中心编排负责决定任务怎么流转引擎注册负责管理每个引擎的能力描述和健康状态共享状态则是一个所有引擎都能读写的上下文存储。为什么不用纯去中心化的多智能体因为去中心化在调试和可观测性上太痛苦了。我试过让几个 Agent 互相发消息协商结果一个任务卡住我花了两个小时才定位到是两个 Agent 对“完成”的定义不一致。中心编排虽然听起来不够“智能”但它把控制流显式化了出问题能一眼看到卡在哪。共享状态我建议用带版本号的键值存储而不是简单的内存字典。因为多引擎并发读写时没有版本控制就会出现“后写的覆盖先写的”这种经典问题。每次引擎更新状态时带上版本号读取时校验版本冲突了就重试或走补偿逻辑。2. 核心引擎的拆解与同步机制设计2.1 编排引擎任务流转的大脑编排引擎是整个系统的心脏它不直接干活只负责“下一步该谁干”。我习惯把它设计成一个状态机每个状态对应一个引擎的调用状态之间的转移条件由引擎的输出决定。举个具体例子一个内容审核 Agent 的编排逻辑是这样的先调分类引擎判断内容类型如果是文本就走文本审核引擎如果是图片就走图像审核引擎审核不通过就调通知引擎通过就调归档引擎。每个转移条件都写得很明确比如“分类引擎返回 text 且置信度大于 0.8”。这里有个关键设计点转移条件要可配置不要硬编码。我早期把条件写死在代码里后来业务方要加一个“置信度低于 0.5 时转人工”的分支我改了半小时代码。后来改成用配置表描述状态转移业务方自己就能加分支。# 状态转移配置示例 transitions [ {from: classify, to: text_review, condition: typetext and confidence0.8}, {from: classify, to: image_review, condition: typeimage and confidence0.8}, {from: classify, to: human_review, condition: confidence0.5}, ]2.2 推理引擎多模型的路由与降级推理引擎负责调用具体的 LLM但它不是简单地把请求转发出去。它要做三件事模型路由、结果聚合、失败降级。模型路由是根据任务类型选最合适的模型。比如代码生成走代码专精模型创意写作走通用大模型简单分类走小模型省钱。我一般会维护一张路由表记录每个模型擅长的任务类型和当前负载。结果聚合是当多个模型同时处理一个任务时怎么把结果合并。常见做法是投票、加权平均、或者让一个“裁判模型”来选。我实测下来对于事实性问题投票法最稳对于开放性问题裁判模型效果更好但成本高。失败降级是当主模型超时或报错时自动切到备用模型。这里要注意降级不是简单换个模型重试而是要把已经产生的中间结果带上否则备用模型得从头再来。我踩过的坑是降级后上下文丢了导致备用模型给出完全不同的答案。2.3 工具引擎外部能力的统一封装工具引擎把数据库查询、API 调用、文件操作这些外部能力封装成统一的接口让推理引擎不用关心底层实现。每个工具都要有清晰的输入输出 schema以及超时和重试策略。我建议每个工具都实现一个health_check方法编排引擎在调用前先检查工具是否可用。有一次我们的数据库工具挂了但编排引擎不知道一直往里发请求结果整个链路卡死。加了健康检查之后工具不可用就直接走降级分支用户体验好很多。工具的输入输出要用 JSON Schema 严格约束。我见过因为工具返回格式不固定导致推理引擎解析失败的案例。比如有的工具返回{result: ok}有的返回{status: ok}推理引擎就得写一堆兼容逻辑。统一 schema 之后这类问题基本消失。2.4 同步机制让引擎们步调一致同步是多引擎系统最难的部分。我把它分成三种同步状态同步、时序同步、语义同步。状态同步是保证所有引擎看到同一份上下文。做法是共享状态存储加版本号每次读写都校验版本。时序同步是保证引擎调用的顺序符合预期比如必须先检索再推理。语义同步最微妙是保证不同引擎对同一个概念的理解一致比如“高优先级”在检索引擎里是分数大于 0.9在决策引擎里也得是同一个阈值。语义同步我推荐用一份共享的“概念字典”所有引擎的阈值、枚举值都从这里取。这样改一个阈值所有引擎同步生效。我早期每个引擎各写各的阈值结果检索引擎认为 0.8 是高决策引擎认为 0.9 才是高中间那 0.1 的区间就成了黑洞。3. 从零搭建完整实操流程与关键代码3.1 环境准备与依赖选型先说环境。Python 3.10 以上是必须的因为要用到一些新的类型语法。核心依赖我推荐这几个fastapi做引擎的 HTTP 接口pydantic做数据校验redis做共享状态存储httpx做异步 HTTP 调用。为什么用 Redis 而不是内存字典因为多引擎可能部署在不同进程甚至不同机器上内存字典没法共享。Redis 的原子操作和过期机制正好适合做状态存储。如果你不想引入 Redis也可以用 SQLite 加文件锁但并发性能会差一些。pip install fastapi uvicorn pydantic redis httpx模型调用方面我建议抽象一层LLMClient把不同厂商的 API 差异屏蔽掉。这样换模型只需要改配置不用动业务代码。3.2 定义引擎接口与注册中心每个引擎都要实现统一的接口我定义了两个核心方法execute和describe。execute是干活的方法describe返回引擎的能力描述供编排引擎做路由决策。from abc import ABC, abstractmethod from pydantic import BaseModel class EngineInput(BaseModel): task_id: str context: dict params: dict class EngineOutput(BaseModel): task_id: str result: dict confidence: float next_hint: str | None None class BaseEngine(ABC): abstractmethod async def execute(self, input: EngineInput) - EngineOutput: pass abstractmethod def describe(self) - dict: pass注册中心是一个简单的字典记录引擎名称到实例的映射。启动时把所有引擎注册进去编排引擎通过名称查找。class EngineRegistry: def __init__(self): self._engines {} def register(self, name: str, engine: BaseEngine): self._engines[name] engine def get(self, name: str) - BaseEngine: return self._engines.get(name)这里有个细节注册中心要支持热更新。我遇到过线上要临时禁用某个引擎的情况如果注册中心不支持动态修改就得重启服务。加一个unregister方法配合配置中心就能做到不重启切换引擎。3.3 编排引擎的实现细节编排引擎的核心是一个循环读当前状态决定下一步调哪个引擎调用更新状态直到到达终止状态。class Orchestrator: def __init__(self, registry: EngineRegistry, state_store): self.registry registry self.state_store state_store async def run(self, task_id: str, initial_state: dict): state initial_state while state.get(status) ! done: engine_name self.decide_next(state) if engine_name is None: state[status] done break engine self.registry.get(engine_name) input_data EngineInput( task_idtask_id, contextstate, paramsstate.get(params, {}) ) output await engine.execute(input_data) state self.merge_state(state, output) self.state_store.save(task_id, state) return statedecide_next是编排逻辑的核心它根据当前状态和转移配置决定下一步。我建议把转移配置存在数据库或配置中心这样改流程不用发版。merge_state要小心处理冲突。如果两个引擎同时更新了同一个字段得有明确的合并策略。我的做法是给每个字段标记来源引擎冲突时按优先级取。3.4 共享状态存储与版本控制共享状态我用 Redis 的 Hash 结构存每个任务一个 key字段是状态项。版本号单独存一个字段每次更新时用WATCH加事务保证原子性。import redis import json class StateStore: def __init__(self, redis_client): self.redis redis_client def save(self, task_id: str, state: dict): key ftask:{task_id} version state.get(_version, 0) 1 state[_version] version self.redis.hset(key, mapping{data: json.dumps(state)}) def load(self, task_id: str) - dict: key ftask:{task_id} data self.redis.hget(key, data) return json.loads(data) if data else {}版本控制的关键是读取时拿到版本号写入时校验版本号是否变化。如果变了说明有其他引擎改过当前写入要重试。这个机制能避免大部分并发冲突。3.5 多引擎协同的完整调用链把上面几块拼起来一个完整的调用链是这样的用户请求进来编排引擎初始化状态然后按配置依次调用分类引擎、检索引擎、推理引擎、校验引擎每个引擎的输出都合并进共享状态最后返回结果。我拿一个实际的内容生成任务举例。用户输入一个主题系统先调检索引擎找相关资料再调推理引擎生成初稿然后调校验引擎检查事实性不通过就回到推理引擎重写通过就调格式化引擎输出。整个链路里检索引擎和推理引擎通过共享状态传递资料校验引擎的反馈也写进状态供推理引擎读取。这里要注意重试要有次数上限。我见过校验一直不通过、推理一直重写、最后 token 烧光的案例。加一个最大重试次数超过就走降级分支比如返回初稿加人工审核标记。4. 常见问题排查与性能优化实录4.1 引擎调用超时与级联失败多引擎系统最怕级联失败一个引擎慢了拖垮整个链路。我遇到过检索引擎响应从 200ms 涨到 5s导致后面所有引擎都在等最后整个请求超时。解决办法是给每个引擎设置独立的超时时间并且超时后走降级分支而不是直接失败。检索超时就返回空结果让推理引擎基于已有知识生成推理超时就返回缓存结果或简化版。import asyncio async def call_with_timeout(engine, input_data, timeout3.0): try: return await asyncio.wait_for(engine.execute(input_data), timeouttimeout) except asyncio.TimeoutError: return EngineOutput( task_idinput_data.task_id, result{error: timeout}, confidence0.0, next_hintdegrade )超时时间怎么定我的经验是取 P99 延迟的 1.5 倍。先跑一段时间收集延迟数据再根据数据调整。不要拍脑袋定 1 秒或 10 秒。4.2 状态不一致的排查思路状态不一致的表现是引擎 A 说任务完成了引擎 B 说还在处理。排查的时候先看共享状态的版本号如果版本号跳跃很大说明有并发写入冲突。我整理了一个排查清单现象可能原因排查方法引擎读到旧状态缓存未失效检查状态存储的 TTL 和刷新策略状态字段丢失合并逻辑覆盖打印每次合并前后的状态 diff版本号不连续并发写入检查是否有引擎绕过 StateStore 直接改状态任务卡住不推进转移条件不满足打印当前状态和所有转移条件的求值结果我踩过最坑的一次是某个引擎在异常分支里直接改了内存里的状态对象没走 StateStore导致其他引擎读到的还是旧状态。后来强制所有状态修改必须走 StateStore问题就没了。4.3 并发场景下的性能瓶颈多引擎系统天然是并发的性能瓶颈通常出现在三个地方模型调用、状态存储、编排循环。模型调用是最大的瓶颈因为 LLM 本身延迟就高。优化手段是并行调用无依赖的引擎。比如检索和情感分析没有依赖关系可以同时跑。我用asyncio.gather把这两个引擎并行整体延迟降了 40%。状态存储的瓶颈在 Redis 的网络往返。优化手段是批量读写把多次hget合并成一次hmget。另外状态不要存太大的对象超过 1MB 的字段考虑单独存对象存储。编排循环的瓶颈在decide_next的求值。如果转移条件很多每次求值都要遍历一遍。优化手段是把条件编译成可复用的表达式或者用决策树预计算。4.4 引擎健康检查与自动摘除生产环境里引擎会时不时挂掉。如果没有健康检查编排引擎会一直往挂掉的引擎发请求浪费时间和资源。我的做法是每个引擎暴露一个/health接口编排引擎定期探测。连续三次失败就把引擎标记为不可用路由时跳过。恢复后再自动加回来。class HealthChecker: def __init__(self, registry, interval30): self.registry registry self.interval interval self.failures {} async def check_all(self): for name, engine in self.registry.all().items(): try: await asyncio.wait_for(engine.health_check(), timeout2.0) self.failures[name] 0 except Exception: self.failures[name] self.failures.get(name, 0) 1 if self.failures[name] 3: self.registry.mark_unavailable(name)健康检查的频率别太高30 秒一次够了。太频繁会给引擎增加不必要的负载尤其是引擎本身就在处理重任务的时候。4.5 日志与可观测性建设多引擎系统的调试难度比单 Agent 高一个数量级因为问题可能出在任何一个引擎也可能是引擎之间的交互。没有好的日志排查就是大海捞针。我建议每个引擎的输入输出都打日志带上task_id和engine_name。这样你可以按task_id把整个链路的日志串起来看。日志级别用 INFO 记录正常流转ERROR 记录异常DEBUG 记录详细状态。另外给每个引擎加指标上报调用次数、成功率、平均延迟、P99 延迟。这些指标用 Prometheus 收集Grafana 展示。我靠这套指标发现过一个引擎的 P99 延迟是平均延迟的 20 倍一查是某个特定输入触发了慢查询。5. 多智能体协同的进阶玩法5.1 从编排式协同到协商式协同前面讲的都是编排式协同中心编排引擎决定一切。进阶玩法是协商式协同引擎之间可以互相发消息、协商任务分配。协商式协同适合任务边界不清晰的场景。比如一个研究助手检索引擎找到资料后可能觉得需要更多背景就主动请求推理引擎先分析一下再继续检索。这种动态协商在编排式里很难表达因为编排逻辑是预先定义的。实现协商式协同的关键是消息总线。每个引擎订阅自己关心的消息类型收到消息后决定是否响应。消息总线可以用 Redis 的 Pub/Sub 或者更专业的消息队列。不过我要提醒一句协商式协同的调试难度很高因为消息流是动态的很难复现。我建议先用编排式把业务跑通确实遇到编排表达不了的场景再上协商式。5.2 引擎能力的动态发现与组合当引擎数量多起来之后手动配置路由就不现实了。这时候需要引擎能力的动态发现每个引擎启动时把自己的能力描述注册到注册中心编排引擎根据任务需求自动匹配引擎。能力描述我建议用结构化的 schema包括输入类型、输出类型、擅长任务、成本、延迟。编排引擎根据这些信息做匹配和排序。{ name: code_review_engine, input_types: [code_diff], output_types: [review_comment], skills: [security_check, style_check], cost_per_call: 0.002, avg_latency_ms: 800 }动态组合的难点在于多个引擎都能干同一件事时怎么选我的策略是综合成本、延迟、历史成功率算一个分数选分数最高的。历史成功率很重要有些引擎虽然便宜但经常出错算下来反而更贵。5.3 记忆机制让 Agent 记住上下文多智能体系统如果没有记忆每次任务都从零开始效率很低。记忆机制分短期记忆和长期记忆。短期记忆就是当前任务的共享状态任务结束就清掉。长期记忆是跨任务的知识比如用户偏好、历史决策、常见问题的解决方案。长期记忆我建议用向量数据库存按语义检索。每次任务开始时先检索相关记忆注入共享状态。任务结束时把有价值的结论写回长期记忆。这里要注意记忆的时效性。有些记忆会过期比如用户三个月前的偏好可能已经变了。给记忆加时间戳检索时按时间衰减加权。5.4 安全边界与权限控制多引擎系统里不同引擎的权限应该不同。检索引擎只能读数据执行引擎才能写数据。如果权限不隔离一个被攻破的引擎可能影响整个系统。我的做法是给每个引擎分配一个角色角色决定它能访问哪些资源。编排引擎在调用引擎时把角色信息一起传过去引擎在执行前校验权限。另外引擎之间的通信要加密和鉴权。即使是内网也不能假设绝对安全。用 mTLS 做双向认证每个引擎有自己的证书。还有一点容易被忽略引擎的输出要过滤。推理引擎可能生成包含敏感信息的内容直接返回给用户就出问题了。加一层输出过滤引擎检查并脱敏。6. 我踩过的坑与实战经验总结6.1 不要过早追求“全自动”我刚开始做多引擎系统时一心想着全自动所有决策都让系统自己做。结果上线后发现很多边界情况系统处理不了用户投诉不断。后来我改成“人机协同”系统处理常规情况遇到低置信度或异常情况就转人工。人工处理的结果又反馈给系统用于优化模型和规则。这样系统越用越准用户满意度也上来了。所以我的建议是先做半自动把人工介入的环节设计好等系统稳定了再逐步减少人工。不要一上来就追求全自动那是给自己挖坑。6.2 引擎粒度不是越细越好我一度把引擎拆得很细一个功能一个引擎结果引擎数量爆炸编排逻辑复杂得没法维护。后来我合并了一些引擎把关联性强的功能放在一起系统反而更稳定。引擎粒度的原则是高内聚、低耦合。经常一起调用的功能放一个引擎需要独立扩缩容的功能拆开。不要为了“看起来架构清晰”而过度拆分。6.3 测试要覆盖引擎组合场景单引擎测试好写多引擎组合测试难写。我吃过亏每个引擎单独测都通过组合起来就出问题因为引擎之间的交互没测到。我的做法是写组合测试用例覆盖常见的引擎调用序列。比如“检索-推理-校验”这条链路要测检索返回空、推理超时、校验不通过等各种分支。这些用例跑起来慢但能发现大部分集成问题。另外用 mock 引擎做单元测试用真实引擎做集成测试。单元测试跑得快集成测试跑得慢但更真实。两者都要有。6.4 成本控制是长期课题多引擎系统调用多个模型成本比单 Agent 高。如果不控制账单会很吓人。我见过一个月烧掉几万块的案例。控制成本的手段有几个小模型干简单活大模型干复杂活缓存常见请求的结果设置每个任务的 token 上限监控成本指标异常时告警。我还会定期分析成本构成看哪个引擎最烧钱然后针对性优化。有一次发现检索引擎的 embedding 调用占了 60% 成本换成更便宜的 embedding 模型后成本降了一半效果几乎没变。6.5 版本管理与灰度发布多引擎系统里每个引擎都可能独立迭代。如果没有版本管理新版本引擎和旧版本编排逻辑不兼容就会出问题。我的做法是给每个引擎的接口加版本号编排引擎按版本号调用。新版本引擎先灰度发布只让少量流量走观察没问题再全量。灰度发布的关键是能快速回滚。我一般保留最近三个版本的引擎出问题一键切回旧版本。回滚时间控制在 1 分钟内这样即使出问题影响也有限。7. 后续可以这样扩展这套系统跑通之后扩展方向很多。我目前在做的是把引擎部署到边缘节点让靠近用户的请求在本地处理降低延迟。另一个方向是引入强化学习让编排引擎根据历史反馈自动优化路由策略而不是靠人工配置。还有一个有意思的方向是引擎的自动生成给定一个任务描述自动生成对应的引擎代码和编排逻辑。这个还在探索阶段但已经能看到一些雏形。等成熟了再单独写一篇分享。如果你也在做多引擎 Agent欢迎交流。我踩过的坑可能你也会遇到提前知道能省不少时间。 多引擎同步优化Agent智能系统：从零搭建到生产级实战
1. 从零理解多引擎同步优化 Agent 智能系统1.1 这套系统到底在解决什么问题先说说我为什么会对“多引擎同步优化 Agent”这个方向这么感兴趣。去年我接手了一个内部知识助手的项目单 Agent 跑起来挺顺可一旦把任务拆成“检索、推理、校验、执行”四段让不同模型分别负责问题就来了检索引擎返回的结果和推理引擎的上下文对不上校验环节拿到的又是过期状态整个链路像四个人各说各话。这就是单引擎 Agent 的天花板——它只能在一个模型、一套上下文里打转遇到需要多能力协作的场景就露怯。多引擎同步优化 Agent 智能系统本质上是把“一个大脑干所有事”改成“多个专职引擎协同干一件事”并且让这些引擎在同一个时间线上保持状态同步。这里的“引擎”可以理解为不同的大模型、不同的工具执行器、不同的检索后端甚至是不同语言写的推理服务。同步优化则是指当某个引擎的输出发生变化时其他引擎能及时感知并调整自己的策略而不是各跑各的。它适合谁如果你已经写过单 Agent 的 demo想往生产级多智能体协同走这篇就是给你准备的。如果你还在纠结“agent 是什么”也没关系我会把每个概念拆开讲保证你能跟上。整套东西落地之后你能拿它做智能客服编排、自动化数据分析流水线、多角色内容生产甚至复杂任务的分解执行。1.2 为什么是“多引擎”而不是“多模型”很多人一上来就把多引擎等同于多模型其实不是。多模型只是换了几个不同的 LLM而多引擎强调的是能力维度的拆分。我举个实际例子一个电商售后 Agent需要理解用户情绪情感引擎、查询订单数据库引擎、生成回复生成引擎、判断是否升级人工决策引擎。这四个引擎里只有生成引擎是纯 LLM其他三个可能是规则引擎、SQL 执行器、分类模型。这么拆的好处很直接每个引擎可以独立优化、独立扩容、独立替换。情感引擎换成更准的模型不影响订单查询数据库引擎加个缓存生成引擎完全无感。这就是“同步优化”的价值——优化是局部的收益是全局的。反过来如果你把所有逻辑塞进一个超大 prompt 里让一个模型干调试的时候你根本不知道是哪段逻辑出了问题改一处又怕影响另一处。我踩过这个坑一个 3000 字的 prompt 改了三天最后发现是中间一句指令和结尾的格式要求冲突了。1.3 整体架构的选型思路架构上我推荐“中心编排 引擎注册 共享状态”的三层结构。中心编排负责决定任务怎么流转引擎注册负责管理每个引擎的能力描述和健康状态共享状态则是一个所有引擎都能读写的上下文存储。为什么不用纯去中心化的多智能体因为去中心化在调试和可观测性上太痛苦了。我试过让几个 Agent 互相发消息协商结果一个任务卡住我花了两个小时才定位到是两个 Agent 对“完成”的定义不一致。中心编排虽然听起来不够“智能”但它把控制流显式化了出问题能一眼看到卡在哪。共享状态我建议用带版本号的键值存储而不是简单的内存字典。因为多引擎并发读写时没有版本控制就会出现“后写的覆盖先写的”这种经典问题。每次引擎更新状态时带上版本号读取时校验版本冲突了就重试或走补偿逻辑。2. 核心引擎的拆解与同步机制设计2.1 编排引擎任务流转的大脑编排引擎是整个系统的心脏它不直接干活只负责“下一步该谁干”。我习惯把它设计成一个状态机每个状态对应一个引擎的调用状态之间的转移条件由引擎的输出决定。举个具体例子一个内容审核 Agent 的编排逻辑是这样的先调分类引擎判断内容类型如果是文本就走文本审核引擎如果是图片就走图像审核引擎审核不通过就调通知引擎通过就调归档引擎。每个转移条件都写得很明确比如“分类引擎返回 text 且置信度大于 0.8”。这里有个关键设计点转移条件要可配置不要硬编码。我早期把条件写死在代码里后来业务方要加一个“置信度低于 0.5 时转人工”的分支我改了半小时代码。后来改成用配置表描述状态转移业务方自己就能加分支。# 状态转移配置示例 transitions [ {from: classify, to: text_review, condition: typetext and confidence0.8}, {from: classify, to: image_review, condition: typeimage and confidence0.8}, {from: classify, to: human_review, condition: confidence0.5}, ]2.2 推理引擎多模型的路由与降级推理引擎负责调用具体的 LLM但它不是简单地把请求转发出去。它要做三件事模型路由、结果聚合、失败降级。模型路由是根据任务类型选最合适的模型。比如代码生成走代码专精模型创意写作走通用大模型简单分类走小模型省钱。我一般会维护一张路由表记录每个模型擅长的任务类型和当前负载。结果聚合是当多个模型同时处理一个任务时怎么把结果合并。常见做法是投票、加权平均、或者让一个“裁判模型”来选。我实测下来对于事实性问题投票法最稳对于开放性问题裁判模型效果更好但成本高。失败降级是当主模型超时或报错时自动切到备用模型。这里要注意降级不是简单换个模型重试而是要把已经产生的中间结果带上否则备用模型得从头再来。我踩过的坑是降级后上下文丢了导致备用模型给出完全不同的答案。2.3 工具引擎外部能力的统一封装工具引擎把数据库查询、API 调用、文件操作这些外部能力封装成统一的接口让推理引擎不用关心底层实现。每个工具都要有清晰的输入输出 schema以及超时和重试策略。我建议每个工具都实现一个health_check方法编排引擎在调用前先检查工具是否可用。有一次我们的数据库工具挂了但编排引擎不知道一直往里发请求结果整个链路卡死。加了健康检查之后工具不可用就直接走降级分支用户体验好很多。工具的输入输出要用 JSON Schema 严格约束。我见过因为工具返回格式不固定导致推理引擎解析失败的案例。比如有的工具返回{result: ok}有的返回{status: ok}推理引擎就得写一堆兼容逻辑。统一 schema 之后这类问题基本消失。2.4 同步机制让引擎们步调一致同步是多引擎系统最难的部分。我把它分成三种同步状态同步、时序同步、语义同步。状态同步是保证所有引擎看到同一份上下文。做法是共享状态存储加版本号每次读写都校验版本。时序同步是保证引擎调用的顺序符合预期比如必须先检索再推理。语义同步最微妙是保证不同引擎对同一个概念的理解一致比如“高优先级”在检索引擎里是分数大于 0.9在决策引擎里也得是同一个阈值。语义同步我推荐用一份共享的“概念字典”所有引擎的阈值、枚举值都从这里取。这样改一个阈值所有引擎同步生效。我早期每个引擎各写各的阈值结果检索引擎认为 0.8 是高决策引擎认为 0.9 才是高中间那 0.1 的区间就成了黑洞。3. 从零搭建完整实操流程与关键代码3.1 环境准备与依赖选型先说环境。Python 3.10 以上是必须的因为要用到一些新的类型语法。核心依赖我推荐这几个fastapi做引擎的 HTTP 接口pydantic做数据校验redis做共享状态存储httpx做异步 HTTP 调用。为什么用 Redis 而不是内存字典因为多引擎可能部署在不同进程甚至不同机器上内存字典没法共享。Redis 的原子操作和过期机制正好适合做状态存储。如果你不想引入 Redis也可以用 SQLite 加文件锁但并发性能会差一些。pip install fastapi uvicorn pydantic redis httpx模型调用方面我建议抽象一层LLMClient把不同厂商的 API 差异屏蔽掉。这样换模型只需要改配置不用动业务代码。3.2 定义引擎接口与注册中心每个引擎都要实现统一的接口我定义了两个核心方法execute和describe。execute是干活的方法describe返回引擎的能力描述供编排引擎做路由决策。from abc import ABC, abstractmethod from pydantic import BaseModel class EngineInput(BaseModel): task_id: str context: dict params: dict class EngineOutput(BaseModel): task_id: str result: dict confidence: float next_hint: str | None None class BaseEngine(ABC): abstractmethod async def execute(self, input: EngineInput) - EngineOutput: pass abstractmethod def describe(self) - dict: pass注册中心是一个简单的字典记录引擎名称到实例的映射。启动时把所有引擎注册进去编排引擎通过名称查找。class EngineRegistry: def __init__(self): self._engines {} def register(self, name: str, engine: BaseEngine): self._engines[name] engine def get(self, name: str) - BaseEngine: return self._engines.get(name)这里有个细节注册中心要支持热更新。我遇到过线上要临时禁用某个引擎的情况如果注册中心不支持动态修改就得重启服务。加一个unregister方法配合配置中心就能做到不重启切换引擎。3.3 编排引擎的实现细节编排引擎的核心是一个循环读当前状态决定下一步调哪个引擎调用更新状态直到到达终止状态。class Orchestrator: def __init__(self, registry: EngineRegistry, state_store): self.registry registry self.state_store state_store async def run(self, task_id: str, initial_state: dict): state initial_state while state.get(status) ! done: engine_name self.decide_next(state) if engine_name is None: state[status] done break engine self.registry.get(engine_name) input_data EngineInput( task_idtask_id, contextstate, paramsstate.get(params, {}) ) output await engine.execute(input_data) state self.merge_state(state, output) self.state_store.save(task_id, state) return statedecide_next是编排逻辑的核心它根据当前状态和转移配置决定下一步。我建议把转移配置存在数据库或配置中心这样改流程不用发版。merge_state要小心处理冲突。如果两个引擎同时更新了同一个字段得有明确的合并策略。我的做法是给每个字段标记来源引擎冲突时按优先级取。3.4 共享状态存储与版本控制共享状态我用 Redis 的 Hash 结构存每个任务一个 key字段是状态项。版本号单独存一个字段每次更新时用WATCH加事务保证原子性。import redis import json class StateStore: def __init__(self, redis_client): self.redis redis_client def save(self, task_id: str, state: dict): key ftask:{task_id} version state.get(_version, 0) 1 state[_version] version self.redis.hset(key, mapping{data: json.dumps(state)}) def load(self, task_id: str) - dict: key ftask:{task_id} data self.redis.hget(key, data) return json.loads(data) if data else {}版本控制的关键是读取时拿到版本号写入时校验版本号是否变化。如果变了说明有其他引擎改过当前写入要重试。这个机制能避免大部分并发冲突。3.5 多引擎协同的完整调用链把上面几块拼起来一个完整的调用链是这样的用户请求进来编排引擎初始化状态然后按配置依次调用分类引擎、检索引擎、推理引擎、校验引擎每个引擎的输出都合并进共享状态最后返回结果。我拿一个实际的内容生成任务举例。用户输入一个主题系统先调检索引擎找相关资料再调推理引擎生成初稿然后调校验引擎检查事实性不通过就回到推理引擎重写通过就调格式化引擎输出。整个链路里检索引擎和推理引擎通过共享状态传递资料校验引擎的反馈也写进状态供推理引擎读取。这里要注意重试要有次数上限。我见过校验一直不通过、推理一直重写、最后 token 烧光的案例。加一个最大重试次数超过就走降级分支比如返回初稿加人工审核标记。4. 常见问题排查与性能优化实录4.1 引擎调用超时与级联失败多引擎系统最怕级联失败一个引擎慢了拖垮整个链路。我遇到过检索引擎响应从 200ms 涨到 5s导致后面所有引擎都在等最后整个请求超时。解决办法是给每个引擎设置独立的超时时间并且超时后走降级分支而不是直接失败。检索超时就返回空结果让推理引擎基于已有知识生成推理超时就返回缓存结果或简化版。import asyncio async def call_with_timeout(engine, input_data, timeout3.0): try: return await asyncio.wait_for(engine.execute(input_data), timeouttimeout) except asyncio.TimeoutError: return EngineOutput( task_idinput_data.task_id, result{error: timeout}, confidence0.0, next_hintdegrade )超时时间怎么定我的经验是取 P99 延迟的 1.5 倍。先跑一段时间收集延迟数据再根据数据调整。不要拍脑袋定 1 秒或 10 秒。4.2 状态不一致的排查思路状态不一致的表现是引擎 A 说任务完成了引擎 B 说还在处理。排查的时候先看共享状态的版本号如果版本号跳跃很大说明有并发写入冲突。我整理了一个排查清单现象可能原因排查方法引擎读到旧状态缓存未失效检查状态存储的 TTL 和刷新策略状态字段丢失合并逻辑覆盖打印每次合并前后的状态 diff版本号不连续并发写入检查是否有引擎绕过 StateStore 直接改状态任务卡住不推进转移条件不满足打印当前状态和所有转移条件的求值结果我踩过最坑的一次是某个引擎在异常分支里直接改了内存里的状态对象没走 StateStore导致其他引擎读到的还是旧状态。后来强制所有状态修改必须走 StateStore问题就没了。4.3 并发场景下的性能瓶颈多引擎系统天然是并发的性能瓶颈通常出现在三个地方模型调用、状态存储、编排循环。模型调用是最大的瓶颈因为 LLM 本身延迟就高。优化手段是并行调用无依赖的引擎。比如检索和情感分析没有依赖关系可以同时跑。我用asyncio.gather把这两个引擎并行整体延迟降了 40%。状态存储的瓶颈在 Redis 的网络往返。优化手段是批量读写把多次hget合并成一次hmget。另外状态不要存太大的对象超过 1MB 的字段考虑单独存对象存储。编排循环的瓶颈在decide_next的求值。如果转移条件很多每次求值都要遍历一遍。优化手段是把条件编译成可复用的表达式或者用决策树预计算。4.4 引擎健康检查与自动摘除生产环境里引擎会时不时挂掉。如果没有健康检查编排引擎会一直往挂掉的引擎发请求浪费时间和资源。我的做法是每个引擎暴露一个/health接口编排引擎定期探测。连续三次失败就把引擎标记为不可用路由时跳过。恢复后再自动加回来。class HealthChecker: def __init__(self, registry, interval30): self.registry registry self.interval interval self.failures {} async def check_all(self): for name, engine in self.registry.all().items(): try: await asyncio.wait_for(engine.health_check(), timeout2.0) self.failures[name] 0 except Exception: self.failures[name] self.failures.get(name, 0) 1 if self.failures[name] 3: self.registry.mark_unavailable(name)健康检查的频率别太高30 秒一次够了。太频繁会给引擎增加不必要的负载尤其是引擎本身就在处理重任务的时候。4.5 日志与可观测性建设多引擎系统的调试难度比单 Agent 高一个数量级因为问题可能出在任何一个引擎也可能是引擎之间的交互。没有好的日志排查就是大海捞针。我建议每个引擎的输入输出都打日志带上task_id和engine_name。这样你可以按task_id把整个链路的日志串起来看。日志级别用 INFO 记录正常流转ERROR 记录异常DEBUG 记录详细状态。另外给每个引擎加指标上报调用次数、成功率、平均延迟、P99 延迟。这些指标用 Prometheus 收集Grafana 展示。我靠这套指标发现过一个引擎的 P99 延迟是平均延迟的 20 倍一查是某个特定输入触发了慢查询。5. 多智能体协同的进阶玩法5.1 从编排式协同到协商式协同前面讲的都是编排式协同中心编排引擎决定一切。进阶玩法是协商式协同引擎之间可以互相发消息、协商任务分配。协商式协同适合任务边界不清晰的场景。比如一个研究助手检索引擎找到资料后可能觉得需要更多背景就主动请求推理引擎先分析一下再继续检索。这种动态协商在编排式里很难表达因为编排逻辑是预先定义的。实现协商式协同的关键是消息总线。每个引擎订阅自己关心的消息类型收到消息后决定是否响应。消息总线可以用 Redis 的 Pub/Sub 或者更专业的消息队列。不过我要提醒一句协商式协同的调试难度很高因为消息流是动态的很难复现。我建议先用编排式把业务跑通确实遇到编排表达不了的场景再上协商式。5.2 引擎能力的动态发现与组合当引擎数量多起来之后手动配置路由就不现实了。这时候需要引擎能力的动态发现每个引擎启动时把自己的能力描述注册到注册中心编排引擎根据任务需求自动匹配引擎。能力描述我建议用结构化的 schema包括输入类型、输出类型、擅长任务、成本、延迟。编排引擎根据这些信息做匹配和排序。{ name: code_review_engine, input_types: [code_diff], output_types: [review_comment], skills: [security_check, style_check], cost_per_call: 0.002, avg_latency_ms: 800 }动态组合的难点在于多个引擎都能干同一件事时怎么选我的策略是综合成本、延迟、历史成功率算一个分数选分数最高的。历史成功率很重要有些引擎虽然便宜但经常出错算下来反而更贵。5.3 记忆机制让 Agent 记住上下文多智能体系统如果没有记忆每次任务都从零开始效率很低。记忆机制分短期记忆和长期记忆。短期记忆就是当前任务的共享状态任务结束就清掉。长期记忆是跨任务的知识比如用户偏好、历史决策、常见问题的解决方案。长期记忆我建议用向量数据库存按语义检索。每次任务开始时先检索相关记忆注入共享状态。任务结束时把有价值的结论写回长期记忆。这里要注意记忆的时效性。有些记忆会过期比如用户三个月前的偏好可能已经变了。给记忆加时间戳检索时按时间衰减加权。5.4 安全边界与权限控制多引擎系统里不同引擎的权限应该不同。检索引擎只能读数据执行引擎才能写数据。如果权限不隔离一个被攻破的引擎可能影响整个系统。我的做法是给每个引擎分配一个角色角色决定它能访问哪些资源。编排引擎在调用引擎时把角色信息一起传过去引擎在执行前校验权限。另外引擎之间的通信要加密和鉴权。即使是内网也不能假设绝对安全。用 mTLS 做双向认证每个引擎有自己的证书。还有一点容易被忽略引擎的输出要过滤。推理引擎可能生成包含敏感信息的内容直接返回给用户就出问题了。加一层输出过滤引擎检查并脱敏。6. 我踩过的坑与实战经验总结6.1 不要过早追求“全自动”我刚开始做多引擎系统时一心想着全自动所有决策都让系统自己做。结果上线后发现很多边界情况系统处理不了用户投诉不断。后来我改成“人机协同”系统处理常规情况遇到低置信度或异常情况就转人工。人工处理的结果又反馈给系统用于优化模型和规则。这样系统越用越准用户满意度也上来了。所以我的建议是先做半自动把人工介入的环节设计好等系统稳定了再逐步减少人工。不要一上来就追求全自动那是给自己挖坑。6.2 引擎粒度不是越细越好我一度把引擎拆得很细一个功能一个引擎结果引擎数量爆炸编排逻辑复杂得没法维护。后来我合并了一些引擎把关联性强的功能放在一起系统反而更稳定。引擎粒度的原则是高内聚、低耦合。经常一起调用的功能放一个引擎需要独立扩缩容的功能拆开。不要为了“看起来架构清晰”而过度拆分。6.3 测试要覆盖引擎组合场景单引擎测试好写多引擎组合测试难写。我吃过亏每个引擎单独测都通过组合起来就出问题因为引擎之间的交互没测到。我的做法是写组合测试用例覆盖常见的引擎调用序列。比如“检索-推理-校验”这条链路要测检索返回空、推理超时、校验不通过等各种分支。这些用例跑起来慢但能发现大部分集成问题。另外用 mock 引擎做单元测试用真实引擎做集成测试。单元测试跑得快集成测试跑得慢但更真实。两者都要有。6.4 成本控制是长期课题多引擎系统调用多个模型成本比单 Agent 高。如果不控制账单会很吓人。我见过一个月烧掉几万块的案例。控制成本的手段有几个小模型干简单活大模型干复杂活缓存常见请求的结果设置每个任务的 token 上限监控成本指标异常时告警。我还会定期分析成本构成看哪个引擎最烧钱然后针对性优化。有一次发现检索引擎的 embedding 调用占了 60% 成本换成更便宜的 embedding 模型后成本降了一半效果几乎没变。6.5 版本管理与灰度发布多引擎系统里每个引擎都可能独立迭代。如果没有版本管理新版本引擎和旧版本编排逻辑不兼容就会出问题。我的做法是给每个引擎的接口加版本号编排引擎按版本号调用。新版本引擎先灰度发布只让少量流量走观察没问题再全量。灰度发布的关键是能快速回滚。我一般保留最近三个版本的引擎出问题一键切回旧版本。回滚时间控制在 1 分钟内这样即使出问题影响也有限。7. 后续可以这样扩展这套系统跑通之后扩展方向很多。我目前在做的是把引擎部署到边缘节点让靠近用户的请求在本地处理降低延迟。另一个方向是引入强化学习让编排引擎根据历史反馈自动优化路由策略而不是靠人工配置。还有一个有意思的方向是引擎的自动生成给定一个任务描述自动生成对应的引擎代码和编排逻辑。这个还在探索阶段但已经能看到一些雏形。等成熟了再单独写一篇分享。如果你也在做多引擎 Agent欢迎交流。我踩过的坑可能你也会遇到提前知道能省不少时间。