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 核心角色定义与系统提示词模板我设计的五个内置角色各有明确职责边界角色职责输入输出Coordinator拆解用户需求分派任务汇总结果用户原始请求任务拆解清单、最终交付Researcher检索资料、整理事实、提炼要点调研任务书结构化的材料清单Writer基于材料撰写内容写作任务书材料初稿Critic攻击逻辑漏洞、事实错误、表达问题待评审文本评审意见Reviewer做最终一致性、格式、合规检查终稿交付校验报告角色定义的三个关键参数我花了不少时间才摸清。第一是职责边界必须用动作动词描述不要用形容词。研究员提示词里应该写搜索、筛选、提炼、标注来源而不是认真负责地收集资料。模型对动词的遵循度远高于对态度的理解。第二是输入输出格式必须非常死板最好给足模板示例模型会严格执行格式。第三是约束条件要写清不要做什么比如评论家要明确只提问题不提供修改方案否则它会越界去改稿。给一个通用角色提示词模板你是一名{角色名称}在数字机构中担任{职责一句话}。 你的任务边界 - 负责{动作性职责列表} - 不负责{明确禁止越界事项} 工作流程 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拿到用户需求后先做任务拆解生成一个动态的有向图。比如写一份新能源汽车竞品分析会拆成调研头部品牌的销量、定价、核心技术 → Researcher分析产品定位和市场策略 → Writer对分析逻辑做严格批判 → Critic汇总团队意见 → Coordinator最终交付格式检查 → ReviewerCoordinator每完成一步就检查后续依赖是否满足满足就分派下一个任务不满足就等待。这个机制本质上是一个由大模型驱动的动态状态机和普通固定流程的区别在于下一步往哪走不是写死的而是Coordinator根据中间产出动态决定。3. 代码实战从零构建一个可运行的Agency框架光说不练假把式。这个项目完整代码大概一千行左右核心模块四个Agent基类、消息协议、调度器、工具注册表。下面按依赖顺序拆开讲。3.1 技术选型与依赖清单技术栈非常克制没有引入重量级框架。我坚持一个原则先用手写代码把逻辑跑通确认机制没问题后再考虑上框架。项目核心只有三个依赖依赖用途版本建议Python主开发语言3.10大模型APIAgent推理引擎任意支持function calling基础工具库JSON解析、HTTP请求requests、pydantic为什么不用现成的多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系统的真正难点不是技术而是组织设计。你怎样定义角色边界怎样设计消息协议怎样设定终止条件这些才是决定成败的因素。代码反而是最不重要的部分。如果你正准备做类似的事我建议先用纸笔画清楚团队架构搞清楚每一步谁做什么事、产出交给谁然后再打开编辑器。这套思路到今天仍然适用也希望这个项目能帮你节省几晚踩坑的时间。