LangGraph 这几年在 Agent 开发里出镜率越来越高。我一直坚持一个判断：一个 Agent 能不能落地，不是看它会不会聊天，而是看它能不能闭环解决实际问题。今天要聊的项目，就是一个基于LangGraph构建的“自我修正”代码生成 Agent：它接收自然语言描述的任务，输出对应代码，然后在真实环境里执行测试，一旦发现 bug，就带着失败信息自动进入修复循环，直到测试通过或达到上限。它把常见的“LLM 写代码—人来改 bug”变成“Agent 写代码—Agent 自己修”，适合正在从链式调用转向图编排的 Agent 开发者，也适合想给自己项目加一个“自动改代码”能力的团队。

严格说，这不是一个新鲜概念，但用 LangGraph 来实现，和以前用 LangChain 链式脚本堆出来的东西完全不是一回事。下面我会从设计思路、状态模型、核心代码到踩坑实录，完整走一遍这个项目。

1. 为什么“自我修正”在代码生成 Agent 里是刚需

1.1 一次性生成的代码为什么不靠谱

很多人第一次让大模型写代码，都会被惊艳到：需求描述清楚，它能直接吐出一个像模像样的函数。但一旦把代码装进真实项目里跑，问题就来了。模型对语法结构把握得还行，但对 API 的拼写、参数的默认值、依赖库的实际行为，经常是“编”出来的。比如它可能生成一个pd.DataFrame.plot()，却传入一个该版本根本不存在的参数；也可能调用某个工具函数时，把返回类型当成另一个类型处理。

更隐蔽的是逻辑错误。代码能运行，但结果不对。边界条件没考虑、浮点精度被忽略、空列表直接取下标……这些问题靠人眼 review 不一定看得出来，但测试用例一跑就现原形。

所以“一次性生成可靠代码”这件事，在模型能力没有质的飞跃之前，基本是伪命题。现实可行的路径是：把代码放进自动校验环境里，用真实执行结果反哺模型修正。这就是自我修正闭环的出发点。

1.2 自我修正闭环的价值

自我修正并不是“让模型多生成几次碰运气”，而是把执行反馈作为新一轮生成的硬约束。

一个最小闭环长这样：

1. 生成候选代码。
2. 编写或附带测试用例。
3. 在受控环境里执行测试，收集 stdout、stderr、异常 traceback。
4. 把失败信息交给模型，让它针对性修复。
5. 重复执行，直到测试通过或达到最大尝试次数。

这里的核心是第 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 不直接写代码，它先想清楚“要做什么、边界在哪、怎么测”。这一步能显著减少 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 终止条件与防死循环策略

自我修正最怕的就是“修到天荒地老”。我在条件边里同时做了三重保险：

1. 硬性轮数上限：attempts >= MAX_ATTEMPTS时强制结束。
2. 测试通过：反馈里出现明确的通过标记，立即结束。
3. 无进展检测：如果连续两轮返回的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 Agent

3.1 环境准备与依赖

我用的环境是 Python 3.10+，核心依赖就这么几个：

pip install langgraph langchain-openai openai

如果你用本地模型或者其它厂商接口，只需要替换 LLM 封装层。我这里用langchain-openai，因为它的接口足够标准，换模型时改动最小。

还要准备一个 API Key。建议环境变量方式，不要硬编码进代码：

export OPENAI_API_KEY="你的key"

3.2 核心代码：节点、条件边与编译

先定义一个统一的 LLM 调用入口。我习惯把模型调用单独封装，方便后续替换模型或加缓存：

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.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"] == 0：

def 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", encoding="utf-8") as f: f.write(state["code"]) with open(test_path, "w", encoding="utf-8") as f: f.write(state["test_code"]) try: result = subprocess.run( [sys.executable, "-m", "pytest", test_path, "-q", "--tb=short"], capture_output=True, text=True, timeout=30 ) 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(checkpointer=checkpointer)

之后每次调用都带上同一个thread_id，Agent 就能从上次中断的地方继续执行：

config = {"configurable": {"thread_id": "task-fib-001"}} result = agent.invoke(initial_state, config)

实际值是：如果某一次测试运行到一半因为外部原因中断，比如网络抖动、超时，你不需要从头开始；再次调用同一个thread_id，LangGraph 会从上一次保存的状态继续。这在接人工审核、故障恢复场景里非常有用。

生产环境建议用SqliteSaver或对应数据库实现，内存版只适合开发调试。

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

4.1 循环不终止或无限修复

最常见的原因有三类：should_continue的返回值没有出现在边字典里；attempts没累加；判断“通过”的逻辑写错，导致明明已经 PASSED 还继续修。

排查顺序：先在条件边函数里把state["feedback"]和attempts打印出来，确认实际状态值；再用agent.get_graph().draw_mermaid()或直接打印图结构，确认边的名称匹配。

我早期在一个项目里把返回字符串写成"fix"但边字典里写成了{"retry": "coder"}，排查了很久才发现是拼写不一致。图框架不会帮你校验这个，全靠自己细心。

4.2 上下文被历史消息撑爆

只要跑过三轮以上，messages里塞的完整代码和日志就会非常夸张。我建议：

1. feedback只保留最近一轮，不要追加到长历史里。
2. messages列表加一个上限，比如保留最近 10 条，超过就把最早的消息移除。
3. 如果日志太长，先用 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 消耗，优化时有据可依。

最后说点我自己的体会。这套架构第一个版本我做得非常糙，Planner、Coder、Tester 全让一个节点干，结果失败信息一多，模型开始自我催眠，前几轮还能定位问题，后面就只会把报错复制一遍。把职责拆开之后，每轮的 prompt 都变得很短、很聚焦，修复率明显上去了。还有一个小技巧：tester 节点里记得把 stdout 和 stderr 原样记录，尤其是 traceback 的最后两三行，一句话都不要剪，模型对异常日志的依赖比我们想象中高很多。目前我又在这条路上加了两样东西：一个是静态检查工具比如 ruff 扫描作为额外的 feedback 来源，另一个是把已修复的样本存下来做 few-shot。如果你也在折腾 Agent，欢迎拿这套骨架改自己的版本。