“agency-agents”这个名字，看着像是一家虚拟公司，其实是一套典型的多智能体协作框架。简单说，它解决的是单个大模型在处理复杂长链路任务时容易断档、上下文丢失、角色混乱的问题——通过引入角色分化、任务编排和跨Agent通信，把一条大任务拆解成多个专用子Agent去执行，再由一个协调层统一调度。我最近在做一个数据处理Demo时，照着这套思路搭了一个“代理事务所”，把数据爬取、清洗、分析和报告生成拆给不同的Agent去做，整体跑下来效果相当稳。这篇文章就围绕agency-agents的拆解、实现和踩坑记录展开，适合那些已经玩过单Agent、想尝试多角色协作方案的开发者参考。

1. 项目整体设计与思路拆解

1.1 为什么需要“代理事务所”而不是一个大而全的Agent

很多初学者一开始的思路是这样的：把所有工具调用、所有Prompt逻辑塞进一个Agent实例里，让它自己判断该用什么工具、该按什么顺序执行。这个方式的优点是简单，但一旦任务变得复杂，问题就开始暴露。比如我做那个数据处理Demo时，早期版本就是单个Agent负责所有事——先让它去读数据文件，再让它决定清洗规则，还要它自己写分析脚本，最后生成报告。结果经常出现两个问题：一是上下文被拉爆，前面读文件的内容把后面分析环节的注意力占满了；二是角色混乱，同一个Agent既要当“数据工程师”又要当“领域专家”，提示词稍长一点，它就开始自己改规则。

agency-agents的核心设计思路就是反着来——把“一个什么都会的Agent”拆成“一群各司其职的Agent”。每个Agent只负责一个环节，只加载自己需要的上下文，只调用自己相关的工具。而一个统一的调度端负责拆解任务、分发任务、收集结果、判断是否需要迭代。这就像现实中的咨询公司：项目经理接单，拆给分析师、设计师、工程师去干活，最后统一汇总。每个成员不需要懂其他人的全部工作，只需要对项目经理负责。

这种设计的好处不只是“看起来专业”，更重要的是它规避了大模型的两个固有短板：上下文窗口有限和单次推理的专注度有限。你让一个Agent连续处理10个环节，它后5个环节的表现一定不如前5个；但如果你让10个Agent各处理1个环节，每一个都在自己最专注的状态下干活，质量就稳定得多。这一点我在实测里感受非常明显，同样的数据源，单Agent完成的分析报告里有不少“幻觉型结论”，拆成多Agent之后，关键结论基本都能在数据里找到出处。

1.2 核心架构：调度层、执行层与记忆层

我搭建的这个demo严格来说算一个轻量版的agency-agents，但它保留了核心的三层架构：

第一层是调度层，对应的是一个TaskRouter。它接收原始任务描述，先做两件事：一是把任务拆解成多个子任务，二是决定每个子任务应该派给哪个Agent。这个层不需要太强的推理能力，关键是拆分逻辑要稳定。我是用一个单独的Prompt模板来实现的，把任务类型限定在“数据处理流水线”这个范围内，让模型输出一个JSON数组，每个元素包含task_id、task_type、dependency和params。这样调度逻辑就是纯代码处理，不依赖模型自由发挥。

第二层是执行层，也就是那批实际干活的Agent。每一类Agent其实就是一个独立的“LLM + 工具集 + 系统Prompt”的组合。数据处理Demo里我设了四个Agent：采集Agent、清洗Agent、分析Agent、报告Agent。每个Agent只能调用自己那套工具，比如采集Agent调文件读取器和网络请求器，清洗Agent只碰格式转换和缺失值处理工具，分析Agent掌握统计计算和图表生成，报告Agent则是把前面的结果组织成一个完整的Markdown报告。

第三层是记忆层，这里我把“记忆”理解成两样东西：一个是各Agent之间传递的结构化消息，另一个是全局上下文存储。Agent之间不直接聊天，而是把产出物写到共享的临时数据库里，调度层通过读取这个数据库来判断当前进度。这个设计避免了很多多Agent框架里常见的“连环对话失控”问题——Agent A跟Agent B聊high了，把任务聊偏了。

1.3 为什么选择“消息队列”而不是“直接函数调用”

