Agent也可以"机构化"：一个基于agency-agents思路的多智能体协作搭建实录

最近在做某跨平台系统重构时，我一直在琢磨一个现象：单Agent处理简单任务已经够用了，但一旦任务链路变长、决策点变多、上下文变杂，单个Agent就开始频频出幺蛾子。要么是顾此失彼，要么是前后矛盾。换了好几个提示词框架，效果依然不稳定。

后来我换了个思路——与其死磕"一个全能的超级Agent"，不如把一个任务拆给"一个机构"去干。这个机构里有执行官、领域专家、执行专员，还有负责流程流转的角色，各管一段、互相汇报、按授权干活。这个思路有个代号，叫agency-agents。

文章写清楚这个思路从哪来、怎么搭、踩过什么坑，以及在什么场景下值得你从单Agent切换到这种"多角色协作"的架构。

1. 为什么单Agent包打天下不靠谱

先说说这个项目出发点。很多人做AI应用第一反应就是"一个Agent搞定一切"。输入需求、输出结果，中间靠提示词引导推理，看起来简单直接。但真实业务场景里，任务根本不是"输入→输出"的单向管道。

我做的一个模拟项目X就是一个典型的例子。这个系统要完成类似"用户提交需求→系统拆解→调用多个模块→汇总结果→生成报告"的流程。数据来源涉及数据库、第三方接口、本地文件，输出包含结构化数据、图表、说明文字三种形态。用单个Agent做，输入端是统一的，但输出端经常出问题。

具体症状大概有三种。问题是Agent在长上下文里会"遗忘"早期约束。第二是工具调用频繁切换时容易"人格漂移"——一会儿强，一会儿稳，一会儿又用片汤话糊弄人。第三是最麻烦的：Agent会自作主张修改既定流程顺序。

后来我意识到，这不是提示词质量的问题，而是职责边界缺失。一个角色一旦又要做计划、又要做执行、又要做校验、又要做输出，它的行为就不可预测。就好比一个小公司里一个人兼着CEO、会计、销售、保洁，一定会在某个环节掉链子。

这个观察直接推动我转向agency-agents模式。我想要的不是"一个更聪明的Agent"，而是一个可编排、可分工、可回退、可审计的Agent集群。

1.1 "机构"这个词的三个含义

先在概念上把这个题目拆透。agency-agents中，"agency"至少有三个可解释的方向：

第一，agency作为"机构/组织"。这对应将多个Agent组织成有角色分工、层级汇报、审批关系的协作体。整个架构像一家公司：有决策层、经理层、执行层。

第二，agency作为"代理关系"。每个Agent都代表系统去执行一部分任务，对用户或上游负责。这要求每个Agent要有话语权边界，不能越权包办。

第三，agency作为"行动力/主动性"。Agent不能总是"等指令然后执行"，而是要能在权限范围内主动推进任务，发现信息缺失时自行决策、向上汇报。

我在实际设计时，三个含义都占了一点。角色分层解决第一种，权限设计解决第二种，任务路由和反馈机制解决第三种。这套体系最终的表现形式是：一个编排中枢，加上若干可独立运行、也可以协作的Agent节点。

1.2 与传统面向对象设计的关键区别

有经验的朋友可能会说，这不就是给每个Agent分角色吗？远不止。区别在协作的"弹性"上。传统类的组装是静态的，调用关系在编译期就定了。Agent的协作是动态的，同一个任务可能在运行期被重新路由给不同的Agent执行。

举个例子。某次我让"数据采集Agent"去抓取一个接口的数据，接口突然变更了返回格式。传统设计会直接报错。但在agency模式下，这个Agent会尝试自我修复：先看看返回格式与预期差异在哪儿，如果只是字段名变化就做映射；如果结构完全改变，它会将情况上报给"编排中枢"，由中枢决定是更新格式规则还是交给"代码修复Agent"处理。

