LiteLLMとLangGraphのSend APIでDeep Researchエージェントを実装してみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langchain-litellm 0.6.4
langchain-core 1.3.2

$ uv pip install \
    litellm==1.83.14 \
    langgraph==1.1.10 \
    langchain-litellm==0.6.4 \
    langchain-core==1.3.2

$ export OPENAI_API_KEY="sk-..."
$ export ANTHROPIC_API_KEY="sk-ant-..."

import operator
from typing import Annotated, TypedDict

from langchain_core.messages import HumanMessage
from langchain_litellm import ChatLiteLLM
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

llm = ChatLiteLLM(model="openai/gpt-5-mini")

class State(TypedDict):
    questions: list[str]
    # 並列ノードからの戻り値を結合するためのリデューサー
    answers: Annotated[list[str], operator.add]

class SubState(TypedDict):
    question: str

def dispatch(state: State) -> list[Send]:
    """questions の各要素を answer ノードに並列ディスパッチする。"""
    return [Send("answer", {"question": q}) for q in state["questions"]]

def answer_node(state: SubState) -> dict:
    """1つの質問に対して回答する。"""
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answers": [f"Q: {state['question']}\nA: {response.content}"]}

graph = StateGraph(State)
graph.add_node("answer", answer_node)
graph.add_conditional_edges(START, dispatch, ["answer"])
graph.add_edge("answer", END)
app = graph.compile()

result = app.invoke(
    {
        "questions": [
            "東京の人口は？",
            "京都の有名な寺は？",
            "大阪のソウルフードは？",
        ],
        "answers": [],
    }
)
for ans in result["answers"]:
    print(ans)
    print("-" * 40)

$ python send_basics.py
Q: 東京の人口は？
A: 東京都全体の人口はおよそ1,400万人です。特別区（23区）の人口はおよそ960万人程度です。
（数値は国勢調査や東京都の推計人口に基づく概算です。最新の正確な人口は東京都公式サイトや総務省統計局の「国勢調査」・「人口推計」をご確認ください。最新値を取得してほしければ、確認してお伝えします。）
----------------------------------------
Q: 京都の有名な寺は？
A: 京都の有名な寺（代表的なもの）と簡単な説明です。
- 金閣寺（鹿苑寺）— 金箔で覆われた楼閣。北山・北部観光の定番。
- 銀閣寺（慈照寺）— 慎ましい東山文化の庭園・哲学の道に近い。
- 清水寺— 清水の舞台で有名。京都東山の大寺院、周辺に産寧坂・二年坂の散策路。
- 龍安寺— 禅の石庭（枯山水）が有名。静かな雰囲気。
（以下略）
----------------------------------------
Q: 大阪のソウルフードは？
A: 大阪のソウルフードと言えば、まず頭に浮かぶのは粉もんや屋台系の料理です。代表的なものをいくつか挙げます：
- たこ焼き：外はカリッと中はトロッ。ソース・マヨ・青のり・かつお節で。道頓堀や商店街に名店多数。
- お好み焼き（大阪風）：具を混ぜて鉄板で焼くタイプ。豚玉が定番。
- 串カツ（新世界）：串に刺した揚げ物。ソースの二度漬け禁止がルール。
（以下略）
----------------------------------------

"""デモ用の擬似検索ツール。

実運用では Tavily / Serper / Bing Search API などに差し替える。
"""

from langchain_core.tools import tool

