“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_retries2): 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系统就能从“玩具”变成“工具”。 
长河君 · 2026/10/26 11:53:53 · 阅读 1.2k 从单Agent到多Agent：agency-agents框架拆解与数据处理实践
“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_retries2): 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系统就能从“玩具”变成“工具”。