这种灵活的"边执行边决策"的能力，是我认为agency-agents模式最核心的价值，也是Class设计替代不了的部分。

2. 整体架构怎么分：麻雀虽小，五脏俱全

在动手写代码之前，先把架构轮廓画出来。

2.1 角色分层：Executive、Domain、Specialist、Workflow

我做拆分时参考了真实机构的组织方式，将Agent划分为四类：

• Executive Agent（执行中枢）：负责接收用户目标、拆解任务、规划执行路径，并协调其他Agent。它不直接处理具体数据，而是"管事"。
• Domain Agent（领域主管）：在各个专业方向（例如数据、图像、文本、运维）充当知识节点。它们知道"本领域有哪些工具和注意事项"。
• Specialist Agent（专项专员）：执行具体动作，比如调用某API、处理某个格式文件、执行某段脚本。它们是"动手干活"的那一层。
• Flow Agent（流转控制）：这个比较特殊，不负责业务，只负责状态流转、重试、超时和异常处理。相当于整个机构里的"流程办公室"。

这种分层直接把"策略"与"执行"分开了，极大地降低了提示词的复杂度。

2.2 为什么不能所有Agent都平级

第一版设计里我犯过一个典型错误：所有Agent平级互联，谁都可以直接调用谁。结果出现了"谁都在管、谁都在推"的局面：一个任务同时被两个Agent执行两次，另一个Agent认为自己没收到指令。

后来我引入了"层级+授权"机制。所有跨域协作必须经过Executor转发，不允许Specialist之间私下对接，这么处理有明确的好处：

• 协作点收敛，路由规则只需在Executor一层实现。
• 可以记录全链路追踪日志，便于回放和定位失误。
• 权限边界更加清晰，不会出现越权调用。

代价是增加了Agent间的通信开销。但考虑到是Agent自身处理的，实际影响可以接受，换来的是可控性的大幅提升。

2.3 每个Agent要有"自我认知"

这是个听起来很虚、实则很重要的设计——就像招聘时怎么让员工明白自己岗位的职责一样。每个Agent的System Prompt不只要写清"你能做什么"，更会写清"哪些事不归你管"，以及"何时该上报"。

我在代码里为每个Agent维护一份配置，类似这样：

Agent: 图像处理专员 职责: 处理格式转换、尺寸调整、滤镜应用 工具: PIL、OpenCV、图像处理函数库 仅限: 输入图片路径与输出路径 禁止: 自行修改文件名、访问数据库、决定存储策略 上报条件: 图片文件损坏、格式不识别、处理超时

白纸黑字写清楚"哪些不能碰"，远比泛泛告诉它"你有能力做很多事"要重要。这也是我踩了很多次坑才悟出来的。

3. 核心机制：授权、任务路由与记忆管理

定完角色，下一步处理协作机制的细节。这部分是实操中真正体现工程能力的地方。

3.1 授权机制

权限分两类。能力授权：哪些Agent能调用哪些工具，哪些操作必须由执行中枢审批；数据授权：Agent能访问哪些数据源、能写入哪些存储位置。

拿常见的"处理用户数据"来讲：

• Specialist A只能读取用户画像，不能写入数据库。
• Specialist B只能读取数据库字段，也不能接触原始文件。
• 汇总工作由Domain Agent完成，输出前要经过中枢确认。

每次授权都带"最小可用范围"原则——在跑通任务的前提下，尽可能地缩小各Agent的权限。不是不信任，而是为了在出错时约束爆炸半径。

3.2 任务路由与协议约定

多Agent协作最怕的是"你说东、它理解西"。所以我在第一版里就定了统一的任务描述协议，格式类似这样：

{ "task_id": "TASK-001", "type": "text_analysis", "input_refs": ["data://profile/2024/user_001"], "required_output": {"format": "json", "schema": "analysis_result_v1"}, "deadline": 30, "callback_path": "/internal/agents/text_domain/result" }

