第一回引言：从“对话”到“推理”的范式转变

传统的LLM应用大多基于“单次问答”或“线性链式（Chains）”模式。在这种模式下，模型如果第一枪没有打中，应用就宣告失败。而在实际工业场景中，我们迫切需要Agent具备像人类一样的反思、推理与迭代纠错能力。

本文将聚焦于如何利用 ReAct 模式、LangGraph 状态机 与 PydanticAI 结构化输出 三大核心技术，从零构建一个具备“自我纠错（Self-Correction）”能力的智能化代码生成 Agent。

---

一、 核心技术栈深度解析

在动手构建复杂系统之前，我们必须先理清底层的技术支柱。一个拥有自主决策能力的 Agent，其大脑由以下三个关键部分构成：

1. 决策范式：ReAct 模式（Reason + Act）

传统的 LLM 习惯于直接输出答案，而 ReAct 模式改变了这一习惯。它强制大模型按照 思考（Thought） $\rightarrow$ 行动（Action） $\rightarrow$ 观察（Observation） $\rightarrow$ 再思考 的闭环逻辑运行。

在代码生成场景中，这套模式的映射非常直观：

• Thought（思考） ：分析用户的需求，或者分析上一次编译器抛出的报错信息。
• Action（行动） ：编写/修改 Python 代码，并将其投入沙箱或本地环境执行。
• Observation（观察） ：获取运行结果，捕获代码成功输出的信息，或是捕获到的系统 Traceback 堆栈。

2. 控制流引擎：LangGraph（状态机）

传统的开发依赖于线性的流程，但“纠错”本质上是一个带循环的非线性工作流。LangGraph 允许我们将 Agent 的执行逻辑抽象为一个 状态机（State Machine） ：

• State（状态） ：全局共享的数据载体，贯穿整个工作流。
• Nodes（节点） ：具体的执行动作，比如“调用大模型生成代码”或“在本地运行测试”。
• Edges（边/条件路由） ：根据当前状态，动态决定下一步是继续循环还是退出流程。

3. 数据契约：PydanticAI（100% 结构化输出）

