多Agent协作系统实战：构建数字机构框架与踩坑指南

1. 为什么是"Agency"：多Agent协作的底层逻辑

最近几个月，AI Agent这个概念几乎被聊烂了，但真正把它用到业务里就会发现，单Agent做Demo很容易，做正经事情很难。我一直在折腾一个叫agency-agents的小项目，核心想法其实一句话：与其训练一个无所不能的超级Agent，不如照着一个真实机构的编制，拉一支各司其职的Agent团队。这里的agency我理解为"数字机构"，一群具备自主能力的智能体通过分工协作完成复杂任务。这篇文章会把整个项目的设计思路、角色划分、核心代码和踩坑过程全部摊开讲，给想搞多Agent系统的朋友做一个参考。

为什么需要这样一套东西？因为实际业务里，很少有任务是单一角色能独立完成的。比如一份市场调研报告，既要检索行业数据，又要整理竞品动态，还要形成有逻辑的结论，最后还得做格式规范检查。如果让一个Agent从头干到尾，它既当研究员又当写手还当审核员，结果往往是四不像：数据有了但逻辑乱，逻辑通了但格式垮。agency-agents的思路是把这个流程拆成多个岗位，让每个Agent只负责自己最擅长的一环，用一套消息协议把结果串起来，最终产出一个经过多轮校验的结果。

这个项目适合谁？两类人最值得看：一类是已经在做单Agent应用、但遇到上下文不够用、输出质量不稳定的开发者；另一类是正在规划多Agent系统、但对角色划分和协作机制还没有清晰框架的产品或技术人员。我会尽量写得具体，代码可以直接抄，踩坑记录也按真实时间线来。

1.1 单一Agent的瓶颈到底在哪

先说清楚单Agent模型的三个硬伤，这是所有多Agent方案成立的前提。

第一是上下文窗口被结构性浪费。一个Agent要扮演多个角色，就意味着系统提示词里要堆入研究员、文案、审核员各自的行为规则。这些规则对当前步骤来说是冗余的，但模型每次推理都要把全部角色背景读一遍，上下文窗口被大量无关信息挤占。实测下来，一个中等复杂度的任务，光角色提示词就能吃掉四五千个token，留给真实业务内容的容量所剩无几。

第二是角色冲突很难调和。模型在同一段对话里既要有发散创造力，又要有严谨批判力，这个要求本身是矛盾的。我试过让同一个Agent先写方案再自己评审，它往往会对自己的作品过度宽容，"自我表扬"的倾向非常明显，结构性问题根本看不出来。这不是模型笨，而是缺乏外部视角。人也是这样，自己写完的东西自己检查，总有盲区。

第三是任务切换的隐性成本。单Agent执行完整流程时，每切换一个工作类型，都要在对话里重新"进入状态"。这种切换并不可见，但消耗token，也会让模型在边界处产生状态混乱。比如从资料整理切换到文章撰写，模型可能把"整理摘要"误当成"输出正文"，导致产出内容形态错误。

1.2 多Agent协作带来的三个本质变化

换成多Agent架构后，这三件事会同时发生质变。

专业化是第一层收益。每个Agent只需要维护自己那一份系统提示词，上下文干净，行为边界清晰。研究员可以尽情加搜索工具和网页解析逻辑，写手不需要关心数据格式，评论家只需要专注找茬。提示词短了，模型对指令的遵循率显著提升，输出可控性比单Agent高一个量级。

并行化是第二层收益。机构里从来不要求一个人按顺序把全部事情做完，而是多个岗位同时开工。多Agent系统同样可以把调研、材料整理、初步撰写拆成三个并行任务，让三个Agent各跑各的，最后汇总。这个能力直接缩短了端到端耗时，尤其在依赖多个外部API的场景下感受明显。

可审计性是第三层，也是被很多人忽略的收益。单Agent是一个黑盒，你只知道输入输出，中间推理过程很难追踪。多Agent架构天然有消息流，每一步谁发给谁、结论是什么、为什么转给下一个角色，全部有记录。出了问题可以直接定位到具体环节，这一点在业务落地时价值非常大。

