基于LangGraph的自我修正代码生成Agent：从链式调用到图编排实践

LangGraph 这几年在 Agent 开发里出镜率越来越高。我一直坚持一个判断一个 Agent 能不能落地不是看它会不会聊天而是看它能不能闭环解决实际问题。今天要聊的项目就是一个基于LangGraph构建的“自我修正”代码生成 Agent它接收自然语言描述的任务输出对应代码然后在真实环境里执行测试一旦发现 bug就带着失败信息自动进入修复循环直到测试通过或达到上限。它把常见的“LLM 写代码—人来改 bug”变成“Agent 写代码—Agent 自己修”适合正在从链式调用转向图编排的 Agent 开发者也适合想给自己项目加一个“自动改代码”能力的团队。严格说这不是一个新鲜概念但用 LangGraph 来实现和以前用 LangChain 链式脚本堆出来的东西完全不是一回事。下面我会从设计思路、状态模型、核心代码到踩坑实录完整走一遍这个项目。1. 为什么“自我修正”在代码生成 Agent 里是刚需1.1 一次性生成的代码为什么不靠谱很多人第一次让大模型写代码都会被惊艳到需求描述清楚它能直接吐出一个像模像样的函数。但一旦把代码装进真实项目里跑问题就来了。模型对语法结构把握得还行但对 API 的拼写、参数的默认值、依赖库的实际行为经常是“编”出来的。比如它可能生成一个pd.DataFrame.plot()却传入一个该版本根本不存在的参数也可能调用某个工具函数时把返回类型当成另一个类型处理。更隐蔽的是逻辑错误。代码能运行但结果不对。边界条件没考虑、浮点精度被忽略、空列表直接取下标……这些问题靠人眼 review 不一定看得出来但测试用例一跑就现原形。所以“一次性生成可靠代码”这件事在模型能力没有质的飞跃之前基本是伪命题。现实可行的路径是把代码放进自动校验环境里用真实执行结果反哺模型修正。这就是自我修正闭环的出发点。1.2 自我修正闭环的价值自我修正并不是“让模型多生成几次碰运气”而是把执行反馈作为新一轮生成的硬约束。一个最小闭环长这样生成候选代码。编写或附带测试用例。在受控环境里执行测试收集 stdout、stderr、异常 traceback。把失败信息交给模型让它针对性修复。重复执行直到测试通过或达到最大尝试次数。这里的核心是第 4 步。模型看到的不是“请检查一下这段代码有没有问题”而是“测试在 line 12 抛出了 AssertionError期望值 55实际拿到 54”。这种具体反馈能把修复从“猜”变成“定位”。我做过一个对比实验同样一个斐波那契函数需求不带测试反馈时模型连续三次生成的代码都因边界条件栽跟头带上 pytest 失败输出后最多两轮就能改对。差距不是模型变聪明了而是反馈信息补上了模型缺失的执行上下文。1.3 为什么是 LangGraph如果只是做一个循环用普通 Python while 也能写。但项目一旦长出多个职责规划、生成、测试、审查代码就会变成一团乱麻。LangGraph 的价值在于把 Agent 流程显式建模成图。它做了几件关键的事有向图编排节点代表处理步骤边代表流转关系循环不是靠 while 硬写而是条件边形成的回边。统一状态管理所有节点共享一个 State 对象跨节点读写都有明确约定不用自己维护一堆全局变量。支持条件分支根据测试结果决定是继续修复还是结束用条件边就能表达改起来非常直观。Checkpointer可以持久化每一步的中间状态中断后能恢复执行这对长任务、人工审核流程非常有用。简单说LangGraph 把 Agent 从“一段有循环的脚本”升级成了“一张可控制、可观察、可恢复的流程图”。这也是我推荐用它而不是堆 if/else 的原因。2. 架构设计与状态模型2.1 节点拆分Planner / Coder / Tester / Critic我在这套架构里把流程拆成四个节点。拆细一点每个节点的 prompt 都更容易优化也方便后续单独替换或升级某一环。节点职责输入来源输出目标Planner拆解需求产出实现方案和测试策略用户任务plan 字段Coder根据方案生成代码或根据反馈修复代码plan feedbackcode 字段Tester在受控环境执行测试收集执行反馈code test_codefeedback 字段Critic可选对反馈做二次分析提取关键错误信息feedback code精简后的诊断结果Planner 不直接写代码它先想清楚“要做什么、边界在哪、怎么测”。这一步能显著减少 Coder 瞎猜的概率。Coder 是唯一生成代码的节点既要产出实现也要在修复轮次里读取上一步的失败信息。Tester 很关键它不调用模型只做真实执行。我建议把它实现成一个纯函数传入代码和测试返回执行反馈。因为不依赖外部模型Tester 可以百分百确定性地验证“代码到底行不行”不会被模型带偏。Critic 我在早期版本里没加后来发现失败信息太长时模型容易迷失在冗长的日志里。Critic 会从 traceback 中提取最后几行关键错误、失败断言对应的变量值再交给 Coder修复命中率明显提升。但它会增加一次额外调用成本敏感的团队可以酌情取舍。2.2 State 怎么设计LangGraph 的所有节点都围绕一个 State 对象工作。我的核心状态定义如下from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): task: str # 原始需求 plan: str # 实现方案 code: str # 当前代码 test_code: str # 测试代码 feedback: str # 最近一次执行反馈 attempts: int # 已尝试次数 messages: Annotated[List[str], operator.add] # 对话历史自动累加字段设计有几个讲究attempts是硬性计数每个测试节点返回时加一用来做终止判断。messages使用Annotated[List[str], operator.add]这是 LangGraph 的 reducer 机制多个节点往同一字段追加内容时不会互相覆盖。feedback只保留最近一轮避免把全部历史日志堆积起来撑爆上下文。这里最容易踩的坑是直接把code也设计成“可追加”的。千万别这么干。代码应当是整体替换不是叠加所以它不需要 reducer普通覆盖赋值就好。2.3 终止条件与防死循环策略自我修正最怕的就是“修到天荒地老”。我在条件边里同时做了三重保险硬性轮数上限attempts MAX_ATTEMPTS时强制结束。测试通过反馈里出现明确的通过标记立即结束。无进展检测如果连续两轮返回的code完全相同或者失败错误类型没变化判定为“修不动了”提前结束。条件边的实现是一个纯函数返回字符串决定下一条边的走向def should_continue(state: AgentState) - str: if PASSED in state[feedback]: return end if state[attempts] MAX_ATTEMPTS: return end return fix这个函数越简单越好。千万不要在里面写大段逻辑否则调试时你会疯。判断条件越明确图的流转越可控。3. 动手实现从零搭一个可运行的 LangGraph Agent3.1 环境准备与依赖我用的环境是 Python 3.10核心依赖就这么几个pip install langgraph langchain-openai openai如果你用本地模型或者其它厂商接口只需要替换 LLM 封装层。我这里用langchain-openai因为它的接口足够标准换模型时改动最小。还要准备一个 API Key。建议环境变量方式不要硬编码进代码export OPENAI_API_KEY你的key3.2 核心代码节点、条件边与编译先定义一个统一的 LLM 调用入口。我习惯把模型调用单独封装方便后续替换模型或加缓存from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0.3)然后是四个节点。Planner 节点def planner(state: AgentState) - dict: prompt f你是一个资深工程师。下面是用户需求 {state[task]} 请给出简要实现方案包括算法思路、主要边界条件、建议的测试用例。不要写完整代码。 resp llm.invoke(prompt) return {plan: resp.content, messages: [f[plan] {resp.content}]}Coder 节点要区分首次生成和修复两种情况。判断依据是state[attempts] 0def coder(state: AgentState) - dict: if state[attempts] 0: prompt f请根据方案实现代码。方案 {state[plan]} 需求{state[task]} 只输出可运行的 Python 代码不要解释。 else: prompt f上一版代码在测试中失败。代码 {state[code]} 失败反馈 {state[feedback]} 请修复代码。只输出可运行的完整 Python 代码不要解释。 resp llm.invoke(prompt) return {code: resp.content, messages: [f[coder] 生成/修复代码长度 {len(resp.content)}]}Tester 节点不调用 LLM。我采用写临时文件、subprocess 执行 pytest 的方式import subprocess, tempfile, os, sys def tester(state: AgentState) - dict: with tempfile.TemporaryDirectory() as tmp: code_path os.path.join(tmp, solution.py) test_path os.path.join(tmp, test_solution.py) # 用户任务没有附带测试时可以让 Coder 顺带生成或走 Planner 输出 with open(code_path, w, encodingutf-8) as f: f.write(state[code]) with open(test_path, w, encodingutf-8) as f: f.write(state[test_code]) try: result subprocess.run( [sys.executable, -m, pytest, test_path, -q, --tbshort], capture_outputTrue, textTrue, timeout30 ) output result.stdout result.stderr except subprocess.TimeoutExpired: output TEST_TIMEOUT: 测试执行超时 30 秒 passed passed in output and failed not in output feedback PASSED\n output if passed else FAILED\n output return {feedback: feedback, attempts: state[attempts] 1}注意我判断通过条件是passed in output and failed not in output因为 pytest 输出中既有 “3 passed” 也可能包含别的单词。这个判断要按自己的测试框架调整。然后构图from langgraph.graph import StateGraph, START, END MAX_ATTEMPTS 4 builder StateGraph(AgentState) builder.add_node(planner, planner) builder.add_node(coder, coder) builder.add_node(tester, tester) builder.add_edge(START, planner) builder.add_edge(planner, coder) builder.add_edge(coder, tester) builder.add_conditional_edges( tester, should_continue, {fix: coder, end: END} ) agent builder.compile()到这里一个可运行的自我修正循环就成型了。调用方式很直接initial_state { task: 实现一个函数 compute_fib(n)返回第 n 个斐波那契数n 从 0 开始。, test_code: from solution import compute_fib def test_fib(): assert compute_fib(0) 0 assert compute_fib(1) 1 assert compute_fib(10) 55 , attempts: 0, messages: [], } result agent.invoke(initial_state) print(result[code]) print(result[feedback][:500])3.3 运行效果与关键日志解读我第一次跑这个流程时输出大致是这样的Planner 给出方案用迭代而非递归避免栈溢出。Coder 生成递归版本。Tester 执行 pytest反馈FAILED test_fib... Expected 55, got 55?之类。实际第一次生成的是递归版测试超时或失败。第二轮 Coder 看到RecursionError或Timeout改为迭代版本。第三轮测试通过。真正考验人的是第三轮以后如果错误信息不明确模型会开始“左右横跳”。这时候我会打开state[messages]看看每一轮传给 Coder 的反馈到底是什么。很多“修复无效”其实是因为反馈里只有一句话“测试失败”没有关键报错行。所以我强烈建议Tester 返回的feedback一定要包含失败用例的名称期望值与实际值异常类型和 traceback 最后 3 行模型对报错尾巴的依赖比很多人想象中更强。3.4 参数调优与模型选择调参这块我没有太多玄学分享几个实际经验temperature代码生成场景我建议 0.2 到 0.4。太低0的话修复时容易原样输出太高0.7则会出现“改对了但多改成错的”情况。模型选择像 gpt-4o-mini 这类小模型处理简单逻辑足够快且便宜复杂项目结构生成上更强模型更划算。如果预算允许gpt-4o或同等水平的模型在“理解失败反馈”上明显更好。上下文窗口每轮修复都会塞入一段失败日志累计多了很容易撞到上下文上限。只保留最近两轮的feedback更早的先压缩成一句摘要。这是最实用的省 token 方法。超时设置Tester 节点一定要设timeout否则遇到死循环代码整个 Agent 会卡死。3.5 让 Agent 记住上一次会话Checkpointer 的使用LangGraph 的 Checkpointer 是它区别于普通编排框架的重要特性。在编译时挂上内存版检查点from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() agent builder.compile(checkpointercheckpointer)之后每次调用都带上同一个thread_idAgent 就能从上次中断的地方继续执行config {configurable: {thread_id: task-fib-001}} result agent.invoke(initial_state, config)实际值是如果某一次测试运行到一半因为外部原因中断比如网络抖动、超时你不需要从头开始再次调用同一个thread_idLangGraph 会从上一次保存的状态继续。这在接人工审核、故障恢复场景里非常有用。生产环境建议用SqliteSaver或对应数据库实现内存版只适合开发调试。4. 常见问题与排查技巧实录4.1 循环不终止或无限修复最常见的原因有三类should_continue的返回值没有出现在边字典里attempts没累加判断“通过”的逻辑写错导致明明已经 PASSED 还继续修。排查顺序先在条件边函数里把state[feedback]和attempts打印出来确认实际状态值再用agent.get_graph().draw_mermaid()或直接打印图结构确认边的名称匹配。我早期在一个项目里把返回字符串写成fix但边字典里写成了{retry: coder}排查了很久才发现是拼写不一致。图框架不会帮你校验这个全靠自己细心。4.2 上下文被历史消息撑爆只要跑过三轮以上messages里塞的完整代码和日志就会非常夸张。我建议feedback只保留最近一轮不要追加到长历史里。messages列表加一个上限比如保留最近 10 条超过就把最早的消息移除。如果日志太长先用 Critic 节点或简单字符串截断只保留 traceback 最后部分。API 层报 token 超限几乎都是历史污染导致的而不是单次 prompt 太长。4.3 测试环境的安全问题这是 Agent 开发最容易踩的雷。Coder 生成的代码是不可信的让它在宿主机上直接执行等于把任意代码执行能力交给了模型。我的做法是使用 Docker 容器或独立的临时目录执行测试代码。必须加timeout防止恶意或者低质量代码卡死。不给测试进程任何网络权限和环境变量中的密钥。安装依赖时使用固定版本避免模型生成的 import 引入意外包。Agent 安全不是事后补的它必须是 Tester 节点的默认设计。尤其当你准备把 Agent 接到 CI/CD 流水线时这条红线不能碰。4.4 状态更新不同步LangGraph 的多节点共享状态本身是有序执行的但如果你在单个节点里多次调用模型或多次给同一字段赋值很容易出现“旧值覆盖新值”。解决办法一个节点只返回一个明确字段集合返回值用字面量或显式变量不要直接用state的整体引用。需要累加的字段如messages用 reducer避免覆盖。我见过有人为了省一次调用在一个节点里连调两次llm.invoke第二次没拿到第一次的结果直接覆盖了状态里的code。拆节点就是拆风险别犯懒。4.5 修复无效Agent 反复输出相同代码如果连续两轮代码一模一样原因基本是两个一是反馈信息里没有指出具体错误点模型只能盲猜二是模型能力不够没法把“错误描述”映射为“代码修改”。解决办法检查feedback是否包含具体行号和期望值。没有就改 Tester 的日志采集。在 Coder 的 prompt 里加一句“务必修改上一版代码不要原样输出”。给 Critic 节点加职责要求它指出“上一版代码的精确错误位置”并输出“建议改动的最小差异”。实在不行就升级模型这不是优化能解决的。4.6 成本与性能控制每个任务最少会调用两次模型Planner Coder如果修复三轮就是五次以上。我实际跑下来的经验简单任务控制在 3 次调用以内超过 MAX_ATTEMPTS 直接放弃别让 Agent 无限烧钱。同一任务的首次生成和多次修复中修复轮次的 prompt 更长token 消耗也更高。用缓存或剪枝控制。加一个简单的日志记录统计每个任务的调用次数和 token 消耗优化时有据可依。问题表现我的排查方向循环不终止一直修复attempts 不涨或边不匹配打印状态 核对边名上下文超限API 报 token 超限裁剪 messages、只留最新 feedback代码被旧值覆盖修复后 code 没变检查节点返回值不要覆盖式赋值测试环境不安全任意代码直接执行Docker/沙箱 timeout 无网络反复输出相同代码修复无进展强化 feedback、加 Critic、换更强模型token 消耗过高账单飙升设 MAX_ATTEMPTS、压缩日志、加缓存最后说点我自己的体会。这套架构第一个版本我做得非常糙Planner、Coder、Tester 全让一个节点干结果失败信息一多模型开始自我催眠前几轮还能定位问题后面就只会把报错复制一遍。把职责拆开之后每轮的 prompt 都变得很短、很聚焦修复率明显上去了。还有一个小技巧tester 节点里记得把 stdout 和 stderr 原样记录尤其是 traceback 的最后两三行一句话都不要剪模型对异常日志的依赖比我们想象中高很多。目前我又在这条路上加了两样东西一个是静态检查工具比如 ruff 扫描作为额外的 feedback 来源另一个是把已修复的样本存下来做 few-shot。如果你也在折腾 Agent欢迎拿这套骨架改自己的版本。