AIエージェント入門

AIエージェントのワークフロー設計パターン5選｜LangGraph実装

15分で読める
AIエージェントのワークフロー設計パターン5選｜LangGraph実装

この記事の結論

線形・分岐・ループ・DAG・マップリデュースなどAIエージェントのワークフロー設計パターンをLangGraphのコード例つきで解説。本番環境での選び方も。

「ワークフローが複雑になるほど、エージェントが暴走する」

10社以上のAIエージェント導入を支援してきた中で、最も多く耳にする悩みです。実際、生産環境でのAIエージェント失敗の60%以上は、モデルの能力不足ではなくワークフロー設計の問題だと言われています（Adopt AI, 2026年）。

問題は多くの場合、同じところにあります。最初から複雑なパイプラインを組もうとする。ループの終了条件を設定しない。並列化できるステップを直列に実行している。

この記事では、実際の導入支援で効果が確認された5つのワークフロー設計パターンを、LangGraphのコード例とともに解説します。どのパターンをどんな状況に使うか、判断基準も明確にします。

LangGraphのワークフロー設計の基礎については、AIエージェント構築完全ガイドでも詳しく解説しています。

まず試したい「5分即効」セットアップ

最もシンプルなLangGraphの線形ワークフローから始めましょう。以下はそのまま動くコードです。


# 動作環境: Python 3.11+, langgraph==0.2.x, langchain-anthropic>=0.3.0
# pip install langgraph langchain-anthropic python-dotenv

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict
import os

# ステートの定義（ワークフロー全体で共有されるデータ）
class AgentState(TypedDict):
    input: str
    research: str
    draft: str
    final: str

# モデルの初期化
llm = ChatAnthropic(model="claude-sonnet-4-5", api_key=os.environ["ANTHROPIC_API_KEY"])

# ノードの定義（各ステップが独立した関数）
def research_node(state: AgentState) -> AgentState:
    response = llm.invoke(f"以下のトピックについて調査してください: {state['input']}")
    return {"research": response.content}

def draft_node(state: AgentState) -> AgentState:
    response = llm.invoke(f"調査結果をもとに記事の下書きを作成: {state['research']}")
    return {"draft": response.content}

def review_node(state: AgentState) -> AgentState:
    response = llm.invoke(f"以下の下書きをレビューして最終版を作成: {state['draft']}")
    return {"final": response.content}

# グラフの構築（線形ワークフロー）
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "review")
workflow.add_edge("review", END)

app = workflow.compile()

# 実行
result = app.invoke({"input": "LangGraphのワークフロー設計", "research": "", "draft": "", "final": ""})
print(result["final"])
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

5つのパターンを「選択基準」で理解する

実際の現場では「どのパターンを選ぶか」が最も重要です。以下の基準で判断してください。

パターン 適用場面 難易度 LangGraph対応
線形（Sequential） ステップが明確に依存、順序固定 ★☆☆☆☆ add_edge()のみ
分岐（Router） 入力タイプに応じた処理振り分け ★★☆☆☆ add_conditional_edges()
ループ（Retry） 品質チェック+再試行が必要 ★★★☆☆ 条件付きエッジで戻り矢印
並列（Parallel/DAG） 独立したサブタスクを同時実行 ★★★★☆ fan-out + fan-in
オーケストレーター 長期・複雑タスクの分解・委任 ★★★★★ マルチグラフ構成

パターン1：線形（Sequential）

最もシンプル。Step 2はStep 1が完了してから始まります。上の即効セットアップコードがそのままこのパターンです。

使うべき場面: リサーチ → 執筆 → レビューのような、前ステップの結果が次ステップの必須入力になるタスク。

一つ注意点があります。線形パターンはどこかが失敗すると全体が止まるため、各ノードにエラーハンドリングが必要です。

パターン2：分岐（Router）

入力の種類を分類して、適切な処理ルートに振り分けます。カスタマーサポートエージェントの実装で最も効果的なパターンです。


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

class SupportState(TypedDict):
    message: str
    category: str
    response: str

