LangGraphで変わるAIエージェント開発 — ステートマシンで制御する自律型ワークフロー

AIエージェントの課題は「賢さ」ではなく「制御」にある

LLMの能力が上がるほど、「エージェントに任せる範囲」を広げたくなる。しかし、本番の業務システムでエージェントを運用する場合、最も問題になるのは推論能力ではない。制御だ。

エージェントが意図しないツールを呼び出す、無限ループに入る、途中で文脈を見失う——プロトタイプ段階では許容できても、本番では許されない。

LangGraphは、この「制御」の問題をステートマシンとして解決するフレームワークだ。2025年に安定版がリリースされ、私たちの受託開発でもエージェント型システムの標準基盤として採用している。

なぜLangChain単体ではダメだったのか

LangChainのAgentExecutorでもエージェントは構築できる。私たちも当初はそちらを使っていた。しかし、本番運用で以下の問題に直面した。

• ループ制御が困難。 AgentExecutorはLLMが「完了」と判断するまで実行し続ける。「3回ツールを呼んだら人間に確認する」のような条件分岐が書きにくい。あるプロジェクトで、100件の文書処理中にAgentExecutorが同一ツールを繰り返し呼び出し、API費用が想定の5倍に達した
• 状態の可視化ができない。 実行中に何が起きているかを外部から観測する仕組みがない。障害時の原因調査に時間がかかる
• チェックポイントがない。 途中で止まったエージェントを再開する機能がなく、長時間タスクに不向き

LangGraphはこれらを「グラフ構造の明示的な定義」で解決する。ノード（処理単位）とエッジ（遷移条件）を宣言的に書くため、実行フローが可視化可能で、条件分岐もコードで制御できる。

パターン1: シーケンシャルワークフロー（直列実行）

最もシンプルなパターン。処理を順番に実行する。

from langgraph.graph import StateGraph, END
from typing import TypedDict

class ReportState(TypedDict):
    raw_data: str
    cleaned_data: str
    analysis: str
    report: str

def clean_data(state: ReportState) -> ReportState:
    state["cleaned_data"] = data_cleaner.invoke(state["raw_data"])
    return state

def analyze(state: ReportState) -> ReportState:
    state["analysis"] = llm.invoke(
        f"以下のデータを分析してください:\n{state['cleaned_data']}"
    ).content
    return state

def generate_report(state: ReportState) -> ReportState:
    state["report"] = llm.invoke(
        f"以下の分析結果からレポートを生成:\n{state['analysis']}"
    ).content
    return state

graph = StateGraph(ReportState)
graph.add_node("clean", clean_data)
graph.add_node("analyze", analyze)
graph.add_node("report", generate_report)
graph.set_entry_point("clean")
graph.add_edge("clean", "analyze")
graph.add_edge("analyze", "report")
graph.add_edge("report", END)
app = graph.compile()

適するケース: ETL→分析→レポート生成のような、処理順序が固定の自動化パイプライン。

パターン2: 条件分岐ワークフロー（ルーター型）

入力内容に応じて処理を振り分けるパターン。

from langgraph.graph import StateGraph, END
from typing import TypedDict

class RequestState(TypedDict):
    input: str
    request_type: str
    response: str

def classify_request(state: RequestState) -> RequestState:
    result = llm.invoke(f"以下のリクエストを分類してください（技術/営業/一般）:\n{state['input']}")
    state["request_type"] = result.content.strip()
    return state

def route_by_type(state: RequestState) -> str:
    if "技術" in state["request_type"]:
        return "tech_handler"
    elif "営業" in state["request_type"]:
        return "sales_handler"
    return "general_handler"

def handle_tech(state: RequestState) -> RequestState:
    state["response"] = llm.invoke(f"技術的な回答を生成:\n{state['input']}").content
    return state

def handle_sales(state: RequestState) -> RequestState:
    state["response"] = llm.invoke(f"営業向けの回答を生成:\n{state['input']}").content
    return state

def handle_general(state: RequestState) -> RequestState:
    state["response"] = llm.invoke(f"一般的な回答を生成:\n{state['input']}").content
    return state

