做项目最怕的不是需求复杂而是把一堆逻辑揉进同一个Agent里最后哪个环节都做不精。去年我接了一个自动化营销方案生成的小需求试过用单条超长Prompt让模型一口气产出完整的推广计划结果来回调了十几版要么前期洞察太浅要么文案和渠道策略对不上。后来我换了个思路干脆仿照小型代理机构的分工搭了一个由多个Agent组成的工作流系统每个Agent只负责一个固定角色互相传递半成品、互相提反馈。这个项目就是agency-agents。这个项目本质上是一套多智能体协作框架核心思路是把“一个全能的Agent”换成“一支分工明确的Agent团队”让它们像真实公司里的小组一样协作完成复杂任务。最典型的落地方式就是模拟一个数字代理机构有项目经理拆解需求有分析师做市场调研有文案写内容有设计师出画面描述再由质检Agent统一把关。整套系统跑下来输出质量明显比单模型硬刚高一个档次。如果你正在做Agent应用或者被“长链路任务”搞得头疼这篇文章应该能给你一个可落地的参考。1. 项目概述与方案选型1.1 项目定位把一个复杂任务拆成一支虚拟团队agency-agents的核心目标是把传统意义上需要多人协作才能完成的业务流程改造成由多个AI Agent分角色执行的自动化流水线。我一开始也不知道这个做法效果如何毕竟单模型直接生成内容已经能应付不少场景了。但真正上手之后发现单模型处理复杂任务有一个很尴尬的问题你让它在同一个上下文里既做行业分析、又写广告文案、还要设计视觉创意它很容易把角色搞混。前期洞察不够深后面文案就像空中楼阁上下文太长前面的信息到后面也记不清了。agency-agents的思路恰好相反。它把每个能力点拆成独立Agent每个Agent只回答自己领域内的问题通过任务编排和消息传递把结果串起来。这样做的好处是每个Agent的上下文保持精简不会被无关信息干扰每个环节可以单独调优文案写得不好不用动分析模块天然支持并行处理比如市场调研和用户画像可以同时跑效果可以量化质检哪一步不合格就单独重跑哪一步这个项目最适合两类人参考一类是在企业里做内部流程自动化的开发者另一类是正在探索复杂AI应用的原型验证团队。不需要分布式系统底子只要懂Python、会基本的异步思想就能把这套框架搭起来。1.2 为什么不用单个大模型硬刚很多人会问既然大模型已经这么强了为什么不一次性把任务描述清楚让模型自己完成这个想法没错但实际体验下来瓶颈很清晰。第一上下文窗口是有限资源。一个完整的营销方案包含行业背景、竞品分析、核心卖点、目标用户、渠道策略、文案素材、视觉建议。把这些内容全部塞进一条Prompt里输入规模轻松超过几万字。大多数模型在长上下文末端都会出现注意力分散关键指令被稀释输出质量明显下降。第二不同阶段对推理温度的要求不一样。前期做分析时希望模型保守、谨慎输出尽量稳定后期写文案时又希望模型有一点发散性创意更足。单个模型只能有一组温度参数没办法兼顾两种模式。第三缺少阶段性把关。营销方案这种任务中间环节一旦跑偏最后输出几乎没法补救。如果没有质检Agent在中间踩刹车等生成完全部内容再让人工检查返工成本就非常高了。用一个更直观的类比一个全能员工确实能把方案写完但效率和质量大概率不如一个五人小组。项目总监拆任务、研究员供数据、文案做表达、设计师管视觉、质检查漏补缺每一环都有明确的验收标准。agency-agents就是把这种管理思路搬到了代码里。1.3 技术选型与架构概览在技术栈上我选择了Python作为主语言原因很实际Python在AI生态里的支持最成熟不管是接模型接口还是后续做向量检索都能很快上手。整个系统的架构分四个核心组件Agent层一堆角色明确的智能体每个Agent有自己的系统提示词和执行逻辑。注册中心负责维护“角色名”和“Agent实例”的对应关系。项目管理者通过角色名调用不需要关心具体实例。编排器负责DAG任务调度决定哪个Agent先跑、哪些任务可以并行、结果如何汇总。消息与状态存储负责在Agent之间传递任务详情、反馈和中间结果。我用进程内消息队列做Agent之间的通信等量级上来了再替换成Redis Streams。这样设计的原因很简单初期开发调试方便打印日志就能看到消息流转不需要额外起中间件。框架跑通之后再换分布式消息组件改动面主要集中在消息层业务逻辑不用动。2. 核心模块与角色分工设计2.1 团队角色设计每个Agent只干一件事agency-agents里最值得花心思的地方不是代码而是角色设计。角色少了没法覆盖完整流程角色多了协作成本直线上升。我第一版的Demo里设了五个角色刚好覆盖一条完整的内容生产链路。角色核心职责输入输出项目经理Agent接收原始需求拆解任务生成执行计划一段模糊的任务描述任务清单和执行顺序市场分析师Agent收集行业信息总结市场趋势和用户痛点任务说明、关键词洞察报告文案Agent基于洞察产出营销文案洞察报告、风格要求文案初稿视觉设计Agent根据文案和调性生成画面描述文案、品牌调性说明画面设计稿描述质检Agent检查各阶段产出输出修改意见任意阶段结果通过或打回意见项目经理Agent是整个团队的总调度。它不直接产出内容而是把需求拆成“要做什么、需要哪些角色、先后顺序是什么”。比如用户说“我们要做一款新咖啡的上市推广”项目经理Agent会拆出市场分析、目标人群定义、渠道策略、文案创作、视觉建议这几个环节并标记好依赖关系。市场分析师Agent负责给后续环节喂弹药。这个Agent的系统提示词格外强调“基于数据说话”要求输出里必须包含具体的用户场景和痛点描述不能只是泛泛的“市场潜力巨大”这类空话。我把它的temperature设得比较低目的是减少幻觉。文案Agent的temperature会稍微调高一点因为写文案需要一点随机性。但质检Agent会在它后面兜底所以不用担心发散过度。视觉设计Agent更特别一点它不直接生成图片而是输出结构化的画面描述方便后续再接文生图模型。质检Agent是中途加进去的。最初我想省掉这个角色结果好几轮测试里文案Agent输出文风偏离品牌调性或者分析报告里出现前后矛盾的数据让人工返工很痛苦。加了一个质检Agent之后效果立刻不一样了。它会根据预设的检查项逐条打分分数不达标就生成反馈打回给上游Agent修改。2.2 任务编排模型顺序执行之外还要支持并行角色确定之后最难的是任务编排。第一个版本我用的是最简单的顺序执行项目经理拆完任务之后一个个调用Agent前一个跑完再跑下一个。这个方法在Demo阶段够用但很快就发现效率问题。比如“市场分析”和“用户画像调研”这两件事互相不依赖完全可以同时跑。顺序执行的话两块任务加起来可能要一分多钟用户等得很烦躁。所以后来我把编排逻辑改成了DAG模型。每个任务都有两个关键属性需要哪个角色执行依赖哪些前置任务。编排器每次先找出所有“前置任务已经完成”的节点扔进线程池里并行执行。一个环节跑完再解锁下一批节点。举个例子一次完整的营销方案任务DAG看起来是这样的项目经理Agent拆解需求输出任务列表市场分析任务和用户调研任务并行执行都依赖步骤1文案创作任务依赖步骤2的分析结果等两份报告都出来才能开跑视觉设计任务依赖步骤3的文案定稿质检任务同时检查步骤3和步骤4的产出这种设计有一个好处不会出现“下游任务空转过早”的问题。只有所有依赖都满足任务才会进入执行队列。要扩展新业务场景时我只需要调整一下任务依赖关系不需要重写编排器。实现DAG调度的时候有一个细节容易踩坑任务状态管理。我维护了一个任务状态表每个节点都有pending、running、completed、failed四种状态。一个节点执行失败时我不会整个流程直接停止而是先记录失败原因然后让编排器根据重试策略决定是重新执行还是标记整体失败。原因是很多失败其实是一过性的比如模型接口超时重试就能解决。2.3 通信协议与上下文传递Agent之间怎么传话是整个系统能否协作起来的关键。我的设计很简单但很实用统一消息格式所有Agent只认四种消息。TaskAssign上层下发的任务包含任务ID、任务说明上下文、期望产出格式。TaskResultAgent执行完毕返回结构化结果。ReviewFeedback质检结果包含通过标记和修改意见。RetryRequest上游Agent收到反馈后请求重新执行原任务。消息格式长这样{ message_id: a8f9c221-3b7e-4a2d-9c1e-77f1a5b6d3f2, type: TaskAssign, task_id: task-2025-001, from_agent: orchestrator, to_agent: market_analyst, payload: { keyword: 即饮咖啡, required_sections: [市场趋势, 用户痛点, 竞品分析], deadline: 2025-03-01T12:00:00 } }这里有个容易被忽视的点上下文传递。Agent之间本身不共享上下文每个Agent都是独立的对话会话。那分析阶段的洞察结果怎么传给下游的文案Agent我的做法是维护一个全局共享状态区。每个任务节点执行完毕后编排器会把它的输出写进状态区命名方式是“任务ID 产出类型”。下游Agent启动时编排器从状态区取出所需的前置结果拼进它的Prompt里。这种设计实际上是把Agent之间的同步从“模型对话内完成”移到了“代码层完成”。好处是可追踪、可回滚每个阶段的输入输出都能查日志。3. 实操过程与核心代码实现3.1 搭建工程骨架我从一个干净的Python项目开始目录结构如下agency-agents/ ├── agents/ │ ├── __init__.py │ ├── base.py │ ├── project_manager.py │ ├── market_analyst.py │ ├── copywriter.py │ ├── visual_designer.py │ └── critic.py ├── core/ │ ├── __init__.py │ ├── message.py │ ├── registry.py │ └── orchestrator.py ├── config.yaml ├── run_demo.py └── requirements.txt依赖包不多核心只要pydantic、pyyaml和openai兼容SDK。pydantic用来做消息结构校验pyyaml用来读配置文件。pip install pydantic pyyaml openai这里提醒一句openai这个库实际上可以兼容很多模型服务只要对方提供OpenAI格式的接口改一下base_url就能用。我不建议自己封装HTTP请求浪费时间还容易在鉴权上踩坑。3.2 实现Agent基类与注册中心所有角色的共同逻辑都抽到BaseAgent里。每个Agent都维护一个消息收件箱循环接收任务执行完毕后返回结构化结果。from typing import Any, Dict import queue class BaseAgent: 所有Agent的基类负责消息接收和通用交互逻辑 def __init__(self, name: str, role: str, model: str chat-model-v1): self.name name self.role role self.model model self.inbox queue.Queue() self.running False def receive(self, message: Dict[str, Any]) - None: self.inbox.put(message) def run(self) - None: self.running True while self.running: message self.inbox.get() if message[type] TaskAssign: result self.execute(message[payload]) self.send_result(message[task_id], result) elif message[type] ReviewFeedback: self.handle_feedback(message[payload]) def execute(self, payload: Dict[str, Any]) - Dict[str, Any]: raise NotImplementedError def handle_feedback(self, feedback: Dict[str, Any]) - None: # 默认实现记录反馈不做其他操作 print(f[{self.name}] 收到反馈: {feedback}) def send_result(self, task_id: str, payload: Dict[str, Any]) - None: from core.registry import get_orchestrator get_orchestrator().collect_result(task_id, self.name, payload)注册中心更简单它就是一个服务定位器。初始化时把所有Agent实例注册进去之后编排器只要通过角色名就能拿到对应的Agent。class Registry: def __init__(self): self._agents: Dict[str, BaseAgent] {} self._roles: Dict[str, str] {} def register(self, agent: BaseAgent) - None: self._agents[agent.name] agent self._roles[agent.role] agent.name print(f[Registry] 注册Agent: {agent.name} (角色: {agent.role})) def get_by_role(self, role: str) - BaseAgent: agent_name self._roles.get(role) if not agent_name: raise KeyError(f角色 {role} 未注册) return self._agents[agent_name]为什么加这一层直接持有Agent实例的引用也能跑但一旦角色数量变多代码就会变得难以维护。有了注册中心我加一个新角色只需要改配置文件不用动编排器代码。这个习惯帮我省了很多事。3.3 实现DAG编排器编排器是整个框架最核心的部分。它负责接收原始任务、生成任务图、调度执行节点、汇总所有输出。from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Any, Dict, List from pydantic import BaseModel class TaskNode(BaseModel): task_id: str agent_role: str depends_on: List[str] [] status: str pending result: Dict[str, Any] None class Orchestrator: def __init__(self, registry: Registry): self.registry registry self.tasks: Dict[str, TaskNode] {} self.state_store: Dict[str, Any] {} self.max_workers 2 def build_task_graph(self, plan: Dict[str, Any]) - None: for task in plan[tasks]: node TaskNode(**task) self.tasks[node.task_id] node def run_dag(self) - Dict[str, Any]: with ThreadPoolExecutor(max_workersself.max_workers) as executor: while not self._all_completed(): ready_tasks self._get_ready_tasks() futures {} for task in ready_tasks: task.status running agent self.registry.get_by_role(task.agent_role) future executor.submit( agent.execute, self._build_agent_payload(task) ) futures[future] task.task_id for future in as_completed(futures): task_id futures[future] task self.tasks[task_id] try: result future.result() task.result result task.status completed self.state_store[task_id] result except Exception as exc: print(f任务 {task_id} 失败: {exc}) task.status failed return { status: success, outputs: self.state_store }这个实现里_get_ready_tasks的作用是找出所有依赖已完成、状态还是pending的节点_build_agent_payload会从state_store里把该节点依赖的上游输出全部取出拼成Prompt上下文。这两个方法不复杂但决定了整个任务流的走向。一个很实用的细节我用state_store而不是让Agent自己传递信息。所有中间结果都以任务ID为key存起来这样相当于给整个执行过程拍了快照。一旦某个下游Agent发挥失常我可以直接从失败的节点重跑不用把前面的流程全部重来一遍。3.4 配置参数与模型接入配置我用的是YAML文件所有经常调整的参数都放到这里尽量少改代码。model: default: chat-model-v1 temperature: 0.4 max_tokens: 1500 timeout_seconds: 30 orchestrator: max_workers: 2 max_retries: 2 retry_interval_seconds: 3 agents: project_manager: model: chat-model-v1 temperature: 0.2 market_analyst: model: chat-model-v1 temperature: 0.2 copywriter: model: chat-model-v1 temperature: 0.7 visual_designer: model: chat-model-v1 temperature: 0.3 critic: model: chat-model-v1 temperature: 0.2每个参数的选择都不是拍脑袋定的。temperature是第一个要调整的参数。项目经理、分析师和质检员这类角色输出需要稳定、可靠所以都设到0.2。文案角色是最需要创作力的设到0.7让它有一点发散空间。视觉设计师我压到0.3因为它在文案定稿的基础上做视觉描述发散过猛反而容易跑题。max_tokens设到1500是因为每个Agent只输出自己负责的那一段内容不太可能出现超长输出。如果某个角色要输出大段报告可以单独给它配置更高的值。max_retries控制在2次。第一次失败可能是网络抖动第二次失败大概率是配置或者模型接口本身出问题再重试意义不大。我也加了一个简单的心跳检查每次任务执行前先测一下模型接口是否可用避免把时间浪费在无意义的重试上。3.5 跑通一个营销方案生成Demo项目搭好以后我用“即饮咖啡新品上市推广”这个需求做了第一轮完整测试。run_demo.py里只写了两件事初始化注册中心把五个Agent都注册进去然后调用编排器执行预先定义的任务图。执行过程中日志大概长这样[Registry] 注册Agent: pm (角色: project_manager) [Registry] 注册Agent: analyst (角色: market_analyst) [Registry] 注册Agent: writer (角色: copywriter) [Registry] 注册Agent: designer (角色: visual_designer) [Registry] 注册Agent: critic (角色: critic) [Orchestrator] 任务 pm-001 开始执行 [PM] 需求拆解完毕共拆出6个任务节点 [Orchestrator] 并发执行: analyst-analysis, analyst-persona [Analyst] 市场分析完成输出3532字洞察报告 [Analyst] 用户画像完成输出3类核心人群特征 [Orchestrator] 任务 writer-draft 开始执行 [Copywriter] 文案初稿完成输出覆盖3个投放场景的文案 [Orchestrator] 任务 designer-visual 开始执行 [Designer] 视觉描述完成输出3套画面脚本 [Orchestrator] 任务 critic-final 开始执行 [Critic] 初审未通过文案缺乏行动号召画面描述缺少品牌露出 [Orchestrator] 打回重做: writer-draft [Copywriter] 根据反馈完成修改 [Critic] 终审通过 [Orchestrator] 全部任务完成输出汇总完毕这次跑通让我对整套系统的信心提升了很多。质检Agent打回文案之后重跑的反馈也验证了一个关键设计由于状态存储里保存了分析师产出的洞察报告文案Agent在二次执行时不需要重新请求上游而是直接把“修改意见 原文案 原始洞察”作为上下文重跑成本非常低。4. 常见问题与排查技巧实录4.1 上下文被冲掉下游Agent丢失关键信息第一版跑通后我随机抽了几次生成结果发现一个高频问题文案Agent虽然配置了分析师的洞察报告但生成的文案里经常抓不住用户痛点显得非常“套话”。排查时发现问题出在上下文拼装顺序上。_build_agent_payload把所有上游结果一股脑拼在同一段Prompt里包括结构化的用户画像、冗长的市场报告。文案模型要处理的文本太长真正关键的痛点信息被埋在了大量背景描述中。解决方法有两个。第一对上游输出做字段级提取只把关键结论传给下游。比如市场分析报告里专门设定一个key_takeaways字段文案Agent只接收这个字段的内容。第二划分Prompt段落把“核心任务”放在最前面背景资料放在后面并明确告诉模型“以下是参考资料请优先依据开头的任务要求进行创作”。这个坑特别典型我要是不加限制地让Agent传原始报告系统看起来能跑但实际输出质量就像在靠运气。4.2 任务死锁与无限重试我在另一个子任务里试过让两个Agent互相依赖比如“文案参考视觉设计的方向视觉设计又依赖文案定稿”。这种循环依赖在DAG里直接造成了死锁编排器一直在等待状态既不推进也不报错整个任务卡死在那里。更隐蔽的是间接死锁。任务A等待任务B任务B等待任务C任务C又等待任务A。三个节点都没有直接环状依赖但组合起来就绕不出来了。我的解决方案是在编排器里加了两道保险。第一构建任务图时做一次环检测发现任何环直接抛异常。第二执行时设置全局超时时间超过设定时间仍没有新节点进入completed状态就强制中止并把现状快照打印出来。这两道保险加完类似问题基本能第一时间暴露。还有一个关联问题重试风暴。当某个依赖模型接口的任务连续失败时编排器如果按固定间隔不断重试会把所有Agent实例都占满其他任务反而得不到执行。我把重试改成指数退避第一次失败等3秒第二次失败等9秒超过两次就不再重试而是把任务标记为failed并通知人工介入。4.3 模型输出格式不稳定结构化输出是我一开始就没太重视的环节。项目经理Agent返回任务清单时我要求它输出JSON数组但大模型偶尔会在JSON前面加一段礼貌回复比如“好的以下是你需要的任务清单”结果json.loads直接抛异常。这个问题一度让我特别崩溃因为格式错误导致整条任务链中断。后来我改用两条策略一起解决。第一在系统提示词里明确输出格式并给一个示例。大模型对样例比描述敏感得多给它看一个完整的小JSON例子输出格式稳定率能到90%以上。第二增加一个容错解析层。如果首次json.loads失败就用正则把JSON片段提取出来再去解析。如果还失败就启用一个“二次修正”调用把原始输出和报错信息一起丢给模型让它自己重新生成一份干净JSON。容错层让我在真实生产环境里少了很多半夜被叫醒的时刻。4.4 成本和性能的平衡多Agent协作比单模型调用要消耗更多Token这是必然的。每个角色的Prompt里都要携带上游输出跑一遍完整任务可能有2到3万的输入消耗。如果每次调用都用大模型成本会很难看。我的做法是给角色配置模型分层。质检Agent承担的任务是“拿检查项核对文本”用轻量模型就能做不需要上最好的模型。文案这种生成类任务稍微提升一个档位。项目经理的任务涉及逻辑拆解也推荐用能力更强的模型。并行度方面我把max_workers设为CPU核心数的一半。并行跑得太快模型接口的限流很容易被触发跑得太慢用户体验又会受影响。以2到3个并发执行最稳定这也是实测下来比较折中的值。5. 实用经验与后续扩展5.1 让配置驱动角色组合现在加一个新角色我只需要写一个配置文件和一个Agent类。比如要加一个“SEO优化Agent”配置里加上角色名、依赖关系、模型参数注册中心在启动时自动扫描并注册编排器不用改一行代码。这种设计对后期扩展特别重要。需求方今天说要加一个“舆情监测Agent”明天可能要加“视频脚本Agent”如果每次都要手改调度逻辑项目很快会失控。配置文件驱动加上模块化实现基本上能做到“加角色不改框架”。5.2 从营销方案扩展到代码审查和工单处理跑通Demo之后我发现这套框架根本不只适用于营销内容。换一套角色和任务依赖就能变成另一种业务系统。代码审查是一个很自然的扩展方向。项目经理Agent变成“代码库变更分析Agent”把一次Pull Request拆成风险检查、性能评估、安全审查、逻辑完整性四块四个Agent并行跑最后由汇总Agent输出评审报告。相比让一个Agent一次性审完分角色审查能显著减少遗漏。客服工单处理也适合这套模式。一个“分类Agent”判断工单类型一个“知识库检索Agent”从文档里搜答案一个“兜底Agent”准备升级路径。三个Agent协作能把原本需要人工分发的流程自动化掉一半。我个人的经验是任何“需要多视角审视、前后步骤有明显依赖”的任务都值得尝试用多Agent拆解。单Agent更像是问答工具多Agent才像业务流程系统。5.3 踩过几次坑之后的心得要我说搭建agency-agents本身不复杂复杂的是如何界定Agent的边界和职责。第一每个Agent必须只做“一个”专业动作。把范围划得过宽又回到了单Agent的老路输出质量照样会塌。宁可多拆一个Agent也不要让一个Agent做两件事。第二日志必须带上任务追踪ID。多Agent系统里一个问题可能横跨好几个角色。如果日志里没有统一的trace_id出了问题根本没办法定位是哪一步产生的脏数据。我后来给所有消息都加了trace_id排查效率提升了一大个台阶。第三保留一条人工兜底路径。Agent系统再稳定也会有预期之外的bad case。我在编排器里增加了一个“人工介入”接口。当重试次数耗尽或质检连续两次不通过时系统会自动生成一条带有完整上下文的“工单”转给人工处理。这个兜底机制保证了系统在绝大多数情况下不会死循环也不会把用户带偏。最后控制成本的一个实用技巧给每个Agent单独配置context大小而不是统一使用一个全局值。像质检Agent只接收关键字段context可以压到几百字市场分析师要泡在大量资料里context需要给大一些。按角色划分context能省下一笔可观的Token费用。到目前为止我把agency-agents从一个营销方案生成Demo扩展成了一个小型任务编排框架。后面我还计划给它接上向量库让Agent可以做更复杂的检索增强生成。但核心设计不会变多个角色各管一段、互相传递上下文、DAG调度统一协调。这套思路在你自己的业务里大概率也能复用。