一句话总结：多Agent的本质不是"多几个模型调用"，而是把一个复杂任务拆成多个有边界、有交付物、可验证的工序，每个工序由独立智能体负责，靠协议连接。

2. 架构设计：一个数字机构需要哪些岗位

想清楚为什么之后，接下来是具体设计。我把agency-agents定义为一套"数字机构"框架，所以第一步不是写代码，而是定编制。一个小型咨询公司大概需要哪些岗位？我的答案是最少四个：能拆解任务的调度者、能查资料的执行者、能生产内容的创作者、能挑毛病的质检者。在此基础上，我加了第五个角色做最终交付审核。

2.1 核心角色定义与系统提示词模板

我设计的五个内置角色各有明确职责边界：

角色定义的三个关键参数，我花了不少时间才摸清。第一是职责边界必须用动作动词描述，不要用形容词。研究员提示词里应该写"搜索、筛选、提炼、标注来源"，而不是"认真负责地收集资料"。模型对动词的遵循度远高于对态度的理解。第二是输入输出格式必须非常死板，最好给足模板示例，模型会严格执行格式。第三是约束条件要写清"不要做什么"，比如评论家要明确"只提问题，不提供修改方案"，否则它会越界去改稿。

给一个通用角色提示词模板：

你是一名{角色名称}，在数字机构中担任{职责一句话}。 你的任务边界： - 负责：{动作性职责列表} - 不负责：{明确禁止越界事项} 工作流程： 1. 接收{输入类型} 2. 执行{核心处理步骤} 3. 输出{输出格式} 输出格式要求： {具体格式模板} 约束条件： - {硬性规则1} - {硬性规则2}

这套模板我复用到所有角色上，效果很稳定。核心规律是：模板里的空位越少，模型的表现越可预测。你要把一个角色的行为模式完全压缩进模板，而不是给一堆自由发挥的空间。

2.2 消息协议与协作流转机制

Agent之间怎么说话？我一开始直接让Agent互相往上下文里扔自然语言，结果一塌糊涂：有的Agent根本不回应，有的一回应就扯远。后来我把消息格式标准化，问题立刻解决。

协议很简单，每条消息五个字段：

{ "sender": "coordinator", "receiver": "researcher", "type": "task", "content": "调研2024年智能家居市场Top10品牌动态", "metadata": { "task_id": "T-001", "deadline": "2024-06-01", "priority": "high" } }

type字段包括task、result、review、report、error五类，所有Agent之间只通过这五类消息通信。task是分派任务，result是返回执行结果，review是评审意见，report是最终交付，error是异常反馈。这个设计参考了真实机构里的工单流：每个人都有固定的话术格式，事情才能高效推进。

协作流转机制是整个数字机构的核心。Coordinator拿到用户需求后，先做任务拆解，生成一个动态的有向图。比如"写一份新能源汽车竞品分析"会拆成：

1. 调研头部品牌的销量、定价、核心技术 → Researcher
2. 分析产品定位和市场策略 → Writer
3. 对分析逻辑做严格批判 → Critic
4. 汇总团队意见 → Coordinator
5. 最终交付格式检查 → Reviewer

Coordinator每完成一步，就检查后续依赖是否满足，满足就分派下一个任务，不满足就等待。这个机制本质上是一个由大模型驱动的动态状态机，和普通固定流程的区别在于：下一步往哪走不是写死的，而是Coordinator根据中间产出动态决定。

3. 代码实战：从零构建一个可运行的Agency框架

光说不练假把式。这个项目完整代码大概一千行左右，核心模块四个：Agent基类、消息协议、调度器、工具注册表。下面按依赖顺序拆开讲。

3.1 技术选型与依赖清单

技术栈非常克制，没有引入重量级框架。我坚持一个原则：先用手写代码把逻辑跑通，确认机制没问题后再考虑上框架。项目核心只有三个依赖：

