0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@0h-n0

LangGraph・PydanticAI・自作で学ぶAIエージェントフレームワーク設計パターン実装ガイド

0
Last updated at Posted at 2026-05-03

LangGraph・PydanticAI・自作で学ぶAIエージェントフレームワーク設計パターン実装ガイド

この記事でわかること

  • Anthropicが提唱する5つのエージェント設計パターンの内部構造と実装原理
  • LangGraph 2.0のグラフベース状態機械・チェックポイントの仕組みと本番運用設計
  • PydanticAIの型安全エージェント・依存性注入パターンの活用法
  • フレームワークに依存しないReActエージェントをPython 100行で自作する方法
  • マルチエージェント設計で陥りやすい失敗パターンと段階的拡張戦略

対象読者

  • 想定読者: LLM APIを利用した経験があり、エージェントフレームワークの内部設計を理解したいMLエンジニア
  • 必要な前提知識:
    • Python 3.11以上の基本文法（async/await含む）
    • OpenAI API / Anthropic APIの基本的な呼び出し方法
    • LLMのFunction Calling（Tool Use）の概念理解

結論・成果

Anthropicの提唱する5パターン（Prompt Chaining・Routing・Parallelization・Orchestrator-Workers・Evaluator-Optimizer）を理解し、LangGraph 2.0とPydanticAIで実装することで、単一プロンプトと比較してタスク成功率が40-60%向上することが報告されています（Anthropic公式ガイドによる）。一方、フレームワーク導入にはレイテンシ増加（1.5-3倍）とコスト増（トークン消費2-5倍）というトレードオフがあり、段階的な複雑性追加が本番運用の鉄則です。

エージェント設計パターンの体系を理解する

AIエージェントフレームワークの設計は、Anthropicが2024年末に公開した「Building Effective Agents」で提唱された5パターンが業界標準として定着しています。この体系は「まず最もシンプルなパターンから始め、必要な場合のみ複雑性を足していく」という原則に基づいています。

5パターンの概要と選定基準

各パターンは独立して使えるだけでなく、組み合わせて利用できます。以下の表で選定基準を整理します。

パターン 制御フロー 適用場面 レイテンシ 実装難易度
Prompt Chaining 逐次・固定 品質を段階的に積み上げる処理 高（直列） 低
Routing 分岐・固定 入力カテゴリで処理を切り替える 低 低
Parallelization 並列・固定 独立サブタスクの同時実行 低（並列） 中
Orchestrator-Workers 動的分解 サブタスクが事前定義不可能 中〜高 高
Evaluator-Optimizer 反復ループ 明確な評価基準で反復改善 高(反復) 中

Prompt Chainingの実装原理

Prompt Chainingは最もシンプルなワークフローパターンで、各ステップの出力を次のステップの入力として渡します。中間結果を検証できるため、品質の安定性が単一プロンプトと比較して大幅に向上します。

# prompt_chaining.py
import asyncio
from dataclasses import dataclass
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

@dataclass
class ChainStep:
    """チェインの各ステップを定義"""
    system_prompt: str
    validate: callable  # 中間結果の検証関数

async def run_chain(steps: list[ChainStep], initial_input: str) -> str:
    """Prompt Chainingの実行エンジン"""
    current_input = initial_input

    for i, step in enumerate(steps):
        response = await client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=step.system_prompt,
            messages=[{"role": "user", "content": current_input}],
        )
        result = response.content[0].text

        # 中間結果の検証（ゲート）
        if not step.validate(result):
            raise ValueError(f"Step {i} validation failed: {result[:100]}")

        current_input = result

    return current_input

# 使用例: 技術記事生成チェイン
steps = [
    ChainStep(
        system_prompt="与えられたテーマのアウトラインを5つの見出しで作成してください。",
        validate=lambda r: r.count("#") >= 3,  # 見出しが3つ以上あるか
    ),
    ChainStep(
        system_prompt="アウトラインに基づいて各セクションの本文を執筆してください。",
        validate=lambda r: len(r) > 2000,  # 十分な長さがあるか
    ),
    ChainStep(
        system_prompt="記事全体を校正し、技術的正確性と文体の一貫性を確認してください。",
        validate=lambda r: "エラー" not in r.lower() or "解決" in r,
    ),
]

