LangGraph 实战：构建具备工具调用与有界执行的代码验证智能体

实现代码的自主修改，仅靠标准的线性提示词链（prompt chain）是远远不够的。当智能体对代码库进行修改时，它必须读取文件、应用代码变更、执行测试套件、检查运行输出，并在引入回归缺陷时具备自我纠错的能力。然而，通常基于 ReAct（推理与行动，Reasoning and Acting）范式的循环智能体模式，如果缺乏约束，就会带来极大的运行隐患。智能体一旦遇到依赖损坏或反复无法修复的测试失败，很可能会陷入耗资巨大的无界循环中。

参考 Google AI 基于 LangGraph 构建的 ReAct 架构文档，本指南将核心的状态图机制改造成一个专用的、有界执行的代码修改智能体。该实现能够执行代码编辑、通过自动化测试套件验证变更，并严格执行迭代预算控制。

---

核心架构：状态、节点与条件边

LangGraph 将智能体工作流建模为有向图。与静态的有向无环图（DAG）不同，LangGraph 原生支持循环结构，这对于智能体的迭代推理至关重要。

          [Entrypoint]
               │
               ▼
        ┌─────────────┐
        │  call_model │◄────────────┐
        └──────┬──────┘             │
               │                    │
     [should_continue]              │
        /             \             │
   (has tools)     (done / max)     │
      /                 \           │
     ▼                   ▼          │
┌───────────┐         ┌─────┐       │
│ call_tool │         │ END │       │
└─────┬─────┘         └─────┘       │
      │                             │
      └─────────────────────────────┘

该图依赖三大架构要素：

1. 状态（State）： 集中式的类型化数据结构，用于存储消息历史、运行元数据以及显式步数计数器。
2. 节点（Nodes）： 普通的 Python 函数，接收当前状态，运行独立的计算逻辑（调用模型或执行工具），并返回更新后的状态属性。
3. 条件边（Conditional Edges）： 决策判断函数，通过检查状态（例如是否存在工具调用请求，或者是否超过迭代步数上限）来决定是路由到工具执行节点还是直接在 END 终止流程。

---

前置准备与环境依赖

请确保您的开发环境中已安装必要的图编排与模型驱动依赖包：

pip install langgraph langchain-core langchain-openai

为了通过兼容 OpenAI 的网关将智能体连接到模型推理服务，需配置相应的环境变量。使用 OneMux 可以简化跨多个模型的统一密钥管理与用量额度控制：

export ONEMUX_API_KEY="your_onemux_api_key"
export ONEMUX_BASE_URL="https://api.onemux.net/v1"

（注意：请务必确认所选的目标模型支持通过 OpenAI 兼容端点传入结构化工具调用参数）。

---

第一步：定义有界智能体状态

在标准对话场景中，通常直接使用 MessagesState 处理对话历史的追加。但在代码编辑任务中，我们使用 Python 的 TypedDict 显式定义 AgentState，以便在记录消息序列的同时追踪明确的计数器（step_count）和最大阈值（max_steps）。

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    step_count: int
    max_steps: int

add_messages 规约器（reducer）确保新传入的消息列表能干净利落地追加到对话历史中，而不会覆盖原有的观察记录。

---

第二步：实现文件系统与验证工具

智能体需要相应的工具来检查源码文件、应用代码修改并执行仓库的测试套件。我们使用 LangChain 的 @tool 装饰器将这些操作封装起来。

import subprocess
from langchain_core.tools import tool

@tool
def read_file(filepath: str) -> str:
    """Reads contents of a file from the repository."""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"Error reading {filepath}: {str(e)}"

@tool
def write_file(filepath: str, content: str) -> str:
    """Overwrites a file with modified source code."""
    try:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully updated {filepath}"
    except Exception as e:
        return f"Error writing {filepath}: {str(e)}"

@tool
def run_test_suite(test_target: str) -> str:
    """Runs pytest on a specific file or test target."""
    try:
        result = subprocess.run(
            ["pytest", test_target],
            capture_output=True,
            text=True,
            timeout=30
        )
        return f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}\nExit Code: {result.returncode}"
    except subprocess.TimeoutExpired:
        return "Error: Test execution timed out after 30 seconds."
    except Exception as e:
        return f"Error executing tests: {str(e)}"

tools = [read_file, write_file, run_test_suite]
tools_by_name = {t.name: t for t in tools}

---

第三步：通过统一 API 配置模型

我们使用 ChatOpenAI 初始化模型，并将其指向统一的代理网关端点。请查阅 OneMux 文档 与 快速入门指南，确保参数配置符合函数调用的兼容要求。

