这篇流水账一样的标题绝对不是让你去找一家中介公司也不是让你训练一群简历投手。拆开来看agency在软件工程体系里通常指代理机构或者代理层agents则是目前最火的AI智能体。合在一起它指向的是一个非常前沿且实战价值极高的架构方向——多智能体协作系统Multi-Agent System。用大白话说就是让多个各司其职的AI代理组成一个团队互相配合、分工协作去完成一个单独的大模型根本搞不定的复杂任务。我花了两周时间搭了一套名为 agency-agents 的模拟系统专门用于内容生产流水线踩了不少坑也总结了一些实操经验。这篇博文就把整个从0到1的过程、核心设计思路、代码细节和避坑指南一次性说透。1. 先把多智能体这件事想清楚1.1 为什么需要多个Agent而不是一个万能Agent你可能会有疑问现在的GPT、Claude这些大模型已经这么强了给它一段复杂的任务提示词它也能完成个七七八八为什么非要多Agent架构关键在于上下文窗口限制和任务耦合度。单个大模型在处理超长任务时比如生成一篇2万字的行业研究报告如果全靠一次对话上下文前半段写得好好的等到写后半段时前文的很多关键数据已经超出了它的记忆范围上下文窗口它就开始编造数据、重复内容或者逻辑前后冲突。你再怎么调提示词也解决不了token容量瓶颈。而多Agent架构把大而全拆成小而美。它的核心逻辑类似于人类公司的运作老板编排Agent定方向、拆任务老张数据分析Agent只负责查数据、算指标老李撰稿Agent只负责把数据顺畅地组织成文章。每位Agent的上下文窗口只需要聚焦自己的专业范围不再被无关信息干扰。这样既不浪费token又能保证每个环节的产出质量都是该角色能力范围内的最优解。1.2 这个项目到底在解决什么问题agency-agents的核心目标只有一个——解放任务编排的手动劳动力。以前我们写复杂报告需要先在对话框里跟AI来回拉扯先让AI列大纲再复制大纲去让它写第一章写完再转给另一个提示词去写第二章全程人工搬运上下文。这套系统把这些重复性的搬运工工作自动化了。它的运行机制如同一条流水线入口接收一个原始需求比如写一篇关于新能源汽车电池市场的分析报告。运维Agent自动清洗需求提取关键要素地区、时间范围、字数、格式。研究Agent调用搜索工具或数据库接口搜集素材并整理成结构化的facts列表。写作Agent依据结构化facts列表分段生成报告正文。审核Agent通读全文检查数据引用冲突、逻辑断点并返回修改意见。如果审核不通过系统自动将意见反馈给写作Agent进行修订循环最多3次。最终成品输出为指定格式文件。整个过程不需要人工干预内部流转你只需要在入口给个任务、在出口收成品剩下的脏活累活交给Agent网络去内部沟通。1.3 适合谁来参考这套方案如果你属于以下任一类人群这篇内容会有实际帮助。独立开发者 / 技术爱好者已经跑通了单Agent调用想试试更复杂、更接近生产环境的系统设计。自媒体运营者需要批量化产出系列文章、短视频脚本但是又不满足于纯粹的单次Prompt生成质量想要一套可控的质控流程。企业内部自动化推进者面临多个业务系统数据流转与自动生成报告的需求多Agent架构是比硬编码脚本更灵活、更容易适应需求变化的方案。这部分的技术门槛不低至少你得会Python基础语法并且能看懂异步代码。但如果只是想来验证架构思路哪怕不写代码光看设计逻辑也会很有启发。2. 核心架构拆解如何设计Agent的网络关系设计多Agent系统最关键的20%工作决定了整套系统80%的成败。在编码之前一定要把协作拓扑结构、通信协议、错误处理机制想明白。我用公司类比来拆解这三大块。2.1 三种主流的协作拓扑结构第一种中心化编排拓扑Orchestrator-Worker这种结构最直观也好理解。一个总控Agent老板接收任务然后把任务碎片分发给不同的技能Agent员工。员工干完活把结果交还给老板老板自己负责汇总、决策、下一轮部署。优点控制流清晰全局状态都在老板手里不容易出现任务漂移。缺点老板Agent容易成为性能瓶颈和单点故障。所有人的prompt都特别长消耗大。第二种管道式拓扑Pipeline把任务固定切成几个阶段阶段按照顺序执行像工厂生产线。文本策划Agent完成大纲然后它输出的内容直接作为输入传给写作Agent写作Agent结果传给编辑Agent。优点血统纯净职责单一特别适合固定流程文案→配图→排版。缺点不适应复杂任务来回修改的需求编辑器Agent改一遍后它没法回头找研究Agent要新数据。第三种协同式拓扑Peer-to-Peer每个Agent都是自由的节点通过消息总线互相通信谁需要数据就向全网广播请求。这种结构具备最强鲁棒性但是实现成本极高且容易陷入消息风暴。提示对于99%的应用场景我都推荐前两种尤其是中心化编排管道式执行的混合体。这套思路既保留了老板的决策权又让阶段性任务可以流水线化流转实际调试和维护成本最低。2.2 关键设计原则分工、反馈、止损在设计Agent系统时有三个容易被忽略的原则同时踩过坑后才真正常挂在心上。原则1Agent消灭个性保留专业这一点很反直觉。设计Agent时不要给它们设置要幽默一点或者要网络感强一点这种人设。这会导致不可复现的输出。Agent需要的是专业字典描述Domain-Specific你需要告诉它你负责使用Pandas处理Excel文件的列名修正不得修改其他数据而不是你是Excel专家。角色边界越清晰系统整体越稳定。原则2反馈闭环必须带状态标记当审核Agent审核文章时它不能只说写得不好重写。它必须输出结构化反馈哪个章节、什么类型的错误数据错误/逻辑错误/语气错误、修改建议。写作Agent拿到结构化Feedback后才能精准定位到代码里的对应模块去处理而不至于整套上下文重新生成。这是决定你能不能让系统稳定跑第二次的关键。原则3异常止损要落在代码层而不是模型层大模型是不可控的你靠提示词请求不要跑题在长链任务里就是扯淡。必须通过代码逻辑来硬限制比如设定某环节最大执行次数、限制每轮生成的Token最大长度、当审核连续两次不通过时强制跳转人为介入。这些安全阀要写死在架构底层。后面第4章我会专门讲止损实现。2.3 通信协议怎么定多个Agent之间交换数据需要结构化建议不要传大段的自然语言文本否则后续Agent解析时会有100种方式抓狂。我在模拟项目X里定义的传输格式是JSON模式如下{ section: market_analysis, data_type: statistical_data, content: {total_market_size: 1200000, unit: units}, feedback_required: false }注意该格式规范了三件事数据归属section明确这是哪一部分的数据。数据类型data_type标明这块内容是数据、文本还是结构化对象。是否必须要审核feedback_required给后续编排模块的判断依据。我强烈建议对Agent之间的通信数据结构用Pydantic或者数据类去继承强约束模板。这看起来增加了前期工作量但实际上能省掉后面掉进数据歧义泥潭的时间。3. 实操过程从零搭建一个可用的多Agent编排系统思路理清了下面进入实操阶段。我在设计这套系统的时候技术栈选的是Python 异步编程asyncio模型调用用LangChain库进行底座封装其实直接用OpenAI SDK也行看个人习惯。核心要把编排逻辑、Agent基类和运行演示跑通。3.1 环境选型与依赖安装工作流管理这块基于常见的实践我用的是LangGraph框架它对于有向图的状态管理能力非常契合多Agent任务流。下面的安装命令使用uv工具来快速构建环境做完这些你就可以继续看后面的代码了。# 使用uv工具快速构建虚拟环境比直接用pip快很多 uv venv .agency_env --python 3.12 source .agency_env/bin/activate # 安装核心依赖 uv pip install langgraph langchain langchain-openai pandas python-dotenv # 检查环境 python --version3.2 构建Agent基类与状态管理万事开头难首要是构造一个Agent基类。所有角色Agent都继承它相当于一个标准工作证。它负责定义每个Agent的输入输出协议一致性以及调用LLM模型的通用方法。from abc import ABC, abstractmethod from typing import Any, TypedDict from langchain_openai import ChatOpenAI import os class AgentState(TypedDict): 多Agent间的公共状态表 task_description: str raw_materials: dict draft_content: str review_feedback: list final_output: str class BaseAgent(ABC): def __init__(self, name: str, system_prompt: str, model_name: str gpt-4o-mini): self.name name self.system_prompt system_prompt self.llm ChatOpenAI(modelmodel_name, temperature0.3) abstractmethod def process(self, state: AgentState) - AgentState: 真正的业务处理逻辑由子类覆盖 pass有了基类和状态定义还不够这里还需要一个关键环节——状态管理。多Agent能协同工作必须依赖一张可以反复读写打补丁的公共账本。全局变量在异步任务里很容易混乱我的解决方法是把状态直接塞进LangGraph的StateGraph里管理方便任意节点读写。给一个简化的伪代码示意from langgraph.graph import StateGraph, END class Orchestrator: 总控编排器负责把各个Agent串起来 def __init__(self, agents: dict): self.graph StateGraph(AgentState) self.agents agents self.graph.add_node(clarify, agents[clarifier].process) self.graph.add_node(research, agents[researcher].process) self.graph.add_node(write, agents[writer].process) self.graph.add_node(review, agents[reviewer].process) self.graph.set_entry_point(clarify) self.graph.add_edge(clarify, research) self.graph.add_edge(research, write) self.graph.add_conditional_edges(write, self.should_rewrite, {accept: END, revise: write}) def should_rewrite(self, state: AgentState): 如果不合格且修改次数在阈值内就给回写节点否则强制退出 feedback_list state[review_feedback] if len(feedback_list) 3: return revise return accept这块是整套系统的心脏后面接线时一定要特别注意conditional_edges的判断函数。它在每次写完内容后触发决定是收工还是回炉。3.3 典型Agent角色的Prompt细节不同的Agent对Prompt的精细度要求完全不同。这里我整理出两类关键Agent的提示词设计对比直接照抄就行。角色1研究员AgentResearcher这个Agent的核心职责是把搜索材料转化为结构化事实清单。我的Prompt设计会把职责边界写得很死你是一名严谨的数据调查员。你的任务仅针对用户提出的调研问题从给到的原始文档中提取相关数据。 要求 1. 每一条数据必须有明确出处原文段落编号。 2. 不允许对数据进行任何推论和联想只做摘录。 3. 数据格式规范为JSON数组字段为{claim, source, page}。 4. 禁止输出任何总结性文字。这样的设计逻辑在于限制大模型的生成欲知乎文体那种随着社会的不断发展一看就跑题了。至于上下文窗口拆解、压缩后只留下规则范围内的硬数据既好做逻辑保证又能省token。角色2写作AgentWriter写作Agent正好和研究员相反它需要足够多的自由度和行文引导。你是一名B2B领域的资深科技记者。你的任务根据给定的结构大纲supplied skeleton和事实列表fact list撰写完整的第一版草稿。 要求 1. 精确锁定目标读者为技术决策者专业词汇不需要解释。 2. 每段落开头使用3W原则What, So What, Now What构建逻辑闭环。 3. 文章中会穿插[[Citation:1]]标记索引号代表该处引用事实列表的第1条。 4. 如遇到事实列表不足禁止编造数据允许使用该领域暂无公开数据占位。这里严重提示一点写作过程允许AI自由发挥但涉及关键数据必须走引用标记。这是为了后端的审核Agent可以有依据地抽查事实也是防幻觉的第一道防线。角色3质检AgentReviewer质检Agent在流水线里角色像合规部。它的Task不是重写而是只读检查。它的输出必须是反馈条码列表。结构化检查项包括是否有引用标记但未完全覆盖文章所有关键事实关键数字是否与前提事实列表完全匹配上下段落是否有明显的逻辑跳跃是否出现过度承诺内容超过prompt要求如果以上检查项全部通过它输出JSON{passed: true, score: 95}如果有问题则相应输出{passed: false, issues: [Line 102: 声称某市场份额为52%与事实列表中的38%冲突, Line 210: 论证了国外市场段但缺少必要的逻辑过渡]}。3.4 打通全流程模拟任务演示我用一个具体任务来验收整套系统演示从输入到最终输出的状态变化。任务描述写一篇面向投资经理的第三代半导体材料SiC行业分析报告字数2000字左右侧重市场规模和竞争格局。状态表流转过程这是个极简的模拟clarify节点结束后状态表step1, raw_materials{target_audience: 投资经理, topic: SiC, depth: medium}research节点向模拟数据接口查询信息返回了16条facts存入状态表的raw_materials。write节点拿facts转化为一篇1892字的初稿并依据提示词规则嵌入了引用标记例如根据行业联盟统计全球导电型SiC衬底市场在2023年达到8亿美元规模[[Citation:2]]。review节点对全文仔细检查后发现第3章的引用和素材不一致打回给write节点要求修订。write节点依据Feedback的issue定位只修改了第3章那两句话并附上更精确的引用再给review节点审核…… 通过后流程结束。在系统跑动过程中我一般在终端实时打印出状态流转日志长这样 [Router] Node clarify completed in 1.2s, state size: 1k tokens [Router] Node research completed in 4.8s, fetched 16 facts [Router] Node write completed in 18.6s, drafted 1892 chars [Router] Node review returns FAILED, issue detected in Section_3: 数据矛盾 [Router] Re-writing attempt 1/3 started...这种可视化日志是事后调试时救命的存在。3.5 关键参数配置的讲究这里有几个参数配置是这套系统能不能在成本、质量和速度之间取得平衡的生死线。配置项推荐值配置理由与实操心得单次LLM温度值0.2 - 0.3多Agent任务流强调的是逻辑稳定性不是创造力。温度调太高会导致风格飘忽调太低会显得像个机器人。最大改写循环次数3次超过3次基本就说明设计有硬伤或者提示词有重大问题继续循环只会浪费钱触发兜底方案。模型选型研究员/质检员用小模型写作用大模型研究员和质检员提取信息、匹配字段用Mini级别的模型性价比极高成文环节因为对语义连贯性要求极高必须用旗舰模型。这是控制成本的最重要手段。超时时间单节点110s设定超时是防止API忙时某节点堵塞整条流水线该节点自动清理并重试一次。Batch Size单轮任务并发控制在3-5个并发太高容易batch超限太低又无法充分利用吞吐量3-5是我测试下来最稳的并发数。4. 排查与避坑多Agent系统常见的那些坑多Agent系统可以说是分布式系统的复杂性 大模型的不确定性双重考验。运行了这两周我踩了一堆坑解决思路整理成速查表直击要害。4.1 常见问题速查表症状根本原因解决方案Agent陷入无限循环对话像两个人在吵架conditional_edges没有设置硬性产量上限导致即使能判断跑了N次也不知道出口在哪在所有循环节点上游加上iteration_count检查器用代码强制截断并走兜底逻辑上下文爆掉API频繁报错某些节点的原始素材太大导致中间状态累计多个超大JSON对象每次节点读操作后立即执行字段投影只保留下一环节需要的字段。比如研究节点输出后把原文长链全部丢弃只留摘要facts改了一版结果改坏了其他章节写作Agent拿到feedback后没有遵循只改局部的原则启用全局重写模式在prompt系统层加重强调修订模式只输出有变更的段落片段并加入具体的行号指令不要让它重写全文两个Agent相互死锁互相等对方数据拓扑结构选择了全互联数据却有环状依赖简化拓扑结构所有数据流强制经过一个中心网关甚至可以把这个网关做成消息队列从根上杜绝环状依赖跑流程到一半某Agent忽然开始输出Markdown语言上一环节Agent被误植入你应该是一个工作效率顾问之类的全能人设审查所有Prompt保证每句话都在岗位说明书边界内不给大模型自由发挥职业的空间4.2 成本失控的预防机制用多Agent系统最怕的不是功能出Bug而是钱花得莫名其妙。我在这套模拟系统里加了两个成本控制闸门。闸门1Token预算挂牌。编排器在每个节点启动前会根据任务类型的字数复杂度提前估算一个max_token阈值。比如写作阶段如果生成内容超过3500 Token系统会强制终止当前任务并上报疑似生产失控。闸门2实时账单展示。在主程序运行同时开一个小携程记录每一次LLM调用的prompt tokens和completion tokens累加成实时费用面板。当某轮任务整体费用超过预设值比如每天设置的5美元警报线系统自动切断新的任务排队只保留当前正在处理的节点完成并退出。注意这个方法可能在写真实论文的场景受到挑战所以我把基本原则再强调一遍——多Agent系统的成本优化核心思想不是砍Prompt字长而是砍调用次数。能用小模型完成的绝不用大模型能合并执行的绝不拆分成两条链路。4.3 可观测性与Debug独门技巧多Agent系统里最痛苦的Debug体验就是不知道哪一步把数据搞坏了。传统打印日志在异步流转时不够用我现在的做法给关键Agent节点都挂上输入输出快照机制。def trace_agent(agent: BaseAgent): def wrapper(state: AgentState): print(f {agent.name} 接收数据:) print(f keys: {list(state.keys())}) result agent.process(state) print(f {agent.name} 输出完成状态变化: {result.get(mailbox)}) return result return wrapper快照能100%复现出Bug发生的现场比起看大模型回复我理解不了要高效太多。复现问题之后常见操作是在不同环节添加字段断言的代码比如某个环节的状态必须包含total_count字段否则直接中断。这种方式可以让数据在链条中任何一步出问题就以最快速度炸出来。5. 进阶扩展让agency-agents走向Agent生态做到这一步你的多Agent系统已经可以处理固定的流水线和报告生成任务。但如果想让它进一步适应复杂业务也就是Agent生态化还有几个方向值得打磨。5.1 让Agent学会使用工具目前例子的研究员Agent是从给定文本中提取数据但在实际应用中它必须能自己搜索互联网、调用内部API、执行数据分析代码。工具调用让我印象最深的是把工具名字用途schema参数说明注入到System Prompt里然后让Agent返回JSON整体工具请求交给代码执行器去执行。这比让大模型直接调代码安全得多资源会被拖在沙箱里。一个典型工具定义Schema如下{ tool_id: web_search, description: 检索公开行业数据。, parameters: { type: object, properties: { query: {type: string, description: 查询语句}, time_range: {type: string, enum: [1M, 3M, 12M]} } } }5.2 引入记忆模块与持久化现在的多Agent每次跑任务都是失忆的它不知道上个项目积累了什么行业背景。在跑第二个写作任务时又得从零开始摸索行业黑话。改进方案很简单设置一个向量数据库在每次任务结束时把该任务的核心术语、行业规律、成功的数据架构全部抽取成嵌入向量存起来。当下次新任务到来时编排Agent先从向量库里进行相似度检索把相关的历史经验注入到写作Agent的前置上下文里。这就是人类员工入职半年后的经验养成效果提升是量级的。5.3 如何过渡到多Agent的自举模式探索完这套系统后可以再往上想一层让一个消费用户提交的任务在经过编排器处理后自动生成一串新的可复用任务流模板DAG。也就是说系统的沉淀能力再上一步就不再是人写工作流Agent跑步了而是变成Agent发现工作流Agent自己跑革命。这个方向当前还很贵动手时建议从小规模流程试起。最后再分享一点个人体会这套多Agent系统调通原理后我又把它应用在工作里一个很小但高频的场景把每周更新的行业数据自动转成团队周报。以前每周五一下午都要用来搬运数字、校准口径现在只要把原始数据包丢给入口20分钟之后就能拿到格式标准、数据引用齐全的周报草稿。而且由于有了审核Agent的把关错漏率比我手动写还要低。最后对想入这个方向的朋友真心的建议先把单体Agent的能力边界摸清楚再来搭建多Agent系统。如果单个Agent本身连一个固定回答都做不好直接上网络编排只会在后期排查到崩溃。多Agent系统的价值不是把垃圾能力复读一遍而是在靠谱的能力单元之间建起高楼。这是你执行力最被放大的一击。