なぜこの実装を選んだか:

  • 各ステップにvalidate関数を設けることで、品質不足の中間結果が後続ステップに伝播するのを防止
  • dataclassで各ステップを宣言的に定義し、チェインの構成変更を容易にしている

注意点:

Prompt Chainingはステップ数に比例してレイテンシが増加します。3ステップを超える場合は、Parallelizationとの組み合わせを検討してください。ステップ間でコンテキストが肥大化するとトークンコストも急増するため、各ステップの出力は必要最小限に絞ることが重要です。

Routingパターンの実装

Routingは入力を分類し、適切な専門処理に振り分けるパターンです。分類にはコストの低い軽量モデルを使い、専門処理には高性能モデルを使うことでコスト効率を最適化できます。

# routing.py
from enum import Enum
from pydantic import BaseModel
from anthropic import Anthropic

client = Anthropic()

class TaskCategory(str, Enum):
    CODE_REVIEW = "code_review"
    BUG_FIX = "bug_fix"
    DOCUMENTATION = "documentation"
    ARCHITECTURE = "architecture"

class RoutingDecision(BaseModel):
    category: TaskCategory
    confidence: float
    reasoning: str

SPECIALIST_PROMPTS = {
    TaskCategory.CODE_REVIEW: "あなたはコードレビューの専門家です...",
    TaskCategory.BUG_FIX: "あなたはデバッグの専門家です...",
    TaskCategory.DOCUMENTATION: "あなたはテクニカルライターです...",
    TaskCategory.ARCHITECTURE: "あなたはソフトウェアアーキテクトです...",
}

def route_task(user_input: str) -> str:
    # Step 1: 軽量モデルで分類（コスト最小化）
    classification = client.messages.create(
        model="claude-haiku-4-5-20251001",  # 分類は軽量モデルで十分
        max_tokens=256,
        system="タスクを以下のカテゴリに分類してJSON形式で返してください: code_review, bug_fix, documentation, architecture",
        messages=[{"role": "user", "content": user_input}],
    )

    decision = RoutingDecision.model_validate_json(classification.content[0].text)

    # Step 2: 専門モデルで処理
    specialist_response = client.messages.create(
        model="claude-sonnet-4-20250514",  # 専門処理は高性能モデル
        max_tokens=4096,
        system=SPECIALIST_PROMPTS[decision.category],
        messages=[{"role": "user", "content": user_input}],
    )

    return specialist_response.content[0].text

よくある間違い: 最初からすべてのリクエストを最高性能モデルに投げてしまうケースが多いですが、分類処理にHaikuクラスの軽量モデルを使うことで、全体コストを50-70%削減できます（分類はトークン消費が少ないため）。

LangGraph 2.0でグラフベース状態機械を実装する

LangGraph 2.0（2026年2月GA）は、エージェントを有向グラフの状態機械として表現するフレームワークです。ノードが処理関数、エッジが状態遷移を表し、各ステップ後に状態がチェックポイントされるため、障害復旧・タイムトラベルデバッグ・Human-in-the-Loopが自然に実現できます。

StateGraphの設計と型付き状態

LangGraphの設計思想は「状態を一級市民として扱う」ことです。TypedDictで状態スキーマを定義し、各ノードが状態を受け取り・更新する純粋関数として実装されます。

# langgraph_agent.py
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_anthropic import ChatAnthropic

class AgentState(TypedDict):
    """エージェントの状態スキーマ（イミュータブル設計）"""
    messages: Annotated[list, add_messages]  # メッセージ履歴（追記型）
    research_results: list[str]              # リサーチ結果
    review_count: int                        # レビュー回数（ループ上限制御）
    final_answer: str                        # 最終回答