在搭建过程中，我其实先尝试过更简单的方案：让Agent之间的调用就是普通的Python函数调用，采集Agent直接调用清洗Agent的清洗函数。结果发现一个致命问题——每个Agent的执行时间差异很大，采集可能只需要几秒，清洗却要跑一两分钟，一个环节卡住，后面的整个调用链就全部阻塞。

后来我改成消息队列模式，其实不复杂，就是用Python标准库的queue模块模拟了一下。调度层把任务封装成带状态的消息，放进一个任务队列，每个Agent从队列里取消息，执行完再往结果队列里放。这样做的好处很明显：Agent之间的耦合度降低了，任何一个Agent超时或者出错，都可以被调度层捕获并触发重试，而不是让整个链路崩溃。

这里我多说一句，很多人在学习agent框架时容易陷入一个误区，觉得Agent之间一定要像人与人之间那样“自然语言对话”。但在实际业务场景里，结构化消息比自然语言可靠得多。自然语言用于描述任务、传递创意没问题，但在传递数值结果、文件路径、统计指标这些精确信息时，就应该用JSON、用字典、用明确的字段名。我自己的设计方案是：Agent之间的消息体统一为{m "task_id", "status", "output", "meta"}这样的格式，任何Agent拿到消息后都能稳定解析。

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

2.1 Agent的提示词设计：角色越窄，效果越好

我在agency-agents项目里最想强调的一个实操心得是：Agent的提示词一定要窄，不要贪心。你看很多人写Agent提示词，一写就是几百字，恨不得把这个Agent能做的所有事情都写进去。这种做法在单Agent时代还能凑合，在多Agent体系里就是灾难，因为其他Agent产生的输入会有各种格式，如果这个Agent的提示词范围太宽，它就会去尝试处理那些不属于它的任务。

我拿清洗Agent举例。最初的版本提示词里写了“你可以处理所有数据质量问题”，结果它经常在拿到采集Agent的数据后，自作主张地去改写原始字段，把不该动的数据也动了。后来我把提示词改成这样：你是一个数据清洗专员，你的输入是一个JSON对象，其中包含data字段和schema字段。你只允许做三件事：识别缺失值并标记、修正数据类型不一致、删除完全重复的记录。你的输出必须是JSON格式，包含清洗后的数据、清洗規則说明和不确定项列表。其余任何操作都不归你管，如果发现输入不在你职责范围内，请直接返回error信息。

这样改完以后，它的行为稳定了非常多。核心原因就是：大模型这个“角色扮演”能力其实需要非常明确的边界，你告诉它“你是某方面的专家”还不够，你必须告诉它“你不是什么”，它才不会越界。

2.2 任务编排的状态机与容错机制

任务编排是agency-agents框架里最容易被低估的部分。很多人以为编排就是“按顺序调几个函数”，但实际业务里任务之间不一定是简单的前后依赖，还可能有并行关系。比如数据处理Demo里，清洗Agent和分析Agent可以并行跑——清洗Agent处理完第一版数据后，分析Agent就可以开始初步探索性分析，不需要等所有清洗工作都结束。

我在项目里用一个简单的状态机来实现编排，每个任务有pending、running、completed、failed四个状态。调度层维护一张任务状态表，每次迭代时扫描所有pending状态的任务，检查它们的依赖任务是否都已经completed。如果依赖都完成了，就把这个任务标记为ready，并投递到执行队列。当某个任务进入failed状态时，调度层会根据预设的策略决定是重试、跳过还是终止整条流水线。

这个状态机的实现代码其实不长，核心部分我放在下面：

class TaskStateMachine: def __init__(self): self.tasks = {} self.status = {} def register(self, task_id, deps, handler): self.tasks[task_id] = {"deps": deps, "handler": handler} self.status[task_id] = "pending" def ready_tasks(self): ready = [] for task_id, task in self.tasks.items(): if self.status[task_id] != "pending": continue if all(self.status[d] == "completed" for d in task["deps"]): ready.append(task_id) return ready def mark_running(self, task_id): self.status[task_id] = "running" def complete(self, task_id): self.status[task_id] = "completed" def fail(self, task_id, max_retries=2): task_info = self.tasks[task_id] retries = task_info.get("retries", 0) if retries < max_retries: task_info["retries"] = retries + 1 self.status[task_id] = "pending" else: self.status[task_id] = "failed"

