多智能体协作框架agency-agents深度拆解:架构设计与工程实践

发布时间:2026/10/25 3:23:13
多智能体协作框架agency-agents深度拆解:架构设计与工程实践

1. 从"agency-agents"这个标题说起:一个多智能体协作框架的完整拆解

第一次看到"agency-agents"这个命名的时候,我的直觉是:这大概率是一个围绕"代理(agent)"概念构建的协作型系统,而且"agency"这个词放在前面,说明它强调的不是单个 agent 的能力,而是 agent 作为一个"能动主体"所具备的自主性、协作性和目标导向性。换句话说,这不是一个简单的"调用大模型 API 返回结果"的玩具项目,而是一套让多个 agent 各司其职、互相配合、共同完成复杂任务的编排框架。

我在过去两年里陆陆续续搭过七八套多智能体系统,从最简单的"两个 agent 互相聊天"到后来带工具调用、带记忆、带任务队列的完整流水线,踩过的坑可以说能写一本小册子。所以当我拿到"agency-agents"这个标题时,脑子里立刻浮现出一整套需要拆解的问题:它的 agent 是怎么定义的?agent 之间怎么通信?任务怎么分配?失败了怎么重试?状态怎么持久化?这些问题的答案,决定了一个多智能体框架到底是"能跑 demo"还是"能上生产"。

这篇文章我打算把"agency-agents"当作一个典型的多智能体协作框架来彻底拆开讲。不管你是刚接触 agent 概念的新手,还是已经写过几个 agent 但总觉得"差点意思"的开发者,我都会从设计思路、核心机制、实操落地、问题排查四个层面,把这类框架的里里外外讲透。你会看到具体的目录结构、关键代码片段、参数配置,以及那些只有真正跑过才会知道的坑。全文基于我在类似项目中的通用实践来展开,所有代码和配置都是可以直接参考复现的。

2. 多智能体框架的整体设计与思路拆解

2.1 为什么需要"agency"而不是单个 agent

先说一个很多人容易忽略的问题:为什么我们要搞多智能体,而不是把一个大模型 prompt 写得足够长、足够全?

我早期也是"单 agent 万能论"的信徒,觉得只要 prompt 写得好,一个 agent 就能干所有事。但实际做下来会发现三个绕不过去的坎。第一是上下文污染:当一个 agent 既要负责理解需求、又要负责写代码、还要负责测试和纠错时,它的上下文里会混杂各种不同性质的信息,导致它在某个具体环节上的判断力下降。第二是角色冲突:让同一个 agent 既当"创作者"又当"审查者",它往往会倾向于认可自己的产出,这就是所谓的"自我肯定偏差"。第三是可维护性:一个巨型 prompt 改一处动全身,根本没法做版本管理和局部优化。

"agency-agents"这类框架的核心价值,就是把这些职责拆开,让每个 agent 只专注一件事,通过明确的协议互相协作。这就像一家公司:你不会让一个人同时当销售、会计和法务,而是分设岗位,用流程和文档把它们串起来。多智能体框架本质上就是在软件层面复刻这种"组织架构"。

2.2 核心架构选型:编排式 vs 自治式

拆解这类框架,第一个要做的决策是架构选型。市面上主流的多智能体架构大致分两派:

架构类型核心特征优势劣势适用场景
编排式(Orchestrated)有一个中心调度器决定谁在什么时候做什么流程可控、易调试、结果稳定灵活性差、中心节点是瓶颈流程固定的生产任务
自治式(Autonomous)agent 之间自由通信,自主决定下一步灵活、能应对开放问题难调试、容易发散、成本不可控探索性、研究性任务
混合式(Hybrid)顶层编排 + 底层自治兼顾可控与灵活实现复杂度高复杂生产系统

"agency-agents"从命名上强调"agency"(能动性),我判断它更偏向混合式:顶层有一个 orchestrator 负责宏观任务分解和流程控制,底层的每个 agent 在自己的职责范围内拥有一定的自主决策权。这也是我在实际项目中最推荐的方案——纯自治的系统听起来很酷,但你很难让它稳定地完成一个需要 20 步的固定流程;纯编排的系统又太死板,遇到预期外的情况就卡住了。

2.3 Agent 的抽象模型设计

一个 agent 在这类框架里到底由什么构成?我总结下来是四个核心部件,缺一不可:

  • 角色定义(Role/Persona):这个 agent 是谁、负责什么、边界在哪。通常是一段系统提示词。
  • 能力集(Tools/Skills):它能调用哪些工具,比如搜索、读写文件、执行代码、调用外部 API。
  • 记忆(Memory):短期记忆(当前对话上下文)和长期记忆(跨会话的知识沉淀)。
  • 决策循环(Reasoning Loop):它如何根据当前状态决定下一步动作,通常是 ReAct 或 Plan-and-Execute 模式。

