做AI Agent开发这段时间最大的感受不是模型能力不够而是** Agent 之间的协作能把人逼疯**。单机单 Agent 跑通一个 demo 很容易一旦涉及多个智能体分工、任务交接、结果汇总你面对的就是一锅浆糊函数调用散落各处、上下文互相污染、超时重试全看运气。我做的 Agent-Reach 就是为了解决这一层问题——把它定位成一个轻量的 Agent 连接与编排层专门处理多 Agent 之间的路由、会话、上下文传递和技能调度让每个 Agent 只专注自己的任务其余交给基础设施。这篇文章把项目的设计思路、核心代码、踩坑记录一次性写清楚希望能给正在被多 Agent 协作折磨的人一个可以直接抄作业的参考。Agent-Reach 适合谁两种人最需要一是已经在用 LangChain、AutoGen 这类框架但发现高层抽象反而限制了自由度的开发者二是刚从单 Agent 升级到多 Agent被消息乱飞和上下文错乱搞得焦头烂烂的团队。项目本身不依赖任何重型框架核心用异步消息传递 注册表路由 会话隔离差不多三百行代码就能搭出一个雏形放生产环境则需要再补上持久化和监控。看完这篇你至少能明白一个核心结论多 Agent 系统不一定要靠框架自己维护一个清晰的消息通道往往比什么都有更省心。1. Agent-Reach 是什么一个被逼出来的连接层1.1 从单 Agent到多 Agent的痛我最初做 Agent 应用时走的也是常规路线一个大 Prompt 塞进模型把工具列表全部挂上让模型自己决定调哪个函数。单 Agent 模式下一切还算听话可一旦任务复杂度上去比如查资料 → 写摘要 → 生成图表 → 发给用户这串流程压在一个 Agent 里会出现几个棘手问题。第一个问题是上下文爆炸。每个工具调用的输入输出都要塞回对话历史一轮操作产生几千 token 的回传内容几次调用之后模型注意力就开始漂回答质量断崖式下降。第二个问题是职责混乱。工具越多模型选错函数的概率越大尤其当工具名称相似时。我曾经在一个 Agent 上挂了 12 个工具结果模型把获取天气和查询日历搞混闹出过凌晨三点给用户推送明天晴适合开会的乌龙。第三个问题是并发能力为零。单 Agent 只能串行执行任务哪怕其中两个子任务彼此毫无依赖也只能排队等待。于是我把任务拆开让多个 Agent 各管一段。但是新的问题立刻出现了Agent A 的输出怎么交给 Agent B它们之间怎么知道彼此的存在上下文如何传递这中间缺的正是 Agent-Reach 要补上的那一层。1.2 为什么不能直接点对点硬连最朴素的方案当然是 Agent 之间直接互相调用。比如 Agent A 直接发 HTTP 请求给 Agent B问它结果。这在只有两三个 Agent 的时候完全可行但一旦 Agent 数量增长这种网状结构会迅速变得无法维护。想象一下你有 8 个 Agent。如果每个 Agent 都要知道其他 Agent 的地址、接口、数据格式那每接入一个新 Agent就要改所有相关 Agent 的代码。更糟糕的是Agent 之间的依赖关系会变成一张乱麻A 依赖 BB 依赖 CC 又依赖 A你盯着调用链路图看半小时也理不出头绪。网络层面还有个不起眼但很致命的问题消息投递的可靠性。直接用 HTTP 调用时如果目标 Agent 正在处理别的任务请求可能会超时如果目标 Agent 崩溃重启请求直接丢失。Agent 越多这种不可靠性被放大得越厉害。Agent-Reach 的思路是把这种网状拓扑收敛成星型拓扑——所有 Agent 只和中心节点通信中心节点负责路由、排队和投递这样每个 Agent 只需要关心自己的输入输出格式对外部世界的认知成本降到了零。1.3 Agent-Reach 的核心定位Agent-Reach 不是一套完整的 Agent 开发框架也不提供大模型调用能力。它更像一个消息总线 路由网关。每个 Agent 接入时向中心节点注册自己的标识和能力描述之后 Agent 之间通过消息通信而不是直接函数调用。中心节点负责任务分发、结果回传、超时重试和上下文隔离。这样的设计有几个显著优势。一是解耦彻底Agent 升级或替换不需要动其他模块二是链路清晰所有消息都有唯一的 trace ID出问题时可以从日志里完整还原一整条调用链路三是扩展容易新 Agent 接入只需要一份配置文件和一次注册不需要改已有代码。这些特性恰好是多 Agent 系统走向生产环境时的硬性要求。2. 核心概念拆解消息路由、会话绑定与技能注册2.1 消息路由与 TAG 寻址机制Agent-Reach 里最核心的概念是TAG。每个 Agent 在注册时会被分配一个全局唯一定位符格式很简单type.group.name。比如agent.nlp.summarizer表示 NLP 类型、摘要分组下的 summarizer 实例。所有 Agent 之间的消息只需要在头部声明目标 TAG 和消息类型路由节点负责把消息送到对应的 Agent 实例。这样做的好处很明显发送方不需要知道目标 Agent 的具体 IP 或进程地址只依赖逻辑标识。哪怕目标 Agent 从 node1 迁移到 node2发送方代码完全不需要变化。路由层内部维护一张从 TAG 到实际连接地址的映射表动态更新、动态感知。路由还承担消息过滤的职责。当一个消息广播给agent.nlp.*时可以匹配所有 NLP 类型的 Agent而精确寻址时只有 TAG 完全匹配的 Agent 才会收到。这样从请求到应答的整个链路都可以通过 TAG 来描述日志清晰排查问题非常直观。2.2 会话上下文怎么做到不串线多 Agent 协作时最隐蔽的坑就是上下文串线。用户给 Agent A 发了个请求A 在处理过程中又向 Agent B 发起子请求B 返回结果后 A 需要把结果和原始对话关联起来。如果没有会话机制多个并发请求同时运行时你根本分不清哪个结果属于哪个请求。Agent-Reach 处理这个问题的方式是引入Session ID Parent ID两级标识。Session ID 表示一条完整的用户请求链路从用户发起开始一直到最终响应结束Parent ID 表示当前消息的上游消息 ID用于构建调用树。每个 Agent 在处理消息时必须将这两个 ID 原样带到后续的所有请求中类似 HTTP Header 里的X-Request-ID和X-Parent-ID。上下文存储上采用按 Session 隔离的槽位。每个 Session 维护一个独立的 KV 存储Agent 之间的中间结果都写入当前 Session 槽位。默认情况下Agent 读取不到其他 Session 的数据即使同一个 Agent 实例在并发处理多个 Session也不会互相污染。这一点在生产环境里救了我无数次——早期没有做隔离时用户 A 的文档摘要经常出现在用户 B 的对话里这是非常严重的生产事故。2.3 技能注册表让 Agent 学会分工要让多个 Agent 像一个整体一样工作光有路由还不够还要有人知道什么任务该派给谁。这个角色就是技能注册表Skill Registry。技能注册表维护一份全局的 Agent 能力清单。每个 Agent 注册时除了指定 TAG还要声明自己能够处理的技能项比如summarize、translate、code_review。路由节点收到任务请求时会根据请求方声明的技能需求结合注册表里每个 Agent 的负载状态选出最合适的接收者。这里有一个非常实际的问题多个 Agent 声称自己支持同一个技能时怎么办我目前的策略是优先级priority 健康度权重health score。每个 Agent 注册时可以声明一个优先级值默认 0。路由时先取 priority 最高且健康度正常的 Agent如果该 Agent 连续失败达到阈值路由节点自动降权并切换到下一个。这种机制不保证全局最优但足够应付绝大多数场景而且实现成本极低。3. 30 分钟搭起第一个 Agent-Reach 节点3.1 环境准备与安装Agent-Reach 的运行时依赖很少核心只需要 Python 3.10 和一个 Redis。Redis 在这里承担三件事消息队列轻量任务分发、Session 存储KV 上下文共享、注册中心TAG 映射表持久化。选 Redis 的原因很简单部署简单、健壮性有保障、生态成熟几乎任何云环境都能快速拉起一个实例。如果你是本机测试一条命令就能启动 RedisDockerdocker run -d --name reach-redis -p 6379:6379 redis:7-alpine然后安装 Agent-Reach 本体我建议直接用 pippip install agent-reach安装完可以验证一下版本reach --version如果输出类似agent-reach 0.4.x环境就绪。接下来做的最重要的一件事给每个 Agent 一个唯一的 TAG 和配置入口。Agent-Reach 的所有 Agent 通用配置都写在 YAML 文件里即使完全不懂底层代码也能通过改配置接入新 Agent。3.2 写一个最简单的 Agent 配置我们先用官方模板初始化一个示例Agent。执行reach init sample_agent --template simple这会在当前目录生成agent.yaml和一个agent.py文件。核心配置如下我会把每个字段的意图都讲清楚agent: tag: agent.sample.hello type: sample group: hello name: greeter version: 1.0.0 transport: type: redis channel: reach:msg skills: - say_hello - echo session: ttl: 3600 # Session 存活时间秒 retry: max_attempts: 3 # 单个消息最大重试次数 backoff_base: 1 # 指数退避基数单位秒这个文件是 Agent 接入中心节点的身份证。注册时路由节点会读取其中的 tag、skills 等信息存入注册中心。transport字段表示通信方式目前最常用的是 Redis 发布订阅模式Agent 处理完消息后把结果发回指定的响应通道。3.3 本地验证把两个 Agent 接通接下来启动两个 Agent。第一个是默认的 greeter第二个我们自己写一个简单的 calculator。先启动路由节点reach router start --config router.yaml然后在两个终端分别启动 Agentreach agent start --config agent.yaml reach agent start --config calculator.yamlRouter 启动时会连接 Redis拉起监听任务Agent 启动时会自动向 Router 注册自己的 TAG。日志中看到registered agent.sample.hello就说明这一步完成。现在我们从测试客户端发一条消息请求 greeter 执行say_hello技能reach send --to agent.sample.hello --skill say_hello --payload {\name\:\张三\}如果一切正常你会看到类似输出response: 你好张三。这看起来像是发了个消息但背后其实已经完成了注册发现、消息路由、会话绑定和结果回传四个步骤。3.4 关键配置参数逐一解释这里挑几个容易被忽略但实际很重要的参数讲。channel 命名reach:msg是默认的消息发布通道。如果你有多个环境dev、staging、prod建议把 channel 命名改成reach:msg:dev或reach:msg:prod避免环境之间的消息互相污染。session.ttlSession 存活时间。如果设置过短比如 60 秒一个长任务还没跑完上下文就被清理了如果设置过长比如 24 小时又非常浪费内存。我的经验是常规对话场景 30 分钟到 1 小时足够长任务流水线才考虑加大。retry.max_attempts最大重试次数。不要设置得过大否则下游 Agent 一旦出现故障消息会在队列里反复重试占用大量资源。3 次是一个合理起点。transportRedis 是最轻量的方案但如果你的 Agent 分散在不同服务器甚至不同机房建议切换为基于消息队列的 transport如 RabbitMQ 或 KafkaRedis 在跨网络场景下的可靠性偏弱。提示第一版不要一上来就想搞高可用。先把单节点跑通再关注重试策略和状态持久化多 Agent 系统的复杂度是一点点涨上来的。4. 实战构建一个提问-检索-执行三 Agent 流水线4.1 场景设定为什么选这个结构为了把 Agent-Reach 的能力发挥出来我设计了一个典型场景用户提出一个自然语言问题系统先理解意图再从知识库检索相关信息最后调工具执行具体操作。这个场景几乎是企业级 Agent 应用最通用的范式。拆成三个 Agent 的理由很直接检索和工具调用的资源消耗不同。检索 Agent 需要访问数据库和向量索引工具执行 Agent 需要调用外部 API两者如果合并在一起Prompt 会非常长而且模型在要不要调工具的判断上容易犹豫。拆开后每个 Agent 只做一件小事Prompt 短、判断快、出错容易定位。4.2 三个 Agent 的职责划分与 Prompt 设计三个 Agent 分别是agent.nlp.router、agent.search.knowledge和agent.tool.executor。agent.nlp.router的 Prompt 只做一件事从用户的原始问题中提取出意图类型和关键词返回一个结构化的 JSON。不要求它回答任何实际内容只做语义解析。这样模型不需要检索工具也不需要考虑回答格式出错率大幅下降。agent.search.knowledge的 Prompt 是接收结构化查询参数访问知识库返回相关片段列表。这里没有大模型的判断任务只是一个接口转换器输入 JSON输出检索结果。agent.tool.executor的 Prompt 是根据检索结果和用户原始问题决定执行哪个工具、传什么参数。这个 Agent 要保证工具调用的正确性是最容易出现幻觉工具名的环节。我在它前面加了一层约束工具名称必须严格来自内置列表禁止模型自行发明工具名。4.3 编排逻辑与回调处理任务的关键在编排层。用户发来一句话Agent-Reach 的工作流workflow会依次完成三步向agent.nlp.router发送analyze_intent消息等 JSON。把 JSON 中的查询关键词传给agent.search.knowledge等检索片段。把检索片段与原始问题打包传给agent.tool.executor最终拿到执行结果。下面是一段简化的编排逻辑Python 伪代码async def handle_user_query(query: str, session_id: str): # Step 1: 意图解析 intent await reach.send_and_wait( targetagent.nlp.router, skillanalyze_intent, payload{query: query}, session_idsession_id, timeout5 ) # Step 2: 知识检索 docs await reach.send_and_wait( targetagent.search.knowledge, skillretrieve, payload{keywords: intent[keywords]}, session_idsession_id, timeout10 ) # Step 3: 工具执行 result await reach.send_and_wait( targetagent.tool.executor, skillexecute, payload{ query: query, context: docs, intent: intent[intent] }, session_idsession_id, timeout15 ) return result这段代码展示了send_and_wait的核心用法发消息、等待响应、拿结果。它会自动处理超时和重试。真正的生产环境我会加一个状态机但即使是这种顺序编排已经能覆盖相当多业务场景了。4.4 常见坑位status 判断、错误码、重试条件跑通流水线是一回事让它稳定工作是另一回事。这里记录几个高频坑位。坑位一把 HTTP 错误码当消息错误码。在 Agent-Reach 里HTTP 层成功不代表业务层成功。消息返回体里必须有明确的业务状态码code和描述message。我的约定是0表示成功非0表示各类失败。编排层判断code 0才继续往下走否则立即终止当前 Session 并向用户返回错误。这个约定早期没定下来时下游 Agent 返回了500字符串而上游 Agent 以为这是正常的状态 500。坑位二重试条件没有区分幂等和非幂等操作。检索操作是天然的幂等操作重复几次没影响但工具执行操作比如发送邮件扣减库存重复执行会出大问题。所以我在工具 Agent 的消息里增加了一个idempotent标志编排层看到idempotent: false时一律不自动重试只把结果挂在待人工处理列表里。坑位三超时时长的选择凭感觉。实际上应该结合 Agent 的平均处理时间做统计取 P90 或 P95 作为超时阈值。上线初期先用默认值运行一周后拉监控数据再调整不要一上来就猜一个可能偏大的值——超时设置过长会让整个链路的问题被掩盖。5. 常见问题与排查技巧实录5.1 消息丢失日志显示已路由但 Agent 没反应这是多 Agent 系统里最诡异的一类问题。Router 显示消息已经投递到目标 Agent但目标 Agent 没有产生任何响应。排查思路要按顺序走不要一上来就怀疑网络。先查目标 Agent 是否在线。Agent 如果实现了心跳机制默认每 30 秒上报一次Router 会在超过 90 秒未收到心跳后标记为离线。如果 Agent 侧线程卡死或者 Redis 断连就会出现Router 不知道 Agent 已死的情况。这是在日志里最容易出现的假象Router 认为消息已发但接收方进程已经僵死。再查 Session 上下文是否被清理。时间较长的任务如果 Session TTL 小于任务实际执行时长上下文会中途消失。Agent 在处理消息时会检查当前 Session 是否存在发现上下文不存在时会静默放弃。日志里通常会有一条session expired记录不显眼但这就是真相。最后查 Redis 的 pub/sub 丢消息。Redis 发布订阅模式是即发即弃的如果 Agent 在消息发出的瞬间处于重连窗口消息就真的丢了。解决方法是切换到 Stream 模式Redis 5.0 后支持Stream 支持消费组和消息持久化严格保证不丢消息代价是多了一些消息积压的风险。5.2 两个 Agent 无限互调循环风暴怎么掐断循环调用是 Agent 协作中非常经典的灾难场景。Agent A 请求 BB 处理不了把请求转发给 CC 有部分结果又回传给 AA 发现缺参数再次请求 B……如此循环往复。如果没有防护这个循环会以指数级速度消耗资金和资源。Agent-Reach 在消息头里增加了hops字段表示消息已经经过的 Agent 数量。每次转发hops 1。在 Router 层设置一个全局的max_hops配置比如默认 10。任何消息的 hops 超过阈值Router 直接丢弃并把一条loop_detected的告警写入日志。我的实际做法更严格一些在 Session 级别维护一个调用路径哈希集合。每到一个 Agent就把当前消息的 (session_id, parent_id, target_tag) 加入集合。如果下次要转发时发现集合中已有相同的三元组直接拒绝并记录告警。这相当于给消息加了一个记忆可以有效防止环状结构在同一个 Session 里反复触发。5.3 超时与重试策略怎么设超时和重试的策略我总结出三条经验。第一条区分整体超时和单次呼叫超时。整体超时是用户能等待的最长时间单次呼叫超时是 Agent 之间单跳的上限。整体超时一般设为 30 秒而单跳超时根据 Agent 类型调整检索类 5-10 秒工具执行类 10-15 秒模型调用类如果 Agent 内部调 LLM10-30 秒。第二条重试必须配指数退避加抖动。指数退避好理解第一次重试等 1 秒第二次 2 秒第三次 4 秒。加上抖动jitter是为了避免风暴——多个请求同时失败时如果不加随机偏移它们会在同一时刻继续发起重试导致下游雪崩。第三条要区分延迟失败和真失败。超时之后消息可能实际上已被目标 Agent 处理完成只是响应没来得及回到调用方。这种情况下直接重试会造成重复处理。我用幂等键消息头里的message_id去解决目标 Agent 在收到重试消息时会先检查当前message_id是否已经处理过如果处理过直接把上次的结果返回。5.4 配置更新不生效缓存与订阅的坑改完agent.yaml里某个 Agent 的配置重启后发现新配置完全没有生效这个问题我在 Agent-Reach 早期版本踩过。原因很典型Router 在内存里缓存了 Agent 注册信息而且缓存没有设置过期时间。即使 Agent 重新注册Router 仍然按旧的配置路由。后来的解决方案是给注册信息增加配置版本号。Agent 启动时除了注册 TAG还会带上一个config_version字段。Router 对比发现版本号不一致时会主动拉取新的配置并替换缓存。这样 Agent 每次更新配置后只需重启一次就能保证 Router 侧立即感知。如果你用的是 Stream 模式还有一个隐藏的坑消费组的 offset 可能会积压。配置更新后要检查消费组的 pending 消息是否过大。我见过一个环境里 pending 消息积压了几万条导致新消息一直在排队表现出来就是配置改了但 Agent 行为像旧版本。定期用XINFO GROUPS命令检查积压情况是一个值得养成的好习惯。6. 项目落地后的几点实在心得聊了这么多机制的细节最后分享几个从 Agent-Reach 实际落地里得到的体会这些比任何配置项都更重要。第一多 Agent 系统的复杂度不是来自 Agent 本身而是来自它们之间的交互协议。协议不清晰Agent 数量越多越混乱。Agent-Reach 能稳定运行最大的功劳不是路由算法而是把消息头字段TAG、Session ID、Parent ID、hops、message_id从第一天就定死并且所有 Agent 都严格遵循。如果想把 Agent-Reach 用到自己项目里第一件事不是部署而是先花半天时间定义清楚你的消息协议。第二监控比功能更重要。Agent-Reach 上线之后我花在加监控上的时间比写业务 Agent 的时间还多。每个环节的耗时、重试次数、失败分布都要建好仪表盘。没有这些数据你根本不知道问题出在哪个 Agent只能靠猜。建议从第一天就埋好 trace 数据最晚不能晚于联调阶段。第三Agent 的拆分粒度不要过细。我之前尝试把意图分类再分成情感分析和关键词提取两个 Agent结果一个查询要多花两跳网络开销响应时间明显变长。拆分的标准很简单如果两个任务几乎总是同时需要执行而且一个的输出直接决定另一个的输入那它们应该在一个 Agent 里。粒度太细会带来大量无意义的调度开销系统整体反而更慢。Agent-Reach 的连接能力再强也不该被用来弥补糟糕的任务设计。第四也是我踩坑最深的一次不要一开始就追求全面的动态编排。我最初设计了一个非常灵活的 DAG 工作流支持任意分支和合并。结果实现复杂不说调试异常困难一个节点报错之后整个执行路径很难直观还原。后来我砍掉了大部分灵活性改用顺序编排 条件分支代码量少了三分之二稳定性反而大幅提升。Agent-Reach 后续版本里我会把重点放在提升默认顺序编排的执行速度和容错能力上而不是继续扩展更复杂的编排形态。如果你正准备搭自己的多 Agent 系统我的建议很简单先把 Agent-Reach 这类消息层跑通用最简单的方式定义协议再逐步丰富你的技能库。这套路可能不炫酷但它是真正不会把人耗死在调试里的路径。