Agent之间不直接传大段自然语言描述，而是传这种半结构化的任务卡。自然语言只用在实在无法结构化的"开放讨论"环节。这套方案的可测性和可调试性，至少提高两个数量级。

3.3 记忆分三类：缓存、长期、隔离

Agent处理复杂任务时"忘事儿"是最让人头疼的。后来我用三级记忆体系解决：

• 工作记忆（短期）：只存在于单个任务的生命周期内，任务结束即清空。它储存当前任务的中间结果、临时标记和约束条件。
• 长期记忆（持久）：存用户偏好、历史决策、任务模式的总结性信息。它不在实时推理上下文里，按需调用。
• 隔离记忆（私有）：给每个Agent单独维护，避免"串台"。A Agent记住的内容不会被B Agent随意调用。

还有个使用技巧：每次路由时，执行中枢会抽取"最小必要记忆"，拼在任务卡里发给对应Agent，而不是把所有历史一骨碌倒给它。

3.4 状态机：让流程不出界

我把每个任务的执行过程建模成一个状态机：待分配 → 执行中 → 校验中 → 已完成 / 已失败 / 待重试 / 待裁决。

整个流转由Flow Agent控制，其他Agent无权自行更改任务状态。比如"执行中"的Agent发现出错了，它只能提交"失败请求"，由Flow Agent决定是重试、降级、还是交由裁决Agent。

这套机制带来的直接收益：多Agent执行的可控性就基本等同于工作流引擎的可控性。一旦出问题，你可以确定性重放、定点排查，而不是面对一片混沌。

4. 实操演示：从零搭一个小型Agency体系

这东西光说不练没有。我拿一个"数据处理中枢"作为例子，演示如何在一个本地框架内搭出一个简化的agency-agents体系。我用的底座是一个标准的多Agent运行框架，它将每个Agent封装成一个可以异步接收信号的服务节点。

4.1 定义四类角色的基础实现

先定义一个BaseAgent类，把通用能力做进去：

class BaseAgent: def __init__(self, name, capabilities, supervisor=None): self.name = name self.capabilities = capabilities self.supervisor = supervisor # 上级Agent，通常是ExecutionAgent self.context = {} self.status = "idle" async def handle_task(self, task): raise NotImplementedError async def report_up(self, event_type, payload): # 向上级提交事件 await self.supervisor.receive_event(self.name, event_type, payload) async def request_action(self, action_name, params): # 向中枢请求授权执行动作 return await self.supervisor.authorize_action(self.name, action_name, params)

这个基类包含了三个关键方法：处理任务、上报事件、请求授权。所有Agent都必须通过这三个途径与外界协作，从源头上杜绝了乱来。

4.2 实现一个事后可观的执行中枢

执行中枢是整个系统的核心。它负责维护任务分发，还要做校验与授权。我写了一个简版逻辑：

class ExecutiveAgent(BaseAgent): def __init__(self, name): super().__init__(name, capabilities=["task_planning", "resource_allocation"]) self.agent_registry = {} self.active_tasks = {} self.audit_trail = [] def register_agent(self, agent): self.agent_registry[agent.name] = agent agent.supervisor = self async def receive_event(self, agent_name, event_type, payload): self.audit_trail.append({ "from": agent_name, "event_type": event_type, "payload": payload, "ts": time.time() }) if event_type == "task_completed": await self.handle_completed(agent_name, payload) elif event_type == "need_clarification": await self.handle_clarification(agent_name, payload) # 其他事件类型，统一走日志 async def dispatch_task(self, task): task_id = task["task_id"] target_capability = task["required_capability"] candidate_agents = [a for a in self.agent_registry.values() if target_capability in a.capabilities] if not candidate_agents: await self.rollback_task(task_id, reason="no_available_agent") return # 简单策略：选第一个有能力处理的Agent agent = candidate_agents[0] self.active_tasks[task_id] = {"agent": agent.name, "status": "running"} await agent.handle_task(task)

