LangGraph：打造具備工具調用與有界執行的程式碼驗證 Agent

自主程式碼修改需要的不僅是標準的線性提示詞鏈（linear prompt chain）。當 Agent 修改程式庫時，必須讀取檔案、套用變更、執行測試套件、檢查輸出結果，並在出現回歸（regression）時自行修正。然而，基於 ReAct（Reasoning and Acting，推理與行動）範式的循環 Agent 模式若未加約束，可能會帶來維運風險。遇到依賴損壞或持續測試失敗的 Agent 可能會陷入高成本的無休止執行迴圈。

本指南參考 Google AI 發布的 LangGraph ReAct 架構文件，將核心狀態圖機制改造成專門且具備執行邊界（bounded）的 Coding Agent。該實作可執行程式碼修改、針對自動化測試套件驗證變更，並強制實施嚴格的迭代上限（iteration budget）。

---

Core Architecture: State, Nodes, and Conditional Edges

LangGraph 將 Agent 工作流程建模為有向圖。與靜態的有向無環圖（DAG）不同，LangGraph 原生支援循環（cycles），這對 Agent 的迭代推理至關重要。

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

該圖表依賴三個架構元素：

1. 狀態（State）： 一個集中式的具型別資料結構，用以儲存訊息歷史、執行元數據以及明確的步數計數器。
2. 節點（Nodes）： 獨立的 Python 函式，接收當前狀態，執行隔離計算（調用模型或執行工具），並回傳更新後的狀態屬性。
3. 條件邊（Conditional Edges）： 決策函式，用以檢查狀態（例如是否存在工具調用，或是是否超過迭代上限），藉此決定路由至工具執行節點還是終止於 END。

---

Prerequisites and Dependencies

請確認您的開發環境已安裝必要的圖編排與模型驅動套件：

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 相容端點下支援結構化工具調用參數）。

---

Step 1: Defining Bounded Agent State

在標準的對話工作流中，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 能確保新傳入的訊息列表俐落地附加至對話歷史中，而非覆寫先前的觀察結果。

---

Step 2: Implementing File System and Verification Tools

Agent 需要工具來檢查原始檔案、套用變更以及執行儲存庫測試。我們使用 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}

---

Step 3: Model Setup via Unified API

我們使用 ChatOpenAI 初始化模型，並將其指向統一的代理端點。請查閱 OneMux 文件 與 快速入門指南，以確保函式調用（function calling）的參數相容性。

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 4: Constructing Graph Nodes and Loop Guards

節點負責執行邏輯並累加計數。若 Agent 在無法修復測試失敗的情況下反覆進入迴圈，邊邏輯將檢測到 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"

---

Step 5: Graph Compilation and Assembly

使用 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()

---

Step 6: Executing Code Changes with Test Verification

若要執行該工作流程，請定義編程任務以及嚴格的執行邊界：

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.

---

Limitations and Production Hardening

在生產環境部署具備程式碼修改能力的 Agent 時，請考量下列防護措施：

• 沙盒隔離執行（Sandboxing Execution）： 在本地端直接執行如 pytest 的任意 shell 指令，容易面臨環境損壞或腳本不安全的風險。應將測試執行隔離在臨時的 Docker 容器或安全沙盒中。
• 狀態 Token 限制（State Token Limits）： 冗長的工具輸出（例如多頁的堆疊追蹤資訊）可能迅速耗盡上下文視窗。在將測試輸出寫入狀態歷史之前，請先行截斷處理。
• 控制模型成本（Managing Model Costs）： 迭代式的 ReAct 迴圈會成倍增加 Token 消耗量。可透過 OneMux 計費與定價 儀表板追蹤用量，並在 OneMux 模型列表 上評估成本較低的推理模型，以優化單一任務成本。

---

Sources

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