大模型原生输出的 Markdown 文本（如带有 ```python 的代码块）虽然对人类友好，但对状态机中的下游节点来说却是一场解析灾难。为了保证应用百分之百不崩溃，我们必须使用 PydanticAI 或大模型的 Structured Outputs 特性，强制 LLM 返回严格符合 Python 类型定义的 JSON 对象。

---

二、 系统架构设计与工作流

这个“自我纠错代码生成 Agent”的底层架构设计如下。大模型将在这个闭环中不断循环迭代，直到代码完全正确或触及我们设定的安全熔断阈值。

+-------------------------+
|       [ 开始 ]          |
+------------+------------+
             |
             v
+-------------------------+
|  节点 A: 生成/修改代码   | <---------+
|  (LLM Based on Pydantic)|           |
+------------+------------+           |
             |                        |
             v                        |
+-------------------------+           |
|  节点 B: 执行 & 测试代码 |           |
|   (Python exec() 环境)  |           |
+------------+------------+           |
             |                        |
             v                        |
  /=====================\             |
 <   代码是否运行成功？  >            |
  /=====================/             |
             |                        |
    (否, 有报错) | (是, 无错误)         |
             |                        |
             v                        |
+-------------------------+           |
| 记录报错(error_message) |-----------+
+-------------------------+
             |
 (若次数超过阈值, 如5次)
             |
             v
+-------------------------+
|       [ 结束 ]          |
+-------------------------+

---

三、 核心代码落地实操

接下来，我们将使用 Python 逐步实现该系统的核心模块。

1. 环境准备

首先安装状态机框架与支持结构化输出的相关依赖：

Bash

pip install langgraph pydantic-ai langchain-openai

2. 定义数据契约（State & Schema）

我们利用 Pydantic 规范大模型的返回格式（Schema），并定义整个 LangGraph 状态机在流转时所依赖的全局状态（State）。

Python

from pydantic import BaseModel, Field
from typing import Optional

# 1. 强制 LLM 返回的严格 JSON 结构
class AgentCodeResponse(BaseModel):
    thought: str = Field(description="你对当前代码需求或上一次报错信息的深入思考与分析")
    code: str = Field(description="完整的、可直接运行的 Python 代码，不要包含任何 markdown 标记")

# 2. LangGraph 状态机全局共享的状态 (State)
class GraphState(BaseModel):
    requirement: str                 # 用户的原始需求
    current_code: Optional[str] = None   # 当前生成的代码
    error_message: Optional[str] = None  # 编译器的报错信息（如果有）
    iterations: int = 0              # 当前已循环重试的次数

3. 编写执行节点（Nodes）

接下来编写状态机里的两个核心动作节点。第一个节点负责与大模型交互（生成/修改），第二个节点负责执行代码并捕获异常。

Python

import sys
import io
import traceback
from langchain_openai import ChatOpenAI

# 实例化支持结构化输出的大模型
model = ChatOpenAI(model="gpt-4o").with_structured_output(AgentCodeResponse)

# 节点一：代码生成与修正
def generate_code_node(state: GraphState) -> dict:
    # 动态组装提示词
    if state.error_message:
        prompt = f"你上一次编写的代码报错了。\n【错误信息】:\n{state.error_message}\n\n请结合你的思考（Thought）并彻底修复这个错误，重新输出完整可运行的代码。"
    else:
        prompt = f"请根据以下需求编写 Python 代码：\n{state.requirement}"
        
    # 调用模型（模型将严格按照 AgentCodeResponse 结构返回）
    response = model.invoke(prompt)
    
    return {
        "current_code": response.code,
        "iterations": state.iterations  # 保持当前循环次数，在执行节点中递增
    }

# 节点二：代码本地动态执行
def execute_code_node(state: GraphState) -> dict:
    code = state.current_code
    old_stdout = sys.stdout
    redirected_output = sys.stdout = io.StringIO()
    
    error_info = None
    try:
        # 动态执行 Agent 编写的代码
        exec_globals = {}
        exec(code, exec_globals)
    except Exception as e:
        # 捕获运行时的所有异常，并提取完整的堆栈跟踪信息
        error_info = traceback.format_exc()
    finally:
        sys.stdout = old_stdout
        
    return {
        "error_message": error_info,
        "iterations": state.iterations + 1 # 执行完毕，计数器加 1
    }

4. 编写条件路由与编排状态机

通过 LangGraph 将节点串联起来，引入决策函数 decide_next_step 来控制逻辑是走向退出还是走向循环。

Python

from langgraph.graph import StateGraph, END

# 条件路由决策函数
def decide_next_step(state: GraphState):
    # 如果没有报错，说明代码运行成功，顺利交付
    if state.error_message is None:
        print(f"[+] 代码在第 {state.iterations} 轮尝试中运行成功！")
        return "end"
    
    # 安全阈值：如果重试超过 5 次，强制结束，防止死循环导致 Token 暴风消耗
    if state.iterations >= 5:
        print("[-] 已达到最大重试次数，纠错失败。")
        return "end"
        
    # 否则，提示状态机回到生成节点进行自我纠错
    print(f"[-] 代码运行失败，准备进入第 {state.iterations + 1} 轮自我纠错...")
    return "generate"

# 构建图结构
workflow = StateGraph(GraphState)

# 注册节点
workflow.add_node("generate_code", generate_code_node)
workflow.add_node("execute_code", execute_code_node)

# 设置工作流起点
workflow.set_entry_point("generate_code")

# 添加普通边（生成代码后必须去执行）
workflow.add_edge("generate_code", "execute_code")

# 添加条件边（执行代码后根据结果决定去向）
workflow.add_conditional_edges(
    "execute_code",
    decide_next_step,
    {
        "generate": "generate_code",
        "end": END
    }
)

# 编译生成可执行的 Agent 应用
app = workflow.compile()

---

四、 工业级落地避坑指南

在实际将这套架构投入生产或进行深度开发时，有三个必须要警惕的行业“雷区”：

1. 安全沙箱（Security Sandbox）

   本文在实操中为了演示方便，使用了 Python 原生的 exec() 动态执行代码。但在生产环境中，绝对不允许直接在宿主机运行 LLM 生成的代码。大模型一旦逻辑跑偏，可能会写出 os.system('rm -rf /') 或恶意网络请求。必须将执行节点（Node B）放入极其严格的隔离沙箱（如 Docker 容器、Wasm 或专属的 Code Execution 环境）中。

2. 防止“逻辑鬼打墙”

   有时候大模型对某个报错的理解存在盲区，它会在两套错误代码之间反复横跳（例如：为了改 A 错误引入 B 错误，为了改 B 错误又改回 A 错误）。为此，硬编码 max_iterations（最大重试次数）进行熔断是刚需。此外，可以在 Prompt 中加入“历史尝试记录”，让模型看清自己前几轮踩过的坑。

3. 结构化输出的极端容错

   虽然 PydanticAI 等工具极大地收敛了 LLM 的输出，但在大模型因连续报错而“智商下降”的极端情况下，它依然可能吐出不合法的 JSON 字符串。在生成节点中，同样需要对大模型的返回动作包裹一层 try-except ValidationError。如果解析失败，要把“JSON 格式解析错误”作为 error_message 塞回给状态机，让它自己先修正格式。

---

五、 总结

通过本阶段的实操，我们打破了传统的单向 Prompt 交互模式，利用 LangGraph 和 PydanticAI 构建起了一个具备推理、行动、观测、反思完整闭环的工业级 Agent 雏形。这种“自我纠错”的控制架构，正是走向高级通用 Agent（如自动化软件工程师）的必经之路。