import os
from langchain_openai import ChatOpenAI

api_key = os.getenv("ONEMUX_API_KEY")
base_url = os.getenv("ONEMUX_BASE_URL", "https://api.onemux.net/v1")

llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.0,
    api_key=api_key,
    base_url=base_url
)

# Bind tools to the model instance
model = llm.bind_tools(tools)

---

第四步：构建图节点与循环防护机制

节点负责执行具体业务逻辑并增加步数。如果智能体在未修复测试的情况下反复陷入死循环，边路由逻辑会检测到 step_count >= max_steps，并直接导向图的终止流程。

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.runnables import RunnableConfig

def call_model(state: AgentState, config: RunnableConfig):
    """Calls the model and increments the iteration step count."""
    response = model.invoke(state["messages"], config)
    current_steps = state.get("step_count", 0) + 1
    return {
        "messages": [response],
        "step_count": current_steps
    }

def call_tool(state: AgentState):
    """Executes tool calls generated by the most recent model response."""
    last_message = state["messages"][-1]
    outputs = []
    
    for tool_call in getattr(last_message, "tool_calls", []):
        tool_inst = tools_by_name.get(tool_call["name"])
        if tool_inst:
            result = tool_inst.invoke(tool_call["args"])
        else:
            result = f"Error: Tool {tool_call['name']} not found."
            
        outputs.append(
            ToolMessage(
                content=str(result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            )
        )
    return {"messages": outputs}

def should_continue(state: AgentState):
    """Evaluates tool requests and halts execution if boundaries are crossed."""
    messages = state["messages"]
    last_message = messages[-1]
    step_count = state.get("step_count", 0)
    max_steps = state.get("max_steps", 5)

    # Boundary check: Halt if loop budget exceeded
    if step_count >= max_steps:
        return "end"

    # Continue if the model requested another tool call
    if getattr(last_message, "tool_calls", None):
        return "continue"
        
    return "end"

---

第五步：图的编译与组装

使用 StateGraph 将上述各个组件组装并完成工作流的编译：

from langgraph.graph import StateGraph, END

workflow = StateGraph(AgentState)

# Register Nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tool)

# Set Entrypoint
workflow.set_entry_point("agent")

# Define Conditional and Static Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile Graph
app = workflow.compile()

---

第六步：执行带有测试验证的代码修改流程

启动工作流时，定义具体的编码任务并附带严格的执行步数限制：

initial_input = {
    "messages": [(
        "user",
        "Fix the bug in utils/math_lib.py where divide_numbers fails on zero values. "
        "Verify your fix by running the test suite on tests/test_math_lib.py. "
        "Only stop once all tests pass."
    )],
    "step_count": 0,
    "max_steps": 6
}

for event in app.stream(initial_input, stream_mode="values"):
    latest = event["messages"][-1]
    if isinstance(latest, AIMessage) and latest.tool_calls:
        print(f"[Action] Invoking tools: {[t['name'] for t in latest.tool_calls]}")
    elif isinstance(latest, ToolMessage):
        print(f"[Observation] {latest.name} -> {latest.content[:100]}...")
    else:
        print(f"[Agent] {latest.content}")

[Action] Invoking tools: ['read_file']
[Observation] read_file -> def divide_numbers(a, b):
    return a / b...
[Action] Invoking tools: ['write_file']
[Observation] write_file -> Successfully updated utils/math_lib.py...
[Action] Invoking tools: ['run_test_suite']
[Observation] run_test_suite -> STDOUT:
============================= test session starts ==============================
... 1 passed in 0.02s...
[Agent] The divide_numbers function has been updated to handle zero division safely, and all tests in tests/test_math_lib.py pass.

---

局限性与生产级加固建议

在生产环境中部署能够修改代码的智能体时，请务必考虑以下安全加固措施：

• 沙箱化隔离执行： 在本地直接运行诸如 pytest 之类的任意 Shell 命令，存在污染系统环境或执行不受信任脚本的风险。建议将测试执行隔离在临时 Docker 容器或受控安全沙箱中。
• 状态 Token 长度控制： 冗长的工具输出（例如多页报错堆栈）会迅速消耗上下文窗口容量。建议在将测试输出写入状态历史前进行必要的内容截断。
• 模型成本管理： 循环迭代的 ReAct 流程会成倍放大 Token 消耗。可通过 OneMux 定价 仪表盘实时监控使用量，并在 OneMux 模型列表 中评估成本更低的高性价比推理模型，以优化每次任务的实际支出。

---

参考资料

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph（更新于 2026 年 9 月 2 日）。