LangGraph实战：构建能自我修正的代码生成Agent

1. 为什么“能跑”的代码生成 Agent 远远不够

代码生成这件事，从大模型能写函数那天起就一直是热门方向。但真正在生产里用过的人都知道，一次性生成的代码“能跑”和“能交付”之间隔着一条巨大的鸿沟。我最早做代码生成工具的时候，思路很朴素：给模型一段需求描述，让它吐出代码，然后人工检查。问题在于，模型生成的代码经常出现变量名拼错、边界条件漏判、依赖库版本对不上、甚至语法都过不了的情况。你让它改，它可能改对一处又弄坏另一处，来回几轮之后你自己都忘了最初的需求是什么。

这就是“自我修正”要解决的核心痛点。所谓自我修正，不是让模型自己说“我检查了一遍没问题”，而是要让 Agent 真正拿到执行反馈——编译报错、测试失败、静态检查告警——然后基于这些客观信号去定位问题、修改代码、重新验证，形成一个闭环。LangGraph 恰好是干这件事的趁手工具，它把 Agent 的执行流程建模成一张有向图，节点是具体的动作（生成、执行、检查、修正），边是状态流转的条件判断。相比传统的链式调用，图结构最大的好处是你可以让流程“回头”——修正节点执行完之后可以回到执行节点重新跑，直到满足退出条件。

这篇文章面向的是已经了解 LangChain 基本用法、想进一步做 Agent 编排的开发者。我会从整体设计思路讲起，把每个节点的职责、状态字段的设计、条件边的判断逻辑拆开说清楚，然后给出一套可以直接复现的实现方案。中间会穿插我在实际搭建过程中踩过的坑，比如状态污染、无限循环、执行超时这些问题怎么处理。读完之后你应该能自己搭一个能编译、能测试、能自动修 bug 的代码生成 Agent，而不是一个只会“看起来像代码”的文本生成器。

2. 整体架构设计与核心思路拆解

2.1 为什么选 LangGraph 而不是简单的 Chain

用 LangChain 的 SequentialChain 也能串起“生成→执行→修正”这几个步骤，但 Chain 的本质是线性流水线，它没有原生的循环能力。你当然可以在外面套一个 while 循环，手动判断要不要再来一轮，但这样状态管理、错误处理、退出条件全都散落在业务代码里，越写越乱。LangGraph 把这些问题收敛到了图的结构里：节点负责干活，边负责决策，状态在节点之间自动传递和合并。

更关键的一点是，LangGraph 支持条件边（conditional edge）。这意味着“执行完之后是去修正还是去结束”这个判断可以作为一个独立的函数存在，逻辑清晰、可测试。而且图结构天然支持并行分支，后面如果想加一个“同时跑单元测试和静态检查”的并行节点，改动成本很低。我实际对比过两种方案，用 Chain 硬凑循环的版本在第三轮修正之后基本就没法维护了，而 LangGraph 版本加到五六个节点依然清晰。

2.2 自我修正闭环的四个核心节点

整个 Agent 的核心就是四个节点构成的闭环：

• 生成节点（generate）：接收需求描述，调用大模型生成代码。第一轮是从零生成，后续轮次是根据修正意见做增量修改。
• 执行节点（execute）：把代码写到临时文件，调用编译器或解释器运行，捕获标准输出、标准错误和退出码。
• 诊断节点（diagnose）：分析执行结果。如果成功且测试通过，标记为完成；如果失败，提取关键错误信息，生成结构化的修正建议。
• 修正节点（fix）：根据诊断结果修改代码。这里不是简单地把错误信息拼到 prompt 里让模型重写，而是要带上原始代码、错误上下文和之前尝试过的修改记录，避免模型反复犯同一个错误。

这四个节点通过条件边连接：generate → execute → diagnose，diagnose 之后根据结果决定是走 fix → execute 的循环，还是直接结束。整个图的状态是一个字典，包含代码、执行结果、错误信息、修正历史、轮次计数等字段。

2.3 状态设计的关键取舍