这里有几处细节值得说：

• audit_trail记录所有Agent上报的事件，相当于给整个机构装了监控。
• dispatch_task里面做了能力匹配，而不是简单轮询。真架构里这里还会加上负载均衡、历史成功率权重、优先级别。
• 找不到可用Agent时，主动进入回滚流程，而不是卡住不动。

4.3 Domain Agent：领域知识与校验关卡

Domain Agent不直接干活，但它负责校验产出正确性。比如在数据处理场景里，DataDomainAgent的职责之一是检查下游返回的数据格式是否符合约定：

class DataDomainAgent(BaseAgent): def __init__(self, name): super().__init__(name, capabilities=["data_schema_validation"]) self.required_schemas = { "user_profile": {"fields": ["user_id", "age", "interests"], "type": "json"} } async def handle_task(self, task): if task["type"] == "validate_schema": ref = task["input_refs"][0] data = load_by_ref(ref) check_result = self.validate_data(data, task["required_output"]["schema"]) if check_result["passed"]: await self.report_up("validation_passed", {"task_id": task["task_id"]}) else: await self.report_up("validation_failed", { "task_id": task["task_id"], "errors": check_result["errors"] }) def validate_data(self, data, schema_name): schema = self.required_schemas.get(schema_name) # 实际此处会加载json schema并做逐字段校验 errors = [] if list(data.keys()) != schema["fields"]: errors.append(f"字段不匹配: 期望{schema['fields']}, 实际{list(data.keys())}") return {"passed": len(errors) == 0, "errors": errors}

这个设计保证了业务逻辑不是直接从"执行Agent"手里产出后就直接上线的，而是要过一道专业校验关卡。

4.4 让Specialist Agent"干活"

真正的执行者反而代码逻辑最简单。它专做单一动作，比如读取一个CSV并转成JSON：

class CSVParserAgent(BaseAgent): def __init__(self, name): super().__init__(name, capabilities=["csv_to_json"]) async def handle_task(self, task): if task["type"] != "csv_to_json": await self.report_up("unsupported_task", {"task_id": task["task_id"], "type": task["type"]}) return try: input_path = task["input_refs"][0] df = parse_csv(input_path) output = df.to_json(orient="records") await self.report_up("task_completed", { "task_id": task["task_id"], "output_ref": f"memory://output/{task['task_id']}", "preview": output[:500] }) except Exception as e: await self.report_up("task_failed", { "task_id": task["task_id"], "error": str(e) })

注意这里几个细节。第一，它不会把完整的输出塞进上报信息里，只上报引用和预览，避免上下文爆炸。第二，出错时没有抛出异常中断进程，而是通过事件将异常上报，让上层决定。这遵循了"执行不过载"的原则。

4.5 接线与运行

这一步来做组装：把所有Agent挂到ExecutiveAgent上，投一个模拟任务进去：

async def main(): executive = ExecutiveAgent("大主管") data_domain = DataDomainAgent("数据主管") csv_parser = CSVParserAgent("CSV专员") json_formatter = JsonFormatterAgent("JSON专员") executive.register_agent(data_domain) executive.register_agent(csv_parser) executive.register_agent(json_formatter) task = { "task_id": "TASK-001", "required_capability": "csv_to_json", "input_refs": ["files://data/raw.csv"], "required_output": {"format": "json", "schema": "user_profile"}, "callback_path": "/internal/agents/data_domain/result" } await executive.dispatch_task(task)

运行这段代码，就像操作一个链条： 执行中枢把任务发给CSV专员；专员解析后上报完成；中枢把产出结果转给数据主管做格式校验；校验通过后再输出。

这套逻辑虽然简化，但已经把"分层、授权、路由、校验、追踪"几个核心要点打通了。你可以在自己的框架上继续加网络接口、持久化日志、重试策略，变成一个真正能上生产环境的体系。

