从一次凌晨三点的事故说起去年做智能客服系统重构我犯了个低级错误——让单个Agent既写代码又自测。结果上线当天它把“用户退款”的SQL写成了DELETE FROM orders WHERE status‘refund’还自信满满地标注“测试通过”。凌晨三点被运维电话叫醒看着后台2000条订单被物理删除我盯着屏幕骂了句“这Agent要是能分角色干活至于出这种幺蛾子”后来我花了三周时间用LangGraph搭了一套多Agent协作框架让三个Agent分别扮演项目经理、开发者和测试者。今天这篇笔记就是这套系统的核心实现——代码可以直接跑但坑我已经替你们踩过了。系统架构别把Agent当人当微服务多Agent协作最忌讳“拟人化”。别想着让Agent像人类团队那样开会、投票、写周报那是浪费token。我的设计原则是每个Agent是一个有状态的计算节点通过结构化消息总线通信。核心组件就三个Orchestrator编排器负责任务分解、状态机流转、异常重试Message Bus消息总线基于Redis Stream的异步队列保证消息不丢Agent Worker Pool每个角色独立部署通过gRPC调用LLM# 别这样写把Agent当对象互相调用耦合度爆炸# class ProjectManager:# def assign_task(self, developer):# developer.do_work()# 正确的做法通过消息总线解耦classOrchestrator:def__init__(self):self.state_machine{INIT:self._handle_init,REQUIREMENT:self._handle_requirement,DEVELOPMENT:self._handle_development,TESTING:self._handle_testing,REVIEW:self._handle_review,COMPLETED:self._handle_completed}self.message_queueRedisStream(hostlocalhost,port6379)self.task_storeSQLiteStore(agent_tasks.db)# 这里踩过坑用内存存储进程重启全丢defrun(self,user_request:str):task_idself._create_task(user_request)self.message_queue.publish(pm_channel,{task_id:task_id,action:ANALYZE_REQUIREMENT,payload:user_request})returntask_id项目经理Agent别让它写需求文档让它拆任务项目经理Agent最容易犯的错是“过度规划”。我见过一个PM Agent写了30页PRD结果开发者Agent根本读不完。PM Agent的核心能力是任务分解和优先级排序不是写文档。classPMAgent:def__init__(self,llm_client):self.llmllm_client self.task_template 你是一个资深项目经理擅长将复杂需求拆解为可执行的任务。 当前需求{user_request} 请按以下格式输出任务列表JSON格式 {{ tasks: [ {{ id: T001, description: 具体任务描述不超过50字, dependencies: [], // 前置任务ID列表 estimated_hours: 4, priority: HIGH|MEDIUM|LOW, acceptance_criteria: 验收标准一句话说清楚 }} ], architecture_notes: 技术架构建议不超过100字 }} 注意 - 任务粒度控制在2-8小时别拆太细 - 依赖关系必须形成DAG不能有循环依赖 - 优先级只给一个HIGH其他都是MEDIUM或LOW defanalyze_requirement(self,user_request:str)-dict:# 这里踩过坑直接让LLM输出JSON经常格式错误# 改用function calling强制结构化输出responseself.llm.chat.completions.create(modelgpt-4,messages[{role:user,content:self.task_template.format(user_requestuser_request)}],response_format{type:json_object}# 强制JSON输出别这样写用正则解析)tasksjson.loads(response.choices[0].message.content)# 验证DAG无环ifnotself._validate_dag(tasks[tasks]):raiseValueError(任务依赖存在循环重新生成)returntasks经验PM Agent的输出一定要结构化别让它写自然语言。我后来把输出格式改成了Pydantic模型配合LangChain的with_structured_output错误率从30%降到了2%。开发者Agent代码生成不是重点上下文管理才是开发者Agent最容易翻车的地方是“忘记上下文”。比如PM说“实现用户登录”开发者写了个登录接口但没考虑PM之前说的“需要支持OAuth2.0”。解决方案是给开发者Agent一个“上下文快照”包含所有相关任务和依赖。classDeveloperAgent:def__init__(self,llm_client,code_repo:str):self.llmllm_client self.repo_pathcode_repo self.context_window[]# 维护最近5个任务的上下文defimplement_task(self,task:dict,project_context:dict):# 构建上下文包含当前任务、依赖任务、架构说明context{current_task:task,dependency_tasks:self._get_dependency_tasks(task[dependencies]),architecture_notes:project_context.get(architecture_notes,),existing_code:self._read_relevant_files(task[description])# 这里踩过坑读整个项目token爆炸}# 限制上下文大小只读取与任务相关的文件promptf 你是一个资深后端开发者正在实现以下任务 任务描述{task[description]}验收标准{task[acceptance_criteria]}项目架构说明{context[architecture_notes]}依赖任务代码摘要{self._summarize_code(context[dependency_tasks])}请生成实现代码注意 1. 遵循项目现有的代码风格从已有代码中学习 2. 添加必要的错误处理和日志 3. 不要重复实现依赖任务中已有的功能 4. 输出格式文件名和完整代码 已有相关代码{context[existing_code]}responseself.llm.chat.completions.create(modelgpt-4,messages[{role:user,content:prompt}],temperature0.3# 代码生成用低温度别这样写用0.7会生成花里胡哨的代码)# 解析代码并写入文件code_blocksself._parse_code_blocks(response.choices[0].message.content)forfilename,codeincode_blocks.items():filepathos.path.join(self.repo_path,filename)os.makedirs(os.path.dirname(filepath),exist_okTrue)withopen(filepath,w)asf:f.write(code)# 更新上下文窗口self.context_window.append({task_id:task[id],files:list(code_blocks.keys()),summary:self._generate_summary(code_blocks)})iflen(self.context_window)5:self.context_window.pop(0)return{status:completed,files:list(code_blocks.keys())}血泪教训开发者Agent的temperature一定要低。我试过0.7它给我生成了带emoji注释的代码还自作主张加了“智能缓存”功能结果把Redis搞崩了。测试者Agent别只测功能要测边界和异常测试者Agent最容易变成“点头机器”——开发者说“我写好了”测试者说“好的通过”。真正的测试Agent应该是个“杠精”专门找茬。classTesterAgent:def__init__(self,llm_client,test_frameworkpytest):self.llmllm_client self.test_frameworktest_framework self.bug_report_template 缺陷报告 #{bug_id} 严重程度{severity} (CRITICAL/MAJOR/MINOR) 任务ID{task_id} 缺陷描述{description} 复现步骤{steps} 期望结果{expected} 实际结果{actual} 建议修复{suggestion} deftest_task(self,task:dict,code_files:list,test_cases:listNone):# 先执行已有的单元测试test_resultsself._run_existing_tests(code_files)# 再让LLM生成边界测试用例ifnottest_cases:test_casesself._generate_test_cases(task,code_files)# 执行新生成的测试new_resultsself._run_generated_tests(test_cases)# 汇总缺陷bugs[]forresultintest_resultsnew_results:ifresult[status]FAILED:bug_reportself._analyze_failure(result,task)bugs.append(bug_report)# 这里踩过坑只报告错误不报告警告# 后来加了代码质量检查quality_issuesself._code_review(code_files)forissueinquality_issues:ifissue[severity]in[CRITICAL,MAJOR]:bugs.append(issue)return{task_id:task[id],passed:len(bugs)0,bugs:bugs,coverage:self._calculate_coverage(code_files)}def_generate_test_cases(self,task:dict,code_files:list)-list:promptf 你是一个严格的测试工程师正在测试以下代码 任务描述{task[description]}验收标准{task[acceptance_criteria]}代码文件{self._read_code_files(code_files)}请生成测试用例重点关注 1. 正常路径Happy Path 2. 边界条件空值、最大值、最小值 3. 异常情况网络超时、数据库连接失败、权限不足 4. 并发场景如果适用 每个测试用例格式 {{ name: test_xxx, type: unit|integration|e2e, input: {{}}, expected_output: {{}}, setup: 前置条件, teardown: 清理操作 }} 至少生成5个测试用例其中至少2个是异常场景。 responseself.llm.chat.completions.create(modelgpt-4,messages[{role:user,content:prompt}],temperature0.5# 测试用例生成可以稍微高一点增加多样性)returnjson.loads(response.choices[0].message.content)关键点测试Agent必须能“自举”——自己生成测试用例自己执行自己报告。我见过最蠢的设计是测试Agent只调用pytest结果开发者没写测试测试Agent就报告“全部通过”。协作流程状态机 重试机制三个Agent不是并行跑的而是通过状态机串行部分并行。核心流程INIT - REQUIREMENT (PM) - DEVELOPMENT (Dev) - TESTING (Tester) - REVIEW (PMDevTester) - COMPLETED如果测试失败状态回退到DEVELOPMENT并带上缺陷报告。classCollaborationEngine:def__init__(self):self.pmPMAgent(llm_client)self.devDeveloperAgent(llm_client,/tmp/repo)self.testerTesterAgent(llm_client)self.max_retries3# 每个任务最多重试3次defexecute_project(self,user_request:str):task_idstr(uuid.uuid4())stateINITretry_count0whilestate!COMPLETED:ifstateINIT:# PM分析需求tasksself.pm.analyze_requirement(user_request)stateREQUIREMENTelifstateREQUIREMENT:# 按依赖顺序执行任务fortaskinself._topological_sort(tasks[tasks]):stateDEVELOPMENTbreakelifstateDEVELOPMENT:# 开发者实现resultself.dev.implement_task(current_task,{architecture_notes:tasks.get(architecture_notes)})stateTESTINGelifstateTESTING:# 测试者测试test_resultself.tester.test_task(current_task,result[files])iftest_result[passed]:stateREVIEWelse:retry_count1ifretry_countself.max_retries:# 这里踩过坑无限重试导致死循环# 超过重试次数标记为失败人工介入self._notify_human(fTask{current_task[id]}failed after{self.max_retries}retries)stateCOMPLETEDelse:# 把缺陷报告传给开发者self.dev.receive_bug_report(test_result[bugs])stateDEVELOPMENTelifstateREVIEW:# PM和测试者联合评审review_resultself._joint_review(current_task,result,test_result)ifreview_result[approved]:# 继续下一个任务stateself._next_task(tasks[tasks])else:stateDEVELOPMENTreturn{task_id:task_id,status:completed}注意状态机一定要有超时机制。我遇到过开发者Agent卡在“思考”状态半小时后来加了每个步骤的timeout超时后自动重试。踩坑记录这些错误我花了三天才定位消息丢失用Redis List做消息队列消费者挂了消息就丢了。换成Redis Stream Consumer Group配合ACK机制才解决。上下文污染开发者Agent的context_window如果不限制大小会越积越多最后token爆炸。我设了5个任务的窗口超过就丢弃最早的。测试用例重复测试者Agent每次测试都生成新的测试用例导致重复执行。后来加了测试用例缓存用任务ID代码hash作为key。PM过度拆分PM Agent把“实现登录”拆成了20个任务每个任务0.5小时。后来加了任务粒度约束最小2小时最大8小时。LLM幻觉开发者Agent经常“发明”不存在的API。解决方案是在prompt里明确说“只使用标准库和requirements.txt里列出的依赖”。性能优化别让Agent闲着多Agent协作最大的性能瓶颈是LLM调用。每个Agent调用一次LLM平均3-5秒一个项目下来可能上百次调用。优化方案# 异步调用别这样写串行调用一个卡住全卡住asyncdefexecute_task_async(self,task):pm_taskasyncio.create_task(self.pm.analyze_requirement_async(task))dev_taskasyncio.create_task(self.dev.implement_task_async(task))tester_taskasyncio.create_task(self.tester.test_task_async(task))# 并行执行不依赖的任务resultsawaitasyncio.gather(pm_task,dev_task,tester_task)returnresults另外缓存LLM响应。同样的prompt同样的输入结果应该一样。我用了Redis缓存TTL设24小时节省了约40%的API调用。个人经验别追求“全自动”留好人工接口这套系统跑了一个月成功率大概85%。剩下的15%包括PM拆任务不合理比如把“优化数据库”拆成“修改索引”和“重写查询”但两者强耦合开发者引入安全漏洞比如SQL注入测试者没测出来测试者漏测边界条件比如并发写入导致数据不一致我的建议多Agent协作系统一定要有“人工介入点”。我在每个关键节点都加了Webhook可以暂停、修改、回滚。比如PM拆完任务后会发到Slack让架构师审核测试发现CRITICAL缺陷时自动创建Jira工单。最后说句大实话多Agent协作不是银弹。如果你的需求是“写个Hello World”单Agent就够了。但如果你要构建一个需要多人协作的复杂系统这套架构能帮你把开发周期从两周缩短到两天——前提是你愿意花一周时间调prompt和修bug。下一篇我会写《Agent的长期记忆如何让Agent记住三个月前的代码逻辑》到时候聊聊向量数据库和知识图谱的坑。