状态字段的设计直接决定了 Agent 能不能正确运转。我一开始图省事，只存了code和error两个字段，结果发现几个问题：第一，模型在修正时看不到自己之前改过什么，容易来回横跳；第二，没有轮次计数就没法设置最大循环次数，遇到死循环直接卡死；第三，执行结果只存错误信息不够，有时候需要看完整输出才能判断是编译错误还是运行时错误。

后来我把状态扩展成了这样几个关键字段：code存当前代码，exec_result存完整的执行输出（包括 stdout、stderr、exit_code），diagnosis存诊断结论（是成功、编译错误还是测试失败），fix_history存每一轮的修改摘要，iteration存当前轮次。这里有个细节要注意：LangGraph 的状态合并默认是覆盖式的，如果你在某个节点只返回了部分字段，其他字段会保留原值，这个行为要理解清楚，否则容易出现状态不一致。

提示：状态字段不要存大模型的完整对话历史，那样会让状态体积膨胀得很快。只存结构化的结论和必要的上下文就够了，需要历史信息时从fix_history里取摘要。

3. 核心细节解析与实操要点

3.1 代码执行沙盒的安全边界

执行节点是整个 Agent 里风险最高的环节，因为你是在运行模型生成的代码。我见过有人直接把生成的代码在宿主机上exec()，这在演示环境里可能没事，但只要模型生成了一句删库或者死循环的代码，后果就很严重。合理的做法是至少做到以下几点：限制执行时间（用subprocess的timeout参数），限制可用的系统调用（用resource模块设置 CPU 和内存上限），把代码写到独立的临时目录里执行，避免污染工作区。

对于 Python 代码，我通常用subprocess.run调起一个独立的 Python 进程，而不是在当前进程里exec。这样即使代码里有sys.exit()或者段错误，也不会影响主进程。超时设置我一般给 10 秒，对于大多数算法题和工具函数来说足够了，超过这个时间基本可以判定是死循环。退出码、stdout、stderr 三个信息都要捕获，因为有些错误信息是打在 stdout 里的，只看 stderr 会漏掉。

3.2 诊断环节如何提取有效错误信号

诊断节点是很多人容易做糙的地方。直接把编译器输出原封不动丢给模型，效果往往不好，因为编译错误信息里有很多噪音，比如路径、行号、内部堆栈，模型容易被这些无关信息带偏。我的做法是做一层轻量解析：对于 Python 的SyntaxError和NameError，提取错误类型和出错行；对于测试失败，提取断言失败的具体表达式和期望值、实际值。

这里有个经验：错误信息里最有价值的是“错误类型 + 出错位置 + 相关变量”，把这三样提取出来，模型定位问题的准确率会明显提升。另外，如果同一类错误连续出现两轮以上，说明模型没有理解问题本质，这时候应该在诊断结论里加上“上一轮已尝试修复但未成功”的提示，引导模型换一个思路。

3.3 修正提示词的构造技巧

修正节点的 prompt 构造直接决定了修正效率。我试过几种写法，最后稳定下来的结构是这样的：先给原始需求，再给当前代码，然后给错误诊断，最后给修改要求。修改要求里明确写“只修改与错误相关的部分，不要重写整个文件”，这一句能显著减少模型把对的代码改错的情况。

还有一个细节：把fix_history里最近两轮的修改摘要带上，格式是“第 N 轮：修改了 X 函数，原因是 Y”。这样模型能看到自己之前的尝试路径，避免重复无效修改。但历史不要带太多，超过三轮的摘要反而会干扰判断，我一般只保留最近两轮。

注意：修正提示词里绝对不要出现“请确保代码正确”这种空泛要求，模型对这类指令基本免疫。要具体到“第 12 行的变量名拼写错误”或者“循环边界应该是range(n)而不是range(n-1)”这种可操作的层面。

4. 完整实操流程与关键环节实现

4.1 环境准备与依赖安装

先把基础环境搭起来。我用的 Python 版本是 3.11，LangGraph 和 LangChain 的版本迭代很快，建议锁定版本避免 API 变动导致代码跑不起来。