5. 实战中的坑：排查技巧与避雷指南

搭建过程中我踩了不少坑，有些在文档里根本找不到。把它们梳理成问题速查表，方便你排查。

5.1 任务循环与死循环

症状：Agent A调用Agent B，B又回调A，循环一段时间后，任务栈溢出。

原因：没有在任务卡里带深度限制或路径标记；Agent在"澄清"时反复触发上报。

解法：

• 在任务卡里加ttl字段，每次流转减一，归零直接丢弃。
• 限制同一任务在两个Agent之间的流转次数，超过直接上报到Executive。
• 全部上报事件落在日志库里，用追踪ID复查循环路径。

def check_cycle(self, task_id, agent_name): if task_id in self.active_tasks and self.active_tasks[task_id].get(agent_name, 0) > 3: raise TaskCycleError(f"任务{task_id}在Agent{agent_name}处循环超过3次")

5.2 上下文爆炸

多Agent各记一段上下文，汇总到执行中枢时，消息量十分惊人。尤其是中间结果带了一大段原文的场景，很快打爆上下文。

解法是"引用替代内容"——Agent之间传递的是ref，不是数据本体。对方按需读取，读取完即可释放。这跟程序设计中指针传参远比值拷贝更高效，道理一致。

5.3 角色混淆与越权行为

没有明确"禁止事项"的Agent，会在意想不到的地方越界。比如让"数据分析Agent"干活，它自己顺手把数据写进了持久化存储，你根本无法预测。

解法就是前述的"能力授权+权限最小化"。每个Agent的实现里做好严格的if type not in capabilities判定，禁止未知操作，更可靠。

5.4 性能开销与成本控制

Agent多了以后性能开销是线性叠加的。每次路由都要通过执行中枢，单次任务的token消耗也随之增长。真实的业务场景里"结果可信度"与"资源消耗"必须做权衡。

我总结的经验是：优先保证执行路径上的Agent少而精，不要在不需要的环节加Agent；只在任务复杂度高、单Agent失败率高时，才启用更大的协作网络。

6. 后续还可以怎么玩

现在这套体系已经在我这边跑了几个月，稳定性明显优于之前单Agent方案。后续有几个明确方向值得继续折腾。

一个是让ExecutiveAgent学会编排自我优化。比如它可以根据历史日志发现"每次让A做格式转换都会失败一次，改成先让B预处理再接A"，进而动态调整路由策略，这本质上是从执行机构进化成自适应组织。

另一个方向是做跨任务的状态复用。多个相似任务如果共享某些中间成果，不要每次都重跑，直接把memory_ref附到新任务卡上即可。

还有一个特别有意思的方向：把Agent的技能变成模块化插件，按需装载。现在的Agent在启动时就固定了能力集，未来我想做成动态能力注册——某个Agent突然多了一个"PDF解析能力"时，它可以主动向Executive报备，注册中心同步更新。

这套体系也适合引入"回放演练"：记录所有运行日志和决策轨迹，出问题后可以像调试程序一样逐步回放，观察哪个环节的决策导致最终偏差。

写在最后的小Tip

如果你第一次尝试搭建agency-agents体系，我建议别贪多。先塞两个Agent：一个Executive、一个Specialist，把任务跑通，确认路由和上报正常后，再加Domain校验层，最后扩展成完整协作网络。直接一步到位搭十个Agent，调试成本非常高，几乎注定会劝退自己。

我个人实际做下来，最顺手的一套组合是：Executive + 3个Domain + 5个Specialist，总共9个角色，能覆盖大多数业务场景。

最后再分享一个小技巧：所有Agent的System Prompt在调试阶段都要加上"你正在参与一次模拟演练，所有动作都会被记录"这样的提醒；它能让Agent在自我推理时更加规范，减少幻觉式输出。正式上线时再移除这句提醒，效果会有一个明显提升。