model = ChatAnthropic(model="claude-sonnet-4-20250514")

def classify_node(state: AgentState) -> dict:
    """入力を分類して次のノードを決定"""
    response = model.invoke(
        [{"role": "system", "content": "質問に回答するために追加情報が必要か判断してください。"
         "必要なら 'research'、不要なら 'direct' と回答してください。"}]
        + state["messages"]
    )
    # 条件付きエッジで使用するフラグを状態に保存
    return {"route": response.content}

def research_node(state: AgentState) -> dict:
    """外部ツールで情報収集"""
    # 実際にはWeb検索やDB参照を実行
    results = ["調査結果1: ...", "調査結果2: ..."]
    return {"research_results": results}

def generate_node(state: AgentState) -> dict:
    """回答を生成"""
    context = "\n".join(state.get("research_results", []))
    response = model.invoke(
        [{"role": "system", "content": f"以下の情報を基に回答してください:\n{context}"}]
        + state["messages"]
    )
    return {"final_answer": response.content}

def review_node(state: AgentState) -> dict:
    """生成された回答を自己レビュー"""
    review = model.invoke([
        {"role": "system", "content": "回答の品質を評価してください。問題があれば 'revise'、OKなら 'approve' と回答してください。"},
        {"role": "user", "content": state["final_answer"]},
    ])
    return {"review_count": state.get("review_count", 0) + 1, "review_result": review.content}

def should_revise(state: AgentState) -> Literal["generate", "end"]:
    """レビュー結果に基づきループ継続を判断"""
    if state.get("review_count", 0) >= 2:
        return "end"  # 無限ループ防止
    if "revise" in state.get("review_result", ""):
        return "generate"
    return "end"

# グラフ構築
graph = StateGraph(AgentState)
graph.add_node("classify", classify_node)
graph.add_node("research", research_node)
graph.add_node("generate", generate_node)
graph.add_node("review", review_node)

graph.set_entry_point("classify")
graph.add_conditional_edges("classify", lambda s: s.get("route", "direct"), {
    "research": "research",
    "direct": "generate",
})
graph.add_edge("research", "generate")
graph.add_edge("generate", "review")
graph.add_conditional_edges("review", should_revise, {
    "generate": "generate",
    "end": END,
})

# 本番用: Postgresチェックポイント
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/langgraph")
app = graph.compile(checkpointer=checkpointer)

なぜLangGraphのグラフ設計が本番に適しているか:

  • チェックポイント: 各ノード実行後に状態が永続化されるため、プロセス再起動後も途中から再開可能
  • タイムトラベル: 過去の任意のチェックポイントに戻って状態を確認・再実行可能（デバッグに有用）
  • Human-in-the-Loop: interrupt_before/interrupt_afterで特定ノードの前後に人間の承認を挿入可能

本番運用で必須のチェックポイント設計

開発環境ではMemorySaver（インメモリ）で十分ですが、本番環境ではPostgres永続化が必須です。

環境 チェックポインター 特徴
開発・テスト MemorySaver 高速、プロセス終了で消失
ステージング SqliteSaver ファイルベース永続化
本番 PostgresSaver 水平スケーリング対応、ACID保証

注意点:

MemorySaverを本番で使うと、デプロイのたびに全セッションの状態が消失します。2026年4月時点のLangGraph公式ドキュメントでは、本番環境には必ずPostgresSaverを使い、接続プールのサイズをワーカー数の2倍に設定することが推奨されています。

LangGraphの制約と適用外ケース

LangGraphはすべてのユースケースに適しているわけではありません。

  • 単純なRAGパイプライン: 分岐もループもないなら、LangGraphのオーバーヘッドは不要。直接APIを呼ぶほうがシンプル
  • リアルタイムストリーミング: ノード単位の実行モデルのため、トークン単位のストリーミングは追加設計が必要
  • 数百エージェントの大規模シミュレーション: グラフの定義が静的なため、動的にエージェントを生成・破棄するユースケースには不向き