# サブトピックに対して返す擬似的な検索結果
FAKE_RESULTS: dict[str, list[str]] = {
    "歴史": [
        "AIエージェントの研究は1950年代の記号論理学から始まり、1980年代のエキスパートシステム、2010年代の強化学習を経て、2020年代のLLMベースのエージェントへと発展した。",
        "2022年のChatGPT登場以降、LLMをコアに据えたエージェント研究が爆発的に増加。AutoGPT・BabyAGIといったOSSプロジェクトが大きな話題となった。",
    ],
    "アーキテクチャ": [
        "代表的なアーキテクチャパターンとして ReAct（Reason + Act）、Plan-and-Execute、Reflection、マルチエージェントが知られている。",
        "LangGraphやAutoGenなどのフレームワークがマルチエージェントの実装を簡素化している。",
    ],
    "ユースケース": [
        "カスタマーサポート、コーディング支援、リサーチ補助、データ分析、業務プロセス自動化など、実業務への適用が拡大している。",
        "AnthropicのClaude Code、OpenAIのOperator、GoogleのProject Astraなど、各社のエージェント製品が登場している。",
    ],
    "課題": [
        "ハルシネーション、コスト、レイテンシ、評価の難しさ、安全性とガバナンス、長期記憶の扱いなどが現在の主要な課題。",
        "特にマルチエージェント環境では、エージェント間の協調プロトコルや責任の追跡可能性が研究されている。",
    ],
    "今後": [
        "A2A（Agent-to-Agent）プロトコルの標準化、Computer Useによるブラウザ・OS操作、長期記憶の標準化、評価基盤の整備が進む見込み。",
        "2026年は『エージェント元年』と呼ばれ、企業導入の本格化が予想されている。",
    ],
}

@tool
def search(query: str) -> str:
    """Webから情報を検索します。"""
    for keyword, results in FAKE_RESULTS.items():
        if keyword in query:
            return "\n".join(f"- {r}" for r in results)
    return f"'{query}' に関する具体的な検索結果は得られませんでした。"

import operator
from typing import Annotated, TypedDict

from fake_search import search
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_litellm import ChatLiteLLM
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

planner_llm = ChatLiteLLM(model="openai/gpt-5.4")
researcher_llm = ChatLiteLLM(model="openai/gpt-5-mini")
synthesizer_llm = ChatLiteLLM(model="openai/gpt-5.1")

class State(TypedDict):
    question: str
    subtopics: list[str]
    findings: Annotated[list[str], operator.add]
    report: str

class SubState(TypedDict):
    subtopic: str

def plan_node(state: State) -> dict:
    """質問を3つのサブトピックに分解する。"""
    response = planner_llm.invoke(
        [
            SystemMessage(
                content=(
                    "ユーザーの質問を効率的に調査するため、3個のサブトピックに分解してください。"
                    "1行1個、番号や記号は付けない。"
                    "可能な限り次のいずれかの観点を含めてください: 歴史 / アーキテクチャ / ユースケース / 課題 / 今後。"
                )
            ),
            HumanMessage(content=state["question"]),
        ]
    )
    subtopics = [
        s.strip() for s in str(response.content).strip().split("\n") if s.strip()
    ]
    print(f"[Plan] サブトピック: {subtopics}")
    return {"subtopics": subtopics}

def dispatch(state: State) -> list[Send]:
    return [Send("researcher", {"subtopic": s}) for s in state["subtopics"]]

def researcher_node(state: SubState) -> dict:
    """1つのサブトピックを調査する。"""
    result = search.invoke({"query": state["subtopic"]})
    response = researcher_llm.invoke(
        [
            SystemMessage(content="検索結果を3行以内で要約してください。"),
            HumanMessage(
                content=f"サブトピック: {state['subtopic']}\n\n検索結果:\n{result}"
            ),
        ]
    )
    print(f"  [Researcher] {state['subtopic']}: 完了")
    return {"findings": [f"## {state['subtopic']}\n{response.content}"]}

def synthesize_node(state: State) -> dict:
    """findings を1つのレポートに統合する。"""
    findings_text = "\n\n".join(state["findings"])
    response = synthesizer_llm.invoke(
        [
            SystemMessage(
                content="複数の調査結果を統合し、Markdownレポートを作成してください。"
            ),
            HumanMessage(content=f"質問: {state['question']}\n\n{findings_text}"),
        ]
    )
    return {"report": str(response.content)}

graph = StateGraph(State)
graph.add_node("plan", plan_node)
graph.add_node("researcher", researcher_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "plan")
graph.add_conditional_edges("plan", dispatch, ["researcher"])
graph.add_edge("researcher", "synthesize")
graph.add_edge("synthesize", END)

app = graph.compile()