这四个部件里,最容易被低估的是记忆。很多人做 agent 只关注 prompt 和工具,结果 agent 每次都是"失忆"状态,无法从之前的交互中学习。我在一个项目里给 agent 加了简单的向量记忆后,它在重复任务上的成功率从 60% 出头提升到了 85% 以上,效果非常明显。

2.4 通信协议:agent 之间怎么"说话"

多智能体系统里,agent 之间的通信协议是骨架。常见的有几种模式:

第一种是消息传递(Message Passing),agent A 把结构化消息发给 agent B,B 处理后回复。这种方式解耦好,但需要定义清晰的消息格式。

第二种是共享黑板(Blackboard),所有 agent 读写同一块共享状态区。这种方式适合需要频繁同步信息的场景,但容易产生竞态。

第三种是任务队列(Task Queue),任务被放进队列,空闲的 agent 去领取。这种方式天然支持并行和负载均衡。

"agency-agents"这类框架我倾向于用消息传递 + 共享状态的组合:agent 之间通过消息通信,但共享一个任务状态对象,记录整个流程的进度。这样既有解耦的好处,又能让每个 agent 知道全局进展。

3. 核心细节解析与实操要点

3.1 目录结构:一个好的框架从清晰的目录开始

我见过太多项目把 agent 定义、工具、配置全塞在一个文件里,跑起来没问题,但一旦要加新 agent 或者改流程就痛苦不堪。一个可维护的多智能体项目,目录结构应该是这样的:

agency-agents/ ├── agents/ # 各个 agent 的定义 │ ├── base.py # agent 基类,定义通用接口 │ ├── planner.py # 规划 agent │ ├── executor.py # 执行 agent │ ├── reviewer.py # 审查 agent │ └── registry.py # agent 注册表 ├── tools/ # 工具集 │ ├── base.py # 工具基类 │ ├── file_ops.py # 文件操作 │ ├── web_search.py # 搜索工具 │ └── code_runner.py # 代码执行 ├── memory/ # 记忆模块 │ ├── short_term.py # 短期记忆 │ └── long_term.py # 长期记忆(向量库) ├── orchestrator/ # 编排器 │ ├── scheduler.py # 任务调度 │ └── state.py # 全局状态管理 ├── configs/ # 配置文件 │ ├── agents.yaml # agent 配置 │ └── models.yaml # 模型配置 ├── tests/ # 测试 └── main.py # 入口

这个结构的关键在于关注点分离:agent 只管自己的逻辑,工具只管具体能力,编排器只管调度。这样你想换一个模型、加一个工具、改一个 agent 的行为,都只需要动对应的那一小块。

3.2 Agent 基类的设计要点

所有 agent 都应该继承同一个基类,这个基类定义了统一的接口。下面是我常用的一个简化版本:

from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any @dataclass class AgentContext: task_id: str shared_state: dict history: list = field(default_factory=list) max_steps: int = 10 class BaseAgent(ABC): def __init__(self, name: str, role: str, tools: list, model: str): self.name = name self.role = role self.tools = {t.name: t for t in tools} self.model = model @abstractmethod def build_prompt(self, context: AgentContext) -> str: """根据上下文构建提示词""" ... @abstractmethod def parse_action(self, response: str) -> dict: """解析模型输出,提取动作""" ... def run(self, context: AgentContext) -> dict: """agent 的主循环""" for step in range(context.max_steps): prompt = self.build_prompt(context) response = self.call_model(prompt) action = self.parse_action(response) if action["type"] == "finish": return action["result"] result = self.execute_tool(action) context.history.append({"action": action, "result": result}) raise RuntimeError(f"{self.name} 超过最大步数未完成")

这里有几个设计细节值得说。max_steps 是必须的,我见过太多 agent 陷入死循环把 token 烧光的案例,一个硬性的步数上限是最后的安全网。build_prompt 和 parse_action 抽象成方法,是为了让不同 agent 可以用不同的提示词模板和输出解析逻辑,比如 planner 输出的是任务列表,executor 输出的是工具调用。context 里带 shared_state,是为了让 agent 能感知全局进度,而不是活在自己的小世界里。

3.3 工具系统的设计:让 agent 真正"能干活"

工具是 agent 的手脚。一个设计良好的工具系统需要满足几个条件:描述清晰(模型能理解什么时候用)、参数明确(模型知道怎么传参)、错误友好(出错时返回可读信息而不是堆栈)。