pip install langgraph==0.2.28 langchain==0.3.7 langchain-openai==0.2.8

如果你用的是其他模型提供商，把langchain-openai换成对应的包就行。环境变量里配好 API Key，这个不用多说。另外建议装一个black用于代码格式化，后面在生成节点里可以顺手做一次格式化，减少因为缩进、空格导致的低级语法错误。

4.2 定义状态结构与图骨架

先定义状态类型。LangGraph 推荐用TypedDict来定义状态，这样类型清晰，也方便后续扩展。

from typing import TypedDict, List, Optional class AgentState(TypedDict): requirement: str # 原始需求 code: str # 当前代码 exec_stdout: str # 执行标准输出 exec_stderr: str # 执行标准错误 exit_code: int # 退出码 diagnosis: str # 诊断结论 fix_history: List[str] # 修正历史摘要 iteration: int # 当前轮次 max_iteration: int # 最大轮次 status: str # running / success / failed

这里status字段很关键，它是条件边判断的依据。max_iteration放在状态里而不是写死在代码里，方便不同任务用不同上限。我一般设 5 轮，超过 5 轮还修不好，基本说明要么需求本身有问题，要么模型能力不够，继续循环也是浪费 token。

然后是图骨架：

from langgraph.graph import StateGraph, END workflow = StateGraph(AgentState) workflow.add_node("generate", generate_node) workflow.add_node("execute", execute_node) workflow.add_node("diagnose", diagnose_node) workflow.add_node("fix", fix_node) workflow.set_entry_point("generate") workflow.add_edge("generate", "execute") workflow.add_edge("execute", "diagnose") workflow.add_conditional_edges( "diagnose", route_after_diagnose, {"fix": "fix", "end": END} ) workflow.add_edge("fix", "execute") app = workflow.compile()

注意fix之后直接回到execute，而不是回到generate。这个设计是有意的：修正阶段只做增量修改，不需要重新理解需求，直接重新执行验证即可。如果修正后还是失败，诊断节点会再次判断，形成循环。

4.3 生成节点的实现细节

生成节点负责第一轮从零生成代码。prompt 里要明确输出格式，我要求模型只输出代码，不要带 markdown 代码块标记，不要带解释文字。这个约束很重要，否则你还得写额外的解析逻辑去剥离代码块标记。

def generate_node(state: AgentState) -> dict: prompt = f"""根据以下需求生成 Python 代码。 需求：{state['requirement']} 要求： 1. 只输出代码，不要任何解释 2. 代码要包含必要的 import 3. 如果是函数，函数名要符合需求描述 """ response = llm.invoke(prompt) code = response.content.strip() # 兜底：如果模型还是带了代码块标记，手动剥离 if code.startswith("```"): code = code.split("\n", 1)[1].rsplit("```", 1)[0] return { "code": code, "iteration": 1, "status": "running", "fix_history": [] }

这里返回的字典只包含需要更新的字段，其他字段保持原值。fix_history初始化为空列表，后续每轮修正往里追加。

4.4 执行节点的超时与资源控制

执行节点是整个流程里最需要小心的地方。我用subprocess调起独立进程，设置超时和资源限制。

import subprocess, tempfile, os, resource def execute_node(state: AgentState) -> dict: with tempfile.TemporaryDirectory() as tmpdir: filepath = os.path.join(tmpdir, "solution.py") with open(filepath, "w") as f: f.write(state["code"]) def limit_resources(): resource.setrlimit(resource.RLIMIT_CPU, (5, 5)) resource.setrlimit(resource.RLIMIT_AS, (256 * 1024 * 1024, 256 * 1024 * 1024)) try: result = subprocess.run( ["python", filepath], capture_output=True, text=True, timeout=10, preexec_fn=limit_resources ) return { "exec_stdout": result.stdout, "exec_stderr": result.stderr, "exit_code": result.returncode } except subprocess.TimeoutExpired: return { "exec_stdout": "", "exec_stderr": "执行超时：代码运行超过 10 秒，可能存在死循环", "exit_code": -1 }

