AIエージェントブログ
コンピュータアクセスシステム

状態機械とエージェント：LangGraphで信頼性の高いワークフローを構築する

agentsarchitecturetutoriallanggraphautonomous-systems

ほとんどのエージェントチュートリアルはシンプルなループを示す：Claudeに質問し、レスポンスを解析し、ツールを呼び出し、繰り返す。デモではこれで十分だ。本番環境では、決定論性、エラー回復、人間の承認ゲート、監査可能性が必要になる。

LangGraphはエージェントワークフローに状態機械をもたらす。if文でつなぎ合わされたアドホックなループの代わりに、明示的なグラフを得る：命名されたノード（ロジックユニット）、型付きエッジ（遷移）、実行全体を流れる共有状態スキーマ。

なぜエージェントに状態機械なのか？

アドホックループの問題

典型的なエージェントループはこのようなものだ：

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

ツールが2〜3個なら読みやすい。5つのツール、条件パス、人間の承認ステップ、リトライロジックを加えると、何百行もの複雑な制御フローになる。

より深い問題は暗黙の状態だ。エージェントはどの段階にいるのか？どのデータを集めたのか？すべてがmessagesに存在する——強制スキーマなしに各ノードが読み書きする型なしのブロブ。

状態機械が解決策として

状態機械は暗黙を明示的にする。定義するのは：

  • ノード — 現在の状態を受け取り、1つのことをして、状態更新を返す離散的なロジックユニット。
  • エッジ — ノード間の遷移。無条件（A → B常に）または条件付き。
  • 状態 — グラフ全体を流れる型付き辞書。各ステップでスキーマが検証される。

LangGraphを使わない方がよい場合

シンプルな1ステップタスクには直接のAPIコールの方が速く明確だ。以下の場合にLangGraphを使う：

  • 順序通りに実行する必要のある複数の異なるステージ
  • 中間結果に基づく条件分岐
  • 人間の関与が必要なステップ
  • エラー回復またはリトライロジック
  • 監査可能性の要件

LangGraphの基礎

状態の定義

from typing import TypedDict


class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

ノード

import anthropic


client = anthropic.Anthropic()


def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"このトピックを徹底的に調査する：{state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}


def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"これらのノートに基づいてドラフトを書く：\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

エッジと条件付きルーティング

from typing import Literal


def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"


workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

グラフの構築と実行

from langgraph.graph import StateGraph, END


workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)


graph = workflow.compile()
result = graph.invoke({"query": "MCPプロトコルとは何か？"})
print(result["draft"])

文書レビューワークフローの構築

状態の設計

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

ノードの実装

import json


def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "この文書からちょうど5つの重要用語を抽出する。"
                "文字列のJSON配列として返す。\n\n"
                f"文書：{state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}




def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "この文書のコンプライアンス問題を確認する。"
                "探すもの：個人情報（名前、マイナンバー、メール、電話番号）、"
                "機密マーキング、未検証の主張。\n\n"
                "問題説明のJSON配列を返す。"
                "問題がなければ空の配列[]を返す。\n\n"
                f"文書：{state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}




def human_review_node(state: DocumentState) -> dict:
    print("\n=== 人間によるレビューが必要 ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "個人情報の墨消し後にレビューし承認",
        "is_approved": True
    }




def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"この文書をちょうど2文で要約する：\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

グラフの組み立て

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"


workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)


workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check", route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)


graph = workflow.compile()


result = graph.invoke({
    "document": "患者・田中太郎（マイナンバー：123456789012）は高血圧を患っている。",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

人間のチェックポイントと割り込み

from langgraph.checkpoint.memory import MemorySaver


checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)


thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)


current_state = graph.get_state(thread_config)
print("問題：", current_state.values["compliance_issues"])


graph.update_state(
    thread_config,
    {"human_feedback": "手動での個人情報墨消し後に承認", "is_approved": True}
)


final_result = graph.invoke(None, config=thread_config)

永続化と本番デプロイ

from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)


import asyncio


async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[
        graph.ainvoke({"document": doc, "key_terms": [], "compliance_issues": [],
                       "summary": "", "human_feedback": "", "is_approved": False})
        for doc in documents
    ])

よくあるパターンと落とし穴

  • ファンアウト/ファンイン：Sendで複数ノードを並列実行して結果をマージする。
  • 条件付きルーティングの過度な複雑化：ルーティング関数の分岐が多すぎる場合、サブグラフに分割する。
  • 状態の肥大化：大きなアーティファクトではなく参照（データベースID、S3キー）を保存する。
  • ノードの独立テスト：ノードは通常の関数なので、直接ユニットテストできる。
def test_extract_terms_node():
    state = {
        "document": "人工知能は現代社会を急速に変革している。",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraphはアドホックループを明示的で、デバッグ可能で、再開可能なワークフローに置き換える。線形グラフ（A → B → C → 終了）から始めよう。分岐が必要なときに条件付きエッジを追加する。承認が必要なときに人間のチェックポイントを追加する。ほとんどの本番エージェントワークフローはこの3つのパターンだけで十分だ。

関連記事