我这里特意留了一个retries的机制。实际运行中，Agent偶尔会出现输出格式不符合预期的情况（比如返回了非JSON的纯文本），这种问题不一定是模型挂了，很可能只是它“那天心情不好”，重试一次可能就好了。但如果连续重试两次还是失败，基本可以确定是提示词或者输入数据本身的问题，这时候再重试没有意义，不如把问题暴露出来。

2.3 结果合并与全局上下文的取舍

在单Agent体系中，所有中间结果都存在同一个上下文里，随时可以引用。但在多Agent体系中，各Agent的结果是分散的，合并起来就需要策略。我在demo里采用了最简单但很有效的方式：每个Agent完成工作后，会把自己的产出物（比如清洗后的数据表、分析结论的摘要、图表文件的路径）写入一个统一的临时工作目录，并往全局存储里追加一条记录。报告Agent在最后阶段，不是去读所有原始数据，而是只读这些记录和摘要。

这里有一个很重要的取舍原则：全局上下文只放“结论性信息”，不放“过程性信息”。比如采集Agent不需要知道报告Agent的写作风格，清洗Agent不需要知道分析Agent用了哪种统计方法。每个Agent要看的，只是自己输入环节所需要的那一小块信息。这个设计和人很像——一个公司里的员工不需要知道所有部门的工作细节，只需要拿到上下游交给自己的交接文件。

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

3.1 项目初始化的配置与目录规划

我在搭建这个demo时，目录结构是这么规划的：

agency-agents/ ├─ orchestrator/ │ ├─ task_router.py # 调度与任务拆解 │ ├─ state_machine.py # 状态机实现 │ └─ memory_store.py # 全局上下文存储 ├─ agents/ │ ├─ base.py # Agent基类 │ ├─ collector.py # 采集Agent │ ├─ cleaner.py # 清洗Agent │ ├─ analyzer.py # 分析Agent │ └─ reporter.py # 报告Agent ├─ tools/ │ ├─ file_reader.py │ ├─ csv_validator.py │ └─ chart_generator.py ├─ workdir/ # 临时产出物目录 └─ config.yaml # 全局配置

这个目录规划看着简单，但有一个点比较重要：把Agent和工具拆开。Agent是“决策大脑”，工具是“手脚”。同一个工具可以被多个Agent调用，但一个Agent绝对不能直接调用另一个Agent的业务函数。这个边界用来保证职责清晰，排查问题的时候也能快速定位是哪一层出的问题。

配置项方面，我在config.yaml里定义了模型参数、每个Agent的最大输出token数、超时時間、重试次数。这里的超时設置特别关键，因为LLM接口的响应时间不稳定，一个Agent卡住，如果整个流水线没有超时控制，后面的任务都会越积越多。我的设计是每个Agent执行阶段有独立的超时时间，超时后直接把该任务标记为failed并触发重试或跳过。

3.2 核心Agent基类的实现方式

所有Agent都继承自同一个基类，基类负责统一的推理调用、结果解析和错误处理。基类的设计思路是：Agent子类只需要实现两个方法——一个是build_prompt，把输入参数转成该Agent的Prompt；另一个是parse_response，把模型输出转成结构化的消息体。其他逻辑，比如调LLM接口、超时重试、日志记录，全部在基类里完成。

class BaseAgent: def __init__(self, name, model_config, tools): self.name = name self.model = model_config self.tools = {tool.name: tool for tool in tools} def run(self, task_input): prompt = self.build_prompt(task_input) for attempt in range(self.model.max_retries + 1): try: raw_response = self._call_llm(prompt) parsed = self.parse_response(raw_response) return {"status": "success", "output": parsed} except Exception as e: if attempt == self.model.max_retries: return {"status": "failed", "error": str(e)} prompt = self._build_retry_prompt(prompt, str(e))

