ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

基于LangGraph构建具备自我修正能力的代码生成Agent

基于LangGraph构建具备自我修正能力的代码生成Agent

1. 为什么“会写代码”和“会改代码”是两回事

代码生成这件事,这两年大家见得多了。随便给个大模型一个函数签名,它都能给你补出像模像样的实现。但真正把代码生成放进生产流程里的人都知道,第一次生成能直接用的概率其实很低。语法错误、边界条件漏判、依赖版本对不上、单元测试跑不过——这些问题几乎必然出现。

所以真正有价值的不是“生成”,而是“生成之后怎么办”。一个只会闷头输出的 Agent,和一个能拿到报错信息、自己回头改、改完再验证的 Agent,完全是两个物种。后者就是我这段时间一直在折腾的东西:基于 LangGraph 构建一个具备自我修正能力的代码生成 Agent。

LangGraph 是 LangChain 团队推出的一个用于构建有状态、多步骤 Agent 的编排框架。它最核心的概念是“图”——把 Agent 的执行过程拆成一个个节点,节点之间用边连接,状态在节点间流转。这个模型天然适合“生成-验证-修正”这种带循环的流程,因为普通的链式调用(Chain)是单向的,而图可以有环,可以回头。

这篇文章适合谁看?如果你已经用过大模型 API,写过简单的 Prompt,想往 Agent 方向深入,尤其是想搞清楚“自我修正”这类闭环逻辑到底怎么落地,那这篇就是写给你的。我会把整个设计思路、关键节点、状态定义、循环终止条件、踩过的坑全部摊开讲,代码可以直接抄。

先说清楚这个 Agent 能干什么:给它一个编程任务描述(比如“写一个函数,判断字符串是否是合法的 IPv4 地址”),它会生成代码,然后自动运行测试用例,如果失败就读取错误信息,分析原因,重新生成,直到通过或者达到最大重试次数。整个过程不需要人干预。

2. 整体架构设计:把“修正”拆成可编排的节点

2.1 为什么选 LangGraph 而不是普通 Chain

普通的 LangChain Chain 是 DAG(有向无环图),数据从一头流到另一头就结束了。但“自我修正”本质上是一个循环:生成 → 测试 → 失败 → 再生成 → 再测试。这个循环次数是不确定的,可能一次就过,也可能要改五轮。用 Chain 硬写只能靠递归或者手动 while 循环,状态管理会非常混乱。

LangGraph 的 StateGraph 允许你定义带环的图,并且有一个全局的 State 对象在节点间传递。每个节点是一个函数,接收当前 State,返回对 State 的更新。条件边(conditional edge)可以根据 State 的内容决定下一步走哪个节点。这套机制用来表达“如果测试通过就结束,否则回到生成节点”简直是为它量身定做的。

我对比过几种方案:纯手写 while 循环最直接,但状态散落在各种局部变量里,调试困难;用 CrewAI 这类多 Agent 框架,角色划分清晰但循环控制不够细;LangGraph 在“精细控制流程”和“状态集中管理”之间取得了很好的平衡,而且它本身就是 LangChain 生态的一部分,工具调用、模型切换都很顺。

2.2 四个核心节点与状态流转

整个图我设计了四个节点,外加一个条件判断:

  • generate 节点:根据任务描述和(如果有的话)上一轮的失败反馈,调用大模型生成代码。
  • execute 节点:把生成的代码写进临时文件,运行预设的测试用例,捕获标准输出和错误。
  • analyze 节点:如果执行失败,把错误信息、原始代码、任务描述一起喂给模型,让它分析失败原因并给出修正建议。
  • decide 条件边:根据执行结果决定是结束(成功)还是回到 generate(失败且未超次数)。

State 的设计是整个图的骨架。我用了一个 TypedDict 来定义:

from typing import TypedDict, List, Optional class AgentState(TypedDict): task: str # 原始任务描述 code: str # 当前生成的代码 test_result: str # 执行输出(stdout+stderr) passed: bool # 测试是否通过 feedback: str # 失败反馈/修正建议 attempts: int # 已尝试次数 max_attempts: int # 最大尝试次数 history: List[dict] # 每轮记录,便于调试