def classify_node(state: SupportState) -> SupportState:
    """入力をカテゴリ分類するルーター"""
    message = state["message"].lower()
    if "請求" in message or "料金" in message:
        category = "billing"
    elif "バグ" in message or "エラー" in message:
        category = "technical"
    else:
        category = "general"
    return {"category": category}

def route_logic(state: SupportState) -> Literal["billing", "technical", "general"]:
    """分岐の判定ロジック（エッジ関数）"""
    return state["category"]

def billing_node(state: SupportState) -> SupportState:
    return {"response": f"請求専門チームが対応します: {state['message']}"}

def technical_node(state: SupportState) -> SupportState:
    return {"response": f"技術サポートが対応します: {state['message']}"}

def general_node(state: SupportState) -> SupportState:
    return {"response": f"一般サポートが対応します: {state['message']}"}

# グラフ構築
workflow = StateGraph(SupportState)
workflow.add_node("classify", classify_node)
workflow.add_node("billing", billing_node)
workflow.add_node("technical", technical_node)
workflow.add_node("general", general_node)

workflow.set_entry_point("classify")
# 条件付きエッジで分岐
workflow.add_conditional_edges("classify", route_logic, {
    "billing": "billing",
    "technical": "technical",
    "general": "general"
})
workflow.add_edge("billing", END)
workflow.add_edge("technical", END)
workflow.add_edge("general", END)

app = workflow.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

パターン3：ループ（Retry / Quality Check）

エージェントが自己評価して、基準を満たすまで繰り返すパターン。コード生成やレポート作成で特に効果的です。


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

class WritingState(TypedDict):
    topic: str
    draft: str
    feedback: str
    retry_count: int
    approved: bool

MAX_RETRIES = 3  # 必ず上限を設定する（無限ループ防止）

def write_node(state: WritingState) -> WritingState:
    """下書き生成"""
    # LLM呼び出し（省略）
    return {"draft": f"下書き v{state['retry_count'] + 1}", "retry_count": state["retry_count"] + 1}

def review_node(state: WritingState) -> WritingState:
    """品質チェック"""
    draft = state["draft"]
    # 基準チェック（文字数、キーワード含有など）
    if len(draft) > 100 or state["retry_count"] >= MAX_RETRIES:
        return {"approved": True}
    return {"feedback": "文字数が不足しています", "approved": False}

def should_retry(state: WritingState) -> Literal["retry", "done"]:
    """ループ継続か終了かの判定"""
    if state["approved"] or state["retry_count"] >= MAX_RETRIES:
        return "done"
    return "retry"

workflow = StateGraph(WritingState)
workflow.add_node("write", write_node)
workflow.add_node("review", review_node)

workflow.set_entry_point("write")
workflow.add_edge("write", "review")
# ループ: 不合格なら write に戻る、合格なら END
workflow.add_conditional_edges("review", should_retry, {
    "retry": "write",
    "done": END
})

app = workflow.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

最重要ルール: `MAX_RETRIES`は必ず設定してください。設定しないと、モデルが自己評価で永遠にループするケースがあります。APIコストが青天井になります。

パターン4：並列（Parallel / DAG）

互いに独立したタスクを同時実行し、全ての結果をまとめます。実行時間を大幅に短縮できます。


# 並列実行の概念コード（fan-out → fan-in）
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class ResearchState(TypedDict):
    query: str
    results_A: str
    results_B: str
    results_C: str
    final_summary: str

# 3つの調査ノードは独立して並列実行
def research_A(state: ResearchState) -> ResearchState:
    return {"results_A": f"Source A の調査結果: {state['query']}"}

def research_B(state: ResearchState) -> ResearchState:
    return {"results_B": f"Source B の調査結果: {state['query']}"}

def research_C(state: ResearchState) -> ResearchState:
    return {"results_C": f"Source C の調査結果: {state['query']}"}