# =============================================
# 実行
# =============================================
result = app.invoke(
    {
        "question": "AIエージェントの歴史と現在のアーキテクチャについて教えてください",
        "subtopics": [],
        "findings": [],
        "report": "",
    }
)
print("\n=== レポート ===")
print(result["report"])

$ python parallel_research.py
[Plan] サブトピック: ['AIエージェントの歴史的発展：初期のルールベースAIやエキスパートシステムから、強化学習、LLMベースの自律エージェントへ至る流れ', '現在のAIエージェントのアーキテクチャ：LLMを中核にした計画・記憶・ツール利用・実行管理の構成要素とその役割', 'AIエージェントの代表的ユースケースと課題・今後：業務自動化、対話支援、ソフトウェア開発支援における活用例と、信頼性・安全性・評価・マルチエージェント化の展望']
  [Researcher] 現在のAIエージェントのアーキテクチャ：LLMを中核にした計画・記憶・ツール利用・実行管理の構成要素とその役割: 完了
  [Researcher] AIエージェントの歴史的発展：初期のルールベースAIやエキスパートシステムから、強化学習、LLMベースの自律エージェントへ至る流れ: 完了
  [Researcher] AIエージェントの代表的ユースケースと課題・今後：業務自動化、対話支援、ソフトウェア開発支援における活用例と、信頼性・安全性・評価・マルチエージェント化の展望: 完了

=== レポート ===
# AIエージェントの歴史と現在のアーキテクチャ概要レポート

## 1. AIエージェントの歴史的発展

### 1-1. 1950–1970年代：記号論理・推論システムの時代
- 1950年代のチューリングによる問題提起を起点に、**記号処理（Symbolic AI）** が主流に。
- 形式論理・推論エンジンを中心とした「推論マシン」。

### 1-2. 1980年代：エキスパートシステムとルールベースAI
- MYCIN、XCONなどが代表例。**if-thenルール**と**知識ベース**に専門知識を明示的に記述。

（中略）

### 1-5. 2020年代：LLMベースの自律エージェントへ
- 2022年の ChatGPT 公開を契機に、自然言語での対話やタスク遂行が一般ユーザーにも解放。
- AutoGPT・BabyAGIといったOSSプロジェクトが登場し、自律エージェント研究が加速。

## 2. 現在のAIエージェントのアーキテクチャ
- **LLM（推論・生成エンジン）** / **計画（Planner）** / **記憶（Memory）** / **ツール利用（Tool / Function Calling）** / **実行管理（Controller）** の5要素が中核。
- 代表パターン: ReAct / Plan-and-Execute / Reflection / マルチエージェント。
- フレームワーク: LangGraph / AutoGen など。

## 3. AIエージェントの代表的ユースケースと課題・今後
- 業務自動化、対話支援、ソフトウェア開発支援、リサーチ補助・データ分析。
- Anthropic（Claude Code）、OpenAI（Operator）、Google（Project Astra）などが製品投入。
- 課題: 信頼性（Hallucination）、安全性（権限管理・悪用防止）、評価手法、マルチエージェント協調。
（以下略）

"""LangGraphでDeep Researchエージェントを構築するサンプル。

フロー:
    [START] -> plan -> [Send] -> researcher (parallel x N)
                                       |
                                       v
                                  reflection -- needs_more --> [Send] -> researcher (loop)
                                       |
                                       v
                                  synthesize -> [END]
"""

import operator
from typing import Annotated, TypedDict

from fake_search import search
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_litellm import ChatLiteLLM
from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

# =============================================
# モデルの定義（ロールごとに最適なモデルを割り当て）
# =============================================
planner_llm = ChatLiteLLM(model="openai/gpt-5.4")
researcher_llm = ChatLiteLLM(model="openai/gpt-5-mini")
reflection_llm = ChatLiteLLM(model="openai/gpt-5.1")
# 最終統合だけClaudeに任せる（プロバイダーを跨いだ役割分担の例）
synthesizer_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

MAX_ITERATIONS = 2

# =============================================
# ステート定義
# =============================================
class Finding(TypedDict):
    subtopic: str
    content: str
    iteration: int