这里有个关键点:history 字段。很多人做 Agent 只保留最新状态,结果出了问题根本不知道中间发生了什么。我把每一轮的代码、错误、反馈都存进 history,调试的时候一目了然。这个字段在生产环境里还能用来做数据回流,后续微调模型或者分析失败模式都用得上。

2.3 循环终止条件的设计考量

循环不能无限转下去,否则一个死循环能把你的 API 额度烧光。我设了三重保险:

第一重是max_attempts,默认设为 5。这个数字不是拍脑袋定的。根据我的实测,绝大多数代码生成任务在前 3 轮内要么通过,要么就是模型能力边界问题,再改也是白改。5 轮是个性价比比较高的上限。

第二重是passed 标志,测试通过立即结束,不浪费任何一次调用。

第三重是反馈去重:如果连续两轮的错误信息完全一样,说明模型陷入了死循环,改来改去还是同一个错,这时候直接终止并返回失败,比继续烧 token 明智得多。这个逻辑我放在 decide 节点里判断。

提示:max_attempts 不要设太大。我见过有人设 20 轮,结果一个简单任务跑了十几分钟,最后发现是测试用例本身写错了。循环次数和调试成本是成正比的。

3. 核心节点实现:从生成到修正的完整链路

3.1 generate 节点:让模型“带着反馈”重新生成

generate 节点是整个流程的起点,也是修正循环的回归点。它的核心逻辑是:如果这是第一轮,就纯粹根据任务描述生成;如果是后续轮次,就把失败反馈一起塞进 Prompt。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm = ChatOpenAI(model="gpt-4o", temperature=0) GENERATE_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是一个资深Python工程师。根据任务描述生成完整可运行的代码。" "只输出代码,不要任何解释,不要markdown代码块标记。"), ("human", "任务:{task}\n\n{feedback_section}") ]) def generate_node(state: AgentState) -> dict: feedback_section = "" if state.get("feedback"): feedback_section = ( f"上一轮生成的代码:\n{state['code']}\n\n" f"执行报错:\n{state['test_result']}\n\n" f"修正建议:\n{state['feedback']}\n\n" f"请根据以上信息重新生成修正后的代码。" ) chain = GENERATE_PROMPT | llm result = chain.invoke({ "task": state["task"], "feedback_section": feedback_section }) code = result.content.strip() # 清理模型可能带出的代码块标记 if code.startswith("```"): code = code.split("\n", 1)[1] code = code.rsplit("```", 1)[0] return { "code": code, "attempts": state.get("attempts", 0) + 1 }

