1. 先说清楚agency-agents 到底在做什么我最近在折腾 agent 相关的个人项目发现很多人在聊 agency-agents 的时候其实说的根本不是一件事。有人以为它只是一个 ros 风格的多 agent 框架有人把它理解成“带权限控制的 API 网关”还有人干脆当它是概念包装。我自己的理解是agency-agents 本质上是解决“多个智能体如何自主协作完成一套真实任务”的那层编排与治理机制——注意关键词不是 agents而是 agency即每个智能体是不是真的具备自主决策权、能力和边界。如果你拆过稍微复杂一点的业务场景比如一个竞品信息采集与周报生成系统、一个客服工单分类与回复起草系统、一个跨部门的数据汇总与分析流水线你会发现单靠一个大模型 Prompt 硬撑根本跑不稳。Prompt 一长上下文就乱工具一多模型不知道调用谁任务一交叠输出格式就漂移。agency-agents 的核心思路是把“一个全能的超级智能体”拆成“一群各司其职、可以互相调用、并且在一个统一调度策略下完成目标的小智能体”每一个小智能体拥有明确职责、工具集和决策边界。这篇文章我不打算写概念论文而是结合我一个真实跑过的项目——一个由采集、分析、撰写三类 agent 组成的竞品周报自动化系统——把 design thought、架构拆解、核心代码、现场踩坑全部过一遍。适合对 agent 编排已经有一定概念、但想真正落地一个“多 agent 协作系统”的朋友。2. 整体设计思路为什么我不直接写一个大 Prompt2.1 大 Prompt 方案看起来省事实际上到处是坑先聊一下我为什么没有把整个“竞品信息采集 分析 周报撰写”塞进一个单 Agent 的 System Prompt。在最早期原型里我确实这么干过。Prompt 里写了“你是一个研究助理你需要先搜索然后总结最后生成报告”然后给这个 agent 挂了一堆工具搜索工具、网页抓取工具、数据库查询工具。跑起来之后问题非常具体第一模型在长任务链条里会“忘记”自己已经做过的中间步骤尤其在上百条搜索结果的上下文里它极可能把上一步的结论当成下一步的事实输出大量幻觉第二工具参数一旦变多模型经常传错参数比如把竞品名称填到了日期字段里第三一旦某个步骤失败比如某个网页抓取 403模型会即兴发挥换一个完全没有验证过的数据来源结果直接污染整份报告。也就是说大 Prompt 方案的问题不是“能不能跑”而是“跑起来之后你无法对它做约束、观察、回滚”。而 agency-agents 的思路恰好能把每一个职能拆出来给每个 agent 只暴露它该用的工具并且在调度层面对它的决策做复核。你可以理解成大 Prompt 方案是让一个实习生全权负责一个项目agency-agents 是一个有分工、有汇报、有复核的项目组。2.2 我最终采用的架构调度中枢 角色集群 工具层 记忆层最终跑通的方案不是去套某个重框架而是自己实现了一套很轻的调度机制。整体结构分四层调度中枢Coordinator ├── 任务拆解把用户目标拆成若干子任务 ├── 优先级管理决定哪些子任务可以并行哪些必须串行 └── 异常转移某个 agent 失败后调度另一条路径 角色集群Workers ├── 采集 Agent只负责搜索、抓取、去重 ├── 分析 Agent只负责数据清洗、结构化、对比 └── 撰写 Agent只负责生成最终报告 工具层Tools ├── 搜索工具、网页抓取工具、数据库读写工具 ├── 所有工具配上 schema 校验 └── 操作日志全部记录可追踪 记忆层Memory ├── 短期当前任务上下文供当前会话内共享 └── 长期历史数据、历史报告摘要、业务规则这套结构大家可能看着眼熟LangGraph、CrewAI、AutoGen 都有类似概念。但我自己实现的好处是没有黑盒行为每一个决策点都能打日志每个 agent 的工具调用都能被审计并且在出问题时我可以精确地知道是哪一环出的问题。对于“agency”这个层面的诉求来说可控性比什么都重要。2.3 为什么多 agent 协作比单体 agent 更适合真实业务这里说点非常实际的理由。首先是上下文隔离。每个 worker 只看到自己职责范围内的数据。采集 agent 只需要关心“搜索结果列表”分析 agent 只接收结构化后的数据它不会被原始网页里的广告文案干扰撰写 agent 拿到的已经是规范的 Markdown 数据块。这直接避免了我前面说的“上下文污染”问题。其次是故障隔离。采集 agent 如果遇到反爬它只需要重试或换数据源绝不会影响到分析 agent 已经完成的工作。而单体 agent 一旦中间某步出错后边所有结果都可能作废。第三是可扩展性。要给系统加一个新能力比如增加一个社交媒体监测 agent我只要在角色集群里加一个 worker给它定义好工具和输入输出格式然后告诉调度中枢它能在哪个环节介入即可。单体 agent 加能力意味着要重新设计 Prompt 和工具集牵一发动全身。3. 核心细节与实操要点agency 落地最容易被忽略的几个点3.1 工具注册表让 agent 的“能力边界”可被调度器感知多 agent 系统里工具注册表不是一个简单字典它决定了两件关键事情一是调度中枢知道“哪个 agent 能干什么”二是 agent 自己知道“我该用什么工具”。我给每个工具定义成这样的结构{ name: search_web, description: 搜索公开网页返回标题、链接、摘要列表。适合用于获取竞品公开新闻。, input_schema: { type: object, properties: { query: {type: string, description: 搜索关键词建议包含品牌名和行业词}, date_range: {type: string, description: 日期范围格式为 YYYY-MM-DD 到 YYYY-MM-DD} }, required: [query] }, agent: collector, timeout_seconds: 30, retry_policy: {max_retries: 2, backoff_seconds: 5} }这个结构有一个很重要的作用它给调度中枢提供了一个“能力清单”调度中枢在做任务拆解时会根据工具描述判断子任务应该分给哪个 agent。同时tool schema 里的 required 字段也间接约束了模型生成参数时的自由度。实操时我发现如果 description 写得含糊比如只写“搜索一些信息”模型调用工具时经常漏参或错传如果 description 写清楚“搜索关键词建议包含品牌名和行业词”参数准确率会明显提升。这听起来像是废话但很多人真的会忽略。3.2 短期上下文与长期记忆的边界管理多 agent 协作系统里记忆层最容易被做成“把所有东西都扔进一个大向量库”然后美其名曰 RAG。我的经验是一定要把短期上下文和长期知识严格分离。短期上下文解决的是当前任务过程中的工作状态。比如采集 agent 已经抓了哪些 URL分析 agent 已经产出了哪些中间结论撰写 agent 正在生成哪一段。这些数据我用一个简单的 JSON 状态对象在调度中枢里维护每次 agent 完成一个子任务就把它的产出写回状态对象。这样即使某个 agent 中途崩溃调度中枢也能根据已保存的状态继续而不是从头再来。长期记忆解决的是跨任务的知识复用。比如“上次周报里提到的定价变化结论”“公司产品名称的官方写法”“分析模型偏好的输出格式”。这些内容我会按业务规则整理成一个只读知识库所有 agent 在生成时会参考它但不会被它无意义地塞爆上下文。区分方式也很简单短期上下文是“这次任务产生的”长期记忆是“无论做多少次任务都应该知道的”。3.3 自主性的边界该让 agent 自由决策的地方和不该让它碰的地方“Agency”这个词容易给人一个错误的暗示——让 agent 尽可能自主。但在真实系统里自主必须是分层级的。我自己的分级方式是这样第一层工具选择自主。agent 自己决定用哪个工具、什么参数、什么时候重试。这一层我会给比较大的自由度。第二层子任务执行顺序自主。比如采集 agent 可以决定先搜哪个关键词、按什么顺序抓取页面。这一层只会给出建议顺序不强制。第三层业务规则决策自主。比如“某个竞品的价格页面是不是可信来源”这个必须受到知识库和校验规则的约束绝对不放开给 agent 自行判断。第四层对外输出决策自主。无论是周报发出去给老板看还是客服回复发给真实用户都必须经过人工确认环节。我在系统里设计了一个“pending_review”状态所有对外内容生成后都挂起只有人工点击通过后才真正发送。这四层边界是我踩了数次坑之后定下来的。早期版本里我让撰写 agent 完全自主生成周报并直接发邮件结果它引用了一个已经被证伪的涨价传闻要不是复盘及时那封周报差点造成错误决策。所以构建 agency-agents 系统优先考虑的永远不是“它能多聪明”而是“它敢犯错时错误能不能被兜住”。3.4 人工干预接口不是一个附属功能而是架构的一部分我在系统里把人工干预设计成一个正式的“agent 角色”它自己有一个待办队列。当任何一个 worker agent 产出了需要确认的结果它会写一条消息到干预队列然后系统进入等待状态。这个设计避免了“写完报告就自动发送”的危险链路。这里面有一个非常实用的细节关键节点的确认请求最好带上 agent 的完整推理过程。比如撰写 agent 提交周报时调度中枢会在确认请求里附上它是基于哪些搜索结果、哪些数据摘要生成结论。这样人工审核时不需要重新看一遍原始数据只需要核对 Agent 引用到的关键论据是否正确。这极大减少了人工审核的心理负担也让审核这件事真正变得高频可执行。4. 实操复盘一个可运行的 agency-agents 最小闭环4.1 场景定义与任务拆解我自己跑的案例是“AI 编程工具竞品周报自动化”。目标每周五上午自动生成一份涵盖 3 个竞品的产品动态、定价变化、社区讨论热点的 Markdown 周报并交付到内部知识库。整个流程拆成四个子任务1. 采集阶段分别搜索三个竞品的最近一周新闻、官方博客更新、社区热帖 2. 清洗阶段把抓到的原始内容去重、过滤广告/无关内容提取发布日期和正文摘要 3. 分析阶段对比三个竞品提取高价值信号新功能、定价调整、重大争议 4. 撰写阶段基于分析结果生成结构化周报这里有个调度设计教训一开始我让采集和分析串行执行整体耗时接近 20 分钟。后来我把三个竞品的采集拆成三个并行的子任务每个竞品一个独立采集 agent只在清洗阶段合并。用时直接降到 6 分钟左右。并行度设计的收益在这里极其明显但并行度高了之后要特别注意共享资源竞争。比如三个采集 agent 同时写同一个去重表就会发生互相覆盖。这个后面我会细说。4.2 核心代码任务循环与角色路由下面这段代码是我整个系统里最关键的一部分负责“接收一个用户目标拆解子任务路由给对应 agent并回收结果”。它定义了一个最小可运行的调度循环import json import uuid from concurrent.futures import ThreadPoolExecutor, as_completed class AgencyAgent: def __init__(self, role, tool_registry, memory): self.role role self.tool_registry tool_registry self.memory memory self.tool_results [] def run(self, task): # 角色 agent 的执行入口接收子任务调用工具返回结构化结果 # 这里的 llm_call 是简化写法项目里替换成了实际的 LLM 接口 plan self._plan_task(task) for step in plan: tool_name step[tool] tool_args step[args] self.tool_results.append(self._call_tool(tool_name, tool_args)) return self._compose_result(task, self.tool_results) def _plan_task(self, task): # 在真实项目里这步由 LLM 根据 task 工具注册表生成 # 这里直接给一个预设步骤作为最小示例 return [ {tool: search_web, args: {query: task[query]}}, {tool: fetch_page, args: {url: task[seed_url]}} ] def _call_tool(self, name, args): tool self.tool_registry.get(name) if tool is None: raise ValueError(funknown tool: {name}) return tool[handler](args) def _compose_result(self, task, results): return {task_id: task[task_id], role: self.role, data: results} class Coordinator: def __init__(self, agents, memory): self.agents agents self.memory memory def dispatch(self, user_intent): # 1. 拆解任务把用户意图转换成子任务列表 subtasks self._decompose(user_intent) # 2. 按角色路由并执行子任务 with ThreadPoolExecutor(max_workers3) as executor: futures {} for sub in subtasks: agent self._route(sub[role]) futures[executor.submit(agent.run, sub)] sub[task_id] results {} for future in as_completed(futures): task_id futures[future] results[task_id] future.result() # 3. 汇总结果写回记忆 self.memory[last_results] results return results def _decompose(self, intent): # 从用户意图 - 子任务列表 # 项目里这一步同样是 llm 调用 规则校验这里省略 return [{ task_id: str(uuid.uuid4()), role: collector, query: intent[query] }] def _route(self, role): return self.agents.get(role)这段代码虽然精简但他把 agency-agents 的骨架表达到位了coordinator 负责任务分解和结果回收worker 只处理自己的子任务所有复杂度被限制在各自的边界里。4.3 关键参数需要哪些配置才能跑稳完整跑通一个多 agent 任务有四个关键配置直接抄就行上下文窗口分配。我给每个 worker 的 LLM 调用窗口设了上限。采集 agent 用 8K 上下文就够分析 agent 给 16K撰写 agent 给 32K。如果所有 agent 都统一给 32K最终调用成本会直线上升而且大部分 token 根本没用到。可以根据自己的场景观察再微调。超时时间。工具调用必须设置超时。网页抓取类工具我统一设 30 秒搜索类工具 15 秒数据库读写 10 秒。超时之后根据重试策略处理最多重试 2 次两次失败直接标记该子任务失败并交给调度中枢决定是否降级。输出格式校验。每个 agent 的返回都必须过 JSON Schema 校验。刚开始这个校验我做得宽松后来发现 agent 偶尔会把数组对象包进 JSON 字符串里返回下游就整个崩溃。加上严格校验后错误会在最早环节暴露。日志级别。多 agent 系统里最容易出的问题就是“谁改了什么数据说不清”。我的做法是每个 agent 实例在初始化时带一个全局唯一的 run_id所有工具调用、中间产出、最终结果都打上 run_id 和 agent role 标签。排查问题时直接按 run_id 过滤日志就能看到完整一条任务链路。4.4 实际运行效果与日志示例一次完整运行的调度日志大致长这样[10:02:13] coordinator: 拆解任务完成共 7 个子任务 [10:02:14] collector-01: 开始搜索竞品 Gemini 最近 7 天新闻 [10:02:15] collector-02: 开始搜索竞品 Copilot 最近 7 天新闻 [10:02:16] collector-03: 开始搜索竞品 Cursor 最近 7 天新闻 [10:02:31] collector-01: search_web 返回 10 条结果命中 6 条 [10:02:33] collector-02: search_web 返回 10 条结果命中 7 条 [10:02:38] collector-03: search_web 返回 10 条结果命中 4 条 [10:02:40] cleaner: 收到 17 条原始记录去重后剩余 14 条 [10:03:12] analyzer: 提取高价值信号 5 条新功能 2定价调整 1社区争议 2 [10:04:15] writer: 周报初稿生成完毕进入 pending_review [10:04:20] coordinator: 任务完成产出报告 ID: rpt_20250117_001从这个日志里能明显看出并行采集和串行分析是两段节奏。前端四个 agent 是并行跑的后端分析、撰写是串行依赖前者的产物。这样设计的好处是可以直观从日志判断瓶颈在哪个环节。比如如果 collector-02 花了 60 秒其他 agent 都在等它那就是采集 agent 的数量或者搜索关键词需要优化。5. 常见问题排查与避坑实录5.1 多 agent 协作系统的高频故障与解决方案我在项目迭代过程中遇到的最典型问题整理成了速查表现象可能原因排查手段解决方案agent 反复调用同一个失败工具工具描述里没有写失败后的替代路径看路由日志里 tool_name 是否一直不变在工具描述中增加“如果失败尝试使用 xx 工具”的兜底说明多 agent 产生的中间数据互相覆盖共享状态没有按 run_id 隔离检查 memory 写入逻辑是否带唯一键所有状态写入强制使用 run_id role 作为 key分析 agent 产出的结论明显违反业务常识知识库没有被注入到分析 prompt查看分析 agent 的输入上下文中是否有业务规则将业务校验规则注入到分析 agent 的 system prompt 中agent 输出格式漂移导致下游 parse 失败JSON Schema 校验缺失或过于宽松查看原始输出和预期 schema 的差异统一使用严格 schema并在失败时返回修正指令任务链路陷入死循环agent 的 self-loop 阈值设置过高统计单 agent 工具调用最大循环次数限制单 agent 最多自循环 3 次超出则上报 coordinator这里我想特别展开说第一条。工具描述里的“失败兜底”是很隐蔽但极有用的细节。比如采集工具的原始描述是“抓取指定 URL 的正文内容”。一旦遇到 403模型会怎么处理常见情况是它不断重试甚至自己编造 URL 变体导致又慢又错。后来我把描述改成“抓取指定 URL 正文若返回 403 或 404尝试搜索该站点内同主题页面”整个失败恢复能力明显提升。5.2 最隐蔽的坑agent 之间隐性互相干扰还有一个我特别想分享的坑很难复现但一旦遇到非常头大。我在某个版本里让采集和分析 agent 共用一个全局配置对象里面存了语言偏好、时间格式、输出风格等全局设置。结果发现某个采集 agent 在任务过程中把“语言偏好”临时改成了英文导致后续所有 agent 的输出风格全部变成英文而代码里没有一个地方记录过这个改动。这个问题的本质是多个 agent 共享可变状态时任何一个 agent 的副作用都会影响全局。解决方式有两个一是所有 agent 的配置输入改成不可变快照初始化时拷贝一份任务中不允许修改二是所有非预期状态变化都要写审计日志后期可以追踪是谁做的变更。我最终两者都做了从此再也没有出现过这类“幽灵式”干扰。5.3 如何判断一套 agency-agents 系统是否健康这个判断标准其实可以量化为几个可观测指标我每次迭代都会看这几个数一个是平台内 exp 函数的内部工具调用成功率如果连续 3 次任务成功率低于 80%优先怀疑工具描述或参数 schema 有问题而不是 model 的问题。第二个是平均单任务执行耗时异常飙升往往意味着某些 agent 进入了无效重试循环。第三个是人工审核界面里“驳回率”如果经常驳回撰写 agent 生成的周报说明分析结果的可信度或者叙述逻辑有待改进。当一个系统在这三个指标上都稳定下来之后真正剩下的问题往往是细节微调而不是架构层面的返工了。6. 还能怎么扩展从“执行任务”到“主动发现任务”实现到这里其实只能算一个“按指令执行”的系统。更高一层的 agency-agents是让 agent 具备主动发现问题的能力。我目前已经在做的扩展是给采集 agent 加了一个“趋势预警”模式它每天自动跑一次关键词搜索把结果和历史基线做对比。如果某个竞品的讨论热度、新闻数量或关键词评分突然出现显著波动它会主动生成一条预警消息推到人工确认队列里而不是等周报日才被动收集。这个扩展不复杂但工作方式已经从“用户发起任务 - agent 执行”变成“agent 持续观察 - 识别变化 - 主动上报”。这也是我认为真正体现“agency”的地方——智能体不是等指令的工具而是能基于自身观察提出“这件事值得关注”的协作成员。前提依然是上报内容经过校验并由人工确认关键决策。这和我前面说的第四层边界并不矛盾主动性体现在“发现问题”而决策权依然在人。最后分享一个我在项目里反复体会到的结论做 agency-agents 系统最重要的不是模型有多强而是你给系统画的边界有多清晰。给 agent 越清晰的工具、越明确的分工、越严格的校验它反而能展现出更令人放心的自主性。边界即自由这句用在多智能体系统上再合适不过。