为什么不用现成的多Agent编排库？不是它们不好，而是它们抽象层级太高，出了问题很难排查。自己写一遍状态机之后，你才能真正理解Agent协作的每一步发生了什么，之后再去用框架，就是降维打击了。这个项目里调度器核心不到两百行代码，但完全可控。

3.2 Agent基类：如何封装模型调用与工具执行

所有角色都继承同一个Agent基类。这个基类做三件事：接收消息、调用模型、执行工具。

from abc import ABC, abstractmethod from typing import Any class Agent(ABC): """所有角色的基类""" def __init__( self, name: str, system_prompt: str, model: Any, tools: dict[str, Any] | None = None, ): self.name = name self.system_prompt = system_prompt self.model = model self.tools = tools or {} self.memory: list[dict] = [] self.token_usage = 0 def send_message(self, message: dict) -> None: """把收到的消息追加进记忆""" self.memory.append(message) self._trim_memory() def act(self) -> dict: """调用模型，决定下一步动作""" messages = self._build_messages() response = self.model.chat(messages) self.token_usage += response.usage.total_tokens if response.has_tool_call(): tool_result = self.execute_tool( response.tool_call.name, response.tool_call.arguments, ) return {"type": "result", "content": tool_result} return {"type": "result", "content": response.text} def execute_tool(self, tool_name: str, arguments: dict) -> str: if tool_name not in self.tools: return "错误：不存在的工具" try: return str(self.tools[tool_name](**arguments)) except Exception as e: return f"工具调用失败：{e}" def _build_messages(self) -> list[dict]: """构造发往模型的完整消息列表""" return [{"role": "system", "content": self.system_prompt}] + self.memory def _trim_memory(self) -> None: """记忆裁剪，防止上下文无限膨胀""" if len(self.memory) > 20: self.memory = self.memory[-20:]

基类设计的两个关键点。第一，所有模型调用统一走chat接口，后续想换模型只需要改model实例，不用动其他代码。第二，工具的调用和执行完全解耦，模型只负责输出"要调用哪个工具、参数是什么"，执行交给execute_tool方法。这个设计隔离了不确定性：模型可能输出错误的参数，但不会导致整个Agent崩溃。

3.3 调度循环：让Agent们有序讨论

调度器是整个项目的发动机。它的核心是一个while循环，每一轮派发一条待处理消息，直到满足终止条件。

class AgencyOrchestrator: def __init__( self, agents: dict[str, Agent], coordinator: Agent, max_rounds: int = 15, ): self.agents = agents self.coordinator = coordinator self.max_rounds = max_rounds self.history: list[dict] = [] def run(self, user_request: str) -> str: # 初始任务分派 self._dispatch_none() task_plan = self.coordinator.act( f"请拆解以下用户需求并输出任务清单：{user_request}" ) # 解析任务清单，逐个分派 for task in task_plan.get("tasks", []): self._dispatch_to(task) for _ in range(self.max_rounds): self._process_pending() if self._is_finished(): break self._check_deadlock() return self._collect_final_report() def _dispatch_to(self, task: dict) -> None: receiver = task["receiver"] if receiver not in self.agents: # 找不到接收者就抛给质检查 receiver = "reviewer" self.agents[receiver].send_message( { "sender": self.coordinator.name, "receiver": receiver, "type": "task", "content": task["description"], "metadata": task, } )

第一次写这个循环时我犯了个错误：让所有Agent异步乱跑，完全没有轮次概念。结果消息顺序不可控，Coordinator还没拆解完毕，Researcher已经收到一堆无头任务。后来老老实实改成"同步轮询+每轮最多处理一条消息"，才稳定下来。这个循环的关键不是性能，而是状态可追踪。

3.4 工具注册表：给Agent装配武器

角色只有提示词是空架子，必须配工具。工具注册表用装饰器实现，任何普通函数都可以变成一个Agent工具。