这里我想单独讲一下_retry_prompt的设计。第一次调用失败后，很多人的做法是把同样的话再问一遍，但效果很差。更好的做法是把错误信息反馈给模型，让它自己修正。比如清洗Agent输出的JSON少了一个右括号，模型其实自己也不知道自己之前错了，但如果你在重试时告诉它“你上次的输出不是合法JSON，请重新输出”，它往往能立刻给出正确结果。这背后的原理是：大模型在生成过程中是有一定随机性的，给它一个明确的纠错信号，比让它盲目重跑更有效。

3.3 一个完整任务的调度执行过程

我用一个具体的例子来讲调度过程。假设我提交给系统的任务是：“请处理2024年某季度销售数据，找出同比和环比的异常波动，并生成一份包含图表的报告。”

调度层收到任务后，先通过TaskRouter拆解：

[ {"task_id": "collect", "task_type": "collector", "deps": []}, {"task_id": "clean", "task_type": "cleaner", "deps": ["collect"]}, {"task_id": "analyze", "task_type": "analyzer", "deps": ["clean"]}, {"task_id": "report", "task_type": "reporter", "deps": ["analyze"]} ]

这个步骤里有一个容易忽略的细节：子任务的描述不必像原始任务那么复杂。TaskRouter在做拆解的同时，会为每个子任务生成更具体的任务参数。比如传给报告Agent的prompt，不是简单的“生成报告”，而是把分析Agent得出的异常结论列表、相关图表文件的路径、报告的受众（管理层/执行层）一并传过去。这个“子任务上下文”的构造，直接影响最终输出质量。

状态机启动后，第一步扫描发现collect任务没有依赖，直接标记为ready，投入执行队列。collectAgent执行完后，调用complete方法，状态机继续扫描发现clean任务依赖已满足，继续投递。这样一级一级推下去，直到report任务完成。

有一回我在实际跑的时候发现一个问题：分析Agent很久都没有返回结果。排查下来发现，是它的输入数据太大了，整个清洗后的表格全文塞进了prompt，导致模型上下文被塞满，响应时间暴增。解决方法是改成只传数据摘要和字段统计信息，分析Agent如果发现数据量超出阈值，会先写一个中间脚本做特征提取，再对提取后的结果做分析。这里透露一个通用原则：永远不要让Agent直接处理原始大数据，让Agent学会使用工具来切分和采样数据，才符合“Agent调用工具”的设计初衷。

3.4 结构化消息的内容与格式约定

在系统跑通了以后，我把消息体格式稳定成了下面这种：

{ "task_id": "clean_001", "agent_name": "cleaner", "status": "success", "output": { "cleaned_rows": 2345, "null_ratio": 0.02, "schema": {"date": "string", "amount": "float"} }, "meta": { "duration_ms": 3400, "model": "some-model-name", "prompt_tokens": 2100, "completion_tokens": 600 } }

如果你仔细看这个结构会发现，output字段放的是一种“半结构化数据”，既有数值又有描述性的schema。这其实是特意设计的——数值信息给后续Agent做判断时用，schema信息给报告Agent组织呈现时用。meta字段则主要留给开发者在调试时查看，不会传給下游Agent。

另一个我踩过的坑是：如果不约束Agent输出为JSON，它会输出一段带有解释性语言的Markdown文本，比如“根据分析，结果显示...”。这些自然语言对下游Agent来说很容易造成干扰。所以我的约束方法是：提示词里严格说明——只准输出JSON，不要输出任何解释文字，所有解释文字放到完成后用单独的日志通道打印。这样做初期的代价是Agent偶尔会“叛逆”，不按规则输出，但通过重试机制和默认值兜底，整体稳定性是可以接受的。

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

4.1 Agent之间的数据格式不一致

这是多Agent框架里最容易出现的问题，几乎每次改动都会踩一遍。采集Agent在处理Excel文件后，把日期字段统一成了字符串格式“2024-01-01”，但是清洗Agent内部假设的输入格式是datetime对象。两个Agent之间没有做字段校验，结果清洗Agent一运行就报错。

这个问题从根上说，是Agent之间的接口契约没有定义清楚。解决办法就是把每个Agent的输入输出格式明确写到提示词里，并且在代码层面加一个校验函数。我在项目里给每个Agent的parse_response后面都接了数据校验，如果输出里缺少必要字段，就直接判定该次任务失败并触发重试。经验数据表明，这个校验流程至少解决了40%的随机性故障，因为模型偶尔就是会漏掉字段，你给它一次“重生的机会”，它就改了。

