做Agent开发的人应该都有过这种体验单机单Agent的Demo跑得飞起一旦进入真实业务要同时协调好几个Agent各干各的活就开始乱套。你给A派了任务它在等B的中间结果B又触达不到C封装好的工具任务在中途丢了上下文最后没人知道该把结果回传给谁。我过去一个季度一直在搞一个代号叫Agent-Reach的轻量级编排层核心目标就是解决多智能体协作时互相“触达不到”的问题——能力触达、上下文触达、结果触达。现在核心流程已经跑通这篇文章把完整的设计思路、关键代码、性能取舍和踩过的坑原原本本整理出来给正在做Agent应用落地的朋友一个参考。1. Agent-Reach到底要解决什么问题1.1 单Agent的边界与多Agent协作的痛点很多人对Agent的理解还停留在一个模型实例上。但实际上一个可用的Agent至少要包含模型、系统提示词、工具集合、记忆管理和执行循环五部分。单体Agent的问题在于所有能力被塞进同一个上下文窗口业务一复杂上下文爆掉、工具冲突、指令漂移全都来了。真实业务里更好的做法是拆成多个专职Agent一个负责查数据一个负责写文本一个负责审核各管一段然后通过某种方式把它们串起来。串起来的第一个难点就是“触达”。你不可能把每个Agent的API地址、认证方式、可用能力都硬编码在另外的Agent里那样每加一个Agent所有相关方都要改一遍。Agent-Reach就是把这层“互相找得到、调得通、回得来”的逻辑抽出来做成独立中间层。另一个痛点是上下文可达性。两个Agent协作时A的最终输出可能就是B的输入。如果中间没有统一的消息协议A传给B的内容经常是截断字符串或者带着一堆模型幻觉出来的假字段。任务排到第N步之后出了错想回溯都无从下手。1.2 “触达”的三层含义我把Reach拆解成三个层次。第一层叫能力触达调度器必须知道所有Agent能干什么、不能干什么。第二层叫上下文触达一个任务在执行中产生的中间状态、片段结果每个参与的Agent都能访问到。第三层叫结果触达任务的最终结果要能可靠地回到发起方手里即使中途某个Agent挂了也要换路径或者降级把结论送回去。这三层缺一个都会出问题。能力触达做不到任务派给一个不擅长它的Agent上下文触达做不到Agent各自为政协作变成两个独立的单体结果触达做不到最致命任务成功了但发起方不知道等于白做。1.3 项目定位与适用场景Agent-Reach的定位非常聚焦它是一个面向业务的轻量级任务编排与触达层不是一个通用大模型框架。适合的使用场景包括企业内部有多个领域Agent想统一接入并做路由一个复杂任务需要多个Agent按流程协作且流程可能动态调整已有Agent希望低成本接入统一调度而不是推翻重写。不适合的场景是单Agent单任务那套编排完全多余它也不能替代RAG、向量库这些基础设施只管触达和编排不负责知识检索。常见问题Agent-Reach给出的解法Agent之间不知道对方能干什么统一注册中心每个Agent注册能力描述任务中间状态丢失断了没法续状态集中存储消息带trace_id全程可追踪一个Agent挂了任务链全断调度层做超时重试、降级路由工具调用参数是模型编造的ToolRunner做schema校验错误结构化回传并发一高队列被冲垮扇出预算限制加并发信号量2. 整体设计思路与核心架构2.1 为什么采用注册中心加调度器的三段式我一开始图省事直接让Agent之间互相调用结果不到两周就改成了注册中心集中式。原因是Agent间直连会让调用拓扑变成一团乱麻排查问题像在玩迷宫。集中式的好处有三个一是全局视图所有Agent的能力、负载、状态一目了然二是统一管控限流、鉴权、重试只在调度层做一遍三是链路追踪所有跨Agent的消息都经过同一个入口日志天然带trace_id。架构上分三段接入层、调度层、执行层。接入层接收外部请求统一做协议转换。调度层是核心包含注册中心、路由器和编排引擎。执行层是真正干活的人它们自己不保存业务状态状态统一放在调度层Agent挂了换一个实例顶上就行。2.2 Agent Schema一个让所有Agent“可被发现”的注册协议核心是Agent描述文件我用JSON定义。这个描述文件至少包含name、description、capabilities、input_schema、output_schema、timeout、concurrency_limit、endpoint。description是关键它是给路由器用的要写清楚这个Agent擅长什么。{ name: report_writer, description: 负责根据结构化数据生成中文周报擅长总结、提炼和格式化输出, capabilities: [text_generation, weekly_report], input_schema: { type: object, required: [data], properties: { data: { type: object, description: 统计数据包含指标名与数值 }, style: { type: string, enum: [detailed, concise] } } }, output_schema: { type: object, properties: { report: { type: string } } }, timeout: 45, concurrency_limit: 5, endpoint: http://agent-report-writer:8080/invoke }这里有个非常容易忽略的点input_schema和output_schema必须是严格JSON而且字段描述越具体越好。后面遇到的参数幻觉问题大半是因为描述不清晰导致模型猜字段。路由时我会把每个Agent的description和capabilities全部做成向量索引用embedding做语义匹配再叠加关键词规则兜底。测下来语义匹配准确率大概在80%左右加上兜底规则能到90%以上。2.3 意图路由从自然语言到Agent调用的映射路由是整个Agent-Reach最容易出错但也最值得优化的模块。我的做法是两段式先用一个轻量分类模型判断意图域再在域内做Agent选择。为什么不用单个大模型一步到位因为成本太高且延迟不可控生产环境里每次路由都要跑一次大模型每天几百万次请求成本完全扛不住。域内选择我用的是“语义向量相似度加关键词兜底”。每个Agent的description和capabilities提前算好向量来一个新任务意图先算向量取Top3的候选再根据关键词规则做一次硬约束修正。举例来说意图是“帮我把销售数据整理成周报”向量匹配大概率命中report_writer和data_analyzer关键词规则再确保涉及“周报”时优先进入report_writer。2.4 编排策略串行、并行、动态分发编排引擎支持三种模式。串行适合流程固定的流水线比如数据采集到清洗到分析。并行适合多个独立子任务可以同时执行的情况比如同时让三个Agent分别生成不同区域的销售报告。动态分发最灵活也最难做适合子任务之间有依赖、且依赖关系要在运行时决定的场景比如写一个新产品方案既有可能需要查竞品又可能需要查询内部价格体系顺序取决于前面的结果。我实际项目中大约七成用的是串行加并行的组合动态分发只在复杂推理流程里启用。动态分发必须配人工审核节点全自动喷涌出来的子任务流一旦方向错了后面越跑越偏。3. 关键实现从零搭建最小可用的Agent-Reach3.1 环境准备与技术选型技术选型我一个个说原因。注册中心用Redis因为要支持服务发现和心跳过期Redis的键过期机制正好用上Agent每隔15秒写一次心跳调度器读不到直接标记离线。路由器用FastAPI轻量且异步支持好。编排引擎直接用Python的asyncio避免引入过重的分布式工作流引擎。消息队列先用Redis Stream生产环境量再大可以换Kafka接口层面做好抽象。依赖清单pip install fastapi redis pydantic2.0 jsonschema httpx numpyPython版本用3.11以上。理由很简单asyncio的TaskGroup和asyncio.timeout在3.11才正式可用写编排代码会顺手很多不用自己封装一堆超时工具。3.2 核心数据结构任务状态机与消息体所有跨Agent的消息统一走同一个数据结构我用Pydantic建模。下面是最重要的任务消息体from enum import Enum from typing import Any, Optional from pydantic import BaseModel, Field from datetime import datetime class TaskStatus(str, Enum): PENDING pending RUNNING running BLOCKED blocked # 等待外部依赖 SUCCEEDED succeeded FAILED failed TIMEOUT timeout CANCELLED cancelled class AgentMessage(BaseModel): task_id: str Field(..., description全局唯一任务ID) trace_id: str Field(..., description链路追踪ID跨Agent传递) parent_task_id: Optional[str] None source_agent: str target_agent: Optional[str] None # None表示由路由器决定 intent: str Field(..., description任务的意图描述给路由器用) payload: dict[str, Any] Field(...) status: TaskStatus TaskStatus.PENDING deadline: datetime Field(...) # 绝对截止时间 retry_count: int 0 max_retry: int 3 created_at: datetime Field(default_factorydatetime.utcnow)deadline和retry_count是血的教训换来的。没有deadline的Agent消息就是一颗定时炸弹模型调用一卡整个任务链全部挂起。我后面把所有执行环节都加上deadline超时一律按失败处理并走重试路径。task状态这里有个关键点blocked状态必须单独存在因为很多Agent会等待外部人工确认这个等待不能和失败混为一谈。3.3 调度器核心循环路由、下发、回收调度器的主循环可以简化成三步从队列取消息、路由定目标Agent、下发并跟踪状态。如果目标为空表示这是一个需要路由的消息进入路由器。已完成的任务结果会被写入Redis的result storekey为task_idvalue为JSON序列化结果TTL设为24小时。24小时足够业务方取走结果同时也不会让Redis无限膨胀。核心代码import asyncio from redis.asyncio import Redis class Scheduler: def __init__(self, redis: Redis, router, agents: dict[str, AgentExecutor]): self.redis redis self.router router self.agents agents self.queue reach:pending self.result_prefix reach:result: async def process(self, msg: AgentMessage): if msg.target_agent is None: msg.target_agent await self.router.route(msg) if msg.target_agent is None: await self.mark_failed(msg, no_route) return executor self.agents.get(msg.target_agent) if executor is None: await self.mark_failed(msg, agent_offline) return if not await executor.check_available(): await self.redis.rpush(self.queue, msg.model_dump_json()) return try: async with asyncio.timeout(msg.deadline - datetime.utcnow()): result await executor.invoke(msg) except TimeoutError: await self.handle_timeout(msg) except Exception as e: await self.handle_error(msg, e) async def run(self): while True: raw await self.redis.blpop(self.queue, timeout1) if raw: msg AgentMessage.model_validate_json(raw[1]) asyncio.create_task(self.process(msg))这里每次队列里Agent不可用时就把它重新放回队尾相当于一个简化版的背压机制防止调度器把消息丢给宕机的Agent。but注意这种无脑放回队尾可能造成活锁所以我在真正实现里加了一个attempted_count超过3次就不再放回直接标记失败。超时用的asyncio.timeout3.11以后不需要自己封装。3.4 工具触达层一个注入到Agent的SDKAgent要触达外部工具我封装了一个极简的ToolRunner SDK。Agent不再自己拼HTTP请求而是调用SDK的tool.call(tool_name, params)SDK负责找到工具、填充参数、调用、把结果格式化回填。这里的坑是很多Agent产生的工具调用参数是模型幻觉出来的。解决办法是在SDK里加一层参数校验对照工具的JSON Schema做校验不通过就返回错误让Agent重新生成参数。class ToolRunner: def __init__(self): self.tools {} def register(self, tool_name, schema, handler): self.tools[tool_name] {schema: schema, handler: handler} async def call(self, tool_name: str, params: dict) - dict: if tool_name not in self.tools: return {error: ftool_not_found: {tool_name}} tool self.tools[tool_name] errors validate_json_schema(params, tool[schema]) if errors: return {error: invalid_params, details: errors} try: result await tool[handler](**params) return {ok: True, result: result} except Exception as e: return {error: str(e)}validate_json_schema我直接用的jsonschema库不用手写校验逻辑。很多AI Agent框架的问题是工具调用的失败不会回传给模型导致模型一遍一遍重复同样的错误调用。ToolRunner把错误以结构化方式返回给Agent的上下文让模型有机会修正自己的行为实测能减少约60%的重复性调用错误。这个点非常关键因为LLM一次调用失败后如果不给它反馈它会基于同样的上下文得出同样的幻觉参数。3.5 超时、重试与降级策略重试必须用指数退避加随机抖动。固定间隔重试在大量Agent同时失败时会形成惊群效应所有Agent同时向同一个故障服务发起重试直接把服务打挂。我的重试策略是import random def next_backoff(retry_count: int) - float: base 2 ** retry_count jitter random.uniform(0, base * 0.5) return min(base * 0.5 jitter, 30)第一次重试等1~2秒第二次2~4秒以此类推封顶30秒。降级的逻辑分两种目标Agent不可用就路由到能力最接近的替代Agent替代也没有就返回结构化错误给发起方由上游决定下一步。特别注意一点降级必须在上游Agent的上下文里说明“当前输入来自备用Agent”否则模型可能会基于错误的来源做出错误判断。4. 踩坑实录Agent触达失败的高频问题4.1 Agent卡死的真凶不是模型慢而是死锁我遇到过最诡异的问题是Agent调用另一个Agent时两边互相等对方的响应等到超时。排查半天才发现A在调B的同时B又在等A之前一个任务的释放锁。这是典型的任务级死锁。解决办法是引入任务依赖图调度器在分发前做一次环检测发现依赖环直接拒绝任务并返回错误描述。环检测的实现并不复杂在任务依赖关系写入Redis时用并查集维护联通映射每次新增依赖关系先查一下两端是否已经在一个集合里如果在说明加上这条边就会成环。这个方案能拦截住百分之九十九的动态依赖成环问题。4.2 上下文丢失模型截断和字段缺失上下文丢失最典型的症状是下游Agent收到的payload缺字段。比如上游明明生成了三个指标下游只收到两个。排查下来原因是有一步的模型输出超长被截断或者JSON序列化时非ASCII字符被转义后解析失败。我这个是在调度层统一做一次schema校验校验不通过就阻塞任务并发一条结构化错误。注意这个校验不能只靠模型自己模型输出完全可能不符合约束。要在消息进入队列前用代码校验一次落库后再校验一次双保险。第一次校验防脏数据进入队列第二次校验防止Agent执行过程中的副作用数据污染结果。两个校验都通过上下文才算“触达成功”。4.3 参数幻觉模型给工具编造参数工具调用最有意思的问题是参数幻觉。模型明明没有拿到日期参数却在调用查询工具时编造了一个日期传给工具。解决思路有两个一个是在prompt中要求“不明确的信息必须标记为unknown”另一个是在ToolRunner里对参数做严格校验两者结合效果最好。另外如果在工具的定义中把参数描述写成“如果用户未提供请询问用户再调用”也能显著减少幻觉。我见过最离谱的一次模型在调用查询员工信息工具时自己编了一个“employee_id9527”进去工具自然返回空结果。模型居然还一本正经地往下游输出“该员工编号9527查无此人”。这个案例后来被我写进了测试用例专门用来验证参数校验模块能不能拦截住这种场景。4.4 消息风暴与并发控制一个任务动态分发成100个子任务每个子任务又触发新的子任务几秒钟Redis队列就堆了几百万条消息。解决办法是给每条任务设置生成预算fanout limit同一个父亲任务的直接子任务数量上限通常是20超过就直接拒绝。并发控制方面我在每个Agent上加了信号量超过并发上限的请求排队而不是直接丢弃。这里用asyncio.Semaphore非常方便。信号量有个坑不能只限制执行中任务的数量还要考虑正在排队等待的数量。我是给每个Agent配了一个双信号量前半段限制执行数后半段限制排队数。排队数满就直接返回忙碌错误让上游去重试而不是把所有积压都堆在内存里。4.5 可观测性让每个Agent的触达路径清晰可见没有可观测性Agent编排就是黑盒。我从第一天就用OpenTelemetry标准的trace_id贯穿全过程。每一条Agent消息都带trace_id在调度层统一记录日志。排查问题时用trace_id就能拉出完整调用链看哪一步耗时最长、哪一步失败。Redis里我存了一份最近24小时的任务日志格式是JSON Lines虽然简陋但非常实用。具体的日志格式我设计成一行一个事件包含timestamp、trace_id、task_id、from、to、event_type、latency_ms、error。事件类型有route、dispatch、invoke_start、invoke_end、timeout、retry、degrade。这套日志上线之后排查问题的速度提升了至少一倍很多之前要靠猜的问题现在看一眼日志就能定位。5. 一些体感与后续想法这套东西从设计到跑通核心流程前后不到一个月但它带给我的最大改变不是代码而是对“Agent怎么协作”这件事的理解。一个多Agent系统能正常工作靠的从来不是某个模型有多聪明而是每个环节之间的契约足够清晰、触达足够可靠。我后面复盘时发现真正拖垮系统的往往不是模型能力而是消息协议不规范、超时策略缺失、失败后没有重试路径这些“基础设施问题”。后续我打算把动态编排的能力再补强尤其是任务依赖环检测和子任务回收机制这两个地方目前还是手工处理。子任务回收尤其重要父任务取消了它派出去的一堆子任务还在跑白白消耗算力。另外还想加一个Agent健康度评分把错误率、延迟、队列堆积量算成一个分数路由时优先选健康度高的Agent这比单纯的轮询要聪明得多。也欢迎正在做Agent编排的朋友一起交流踩坑经验。 Agent-Reach：解决多智能体协作触达问题的轻量级编排层
做Agent开发的人应该都有过这种体验单机单Agent的Demo跑得飞起一旦进入真实业务要同时协调好几个Agent各干各的活就开始乱套。你给A派了任务它在等B的中间结果B又触达不到C封装好的工具任务在中途丢了上下文最后没人知道该把结果回传给谁。我过去一个季度一直在搞一个代号叫Agent-Reach的轻量级编排层核心目标就是解决多智能体协作时互相“触达不到”的问题——能力触达、上下文触达、结果触达。现在核心流程已经跑通这篇文章把完整的设计思路、关键代码、性能取舍和踩过的坑原原本本整理出来给正在做Agent应用落地的朋友一个参考。1. Agent-Reach到底要解决什么问题1.1 单Agent的边界与多Agent协作的痛点很多人对Agent的理解还停留在一个模型实例上。但实际上一个可用的Agent至少要包含模型、系统提示词、工具集合、记忆管理和执行循环五部分。单体Agent的问题在于所有能力被塞进同一个上下文窗口业务一复杂上下文爆掉、工具冲突、指令漂移全都来了。真实业务里更好的做法是拆成多个专职Agent一个负责查数据一个负责写文本一个负责审核各管一段然后通过某种方式把它们串起来。串起来的第一个难点就是“触达”。你不可能把每个Agent的API地址、认证方式、可用能力都硬编码在另外的Agent里那样每加一个Agent所有相关方都要改一遍。Agent-Reach就是把这层“互相找得到、调得通、回得来”的逻辑抽出来做成独立中间层。另一个痛点是上下文可达性。两个Agent协作时A的最终输出可能就是B的输入。如果中间没有统一的消息协议A传给B的内容经常是截断字符串或者带着一堆模型幻觉出来的假字段。任务排到第N步之后出了错想回溯都无从下手。1.2 “触达”的三层含义我把Reach拆解成三个层次。第一层叫能力触达调度器必须知道所有Agent能干什么、不能干什么。第二层叫上下文触达一个任务在执行中产生的中间状态、片段结果每个参与的Agent都能访问到。第三层叫结果触达任务的最终结果要能可靠地回到发起方手里即使中途某个Agent挂了也要换路径或者降级把结论送回去。这三层缺一个都会出问题。能力触达做不到任务派给一个不擅长它的Agent上下文触达做不到Agent各自为政协作变成两个独立的单体结果触达做不到最致命任务成功了但发起方不知道等于白做。1.3 项目定位与适用场景Agent-Reach的定位非常聚焦它是一个面向业务的轻量级任务编排与触达层不是一个通用大模型框架。适合的使用场景包括企业内部有多个领域Agent想统一接入并做路由一个复杂任务需要多个Agent按流程协作且流程可能动态调整已有Agent希望低成本接入统一调度而不是推翻重写。不适合的场景是单Agent单任务那套编排完全多余它也不能替代RAG、向量库这些基础设施只管触达和编排不负责知识检索。常见问题Agent-Reach给出的解法Agent之间不知道对方能干什么统一注册中心每个Agent注册能力描述任务中间状态丢失断了没法续状态集中存储消息带trace_id全程可追踪一个Agent挂了任务链全断调度层做超时重试、降级路由工具调用参数是模型编造的ToolRunner做schema校验错误结构化回传并发一高队列被冲垮扇出预算限制加并发信号量2. 整体设计思路与核心架构2.1 为什么采用注册中心加调度器的三段式我一开始图省事直接让Agent之间互相调用结果不到两周就改成了注册中心集中式。原因是Agent间直连会让调用拓扑变成一团乱麻排查问题像在玩迷宫。集中式的好处有三个一是全局视图所有Agent的能力、负载、状态一目了然二是统一管控限流、鉴权、重试只在调度层做一遍三是链路追踪所有跨Agent的消息都经过同一个入口日志天然带trace_id。架构上分三段接入层、调度层、执行层。接入层接收外部请求统一做协议转换。调度层是核心包含注册中心、路由器和编排引擎。执行层是真正干活的人它们自己不保存业务状态状态统一放在调度层Agent挂了换一个实例顶上就行。2.2 Agent Schema一个让所有Agent“可被发现”的注册协议核心是Agent描述文件我用JSON定义。这个描述文件至少包含name、description、capabilities、input_schema、output_schema、timeout、concurrency_limit、endpoint。description是关键它是给路由器用的要写清楚这个Agent擅长什么。{ name: report_writer, description: 负责根据结构化数据生成中文周报擅长总结、提炼和格式化输出, capabilities: [text_generation, weekly_report], input_schema: { type: object, required: [data], properties: { data: { type: object, description: 统计数据包含指标名与数值 }, style: { type: string, enum: [detailed, concise] } } }, output_schema: { type: object, properties: { report: { type: string } } }, timeout: 45, concurrency_limit: 5, endpoint: http://agent-report-writer:8080/invoke }这里有个非常容易忽略的点input_schema和output_schema必须是严格JSON而且字段描述越具体越好。后面遇到的参数幻觉问题大半是因为描述不清晰导致模型猜字段。路由时我会把每个Agent的description和capabilities全部做成向量索引用embedding做语义匹配再叠加关键词规则兜底。测下来语义匹配准确率大概在80%左右加上兜底规则能到90%以上。2.3 意图路由从自然语言到Agent调用的映射路由是整个Agent-Reach最容易出错但也最值得优化的模块。我的做法是两段式先用一个轻量分类模型判断意图域再在域内做Agent选择。为什么不用单个大模型一步到位因为成本太高且延迟不可控生产环境里每次路由都要跑一次大模型每天几百万次请求成本完全扛不住。域内选择我用的是“语义向量相似度加关键词兜底”。每个Agent的description和capabilities提前算好向量来一个新任务意图先算向量取Top3的候选再根据关键词规则做一次硬约束修正。举例来说意图是“帮我把销售数据整理成周报”向量匹配大概率命中report_writer和data_analyzer关键词规则再确保涉及“周报”时优先进入report_writer。2.4 编排策略串行、并行、动态分发编排引擎支持三种模式。串行适合流程固定的流水线比如数据采集到清洗到分析。并行适合多个独立子任务可以同时执行的情况比如同时让三个Agent分别生成不同区域的销售报告。动态分发最灵活也最难做适合子任务之间有依赖、且依赖关系要在运行时决定的场景比如写一个新产品方案既有可能需要查竞品又可能需要查询内部价格体系顺序取决于前面的结果。我实际项目中大约七成用的是串行加并行的组合动态分发只在复杂推理流程里启用。动态分发必须配人工审核节点全自动喷涌出来的子任务流一旦方向错了后面越跑越偏。3. 关键实现从零搭建最小可用的Agent-Reach3.1 环境准备与技术选型技术选型我一个个说原因。注册中心用Redis因为要支持服务发现和心跳过期Redis的键过期机制正好用上Agent每隔15秒写一次心跳调度器读不到直接标记离线。路由器用FastAPI轻量且异步支持好。编排引擎直接用Python的asyncio避免引入过重的分布式工作流引擎。消息队列先用Redis Stream生产环境量再大可以换Kafka接口层面做好抽象。依赖清单pip install fastapi redis pydantic2.0 jsonschema httpx numpyPython版本用3.11以上。理由很简单asyncio的TaskGroup和asyncio.timeout在3.11才正式可用写编排代码会顺手很多不用自己封装一堆超时工具。3.2 核心数据结构任务状态机与消息体所有跨Agent的消息统一走同一个数据结构我用Pydantic建模。下面是最重要的任务消息体from enum import Enum from typing import Any, Optional from pydantic import BaseModel, Field from datetime import datetime class TaskStatus(str, Enum): PENDING pending RUNNING running BLOCKED blocked # 等待外部依赖 SUCCEEDED succeeded FAILED failed TIMEOUT timeout CANCELLED cancelled class AgentMessage(BaseModel): task_id: str Field(..., description全局唯一任务ID) trace_id: str Field(..., description链路追踪ID跨Agent传递) parent_task_id: Optional[str] None source_agent: str target_agent: Optional[str] None # None表示由路由器决定 intent: str Field(..., description任务的意图描述给路由器用) payload: dict[str, Any] Field(...) status: TaskStatus TaskStatus.PENDING deadline: datetime Field(...) # 绝对截止时间 retry_count: int 0 max_retry: int 3 created_at: datetime Field(default_factorydatetime.utcnow)deadline和retry_count是血的教训换来的。没有deadline的Agent消息就是一颗定时炸弹模型调用一卡整个任务链全部挂起。我后面把所有执行环节都加上deadline超时一律按失败处理并走重试路径。task状态这里有个关键点blocked状态必须单独存在因为很多Agent会等待外部人工确认这个等待不能和失败混为一谈。3.3 调度器核心循环路由、下发、回收调度器的主循环可以简化成三步从队列取消息、路由定目标Agent、下发并跟踪状态。如果目标为空表示这是一个需要路由的消息进入路由器。已完成的任务结果会被写入Redis的result storekey为task_idvalue为JSON序列化结果TTL设为24小时。24小时足够业务方取走结果同时也不会让Redis无限膨胀。核心代码import asyncio from redis.asyncio import Redis class Scheduler: def __init__(self, redis: Redis, router, agents: dict[str, AgentExecutor]): self.redis redis self.router router self.agents agents self.queue reach:pending self.result_prefix reach:result: async def process(self, msg: AgentMessage): if msg.target_agent is None: msg.target_agent await self.router.route(msg) if msg.target_agent is None: await self.mark_failed(msg, no_route) return executor self.agents.get(msg.target_agent) if executor is None: await self.mark_failed(msg, agent_offline) return if not await executor.check_available(): await self.redis.rpush(self.queue, msg.model_dump_json()) return try: async with asyncio.timeout(msg.deadline - datetime.utcnow()): result await executor.invoke(msg) except TimeoutError: await self.handle_timeout(msg) except Exception as e: await self.handle_error(msg, e) async def run(self): while True: raw await self.redis.blpop(self.queue, timeout1) if raw: msg AgentMessage.model_validate_json(raw[1]) asyncio.create_task(self.process(msg))这里每次队列里Agent不可用时就把它重新放回队尾相当于一个简化版的背压机制防止调度器把消息丢给宕机的Agent。but注意这种无脑放回队尾可能造成活锁所以我在真正实现里加了一个attempted_count超过3次就不再放回直接标记失败。超时用的asyncio.timeout3.11以后不需要自己封装。3.4 工具触达层一个注入到Agent的SDKAgent要触达外部工具我封装了一个极简的ToolRunner SDK。Agent不再自己拼HTTP请求而是调用SDK的tool.call(tool_name, params)SDK负责找到工具、填充参数、调用、把结果格式化回填。这里的坑是很多Agent产生的工具调用参数是模型幻觉出来的。解决办法是在SDK里加一层参数校验对照工具的JSON Schema做校验不通过就返回错误让Agent重新生成参数。class ToolRunner: def __init__(self): self.tools {} def register(self, tool_name, schema, handler): self.tools[tool_name] {schema: schema, handler: handler} async def call(self, tool_name: str, params: dict) - dict: if tool_name not in self.tools: return {error: ftool_not_found: {tool_name}} tool self.tools[tool_name] errors validate_json_schema(params, tool[schema]) if errors: return {error: invalid_params, details: errors} try: result await tool[handler](**params) return {ok: True, result: result} except Exception as e: return {error: str(e)}validate_json_schema我直接用的jsonschema库不用手写校验逻辑。很多AI Agent框架的问题是工具调用的失败不会回传给模型导致模型一遍一遍重复同样的错误调用。ToolRunner把错误以结构化方式返回给Agent的上下文让模型有机会修正自己的行为实测能减少约60%的重复性调用错误。这个点非常关键因为LLM一次调用失败后如果不给它反馈它会基于同样的上下文得出同样的幻觉参数。3.5 超时、重试与降级策略重试必须用指数退避加随机抖动。固定间隔重试在大量Agent同时失败时会形成惊群效应所有Agent同时向同一个故障服务发起重试直接把服务打挂。我的重试策略是import random def next_backoff(retry_count: int) - float: base 2 ** retry_count jitter random.uniform(0, base * 0.5) return min(base * 0.5 jitter, 30)第一次重试等1~2秒第二次2~4秒以此类推封顶30秒。降级的逻辑分两种目标Agent不可用就路由到能力最接近的替代Agent替代也没有就返回结构化错误给发起方由上游决定下一步。特别注意一点降级必须在上游Agent的上下文里说明“当前输入来自备用Agent”否则模型可能会基于错误的来源做出错误判断。4. 踩坑实录Agent触达失败的高频问题4.1 Agent卡死的真凶不是模型慢而是死锁我遇到过最诡异的问题是Agent调用另一个Agent时两边互相等对方的响应等到超时。排查半天才发现A在调B的同时B又在等A之前一个任务的释放锁。这是典型的任务级死锁。解决办法是引入任务依赖图调度器在分发前做一次环检测发现依赖环直接拒绝任务并返回错误描述。环检测的实现并不复杂在任务依赖关系写入Redis时用并查集维护联通映射每次新增依赖关系先查一下两端是否已经在一个集合里如果在说明加上这条边就会成环。这个方案能拦截住百分之九十九的动态依赖成环问题。4.2 上下文丢失模型截断和字段缺失上下文丢失最典型的症状是下游Agent收到的payload缺字段。比如上游明明生成了三个指标下游只收到两个。排查下来原因是有一步的模型输出超长被截断或者JSON序列化时非ASCII字符被转义后解析失败。我这个是在调度层统一做一次schema校验校验不通过就阻塞任务并发一条结构化错误。注意这个校验不能只靠模型自己模型输出完全可能不符合约束。要在消息进入队列前用代码校验一次落库后再校验一次双保险。第一次校验防脏数据进入队列第二次校验防止Agent执行过程中的副作用数据污染结果。两个校验都通过上下文才算“触达成功”。4.3 参数幻觉模型给工具编造参数工具调用最有意思的问题是参数幻觉。模型明明没有拿到日期参数却在调用查询工具时编造了一个日期传给工具。解决思路有两个一个是在prompt中要求“不明确的信息必须标记为unknown”另一个是在ToolRunner里对参数做严格校验两者结合效果最好。另外如果在工具的定义中把参数描述写成“如果用户未提供请询问用户再调用”也能显著减少幻觉。我见过最离谱的一次模型在调用查询员工信息工具时自己编了一个“employee_id9527”进去工具自然返回空结果。模型居然还一本正经地往下游输出“该员工编号9527查无此人”。这个案例后来被我写进了测试用例专门用来验证参数校验模块能不能拦截住这种场景。4.4 消息风暴与并发控制一个任务动态分发成100个子任务每个子任务又触发新的子任务几秒钟Redis队列就堆了几百万条消息。解决办法是给每条任务设置生成预算fanout limit同一个父亲任务的直接子任务数量上限通常是20超过就直接拒绝。并发控制方面我在每个Agent上加了信号量超过并发上限的请求排队而不是直接丢弃。这里用asyncio.Semaphore非常方便。信号量有个坑不能只限制执行中任务的数量还要考虑正在排队等待的数量。我是给每个Agent配了一个双信号量前半段限制执行数后半段限制排队数。排队数满就直接返回忙碌错误让上游去重试而不是把所有积压都堆在内存里。4.5 可观测性让每个Agent的触达路径清晰可见没有可观测性Agent编排就是黑盒。我从第一天就用OpenTelemetry标准的trace_id贯穿全过程。每一条Agent消息都带trace_id在调度层统一记录日志。排查问题时用trace_id就能拉出完整调用链看哪一步耗时最长、哪一步失败。Redis里我存了一份最近24小时的任务日志格式是JSON Lines虽然简陋但非常实用。具体的日志格式我设计成一行一个事件包含timestamp、trace_id、task_id、from、to、event_type、latency_ms、error。事件类型有route、dispatch、invoke_start、invoke_end、timeout、retry、degrade。这套日志上线之后排查问题的速度提升了至少一倍很多之前要靠猜的问题现在看一眼日志就能定位。5. 一些体感与后续想法这套东西从设计到跑通核心流程前后不到一个月但它带给我的最大改变不是代码而是对“Agent怎么协作”这件事的理解。一个多Agent系统能正常工作靠的从来不是某个模型有多聪明而是每个环节之间的契约足够清晰、触达足够可靠。我后面复盘时发现真正拖垮系统的往往不是模型能力而是消息协议不规范、超时策略缺失、失败后没有重试路径这些“基础设施问题”。后续我打算把动态编排的能力再补强尤其是任务依赖环检测和子任务回收机制这两个地方目前还是手工处理。子任务回收尤其重要父任务取消了它派出去的一堆子任务还在跑白白消耗算力。另外还想加一个Agent健康度评分把错误率、延迟、队列堆积量算成一个分数路由时优先选健康度高的Agent这比单纯的轮询要聪明得多。也欢迎正在做Agent编排的朋友一起交流踩坑经验。