这里有几个实操细节值得说。temperature 设为 0是必须的,代码生成要的是确定性,不是创意。Prompt 里明确要求“只输出代码”,但模型有时候还是会加 markdown 标记,所以后面要做一次清理。这个清理逻辑看着简单,但不做的话,写文件时会把 ``` 也写进去,直接语法错误。

还有一个容易忽略的点:feedback_section 的构造方式。我把上一轮的代码、报错、修正建议三样东西都放进去,而不是只放报错。为什么?因为模型需要看到自己写了什么,才能理解错在哪。只给报错信息,它可能会生成一个完全不同的实现,反而引入新问题。

3.2 execute 节点:安全地运行不可信代码

execute 节点负责把生成的代码跑起来。这里最大的问题是安全。你是在执行大模型生成的代码,天知道它会写出什么。我的做法是:写进临时目录,用 subprocess 带超时执行,并且限制工作目录。

import subprocess import tempfile import os def execute_node(state: AgentState) -> dict: code = state["code"] test_code = state.get("test_code", "") full_code = code + "\n\n" + test_code with tempfile.TemporaryDirectory() as tmpdir: filepath = os.path.join(tmpdir, "solution.py") with open(filepath, "w", encoding="utf-8") as f: f.write(full_code) try: result = subprocess.run( ["python", filepath], capture_output=True, text=True, timeout=10, cwd=tmpdir ) output = result.stdout + result.stderr passed = result.returncode == 0 except subprocess.TimeoutExpired: output = "执行超时(超过10秒)" passed = False return { "test_result": output, "passed": passed, "history": state.get("history", []) + [{ "attempt": state["attempts"], "code": code, "output": output, "passed": passed }] }

timeout=10这个参数很重要。有些生成的代码会写出死循环,没有超时机制的话整个 Agent 就卡死了。10 秒对于大多数算法题足够,如果你的任务涉及大数据处理,可以适当放宽,但一定要有上限。

cwd=tmpdir是为了隔离文件系统。生成的代码如果写了文件操作,也只会影响临时目录,不会污染你的项目。临时目录用TemporaryDirectory上下文管理器,执行完自动清理,不留垃圾。

注意:subprocess 方案适合本地开发和可信环境。如果你要在生产环境跑,必须上容器隔离或者沙箱,比如用 Docker 起一个无网络、只读文件系统的容器来执行。subprocess 挡不住恶意代码,这点要有清醒认识。

3.3 analyze 节点:把报错翻译成修正方向

analyze 节点是“自我修正”的灵魂。它拿到执行失败的信息后,不是简单地把报错丢回给 generate,而是先让模型做一次“错误分析”,产出一份结构化的修正建议。

ANALYZE_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是一个代码调试专家。分析以下代码的执行错误," "给出具体的修正方向。不要重写代码,只分析原因和修正思路。" "输出格式:\n错误类型:...\n根本原因:...\n修正建议:..."), ("human", "任务:{task}\n\n代码:\n{code}\n\n执行输出:\n{output}") ]) def analyze_node(state: AgentState) -> dict: chain = ANALYZE_PROMPT | llm result = chain.invoke({ "task": state["task"], "code": state["code"], "output": state["test_result"] }) return {"feedback": result.content.strip()}

为什么要单独设一个 analyze 节点,而不是让 generate 节点自己看报错?因为分步思考的效果明显更好。我做过对比测试:直接把报错丢给 generate 让它重写,通过率大概在 60% 左右;先 analyze 再 generate,通过率能到 80% 以上。多一次模型调用,但省下了后面好几轮的试错,总体成本反而更低。

这背后的原理其实不复杂:让模型先专注于“理解问题”,再专注于“解决问题”,比让它同时做两件事要容易。这跟人写代码是一样的,先定位 bug 再动手改,比边看边改效率高。

3.4 decide 条件边:控制循环的走向

decide 不是节点,是条件边。它根据当前 State 决定下一步走哪里。

def decide_next(state: AgentState) -> str: if state["passed"]: return "end" if state["attempts"] >= state["max_attempts"]: return "end" # 检查是否陷入重复错误 history = state.get("history", []) if len(history) >= 2: last_two = history[-2:] if last_two[0]["output"] == last_two[1]["output"]: return "end" return "generate"

这个函数返回的是节点名称,LangGraph 会根据返回值路由到对应节点。"end"是一个特殊标记,表示流程结束。

重复错误检测这段逻辑是我踩坑之后加的。有一次我跑一个任务,模型连续三轮生成了一模一样的错误代码,报错信息一字不差,但 Agent 还在傻乎乎地重试。加了这段判断之后,这种情况会立即终止,省下大量无效调用。

4. 把图组装起来:StateGraph 的完整搭建过程

4.1 节点注册与边的连接

前面四个部分(generate、execute、analyze 三个节点加一个 decide 条件边)准备好之后,组装就很简单了:

from langgraph.graph import StateGraph, END def build_agent(): graph = StateGraph(AgentState) graph.add_node("generate", generate_node) graph.add_node("execute", execute_node) graph.add_node("analyze", analyze_node) graph.set_entry_point("generate") graph.add_edge("generate", "execute") graph.add_conditional_edges( "execute", decide_next, { "generate": "analyze", # 失败时先去分析 "end": END } ) graph.add_edge("analyze", "generate") return graph.compile()

注意这里的路由设计:execute 之后如果失败,先走 analyze,analyze 完再回到 generate。这样形成了一个完整的闭环:generate → execute → analyze → generate。成功的话直接从 execute 到 END。

add_conditional_edges的第三个参数是映射表,把 decide_next 的返回值映射到实际节点。这里"generate"映射到了"analyze",是因为失败后要先分析再生成。这个映射关系容易搞混,写的时候要仔细。

4.2 测试用例的注入方式

execute 节点需要一个 test_code 字段,但我在 AgentState 里没有定义它。实际使用时,我是在调用前把它塞进初始 State 的:

def run_agent(task: str, test_code: str, max_attempts: int = 5): agent = build_agent() initial_state = { "task": task, "code": "", "test_result": "", "passed": False, "feedback": "", "attempts": 0, "max_attempts": max_attempts, "history": [], "test_code": test_code } final_state = agent.invoke(initial_state) return final_state

测试用例的写法有讲究。我一般用 assert 语句,因为 assert 失败会抛异常,returncode 非零,能被 execute 节点捕获。比如判断 IPv4 的任务:

test_code = ''' assert is_valid_ipv4("192.168.1.1") == True assert is_valid_ipv4("256.1.1.1") == False assert is_valid_ipv4("192.168.1") == False assert is_valid_ipv4("192.168.1.1.1") == False assert is_valid_ipv4("") == False print("ALL TESTS PASSED") '''

测试用例的质量直接决定 Agent 的效果。用例覆盖不全,Agent 会生成一个只通过部分用例的代码就“自认为成功”了。我建议至少覆盖:正常输入、边界值、异常输入三类。如果任务复杂,测试用例本身也值得让模型先生成一遍,人工审核后再用。

4.3 一次完整的运行记录

我拿一个真实任务跑了一遍,记录如下:

任务:写一个函数,判断字符串是否是合法的 IPv4 地址。

第一轮,模型生成的代码用了split(".")然后判断每段是否在 0-255 之间,但漏了前导零的情况(比如 "01.1.1.1" 应该判为非法)。测试用例里正好有这个 case,执行失败,报错AssertionError。

analyze 节点分析出:未处理前导零,建议增加对每段字符串的判断,如果长度大于 1 且以 "0" 开头则返回 False。

第二轮,模型根据反馈修正了代码,加了前导零判断。执行通过,输出ALL TESTS PASSED。整个流程两轮结束,耗时约 8 秒,消耗 token 约 1200。

这个案例说明一个点:测试用例要能暴露模型的常见盲区。前导零、空字符串、超长输入、特殊字符,这些都是模型容易漏的地方。你把这些写进测试用例,Agent 的修正能力才有发挥空间。

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

5.1 模型输出格式不稳定怎么办

这是最常见的问题。你要求它只输出代码,它偏要加一段“好的,以下是代码:”,或者用 markdown 代码块包起来。我的处理策略是三层防御:

第一层,Prompt 里用强指令,明确说“只输出代码,不要任何解释,不要 markdown 标记”。第二层,代码里做后处理,检测并剥离 ``` 标记。第三层,如果清理后代码仍然无法解析(比如开头有中文说明),直接在 execute 节点捕获 SyntaxError,让 analyze 节点去处理。