def aggregate_node(state: ResearchState) -> ResearchState:
    """全結果を統合（fan-in）"""
    summary = f"統合結果:n{state['results_A']}n{state['results_B']}n{state['results_C']}"
    return {"final_summary": summary}

workflow = StateGraph(ResearchState)
workflow.add_node("research_A", research_A)
workflow.add_node("research_B", research_B)
workflow.add_node("research_C", research_C)
workflow.add_node("aggregate", aggregate_node)

# fan-out: 開始点から3つのノードに同時分岐
workflow.set_entry_point("research_A")  # LangGraphは同一レイヤーのノードを並列実行

# fan-in: 全ノードがaggregateに収束
for node in ["research_A", "research_B", "research_C"]:
    workflow.add_edge(node, "aggregate")
workflow.add_edge("aggregate", END)

app = workflow.compile()
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

【要注意】よくある設計ミスと回避策

失敗1：ループに終了条件を設定しない

❌ ループノードに`MAX_RETRIES`も終了条件も設定しない

⭕ 必ず最大反復回数をステートに持ち、超過したら強制終了

なぜ重要か: モデルの自己評価は必ずしも収束しません。条件なしループは数百回のAPI呼び出しを引き起こし、コストが数万円単位になるケースがあります。

失敗2：全ステップを線形に並べる

❌ 独立したタスクAとBを、A→B→C…と直列に接続する

⭕ 依存関係を分析し、独立したステップは並列実行に変更する

なぜ重要か: 5ステップすべてが独立している場合、並列化で実行時間を最大5分の1に短縮できます。

失敗3：エラー時のフォールバックがない

❌ ノードがエラーを返した時、グラフ全体がクラッシュする設計

⭕ 各ノードにtry/exceptとエラー状態をステートで伝播する設計


def safe_research_node(state: AgentState) -> AgentState:
    try:
        # LLM呼び出し
        response = llm.invoke(state["input"])
        return {"research": response.content, "error": None}
    except Exception as e:
        # エラーをステートで伝播（クラッシュさせない）
        return {"research": "", "error": str(e)}
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

パターン選択のフローチャート

実際の設計現場では、以下の順序で考えると判断が速くなります：

  1. タスク間に依存関係があるか？ → ある → 線形 or 分岐
  2. 品質基準を繰り返しチェックが必要か？ → ある → ループ追加
  3. 独立したサブタスクが複数あるか？ → ある → 並列化
  4. タスク自体が動的に変わるか？ → ある → オーケストレーター

まとめ：今日から始める3つのアクション

  1. 今日やること: 線形パターンのサンプルコードを実際に動かしてみる。LangGraph v1.0の基本構造を手で書いて理解する
  2. 今週中: 現在のワークフローを見直して、並列化できるステップを特定する。ループがある場合はMAX_RETRIESを設定する
  3. 今月中: 本番ワークフローに分岐パターンを導入し、エラーログを確認して品質を測定する

あわせて読みたい:

著者: 佐藤傑（さとう・すぐる）
株式会社Uravation代表取締役。早稲田大学法学部在学中に生成AIの可能性に魅了され、X（旧Twitter）で活用法を発信（@SuguruKun_ai、フォロワー10万人超）。100社以上の企業向けAI研修・導入支援を展開。著書累計3万部突破。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。

参考・出典

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

30分無料相談を予約 支援実績を見る

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事

AIエージェント最小権限設計｜AWS/GCP/Azure IAM実装ガイド AIエージェント入門

AIエージェント最小権限設計｜AWS/GCP/Azure IAM実装ガイド

AIエージェントへの過剰権限付与はセキュリティリスクの根本原因です。AWS IA...

13分で読める
Claude Code Agent Teams Q&A｜詰まりポイント完全解消 AIエージェント入門

Claude Code Agent Teams Q&A｜詰まりポイント完全解消

CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1の使...

8分で読める
AIエージェント入門

AIエージェントの認証設計入門：APIキー・OAuth・Vaultの使い分け

AIエージェントが外部APIを安全に呼び出すための認証設計を解説。環境変数→JW...

12分で読める