1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@miyaguchi_kioku(宮口 公寿)

LangGraphのStateとConditional Edgeをブラウザで見える化して理解する

1
Posted at

はじめに

LangGraphを学んでいると、よく次のような説明が出てきます。

  • State: グラフ全体で共有される状態
  • Node: Stateを受け取って処理し、Stateの差分を返す関数
  • Edge: 次にどのNodeへ進むかを決める接続
  • Conditional Edge: Stateを見て分岐先を決める条件付きEdge

説明としては理解できるのですが、最初は「普通のループと何が違うのか」が少し分かりにくいです。

そこで今回は、LangGraphで「コード生成 -> 安全確認 -> テスト -> 修正/停止」という流れを作り、それをブラウザ上の小さなダッシュボードで見える化しました。

なお、このサンプルではLLMによるコード生成部分はモックにしています。目的は、LangGraphのState、Node、Conditional Edgeによって処理フローをどう制御できるかを理解することです。

作ったもの

ブラウザ上で、LangGraphの実行状態をリアルタイム風に確認できるデモです。

langgraph-dashboard.png

画面では、次のようなノードが表示されます。

コード生成
  ↓
安全確認
  ├─ 問題なし -> テスト
  ├─ 要件が曖昧 -> 人間確認
  └─ 危険コード -> ブロック

テスト
  ├─ 成功 -> 終了
  └─ 失敗 -> エラー分類 -> コード生成へ戻る

実行中は、現在動いているノードが光り、右側にState、生成コード、試行回数、エラー種別などが表示されます。

なぜ普通のwhileではなくLangGraphなのか

単純な「失敗したら再実行」だけなら、普通のwhileで十分です。

例えば、次のような処理です。

while True:
    code = generate_code()
    success, error = run_test(code)

    if success:
        break

    if retry_count >= 3:
        break

ただ、実際のコード生成フローでは、分岐が増えていきます。

  • テストに成功したら終了する
  • テストに失敗したらエラーを分類する
  • 文法エラーなら再生成する
  • 要件が曖昧なら人間確認に回す
  • 危険なコードなら即ブロックする
  • 何度も失敗したら人間に渡す

このような処理をすべてwhileとifで書くと、「今どの状態なのか」「次にどこへ進むのか」が読みにくくなります。

LangGraphでは、処理を次の3つに分けて考えられます。

State = 今の状態
Node = 何をするか
Edge = 次にどこへ進むか

この分離が、LangGraphの分かりやすいところです。

Stateの定義

今回のStateは次のように定義しました。

from typing import TypedDict


class GraphState(TypedDict):
    task: str
    scenario: str
    code: str
    error_msg: str
    error_type: str
    iteration: int
    status: str
    next_action: str

主な意味は次の通りです。

キー 意味
task ユーザー要件
scenario デモ用の実行パターン
code 生成されたコード
error_msg テストや安全確認のエラー
error_type エラー分類
iteration コード生成の試行回数
status 現在の状態
next_action 次に進む方向

ポイントは、Node同士が直接やり取りするのではなく、共有されたStateを更新しながら進むことです。

Nodeの例

コード生成Node

このNodeは、コードを生成してStateに返します。

今回のデモではLLMを呼ばず、シナリオに応じて固定のコードを返しています。

def generate_code_node(state: GraphState) -> dict[str, str | int]:
    iteration = state["iteration"] + 1
    scenario = state["scenario"]

    if scenario == "dangerous_code":
        code = """\
import os

def add(a, b):
    os.remove("important_file.txt")
    return a + b
"""
    elif iteration == 1:
        code = """\
def add(a, b):
    return a - b
"""
    else:
        code = """\
def add(a, b):
    return a + b
"""

    return {
        "code": code,
        "iteration": iteration,
        "status": "generated",
        "next_action": "test",
    }

本物のLLMを使う場合は、このNodeの中をLLM呼び出しに差し替えます。

安全確認Node

安全確認Nodeでは、生成コードに危険な処理が含まれていないかを見ます。

def safety_check_node(state: GraphState) -> dict[str, str]:
    code = state["code"]
    dangerous_words = ["os.remove", "shutil.rmtree", "subprocess.run"]

    if not code:
        return {
            "error_type": "unclear_requirement",
            "error_msg": "コード生成前に要件確認が必要です。",
            "status": "needs_human",
            "next_action": "human_review",
        }

    for word in dangerous_words:
        if word in code:
            return {
                "error_type": "dangerous_code",
                "error_msg": f"Blocked dangerous code: {word}",
                "status": "blocked",
                "next_action": "block",
            }

    return {
        "error_type": "",
        "status": "safe",
        "next_action": "test",
    }

ここでは簡単な文字列検出だけですが、実運用ではAST解析、サンドボックス、許可リスト、権限分離などが必要になります。

テスト実行Node

テストNodeでは、生成されたコードを一時ファイルに保存し、実際にPythonでテストを実行します。

def run_test_node(state: GraphState) -> dict[str, str]:
    code = state["code"]
    test_code = """\
from solution import add

assert add(1, 2) == 3
assert add(-1, 1) == 0
assert add(10, 5) == 15
"""

    with tempfile.TemporaryDirectory() as temp_dir:
        temp_path = Path(temp_dir)
        solution_path = temp_path / "solution.py"
        test_path = temp_path / "test_solution.py"

        solution_path.write_text(code)
        test_path.write_text(test_code)

        result = subprocess.run(
            ["python3", str(test_path)],
            cwd=temp_path,
            text=True,
            capture_output=True,
            check=False,
        )

    if result.returncode == 0:
        return {
            "error_msg": "",
            "error_type": "",
            "status": "passed",
            "next_action": "end",
        }

    error_msg = result.stderr.strip() or result.stdout.strip()
    return {
        "error_msg": error_msg,
        "status": "failed",
        "next_action": "classify_error",
    }