class ResearchState(TypedDict):
    question: str
    subtopics: list[str]
    findings: Annotated[list[Finding], operator.add]
    iteration: int
    needs_more: bool
    report: str

class SubResearchState(TypedDict):
    subtopic: str
    iteration: int

# =============================================
# Plan ノード
# =============================================
PLANNER_PROMPT = """\
あなたはリサーチプランナーです。
ユーザーの質問を効率的に調査するため、3〜5個の具体的なサブトピックに分解してください。
各サブトピックは独立して調査できる粒度にし、互いに重複しないようにしてください。
可能な限り次のいずれかの観点を含めてください: 歴史 / アーキテクチャ / ユースケース / 課題 / 今後。

出力形式: 1行に1つのサブトピックのみ。番号や記号は付けないでください。
"""

def plan_node(state: ResearchState) -> dict:
    """元の質問をサブトピックに分解する。"""
    response = planner_llm.invoke(
        [
            SystemMessage(content=PLANNER_PROMPT),
            HumanMessage(content=state["question"]),
        ]
    )
    subtopics = [
        s.strip() for s in str(response.content).strip().split("\n") if s.strip()
    ]
    print(f"\n[Plan] {len(subtopics)} 個のサブトピックを生成:")
    for s in subtopics:
        print(f"  - {s}")
    return {"subtopics": subtopics, "iteration": 1, "findings": []}

# =============================================
# Send による並列ディスパッチ
# =============================================
def dispatch_initial(state: ResearchState) -> list[Send]:
    return [
        Send("researcher", {"subtopic": s, "iteration": state["iteration"]})
        for s in state["subtopics"]
    ]

# =============================================
# Researcher ノード（並列実行）
# =============================================
RESEARCHER_PROMPT = """\
あなたは調査担当エージェントです。
search ツールで情報を集め、与えられたサブトピックについて事実ベースで簡潔にまとめてください。
推測は避け、検索で得られた情報のみを使ってください。
"""

def researcher_node(state: SubResearchState) -> dict:
    subtopic = state["subtopic"]
    search_result = search.invoke({"query": subtopic})
    response = researcher_llm.invoke(
        [
            SystemMessage(content=RESEARCHER_PROMPT),
            HumanMessage(
                content=f"サブトピック: {subtopic}\n\n検索結果:\n{search_result}\n\n上記をもとに3〜5行で簡潔にまとめてください。"
            ),
        ]
    )
    finding: Finding = {
        "subtopic": subtopic,
        "content": str(response.content).strip(),
        "iteration": state["iteration"],
    }
    print(f"  [Researcher#{state['iteration']}] {subtopic}: 完了")
    return {"findings": [finding]}

# =============================================
# Reflection ノード
# =============================================
REFLECTION_PROMPT = """\
あなたはリサーチの品質を判定する評価担当です。
これまでの調査結果が、元の質問に十分に答えられる量・質に達しているかを判定してください。

判定: SUFFICIENT（十分）または INSUFFICIENT（不足）。
INSUFFICIENT の場合は、追加で調査すべき新しいサブトピックを 1〜3個、改行区切りで列挙してください。
追加サブトピックには可能な限り次のいずれかの観点を含めてください: 歴史 / アーキテクチャ / ユースケース / 課題 / 今後。
SUFFICIENT の場合は判定だけで終わってください。

出力形式:
判定: SUFFICIENT または INSUFFICIENT
追加サブトピック:
（INSUFFICIENT のときのみ。1行1サブトピック。番号や記号なし。）
"""