4.2 模型输出格式随机性引发的JSON解析失败

我在多次实操中发现，无论提示词怎么写，模型偶尔就是会输出一些带注释的JSON，或者JSON里多了一个尾逗号，造成json.loads直接报错。解决方法不是在重试时无脑重新解析，而是要做一个“容错解析器”。

我的做法是先尝试json.loads，失败后用正则把常见的尾逗号、注释、多余的换行清理掉再解析一次，再不行就提取代码块包裹的内容，如果里面是JSON就解析它。这个容错解析器帮我把整个流水线的成功率从72%拉到了95%以上。但要注意，容错解析只是最后一道防线，不应该依赖它——最好的情况仍然是提示词设计得当，让模型稳定输出标准JSON。

4.3 并行执行时的资源占用与限流

当任务可以并行执行时，比如同时起采集Agent和清洗Agent，系统性能开销并不是简单的相加关系。实际运行中我发现，如果同时让4个Agent调用同一个大模型接口，接口的速率限制会频繁触发，而且模型在并发状态下的响应质量会有波动。

我的建议是给每个Agent加一个全局的并发信号量，控制同一时间最多两个Agent在跑。你可以根据自己使用的模型服务容量来调整，但总的原则是：不要贪。多Agent的关键优势在于任务解耦和上下文优化，而不在于无脑并发。真需要提升吞吐量的时候，优先考虑缓存中间结果，而不是盲目加线程。

另外，日志系统一定要带trace_id，把整个流水线从开始到结束串联起来。我一开始没有做这个，结果出了问题根本不知道是哪个环节的Agent在调哪个任务。后来在调度层分配任务的时候给每个task_id挂上一个trace_id，日志里打印所有相关链路，排查效率提升了不止一个量级。

4.4 提示词过长时的性能衰退

最后一个常见问题是，当某个Agent的提示词包含太多历史记录或太多过时的示例时，模型的性能会明显下降，而且响应时间暴增。比如报告Agent如果每次都把上一次的报告样例塞进去，前几次还好，第10次以后prompt长度几乎翻倍，响应速度却慢了一半。

我的解决办法是给每个Agent单独维护一个缓存池，只放最近一次的成功输出示例，而且示例不要放完整版，只放结构和格式骨架。比如报告Agent的提示词里放的示例，是那种“报告标题 + 摘要列表 + 图表引用占位符”的骨架，而不是一个完整的上次报告。这样既能约束格式，又不会把上下文撑爆。

5. 一些关于扩展性的想法

agency-agents这套东西，做完以后我最大的感受是：它不是一个写死的项目，而是一种可以不断往里面加Agent、加工具、加任务类型的框架。比如我后来在这个demo上额外加了一个“质量检查Agent”，它的职责是在报告生成之后，根据原始数据再校验一遍报告里的所有数字结论是否一致——不需要很复杂的逻辑，只是做数值层面的比对，发现不一致就触发报告Agent重新生成。这个想法直接模仿了“审计”的角色，效果出奇地好。

扩展的时候有一点要注意：新增Agent时，不要急于给它写复杂的业务逻辑，先把它的输入输出接口定义好，让它能跟现有Agent通信，再逐步丰富它的提示词和工具集。这个思路和做微服务是一样的——先保证通信，再保证业务。多Agent体系里最忌讳的就是一开始就把几个Agent绑死在彼此的内部实现上，一旦业务变化，改起来能让你怀疑人生。

我个人在实际操作中还有一个小技巧：每次修改Agent的提示词之后，都保留一个旧版本放在版本目录里，标注好改动原因和改动时间。因为提示词改动带来的效果变化不一定立竿见影，有时候要跑完整个流水线才能发现。如果你没有版本对比，出了问题根本不知道是“提示词改坏了”还是“模型本身抽风”。

这个方法后续还可以继续扩展的方向，包括给Agent接入外部知识库、增加Agent之间的互评机制、引入更复杂的条件分支逻辑。但无论加什么功能，核心的原则不变：职责边界清晰、消息结构稳定、重试机制兜底。这三件事做好了，多Agent系统就能从“玩具”变成“工具”。