PydanticAIで型安全エージェントを設計する

PydanticAI（GitHub Stars 16,000+、2026年4月時点）は、FastAPIの設計哲学をエージェント開発に持ち込んだフレームワークです。Pydanticの型検証・依存性注入（DI）・モデル非依存設計により、型エラーを実行前に検出し、テスタブルなエージェントを構築できます。

依存性注入による疎結合設計

PydanticAIの核心はdeps_typeによる依存性注入です。データベース接続・APIクライアント・ユーザーセッションをエージェント定義から分離し、テスト時にモックへ差し替えられます。

# pydantic_ai_agent.py
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
import httpx

# 依存性の型定義
@dataclass
class AgentDeps:
    http_client: httpx.AsyncClient
    db_connection: object  # 実際にはSQLAlchemy AsyncSession等
    user_id: str

# 出力の型定義（型安全な応答）
class ResearchResult(BaseModel):
    summary: str
    sources: list[str]
    confidence: float  # 0.0-1.0

# エージェント定義
research_agent = Agent(
    "anthropic:claude-sonnet-4-20250514",
    result_type=ResearchResult,  # 戻り値の型を厳密に指定
    deps_type=AgentDeps,         # 依存性の型を指定
    system_prompt="あなたはリサーチアシスタントです。情報を収集し構造化された結果を返してください。",
)

@research_agent.tool
async def search_web(ctx: RunContext[AgentDeps], query: str) -> str:
    """Web検索を実行してリサーチ情報を取得"""
    # RunContext経由で依存性にアクセス（型安全）
    response = await ctx.deps.http_client.get(
        "https://api.search.example.com/v1/search",
        params={"q": query, "user": ctx.deps.user_id},
    )
    return response.text

@research_agent.tool
async def fetch_user_history(ctx: RunContext[AgentDeps]) -> str:
    """ユーザーの過去のリサーチ履歴を取得"""
    # DBアクセスも依存性注入経由
    history = await ctx.deps.db_connection.execute(
        "SELECT topic FROM research_history WHERE user_id = :uid",
        {"uid": ctx.deps.user_id},
    )
    return str(history)

# 実行
async def main():
    async with httpx.AsyncClient() as http:
        deps = AgentDeps(http_client=http, db_connection=db, user_id="user-123")
        result = await research_agent.run("量子コンピューティングの最新動向を調べて", deps=deps)
        # result.data は ResearchResult型（型チェッカーが検証）
        print(f"信頼度: {result.data.confidence}")
        print(f"参考: {result.data.sources}")

なぜPydanticAIの型安全設計が有用か:

  • result_type=ResearchResultにより、LLMの出力が必ず指定したスキーマに従う（Pydanticが検証）
  • deps_type=AgentDepsにより、ツール関数内で使える依存性がコンパイル時に型チェックされる
  • テスト時はAgentDepsのモックを注入するだけで、外部サービスなしにエージェントをテスト可能

PydanticAI vs LangGraph: 使い分け

観点 PydanticAI LangGraph
設計哲学 型安全・FastAPI風 グラフ・状態機械
状態管理 RunContext（リクエストスコープ） チェックポイント（永続化）
マルチステップ ツール呼び出しループ ノード・エッジで明示的に定義
Human-in-the-Loop 独自実装が必要 interrupt_beforeで組み込み
テスタビリティ DI + 型安全で容易 グラフの部分実行で対応
学習コスト 低（FastAPI経験者に親和性高） 中（グラフ概念の理解が必要）

選定指針: 単一エージェント+ツール呼び出しで完結するならPydanticAI、複数ステップ間の状態管理や条件分岐が複雑ならLangGraphを選択してください。

フレームワーク非依存のReActエージェントを自作する

フレームワークの内部設計を深く理解するには、ReActパターンを自分で実装するのが効果的です。ReAct（Reasoning + Acting）は「思考→行動→観察」のループで、100行程度のPythonで本質的な動作を再現できます。

