我最近把一个代码生成 Agent 从 LangChain 的 AgentExecutor 迁移到了 LangGraph前后折腾了大概两周。最大的感受是以前靠 prompt 硬撑的自我修正现在变成了图里一条实打实的循环边运行逻辑一眼就能看穿。这篇是我在真实项目里跑通生成代码→执行测试→识别错误→反思修复整个闭环之后的完整记录包含可复现的代码、踩过的坑和取舍过程。如果你正准备写一个代码生成 Agent或者已经受够了工具调用循环黑盒到没法干预的旧写法这篇文章应该能帮你省下不少弯路。它适合两类人一类是想用 LangGraph 做复杂 Agent 编排的开发者另一类是被模型直接生成代码、但经常跑不通折磨的工程师。我会尽量把原理讲明白也把完整的实操代码放出来你照着抄就能跑起来。这个 Agent 能做什么给它一个自然语言任务比如写一个函数输入整数 n返回斐波那契数列前 n 项它会自动生成 Python 代码、自动执行测试如果测试失败就读取错误信息重新修复代码如此迭代直到通过或达到最大尝试次数。它不是一个只会写代码的机器人而是一个会写完再验证、验证不过就改的闭环系统。1. 为什么选 LangGraph从链式调用到可编排的有向图1.1 代码生成 Agent 的三个核心痛点先聊一下背景。单纯的 LLM 生成代码第一次能跑通的比例其实不高。我实测过一批常见的编程题直接把生成代码和执行测试串成一条链通过率大概在五成到七成之间任务越复杂成功率越低。真正让代码生成 Agent 可用靠的不是让模型一次性写对而是给它一个试错—反馈—修正的循环让它在几轮迭代里自己把错误磨平。但做这个循环有三个绕不开的坑。第一个是状态管理。每一轮迭代都要记录当前代码、上一次测试输出、已经尝试了几次、历史修改过哪些版本这些状态散落在代码里很容易乱尤其当你想支持并发调用的时候一个全局变量写进去就是事故现场。第二个是控制流。自我修正不是一个线性的链它是一个生成→执行→判断→再生成的循环循环什么时候退出、出错多少次放弃这些逻辑在普通代码里会变成一堆 if-else 和 while 嵌套维护起来很痛苦。第三个是可观测性。线上 Agent 出问题的时候你要能定位它是在哪一步失败的、模型当时收到了什么上下文如果整个流程是黑盒排查起来就是大海捞针。1.2 LangGraph 的核心设计状态、节点、边LangGraph 解决上面三个问题的方式很直接它把整个流程建模成一张有向图图上有节点Node和边Edge数据在节点之间流动时统一放进一个**共享状态State里。节点就是你的业务函数边决定节点之间的跳转顺序而是否继续修正这种决策可以挂在条件边Conditional Edge**上让一个判断函数来决定下一个进入哪个节点。这个模型和我们熟悉的 React 状态管理有几分相似所有状态集中维护更新方式通过 Reducer 定义节点只负责消费和产生新的状态片段不关心全局发生了什么。好处是每个节点都是纯函数式的方便单独测试状态的变化有明确轨迹方便 debug多个 Agent 实例之间只要初始化不同的状态对象天然就能做到会话隔离。LangGraph 还有一个我很看重的能力Checkpointer记忆检查点。它可以把每一步执行后的状态持久化下来这意味着你可以在任意一轮迭代之后恢复执行、重放历史、甚至手动干预某一轮的输出再继续跑。很多团队拿它做 Agent 的时光回溯对于代码生成这种多轮修正场景特别有用——可以在某次生成的代码出错时回退到之前那个没出错的版本。1.3 自我修正在图上长什么样把自我修正翻译成图语言最简单的形式就是一条带环的执行路径生成节点 → 执行节点 → 判断节点 →通过→ 收尾节点 → END判断节点 →失败且未超次数→ 回到生成节点生成节点接收任务和上一轮的报错信息产出新代码执行节点在隔离环境里运行这段代码捕获输出和异常判断节点检查执行结果决定是继续修正还是直接收尾。整个闭环里最有价值的设计就是这个判断节点它把要不要继续修这个模糊的语义变成了一个可编程的布尔判断你可以根据退出码、错误类型、尝试次数来自由控制收敛策略。2. 系统架构与状态设计把写码-运行-改错变成一张图2.1 节点与边的总体规划我最终采用的方案是四个节点、一条条件边。先看全貌节点职责输入重点输出重点generate_node调用 LLM 生成或修正代码任务描述、上一轮代码、报错信息新代码、尝试次数1execute_node在隔离环境执行代码完整代码stdout、stderr、returncodedecide_route判断继续还是结束执行结果、尝试次数路由决策finalize_node产出最终答案代码、执行结果最终回复边的布局是这样的START 进入 generate_nodegenerate_node 完成后进入 execute_nodeexecute_node 完成后进入 decide_route。decide_route 是一条条件边返回finalize时进入 finalize_node返回generate时回到 generate_node。finalize_node 执行完就到达 END。你可能注意到我少画了一个用户输入节点因为 LangGraph 的 State 初始化时就可以把任务写进去不需要单独节点。2.2 状态数据结构设计状态是整个图的总线结构设计直接影响后续扩展。我用 TypedDict 来定义字段里有几个值得多讲两句from typing import TypedDict, Annotated, List, Optional import operator class CodeState(TypedDict): task: str # 用户任务描述全程不变 code: str # 当前最新版代码 stdout: str # 最近一次执行的标准输出 stderr: str # 最近一次执行的标准错误 returncode: int # 最近一次执行的退出码 attempts: int # 已经尝试生成/修正的次数 max_attempts: int # 允许的最大尝试次数 history: Annotated[List[dict], operator.add] # 完整迭代历史 final_answer: str # 最终输出给用户的内容其中 history 加了一个Annotated[List[dict], operator.add]的标注这告诉 LangGraph这个字段的新值要和旧值拼接而不是整体覆盖。简单说operator.add就是一个 Reducer 函数它定义了状态更新的合并方式。我不把历史放在一个普通列表里手动 append是因为 LangGraph 在每次节点返回后都会做状态合并用 Reducer 声明之后节点只需要返回新增的部分合并逻辑由框架保证。attempts和max_attempts是控制循环收敛的关键参数。max_attempts我在初始化时根据任务复杂度设置简单任务 3 次复杂任务 5 次太多次数会让响应时间变得不可接受后面我会单独讲这个参数怎么调。2.3 条件边设计用什么标准判断修好了decide_route 是整个自我修正逻辑的核心裁判。我的判断逻辑很简单优先级从高到低如果returncode 0说明代码执行成功直接进入 finalize。如果attempts max_attempts说明已经修了太多次还没好放弃进入 finalize把最后一次的错误信息如实返回。否则回到 generate_node 继续修。用代码表示就是这样from typing import Literal def decide_route(state: CodeState) - Literal[generate, finalize]: if state[returncode] 0: return finalize if state[attempts] state[max_attempts]: return finalize return generate这里有一个容易被忽略的细节returncode 为 0 不代表结果一定正确只是代表代码没有崩溃。比如我要生成一个排序函数模型写了个每次都返回空列表的代码它也能 returncode 0。所以更严格的场景我会在 execute_node 里嵌入一段用户提供的断言检查脚本执行的是被测函数 断言当断言失败时 returncode 非 0。简单说判断依据不应该是程序跑没跑起来而是程序跑完之后结果符不符合预期。3. 完整代码实现从零搭起一个能自我修正的 Agent3.1 环境准备与依赖安装我的运行环境是 Python 3.10核心依赖如下。版本号是大版本具体小版本建议装最新的LangGraph 更新很快API 有细微变化。pip install langgraph langchain langchain-openai python-dotenv如果你要执行的不是 Python 而是 JS 代码还需要 node。接下来把所有示例代码放进一个文件code_agent.py你可以直接跑通。3.2 工具函数安全的代码执行器代码执行是自我修正循环的基石也是安全风险最大的环节。生产环境我的建议是用 Docker 沙箱把每次执行放在一个容器里限制 CPU、内存、网络和文件系统跑完直接销毁容器。本文为了让大家能快速跑通先用子进程加超时控制的方式做一个简化版执行器原理是一样的隔离进程 限制资源 控制超时。import subprocess import tempfile import os import textwrap def execute_python(code: str, timeout: int 15) - dict: 在独立子进程中执行Python代码并返回执行结果。 - 通过 subprocess 隔离执行环境 - tempfile 保证临时文件不污染当前目录 - timeout 防止模型生成的死循环代码 wrapped_code textwrap.dedent(code).strip() with tempfile.NamedTemporaryFile( modew, suffix.py, deleteFalse, encodingutf-8 ) as tmp: tmp.write(wrapped_code) tmp.flush() tmp_path tmp.name try: proc subprocess.run( [ python, -u, tmp_path ], capture_outputTrue, textTrue, timeouttimeout, cwdos.path.dirname(tmp_path), env{**os.environ, PYTHONIOENCODING: utf-8} ) return { stdout: proc.stdout[-4000:], stderr: proc.stderr[-4000:], returncode: proc.returncode, } except subprocess.TimeoutExpired: return { stdout: , stderr: f执行超时超过 {timeout} 秒, returncode: -1, } finally: os.unlink(tmp_path)这里有几个细节。textwrap.dedent会去掉模型输出代码里多余的缩进避免模型在回复里用 markdown 代码块包裹时产生的缩进错误。proc.stdout[-4000:]截断输出是因为模型生成的代码里如果 print 了大量数据直接全部塞回 prompt 会把上下文撑爆尤其当错误信息本身很长时这个截断能让修复循环保持稳定。timeout参数建议从 10 秒起步给简单任务用够了复杂任务可以到 30 秒但不宜太长否则用户体验直线下降。3.3 定义图结构和节点函数接下来是核心部分。状态定义我已经在第 2.2 节给出了直接进入节点实现。generate_node让模型产出代码from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage SYSTEM_PROMPT 你是一名资深Python工程师。根据用户需求生成或修正Python代码。 要求 1. 只输出纯代码禁止使用markdown代码块包裹。 2. 代码必须包含函数定义和必要的测试调用保证直接运行不报错。 3. 如果提供了历史错误信息请根据错误对症下药不要重复同样的错误。 4. 输出完整代码不是补丁不是diff。 def generate_node(state: CodeState) - dict: llm ChatOpenAI(modelgpt-4o-mini, temperature0.1) prompts [SystemMessage(contentSYSTEM_PROMPT)] user_content f任务{state[task]}\n if state[history]: user_content \n历史迭代记录\n for item in state[history][-2:]: # 只保留最近两轮 user_content f第{item[attempt]}轮\n代码\n{item[code]}\n user_content f错误信息\n{item[stderr] or item[stdout]}\n\n else: user_content 这是第一次生成请直接输出能解决任务的最优代码。\n prompts.append(HumanMessage(contentuser_content)) response llm.invoke(prompts) raw_code response.content.strip() # 清理意外的markdown代码块标记 if raw_code.startswith(): lines raw_code.splitlines() lines lines[1:] if lines and lines[-1].strip() : lines lines[:-1] raw_code \n.join(lines).strip() history_item { attempt: state[attempts], code: raw_code, stderr: state[stderr], stdout: state[stdout], } return { code: raw_code, attempts: state[attempts] 1, history: [history_item], }这个节点里我把temperature调到 0.1对代码生成场景非常关键。代码生成不像写诗需要的是稳定性和确定性温度越高越容易出现这轮修好了、下一轮又改坏了的震荡。如果你对任务有把握也可以试试 0但要留意有些模型在 temperature0 时反而会陷入重复输出。history[-2:]是另一个经过实测有效的取舍。完整历史对模型有参考价值但轮次多了之后 token 消耗太大而且最新一轮的错误信息往往足以定位问题保留最近两轮足够。这个值你可以调成 3 或 5但总体原则是历史不是越多越好。execute_node跑代码拿结果def execute_node(state: CodeState) - dict: result execute_python(state[code]) return { stdout: result[stdout], stderr: result[stderr], returncode: result[returncode], }这个节点简单到只有三行但它是自我修正闭环里最重要的事实来源。模型说什么不重要跑出来的 returncode 才算数。finalize_node收尾总结def finalize_node(state: CodeState) - dict: if state[returncode] 0: answer ( f已经生成并验证通过。尝试次数{state[attempts]}。\n\n fpython\n{state[code]}\n\n f运行输出\n{state[stdout][-1000:]} ) else: answer ( f代码经过 {state[attempts]} 次尝试仍未通过验证。\n f最后一次错误\n{state[stderr][-1000:]} ) return {final_answer: answer}编译成图并跑起来from langgraph.graph import StateGraph, START, END def build_graph(): g StateGraph(CodeState) g.add_node(generate, generate_node) g.add_node(execute, execute_node) g.add_node(finalize, finalize_node) g.add_edge(START, generate) g.add_edge(generate, execute) g.add_conditional_edges( execute, decide_route, {generate: generate, finalize: finalize}, ) g.add_edge(finalize, END) return g.compile()add_conditional_edges的第一个参数是从哪个节点出发第二个参数是路由函数第三个参数是路由函数返回值到目标节点的映射。这个映射最大意义是让图的调用关系显式化你一眼能看出 generate 是从 execute 被决策过去的不是无条件过去的。3.4 跑一次完整流程看看状态怎么流转我用一个真实任务来验证让 Agent 写一个输入整数 n返回第 n 个斐波那契数的函数。def main(): app build_graph() initial_state { task: 写一个Python函数fib(n)返回第n个斐波那契数n从0开始fib(0)0fib(1)1。 在if __name__ __main__:中调用fib(10)并打印结果期望输出55。, code: , stdout: , stderr: , returncode: -1, attempts: 0, max_attempts: 4, history: [], final_answer: , } result app.invoke(initial_state) print(result[final_answer])第一次运行时模型给了一个递归解法def fib(n): if n 1: return n return fib(n-1) fib(n-2) if __name__ __main__: print(fib(10))执行节点返回 returncode 0stdout 是55decide_route 直接走到 finalize一次通过。说明闭环没问题。但作为压力测试我又让它生成一个用正则表达式提取HTML中所有链接的函数并在测试里故意用了 HTML 特殊字符第一次生成的代码由于转义不完整抛出了re.error: bad escape这时你看状态流转就能体会到自我修正的价值generate_node 收到stderr里的bad escape后第二轮生成的代码把字符串改成了 raw string执行通过。4. 进阶实战用 FastAPI 把 Agent 包装成可用服务4.1 从同步调用切换到异步调用本地跑通只是第一步。把 Agent 暴露成 HTTP 服务绕不开两个问题并发和流式。LangGraph 的invoke是同步的在 FastAPI 的异步线程池里也能跑但如果你想同时处理几十个请求纯同步调用会占满线程池。LangGraph 提供了ainvoke异步方法配合 FastAPI 的async def接口就能做到真正的异步处理。用法很简单把app.invoke(initial_state)换成await app.ainvoke(initial_state)即可。注意ainvoke返回的也是最终状态 dictAPI 面没有变化。4.2 会话隔离与并发控制的正确姿势这里有个关键陷阱不要每次都build_graph()新建一个编译图。图结构编译一次就够了编译后的对象是可复用的LangGraph 内部会通过 checkpointer 来隔离不同会话的上下文。你需要的是一个 checkpointer 实例然后在每个请求里传不同的thread_id。from langgraph.checkpoint.memory import MemorySaver from fastapi import FastAPI from pydantic import BaseModel import asyncio app FastAPI() checkpointer MemorySaver() graph_app build_graph_with_checkpointer(checkpointer) semaphore asyncio.Semaphore(10) class CodeRequest(BaseModel): task: str max_attempts: int 3 app.post(/generate) async def generate_code(req: CodeRequest): async with semaphore: # 用uuid作为thread_id实现会话隔离 config {configurable: {thread_id: freq-{uuid.uuid4()}}} result await graph_app.ainvoke( { task: req.task, code: , stdout: , stderr: , returncode: -1, attempts: 0, max_attempts: req.max_attempts, history: [], final_answer: , }, configconfig, ) return {answer: result[final_answer], attempts: result[attempts]}build_graph_with_checkpointer和之前的build_graph只差一行g.compile(checkpointercheckpointer)。thread_id是 LangGraph 隔离会话的关键相同 thread_id 的多次调用共享历史状态。这里每个请求一个新 ID天然做到会话隔离。Semaphore(10)是同时最多处理 10 个请求防止模型 API 配额被打爆也防止子进程并发数量过多耗光内存。4.3 流式输出把修正过程实时送给用户代码生成 Agent 有个体验问题完整跑完可能需要 20 到 40 秒用户盯着空白页面干等很煎熬。LangGraph 支持流式执行你可以把每一步的状态变化推给前端让用户实时看到正在生成代码→正在执行测试→发现错误→正在修复。async def stream_generate(req: CodeRequest): config {configurable: {thread_id: freq-{uuid.uuid4()}}} async for event in graph_app.astream_events( initial_state, configconfig, versionv1 ): # 根据事件名过滤出节点级别的状态更新 if event[event] on_chain_end: node_name event.get(name) result event.get(data, {}).get(output) if isinstance(result, dict) and code in result: yield {event: code_generated, code: result[code]} if isinstance(result, dict) and stderr in result: yield {event: executed, stderr: result[stderr]}SSEServer-Sent Events或 WebSocket 都可以配合这种流式输出。实际项目中我用的是 SSE原因是实现简单前端EventSource直接接。5. 常见问题与排查技巧实录5.1 死循环修复次数到了还在继续修这是我的 Agent 上线后遇到的第一个故障。现象某个任务已经达到 max_attemptsdecide_route 也返回了 finalize但图还在 continue。排查后发现是条件边映射写错了——我在add_conditional_edges的映射里把finalize写成了generate导致所有返回值都走进了 generate。这种错误直接用 LangGraph 的调试模式能看到边的走向但更快的排查方式是打印每一步的decide_route返回值和图上边的映射对一遍。另一种死循环来自模型输出格式问题。模型偶尔会在修复轮次里输出空代码execute_node 执行空文件returncode 还是 0于是图误以为成功了。我的对策是在 execute_node 里先检查代码是否为空或长度小于 10直接返回 returncode -1。5.2 状态污染上一个任务的错误出现在下一个任务里MemorySaver 的默认行为是按 thread_id 保存历史。如果你在服务里复用了同一个 thread_id那么这个 session 的所有历史都会累积进状态。看起来有用实际很危险上一个任务生成的代码和错误会出现在下一个任务的 prompt 里模型会借鉴之前的错误模式反而降低成功率。解决办法每个请求生成一个全新 thread_id如果确实需要多轮对话也必须在每次新任务开始时把 code、stderr、attempts 重置为空/0只保留对话类的消息。5.3 上下文爆炸错误信息把 prompt 塞满了代码生成 Agent 的上下文膨胀非常快。模型生成的代码动辄 1000 到 3000 token执行错误堆栈可能又是 1000 token三轮迭代下来prompt 可能超过 10000 token。我不是说 prompt 长一定不行但 token 越多延迟越高、成本越贵而且模型在长上下文中更容易迷失。我的做法是三层防线第一层执行器里截断 stdout/stderr最多 4000 字符第二层generate_node 里 history 只保留最近两轮不把全部历史喂给模型第三层当某轮错误信息本身非常长时只提取最后 2000 字符因为 Python 异常信息通常是越靠后越接近真正的错误原因。5.4 模型不按指令走输出乱码或重复代码用 LLM 做代码生成最头疼的就是它偶尔不遵守只输出代码的指令。我遇到过的状况有用 markdown 包裹、在代码后面解释这段代码可以……、甚至把修正建议也写进代码里导致语法错误。我已经在 generate_node 里做了 markdown 清理这能解决 80% 的问题。剩下 20% 靠的是结构化输出。如果你想更严格可以让模型输出 JSON再用 JSON parser 提取代码字段。这样即便模型想自由发挥数据结构也会约束它。from langchain_core.utils.function_calling import convert_to_openai_function from langchain.pydantic_v1 import BaseModel, Field class CodeOutput(BaseModel): code: str Field(description完整的纯Python代码) llm_with_struct ChatOpenAI(modelgpt-4o-mini, temperature0.1).with_structured_output(CodeOutput)用结构化输出后我测试了 40 个不同难度的任务格式违规率从原来的 15% 降到了 0值得一试。5.5 代码执行安全性问题最后聊一个严肃的问题让 Agent 执行任意代码本质上是让不受信任的代码在你的机器上跑。我的建议分三个层次本地调试用 subprocess timeout只跑自己信任的代码不碰敏感数据。生产环境必须 Docker 沙箱限制内存、CPU、网络、文件系统挂载设置 runc 的能力白名单。高安全场景考虑用 gVisor 或 Firecracker 这类更强隔离的运行时或者干脆只允许执行静态分析不真正运行代码。strip 掉危险能力比事后补救重要得多。不要在裸机上跑用户提交的代码这句话值得用加粗贴在公司墙上。6. 一点实操体会供你参考做完这个项目我对 LangGraph 最深的体会是它把 Agent 的动态性从代码逻辑里抽离出来放到了图这一层。过去我想调整修复策略要去改模型 prompt、改循环条件、改状态清理逻辑牵一发动全身现在策略调整通常是改一条边的映射或者改一个路由函数的返回值改动面小也更容易测试。如果你在规划自己的代码生成 Agent有几点可以提前设计避免后面返工。第一max_attempts不要拍脑袋定建议根据线上任务的通过率和响应延迟来调任务简单用 2~3 次复杂任务 4~5 次超过 6 次收益会急剧下降因为大部分错误在前两轮就能修好修不好基本是任务描述有问题或模型能力不足。第二尽量把验证和生成分开设计验证逻辑的表现形式就是断言和测试脚本它可以和生成逻辑完全解耦以后想替换成别的语言或别的测试框架都很方便。第三代码执行环境一定要从第一天就考虑隔离我见过太多项目先在裸机环境跑得飞起上线前补安全措施时发现执行方式全要重写。最后一个小技巧如果你期望的是带完整断言的测试体系可以给 execute_node 增加一个test_code字段让模型生成代码的同时生成一组测试用例执行节点把被测代码 测试用例拼在一起跑这样自我修正就从修到能运行升级成了修到通过测试离真正可用的代码生成 Agent 又近了一步。这个方向的扩展空间很大跑顺手之后你可以继续玩多 Agent 协作、外部工具集成甚至代码评审 Agent。