1. 从“agency-agents”这个标题说起它到底是什么第一次看到“agency-agents”这个词很多人会愣一下——它不像某个具体框架的名字也不像某个开源库的标准命名。但如果你在自动化、任务编排、智能体协作这些圈子里待过一阵子就会意识到这个词其实指向一个非常具体的东西一套让多个智能体agents像一家代理机构agency那样协同工作的系统或模式。拆开来看“agency”在这里不是“代理公司”的字面意思而是强调一种组织化的协作结构——有分工、有角色、有调度、有交付。“agents”则是执行单元可以是基于大模型的智能体也可以是脚本化的任务执行器甚至可以是人工节点的抽象封装。合在一起agency-agents 描述的是把一群各自独立的执行单元组织成一个能接活、能拆活、能干活、能交活的协作网络。它能解决什么问题最直接的痛点是单个智能体能力再强面对复杂任务时也会力不从心。比如你要做一个完整的市场调研报告涉及数据采集、清洗、分析、可视化、文案撰写、格式排版单靠一个 agent 串行做完不仅慢而且每一步的上下文会互相干扰。agency-agents 的思路就是把这些环节拆开交给不同的专职 agent再通过一个调度层把它们串起来。适合谁来参考三类人最值得看一是正在做多智能体编排的开发者二是想把重复性工作流自动化的效率工具爱好者三是需要理解“任务分解与协作”这套思维的产品经理和项目负责人。哪怕你暂时不写代码理解这套协作逻辑对设计任何复杂流程都有帮助。2. 整体设计思路为什么不是“一个大而全的智能体”2.1 单智能体方案的三个硬伤在 agency-agents 这套思路流行之前大多数人的做法是写一个超级 prompt把任务的所有要求、背景、输出格式全塞进去让一个模型一次性完成。我早期也这么干过实测下来有三个绕不过去的坑。第一个坑是上下文污染。当你让同一个 agent 先做数据整理再做文案撰写时它在整理阶段产生的中间推理、试错记录、废弃方案都会留在上下文里干扰后续步骤的判断。就像让一个人先算账再写诗算账时的数字思维会拖累写诗的状态。第二个坑是能力错配。不同任务对模型能力的要求不一样。数据抽取需要强结构化输出能力文案润色需要强语言生成能力代码生成需要强逻辑推理能力。用一个通用 prompt 去覆盖所有环节等于让一个全科医生去做心脏手术不是不能做但风险高。第三个坑是无法并行。单智能体天然是串行的第一步没做完第二步没法开始。而实际工作中很多环节是可以并行的比如数据采集和模板设计完全可以同时进行。2.2 agency-agents 的核心设计原则针对上面三个问题agency-agents 的设计遵循四条原则。角色隔离每个 agent 只负责一个明确的职能有自己的系统提示词、工具集和输出规范。角色之间通过标准化的消息格式通信不共享内部推理过程。这就像公司里不同部门用邮件沟通而不是所有人挤在一个脑子里想事情。调度中心化有一个 orchestrator调度器负责接收总任务、拆解子任务、分配给对应 agent、收集结果、处理异常。调度器不干具体的活只做路由和状态管理。这个设计的关键在于调度逻辑和执行业务逻辑分离方便单独优化。消息标准化agent 之间的输入输出必须遵循统一的结构通常是一个包含 task_id、role、payload、status 的 JSON 对象。标准化带来的好处是任何一个 agent 都可以被替换或升级只要它遵守消息协议整个系统不受影响。状态可追溯每个子任务的执行状态、耗时、输出结果都要落盘记录。这不是为了好看而是为了排查问题。多 agent 系统出问题时最难定位的就是“到底是哪个环节出了错”没有状态记录基本没法调试。2.3 和常见编排方案的对比方案类型代表思路优势劣势单智能体串行一个 prompt 走天下实现简单无需调度上下文污染无法并行能力错配固定流水线硬编码步骤顺序稳定可控易调试缺乏灵活性无法动态调整agency-agents角色隔离动态调度灵活、可扩展、可并行实现复杂度高需要状态管理纯人工分工人肉拆解任务判断力最强慢成本高无法规模化选择 agency-agents 而不是固定流水线核心考量是任务的不确定性。如果你的任务每次都是完全一样的步骤那硬编码流水线就够了。但现实中很多任务需要根据中间结果动态决定下一步比如调研发现某个方向数据不足需要临时增加一个采集 agent这种动态性只有调度中心化才能支撑。3. 核心细节解析角色设计、通信协议与调度逻辑3.1 角色怎么划分才合理角色划分是 agency-agents 最考验设计功力的地方。划得太粗等于没拆划得太细调度开销比干活还大。我踩过的坑是一开始按“数据 agent、分析 agent、写作 agent”三分结果发现数据 agent 既要采集又要清洗内部逻辑还是很重。后来调整成按输入输出边界来划分一个 agent 的输入和输出如果可以用同一种数据格式描述就合并如果格式差异大就拆开。比如采集 agent 输出原始 HTML清洗 agent 输出结构化 JSON这两个格式差异大必须拆。而清洗和标准化都输出 JSON可以合并成一个数据处理 agent。常见的角色划分参考采集 agent负责从各种来源获取原始数据输出原始内容处理 agent负责清洗、转换、结构化输出规范数据分析 agent负责计算、推理、归纳输出结论生成 agent负责文案、代码、报告撰写输出成品校验 agent负责检查输出质量输出通过/不通过及原因调度 agent负责拆解和路由不直接产出业务内容注意校验 agent 很容易被忽略但它是保证质量的关键。没有校验环节错误会一路传递到最终输出返工成本极高。3.2 通信协议的设计要点agent 之间怎么说话决定了系统能不能稳定运行。我推荐用最朴素的 JSON 结构不要一上来就搞复杂的消息队列或 RPC。{ task_id: t-20240101-001, from: orchestrator, to: collector, action: collect, payload: { source: https://example.com/data, format: html }, status: pending, timestamp: 2024-01-01T10:00:00Z }这个结构里task_id 是全局唯一标识用于追踪from/to 标明消息流向action 是动作类型payload 是具体参数status 是当前状态timestamp 用于超时判断。关键设计点payload 里不要塞大段文本。如果采集 agent 抓回来一篇长文不要把全文放进 payload 传给下一个 agent而是存到共享存储文件或数据库payload 里只放存储路径。这样做的好处是消息体小、传输快、也方便单独查看中间产物。3.3 调度逻辑的三种模式调度器怎么决定下一步做什么直接决定了系统的智能程度。我实践下来总结了三档模式可以按需选择。静态模式预先定义好步骤顺序调度器按顺序执行。适合流程固定的场景实现最简单但灵活性差。条件模式在静态基础上增加判断节点根据上一步的输出决定下一步走哪个分支。比如校验不通过就回到处理环节重做。适合有明确质量标准的场景。动态模式调度器本身也是一个 agent根据当前状态和剩余任务动态决定下一个该调用谁。灵活性最高但调试难度也最大容易陷入死循环。我的建议是从静态模式开始遇到瓶颈再升级。很多团队一上来就追求动态调度结果系统复杂度爆炸连基本流程都跑不稳。4. 实操过程从零搭一个最小可用的 agency-agents 系统4.1 环境准备与目录结构先说明这里给的是一个最小可用版本不依赖任何特定框架用 Python 标准库加一个 HTTP 请求库就能跑起来。这样做的原因是框架会掩盖很多细节而理解细节才是掌握这套思路的关键。目录结构建议这样组织agency-agents/ ├── orchestrator.py # 调度器主逻辑 ├── agents/ │ ├── collector.py # 采集 agent │ ├── processor.py # 处理 agent │ ├── analyzer.py # 分析 agent │ └── writer.py # 生成 agent ├── shared/ │ ├── storage/ # 中间产物存储 │ └── protocol.py # 消息协议定义 └── config.yaml # 角色配置与路由规则每个 agent 是一个独立的 Python 脚本通过命令行参数或标准输入接收消息处理完后把结果写到标准输出或存储路径。调度器负责按顺序调用它们。4.2 消息协议的具体实现在shared/protocol.py里定义消息的构造和解析函数import json import uuid from datetime import datetime def make_message(from_role, to_role, action, payload): return { task_id: str(uuid.uuid4()), from: from_role, to: to_role, action: action, payload: payload, status: pending, timestamp: datetime.utcnow().isoformat() } def parse_message(raw): return json.loads(raw) def save_artifact(task_id, name, content): path fshared/storage/{task_id}_{name}.json with open(path, w, encodingutf-8) as f: json.dump(content, f, ensure_asciiFalse, indent2) return path这里save_artifact是关键它把中间产物落盘返回路径给消息 payload 使用。这样消息体始终保持轻量。4.3 调度器的最小实现调度器不需要很复杂核心就是一个循环读任务、拆步骤、调 agent、收结果、判断是否继续。import subprocess import json from shared.protocol import make_message, parse_message def call_agent(agent_name, message): result subprocess.run( [python, fagents/{agent_name}.py], inputjson.dumps(message), capture_outputTrue, textTrue ) return parse_message(result.stdout) def run_task(initial_input): task_id t-001 # 第一步采集 msg make_message(orchestrator, collector, collect, {input: initial_input}) msg[task_id] task_id result call_agent(collector, msg) # 第二步处理 msg make_message(orchestrator, processor, process, result[payload]) msg[task_id] task_id result call_agent(processor, msg) # 第三步分析 msg make_message(orchestrator, analyzer, analyze, result[payload]) msg[task_id] task_id result call_agent(analyzer, msg) # 第四步生成 msg make_message(orchestrator, writer, write, result[payload]) msg[task_id] task_id result call_agent(writer, msg) return result[payload]这个版本是静态模式步骤写死了。但结构已经清晰每个 agent 只关心自己的输入和输出调度器负责串联。4.4 单个 agent 的写法示例以采集 agent 为例它只做一件事根据输入获取原始内容存盘返回路径。import sys import json import requests from shared.protocol import parse_message, save_artifact def main(): raw sys.stdin.read() msg parse_message(raw) source msg[payload][input] response requests.get(source, timeout30) artifact_path save_artifact( msg[task_id], raw_content, {content: response.text, status_code: response.status_code} ) output { task_id: msg[task_id], from: collector, to: orchestrator, action: collect_done, payload: {artifact: artifact_path}, status: success, timestamp: msg[timestamp] } print(json.dumps(output, ensure_asciiFalse)) if __name__ __main__: main()注意这里 agent 不直接返回内容而是返回存储路径。这是 agency-agents 的一个核心习惯消息传引用不传实体。4.5 参数选择与性能考量超时时间设多少我一般设 30 秒给采集类 agent60 秒给分析类 agent。依据是采集主要是网络 IO30 秒足够分析涉及模型推理给足 60 秒避免误杀。并发怎么处理最小版本是串行的但如果你的采集 agent 需要抓多个来源可以在采集 agent 内部用线程池并发而不是在调度层并发。原因是调度层并发会带来状态同步问题而单个 agent 内部的并发是局部的好控制。存储用什么小规模用本地文件就够了task_id 做前缀避免冲突。规模大了再换 SQLite 或对象存储。不要一上来就上数据库增加不必要的复杂度。5. 常见问题与排查技巧实录5.1 问题速查表现象可能原因排查方向解决思路某个 agent 无输出脚本报错但异常被吞检查 stderr在调度器里捕获并打印 stderr消息解析失败JSON 格式不合法打印原始输出统一用 json.dumps 生成禁止手工拼接任务卡住不结束某步超时未处理加超时机制subprocess 加 timeout 参数中间产物找不到路径拼接错误检查 task_id 一致性统一用 save_artifact 生成路径结果质量差上游数据有问题逐环节检查产物从采集开始逐步验证重复执行同一任务状态未持久化检查状态记录每步完成后写状态文件5.2 三个我踩过的坑第一个坑agent 之间直接互相调用。早期我图省事让处理 agent 直接调用分析 agent跳过了调度器。结果流程一复杂调用链变成一团乱麻根本理不清谁调了谁。后来强制规定所有 agent 只能和调度器通信agent 之间禁止直接调用。这条规矩立下来之后系统清晰多了。第二个坑payload 里塞大文本。有一次采集 agent 把整篇几万字的文章塞进 payload 传给下一个 agent结果消息体过大解析慢日志也没法看。改成存盘传路径后不仅速度快了调试时还能直接打开中间文件看内容。第三个坑没有校验环节。有段时间系统跑得挺顺但最终输出质量忽好忽坏。后来加了一个校验 agent专门检查上游输出是否符合预期格式和基本质量要求不合格就打回重做。加了这一环之后最终输出的稳定性明显提升。提示校验 agent 的判断标准要写清楚不要用“质量好不好”这种模糊描述而是用“是否包含必要字段”“长度是否在范围内”“是否包含禁用词”这类可量化规则。5.3 调试技巧单步执行与产物回放多 agent 系统最难的是调试。我的做法是给调度器加一个--step参数每执行完一个 agent 就暂停打印当前消息和产物路径人工确认后再继续。这样能精确定位是哪一步出了问题。另一个技巧是产物回放把某次任务的中间产物全部保留下次调试时直接从某个环节开始不用每次都从头跑。这在优化某个特定 agent 时特别有用。6. 扩展方向从最小系统到生产可用6.1 加入重试与降级机制最小版本没有重试一个 agent 失败整个任务就挂了。生产环境必须加重试同一个 agent 失败后隔几秒重试最多重试三次。三次都失败要么降级到备用 agent要么标记任务失败并通知人工介入。重试的关键是幂等性同一个消息重试多次结果应该一致。所以 agent 内部如果有写操作要用 task_id 做去重判断。6.2 引入优先级与队列当任务多了之后需要区分优先级。比如用户直接触发的任务优先于定时批处理任务。实现方式是在消息里加 priority 字段调度器维护一个优先队列高优先级的先出队。队列的另一个好处是削峰填谷任务提交速度超过处理速度时先入队慢慢消化避免系统被压垮。6.3 监控与告警生产系统必须知道每个 agent 的健康状况。最基本的监控指标每个 agent 的调用次数、成功率、平均耗时、失败原因分布。这些数据可以从状态记录里统计出来。告警规则可以设某个 agent 连续失败超过 5 次或者平均耗时突然翻倍就触发告警。告警渠道用邮件或即时通讯工具都行关键是有人能看到并处理。6.4 角色动态注册最小版本里 agent 是写死在调度器里的。更灵活的做法是让 agent 启动时向调度器注册自己声明自己能处理哪些 action。调度器维护一个注册表收到任务时查表决定调用谁。这样新增 agent 不需要改调度器代码扩展性更好。注册消息可以长这样{ role: collector, actions: [collect, collect_batch], endpoint: agents/collector.py, max_concurrency: 3 }调度器收到注册后把信息存到注册表。任务来了根据 action 查注册表找到对应的 agent 列表再按负载情况选一个调用。这套机制搭起来之后整个系统就从“写死的流水线”变成了“可插拔的协作网络”这才是 agency-agents 这个思路真正发挥价值的地方。我自己的体会是前期多花点时间把协议和调度逻辑设计好后期加功能就是加一个 agent 文件的事几乎不用动核心代码。反过来如果一开始图快把逻辑写死后面每加一个功能都要伤筋动骨。 
长河君 · 2026/10/26 9:23:59 · 阅读 1.2k agency-agents：多智能体协作编排系统设计与实现
1. 从“agency-agents”这个标题说起它到底是什么第一次看到“agency-agents”这个词很多人会愣一下——它不像某个具体框架的名字也不像某个开源库的标准命名。但如果你在自动化、任务编排、智能体协作这些圈子里待过一阵子就会意识到这个词其实指向一个非常具体的东西一套让多个智能体agents像一家代理机构agency那样协同工作的系统或模式。拆开来看“agency”在这里不是“代理公司”的字面意思而是强调一种组织化的协作结构——有分工、有角色、有调度、有交付。“agents”则是执行单元可以是基于大模型的智能体也可以是脚本化的任务执行器甚至可以是人工节点的抽象封装。合在一起agency-agents 描述的是把一群各自独立的执行单元组织成一个能接活、能拆活、能干活、能交活的协作网络。它能解决什么问题最直接的痛点是单个智能体能力再强面对复杂任务时也会力不从心。比如你要做一个完整的市场调研报告涉及数据采集、清洗、分析、可视化、文案撰写、格式排版单靠一个 agent 串行做完不仅慢而且每一步的上下文会互相干扰。agency-agents 的思路就是把这些环节拆开交给不同的专职 agent再通过一个调度层把它们串起来。适合谁来参考三类人最值得看一是正在做多智能体编排的开发者二是想把重复性工作流自动化的效率工具爱好者三是需要理解“任务分解与协作”这套思维的产品经理和项目负责人。哪怕你暂时不写代码理解这套协作逻辑对设计任何复杂流程都有帮助。2. 整体设计思路为什么不是“一个大而全的智能体”2.1 单智能体方案的三个硬伤在 agency-agents 这套思路流行之前大多数人的做法是写一个超级 prompt把任务的所有要求、背景、输出格式全塞进去让一个模型一次性完成。我早期也这么干过实测下来有三个绕不过去的坑。第一个坑是上下文污染。当你让同一个 agent 先做数据整理再做文案撰写时它在整理阶段产生的中间推理、试错记录、废弃方案都会留在上下文里干扰后续步骤的判断。就像让一个人先算账再写诗算账时的数字思维会拖累写诗的状态。第二个坑是能力错配。不同任务对模型能力的要求不一样。数据抽取需要强结构化输出能力文案润色需要强语言生成能力代码生成需要强逻辑推理能力。用一个通用 prompt 去覆盖所有环节等于让一个全科医生去做心脏手术不是不能做但风险高。第三个坑是无法并行。单智能体天然是串行的第一步没做完第二步没法开始。而实际工作中很多环节是可以并行的比如数据采集和模板设计完全可以同时进行。2.2 agency-agents 的核心设计原则针对上面三个问题agency-agents 的设计遵循四条原则。角色隔离每个 agent 只负责一个明确的职能有自己的系统提示词、工具集和输出规范。角色之间通过标准化的消息格式通信不共享内部推理过程。这就像公司里不同部门用邮件沟通而不是所有人挤在一个脑子里想事情。调度中心化有一个 orchestrator调度器负责接收总任务、拆解子任务、分配给对应 agent、收集结果、处理异常。调度器不干具体的活只做路由和状态管理。这个设计的关键在于调度逻辑和执行业务逻辑分离方便单独优化。消息标准化agent 之间的输入输出必须遵循统一的结构通常是一个包含 task_id、role、payload、status 的 JSON 对象。标准化带来的好处是任何一个 agent 都可以被替换或升级只要它遵守消息协议整个系统不受影响。状态可追溯每个子任务的执行状态、耗时、输出结果都要落盘记录。这不是为了好看而是为了排查问题。多 agent 系统出问题时最难定位的就是“到底是哪个环节出了错”没有状态记录基本没法调试。2.3 和常见编排方案的对比方案类型代表思路优势劣势单智能体串行一个 prompt 走天下实现简单无需调度上下文污染无法并行能力错配固定流水线硬编码步骤顺序稳定可控易调试缺乏灵活性无法动态调整agency-agents角色隔离动态调度灵活、可扩展、可并行实现复杂度高需要状态管理纯人工分工人肉拆解任务判断力最强慢成本高无法规模化选择 agency-agents 而不是固定流水线核心考量是任务的不确定性。如果你的任务每次都是完全一样的步骤那硬编码流水线就够了。但现实中很多任务需要根据中间结果动态决定下一步比如调研发现某个方向数据不足需要临时增加一个采集 agent这种动态性只有调度中心化才能支撑。3. 核心细节解析角色设计、通信协议与调度逻辑3.1 角色怎么划分才合理角色划分是 agency-agents 最考验设计功力的地方。划得太粗等于没拆划得太细调度开销比干活还大。我踩过的坑是一开始按“数据 agent、分析 agent、写作 agent”三分结果发现数据 agent 既要采集又要清洗内部逻辑还是很重。后来调整成按输入输出边界来划分一个 agent 的输入和输出如果可以用同一种数据格式描述就合并如果格式差异大就拆开。比如采集 agent 输出原始 HTML清洗 agent 输出结构化 JSON这两个格式差异大必须拆。而清洗和标准化都输出 JSON可以合并成一个数据处理 agent。常见的角色划分参考采集 agent负责从各种来源获取原始数据输出原始内容处理 agent负责清洗、转换、结构化输出规范数据分析 agent负责计算、推理、归纳输出结论生成 agent负责文案、代码、报告撰写输出成品校验 agent负责检查输出质量输出通过/不通过及原因调度 agent负责拆解和路由不直接产出业务内容注意校验 agent 很容易被忽略但它是保证质量的关键。没有校验环节错误会一路传递到最终输出返工成本极高。3.2 通信协议的设计要点agent 之间怎么说话决定了系统能不能稳定运行。我推荐用最朴素的 JSON 结构不要一上来就搞复杂的消息队列或 RPC。{ task_id: t-20240101-001, from: orchestrator, to: collector, action: collect, payload: { source: https://example.com/data, format: html }, status: pending, timestamp: 2024-01-01T10:00:00Z }这个结构里task_id 是全局唯一标识用于追踪from/to 标明消息流向action 是动作类型payload 是具体参数status 是当前状态timestamp 用于超时判断。关键设计点payload 里不要塞大段文本。如果采集 agent 抓回来一篇长文不要把全文放进 payload 传给下一个 agent而是存到共享存储文件或数据库payload 里只放存储路径。这样做的好处是消息体小、传输快、也方便单独查看中间产物。3.3 调度逻辑的三种模式调度器怎么决定下一步做什么直接决定了系统的智能程度。我实践下来总结了三档模式可以按需选择。静态模式预先定义好步骤顺序调度器按顺序执行。适合流程固定的场景实现最简单但灵活性差。条件模式在静态基础上增加判断节点根据上一步的输出决定下一步走哪个分支。比如校验不通过就回到处理环节重做。适合有明确质量标准的场景。动态模式调度器本身也是一个 agent根据当前状态和剩余任务动态决定下一个该调用谁。灵活性最高但调试难度也最大容易陷入死循环。我的建议是从静态模式开始遇到瓶颈再升级。很多团队一上来就追求动态调度结果系统复杂度爆炸连基本流程都跑不稳。4. 实操过程从零搭一个最小可用的 agency-agents 系统4.1 环境准备与目录结构先说明这里给的是一个最小可用版本不依赖任何特定框架用 Python 标准库加一个 HTTP 请求库就能跑起来。这样做的原因是框架会掩盖很多细节而理解细节才是掌握这套思路的关键。目录结构建议这样组织agency-agents/ ├── orchestrator.py # 调度器主逻辑 ├── agents/ │ ├── collector.py # 采集 agent │ ├── processor.py # 处理 agent │ ├── analyzer.py # 分析 agent │ └── writer.py # 生成 agent ├── shared/ │ ├── storage/ # 中间产物存储 │ └── protocol.py # 消息协议定义 └── config.yaml # 角色配置与路由规则每个 agent 是一个独立的 Python 脚本通过命令行参数或标准输入接收消息处理完后把结果写到标准输出或存储路径。调度器负责按顺序调用它们。4.2 消息协议的具体实现在shared/protocol.py里定义消息的构造和解析函数import json import uuid from datetime import datetime def make_message(from_role, to_role, action, payload): return { task_id: str(uuid.uuid4()), from: from_role, to: to_role, action: action, payload: payload, status: pending, timestamp: datetime.utcnow().isoformat() } def parse_message(raw): return json.loads(raw) def save_artifact(task_id, name, content): path fshared/storage/{task_id}_{name}.json with open(path, w, encodingutf-8) as f: json.dump(content, f, ensure_asciiFalse, indent2) return path这里save_artifact是关键它把中间产物落盘返回路径给消息 payload 使用。这样消息体始终保持轻量。4.3 调度器的最小实现调度器不需要很复杂核心就是一个循环读任务、拆步骤、调 agent、收结果、判断是否继续。import subprocess import json from shared.protocol import make_message, parse_message def call_agent(agent_name, message): result subprocess.run( [python, fagents/{agent_name}.py], inputjson.dumps(message), capture_outputTrue, textTrue ) return parse_message(result.stdout) def run_task(initial_input): task_id t-001 # 第一步采集 msg make_message(orchestrator, collector, collect, {input: initial_input}) msg[task_id] task_id result call_agent(collector, msg) # 第二步处理 msg make_message(orchestrator, processor, process, result[payload]) msg[task_id] task_id result call_agent(processor, msg) # 第三步分析 msg make_message(orchestrator, analyzer, analyze, result[payload]) msg[task_id] task_id result call_agent(analyzer, msg) # 第四步生成 msg make_message(orchestrator, writer, write, result[payload]) msg[task_id] task_id result call_agent(writer, msg) return result[payload]这个版本是静态模式步骤写死了。但结构已经清晰每个 agent 只关心自己的输入和输出调度器负责串联。4.4 单个 agent 的写法示例以采集 agent 为例它只做一件事根据输入获取原始内容存盘返回路径。import sys import json import requests from shared.protocol import parse_message, save_artifact def main(): raw sys.stdin.read() msg parse_message(raw) source msg[payload][input] response requests.get(source, timeout30) artifact_path save_artifact( msg[task_id], raw_content, {content: response.text, status_code: response.status_code} ) output { task_id: msg[task_id], from: collector, to: orchestrator, action: collect_done, payload: {artifact: artifact_path}, status: success, timestamp: msg[timestamp] } print(json.dumps(output, ensure_asciiFalse)) if __name__ __main__: main()注意这里 agent 不直接返回内容而是返回存储路径。这是 agency-agents 的一个核心习惯消息传引用不传实体。4.5 参数选择与性能考量超时时间设多少我一般设 30 秒给采集类 agent60 秒给分析类 agent。依据是采集主要是网络 IO30 秒足够分析涉及模型推理给足 60 秒避免误杀。并发怎么处理最小版本是串行的但如果你的采集 agent 需要抓多个来源可以在采集 agent 内部用线程池并发而不是在调度层并发。原因是调度层并发会带来状态同步问题而单个 agent 内部的并发是局部的好控制。存储用什么小规模用本地文件就够了task_id 做前缀避免冲突。规模大了再换 SQLite 或对象存储。不要一上来就上数据库增加不必要的复杂度。5. 常见问题与排查技巧实录5.1 问题速查表现象可能原因排查方向解决思路某个 agent 无输出脚本报错但异常被吞检查 stderr在调度器里捕获并打印 stderr消息解析失败JSON 格式不合法打印原始输出统一用 json.dumps 生成禁止手工拼接任务卡住不结束某步超时未处理加超时机制subprocess 加 timeout 参数中间产物找不到路径拼接错误检查 task_id 一致性统一用 save_artifact 生成路径结果质量差上游数据有问题逐环节检查产物从采集开始逐步验证重复执行同一任务状态未持久化检查状态记录每步完成后写状态文件5.2 三个我踩过的坑第一个坑agent 之间直接互相调用。早期我图省事让处理 agent 直接调用分析 agent跳过了调度器。结果流程一复杂调用链变成一团乱麻根本理不清谁调了谁。后来强制规定所有 agent 只能和调度器通信agent 之间禁止直接调用。这条规矩立下来之后系统清晰多了。第二个坑payload 里塞大文本。有一次采集 agent 把整篇几万字的文章塞进 payload 传给下一个 agent结果消息体过大解析慢日志也没法看。改成存盘传路径后不仅速度快了调试时还能直接打开中间文件看内容。第三个坑没有校验环节。有段时间系统跑得挺顺但最终输出质量忽好忽坏。后来加了一个校验 agent专门检查上游输出是否符合预期格式和基本质量要求不合格就打回重做。加了这一环之后最终输出的稳定性明显提升。提示校验 agent 的判断标准要写清楚不要用“质量好不好”这种模糊描述而是用“是否包含必要字段”“长度是否在范围内”“是否包含禁用词”这类可量化规则。5.3 调试技巧单步执行与产物回放多 agent 系统最难的是调试。我的做法是给调度器加一个--step参数每执行完一个 agent 就暂停打印当前消息和产物路径人工确认后再继续。这样能精确定位是哪一步出了问题。另一个技巧是产物回放把某次任务的中间产物全部保留下次调试时直接从某个环节开始不用每次都从头跑。这在优化某个特定 agent 时特别有用。6. 扩展方向从最小系统到生产可用6.1 加入重试与降级机制最小版本没有重试一个 agent 失败整个任务就挂了。生产环境必须加重试同一个 agent 失败后隔几秒重试最多重试三次。三次都失败要么降级到备用 agent要么标记任务失败并通知人工介入。重试的关键是幂等性同一个消息重试多次结果应该一致。所以 agent 内部如果有写操作要用 task_id 做去重判断。6.2 引入优先级与队列当任务多了之后需要区分优先级。比如用户直接触发的任务优先于定时批处理任务。实现方式是在消息里加 priority 字段调度器维护一个优先队列高优先级的先出队。队列的另一个好处是削峰填谷任务提交速度超过处理速度时先入队慢慢消化避免系统被压垮。6.3 监控与告警生产系统必须知道每个 agent 的健康状况。最基本的监控指标每个 agent 的调用次数、成功率、平均耗时、失败原因分布。这些数据可以从状态记录里统计出来。告警规则可以设某个 agent 连续失败超过 5 次或者平均耗时突然翻倍就触发告警。告警渠道用邮件或即时通讯工具都行关键是有人能看到并处理。6.4 角色动态注册最小版本里 agent 是写死在调度器里的。更灵活的做法是让 agent 启动时向调度器注册自己声明自己能处理哪些 action。调度器维护一个注册表收到任务时查表决定调用谁。这样新增 agent 不需要改调度器代码扩展性更好。注册消息可以长这样{ role: collector, actions: [collect, collect_batch], endpoint: agents/collector.py, max_concurrency: 3 }调度器收到注册后把信息存到注册表。任务来了根据 action 查注册表找到对应的 agent 列表再按负载情况选一个调用。这套机制搭起来之后整个系统就从“写死的流水线”变成了“可插拔的协作网络”这才是 agency-agents 这个思路真正发挥价值的地方。我自己的体会是前期多花点时间把协议和调度逻辑设计好后期加功能就是加一个 agent 文件的事几乎不用动核心代码。反过来如果一开始图快把逻辑写死后面每加一个功能都要伤筋动骨。