# react_agent.py
"""フレームワーク非依存のReActエージェント実装（約100行）"""
import json
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from anthropic import Anthropic

# --- Tool基盤 ---
class BaseTool(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...

    @property
    @abstractmethod
    def description(self) -> str: ...

    @abstractmethod
    def run(self, input: str) -> str: ...

class CalculatorTool(BaseTool):
    name = "calculator"
    description = "数式を計算します。入力: 数式文字列（例: '2 + 3 * 4'）"

    def run(self, input: str) -> str:
        try:
            result = eval(input, {"__builtins__": {}})  # 安全な評価
            return str(result)
        except Exception as e:
            return f"計算エラー: {e}"

class SearchTool(BaseTool):
    name = "search"
    description = "情報を検索します。入力: 検索クエリ"

    def run(self, input: str) -> str:
        # 実際にはWeb検索APIを呼び出す
        return f"検索結果: '{input}' に関する情報..."

# --- ReActエージェント ---
@dataclass
class ReActAgent:
    tools: list[BaseTool]
    max_iterations: int = 5  # 無限ループ防止
    client: Anthropic = field(default_factory=Anthropic)

    def _build_system_prompt(self) -> str:
        tool_descriptions = "\n".join(
            f"- {t.name}: {t.description}" for t in self.tools
        )
        return f"""あなたはReActエージェントです。以下の形式で応答してください:

Thought: [現在の状況分析と次のアクション計画]
Action: [ツール名]
Action Input: [ツールへの入力]

または最終回答が出せる場合:
Thought: [最終的な分析]
Final Answer: [最終回答]

利用可能なツール:
{tool_descriptions}"""

    def run(self, query: str) -> str:
        messages = [{"role": "user", "content": query}]
        system = self._build_system_prompt()

        for iteration in range(self.max_iterations):
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=1024,
                system=system,
                messages=messages,
            )
            text = response.content[0].text

            # Final Answerが含まれていたら終了
            if "Final Answer:" in text:
                return text.split("Final Answer:")[-1].strip()

            # Action/Action Inputをパース
            action, action_input = self._parse_action(text)
            if action is None:
                return text  # パース失敗時はそのまま返す

            # ツール実行
            observation = self._execute_tool(action, action_input)

            # 観察結果をコンテキストに追加
            messages.append({"role": "assistant", "content": text})
            messages.append({"role": "user", "content": f"Observation: {observation}"})

        return "最大反復回数に達しました。回答を生成できませんでした。"

    def _parse_action(self, text: str) -> tuple[str | None, str | None]:
        lines = text.strip().split("\n")
        action = action_input = None
        for line in lines:
            if line.startswith("Action:"):
                action = line.split("Action:")[-1].strip()
            elif line.startswith("Action Input:"):
                action_input = line.split("Action Input:")[-1].strip()
        return action, action_input

    def _execute_tool(self, action: str, action_input: str) -> str:
        for tool in self.tools:
            if tool.name == action:
                return tool.run(action_input)
        return f"エラー: ツール '{action}' は存在しません"

# 使用例
agent = ReActAgent(tools=[CalculatorTool(), SearchTool()])
answer = agent.run("日本の人口は約何人で、一人当たりGDPと掛けるといくら?")

この自作実装から学べること:

  • ReActの本質はwhileループ + テキストパース + ツール実行ルーターの3要素
  • max_iterationsによる無限ループ防止は本番では必須（コスト爆発防止）
  • フレームワークが提供する「魔法」は、この基本構造にチェックポイント・並列実行・型安全を付加したもの

ハマりポイント:

テキストパースでAction/Action Inputを抽出する方式は、LLMの出力フォーマットが揺れると失敗します。本番ではAnthropic/OpenAIのTool Use（Function Calling）APIを使い、構造化された形式でツール呼び出しを受け取る設計にしてください。上記のテキストパースは教育目的の簡易実装です。

