ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

基于LangGraph的多智能体测试与调试闭环设计

基于LangGraph的多智能体测试与调试闭环设计 我最近在一个多智能体项目里把“自动测试”和“调试闭环”做进了系统本身踩了不少坑也沉淀了一套能直接跑通的做法。核心框架用的就是标题里说的 LangGraph整体方案不是简单的多Agent串一条链而是让Agent自己写测试用例、自己执行、自己根据失败结果修代码全程还能断点、回放、人工介入。这套东西做下来最大的体感是测试不再是项目结束后的补丁调试也不再是出了错才去翻日志而是整个多智能体系统的“基础设施”。这篇文章就把我的设计思路、关键代码、执行流程和排查经验完整写出来适合已经在用LangChain但觉得编排不够灵活的人也适合想用多智能体做自动化开发闭环但不知道从哪下手的团队。1. 项目背景为什么多智能体系统需要“测试与调试闭环”1.1 多智能体不是聊天机器人不能靠“感觉”交付先泼一盆冷水如果多个Agent只是用LangChain依次调用那本质上还是一个“伪多智能体”——因为缺少状态流转、条件分支和循环控制Agent之间的协作只能靠硬编码拼接一旦中间某一步输出不符合预期后面的流程全崩。真正的问题是在一个多智能体系统里前面Agent的输出就是后面Agent的输入这种流水线式的信息传递会让错误像滚雪球一样越滚越大。比如让一个Agent写代码另一个Agent做Code Review如果第二个Agent给出的审查意见本身就是错的那第三个Agent的修改方向也会跟着错。所以我接手这个项目时第一件事不是写Agent提示词而是给系统设计“可观察、可打断、可回放”的运行时。这个运行时决定了每一步谁在执行、输入是什么、输出是什么、状态走到哪了、我要在哪个节点介入。LangGraph恰好就是干这个的。1.2 LangGraph和LangChain有什么区别为什么选它热词里一直在问LangGraph和LangChain的区别这里直接给结论。LangChain更像组件库提供模型封装、工具调用、检索、记忆这些积木但它默认的执行模型是“链式”也就是一个输出接一个输入流程固定遇到分支和循环写起来很别扭。LangGraph的本质是一个图执行引擎——你定义一张有向图节点是逻辑单元边是流转方向状态在全局共享支持条件边、循环、断点、持久化和人工介入。我选LangGraph的原因很实际多智能体系统的本质就是“多个节点 灵活的路由 循环修复”这是标准链式结构表达不了的。比如我们的系统里“测试失败后回到工程师Agent修复”是个循环而“测试通过后进入Review节点”是条件分支用LangGraph的StateGraph画出来非常直观语义也清楚。LangChain里不是不能做但做出来的是回调地狱。另外LangGraph能复用LangChain里的模型封装和工具所以老项目迁移成本并不高。1.3 自动测试和代码Review本质上是“Harness工程化”热搜词里有一句说得挺准“AI自动写测试用例做自动测试以及代码Review这些都是Harness工程化的功能。”我把这句话翻译一下在实际工程里Harness是测试夹具、执行环境和护栏的统称你写代码光看逻辑对没用必须有一套东西能自动验证、自动约束、自动反馈。对多智能体系统来说Harness就是把“人类工程师平时做的验证工作”自动化掉。具体到这个项目我给系统设计了两条闭环测试闭环QA Agent根据任务描述和代码自动生成pytest测试用例在沙箱里执行收集通过率、错误堆栈、覆盖率把结果反馈给工程师Agent。调试闭环系统运行到关键节点时支持断点暂停支持从指定thread恢复每一步的输入输出都落盘追踪失败后可以原样回放定位是哪一步Agent“脑子进水”了。我把这两个闭环称为“内建质量体系”——不是跑完以后拿结果去人工检查而是让测试结果直接决定下一个节点形成一个自动修复的回路。这个设计对多智能体系统尤其重要因为Agent生成的东西本来就不稳定没有自动验证的闭环系统输出质量就失控。2. 整体设计把测试与调试设计成系统的一等公民2.1 系统架构与Agent分工整体上这个系统模拟一个“小型研发团队”。我给它取名最简单的角色划分避免概念上的过度设计Triage Agent接收用户任务拆解需求制定计划决定要动哪些文件、做什么范围的改动。Engineer Agent负责写代码和改代码实现Triage Agent拆解的计划。QA Agent根据任务和当前代码自动生成测试用例并执行测试返回测试结果、错误日志和覆盖率摘要。Reviewer Agent对代码做静态Review重点看逻辑漏洞、边界条件、潜在bug和安全问题。Reporter Agent把修复和测试结果汇总输出给调用方。这里不搞“每个Agent一个模型”的铺张浪费除了Reporter用轻量模型其余角色我用同一个较强模型跑不同提示词就够了。多智能体的重点不是模型数量而是状态分工与协作逻辑。2.2 状态设计一张State定义整个团队的工作台LangGraph里的状态State就是所有节点共享的工作台每个节点都能往里面写字段后续节点从此读取数据。状态设计做得好不好直接决定系统是否容易被调试。我定义的AgentState大概长这样from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class RepoContext(TypedDict, totalFalse): file_path: str language: str source_code: str class AgentState(TypedDict): messages: Annotated[list, add_messages] # Agent之间的对话历史 task: str # 用户原始任务 plan: str # Triage拆解出的步骤 repo: List[RepoContext] # 涉及的文件与当前内容 code: str # 工程师产出的最新代码 test_code: str # QA生成的测试代码 test_result: str # 测试执行结果摘要 coverage: str # 覆盖率摘要 review_comments: str # Review意见 iteration: int # 当前修复轮次 max_iterations: int # 最大轮次防止死循环 final_report: str # 最终交付结果这里有个细节值得强调为什么用Annotated[list, add_messages]而不是普通list因为多智能体里多个节点都会往对话消息里追加内容普通list会把整个旧消息列表覆盖掉或导致并发写冲突add_messages是LangGraph内置的Reducer能把新消息追加进去而不是覆盖。Reducer机制一开始容易被忽略但它是状态正确性的核心。2.3 为什么方案里必须有“环”而不是一条线如果系统只是一个DAG有向无环图那“测试失败后回修”这个动作就只能靠外部循环触发图本身表达不了。LangGraph的价值恰恰在支持“环”Engineer Agent写完代码QA节点执行测试如果失败条件边回到Engineer节点如果通过条件边去Reviewer节点。这种环形流转表达的是“自动修复闭环”。没有环就谈不上闭环系统就只能做单程航班。所以构建多智能体系统时我建议先画出业务闭环再定义图结构——哪些节点会重复执行、什么条件下退出循环、最多循环几次都必须在设计阶段想清楚。我还给图设置了两个“侧门”一个是持久化使用Checkpointer保存每一步状态一个是中断点可以在某个节点执行前暂停等人确认或修改后继续。这两个能力是把“调试闭环”落到实处的关键后面第三节和第四节详细讲。2.4 方案选型Checkpointer、中断点和追踪工具怎么配具体选型上开发环境我用MemorySaver做Checkpointer因为它轻量、内存态、重启即清空适合调试和单机Demo生产环境我建议换Postgres或SQLite持久化方便多个worker共享状态和恢复现场。中断点用LangGraph编译后传入interrupt_before参数即可比如在QA执行前拦一下人工确认测试计划合理再放行。追踪方面我开发期用LangSmith看完整trace同时自己也写了一个轻量的节点执行日志函数记录每个节点名称、输入长度、输出摘要、耗时和token数。后面讲调试时会展开这些配置。3. 核心实现从零到一跑通自动测试闭环3.1 定义图结构和基础节点先看整体图的搭建方式。基于前面定义的AgentState我构建了一个包含六个节点、三条条件边的图from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver def build_graph(): g StateGraph(AgentState) g.add_node(triage, triage_node) g.add_node(engineer, engineer_node) g.add_node(qa, qa_node) g.add_node(reviewer, reviewer_node) g.add_node(reporter, reporter_node) g.set_entry_point(triage) g.add_edge(triage, engineer) g.add_edge(engineer, qa) g.add_conditional_edges( qa, decide_after_qa, { repair: engineer, review: reviewer, done: reporter, }, ) g.add_conditional_edges( reviewer, decide_after_review, { repair: engineer, pass: reporter, }, ) g.add_edge(reporter, END) return g.compile( checkpointerMemorySaver(), interrupt_before[qa], # 断点示例执行测试前暂停 )这里最需要注意的是add_conditional_edges的用法它接受一个函数名和一个映射表函数返回映射表的Key图根据Key决定走哪条边。这种写法把路由逻辑从业务节点里抽离出来非常利于调试——我可以在路由函数里打印当前条件看清系统为什么走了某条边。3.2 Engineer节点让Agent真正“动手改代码”工程节点不仅负责写代码还负责根据测试结果和Review意见修改代码。核心思路是把上下文组装成完整prompt然后调用模型得到修改后的代码。def engineer_node(state: AgentState) - dict: context_parts [] for item in state.get(repo, []): context_parts.append(f文件{item[file_path]}\n内容\n{item[source_code]}) instructions f 你是一名工程师负责实现任务并修复测试问题。 任务{state[task]} 整体计划{state.get(plan, )} 当前代码 {state.get(code, state[repo][0][source_code])} 上轮测试结果如果有 {state.get(test_result, 无)} 上轮Review意见如果有 {state.get(review_comments, 无)} 当前修复轮次{state.get(iteration, 1)} 最大轮次{state[max_iterations]} 要求 1. 只输出最终完整的代码不要解释。 2. 保证代码语法正确函数签名与任务要求一致。 3. 如果上轮有测试失败或Review意见必须逐条修复。 # 这里可换用任何Code模型我用的是支持工具调用的Chat模型 response model.invoke([ {role: system, content: instructions}, {role: user, content: state[task]}, ]) new_code extract_code_block(str(response.content)) return { code: new_code, iteration: state.get(iteration, 0) 1, messages: [{role: assistant, content: f工程师Agent输出新代码长度{len(new_code)}}], }实操上有个比较大的坑模型输出经常会带markdown代码块或解释文字必须用extract_code_block把真正的代码抽出来。我的做法是用正则python ... 提取如果提取不到再按全文返回兜底。这一步看起来简单但漏了就会导致测试执行时报语法错误而且错误信息会让人误以为是模型写出了烂代码。3.3 QA节点让AI写测试用例并自动执行QA节点是“自动测试闭环”的核心拆成两步第一步让LLM生成pytest测试文件第二步用工具函数在子进程里执行。def qa_node(state: AgentState) - dict: prompt f 你是一名测试工程师。请根据任务要求和当前代码生成pytest测试用例。 任务{state[task]} 被测代码 {state[code]} 要求 1. 覆盖正常路径、边界路径、异常路径。 2. 使用assert断言一个用例一个函数函数以test_开头。 3. 只输出完整测试代码不要解释。 4. 如果代码里有明显的外部依赖网络、数据库优先用mock或打桩。 test_code model.invoke([ {role: system, content: prompt}, {role: user, content: f请为以下代码生成测试\n{state[code]}}, ]).content test_code extract_code_block(str(test_code)) test_result, coverage run_tests_with_coverage(test_code) return { test_code: test_code, test_result: test_result, coverage: coverage, messages: [{role: assistant, content: fQA Agent执行测试通过率摘要{test_result[:200]}}], }run_tests_with_coverage的实现关键是安全与超时控制。我在子进程里通过exec不直接执行而是把测试代码写到临时文件用subprocess.run跑pytest并设置5秒超时import subprocess, tempfile, json, os, pathlib def run_tests_with_coverage(test_code: str): with tempfile.TemporaryDirectory() as tmpdir: test_file pathlib.Path(tmpdir) / test_generated.py test_file.write_text(test_code, encodingutf-8) try: result subprocess.run( [python, -m, pytest, str(test_file), -p, no:cacheprovider, --cov., --cov-reportjson, -q, --timeout5], capture_outputTrue, textTrue, timeout30, ) stdout result.stdout[-1500:] stderr result.stderr[-800:] summary stdout stderr coverage_summary 无 cov_json_path pathlib.Path(tmpdir) / .coverage if cov_json_path.exists(): # 实际项目中要解析coverage json这里做简化占位 coverage_summary parse_coverage(cov_json_path) return summary, coverage_summary except subprocess.TimeoutExpired: return 测试执行超时, 无 except Exception as e: return f执行失败: {str(e)}, 无注意这里的坑pytest-cov对源码路径有要求必须保证被测代码在临时目录里也能被导入。如果临时目录和当前代码目录不在同一层--cov.会扫描错目录。我在实际项目里是先把被测文件也拷贝到临时目录再以临时目录为工作目录执行这样测试产物和源码在一起导入路径不会乱。另一个坑是pytest插件--timeout5依赖pytest-timeout没装会在执行时报错建议要么统一装好要么在命令里不写这个参数。3.4 自动生成测试用例的Prompt设计细节模型自动写测试用例最大的问题是“泛泛而谈”——生成的用例虽然语法合法但没有针对业务逻辑做断言测试跑完永远是绿的没意义。我试过几版Prompt后有效的写法是“给签名、给样例、给断言目标”给函数签名和类型提示模型不需要猜函数名和入参。给输入输出样例至少两组正常情况和边界情况。明确要求Assert具体值不要Assert无异常。对外部依赖显式要求mock。举例如果被测函数是def calculate_discount(price: float, vip: bool) - float我不让QA只看“生成测试”而是给出样例并指明“VIP用户8折、普通用户原价、价格小于等于0应抛异常”模型生成的用例质量会高很多。本质上自动测试生成不是推理任务而是要求模型把需求转成确定性验证逻辑的任务提示词里的确定性约束越多输出越可靠。3.5 条件路由与循环控制两个条件边函数是闭环的关键。decide_after_qa的伪逻辑def decide_after_qa(state: AgentState) - str: test_summary state.get(test_result, ) # 1. 测试超时或执行报语法错误 → 修 if 超时 in test_summary or SyntaxError in test_summary: return repair # 2. 失败用例数超过0 → 修 if failed in test_summary and 0 failed not in test_summary: return repair # 3. 超过最大迭代轮次 → 不再修直接去reporter兜底 if state.get(iteration, 0) state.get(max_iterations, 5): return done # 4. 测试全通过 → 转Review return reviewdecide_after_review同理如果review_comments里包含“严重问题”这类信号且未超过轮次就回repair否则pass。这里我建议在路由函数里加上日志打印本次判定依据比如“测试通过率100%转Review”否则系统行为不透明排查起来很累。4. 调试闭环断点、单步、重放与人工介入4.1 先让系统支持“打断”Checkpointer与interrupt_before自动测试闭环解决的是“系统自己验证”但开发期不够还必须在出问题时能打断。LangGraph里打断的方式是编译时指定interrupt_before或interrupt_after配合Checkpointer图会在指定节点执行前保存所有状态并等待外部输入。我在实际项目里这样执行config {configurable: {thread_id: task-42}} # 执行到qa节点前停下 events graph.stream({task: 写一个函数输入秒数返回时分秒}, config) for event in events: print(事件:, event)当系统停在qa节点前我可以检查当前状态里的code是否符合预期如果不想继续测试就修改状态后恢复from langgraph.types import Command graph.update_state(config, {code: corrected_code}) # 继续执行 for event in graph.stream(Command(resume{approved: True}), config): print(继续执行:, event)这跟调试一个普通程序太不一样了——普通程序打断点后看的是变量而这里打断点后看到的是“整个Agent的工作台”包括计划、代码、测试结果、上下文消息。对多智能体系统来说这种“看全局状态”的能力比单点debug重要得多。4.2 观察与追踪像串口调试助手一样看每一步状态我是在做一个串口工具项目时萌生这个想法的硬件工程师用串口调试助手看设备上报的每一个字节调试LLM系统其实需要同样的“逐包观察”。我对LangGraph的洞察是每个节点的输入输出都是结构化数据完全可以在节点外层做统一拦截而不是依赖LangSmith。做法是在每个节点函数外层加一个装饰器或包装函数def traced_node(name: str, func): def wrapper(state: AgentState) - dict: start time.time() result func(state) used time.time() - start print(f[TRACE] {name} 完成耗时{used:.2f}s输出字段{list(result.keys())}) for key, value in result.items(): if isinstance(value, str) and len(value) 300: print(f[TRACE] {key}: 长度{len(value)}前300字{value[:300]}) return result return wrapper有了这个Trace每个节点跑完我一眼就能看到“哪个节点耗时异常”“哪个节点输出为空”“哪个节点动了哪些字段”。LangSmith虽然更全面但它偏“事后看”而这种本地打印更接近开发期的即时反馈。生产环境我会同时启用LangSmith做全量追踪本地打印反而会关闭避免日志刷屏。4.3 失败回放与回归用例沉淀做了追踪之后最后一个调试闭环环节是“回放”。LangGraph搭配Checkpointer的好处是只要保留thread_id相关的检查点我就能在任意节点恢复执行。遇到一次线上失败我不需要重新让Agent跑一遍而是直接读取上次的状态重放最后几步定位是哪一轮循环把状态写坏了。更狠的用法是把失败案例沉淀成回归测试每次项目里某个任务的执行结果不符合预期我就在测试集里加一条“回归任务”要求系统必须跑出符合预期结果。然后每次图结构或Prompt改动后先跑一遍回归集确保旧的坑不会因为优化别的Agent又踩回去。这套做法特别像我们在自动化测试里说的“从bug到回归用例”只是这里的“单元”是整个多智能体流程。4.4 传统调试思路的迁移从断点、单步到Agent状态回放前面说了串口调试助手、GDB这些热词其实它们背后的调试套路完全可以迁移到LLM系统传统调试手段对应LangGraph能力用途串口/日志打印traced_node日志、节点输入输出落盘观察每一步发生了什么断点暂停interrupt_before / interrupt_after在关键节点前人工检查状态单步执行stream按节点逐步产出事件一次只走一个节点变量查看graph.get_state(config)查看全局Agent状态Core dump回放Checkpointer 同thread恢复失败现场原样复现修改后继续update_state Command(resume...)人工修正后恢复流程这个对应关系让我团队里的硬件工程师也很快理解了LangGraph的设计——不再觉得“AI系统没法调试”而是把多智能体当成一个分布式状态机来处理。状态机上任何一步都是可观测、可干预、可重放的。5. 常见问题与排查技巧实录5.1 多智能体状态污染字段被莫名覆盖状态污染是我遇到最多的问题。具体表现是Engineer Agent修改的code字段在QA节点执行后莫名变成旧值或者多个节点同时向messages追加内容时丢消息。原因基本是没有使用Reducer。LangGraph里的普通字段默认是“覆盖式赋值”后写的节点会覆盖先写的值如果两个节点都通过return {code: ...}写同一个字段后者一定会覆盖前者。解决办法想保留追加语义的字段必须声明Reducer典型就是messages: Annotated[list, add_messages]。其他自定义字段如果多个节点都要写也要自定义Reducer或设计独立的写入节点避免并发写冲突。排查手法也分享一个一旦发现状态不对在可疑节点前后各打印一次完整状态对比字段长度和值的变化基本能锁定是谁覆盖的。5.2 测试用例“假阳性”全绿但逻辑是错的自动生成的测试用例容易出现假阳性——Model写的断言太弱比如只断言函数返回不是None或者断言一个内部不会走到抛异常路径的异常。测试全绿给后面Review节点传递了错误信号Review也会觉得代码没问题最后交付的代码质量完全没保障。我的规避办法是双管齐下一是Prompt强制要求“每个测试必须包含至少一个具体值断言禁止只做类型断言”二是加一个成本很低的“测试用例评审”步骤让QA Agent在生成测试后先自评一轮列出每个用例覆盖的路径和断言目标连自己都说不出测了什么路径的测试用例直接重写。这步不额外增加Agent只是在同一个节点里多调一次模型。5.3 沙箱执行与超时问题小心子进程把宿主机搞挂AI生成的测试代码可能是恶意的也可能是写着玩的死循环所以在子进程里跑测试必须考虑隔离。我开发环境的经验是三层保障第一subprocess.run必须设置timeout并给内部pytest配置--timeout第二临时目录每次清理不允许测试代码访问项目源码目录之外的敏感路径第三如果测试要调用网络接口优先让QA Agent加mock必要时在子进程里用环境变量断开外网。最后一条更重要AI生成的测试里经常出现while True或巨大内存分配就算有超时我也建议用resource模块限制子进程的CPU和内存上限防住极端情况。这不是杞人忧天模型生成的代码一旦跑飞它造成的破坏和普通程序员写死循环是一样的。5.4 路由失效与迭代爆循环条件边不按预期走decide_after_qa是基于字符串匹配的判断一旦LLM生成的测试结果格式变化比如pytest输出从英文变中文、失败信息里没有failed字样条件边就会走错分支甚至在一个循环里出不来。我的经验是路由函数只做“关键信号提取”不要依赖完整摘要文本。具体做法是让QA节点在写state时额外写一个结构化字段比如test_metrics: {passed: 5, failed: 1, error: AssertionError}路由函数直接读取结构化字段判断而不是去字符串里碰运气。同时图结构里必须有max_iterations兜底。我踩过一次死循环当时Engineer修了一个问题QA又暴露了新问题两个Agent陷入反复修复循环最后靠外部手动杀了进程。从那以后我在decide_after_qa里第一行就检查迭代次数超过上限立刻转去Reporter输出半成品报告也比无限循环好。5.5 排查技巧速查表症状优先检查项常用处理某节点不执行条件边映射和路由函数返回值在路由函数加日志打印判断依据字段无变化/被覆盖是否缺Reducer是否多节点写同字段为字段声明Reducer或拆分节点测试全部报语法错误代码提取是否失败模型是否输出markdown多余块强化extract_code_block增加兜底清理循环不退出max_iterations是否生效路由判定是否命中增加结构化metrics防死循环恢复执行时状态丢失Checkpointer是否配置thread_id是否一致统一thread_id生产用持久化CheckpointerAgent回答质量突然下降上下文是否过长截断messages是否重复膨胀定期清理旧消息只保留必要摘要写在最后的个人体会这套系统从设计到跑通我最深的感觉是“多智能体系统的工程难度不在模型而在可控性”。LangGraph把图执行、状态持久化、断点恢复这些能力给齐了剩下的就看开发者怎么用好这套基础设施。现在我自己养成了一个固定习惯每次给图加新节点第一件事不是写节点逻辑而是先想清楚它的输入字段、输出字段、失败时的fallback路径再决定要不要在这里加断点。第二个习惯是越复杂的自动测试闭环越需要“小步快跑”先保证一个最小用例能跑通再逐渐放开Agent的自主性而不是一开始就让AI全权负责写代码又写测试又审核否则出了错根本分不清是哪一层的责任。这套设计目前已经在我们内部多个编码辅助场景里稳定跑了很久后续我也会把状态压缩、多线程并发执行、以及基于历史失败自动调整提示词的内容陆续整理出来欢迎一起讨论。
返回列表