小白也能玩转多Agent协作开发：300行代码搭建月度总结系统

1. 为什么小白也能玩转多Agent协作开发先说个扎心的事实单Agent能做的东西越来越窄了。早期我们玩大模型无非是写个Prompt调API让模型帮你总结文档、补全代码、算个Excel公式。可一旦任务稍微复杂点——比如“帮我做一份本周销售复盘并发给团队”单个Agent就开始拉胯它既要分析数据、又要写结论、还得生成汇报表格一个Prompt塞进去又长又乱模型精力一分散输出质量直线下降。我前两年刚接触多Agent协作时第一反应也是这玩意儿要写一堆复杂调度代码吧结果真正上手才发现本质上就是“多个模型实例各干各的活再用一个共享缓冲区接力传递结果”。底层逻辑比我预想中简单得多。很多小白程序员其实已经会调API、会写基础Python了卡住的点往往是“不知道从哪下手组织交互结构”。这篇内容我就围绕一个真实场景来拆用多个Agent协作搭建一个能自动生成月度总结的小系统。规划Agent负责拆任务、检索Agent负责查资料、写作Agent负责出文案三个角色接力最终产出一份可以直接用的总结文档。整个工程不到300行代码不需要任何分布式基础一台开发机跑通这是我验证过最适合入门多Agent开发的第一个练手项目。照我自己的经验适合看这篇内容的人包括会写基础Python、用过至少一次大模型API、想让程序不只“回答问题”而是“完成任务”的程序员。如果你还没跑通过API调用先去把key申请下来、跑通一个最简单的对话请求再回来读体验会顺畅很多。2. 多Agent协作的核心设计思路与整体拆分2.1 先想明白多Agent解决的是“任务膨胀”问题多Agent协作不是噱头它解决的是一个特别具体的痛点——单线程提示词的能力边界。你让一个大模型对话一次它的上下文窗口就那么多注意力会被长任务稀释。也就是说如果一次性告诉它“去读这五个文档然后统计数字再写一份分析报告”它大概率前面看完后面忘中途还容易把指令执行成四不像。多Agent的思路其实特别像职场分工。你不可能让一个新人同时干策划、执行、汇总、汇报四件事但你可以让四个人各管一段交接顺畅整条流水线就能跑起来。落到代码层面规划AgentDirector负责接收任务、拆解成子任务、判断每个子任务需要的工具或信息类型。检索AgentResearcher根据子任务去调用搜索API、读取本地文档、拉取数据库数据把原始信息收集回来。写作AgentWriter接收检索结果按规划Agent给出的格式要求生成最终文案并对结果做校验自查。这三个角色加起来每一个的单点任务仍然很小但串联之后却能执行一个复杂得多的完整业务流程。这才是多Agent真正值钱的地方用简单单元的协作去逼近复杂系统的能力。提示你可以把多Agent理解成“流水线”而不是“一群人在一个房间里开会”。流水线强调的是交接和顺序开会强调的是讨论和共识。入门阶段先做流水线别一上来就做复杂辩论式协作后者调度成本高到你会怀疑人生。2.2 编排方案选型自己写循环调度还是直接上框架这里是我踩过最深的坑。刚起步那会儿我第一反应是找个现成的多Agent框架比如某些编排库觉得“框架肯定比我手撸的稳”。然后我就陷入了配置地狱注册Agent、定义边、设置条件路由、管理节点状态……光读文档就读了半天最后写出来的东西我自己都不敢改。后来我反而不纠结框架了问了自己一个问题一个多Agent系统最核心的运行骨架是什么答案就三个字消息循环。每个Agent轮流被唤醒从共享缓冲区里取自己需要的输入调一次大模型API把结果写回缓冲判断是否结束。就这么简单。这个逻辑用纯Python写也就50行核心代码。对整个流程有了手感之后再去看框架你会发现框架不过是在帮你处理更复杂的并发、回溯、多分支路由你自己心里有底。所以我给小白程序员的建议非常直白第一版别用框架自己写调度循环。等你明确了协同流程发现自己在反复造轮子、处理并发和断点续跑时再引入现成编排框架那是锦上添花不是雪中送炭。我整理了两种方案的适用场景对比方案适合阶段核心优势常见坑自写Python调度循环初学者第一版流程透明能彻底理解状态流转并发和容错需要自己补状态图编排框架流程变动频繁、分支复杂内置状态管理、支持条件路由配置感重调试门槛高消息队列解耦多服务部署、跨进程通信Agent可独立部署扩展性强起步阶段反而增加运维负担根据我的实际经验第一版永远选择最上面的方案。哪怕你的最终目标是框架先用循环跑通一次后面学框架时会发现所有概念都变得“原来如此”。2.3 系统角色设计每个Agent只做一件小事在设计Agent角色时最容易犯的错误是给一个Agent塞太多职责。我自己写坏过一个版本让一个“全能Agent”既负责拆任务又负责写最终报告结果这个Agent的Prompt被我越写越长因为要覆盖太多场景。最后它进入了一种“什么都想管却什么都做不精”的状态。正确做法是每个Agent的Prompt保持短短到只描述一个单一职责。规划Agent不需要知道报告怎么写检索Agent不需要了解用户审美偏好写作Agent不需要关心搜索词怎么构造。职责边界像API接口一样清晰Agent之间不共享prompt只通过结构化的数据交互。角色Prompt参考结构规划Agent Prompt核心你是任务规划专家。收到用户目标后输出JSON格式的子任务列表每个子任务包含id、类型、依赖、期望产出。检索Agent Prompt核心你是信息检索专家。根据给定的子任务描述调用搜索工具收集信息输出结构化信息列表包含来源和摘要。写作Agent Prompt核心你是文案写作专家。根据规划要求和检索信息生成最终报告。输出必须是完整的Markdown文档。这个结构让每个Prompt都短小精悍也给后面的调试保留了极大的操作空间。某个Agent出了质量问题你只需要单独优化那一个角色的Prompt完全不需要动其他逻辑。3. 核心细节解析会话消息、全局上下文和工具注册3.1 为什么SharedContext是全系统的“生命线”多Agent协作系统里最容易被忽略、却最关键的设计就是共享上下文SharedContext。打个比方三个Agent接力跑任务如果A的结果用了一个变量保存B的代码里又定义了另一个变量这两段代码根本对不上任务就断了。SharedContext就是解决这个“对不上”问题的地方。我常用的设计是一个全局字典对象# shared_context.py from dataclasses import dataclass, field from typing import Any, Dict, List dataclass class Task: id: str type: str # 子任务类型plan / search / write params: Dict[str, Any] # 子任务参数 result: Any None # Agent执行结果 status: str pending # pending / running / done / failed dataclass class SharedContext: goal: str # 用户的最终目标描述 plan: List[Task] field(default_factorylist) messages: List[Dict[str, str]] field(default_factorylist) final_output: str max_rounds: int 5 # 最大协作轮次防止死循环 def add_message(self, role: str, content: str): self.messages.append({role: role, content: content}) def get_task_by_id(self, task_id: str) - Task: for t in self.plan: if t.id task_id: return t raise ValueError(ftask {task_id} not found)这个Context字典的价值在于它是全系统唯一的数据源。所有Agent只和这个字典交流不直接互相调用函数。这样哪怕后面加了第四个Agent、第五个Agent或者把某个Agent换成不同的模型供应商整个系统骨架都不用动只是往里加数据和处理函数。你可能会问为什么不用数据库因为入门阶段数据量小、并发小一个内存字典的读写速度是毫秒级的而且断点调试特别方便。你随时可以打印Context里的所有状态看到底是哪一步、哪个Agent出了问题。这是数据库做不到的开发体验。3.2 消息路由机制每个Agent怎么知道自己该干什么有了SharedContext下一个关键问题就是调度循环如何决定轮到哪里个Agent。最直观的做法是“按预定义顺序跑一圈”规划Agent先跑跑完生成子任务列表然后检索Agent依次处理所有检索类任务最后写作Agent汇总输出。但真实场景更复杂。比如检索Agent跑完之后发现信息不足以支撑写作文案它需要回头找规划Agent重新规划。这时候就需要一个简单的“路由规则”来判断下一步执行谁。# router.py from typing import Callable from shared_context import SharedContext, Task def route_next(context: SharedContext) - str | None: # 返回下一步要执行的agent名称None表示流程结束 for task in context.plan: if task.status pending: return researcher # 所有子任务执行完毕轮转写手 if context.final_output : return writer return None这个路由函数写的规则很粗糙但够用只要还有没执行的检索子任务就继续让检索Agent干活都干完了交给写作Agent。真正的系统中你可以在Task上增加depends_on字段来表达依赖关系比如“写作Agent只能在所有检索任务成功后启动”这是后续优雅扩展的方向但第一版基于轮询的关键在于让流程首先跑通。注意路由逻辑的核心禁忌是“让Agent自己决定下一步干什么”。如果每次Agent调完大模型API后让模型输出一个“下一步行动”你确实得到了灵活行为但也收获了不确定性。小白阶段先把路由逻辑固化在代码里调模型只负责产出内容不负责决策调度。3.3 工具调用与函数注册给检索Agent装上手和脚模型本身不能上网、不能读文件只能“想”。要让Agent真正干活必须让你自己的Python函数被Agent调用。这就是函数注册机制。原理很简单你定义一个工具函数把这个函数的名称、描述、参数JSON Schema塞给大模型API模型输出一个“调用某函数、参数是什么”的结构你的代码再按这个结构去执行对应的函数。以下面这个“按关键词搜索内部资料库”的工具为例# tools.py from typing import Dict, List def search_knowledge_base(keyword: str, limit: int 5) - List[Dict[str, str]]: # 这里模拟搜索本地文档实际可替换为数据库查询或搜索接口 docs [ {title: 销售数据周报, content: 3月第一周销售额120万环比增长8%}, {title: 用户反馈汇总, content: 用户普遍关注加载速度和移动端适配}, ] result [d for d in docs if keyword in d[title] or keyword in d[content]][:limit] return result TOOL_SCHEMA { type: function, function: { name: search_knowledge_base, description: 在内部知识库中按关键词检索相关文档返回文档标题和内容摘要, parameters: { type: object, properties: { keyword: {type: string, description: 检索关键词}, limit: {type: integer, description: 返回的最大文档数默认5} }, required: [keyword] } } } TOOL_FUNCTIONS { search_knowledge_base: search_knowledge_base, }这一步做完你就完成了“给大模型装外挂”的闭环。模型不再只回答空对空的问题它可以真正去查数据、拿到了再回答。工具调用的返回结果也会追加进模型的上下文让后续推理依据真实数据。我给几个实操心得工具函数命名不要用缩写模型是靠“描述”理解工具用法的名字越直白越好。参数尽量少超过三个必填参数的函数模型调用出错率会明显上升。每个工具函数必须要try-except兜底。模型生成参数时偶尔会冒出奇怪的类型比如把数字写成字符串你的函数要能稳健处理。4. 实操从零搭建一个可运行的多Agent月度总结系统4.1 项目结构与依赖准备实操部分我直接给完整可跑的版本实现。整个项目文件结构如下agent_demo/ ├── shared_context.py # SharedContext与Task定义 ├── tools.py # 工具注册与函数实现 ├── agents.py # 三个Agent的调用逻辑 ├── router.py # 路由调度逻辑 ├── main.py # 入口运行整个协作流程 └── requirements.txt # 依赖清单依赖方面只需要一个数组模型SDK和标准库。我这边假设你用的是常见的大模型API SDK支持工具调用function calling。如果你用的是其他模型服务逻辑完全一样只是把客户端替换成对应SDK。# requirements.txt openai1.0.0装依赖pip install -r requirements.txt4.2 Agent实现代码把大模型API封装成可复用单元三个Agent的封装逻辑高度相似我建议写成一个基类统一管理再分别定义Prompt。核心函数只有一个run(task, context)接收子任务和全局上下文返回执行结果。# agents.py from openai import OpenAI from abc import ABC, abstractmethod from shared_context import SharedContext, Task from tools import TOOL_SCHEMA, TOOL_FUNCTIONS client OpenAI() # 从环境变量读取API Key class BaseAgent(ABC): Agent基类封装了大模型调用和工具执行 system_prompt: str def __init__(self): pass abstractmethod def run(self, task: Task, context: SharedContext) - None: ... def call_model(self, user_content: str, use_tools: bool False): messages [ {role: system, content: self.system_prompt}, {role: user, content: user_content}, ] params { model: gpt-4o-mini, temperature: 0.3, messages: messages, } if use_tools: params[tools] [TOOL_SCHEMA] resp client.chat.completions.create(**params) return resp.choices[0].message def exec_tool_calls(self, message) - str: tool_calls message.tool_calls or [] results [] for tc in tool_calls: func_name tc.function.name args eval(tc.function.arguments) # 注意内容来自模型严格场景需改用json if func_name in TOOL_FUNCTIONS: results.append(TOOL_FUNCTIONS[func_name](**args)) return \n.join([str(r) for r in results])这部分有两点需要留意。第一temperature我设成0.3多Agent协作场景我们更希望模型保持一致、理性、不发挥所以温度要低。第二exec_tool_calls里我用了eval解析参数演示可以生产环境请换成json.loads并且对参数做类型校验避免模型返回恶意或错误结构。接下来定义三个具体Agentclass PlannerAgent(BaseAgent): system_prompt 你是任务规划专家。请把用户的目标拆分为多个可执行的子任务。 所有子任务必须按JSON数组格式输出每个子任务包含 - id: 唯一字符串 - type: 只能是 search 或 write - params: 包含执行该子任务需要的参数描述 - description: 一句话说明这个子任务的产出目标 要求拆分合理、粒度适中不要超过5个子任务。 def run(self, task, context): resp self.call_model(f用户目标{context.goal}\n请输出拆分的子任务JSON。) content resp.content # 以下是一个简易解析逻辑提取[]内的内容 import json start content.find([) end content.rfind(]) 1 plan_data json.loads(content[start:end]) context.plan [Task(**item) for item in plan_data]规划Agent的输出解析是我实操中踩坑最多的地方。大模型的输出不一定总是纯净的JSON它偶尔会在前后加一些解释性文字。所以解析逻辑必须加“提取JSON片段”这一步而不是直接json.loads(整个输出)。检索Agentclass ResearcherAgent(BaseAgent): system_prompt 你是信息检索专家。根据给定的子任务描述调用检索工具收集信息。 注意必须实际调用工具不能凭空编造数据。 工具返回结果后请整理成信息列表每条包含来源标题和核心内容摘要。 def run(self, task, context): user_content f子任务描述:{task.description}\n参数:{task.params} first_message self.call_model(user_content, use_toolsTrue) if first_message.tool_calls: tool_result self.exec_tool_calls(first_message) # 把工具结果传给模型让它整理格式 second_message self.call_model( f工具返回了以下信息请整理成结构化列表\n{tool_result} ) task.result second_message.content task.status done else: task.result 未调用工具任务失败 task.status failed检索Agent有一个看似多余实则关键的设计模型先调用工具然后我再让模型“整理工具结果”。为什么不在第一次调用时直接让它边调用边输出因为工具调用和文字输出在同一轮里可能发生冲突模型可能写了半截文章又说要调用工具结果两样都没做好。分两步走稳定。写作Agentclass WriterAgent(BaseAgent): system_prompt 你是文案写作专家。根据规划目标和检索信息生成最终报告。 要求 1. 结构清晰使用Markdown标题。 2. 数据必须来源于检索信息不得编造。 3. 结尾附上信息来源列表。 def run(self, task, context): search_results [ f任务ID:{t.id}\n结果:{t.result} for t in context.plan if t.type search ] user_content f用户目标:{context.goal}\n\n检索信息:\n \n.join(search_results) resp self.call_model(user_content) context.final_output resp.content task.status done4.3 调度主循环让整条流水线自动跑起来主循环代码是整个系统的“发动机”。它的职责是反复询问路由函数找到下一个该执行的Agent执行它更新状态再问下一轮。注意这里有个关键点路由执行一轮可能只跑一个Agent也可能连续跑多个Agent这取决于子任务数量。# main.py from shared_context import SharedContext from agents import PlannerAgent, ResearcherAgent, WriterAgent from router import route_next def run_pipeline(user_goal: str): context SharedContext(goaluser_goal) agents { planner: PlannerAgent(), researcher: ResearcherAgent(), writer: WriterAgent(), } # 第一轮总是先规划 agents[planner].run(None, context) print(规划完成子任务列表) for task in context.plan: print(f - {task.id} [{task.type}] {task.description}) round_count 0 while True: if round_count context.max_rounds: print(超出最大轮次强制终止) break next_agent route_next(context) if next_agent is None: break print(f第{round_count 1}轮 - 执行Agent: {next_agent}) # 路线researcher处理第一个pending的search任务writer处理收尾 if next_agent researcher: task next((t for t in context.plan if t.status pending), None) agents[researcher].run(task, context) elif next_agent writer: task context.get_task_by_id(write_final) agents[writer].run(task, context) round_count 1 print(\n 最终输出 ) print(context.final_output) return context if __name__ __main__: run_pipeline(请生成一份3月份开发团队的月度工作总结报告)跑过这段代码的读者应该能体会到那个“魔法时刻”你只输入一个目标三个Agent像车间工人一样接力执行最后吐出完整报告。整个过程没有任何人干预每一个环节都是自动流转的。这个系统不复杂但它完整展示了多Agent开发的骨架。4.4 日志与中间状态可视化多Agent系统调试最大的困难是“看不到中间过程”。我的经验是从一开始就把每一步的输入输出都打印出来。你慢慢会发现大部分bug根本不需要断点调试只要看日志就能定位是哪个Agent说了胡话、哪个工具返回了空结果。我建议在关键节点都打印结构化信息def log_step(task_id: str, agent_name: str, status: str, result_preview: str): print(f[{status.upper()}] agent{agent_name} task{task_id} result{result_preview[:100]}...)这个简单的日志函数能救你无数次。比如你发现检索Agent的返回结果一直为空通过日志你马上能看出来是工具调用失败、模型参数传错、还是知识库本身没数据。不看日志瞎猜你会陷入“改了Prompt重新跑还是不对再改Prompt”的死循环。5. 常见问题与排查技巧实录5.1 六个高频异常与排查速查表多Agent系统跑不起来的原因和普通程序不太一样很多时候不是语法错误而是“模型理解偏差”或“状态流转不对”。我给你列一个我实践里出现频率最高的问题清单可以直接当查错手册用。现象可能原因排查与解决办法Agent输出不是合法JSON模型生成了多余解释文字提取JSON片段后再解析参考上面的start/end切片逻辑工具调用后结果没生效工具返回结果没有追加进模型上下文确保把tool_result拼接到下一次call_moel的用户消息里轮次耗尽但任务没完成子任务拆得太多Agent卡在失败的pending状态调高max_rounds给failed任务设置重试逻辑某个Agent输出质量突然变差上下文塞入了无关历史信息每个Agent只取自己需要的子集不要全量传多个Agent并发写同一个字段调度循环存在并发或顺序问题第一版保持单线程按路由顺序执行别提前优化模型调用报401/限流API Key配置错误或超出请求额度检查环境变量增加退避重试limit参数调低5.2 “模型突然不听话”的应对策略在实际运行中最让人崩溃的不是代码崩溃而是模型“自主发挥”。比如规划Agent明明只需要输出JSON子任务它非要附带一段自我评价“我认为这个计划很好”……这种多余输出会让JSON解析直接报错。应对方法我测试过最有效的是“双重保险”。第一重在Prompt中强约束输出格式并且给一个示例。第二重代码里做容错解析找不到合法JSON就从文本里“抠”出JSON。双重保险配合成功率从70%能提到95%以上。另外一个常用技巧是“让Agent输出结构化格式更容易解析”。与其让模型输出自由文本你再去解析不如引导它输出极度规整的内容。比如请严格按照下面格式输出不要添加任何解释 {plan: [{id: 1, type: search, description: ...}]}模型对这种“不考虑解释、直接给格式”的指令遵守度会显著提升。我自己实测带示例的Prompt比不带示例的成功率高将近20个百分点。5.3 如何给Agent加“反思反馈”机制如果你觉得上面的系统已经跑得很流畅那下一步就可以引入“验证环节”。目前的流水线有一个隐患写作Agent拿到检索结果后万一检索结果本身是空或者歧义大它照样会“硬写”一份报告出来里面全是编造内容。这个问题的解法是加一个“检查Agent”或“反馈回路”。核心逻辑是写作Agent写完后一个检查Agent对内容做校验。校验项目包括是否包含关键数据、是否引用了检索结果、是否符合字数要求、有没有明显逻辑断裂。如果校验不通过就把反馈意见交给写作Agent让它修改重写。class ReviewerAgent(BaseAgent): system_prompt 你是质量审核专家。对给定报告进行审核。 输出格式为JSON{pass: true/false, suggestions: [修改建议]} 如果数据缺失或内容空泛必须返回false。 def run(self, task, context): resp self.call_model(f报告内容:\n{context.final_output}) import json start resp.content.find({) end resp.content.rfind(}) 1 review json.loads(resp.content[start:end]) context.review_result review在路由函数里加入判断即可检查Agent返回false时重新触发writer。最多允许重写两轮防止死循环。这一个小改动会让系统输出从“能用”升级到“基本可靠”。6. 小白进阶路线从Demo到可维护项目的提升方向6.1 并发与异步改造让多个Agent不再空等目前的Demo是单线程流水线一个时间点只有一个Agent在运行。如果你的任务量上来了比如有10个子任务要检索逐个排队太慢。这时候值得引入asyncio做并发改造。改造思路不复杂把检索Agent的run改成异步函数用asyncio.gather并发执行所有无依赖的检索子任务。需要注意一个关键点SharedContext的写入操作要加锁保护避免多个Agent同时修改同一个字段造成数据覆盖。import asyncio from concurrent.futures import ThreadPoolExecutor async def run_researchers_concurrently(context): tasks [t for t in context.plan if t.type search] async def run_one(task): agent ResearcherAgent() agent.run(task, context) await asyncio.gather(*[run_one(t) for t in tasks])并发改造带来的收益非常明显。我优化后原来10个子任务串行需要80秒并发后直接降到12秒。代价就是调试难度增大建议你至少先把串行版本稳定跑通一周再考虑并发。6.2 上下文隔离与任务追溯让系统更工程化随着Agent增多共享一个全局Context会变得臃肿而且存在数据污染风险。比如写作Agent读了某个检索Agent的历史任务信息但这个信息其实已经过时了。工程化的解法是给每个子任务一个独立的数据域。dataclass class Task: id: str type: str params: Dict[str, Any] result: Any None status: str pending trace: List[str] field(default_factorylist) # 记录每一步操作日志 created_at: datetime field(default_factorydatetime.now)增加trace字段后你可以回溯每一个任务从创建到完成的完整路径。真正出问题时排查成本大幅下降。这个设计在Agent数量超过5个之后变得非常必要。6.3 善用评估集用自动化手段保证Agent质量最后给一个我的压箱底经验多Agent系统一定要做回归评测。很多人改完Prompt后只跑一遍案例发现效果好了就以为万事大吉。实际上这个Prompt可能把另一个案例改坏了。建议准备一套固定测试用例比如10个不同难度的任务每次修改Prompt或代码后全部跑一遍记录每个任务的输出是否通过检查。把这个评测脚本纳入你的日常开发流程你就有了一张“质量安全网”。具体做法判断输出里是否包含预期的关键字段、是否调用过预期工具、整体响应是否超时。用这套标准去衡量比肉眼判断靠谱得多。我一直坚持的做法是把评测集放到一个独立目录里日常版本变更时手动跑一遍。之后如果要做更高阶的自动化回测还可以玩出更多花样但基础思路就是“构建一个反馈回路让Agent的每次改动都能被量化验证”。7. 最后分享的一点心得这套多Agent协作系统我从第一次写出能跑的版本到现在已经迭代了三四茬。回头看最初那个版本问题一堆规划Agent经常输出不可解析的格式检索Agent偶尔会幻觉出错误的数据写作Agent上下文串味更是家常便饭。但正因为它足够简单每一个问题都能快速定位我反而是在反复修bug的过程中真正理解了多Agent体系的运转规律。所以如果你正准备接触这个方向我的建议就是别做太多设计先写一个最单纯的三Agent流水线让它在你的机器上跑出第一份报告。那份报告可能质量一般但你已经完整走通了这条链路。后面加的每一个机制无论是反思反馈、并发改造还是评测集全都是基于同一个骨架打补丁。先有一个骨架这个方向你一定走得下去。