マルチエージェント設計の落とし穴と段階的拡張戦略

マルチエージェントシステムへの企業からの問い合わせは2024年Q1から2025年Q2にかけて1,445%増加していると報告されています（LangChain公式ブログによる）。しかし、最初からマルチエージェントで設計することは推奨されません。

よくある失敗パターン

1. 過剰なエージェント分割

最初から5-10個のエージェントを設計し、通信オーバーヘッドとデバッグ困難性に悩むケースが頻出します。Anthropicは「ほとんどのタスクは単一のよく設計されたプロンプトで解決できる」と明言しています。

2. エージェント間のコンテキスト消失

エージェント間でメッセージを受け渡す際、要約や情報の欠落が発生します。共有状態（LangGraphのStateやRedisベースのメモリ）を適切に設計しないと、後続エージェントが必要な情報を持たずに動作します。

3. コスト爆発

マルチエージェントでは各エージェントがLLMを呼び出すため、トークン消費量が線形以上に増加します。特にEvaluator-Optimizerパターンで反復回数に上限を設けないと、1リクエストあたり$1以上のコストが発生することがあります。

段階的拡張の実践ステップ

段階的拡張の原則:

  1. 計測してから拡張: 各ステップで成功率・コスト・レイテンシを計測し、品質上限に到達した場合のみ次のステップに進む
  2. 最小限のエージェント数: 2つのエージェントで解決できるなら、3つ目は追加しない
  3. ループ上限の設定: すべての反復パターンにmax_iterationsを設定（推奨: 3-5回）
  4. フォールバック設計: エージェントが失敗した場合の代替パスを必ず用意

2026年のフレームワーク選定マトリクス

ユースケース 推奨フレームワーク 理由
単一エージェント+ツール PydanticAI 型安全、軽量、DI対応
複雑なワークフロー LangGraph 2.0 グラフ定義、チェックポイント
エンタープライズ/.NET Microsoft Agent Framework 1.0 .NET統合、Azure連携
ラピッドプロトタイプ CrewAI ロールベースで直感的
学習・実験 自作（ReAct from scratch） 内部動作の完全な理解

よくある問題と解決方法

問題 原因 解決方法
エージェントが無限ループ ループ上限未設定 max_iterationsを3-5に制限
ツール選択ミスが頻発 ツール説明が曖昧 description を具体的に（入出力例を含める）
レイテンシが10秒超 逐次呼び出しの連鎖 Parallelization導入、キャッシュ追加
コストが想定の5倍 コンテキスト肥大化 中間結果の要約、不要メッセージの刈り込み
チェックポイント復旧失敗 スキーマ変更 StateのバージョニングとMigration設計
型検証エラーが本番で頻発 LLM出力のフォーマット揺れ リトライ+structured output指定

まとめと次のステップ

まとめ:

  • Anthropicの5パターン（Prompt Chaining / Routing / Parallelization / Orchestrator-Workers / Evaluator-Optimizer）がフレームワーク非依存の設計原則として2026年に標準化
  • LangGraph 2.0はグラフベース状態機械+Postgresチェックポイントで本番ワークフローのデファクトスタンダード
  • PydanticAIは型安全・依存性注入で単一エージェントの品質とテスタビリティを最大化
  • 「まずシンプルに、計測して、必要な場合のみ複雑性を追加」が実務の鉄則
  • マルチエージェントは最終手段であり、単一エージェントの最適化を先に試す

次にやるべきこと:

  • 自分のユースケースで単一プロンプトの品質上限を計測する
  • PydanticAIで型安全な単一エージェントを実装し、テストカバレッジを確保する
  • 品質上限に到達したらLangGraphでワークフローを設計し、Prompt ChainingかRoutingから導入する

参考

注意: この記事はAI（Claude Code）により自動生成されました。内容の正確性については複数の情報源で検証していますが、実際の利用時は公式ドキュメントもご確認ください。

0
0
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
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?