class BaseTool(ABC): name: str description: str parameters: dict # JSON Schema 格式 @abstractmethod def execute(self, **kwargs) -> str: ... class FileReadTool(BaseTool): name = "read_file" description = "读取指定路径的文件内容,用于查看已有代码或文档" parameters = { "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"} }, "required": ["path"] } def execute(self, path: str) -> str: try: with open(path, "r", encoding="utf-8") as f: return f.read() except FileNotFoundError: return f"错误:文件 {path} 不存在,请检查路径" except Exception as e: return f"读取失败:{str(e)}"

注意execute里我捕获了所有异常并返回字符串,而不是让它抛出去。这是血泪教训:如果工具抛异常,整个 agent 循环就崩了;返回一个可读的错误信息,模型反而能根据错误调整策略,比如换个路径重试。这个小小的改动,让我的 agent 在文件操作类任务上的鲁棒性提升了一大截。

3.4 记忆模块:短期与长期的分工

记忆分两层。短期记忆就是当前任务的对话历史,通常直接放在 context 里,随任务结束而销毁。长期记忆是跨任务的知识,需要持久化,一般用向量数据库存储。

短期记忆的关键是裁剪策略。上下文窗口是有限的,当历史消息太多时必须裁剪。我的做法是保留最近 N 轮完整对话,加上更早对话的摘要。摘要用一个便宜的模型生成,成本可控。

长期记忆的关键是写入时机。不是所有信息都值得存。我通常只在两种情况下写入长期记忆:任务成功完成时,把"任务描述 + 成功方案"存进去;遇到并解决了某个典型错误时,把"错误特征 + 解决方案"存进去。这样长期记忆库会越来越"聪明",而不是被垃圾信息淹没。

3.5 编排器:整个系统的大脑

编排器负责三件事:任务分解、agent 调度、状态管理。任务分解通常交给一个专门的 planner agent,它把用户的模糊需求拆成一系列具体子任务。调度则是根据子任务的类型,分派给对应的 agent。状态管理维护一个全局的 task state,记录每个子任务的状态(pending/running/done/failed)。

class Orchestrator: def __init__(self, agents: dict, state_store): self.agents = agents self.state_store = state_store def run(self, user_request: str): state = self.state_store.create(user_request) subtasks = self.agents["planner"].run(state.context) for task in subtasks: agent = self.select_agent(task) try: result = agent.run(state.context) state.mark_done(task.id, result) except Exception as e: state.mark_failed(task.id, str(e)) if not self.try_recover(task, state): break return state.final_result()

try_recover是编排器里最容易被忽略但最重要的部分。任务失败时,是重试、换 agent、还是降级处理?我的经验是:先重试一次(可能是偶发错误),再换一个 agent 试(可能是能力不匹配),最后才标记失败并向上汇报。这套三级恢复机制让我的系统在真实场景下的任务完成率从 70% 左右提升到了 90% 以上。

4. 实操过程与核心环节实现

4.1 环境准备与依赖安装

动手之前先把环境搭好。我推荐用 Python 3.10 以上版本,因为要用到一些新的类型语法。依赖管理用虚拟环境,别偷懒直接装全局。

python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install openai pydantic pyyaml chromadb tenacity

这里几个依赖的作用:pydantic用来做配置和消息的校验,pyyaml读配置文件,chromadb做长期记忆的向量存储,tenacity做重试。如果你不想引入向量库,长期记忆也可以先用简单的 JSON 文件加关键词匹配,后面再升级。

4.2 配置文件:把易变的东西抽出来

agent 的角色、模型、工具这些容易变的东西,全部放到 YAML 里,代码里只读配置。这样改行为不用动代码,也方便做多套配置切换。

agents: planner: model: "gpt-4o-mini" role: "你是一个任务规划专家,负责把用户需求拆解成清晰的子任务列表" tools: [] max_steps: 3 executor: model: "gpt-4o" role: "你是一个执行专家,负责根据任务描述调用工具完成具体操作" tools: ["read_file", "write_file", "run_code", "web_search"] max_steps: 15 reviewer: model: "gpt-4o" role: "你是一个严格的审查者,负责检查执行结果是否符合要求" tools: ["read_file"] max_steps: 5

注意我给 planner 用的是便宜的小模型,因为它只需要做拆解,不需要复杂推理;executor 和 reviewer 用强模型,因为它们要处理具体细节。这种按角色分配模型的做法,能在保证效果的同时把成本压下来。实测下来,整体成本能比全用强模型降低 40% 到 60%。

4.3 一个完整任务的执行流程