graph = StateGraph(RequestState)
graph.add_node("classify", classify_request)
graph.add_node("tech_handler", handle_tech)
graph.add_node("sales_handler", handle_sales)
graph.add_node("general_handler", handle_general)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_type)
graph.add_edge("tech_handler", END)
graph.add_edge("sales_handler", END)
graph.add_edge("general_handler", END)
app = graph.compile()

適するケース: カスタマーサポートの自動振り分け、ドキュメントの種類別処理。ノード関数（状態を更新するclassify_request）と条件分岐関数（遷移先を決定するroute_by_type）を明確に分離することで、LangGraphの設計思想——ノードは状態変更、エッジは遷移判定——に準拠する。

パターン3: 自己修正ループ（Human-in-the-Loop）

エージェントが出力を自己評価し、不十分なら再実行する。一定条件で人間に判断を委ねる。

from langgraph.graph import StateGraph, END
from typing import TypedDict

class AgentState(TypedDict):
    task: str
    output: str
    score: float
    retry_count: int

def execute_task(state: AgentState) -> AgentState:
    state["output"] = llm.invoke(f"以下のタスクを実行:\n{state['task']}").content
    state["retry_count"] = state.get("retry_count", 0) + 1
    return state

def evaluate_quality(state: AgentState) -> AgentState:
    eval_result = llm.invoke(f"以下の出力を0.0-1.0で品質評価:\n{state['output']}")
    state["score"] = float(eval_result.content.strip())
    return state

def decide_next(state: AgentState) -> str:
    if state["score"] >= 0.8:
        return "complete"
    elif state["retry_count"] >= 3:
        return "human_review"
    return "retry"

def escalate(state: AgentState) -> AgentState:
    state["output"] = f"[要人間レビュー] {state['output']}"
    return state

graph = StateGraph(AgentState)
graph.add_node("execute", execute_task)
graph.add_node("evaluate", evaluate_quality)
graph.add_node("human_review", escalate)
graph.set_entry_point("execute")
graph.add_edge("execute", "evaluate")
graph.add_conditional_edges("evaluate", decide_next, {
    "complete": END,
    "retry": "execute",
    "human_review": "human_review",
})
graph.add_edge("human_review", END)
app = graph.compile()

適するケース: 文書レビューの自動化、コード生成の品質担保、判断が微妙なケースを人間にエスカレートする業務。

設計上の要点: retry_count を状態に持たせて無限ループを防止する。Human-in-the-LoopはLangGraphの interrupt_before / interrupt_after で実装可能。チェックポイントと組み合わせて、人間の判断後に再開できる。

LangGraphを選ばなかった代替手段

• CrewAI: マルチエージェントの「役割分担」に特化。しかし、ステートマシンとしての制御粒度が粗く、条件分岐やHuman-in-the-Loopの柔軟性が不足する
• AutoGen: Microsoft製。会話型マルチエージェントに強いが、ワークフロー制御の宣言的な記述がしにくい。デバッグも難しい
• 自前実装: Python + asyncioで状態遷移を書く方法。小規模なら有効だが、チェックポイント・可視化・再開機能を自前で作るコストが大きい

LangGraphを選んだ理由は「制御の明示性」と「LangChainエコシステムとの連続性」。RAG（B-01で解説）と同じ基盤でエージェントを構築でき、プロジェクト間でコンポーネントを再利用できる。

まだ解決できていないこと

テストの難しさ。 LLMを含むグラフのテストは非決定論的だ。条件分岐の網羅テストは書けるが、LLMノードの出力品質テストはRAGAS等の評価パイプラインに依存している。

デバッグの視認性。 LangSmithで実行トレースを可視化しているが、複雑なグラフでは分岐の追跡に時間がかかる。ログの構造化がまだ十分でない。

コスト予測。 自己修正ループではLLM呼び出し回数が可変のため、1リクエストあたりのコストが予測しにくい。retry上限で最悪ケースは管理しているが、平均コストの見積もりは実績データが必要。

AIエージェント開発の本質は、LLMの推論能力ではなく、その能力をいかに制御するかにある。賢さの前に、まず制御を設計すべきだ。

関連記事