RLIMIT_CPU限制 CPU 时间为 5 秒，RLIMIT_AS限制内存为 256MB。这两个限制配合timeout=10使用，基本能挡住大多数恶意或失控代码。注意preexec_fn在 Windows 上不可用，如果你在 Windows 上开发，需要换用job objects或者干脆用 Docker 容器来隔离执行环境。

4.5 诊断节点的错误分类逻辑

诊断节点要把执行结果翻译成结构化的结论。我把它分成三种情况：成功、编译/语法错误、运行时错误。每种情况的处理策略不同。

def diagnose_node(state: AgentState) -> dict: if state["exit_code"] == 0 and not state["exec_stderr"]: return {"diagnosis": "success", "status": "success"} stderr = state["exec_stderr"] if "SyntaxError" in stderr or "IndentationError" in stderr: diag = f"语法错误：{extract_error_line(stderr)}" elif "NameError" in stderr or "TypeError" in stderr: diag = f"运行时错误：{extract_error_line(stderr)}" elif "AssertionError" in stderr: diag = f"测试失败：{extract_assertion(stderr)}" else: diag = f"未知错误：{stderr[:500]}" return {"diagnosis": diag, "status": "running"}

extract_error_line和extract_assertion是两个辅助函数，负责从原始错误信息里提取关键行。这里不要偷懒直接截取前 500 字符，因为 Python 的 traceback 前面往往是一堆无关的调用栈，真正的错误信息在最后几行。我的做法是从后往前找第一个包含Error或assert的行。

4.6 修正节点的增量修改策略

修正节点的 prompt 构造是决定修正成功率的关键。我的模板是这样的：

def fix_node(state: AgentState) -> dict: history_text = "\n".join(state["fix_history"][-2:]) if state["fix_history"] else "无" prompt = f"""原始需求：{state['requirement']} 当前代码： {state['code']} 错误诊断：{state['diagnosis']} 之前的修改记录： {history_text} 请修复上述错误。要求： 1. 只修改与错误相关的部分，不要重写整个文件 2. 保持其他部分的逻辑不变 3. 只输出修改后的完整代码，不要解释 """ response = llm.invoke(prompt) new_code = response.content.strip() if new_code.startswith("```"): new_code = new_code.split("\n", 1)[1].rsplit("```", 1)[0] summary = f"第 {state['iteration']} 轮：{state['diagnosis'][:100]}" return { "code": new_code, "iteration": state["iteration"] + 1, "fix_history": state["fix_history"] + [summary] }

注意iteration在这里加一，而不是在执行节点加。这样轮次计数和修正次数是对应的，方便判断是否达到上限。

4.7 条件边的路由判断

路由函数决定诊断之后是继续修正还是结束：

def route_after_diagnose(state: AgentState) -> str: if state["status"] == "success": return "end" if state["iteration"] >= state["max_iteration"]: return "end" return "fix"

这里有个容易忽略的点：达到最大轮次时直接结束，但状态里的status还是running。你需要在图执行完之后检查最终状态，如果status不是success，说明任务失败，需要人工介入或者记录日志。我一般会在外层包一个函数，执行完图之后根据status决定是返回代码还是抛异常。

5. 常见问题与排查技巧实录

5.1 无限循环与状态污染

最常见的坑是 Agent 陷入无限循环。表现是iteration一直涨但status永远不是success，最后要么被max_iteration拦住，要么直接把 token 烧光。原因通常有两个：一是修正节点每次改的地方不对，模型在同一个错误上反复横跳；二是状态污染，比如fix_history没有正确追加，导致模型看不到之前的尝试。

排查方法很简单：把每一轮的diagnosis和code打印出来，看模型是不是在重复同样的修改。如果是，说明诊断信息不够具体，需要加强错误提取的粒度。我遇到过一次，模型连续三轮都在改同一个变量名，原因是诊断信息里只说了“NameError”，没说是哪个变量。后来在诊断里加上了具体变量名，一轮就修好了。