def reflection_node(state: ResearchState) -> dict:
    findings_text = "\n\n".join(
        f"## {f['subtopic']}\n{f['content']}" for f in state["findings"]
    )
    response = reflection_llm.invoke(
        [
            SystemMessage(content=REFLECTION_PROMPT),
            HumanMessage(
                content=f"元の質問: {state['question']}\n\nこれまでの調査結果:\n{findings_text}"
            ),
        ]
    )
    text = str(response.content)
    is_insufficient = "INSUFFICIENT" in text.upper()

    if is_insufficient and state["iteration"] < MAX_ITERATIONS:
        lines = text.split("\n")
        new_subtopics: list[str] = []
        in_section = False
        for line in lines:
            stripped = line.strip()
            if "追加サブトピック" in stripped:
                in_section = True
                continue
            if in_section and stripped and "判定" not in stripped:
                new_subtopics.append(stripped.lstrip("- *・").strip())
        new_subtopics = [s for s in new_subtopics if s][:3]
        # パース失敗で追加トピックが0件のときは、無限ループを避けるため synthesize へ進める
        if not new_subtopics:
            print("\n[Reflection] INSUFFICIENT だが追加トピックが取得できず、レポート作成へ")
            return {"needs_more": False}
        print(f"\n[Reflection] INSUFFICIENT - 追加調査 {len(new_subtopics)} 件:")
        for s in new_subtopics:
            print(f"  - {s}")
        return {
            "subtopics": new_subtopics,
            "iteration": state["iteration"] + 1,
            "needs_more": True,
        }

    if is_insufficient:
        print(f"\n[Reflection] INSUFFICIENT だが MAX_ITERATIONS={MAX_ITERATIONS} に到達、レポート作成へ")
    else:
        print("\n[Reflection] SUFFICIENT - 調査完了、レポート作成へ")
    return {"needs_more": False}

def route_after_reflection(state: ResearchState):
    if state["needs_more"]:
        return [
            Send("researcher", {"subtopic": s, "iteration": state["iteration"]})
            for s in state["subtopics"]
        ]
    return "synthesize"

# =============================================
# Synthesizer ノード
# =============================================
SYNTHESIZER_PROMPT = """\
あなたはリサーチレポートの執筆担当です。
複数の調査結果を統合し、ユーザーの元の質問に対する読みやすいレポートを作成してください。

レポートの構成:
1. 概要（2〜3文）
2. 主要なポイント（見出し付きで複数）
3. まとめ

Markdown形式で記述してください。
"""

def synthesize_node(state: ResearchState) -> dict:
    findings_text = "\n\n".join(
        f"## {f['subtopic']} (iteration {f['iteration']})\n{f['content']}"
        for f in state["findings"]
    )
    response = synthesizer_llm.invoke(
        [
            SystemMessage(content=SYNTHESIZER_PROMPT),
            HumanMessage(
                content=f"元の質問: {state['question']}\n\n調査結果:\n{findings_text}"
            ),
        ]
    )
    print("\n[Synthesize] 最終レポートを生成しました")
    return {"report": str(response.content)}

# =============================================
# グラフ構築
# =============================================
graph = StateGraph(ResearchState)

graph.add_node("plan", plan_node)
graph.add_node("researcher", researcher_node)
graph.add_node("reflection", reflection_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "plan")
graph.add_conditional_edges("plan", dispatch_initial, ["researcher"])
graph.add_edge("researcher", "reflection")
graph.add_conditional_edges(
    "reflection",
    route_after_reflection,
    ["researcher", "synthesize"],
)
graph.add_edge("synthesize", END)

app = graph.compile()

# =============================================
# 実行
# =============================================
question = "AIエージェントの現状と今後の展望について調査してレポートを作成してください"

print(f"質問: {question}")
print("=" * 70)

result = app.invoke(
    {
        "question": question,
        "subtopics": [],
        "findings": [],
        "iteration": 0,
        "needs_more": False,
        "report": "",
    }
)

print("\n" + "=" * 70)
print("最終レポート")
print("=" * 70)
print(result["report"])
print("\n" + "=" * 70)
print(f"総調査件数: {len(result['findings'])} 件 / {result['iteration']} イテレーション")

def route_after_reflection(state: ResearchState):
    if state["needs_more"]:
        return [
            Send("researcher", {"subtopic": s, "iteration": state["iteration"]})
            for s in state["subtopics"]
        ]
    return "synthesize"

$ python deep_research.py
質問: AIエージェントの現状と今後の展望について調査してレポートを作成してください
======================================================================

