LangGraph: ツール呼び出しと有界実行によるコード検証エージェントの構築

自律的なコード修正には、標準的な線形プロンプトチェーン以上の仕組みが求められます。エージェントがリポジトリに変更を加える際、ファイルの読み込み、編集の適用、テストスイートの実行、出力の検査、そしてリグレッション発生時の自己修正を行う必要があります。しかし、ReAct（Reasoning and Acting）パラダイムに基づく循環型エージェントパターンは、制約がない場合、運用上の重大なリスクを伴います。依存関係の破損や解消されないテスト失敗に直面したエージェントは、コストのかかる無限の実行ループに突入してしまう恐れがあります。

本ガイドでは、Google AIが公開しているLangGraph向けReActアーキテクチャをベースに、中核となるステートグラフのメカニズムを応用して、安全上限（ステップ制限）を備えた特化型コーディングエージェントを構築します。この実装はコード編集を実行し、自動テストスイートに照らして変更を検証し、厳格な反復バジェットを適用します。

---

コアアーキテクチャ: ステート、ノード、条件付きエッジ

LangGraphは、エージェントのワークフローを有向グラフとしてモデル化します。静的な有向非巡回グラフ（DAG）とは異なり、LangGraphはエージェントの反復的な推論に不可欠なサイクル（循環）をネイティブにサポートしています。

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

このグラフは3つのアーキテクチャ要素で成り立っています:

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互換エンドポイント経由での構造化ツール呼び出しパラメータをサポートしていることを必ず確認してください)。

---

ステップ1: 有界エージェントステートの定義

通常の対話ワークフローでは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レデューサーにより、新しいメッセージリストが過去の観察結果を上書きすることなく、会話履歴へ適切に追加されます。

---

ステップ2: ファイルシステムおよび検証ツールの実装

エージェントには、ソースファイルの検査、変更の適用、リポジトリのテスト実行を行うツールが必要です。これらの操作を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}

---

ステップ3: 統合APIを介したモデルのセットアップ

統合プロバイダーのプロキシエンドポイントを指定し、ChatOpenAIを使用してモデルを初期化します。ファンクションコーリングのパラメータ互換性については、OneMux DocsおよびQuickstart Guideを参照してください。

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)

---

ステップ4: グラフノードとループガードの構築

ノードはロジックを実行し、バウンドをインクリメントします。テストの失敗を修正できずにエージェントが繰り返しループする場合、エッジのロジックが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"

---

ステップ5: グラフのコンパイルとアセンブリ

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

---

ステップ6: テスト検証を伴うコード変更の実行

ワークフローを実行するため、厳格な実行上限とともにコーディングタスクを定義します:

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などの任意のシェルコマンドをローカル環境で直接実行すると、環境の破壊や安全でないスクリプト実行のリスクが生じます。テストの実行はエフェメラルなDockerコンテナやセキュアなサンドボックス内に分離してください。
• ステートのトークン制限: 複数ページにわたるスタックトレースなど、ツールの肥大化した出力はコンテキストウィンドウを急速に圧迫します。テストの出力メッセージはステート履歴に書き込む前に適切に切り詰めてください。
• モデルコストの管理: 反復的なReActループはトークン消費を増大させます。OneMux Pricingダッシュボードで利用状況を追跡し、OneMux Modelsでより低コストな推論モデルを評価して、タスクあたりのコストを最適化してください。

---

参考資料

• Google AI for Developers: ReAct agent from scratch with Gemini and LangGraph (Updated Sept 2, 2026).