提示：在开发阶段把max_iteration设小一点（比如 3），方便快速验证流程。上线前再根据任务复杂度调整到 5 到 8。

5.2 执行超时与资源耗尽

执行节点超时是另一个高频问题。除了死循环，还有一种情况是代码里调用了需要网络请求的库，而沙盒环境没有网络，导致进程一直等待直到超时。我的处理方式是在执行前做一次静态检查，如果代码里出现了requests、urllib、socket这些网络相关的 import，直接在诊断阶段标记为“不允许网络操作”，不进入执行环节。

资源耗尽方面，内存限制设得太小会导致正常代码也跑不起来，设得太大又失去保护意义。256MB 对于大多数算法题和工具函数是够的，但如果你要处理大数据集，需要适当放宽。我的经验是先用一个宽松的限制跑通流程，然后逐步收紧，观察什么时候开始出现误杀，取那个临界值再往上浮 20%。

5.3 模型输出格式不稳定

即使 prompt 里明确要求“只输出代码”，模型有时候还是会带上解释文字或者 markdown 标记。这种情况在换模型或者调整 temperature 之后尤其明显。我的应对策略是写一个健壮的解析函数，按优先级依次尝试：先找 markdown 代码块标记，找到就提取块内内容；没有标记就找第一个import或def作为起点；如果都没有，就把整个输出当代码，但在诊断阶段会自然报语法错误，进入修正循环。

这个解析函数要放在生成节点和修正节点里共用，避免两处逻辑不一致。我一开始就是两处各写了一套，结果修正节点漏了 markdown 剥离逻辑，导致代码里带着 ``` 符号去执行，报了一堆莫名其妙的语法错误。

5.4 常见问题速查表

5.5 几个提升修正成功率的实操心得

第一，在诊断信息里加上出错行的上下文代码。比如错误是第 12 行的NameError，那就把第 10 到 14 行的代码一起给模型看。这样模型不需要自己去数行号，定位准确率会高很多。

第二，如果连续两轮修正都失败，第三轮换一个策略：让模型先解释错误原因，再给修改方案，最后输出代码。这种“先想后写”的方式对复杂错误特别有效，我实测能把修正成功率从 40% 左右提升到 70% 以上。

第三，对于测试失败的情况，把期望值和实际值都提取出来。模型看到assert result == 10失败，实际值是 8，它就能推断出是计算逻辑少了某个步骤，而不是盲目地改比较运算符。

6. 从单文件到多文件项目的扩展思路

上面这套流程跑通的是单文件场景，但实际项目里代码往往是多文件的。扩展的方向有几个：一是把状态里的code从字符串改成文件路径到内容的字典，执行节点负责把多个文件写到临时目录再运行；二是诊断节点要能定位到具体是哪个文件哪一行出错，这需要解析 traceback 里的文件路径；三是修正节点只修改出错的那个文件，其他文件保持不变。

另一个扩展方向是加入测试用例生成。现在的流程是执行代码看有没有报错，但代码不报错不代表逻辑正确。可以在生成节点之后加一个“生成测试”节点，让模型根据需求自动写几个测试用例，然后执行节点同时跑代码和测试，诊断节点根据测试通过率来判断是否成功。这个改动会让 Agent 的自我修正能力从“语法正确”提升到“逻辑正确”，实用性会强很多。

我在实际项目里还加了一个“代码审查”节点，用另一个模型实例对生成的代码做静态审查，检查是否有明显的逻辑漏洞或者边界条件遗漏。这个节点不参与循环，只在最终成功之后跑一次，作为质量兜底。多一个模型视角，能发现不少单模型自我修正时忽略的问题。

最后分享一个小技巧：把每一轮的代码和执行结果都持久化到本地文件，按iteration编号。这样当 Agent 最终失败时，你可以回溯整个修正过程，看看模型是在哪一步走偏的。这个记录对于调优 prompt 和诊断逻辑非常有价值，我靠这些记录定位了好几个隐蔽的 prompt 问题。