「agency-agents」这个项目标题，我第一次看到的时候第一反应是：这又是一个套壳的 agent 框架吧？但真正把它拆开之后我发现，它跟我预想的不太一样——它确实是在认认真真地研究「多个智能体怎么像一家代理机构那样协同工作」这件事。这两年大模型智能体（Agent）很火，但大多数项目都停留在单 Agent 的“对话即服务”层面，而一旦任务稍微复杂一点，比如要你同时完成市场调研、内容策划、文案写作、质量审核一整条链路，单 Agent 就会明显力不从心：上下文越拖越长、中间结果没人管、错误一路传导无法收敛。
「agency-agents」要解决的，正是这个痛点。它把多个智能体组织成一个“虚拟代理机构”，让不同角色各司其职，再通过一个核心的编排层去拆解任务、调度角色、汇总成果。这篇文章我会完整拆解它的设计思路、核心机制和实操方法，并且给出一套可以直接跑起来的 Python 示例，带你从零构建一个“文本写作代理机构”。无论你是做 AI 应用开发的工程师、研究 Agent 架构的产品经理，还是想把大模型真正落地到业务流程中的决策者，这篇内容都值得认真看完。
1. 项目定位：它不是一个 Agent，而是一整套“代理机构”
「agency-agents」这个名字里的 agency 一词语义很关键。在英文语境里，agency 既有“代理”的意思，也有“机构”的意思。这个项目想做的是后者：把多个 Agent 组织成像公司咨询部门那样的专业机构，让每个成员只负责自己最擅长的那一段工作。
1.1 从“单个助手”到“虚拟机构”的思路转变
单一 Agent 的工作模式，本质上是一个“无限上下文的对话助手”。你把整个任务丢给它，它在一段密密麻麻的对话里完成所有事。短期任务没问题，但一旦任务周期拉长、子任务变多，单 Agent 的短板就暴露出来了：
- 上下文窗口是有限的，尤其是当你需要处理长文档、多轮外部反馈时，早期的关键信息会被挤出去。
- 所有角色能力都被塞进同一个 prompt 里，容易产生角色混淆（我叫你写代码，你却顺带把测试也写了）。
- 中间产物不可复用，出了问题只能从头再来。
- 无法并行，一条链路走到底，效率很低。
而「agency-agents」的思路更像一家真实的公司：你有一个老板角色（Orchestrator）负责接活和拆活；下面有策划 Agent 负责出方案；有执行 Agent 负责把方案落地成具体产出物；还有质检 Agent 负责检查交付物是否合格。每个角色都有自己的独立上下文、独立工具权限和独立输出标准，任务以工作流的形式在角色之间流转。
这个转变的价值不是“更酷”，而是“更可控”。代理机构模式让每个环节都有对应的负责人，哪一步出了问题你直接定位到那个角色就行，而不是在成千上万行对话记录里大海捞针。
1.2 “agency-agents”解决的核心问题
拆解这个项目，我认为它重点解决了四个问题，这也是你在任何多智能体系统设计时都会遇到的共性问题：
- 任务拆解与角色分配：老板 Agent 需要把“帮我写一份季度营销方案”拆解成“分析目标用户”“整理竞品数据”“撰写策略框架”“生成最终方案”等可控的子任务，并且知道该分配给谁。
- 上下文隔离：每个 Agent 只看得到自己需要的输入和输出，而不是整个任务的所有历史记录。这样既节省了 token，也避免了无关信息干扰判断。
- 可追踪的中间状态：每一步谁在做、做到什么程度、产出物是什么，都有明确的记录和校验机制，而不是全靠 LLM 自由发挥。
- 权限边界与工具隔离：不是每个 Agent 都能调用所有工具。比如负责写文案的 Agent 不应该有删除文件的权限，这是工程安全层面的基本要求。
这几点听起来简单，但真正落地的时候每一处都容易出问题。我自己在尝试复刻这个项目的核心逻辑时，踩过不少坑，后面我会专门用一章节来说。
1.3 哪些人适合从「agency-agents」入手
如果你属于下面某一类人，这个项目会是一个很好的参考：
- 你正在做 AI 自动化流程，比如把“内容生产—审核—发布”这一整条链路交给 LLM 去跑。
- 你研究 Agent 的架构编排，想知道除了 LangChain 这种重框架之外，有没有更轻量、更符合业务组织逻辑的写法。
- 你负责技术决策，想在团队里推进多智能体协作，需要一份能讲清楚“为什么这样做”的参考设计。
- 你带了不少搞 Agents 的实习生，发现大家都在堆 prompt，没有一个系统化的协作机制。
这个项目不是银弹，它不会让你“一行代码获得一个万能智能体团队”，但它提供了一套结构化的思考框架和实现路径。我推荐你把它当作一个多智能体系统的“最小可行样例”来读，而不是一个开箱即用的产品。
2. 整体架构与核心设计拆解
「agency-agents」的整体架构并不复杂，关键就一句话：分层 + 状态机 + 消息机制。下面我把这三层摊开讲。
2.1 分层架构：编排层-角色层-工具层
我把这个项目的核心架构抽象成三层：
- 编排层（Orchestration Layer）：负责接收用户任务，拆解成子任务，决定执行顺序和谁先谁后，并汇总结论。
- 角色层（Agent Layer）：具体干活的一群 Agent。每个 Agent 有自己的系统提示词、上下文窗口、可用技能列表和输出格式约束。
- 工具层（Tool Layer）：Agent 可以调用的外部能力，比如搜索引擎、代码解释器、内部 API、文件读写工具等。
这三层之间是严格调用的：编排层不直接碰工具，角色层不直接接收用户原始需求，工具层只服务角色。
为什么要这样做？用生活里的例子类比就很好理解：你在公司里不会直接跑到前台小妹那里说“我要一份全年的财务报告”，你会找部门负责人，由他拆解需求并派发到财务岗位，前端岗位只负责自己那部分数据。这样每一层的职责都清晰，出了问题也好追责。
在代码层面，这三层通常对应三个基类：
class AgentBase: """角色层基类，所有角色的公共能力""" def __init__(self, name: str, system_prompt: str, tools: list, memory: MemoryStore): self.name = name self.system_prompt = system_prompt self.tools = tools self.memory = memory async def run(self, task: Task) -> AgentResult: raise NotImplementedError class OrchestratorBase: """编排层基类，管理任务拆解、调度、汇合""" def __init__(self, agents: dict[str, AgentBase], max_steps: int = 10): self.agents = agents self.max_steps = max_steps self.task_history = [] async def execute(self, user_request: str) -> dict: raise NotImplementedError class ToolBase: """工具层基类，定义外部能力的最小接口""" async def execute(self, **kwargs) -> ToolResult: raise NotImplementedError这套代码写得非常朴素，但结构很稳。你可以在不改变其他层的情况下，随时新增一个 Agent 角色，或者替换底层工具实现，灵活性很高。
2.2 核心机制：任务状态机与消息传递
多智能体系统最容易踩的坑，就是“没有流程感”。多个 Agent 各说各话，对话结果随机性很大，系统无法判断到底哪个环节算完成。所以「agency-agents」引入了任务状态机，每个子任务都会经历这几个状态：
| 状态 | 含义 | 触发条件 |
|---|---|---|
| pending | 任务已创建，等待调度 | 拆解任务完成 |
| planning | 负责该任务的 Agent 正在制定执行计划 | 编排器派发到某个角色 |
| executing | Agent 正在调用工具或模型生成结果 | 计划通过评审 |
| reviewing | 结果进入质检环节 | 执行产出物生成 |
| done | 任务验收通过，结果归档 | 质检 Agent 返回 pass |
| failed | 任务在多轮重试后仍无法通过 | 达到最大重试次数或抛出致命错误 |
状态机并不是什么新鲜东西，但它极大提升了系统的可控性。有了状态，你就可以在关键时刻做干预：如果某个任务在 reviewing 状态连续失败两次，你可以让编排器换人重做，而不是让它无限循环下去。
消息传递机制则定义了 Agent 之间如何沟通。常见的消息类型有三种：
- direct message：点对点发送，比如编排器把任务单发给策划 Agent。
- broadcast：广播消息，适合“全员注意，需求变更了”这种场景。
- tool call：工具调用消息，表示某个 Agent 正在请求使用外部能力。
把消息机制显式化，最大的好处是方便做日志追踪。你不会看到一堆“天书”一样的对话记录，而是一条条结构化的事件流：某某 Agent 在某时接收了某消息，调用了某工具，返回了某结果。这才能支撑后续的监控、回溯和调优。
2.3 对比主流方案：为什么不用现成的 Agent 框架
你可能会问：LangChain、AutoGen、CrewAI 都现成，为什么还要折腾「agency-agents」？我的看法是，它解决的是“组织逻辑”的抽象问题，而不是“调用接口”的工具问题。
LangChain 和 AutoGen 这类框架，功能非常强，但也非常重。它们提供了一堆高度抽象的类，你要花很多时间搞清楚该继承哪个、该配置什么，而且它们默认的单 Agent / 对话式交互模式，跟机构协作的节奏并不一致。CrewAI 虽然也是多 Agent 协作，但它的设计重心是“角色扮演和任务委派”，而不是“带状态流转和职责隔离的代理机构”。
「agency-agents」的核心参考价值，在于它用极少的代码量、极清晰的边界，向你展示了多智能体协作的最小完整架构。它没有一堆花哨的接口，所以你很容易把它改成自己想要的样子。如果让我一句话总结：它更像一张“多智能体机构化协作的参考图纸”，而其他框架更像“已经拼好的乐高积木”。图纸的价值在于你可以理解每一个零件的连接逻辑，而积木的价值在于上手快。
3. 核心环节实操：30 分钟搭建一个“文本写作代理机构”
有了架构层面的理解，下面直接实操。我来带你把「agency-agents」的核心思想落成一个可运行的“文本写作代理机构”。这个示例会包含三个角色：策划 Agent、写作 Agent、校对 Agent，外加一个编排器。
3.1 环境准备与依赖安装
我建议你用 Python 3.10 及以上版本。需要一个异步框架的支持，我直接用asyncio标准库，再配合 openai SDK 来调用大模型。项目本身只用几个关键库，安装命令如下：
pip install openai asyncio pydantic python-dotenv在项目根目录创建.env文件，写入你的 API Key 和模型配置：
OPENAI_API_KEY=你的key MODEL_NAME=gpt-4o-mini这个示例不绑定任何特定服务商，理论上任何兼容 OpenAI API 的大模型接口都能跑。你要是用国内模型，只要把 endpoint 和 key 换成对应的就行。
3.2 定义三个角色：策划 Agent、写作 Agent、校对 Agent
第一步先实现角色层。每个角色就是一个独立的 Agent 类，它们在初始化时拿到自己的系统提示词、可用工具和记忆存储。
# agents.py from dataclasses import dataclass, field from typing import List, Optional @dataclass class Task: id: int type: str # planning / writing / reviewing content: str # 任务说明或产出物内容 metadata: dict = field(default_factory=dict) @dataclass class AgentResult: status: str # success / failed / need_retry output: str = "" feedback: str = "" error_msg: str = "" class PlanningAgent: name = "策划Agent" def __init__(self, llm_backend): self.llm = llm_backend async def run(self, task: Task) -> AgentResult: prompt = self._build_prompt(task.content) result = await self.llm.generate(prompt) return AgentResult(status="success", output=result) class WritingAgent: name = "写作Agent" def __init__(self, llm_backend): self.llm = llm_backend async def run(self, task: Task) -> AgentResult: prompt = self._build_prompt(task.content) result = await self.llm.generate(prompt) return AgentResult(status="success", output=result) class ReviewingAgent: name = "校对Agent" def __init__(self, llm_backend): self.llm = llm_backend async def run(self, task: Task) -> AgentResult: prompt = self._build_prompt(task.content) result = await self.llm.generate(prompt) # 简单处理：如果大模型反馈包含 "PASS" 就视为通过，否则需要重写 if "PASS" in result.upper(): return AgentResult(status="success", output=task.content, feedback="通过") return AgentResult(status="need_retry", feedback=result)这里只展示了类的骨架，实际你在写_build_prompt方法时，要把每个角色的专业要求和输出格式写清楚。比如策划 Agent 的 prompt 要包含“给出文章结构、大纲、目标读者和关键信息点”，而校对 Agent 的 prompt 要包含“检查逻辑漏洞、语法错误、与大纲的一致性，通过则返回 PASS，否则给出修改意见”。
3.3 实现编排器：把任务拆解并派发下去
编排器是「agency-agents」的真正大脑。它负责四件事：
- 接收用户请求。
- 把请求拆解成一个任务清单。
- 按照依赖顺序把任务派发给对应角色。
- 在最终结果产出前，经过质检闭环。
# orchestrator.py class WritingOrchestrator: def __init__(self, planning: PlanningAgent, writing: WritingAgent, reviewing: ReviewingAgent): self.agents = { "planning": planning, "writing": writing, "reviewing": reviewing, } self.max_retries = 3 async def execute(self, user_request: str) -> dict: # Step 1: 拆解任务 plan_task = Task(id=1, type="planning", content=f"为以下主题制定写作大纲：{user_request}") plan_result = await self.agents["planning"].run(plan_task) if plan_result.status != "success": return {"status": "failed", "error": "策划环节出错"} # Step 2: 写作执行 write_task = Task(id=2, type="writing", content=plan_result.output) write_result = await self.agents["writing"].run(write_task) if write_result.status != "success": return {"status": "failed", "error": "写作环节出错"} # Step 3: 校对与重试闭环 review_task = Task(id=3, type="reviewing", content=write_result.output) for attempt in range(self.max_retries): review_result = await self.agents["reviewing"].run(review_task) if review_result.status == "success": return {"status": "success", "final_output": review_result.output} # 需要重写：把校对反馈拼接进改写指令 rewrite_task = Task(id=4, type="writing", content=f"根据反馈修改文稿：\n{review_result.feedback}") write_result = await self.agents["writing"].run(rewrite_task) review_task.content = write_result.output return {"status": "failed", "error": "多次审核未通过"}这个编排逻辑兼顾了执行和质检。你可能会问：这个流程会不会太简单了？现实中的文案写作任务当然比这复杂，但它已经具备了「agency-agents」最核心的闭环要素：拆解、分工、执行、质检、反馈重试。
3.4 运行效果演示与关键参数调优
你可以在主程序里跑一下，输入一个实际需求：
python main.py运行后会看到类似这样的输出（这是我本地跑的一个示例，主题是“智能家居行业分析”的报告写作）：
[编排器] 收到用户需求：写一篇1000字左右的智能家居市场趋势分析 [编排器] 已派发策划任务给策划Agent [策划Agent] 产出大纲：1. 行业规模 2. 技术趋势 3. 用户需求变化 4. 未来五年预测 [编排器] 已派发写作任务给写作Agent [写作Agent] 完成初稿，字数约1100字 [编排器] 已派发校对任务给校对Agent [校对Agent] 返回建议：第三小节数据与开头引用不一致，建议补充具体出处；结尾缺乏明确结论，建议增加两句总结 [编排器] 发起第2轮改写 [写作Agent] 完成二稿，已修正引用问题与结论 [校对Agent] 返回 PASS [编排器] 任务完成，最终输出已写入 result.md要跑得更稳，有几个参数值得关注：
temperature：写作类 Agent 建议设为 0.7 左右，保证一定的创造性；校对 Agent 建议设为 0.2 以下，减少误判和随机性。max_retries：我一般设为 2~3。尽量让质检 Agent 给出结构化反馈，而不是简单说“不行”。timeout：工具调用和模型响应都需要设置超时，避免因为某个环节卡死导致整个任务挂起。max_steps：编排器里的整体最大步数限制，防止任务在极端情况下死循环。
执行方式要强调一个细节：因为角色之间是异步流转，你在定义 Agent 的run方法时必须用async/await。如果某个环节用了同步阻塞的 SDK（比如某些文件读取库），记得用asyncio.to_thread包装一下，否则会阻塞整个事件循环，多智能体并行效率就废了。
4. 常见问题与排查技巧实录
这部分是我最想写的内容。因为「agency-agents」这类多智能体系统，demo 跑起来很容易，一上线就各种毛病。我把实际操作中遇到的典型问题，以及排查思路整理成一个速查清单。
4.1 任务陷入死循环，怎么办？
这是遇到最多的一个坑。比如校对 Agent 反复要求“再补充一个例子”，写作 Agent 真就反复加，但永远不会满足，于是两个 Agent 来回拉扯，直到 token 烧完。
排查思路：
- 查看任务状态机是不是卡在
reviewing和executing之间反复跳转。 - 设置全局最大步数和最大 token 限制，比如
orchestrator.max_steps=5，超了直接置为失败。 - 检查校对 Agent 的 prompt 是否给了明确的“通过标准”。最好要求它输出结构化的结果：
{"verdict": "PASS"}或{"verdict": "NEED_FIX", "issues": [...]}，而不是自由文本。
我自己的经验是，每次在校对环节给出具体的通过标准（比如“逻辑结构是否与大纲一致，是/否；有没有明显的事实错误，有/无；结尾是否提出明确的结论，是/否”），死循环的概率会大幅下降。
4.2 角色上下文串扰，输出张冠李戴
另一个高频问题：写作 Agent 一不小心就把策划大纲里的内容当成自己的正文输出，或者策划 Agent 在收到反馈后开始改写具体段落，角色职责完全乱套。
根源：所有角色共享了同一个上下文库，或者你在构建消息时没有加角色前缀。
解决办法：
- 在 MemoryStore 里按
agent_name做命名空间隔离，每个 Agent 只允许读取自己的记忆。 - 任务传递时，只传“上一个环节的最终输出”，不要把所有中间草稿都丢过去。
- 在系统提示词里明确“你的职责是X，其他事情一概不做”。
4.3 Token 超限，成本飞涨
多智能体协作框架看着漂亮，成本账单出来的时候才是真的头疼。尤其是多轮重试机制，一次写作任务可能调用几十次大模型接口。
降本思路：
- 中间过程用便宜的模型（如 mini / 轻量版本），最终整合输出时再用更强模型。
- 开启消息压缩：在任务流转到下一环节前，让当前 Agent 先输出一份 200 字以内的“要点摘要”，替代原始全文进入后续环节。
- 结果缓存：如果某个任务在近期被处理过，且输入参数完全一致，直接复用之前的产出物。
举个例子：我在构建一个“周报生成机构”时，策划 Agent 每次都会拿到一堆聊天记录做输入。如果不压缩，这堆记录会完整传给写作 Agent，再传给校对 Agent，一份周报烧掉的 token 是正常需求的 6 倍。后来我在策划 Agent 结束时加了一个“提炼关键信息”的步骤，成本直线下降。
4.4 工具调用权限越界，安全风险
多 Agent 系统最常见的风险是，某个 Agent 获得了过大的工具权限，比如文件读写、代码执行、发送消息的 API。一旦 prompt 注入或者模型幻觉爆发，轻则删错文件，重则泄露数据。
基本安全规范：
- 每个 Agent 的可调工具必须白名单化，和它的角色职责强相关。比如写作 Agent 只能调用“搜索资料”和“读取本地文件”，不能调用“删除文件”和“发送 HTTP 请求”。
- 工具执行前加一道人工审批钩子（类似
require_human_approve=True）用于高危操作，比如生产环境的代码执行。 - 所有工具执行的输入输出都要记日志，方便事后审计。
我把常见问题整理成一个速查表，方便你对照排查：
| 现象 | 大概率原因 | 处理建议 |
|---|---|---|
| Agent 之间来回重复对话 | 缺少全局步数限制 | 设 max_steps，超时强制失败 |
| 输出内容与角色职责不符 | 上下文串扰 | 内存按 Agent 命名空间隔离，只传必要字段 |
| 任务很快完成但结果质量差 | 编排器拆解颗粒度太粗 | 增加中间质检环节，拆分子任务粒度 |
| 某一步一直超时 | Agent 调用了慢工具且无超时设置 | 给所有工具调用加 timeout |
| 成本远超预期 | 中间消息未压缩 | 增加摘要机制，使用轻量模型做中间环节 |
| 校对总是不通过 | 质检标准模糊 | 让校对 Agent 输出结构化 verdict |
4.5 实测下来的独家避坑技巧
最后补充一条我在实际复刻中发现的小技巧：不要让编排器承担“太聪明的角色”。编排器的主要职责是流程控制，不是内容创造。如果你让编排器去判断“写作 Agent 的输出好不好”，它既没有专业上下文，又容易给出廉价判断。更好的做法是引入一个独立的“评审 Agent”来做质量把关，编排器只负责把评审结果翻译成下一步指令。
5. 从 Demo 到生产：更工程化的思考
很多项目跑通 demo 就完了，但「agency-agents」更大的价值，是它提供了一个可以往生产方向走的起点。我觉得有几个点值得继续深入。
5.1 可观测性：日志、追踪与效果评估
生产环境的多智能体系统，最怕的就是“黑盒”。一个任务到底经历了哪些 Agent，每个 Agent 的状态是什么，中间产出了什么，这些都要有完整的日志链路。你可以给每个任务分配一个trace_id，在消息流转时一路透传，再用一个日志服务把所有节点信息收拢。
具体来说，我建议至少要记录四类信息：
- 任务元信息（输入需求、最终输出、完成耗时、总 token 数）。
- 每个 Agent 的输入输出快照。
- 每次工具调用的请求和响应。
- 异常事件（重试、超时、失败），以及当时的上下文摘要。
有这些数据，做效果评估才有依据。评估指标不只有“任务完成率”，你还得看“平均重试次数”“每任务 token 消耗”“返回结果与用户需求的语义匹配度”等。
5.2 动态角色扩展与人工审批节点
真实业务不会像 demo 这么固定。我建议在设计时把“角色列表”做成可配置的，甚至可以让编排器根据任务类型动态增删角色。
比如，你可以在一个工作流中动态添加“合规审核 Agent”，专门检测输出内容是否符合合规要求，再添加“数据准确性验证 Agent”，在最终输出前做一轮事实核查。这些角色都不是最初就固定在内存里的，而是通过“注册-发现-使用”机制按需加载。
另外，人工审批节点非常值钱。不是所有环节都适合全自动。比如在对外发布的文案、资金操作、系统变更等场景，你可以在某个状态节点上插入awaiting_human状态，让真实的人在关键节点踩一脚刹车。
5.3 安全合规方面的注意事项
多智能体系统意味着自动化程度更高，但也意味着出错的波及面更广。在合规层面，我强烈建议把以下约束写进系统设计：
- 所有训练数据和业务数据都必须经过授权，Agent 不能擅自访问或外传。
- 对用户输入要做注入攻击检测，防止恶意 prompt 绕过角色约束，诱导 Agent 执行高危操作。
- 重要的输出结果要有审计记录，保证可以追溯“这个结论是怎么来的”。
- 内容输出前，最好经过一次“违规内容过滤”，避免把模型幻觉直接暴露给用户。
这里我不展开讲具体某种技术方案，只说一点：不要在项目里把“违规内容检测”和“内容生成”混在一个 Agent 里。让生成 Agent 管生成，让检测 Agent 管检测，责任分离，效果会好得多。
5.4 最后的实操建议：从小而精开始
如果你看完这篇想动手实践，我给你的最大建议是：不要一上来就做“全公司智能化改造”。先选一个极窄的场景，用「agency-agents」的思路搭一个 3 个以内的 Agent 小机构，把状态机、消息机制、质检闭环全部跑通。等框架稳定了、你还想说再扩展，再慢慢加角色、加工具。
我个人在实际操作中的体会是，多智能体系统不怕角色少，就怕流程乱。只要你有清晰的状态流转、明确的消息格式、强制的质检节点，哪怕只有两个 Agent 协同，也能产生实实在在的业务价值。反过来，角色再多、能力再强，没有这套“机构化”的组织约束，也只是一堆各说各话的聊天机器人而已。
「agency-agents」这个项目给了一个很好的起点，但不要把它当成现成的库用。把它当成一种思维模型：每个 Agent 不是孤立的助手，而是一家“虚拟代理机构”里的员工，需要有岗位职责、有工作流、有质检标准、有审计记录。你能把这个模型想通透，那比抄多少行代码都有用。
