这半年我绝大部分业余时间都扔在了一个叫Agent-Reach的项目上。起因很直接团队里的智能体越上越多客服一个、知识库一个、报表一个、审批一个各自单拎出来都能跑可一旦遇到要跨多个Agent协作的复杂问题就暴露出了问题——任务该交给谁、多个Agent执行的先后顺序怎么定、其中一个挂了怎么处理、结果怎么可靠收回来。这些问题没有统一答案每次都是硬编码调用链加一个Agent就要改一圈代码。Agent-Reach就是冲着这个痛点去的核心解决三件事能力路由、任务编排、触达回执。如果你也在做多Agent协作或者想把Agent接入生产环境这篇文章里写的架构取舍、核心机制、部署压测数据、还有我踩过的六个大坑应该能帮你少走不少弯路。1. 为什么需要Agent-Reach多智能体协作的调度之痛1.1 我遇到的实际场景先还原一个让我决定动手的场景。用户问“帮我查一下上个月华东区的销售额和去年同期对比然后简单分析一下原因。”这个问题本身不复杂但它要拆成三步报表Agent先拉数据知识库Agent再结合行业常识做归因最后客服Agent把结论润色成自然语言回复。三个Agent串行调用中间任何一步超时整个流程就卡死。更麻烦的是当时我们用硬编码把这三个Agent绑在一条调用链里下一次需求换成“对比华北区”又要改代码。这种问题并非个例。Agent一多就会冒出三类高频问题一个任务有多种Agent能处理选谁按什么规则选任务之间有依赖关系先A后B或者C失败才执行D怎么表达Agent处理完以后结果怎么可靠通知回来Agent崩溃了怎么办我调研过市面上的编排框架比如LangGraph、Temporal它们确实强大但对一个中小团队来说太重了——要引入一堆概念学习成本高和现有系统的集成也是大工程。直接用消息队列呢又太底层消息里只有字节没有“Agent能力”这种语义。Agent-Reach就是在两者之间找一条轻量路径做一个中心调度、边车注册的轻量引擎让Agent像打电话一样被触达——拨过去有响应、有回执、有状态挂断了还能重拨。1.2 市面方案的缺口我把问题拆开看发现现有工具各有盲区。Agent框架如LangGraph擅长把大模型调用组织成图但很少管“Agent部署在哪、如何被可靠通知”消息中间件擅长可靠投递但不懂“哪个Agent有处理这个任务的能力”。Agent-Reach的定位就是填补这个空白它不是一个Agent开发框架而是一个Agent之间的触达与调度层。你完全可以用LangGraph写单个Agent内部的逻辑再用Agent-Reach把多个Agent串起来。所以项目名字里的“Reach”是我刻意选的。Reach表达的是“触达”——我要让一个Agent能稳定、可靠地触达另一个Agent并且知道触达的结果是什么。1.3 Agent-Reach要解决的三个核心问题整个项目围绕三个问题展开后面所有设计都是对这三个问题的回应任务该交给谁需要一个能力路由机制Agent启动时描述自己会什么任务进来时按规则和语义匹配最佳处理者。多个Agent怎么配合需要编排能力用声明式的方式描述任务之间的顺序、分支和依赖。结果怎么可靠收回需要触达回执包括成功、失败、进度上报以及超时、重试等异常兜底。这三个问题对应Agent-Reach的三个核心模块注册中心、调度中枢、触达网关。下面逐个讲。2. 整体架构与设计取舍中心调度加边车注册2.1 核心组件调度中枢、注册中心、触达网关Agent-Reach由三个语义清晰的组件组成调度中枢Core负责接收任务、做路由决策、驱动编排状态机、记录任务生命周期。它是唯一理解“任务应该怎么走”的地方。注册中心Registry维护所有Agent的在线状态和能力描述。Agent启动时上报自己的endpoint和能力清单然后通过心跳保活。触达网关Gateway负责与外部Agent的实际通信支持HTTP和WebSocket两种接入方式。调度中枢不直接连接Agent所有出站请求都走网关这样便于统一做超时控制、重试和流式转发。这三个组件在物理部署上可以分开也可以合并后面第5章会讲我实际部署的演进过程。数据层面我用PostgreSQL存Agent注册表、任务表和事件表用Redis做任务队列和临时状态缓存。选PostgreSQL是故意的任务状态是最重要的数据不能丢用关系型数据库能拿到事务和唯一索引来做幂等。2.2 为什么不选完全去中心化做架构评审时好几个人建议用P2P模式——每个Agent互相发现、互相调用不设中心节点。理论上这样没有单点扩展性更好。但我在实际调研后否掉了这个方案原因很现实调试困难Agent互相直连任务流转路径飘忽不定出了问题根本不知道去哪看链路上的完整记录。语义复杂每个Agent都要实现一套发现、握手、心跳、重试逻辑工程量大到失控。审计缺失生产环境里合规要求你必须能回答“这个任务经过了哪些Agent、每一步谁处理的”。中心化天然留痕。中心化的代价是单点但这可以在部署层解决。调度中枢可以起多个副本任务表使用PostgreSQL的FOR UPDATE SKIP LOCKED来实现多副本抢占空闲任务做到“逻辑中心、物理多活”。这比P2P带来的复杂度要划算得多。2.3 关键数据模型Agent-Reach的核心表就三张设计得尽可能简单方便业务方二次开发。Agent注册表agent_registry字段类型说明idUUIDAgent唯一IDnametextAgent名称capabilitiesjsonb能力描述JSON Schema格式endpointtext回调地址statustextonline / offline / degradedlast_heartbeattimestamptz最近心跳时间能力描述用JSONB是刻意的。Agent的能力是千差万别的不可能用关系型字段建模。比如客服Agent的能力描述是这样的{ services: [ {name: chat_reply, params_schema: {user_message: string}}, {name: sentiment_analyze, params_schema: {text: string}} ], metadata: {department: customer_service, region: all} }任务表tasks字段类型说明task_idUUID全局唯一任务IDdag_idtext编排定义IDsourcetext发起来源payload_reftext任务数据引用存在对象存储current_statustextpending / dispatching / executing / success / failedretry_countint已重试次数timeout_msint超时阈值created_at / finished_attimestamptz生命周期事件表task_events记录任务从创建到结束的每一步包括dispatched、ack、progress、terminal_success、terminal_failure等事件。这张表是审计和排查问题的关键。数据模型里的一个细节是payload_ref。最初我图省事直接把任务数据塞JSONB字段后来压测发现Payload过大会导致整个任务表膨胀。后来改成对象存储只存引用任务表只持有路径和大小查询性能和存储成本都好很多。这个放到第6章的踩坑里细说。3. 核心机制拆解能力路由、任务编排、触达回执3.1 能力路由规则过滤加语义匹配的融合方案路由是Agent-Reach的入口也是第一个要回答“为什么这样做”的机制。任务进来后第一步是规则过滤。比如某个报表任务的元数据标明“仅限华东区数据”那么注册表里metadata声明region: cn-east之外的Agent直接排除。规则过滤解决的是硬约束权限、地域、业务线这些是定死的不能靠模糊匹配来碰运气。第二步是语义匹配。剩下的候选Agent可能还有好几个比如“分析用户情绪”这个需求客服Agent能做一个专门的自然语言处理Agent也能做。这时就需要算一下任务描述和Agent能力描述之间的语义相似度。实际代码里我用的是sentence-transformers的预训练模型把两者都向量化然后算余弦相似度取最高分。这类模型常见的有BAAI/bge-small-zh-v1.5中文场景效果不错模型体积也小。from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(BAAI/bge-small-zh-v1.5) def route_decision(task_description, candidates): task_vec model.encode(task_description, normalize_embeddingsTrue) best_agent None best_score -1.0 for agent in candidates: cap_vec model.encode(agent.capabilities_text, normalize_embeddingsTrue) score float(np.dot(task_vec, cap_vec)) if score best_score: best_score score best_agent agent return best_agent, best_score为什么用“规则语义”融合而不是纯语义匹配因为我踩过纯向量匹配的坑。有次“查销售额”的任务匹配到了一个能做数据可视化的Agent因为它的能力描述里也有“销售额”三个字实际上它根本拉不了数。规则过滤把这类不满足硬约束的候选先干掉语义匹配只管模糊的排序两者各司其职。这里还有个阈值问题。相似度低于某个阈值比如0.5时路由不应该硬选一个而是返回“无法匹配合适Agent”并结束任务。宁可不做也不要错做。3.2 任务编排用DAG加状态机表达复杂依赖路由解决“单个任务给谁”编排解决“多个任务怎么串”。Agent-Reach用DAG有向无环图描述任务依赖关系每个节点是发给某个Agent的一次调用节点之间的边是依赖关系。下面这个DAG对应文章开头的场景先拉数据再分析原因最后润色回复。注意analyze_cause节点带了一个condition意思是只有数据拉取成功才执行。{ dag_id: sales_analysis_flow, nodes: [ {id: fetch_sales_data, agent: report_agent, timeout_ms: 10000}, {id: analyze_cause, agent: knowledge_agent, timeout_ms: 15000, condition: fetch_sales_data.status success}, {id: compose_reply, agent: chat_agent, timeout_ms: 10000, dependencies: [analyze_cause]} ] }执行引擎用一个简单的拓扑排序加状态机驱动。任务每完成一个节点就把该节点的后继节点加入就绪队列等依赖全部完成后触发下一个节点。import asyncio from collections import defaultdict async def run_dag(dag, ctx): pending defaultdict(int) children defaultdict(list) for node in dag[nodes]: parent_id node.get(dependencies, [None])[0] pending[node[id]] 0 if parent_id: pending[node[id]] 1 children[parent_id].append(node[id]) ready [n[id] for n in dag[nodes] if pending[n[id]] 0] results {} while ready: node_id ready.pop(0) node next(n for n in dag[nodes] if n[id] node_id) results[node_id] await dispatch_to_agent(node, ctx) for child in children.get(node_id, []): pending[child] - 1 if pending[child] 0: ready.append(child) return results这里有个设计原则值得多说一句不要在Agent内部写编排逻辑所有依赖判断都放在调度中枢。Agent只负责“收到一个任务、处理完返回一个结果”不知道自己在整个DAG里的位置。这样Agent可以完全无状态方便复用和替换。这和微服务里“服务无状态、编排在网关”是一个道理。3.3 触达回执outbox模式加幂等键Agent处理完任务后怎么通知中枢最直接的方式是Agent直接调中枢的回调接口但这存在问题如果中枢刚发起调用就崩溃重启或者Agent回调时网络抖动回调可能丢。Agent-Reach的解决方案是outbox模式加幂等回执。outbox模式的思路是网关在发起Agent调用前先在本地数据库的事件表里写一行“待发送”记录定时任务扫描这张表把未确认的消息发出去Agent成功接收并处理完成后回执一个完成事件事件表里对应记录标记为已确认。这样即使发送过程中宕机重启后定时任务能捞回未确认的消息继续发保证不丢。-- 定时任务捞取待发送的调用记录 SELECT * FROM outbox_events WHERE status pending OR (status sending AND updated_at NOW() - INTERVAL 30 seconds) ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED;幂等键用task_id attempt_seq组合。同一个任务重试多次时每次attempt_seq不同Agent侧拿到相同键时直接返回上一次的结果不重复执行。这是所有可靠消息系统都要解决的基础问题——网络请求可能超时但处理方可能已经成功执行了发送方不知道只能靠幂等键去重。4. 关键代码实现注册、路由、流式转发的落地细节4.1 Agent注册与心跳的实现Agent接入Agent-Reach第一件事是注册。我写了一个很轻的客户端SDK核心就两个方法register()和start_heartbeat()。import requests import threading import time import uuid AGENT_ID str(uuid.uuid4()) REGISTRY_URL http://core-host:8000/registry HEARTBEAT_INTERVAL 15 capabilities { services: [ {name: chat_reply, params_schema: {user_message: string}} ] } def register(): resp requests.post(f{REGISTRY_URL}/register, json{ agent_id: AGENT_ID, name: chat-agent, capabilities: capabilities, endpoint: http://agent-host:9001/callback, callback_mode: http }, timeout5) resp.raise_for_status() def heartbeat_loop(): while True: try: requests.post(f{REGISTRY_URL}/heartbeat, json{agent_id: AGENT_ID}, timeout5) except Exception as e: print(fheartbeat failed: {e}) time.sleep(HEARTBEAT_INTERVAL)注册接口只做两件事写入Agent信息和启动心跳。心跳时间间隔为什么是15秒这是权衡后的结果。5秒太频繁对Agent和注册中心压力都大30秒以上又影响故障感知速度。15秒能满足大部分场景检测到Agent掉线最多滞后15秒。4.2 路由决策和超时控制调度中枢收到新任务后执行路由决策然后通过网关发起调用。这里最容易踩的问题是超时没设或设得太大。Agent调用的超时必须按任务类型分别设置不能全局一个值。比如报表Agent要拉取大量数据超时可以放宽到30秒客服回复这种交互型任务2秒就该返回否则用户侧早就等得不耐烦了。路由和发起的核心逻辑用asyncio写得非常清爽import asyncio async def handle_task(task): # 1. 规则过滤 candidates filter_by_rules(task) if not candidates: return fail_task(task, no candidate agent matched rules) # 2. 语义匹配 agent, score route_decision(task.description, candidates) if score 0.5: return fail_task(task, similarity below threshold) # 3. 发起调用带超时 try: result await asyncio.wait_for( gateway.call_agent(agent, task), timeouttask.timeout_ms / 1000 ) return success_task(task, result) except asyncio.TimeoutError: return retry_or_fail(task, agent call timeout)asyncio.wait_for是这里的关键它不只是设了一个超时还负责在超时时取消正在等待的协程不会让调用悬挂在内存里浪费连接。4.3 流式返回在网关层的处理Agent执行长任务时比如报表Agent跑一个复杂的SQL可能要几十秒期间用户侧想知道进度。Agent-Reach的网关支持SSE流式转发Agent在处理过程中通过回调接口上报progress事件网关直接把进度透传给发起方。from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx app FastAPI() app.post(/v1/tasks/{task_id}/stream) async def stream_task(task_id: str, request: Request): async def event_stream(): async with httpx.AsyncClient(timeout60) as client: async with client.stream(POST, fhttp://agent-host:{task_id}/run) as resp: async for line in resp.aiter_lines(): yield fdata: {line}\n\n return StreamingResponse(event_stream(), media_typetext/event-stream)这个转发层看着简单但它承担了两个职责一是把Agent的进度翻译成标准SSE协议二是代理HTTP连接避免Agent直接暴露给调用方导致安全风险。5. 部署与压测从512并发到2000并发的调优过程5.1 部署形态单机到多节点的演进Agent-Reach的部署我分了三个阶段。第一阶段单机全组件。一台8核16G的虚机Core、Gateway、PostgreSQL、Redis全放一起连演示Agent都跑在这台机器上。这个阶段的目的是验证逻辑正确性不追求性能。跑通后我用locust做了简单压测512并发时问题立刻暴露出来PostgreSQL的连接数先爆了。每个任务的路由决策都要查库、每个并发协程都可能占一个连接直接把max_connections打到上限。第二阶段拆分数据库并加连接池。PostgreSQL移到了独立的机器引入PgBouncer做连接池把应用直接建连改成走池。这里一个关键经验PostgreSQL的max_connections不是设得越大越好设太大反而会因为上下文切换导致性能下降。我用的是max_connections200PgBouncer的pool_modetransaction池大小设为应用并发数的1.2倍左右。第三阶段Core多副本加SKIP LOCKED。单个Core实例成为瓶颈后我起了第二个Core副本。两个副本对同一个PostgreSQL任务表做抢占用FOR UPDATE SKIP LOCKED避免互相等待同一行。这样处理任务的水平扩展问题变成了增加Core副本代价是任务表偶尔出现少量锁竞争但实测影响很小。提示FOR UPDATE SKIP LOCKED是PostgreSQL 9.5才有的特性如果你用MySQL没有直接等价语法需要借助其他方案比如手动维护锁状态字段选型时要考虑到这一点。5.2 压测数据与瓶颈定位我用locust模拟任务提交任务内容是调用一个模拟Agent做1秒延迟的假处理。测试结果如下并发数P99延迟瓶颈表现512482 ms正常无报错10001.1 sPostgreSQL连接数告急20001.8 s连接耗尽大量任务超时2000调优后920 ms稳定运行瓶颈定位的思路是由上到下逐层排查。先看网关CPU网关是IO密集型节点如果CPU不高但延迟高说明瓶颈不在计算再看PostgreSQL的连接数和活跃查询连接数打满基本就是它了。用pg_stat_activity看活跃连接时我发现大量连接处于idle in transaction状态说明是连接池配置问题而不是查询本身慢。修复连接池后再把瓶颈定位到Core的单线程事件循环。路由决策里的sentence-transformers模型推理虽然是IO密集但CPU占用在2000并发时涨到了85%左右因为每个任务都要做一次向量编码。优化方案是在模型推理前面加了一个简单的关键词预过滤设计上先排除明显不匹配的Agent能省掉约40%的编码调用。5.3 三处不得不提的配置调优HTTP连接的超时参数发往Agent的HTTP请求要设置connect、read、write三个独立超时而不是笼统一个。connect超时建议3秒read超时根据任务类型从2到30秒不等。只用一个总超时会导致慢读取和慢连接混在一起难以定位问题。心跳容忍度注册中心认为Agent掉线的最长时间我设置为心跳间隔的3倍45秒。少于3倍会因为网络抖动误判离线大于3倍又拖慢了故障切换速度。PostgreSQL自动清理事件表写入量大如果不对历史事件做归档表会无限膨胀。我写了一个每天定时清理30天前终端事件的任务这样查询性能稳定任务表的体积可控。6. 踩坑实录六个让你排查到崩溃的边界问题6.1 Agent回调地址被NAT吞掉现象任务状态一直停留在dispatchingAgent日志显示任务早就收到了也处理完了但回调永远到不了调度中枢。排查链路我先查事件表发现dispatched事件有记录但terminal_success事件一条都没有。再登到Agent机器上看日志发现Agent回调用的是自己注册时报的内网IP。问题在于Agent部署在客户的NAT后面它上报的endpoint是172.16.0.10:9001调度中枢在云上根本路由不到这个地址。解决方案Agent端注册时增加一个配置项public_endpoint让运维在部署时手工指定外网可达的地址优先使用这个地址回调。更好的做法是让Agent主动和网关建立WebSocket长连接回调直接走这条已建立的通道彻底绕开NAT入站问题。这两种方式现在都支持。教训分发路径和回调路径要分开设计。分发是调度中枢主动发起要求Agent入口可达回调是Agent主动上报要求Agent出口可达。两个方向涉及的网络假设完全不同不能假设一个endpoint两个方向都通。6.2 Agent进程崩溃导致任务悬挂现象任务状态变成executing后再也不动了既没有成功回执也没有失败回执。这个问题的根因很简单Agent进程在跑一个长任务到一半时崩溃了它来不及上报任何事件。调度中枢一开始没有任何机制检测这种“既没成功也没失败”的悬挂状态。解决方案是加了一个watchdog协程定期扫描状态为executing且updated_at超过timeout_ms 30秒的任务强制标记为失败并触发重试UPDATE tasks SET current_status failed, error_message watchdog: agent no response WHERE current_status executing AND updated_at NOW() - (timeout_ms || milliseconds)::interval RETURNING task_id;这个机制看起来笨但在生产环境里它就是兜底安全网。调度中枢的功能完善很多时候不是加了多少高级特性而是把这些异常的边缘情况堵住。6.3 幂等表失效导致重复执行现象网络抖动触发重试后同一个任务在Agent侧被执行了两次产生了重复数据。排查发现我最初用Redis的SETNX做幂等去重Agent收到重试请求时检查Redis里有没有这个键。理论上是可行的但问题出在Redis的持久化配置上。我用的是默认配置不持久化或快照方式Redis重启时内存里的幂等键全部丢失旧的重试请求就能再次通过检查。解决方案是把幂等表移到PostgreSQL用唯一索引保证不重复CREATE UNIQUE INDEX idx_agent_idempotency ON agent_results(agent_id, task_id, attempt_seq);这样即使PostgreSQL也重启只要数据有持久化唯一索引依然能生效。涉及幂等的关键数据绝不要只放在Redis里除非你能接受它丢失。6.4 大上下文导致内存暴涨现象压测时发现Core节点内存持续上涨最终OOM了几次。定位过程用top观察内存大户再结合监控面板看哪些任务在执行。发现某些Agent的payload_ref指向的对象存储文件特别大——一个任务上下文有20MB里面有历史对话记录、报表原始数据、图片Base64等。当20个这样的任务并发时光Payload就占了几百MB内存。解决方案是三管齐下限制单个任务Payload大小不超过1MB超出就要求调用方把数据放在外部存储、只传引用任务执行完以后立刻释放Payload引用不让它留在内存里调度中枢的内存监控做了阈值告警超过80%就拒绝新任务而不是硬扛。6.5 时钟不一致引发的调度错乱现象注册中心频繁把健康的Agent标记为offline导致路由器跳过这些Agent即使它们明明活着。原因让人哭笑不得Agent部署的机器和调度中枢的机器时钟不一样。Agent的机器时钟比中枢快了3分钟它发的last_heartbeat时间戳比中枢的当前时间还早中枢判断时间差已经超过了3倍心跳间隔于是误判离线。解决方案很简单但容易忽略时间戳一律以调度中枢为准。Agent心跳只上报“我还活着”这个信号不携带具体时间由中枢收到心跳时取当前时间写入last_heartbeat。跨机房的场景可以考虑统一NTP校准但架构上不应依赖机器时钟一致。6.6 重试风暴现象一个Agent短暂不可用触发了大量任务的重试逻辑每个任务重试3次结果是几十个任务在同一秒内疯狂请求同一个不健康的Agent把它的恢复过程又拖长了。我在这上面体会到重试策略必须加退避和抖动。固定间隔重试会导致所有失败请求在时间轴上对齐形成尖峰脉冲。正确的做法是使用指数退避加随机抖动import random def next_retry_delay(attempt): base min(2 ** attempt, 30) # 1, 2, 4, 8, 16, 30, 30... jitter random.uniform(0, 0.3 * base) return base jitter同时还要限制全局限流同一个Agent在30秒窗口内最多收到N个重试请求超出后直接在调度侧失败快不再发起网络调用。另外同一个DAG内的任务最多只能重试3次超出后整个DAG标记为失败这样不会让下游Agent重复收到同一批任务。这六个坑填完之后Agent-Reach才算真正能扛事。回头看这个项目最深的体会是Agent协作真正难的不是写Agent而是在Agent之间建立一套稳定的通信和调度机制。规则的力度、语义匹配的阈值、重试的退避策略这些参数没有一个公式可以直接套只能根据你的业务场景和实测数据一点一点调。如果你也在做类似的事情建议从最简单的中心调度开始先保证任务能被可靠执行和回执再去追求高级特性。Agent-Reach这套思路你可以直接参考组件拆开单独用也有价值比如注册中心和事件表本身就是一个Agent能力目录。