假设用户的需求是"帮我分析这个日志文件,找出所有报错并生成一份报告"。整个流程会这样走:

第一步,planner 拆解任务。planner 收到需求后,输出类似这样的子任务列表:

[ {"id": 1, "type": "read", "desc": "读取日志文件内容"}, {"id": 2, "type": "analyze", "desc": "提取所有 ERROR 级别的日志行"}, {"id": 3, "type": "analyze", "desc": "对错误进行分类统计"}, {"id": 4, "type": "write", "desc": "生成分析报告并写入文件"} ]

第二步,编排器分派任务。根据 type 字段,read 和 write 类任务分给 executor,analyze 类任务也分给 executor(因为需要工具),但会带上不同的提示词。

第三步,executor 逐个执行。每个子任务,executor 都会进入自己的决策循环:思考该用什么工具、调用工具、看结果、决定下一步。比如任务 2,它会先调用 read_file 读取日志,然后在返回内容里用代码工具做正则匹配。

第四步,reviewer 审查。所有子任务完成后,reviewer 检查最终报告是否包含所有必要信息,格式是否正确。如果发现问题,它会生成一个"修正任务"重新丢回给 executor。

第五步,状态落库。整个流程的每一步都记录在 state 里,任务结束后可以回放整个执行过程,方便调试和优化。

4.4 关键参数的计算与选择

多智能体系统里有几个参数直接决定成本和效果,我一个个说。

max_steps 怎么定?我的经验公式是:max_steps = 预期工具调用次数 × 2 + 3。比如一个任务预计要调用 5 次工具,那 max_steps 设 13 左右比较合适。设太小任务做不完,设太大浪费 token 且增加死循环风险。

上下文窗口怎么分配?假设模型窗口是 128K token,我会这样分配:系统提示词占 5%,工具定义占 10%,历史对话占 50%,当前任务描述占 10%,留给输出的空间占 25%。历史对话超过 50% 就触发裁剪。

重试次数怎么定?我一般设 2 次重试。第一次重试用相同参数(应对偶发错误),第二次重试换参数或换 agent(应对系统性问题)。超过 2 次还不成功,基本可以判定是任务本身有问题,继续重试只是烧钱。

并发度怎么控制?如果子任务之间没有依赖,可以并行执行。但并发度不要超过 5,否则容易触发 API 限流,而且日志会乱成一团难以调试。我通常设 3。

4.5 实操现场:一次真实的调试记录

说个我印象深刻的调试经历。有一次我的 executor 在处理"读取文件并统计行数"这个简单任务时,反复失败。日志显示它一直在调用 read_file,但每次都返回"文件不存在"。

我先检查了文件路径,没问题。然后我把 executor 的完整对话历史打出来看,发现问题出在路径拼接上:planner 拆解任务时给的描述是"读取 data 目录下的日志文件",executor 理解成了相对路径data/log.txt,但实际工作目录变了,导致找不到文件。

解决方案有两个:一是让 planner 在拆解任务时就输出绝对路径,二是给 executor 加一个"路径解析"工具,专门处理相对路径到绝对路径的转换。我选了第二个,因为更通用。加了这个工具后,类似的路径问题再没出现过。

这个案例的教训是:多智能体系统里,信息在 agent 之间传递时会发生"语义损耗"。planner 脑子里的"data 目录"和 executor 理解的"data 目录"可能不是一回事。解决办法是在关键信息上做显式化,别指望 agent 之间能"心领神会"。

5. 常见问题与排查技巧实录

5.1 常见问题速查表

问题现象可能原因排查方法解决方案
agent 陷入死循环max_steps 过大或提示词有歧义打印完整对话历史降低 max_steps,明确提示词中的终止条件
任务完成但结果不对子任务拆解有误或信息传递丢失检查 planner 输出的子任务列表优化 planner 提示词,增加关键信息显式传递
成本异常高上下文过长或重试过多统计每次调用的 token 数加裁剪策略,限制重试次数
agent 之间互相推诿职责边界不清看哪个 agent 在"踢皮球"在角色定义里明确"什么情况下必须处理"
工具调用参数错误工具描述不清或 schema 不严检查工具定义和模型输出完善 JSON Schema,加参数校验
结果不稳定,时好时坏温度参数过高或缺乏确定性同一任务跑多次对比降低 temperature,关键步骤用确定性逻辑

5.2 独家避坑技巧

技巧一:给每个 agent 加"退出条件"。很多 agent 卡住是因为它不知道该什么时候停。在提示词里明确写"当你完成了 X 或者确认无法完成时,输出 FINISH 并说明原因",能大幅减少无效循环。