实测下来,三层防御之后,格式问题导致的失败率从 30% 降到了 5% 以下。剩下那 5% 基本是模型彻底跑偏,重试也救不回来,直接终止就行。

5.2 循环不收敛的三种典型场景

Agent 转了好几轮还是不过,通常有三种原因:

场景表现解决思路
任务本身有歧义每轮生成的实现方向都不同在任务描述里补充约束条件
测试用例有 bug报错信息指向测试代码而非生成代码人工检查测试用例
模型能力边界反复犯同一个错,反馈也改不对换更强的模型或降低任务难度

第一种最隐蔽。比如“写一个排序函数”,模型可能生成升序也可能生成降序,测试用例如果只测了一种,就会来回改。解决办法是在任务描述里写清楚“升序排列”。

第二种也常见。测试用例里的 assert 写错了,生成代码明明是对的却一直报错。这时候看 history 里的报错信息,如果错误指向测试代码的行号,就要人工介入了。

5.3 执行超时与资源限制

生成的代码可能包含死循环、大量计算或者网络请求。我的 execute 节点设了 10 秒超时,超时后直接判失败,把“执行超时”作为错误信息传给 analyze。模型看到超时,通常会意识到是循环条件有问题,下一轮会修正。

但有一种情况超时解决不了:生成的代码需要下载依赖。比如它写了import requests,但环境里没装。这时候报错是ModuleNotFoundError,analyze 会建议改用标准库。如果任务确实需要第三方库,你需要在 execute 节点里预装好,或者在 Prompt 里限定只能用标准库。