class ToolRegistry: """全局工具注册中心""" def __init__(self): self._tools: dict[str, callable] = {} def register(self, name: str, description: str, schema: dict): def decorator(func): self._tools[name] = { "function": func, "description": description, "schema": schema, } return func return decorator registry = ToolRegistry() @registry.register( "web_search", "搜索指定关键词并返回结果列表", { "type": "object", "properties": { "keyword": {"type": "string", "description": "搜索关键词"}, "limit": {"type": "integer", "description": "返回条数"} }, }, ) def web_search(keyword: str, limit: int = 5) -> str: # 这里接某个搜索服务API return f"搜索到 {limit} 条关于 {keyword} 的结果"

工具schema一定要严格，因为大模型依赖schema生成合法的工具调用参。我踩过的一个坑是，schema里不写description，模型就经常漏传参数；写了description之后，漏传率大幅下降。工具返回结果最好统一成字符串，这样Agent能直接把它塞进上下文里做分析，不用做二次类型转换。

4. 实操踩坑实录：这些问题我花了三个晚上才解决

这套框架运行起来之后，真正的磨难才开始。这一节把所有踩过的坑按严重程度排个序，帮助后面的人少走弯路。

4.1 上下文爆炸：记忆该存不该全塞

第一版实现里，我让每个Agent保存全部历史消息，以为这样信息最全。结果跑了三轮协作，消息量直接翻倍，模型开始把几天前的旧消息当作当前指令执行，输出质量断崖式下跌。

排查之后发现两件事。一是单个Agent的memory需要裁剪，我最终限定最多保留最近20条消息。二是Agent之间不能盲目转发全部历史，而应该基于当前任务的相关性，构造一个"精简包":只包含任务描述、当前待处理内容、必要的背景摘要。这个摘要可以由Coordinator在分派时生成，也可以由接收方自己提炼。我的做法是加了一步"记忆压缩"：每次长对话结束后，用一个小模型把历史消息压缩成300字以内的摘要，后续轮次只传摘要。

def compress_history(messages: list[dict]) -> str: # 提取关键事实、结论、待办事项 # 压缩为结构化摘要 summary = [] for m in messages: if m["type"] in ("result", "review"): summary.append(f"{m['sender']} 提到: {m['content'][:100]}") return "\n".join(summary[-20:])

这个压缩摘要的效果非常显著，token成本直接降了40%，而且信息完整性没有明显下降。

4.2 死循环与幻觉：最大迭代次数要卡死

多Agent协作最让人崩溃的问题是无休止的循环。Critic批评Writer的稿子，Writer根据意见修改，Critic再次批评，Writer发现Critic的新意见和自己原来的想法矛盾，开始反驳，Critic再辩论。如果没有终止条件，这两个Agent能吵到天荒地老。

解决方案分两层。第一层是硬性保护：max_rounds设置上限，我默认15轮，超过立即终止并返回当前最优版本。这个数字不是拍脑袋定的，我统计过正常任务平均需要7到10轮，15轮足够覆盖绝大多数情况。第二层是软性仲裁：给Coordinator加一个特殊指令，当检测到两个Agent陷入重复争论时，由Coordinator拍板，不再让双方继续互喷。

def _check_deadlock(self) -> None: """检测是否出现重复争论""" last_two_critics = [ m for m in self.history if m["type"] == "review" ][-2:] if len(last_two_critics) >= 2: c1 = last_two_critics[-2]["content"] c2 = last_two_critics[-1]["content"] if self._similar(c1, c2) > 0.85: # 争议内容高度重复，强制进入交付流程 self.force_finish = True

这个"强制收敛"机制非常有用，可以把项目从无限讨论中解救出来。

4.3 工具调用失败：解析要足够鲁棒

模型输出不规范的JSON是家常便饭。工具调用参数里多一个逗号、少一个引号、字符串里混入换行符，都会导致解析失败。第一版我的Agent遇到解析失败就直接error返回，结果一个简单的调研任务因为某个临时工具崩溃，整条链路都断了。

后来我写了一个三方容错解析器：