技巧二:用"结构化输出"代替"自由文本"。让 agent 输出 JSON 而不是自然语言,解析起来稳定得多。配合 pydantic 做校验,格式错误直接打回重来,比让模型自己纠错靠谱。

技巧三:日志要记全,但要有层次。我通常分三级日志:DEBUG 记每次模型调用的完整输入输出,INFO 记每个子任务的开始结束,ERROR 记异常。平时看 INFO,出问题看 DEBUG。日志量大的话,DEBUG 级别只在需要时开启。

技巧四:给关键 agent 加"自检"步骤。比如 executor 在调用写文件工具前,先让它自己确认"我要写的内容是否完整、路径是否正确"。这个自检步骤会增加一点成本,但能避免很多低级错误。

技巧五:定期回放历史任务。把成功和失败的任务都存下来,定期回放分析。你会发现很多失败是有规律的,针对这些规律优化提示词或流程,比盲目调参有效得多。

5.3 性能优化的几个方向

当系统跑起来后,优化永无止境。我总结下来有三个方向收益最大。

第一是减少不必要的模型调用。很多步骤其实可以用规则或代码完成,不需要模型。比如判断一个字符串是不是 JSON,用代码判断就行,别让模型去判断。我统计过,把这类"确定性判断"从模型手里拿回来,能减少 20% 到 30% 的调用量。

第二是缓存。相同的输入会得到相同的输出(在 temperature 为 0 时),可以缓存起来。特别是那些重复性高的子任务,缓存命中率能到 40% 以上。

第三是并行化。没有依赖关系的子任务并行执行,能把总耗时压缩到原来的三分之一甚至更少。但要注意控制并发度,别把 API 打爆了。

5.4 从 demo 到生产的最后一公里

很多人能把多智能体系统跑出 demo,但上不了生产,卡在几个地方。

可观测性不足。生产系统必须能回答"这个任务为什么失败""花了多少钱""哪一步最慢"。这需要完整的埋点和监控。我通常会给每个 agent 调用打上 trace_id,用统一的日志系统收集,再配一个简单的看板。

错误处理不完善。demo 里出错就崩,生产里出错要能优雅降级。比如搜索工具挂了,agent 应该能切换到"基于已有知识回答"的模式,而不是直接失败。

缺乏人工介入点。有些关键决策,比如"是否删除文件""是否发送邮件",应该设置人工确认环节。完全自动化的系统在关键操作上风险太大。

配置管理混乱。生产环境有多个部署,每个部署的配置可能不同。用环境变量加配置文件的方式管理,别把配置硬编码在代码里。

6. 关于多智能体协作,我的一些真实体会

写到这里,关于"agency-agents"这类多智能体框架的核心内容基本讲完了。最后分享几个我在实际项目中反复验证过的体会,可能比具体的技术细节更重要。

第一个体会是:多智能体不是银弹,很多任务单 agent 加好工具就够了。我见过不少人一上来就搞五六个 agent,结果系统复杂到自己也维护不了,效果还不如一个精心设计的单 agent。判断标准很简单:如果任务能被清晰地拆成几个职责分明的阶段,且每个阶段需要不同的能力或视角,那多智能体有价值;如果任务本身就是连贯的、难以分割的,那单 agent 更合适。

第二个体会是:agent 之间的"接口"比 agent 本身更重要。我花在定义消息格式、任务 schema、状态结构上的时间,往往比写 agent 逻辑的时间还多。但这些投入是值得的,因为接口清晰了,每个 agent 才能独立开发和测试,整个系统才能演进。

第三个体会是:别追求全自动,追求"人机协作的最优解"。完全无人值守的多智能体系统在开放场景下还很脆弱。更务实的做法是让 agent 处理 80% 的常规工作,把 20% 的关键决策留给人。这样既享受了自动化的效率,又保留了人类判断的可靠性。

第四个体会是:持续迭代比一次做对更重要。多智能体系统的行为很难在写代码时就完全预测,必须通过实际运行、观察、调整来逐步优化。我现在的习惯是每周回放一次历史任务,找出失败案例,针对性地改提示词或流程。这种小步快跑的方式,比憋大招一次重构有效得多。

如果你正准备动手搭自己的多智能体系统,我的建议是从最小的可用版本开始:一个 planner、一个 executor、两三个工具,先跑通一个简单任务。跑通之后再逐步加 agent、加工具、加记忆。别一上来就照着最复杂的架构抄,那样你会在调试的泥潭里挣扎很久。先把骨架搭起来,让系统能跑,然后再一点点长肉,这是我从无数次踩坑中总结出的最靠谱的路径。