ここはモックではなく、本当にPythonコードを実行しています。

Conditional Edgeで分岐する

LangGraphで面白いのは、Nodeの処理結果を見て、次のNodeを決定できるところです。

例えば、安全確認後の分岐は次のように書けます。

def route_after_safety(state: GraphState):
    if state["next_action"] == "human_review":
        return "human_review"

    if state["next_action"] == "block":
        return "block"

    return "test"

これをadd_conditional_edgesで登録します。

workflow.add_conditional_edges(
    "SafetyCheck",
    route_after_safety,
    {
        "test": "Tester",
        "human_review": "HumanReview",
        "block": "Blocked",
    },
)

この設計にすると、AIの出力が危険だった場合でも、AIに次の行動を決めさせる必要がありません。

Pythonの決定論的なロジックで、必ずBlockedへ送れます。

グラフ全体の構築

全体は次のように構築します。

from langgraph.graph import END, StateGraph


workflow = StateGraph(GraphState)

workflow.add_node("Coder", generate_code_node)
workflow.add_node("SafetyCheck", safety_check_node)
workflow.add_node("Tester", run_test_node)
workflow.add_node("ErrorClassifier", classify_error_node)
workflow.add_node("HumanReview", human_review_node)
workflow.add_node("Blocked", blocked_node)

workflow.set_entry_point("Coder")
workflow.add_edge("Coder", "SafetyCheck")

workflow.add_conditional_edges(
    "SafetyCheck",
    route_after_safety,
    {
        "test": "Tester",
        "human_review": "HumanReview",
        "block": "Blocked",
    },
)

workflow.add_conditional_edges(
    "ErrorClassifier",
    route_after_error,
    {
        "retry": "Coder",
        "human_review": "HumanReview",
    },
)

workflow.add_edge("HumanReview", END)
workflow.add_edge("Blocked", END)

app = workflow.compile()

ブラウザ表示

今回は、Pythonの小さなHTTPサーバーを用意し、Server-Sent Eventsで実行イベントをブラウザへ流しています。

ブラウザ側では、イベントを受け取るたびに次の表示を更新します。

  • 現在動いているNode
  • 通過したEdge
  • Stateの中身
  • 生成コード
  • イベントログ

これにより、LangGraphが「ただのループ」ではなく、「状態に応じて経路が変わるステートマシン」であることが見やすくなります。

動くシナリオ

1. テスト失敗から自動修正

最初はわざと間違ったコードを生成します。

def add(a, b):
    return a - b

テストでAssertionErrorになり、エラー分類後にコード生成へ戻ります。

2回目で正しいコードになります。

def add(a, b):
    return a + b

2. 文法エラーから自動修正

最初は文法エラーのあるコードを生成します。

def add(a, b):
    return a +

テスト実行時にSyntaxErrorとなり、エラー分類後にコード生成へ戻ります。

3. 曖昧な要件で人間確認

要件が曖昧な場合は、コード生成を無理に進めず、人間確認へ送ります。

コード生成 -> 安全確認 -> 人間確認 -> 終了

4. 危険コードをブロック

生成コードに危険な処理が含まれている場合は、テストに進めずブロックします。

os.remove("important_file.txt")

この場合は、次の経路になります。

コード生成 -> 安全確認 -> ブロック -> 終了

実行方法

依存関係をインストールします。

python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt

コンソール版を実行する場合:

.venv/bin/python langgraph_code_test_sample.py

ブラウザ版を実行する場合:

.venv/bin/python dashboard_app.py

その後、ブラウザで次を開きます。

http://127.0.0.1:8001

このサンプルで本当にやっていること

このデモは、すべてが見せかけというわけではありません。

実際にやっていること:

  • LangGraphのStateGraphでグラフを構築している
  • NodeとConditional Edgeで経路を制御している
  • Stateを更新しながら処理している
  • 生成コードを一時ファイルに書き出している
  • Pythonでテストコードを実行している
  • テスト結果に応じて再生成、終了、人間確認、ブロックへ分岐している
  • 実行イベントをブラウザへ配信している

一方で、LLMによるコード生成はまだモックです。

つまり、この記事の主題は「本物のAIコード生成エージェントを作った」ではなく、「LangGraphのStateとConditional Edgeによる制御構造を理解する」です。

まとめ

LangGraphの良さは、単にループを書けることではありません。

重要なのは、AIのように出力が揺れやすい処理をNodeに閉じ込めつつ、システム全体の進行ルールをEdgeで決定論的に制御できることです。

今回のようなコード生成フローでは、次のような制御を明示できます。

  • 成功したら終了
  • テスト失敗なら修正へ戻す
  • 文法エラーなら修正へ戻す
  • 要件が曖昧なら人間確認へ送る
  • 危険コードなら即ブロックする

このように、LangGraphは「AIに全部任せる」ための道具というより、AIを含む処理全体を安全に制御するためのステートマシンとして使うと分かりやすいです。

次に発展させるなら、モックのコード生成Nodeを実際のLLM呼び出しに置き換えたり、チェックポイントやHuman-in-the-loopの再開処理を追加すると、より実運用に近い構成になります。

1
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?