import json import re def robust_json_parse(raw: str) -> dict: """三段式解析：直接解析→清理后解析→正则提取""" if not raw: return {} # 尝试1：直接解析 try: return json.loads(raw) except json.JSONDecodeError: pass # 尝试2：去除包装代码块 cleaned = re.sub(r"^```(?:json)?|```$", "", raw.strip()) # 尝试3：提取最外层花括号并修复尾逗号 match = re.search(r"\{.*\}", cleaned, re.DOTALL) if match: candidate = match.group(0) candidate = re.sub(r",\s*}", "}", candidate) try: return json.loads(candidate) except json.JSONDecodeError: pass return {"error": "无法解析", "raw": raw}

三层解析兜底之后，工具调用成功率从70%左右提升到了95%以上。剩下的5%，我就让Agent把原始输出原样返回，Coordinator收到error消息后会自动重试一次。这个策略简单但有效。

4.4 调试技巧：可视化追踪每个Agent的思考

多Agent系统调试最大的痛点是看不见内部过程。我花了好几天加日志，最终沉淀了一套非常好用的可视化方案。

每个Agent在日志里分配一个独立前缀和颜色码。Coordinator用青色，Researcher用绿色，Writer用蓝色，Critic用红色，Reviewer用紫色。每轮消息打印时，带上时间戳 角色 类型 内容摘要。这样一眼就能看清当前是哪个角色在发言、在做什么。

def log_message(message: dict) -> None: color_map = { "coordinator": "\033[96m", "researcher": "\033[92m", "writer": "\033[94m", "critic": "\033[91m", "reviewer": "\033[95m", } color = color_map.get(message["sender"], "\033[0m") timestamp = time.strftime("%H:%M:%S") content_preview = message["content"][:80].replace("\n", " ") print( f"{color}[{timestamp}] {message['sender']}" f" -> {message['receiver']} " f"[{message['type']}] {content_preview}\033[0m" )

实际操作中，我很推荐再加一个token计数器，每个Agent调用模型后打印累计token消耗。多Agent系统的成本往往出乎意料，没有这个计数器，项目跑完才知道账单，根本调整不过来。

5. 后续可以怎么玩：至少三个扩展方向

现在的版本已经能稳定跑通调研、写作、评审闭环。如果有兴趣继续扩展，我给三个可以深入的方向。

5.1 持久化记忆

当前所有Agent的记忆都在内存里，进程一结束就全没了。下一步可以把消息历史序列化存入本地文件或向量数据库，做成"带记忆的数字机构"。这样下次跑相同的任务时，Coordinator能回忆起上次的调研结论，避免重复工作。这个扩展对长期运营类任务价值很大。

5.2 多模态与工具扩展

整个框架的Agent基类并没有绑定文本模态，只要模型支持，就可以接入图片理解、语音转写、文档解析等工具。比如让Researcher接入OCR，直接把扫描件变成结构化文本；让Writer接入图片生成，产出配图。工具的扩展成本非常低，因为工具注册表已经把接口统一了。

5.3 人在环路（Human-in-the-loop）

纯粹让Agent自主运转，在关键决策节点上还是有点让人不放心。我给框架加了一个人工审批钩子：当Coordinator准备交付最终报告前，会先发一条pending_review消息给一个人工API。人工可以一键通过，也可以附上修改意见返回给Writer。这个钩子控制成本不高，但对质量保障非常关键，尤其在面向外部客户的场景下，人工审批几乎是必须的。

我个人在这个项目上最大的体会是，多Agent系统的真正难点不是技术，而是"组织设计"。你怎样定义角色边界，怎样设计消息协议，怎样设定终止条件，这些才是决定成败的因素。代码反而是最不重要的部分。如果你正准备做类似的事，我建议先用纸笔画清楚团队架构，搞清楚每一步谁做什么事、产出交给谁，然后再打开编辑器。这套思路到今天仍然适用，也希望这个项目能帮你节省几晚踩坑的时间。