[Plan] 5 個のサブトピックを生成:
  - AIエージェントの定義と発展の歴史、従来のチャットAIとの違い
  - AIエージェントの主要なアーキテクチャと構成要素（計画、ツール利用、記憶、実行、評価）
  - AIエージェントの現在のユースケースと導入が進む業界別の活用事例
  - AIエージェントの技術的・運用的・社会的な課題とリスク
  - AIエージェントの今後の展望、進化の方向性、市場・規制・働き方への影響
  [Researcher#1] AIエージェントの技術的・運用的・社会的な課題とリスク: 完了
  [Researcher#1] AIエージェントの今後の展望、進化の方向性、市場・規制・働き方への影響: 完了
  [Researcher#1] AIエージェントの主要なアーキテクチャと構成要素（計画、ツール利用、記憶、実行、評価）: 完了
  [Researcher#1] AIエージェントの現在のユースケースと導入が進む業界別の活用事例: 完了
  [Researcher#1] AIエージェントの定義と発展の歴史、従来のチャットAIとの違い: 完了

[Reflection] INSUFFICIENT - 追加調査 3 件:
  - 代表的なAIエージェントフレームワーク・プラットフォーム（LangChain, LangGraph, AutoGen, CrewAI など）の比較とアーキテクチャの違い
  - AIエージェントの評価手法・ベンチマークの現状（タスク完遂率、安全性評価、長期タスクでのパフォーマンスなど）と今後の課題
  - エージェント導入に伴う業務プロセス・働き方の変化の具体例と、その歴史的な自動化波との比較（RPA・従来のチャットボットとの違いを含む）
  [Researcher#2] AIエージェントの評価手法・ベンチマークの現状...: 完了
  [Researcher#2] 代表的なAIエージェントフレームワーク・プラットフォーム...: 完了
  [Researcher#2] エージェント導入に伴う業務プロセス・働き方の変化の具体例...: 完了

[Reflection] INSUFFICIENT だが MAX_ITERATIONS=2 に到達、レポート作成へ

[Synthesize] 最終レポートを生成しました

======================================================================
最終レポート
======================================================================
# AIエージェントの現状と今後の展望レポート

## 概要

AIエージェントは、大規模言語モデル（LLM）を中核に据え、自律的に計画・実行・評価を行う知的システムとして急速に進化しており、2022年のChatGPT登場以降、研究・実用化の両面で爆発的な成長を遂げている。2026年は「エージェント元年」と称され、企業導入の本格化が世界的に見込まれている。

## 主要なポイント

### 1. AIエージェントの発展の歴史と従来のAIとの違い
従来のチャットAIが「質問に答える」単発的な対話を中心としていたのに対し、AIエージェントは**自律的な計画立案・ツール利用・継続的な実行**を行える点で本質的に異なる。

### 2. 主要なアーキテクチャと構成要素

| パターン | 特徴 |
|---|---|
| **ReAct**（Reason + Act） | 推論と行動を交互に繰り返す基本パターン |
| **Plan-and-Execute** | 事前に計画を立て、順次実行する構造 |
| **Reflection** | 自己評価・修正を繰り返す反省型 |
| **マルチエージェント** | 複数エージェントが協調してタスクを処理 |

実装フレームワークとしては、LangChain・LangGraph・AutoGen・CrewAIなどが広く利用されている。

### 3. 現在のユースケースと業界別の活用事例
カスタマーサポート、コーディング支援（Claude Code）、リサーチ補助、データ分析、業務プロセス自動化などで導入が進む。Anthropic、OpenAI（Operator）、Google（Project Astra）が主要プレイヤー。

### 4. 技術的・運用的・社会的な課題とリスク
ハルシネーション、長期記憶、レイテンシ、評価手法の難しさ、安全性とガバナンス、責任の所在の明確化が主要課題。

### 5. 今後の展望と技術・社会への影響
- **A2A（Agent-to-Agent）プロトコルの標準化**
- **Computer Use（ブラウザ・OS操作）**
- **長期記憶の標準化**
- **評価基盤の整備**

## まとめ

AIエージェントは、LLMの急速な発展を背景に、単なる対話AIを超えた「自律的な知的作業者」として進化を続けている。
（以下略）
======================================================================
総調査件数: 8 件 / 2 イテレーション