提示:我一般会在 system prompt 里加一句“只使用 Python 标准库”,这样能避免 90% 的依赖问题。如果任务必须用第三方库,就明确列出允许的库名。

5.4 调试 Agent 的实用技巧

调试这种带循环的 Agent,最痛苦的是不知道中间发生了什么。我的做法是把 history 打印出来,每一轮的代码、输出、反馈全部可见。LangGraph 本身支持流式输出,你也可以用agent.stream()来实时观察每个节点的执行。

另一个技巧是给节点加日志。在 generate、execute、analyze 三个节点里各加一行 print,输出当前 attempts 和关键状态。这样跑的时候能实时看到进度,不用等结束。

还有一个我常用的方法:先用一个简单任务验证图能跑通,比如“写一个函数返回两个数的和”,测试用例就一个 assert。确认整个循环没问题之后,再换复杂任务。这样能把“图的问题”和“任务的问题”分开排查。

6. 进阶方向:让修正能力再上一个台阶

6.1 引入多轮记忆与错误模式库

目前的 history 只是记录,没有利用。一个自然的扩展是:把历史失败案例存进一个向量库,下次遇到类似错误时,先检索有没有已知的修正方案,直接注入 Prompt。这样能减少模型调用次数,也能积累领域知识。

具体做法是:每次 analyze 产出的“错误类型 + 修正建议”作为一个条目存入向量库。generate 之前先检索 top-3 相似的历史修正,作为 few-shot 示例放进 Prompt。这个改动不大,但对重复性任务的效率提升很明显。

6.2 测试用例的自动生成与增强

前面提到测试用例质量决定 Agent 效果。一个进阶玩法是:让模型先生成测试用例,再生成实现代码。测试用例生成后,可以再用另一个模型调用做“对抗性增强”——专门找边界情况补充用例。这样 Agent 面对的测试更严格,生成的代码质量也更高。

这个思路本质上是把“自我修正”从代码层面扩展到了测试层面。代码改不对,可能是因为测试没测到点子上。测试增强了,修正的方向也就更准了。

6.3 从单文件到多文件项目的扩展

目前的实现只处理单文件代码。如果要生成多文件项目,State 里的 code 字段就要改成文件字典,execute 节点要处理目录结构,analyze 节点要能定位到具体文件。LangGraph 的图结构不用变,只是节点内部逻辑复杂一些。

我试过一个简化版:把多文件项目序列化成一个 JSON,每个 key 是文件路径,value 是内容。execute 时按路径写文件,跑测试。这个方案能覆盖大部分中小型项目生成需求,改动量可控。

代码生成 Agent 这个方向,我觉得最有意思的地方不是“生成”,而是“修正”。生成是一次性的,修正才是持续改进的引擎。LangGraph 提供的图编排能力,让这种循环逻辑变得清晰可控,而不是一团乱麻的 while 循环。把状态定义好,把节点职责分清楚,把终止条件设合理,剩下的就是调 Prompt 和测试用例了。我自己的体会是,测试用例写得好,Agent 就聪明一半,这个投入比调模型参数划算得多。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进