Home 論文解説: MindSearch — DAGベース並列検索エージェントによるマルチソース情報統合
投稿
キャンセル

📄 論文解説: MindSearch — DAGベース並列検索エージェントによるマルチソース情報統合

投稿日 22/02/2026
投稿者 0h-n0
16 分読了時間

論文概要（Abstract）

MindSearchは、人間の認知過程を模倣したマルチエージェント型Web検索フレームワークです。WebPlanner（DAG構造でサブクエリを計画）とWebSearcher（階層的情報検索を実行）の2エージェント構成で、複雑な質問に対して人間が3時間かかるタスクを3分で完遂します。HotpotQA（マルチホップQA）でReActベースラインを+7.0% EM上回り、ChatGPT-WebやPerplexity.aiを凌駕する性能を達成しています。

この記事は Zenn記事: LangGraph Agentic RAGの本番運用設計：マルチソースルーティングと評価駆動リランキング の深掘りです。

情報源

背景と動機（Background & Motivation）

複雑なWeb検索タスク（例：「電気自動車と水素燃料電池の環境影響を比較せよ」）は、以下の認知能力を要求します。

  1. 分解: 複雑な質問をサブクエリに分割する
  2. 並列実行: 独立したサブクエリを同時に検索する
  3. 依存関係管理: あるサブクエリの結果が別のサブクエリの入力になる
  4. 統合: 複数ソースの情報を矛盾なく統合する

既存アプローチの限界は明確です。単発RAGはマルチホップ問題に対応できず、Chain-of-Thought検索は逐次処理で遅く、ReActエージェントは計画能力が限定的です。

Zenn記事で解説したSend()APIによるマルチソースルーティングは「どのソースに振り分けるか」を扱いますが、MindSearchはそもそも何を検索すべきかの計画と並列実行を扱います。この2つは相補的であり、MindSearchのDAG計画をLangGraphのグラフ構造で実装するのが自然な統合パターンです。

主要な貢献（Key Contributions）

  • DAGベースの認知グラフ計画（WebPlanner）: サブクエリの依存関係をDAGで明示化し、人間の調査プロセスを模倣
  • 階層的WebSearcher: Web検索→ページ読み取り→要約の3段階で各サブクエリを処理
  • 並列実行による4倍速化: 独立したサブクエリをasyncio.gatherで同時実行
  • オープンソースで小規模モデル対応: InternLM2.5-7B-Chat（4-bit量子化）でGPT-4oに匹敵する性能
  • 商用検索エンジンを凌駕: ChatGPT-WebとPerplexity.aiを複数ベンチマークで上回る

技術的詳細（Technical Details）

システムアーキテクチャ

MindSearchは2つの協調エージェントで構成されます。

1
2
3
4
5
6
7
8
9
10
11

User Query
     |
     v
[WebPlanner] — DAG G = (V, E) を管理
     |
     |-- 並列 --> [WebSearcher_1] (サブクエリ1)
     |-- 並列 --> [WebSearcher_2] (サブクエリ2)
     |-- 待機 --> [WebSearcher_3] (サブクエリ3: 1の結果に依存)
     |
     v
統合された最終回答

WebPlanner: DAG構造の検索計画

WebPlannerは、情報要求をDAG（Directed Acyclic Graph）として構築します。

グラフ状態の定義:

時刻$t$でのグラフ: $G_t = (V_t, E_t)$

各ノード$v_i \in V$は以下の属性を持ちます。

  • query: 検索すべきサブクエリ
  • state $\in {\text{PENDING}, \text{SEARCHING}, \text{DONE}}$
  • content: 検索結果（WebSearcher完了後に設定）

有向辺$(v_i, v_j) \in E$は、$v_j$が$v_i$の結果に依存することを表します。

フロンティアの定義:

\[\mathcal{F}_t = \{v \in V_t \mid \text{state}(v) = \text{PENDING} \wedge \forall u \in \text{pred}(v): \text{state}(u) = \text{DONE}\}\]

フロンティアは、全依存ノードが完了済みかつ自身が未処理のノード集合です。

計画ループ:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

import asyncio
from dataclasses import dataclass, field

@dataclass
class SearchNode:
    """DAG内の検索ノード"""
    node_id: str
    query: str
    state: str = "PENDING"  # PENDING, SEARCHING, DONE
    content: str = ""
    depends_on: list[str] = field(default_factory=list)

class WebPlanner:
    """DAGベースの検索計画エージェント"""
    def __init__(self, llm, max_nodes: int = 12, max_depth: int = 4):
        self.llm = llm
        self.max_nodes = max_nodes
        self.max_depth = max_depth
        self.nodes: dict[str, SearchNode] = {}
        self.depth = 0

    def get_frontier(self) -> list[SearchNode]:
        """実行可能なフロンティアノードを取得"""
        return [
            n for n in self.nodes.values()
            if n.state == "PENDING"
            and all(
                self.nodes[dep].state == "DONE"
                for dep in n.depends_on
            )
        ]

    async def plan_and_execute(
        self, query: str, searcher: "WebSearcher"
    ) -> str:
        """DAGを構築しながら検索を実行"""
        # 初期ノード生成
        initial_nodes = self._decompose_query(query)
        for node in initial_nodes:
            self.nodes[node.node_id] = node

        while True:
            frontier = self.get_frontier()
            if not frontier:
                break  # 全ノード完了
            if self.depth >= self.max_depth:
                break  # 深さ上限

            # フロンティアを並列実行
            results = await asyncio.gather(*[
                searcher.search(
                    n.query,
                    context=self._get_dependency_context(n)
                )
                for n in frontier
            ])

            for node, result in zip(frontier, results):
                node.content = result
                node.state = "DONE"

            # 新しいサブクエリが必要か判断
            new_nodes = self._expand_graph(query)
            for node in new_nodes:
                if len(self.nodes) < self.max_nodes:
                    self.nodes[node.node_id] = node

            self.depth += 1

        return self._synthesize_answer(query)

終了条件:

  1. 全ノードがDONE状態
  2. 最大深度（デフォルト4）到達
  3. 最大ノード数（デフォルト12）到達

WebSearcher: 階層的情報検索

各サブクエリに対して3段階の検索を実行します。

Level 1 — Web検索: 検索APIで上位10件のURL+スニペットを取得

Level 2 — ページ読み取り: 上位3ページのHTMLを取得し、256トークンのチャンクに分割。各チャンクのスコアを計算：

\[\text{score}(s) = \cos(\text{embed}(s), \text{embed}(q_{\text{sub}}))\]

上位チャンクを最大4096トークンまで抽出。

Level 3 — 要約合成: LLMが抽出チャンクを512トークンの構造化回答に要約。ソースURLも記録。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

class WebSearcher:
    """階層的Web検索エージェント"""
    def __init__(self, llm, search_api, embedding_model):
        self.llm = llm
        self.search_api = search_api
        self.embedding_model = embedding_model

    async def search(
        self, query: str, context: str = ""
    ) -> str:
        """3段階の階層的検索"""
        # Level 1: Web検索
        search_results = await self.search_api.search(
            query, max_results=10
        )

        # Level 2: ページ読み取り + チャンクスコアリング
        top_urls = search_results[:3]
        all_chunks = []
        for url in top_urls:
            page_content = await self._fetch_page(url)
            chunks = self._chunk_text(page_content, size=256)
            scored = [
                (c, self._compute_similarity(query, c))
                for c in chunks
            ]
            all_chunks.extend(scored)

        # 上位チャンク選択（4096トークン以内）
        all_chunks.sort(key=lambda x: x[1], reverse=True)
        selected = self._select_within_budget(all_chunks, 4096)

        # Level 3: 要約合成
        context_text = "\n".join([c[0] for c in selected])
        summary = self.llm.invoke(
            f"Context:\n{context}\n{context_text}\n\n"
            f"Sub-question: {query}\n"
            f"Provide a comprehensive answer with citations."
        )
        return summary.content

並列実行の効果

独立したサブクエリをasyncio.gatherで同時実行することで、逐次処理と比較して最大4倍の速度向上を達成しています。

方式複雑なクエリの処理時間
人間の専門家~3時間
ReAct（逐次）~12分
MindSearch（並列）~3分

実装のポイント（Implementation）

LangGraphとの統合

MindSearchのDAG計画は、LangGraphのStateGraphとSend()で自然に実装できます。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

def plan_and_dispatch(state: MindSearchState) -> list[Send]:
    """WebPlannerの計画結果からWebSearcherにディスパッチ"""
    frontier = get_frontier_nodes(state["dag"])
    return [
        Send("web_searcher", {
            "sub_query": node.query,
            "context": get_dependency_context(node, state["dag"]),
            "node_id": node.node_id,
        })
        for node in frontier
    ]

# LangGraph構築
workflow = StateGraph(MindSearchState)
workflow.add_node("planner", plan_and_dispatch)
workflow.add_node("web_searcher", web_searcher_node)
workflow.add_node("synthesizer", synthesize_results)
workflow.add_edge(START, "planner")
workflow.add_conditional_edges("planner", plan_and_dispatch,
    ["web_searcher"])
workflow.add_edge("web_searcher", "planner")  # ループ
workflow.add_edge("planner", "synthesizer")  # 終了条件時
workflow.add_edge("synthesizer", END)

設定パラメータ

パラメータデフォルト値推奨範囲
max_nodes128-15
max_depth43-5
search_results_per_query105-15
pages_read_per_search32-5
max_page_length4096トークン2048-8192
chunk_size256トークン128-512
summary_max_length512トークン256-1024

よくある落とし穴

  • 冗長サブクエリ: WebPlannerが重複するサブクエリを生成することがある。プロンプトで「既に検索済みの内容を確認してから新しいサブクエリを生成」と明記
  • 検索API制限: Brave/Google APIのレート制限に注意。並列実行時にThrottlingが発生する場合はセマフォで制御
  • 最大ノード数のチューニング: 12ノードでは5ホップ以上の深い推論に不足する場合がある。ただし増やすとコストが線形増加

実験結果（Results）

マルチホップQAベンチマーク

手法HotpotQA EMHotpotQA F12WikiMultiHop EMMuSiQue EM
ChatGPT-Web0.4120.5570.3410.198
Perplexity.ai0.4380.5780.3620.214
ReAct (GPT-4o)0.4510.5890.3780.223
MindSearch (GPT-4o)0.5210.6610.4470.289
MindSearch (InternLM2.5-7B)0.4870.6310.4210.261

同じGPT-4oバックボーンでMindSearchはReActを+7.0% EM（HotpotQA）上回り、差はDAG計画の効果に帰着します。

オープンエンド質問（人間評価、N=100）

手法深さ幅事実性総合
ChatGPT-Web3.213.143.453.27
Perplexity.ai3.483.523.613.54
MindSearch (GPT-4o)4.124.234.084.14

3名の独立評価者による5段階評価。MindSearchは特に幅（4.23）で優位であり、マルチソース統合の効果が顕著です。

アブレーション

構成HotpotQA EMオープンQA
Full MindSearch0.5214.14
DAG→線形チェーン0.478 (-4.3%)3.87
並列→逐次実行0.501 (-2.0%)3.98
階層検索→単層検索0.492 (-2.9%)3.91

DAG計画が最大の貢献（+4.3% EM）。並列実行は品質にも影響し（+2.0% EM）、これは時間制約内でより多くの情報を統合できるためです。

実運用への応用（Practical Applications）

MindSearchの知見はZenn記事のマルチソースルーティングと以下のように統合できます。

  1. クエリ分解の自動化: LangGraphのclassify_queryの前段階に、MindSearch的なクエリ分解を導入。複合クエリを複数のサブクエリに分割
  2. 並列ルーティング: 分解されたサブクエリをSend()で並列にルーティング。各サブクエリに最適なリトリーバーを割り当て
  3. 結果統合: MindSearchの階層的要約をフォールバック戦略に組み込み、検索結果の品質が低い場合にクエリを再分解

関連研究（Related Work）

  • ReAct (Yao et al., 2023): 推論と行動の交互実行。MindSearchのDAG計画が大幅に上回る
  • Self-Ask (Press et al., 2022): 反復的な質問分解。並列実行なし
  • WebGPT (Nakano et al., 2021): LLMによるWeb検索。MindSearchのマルチエージェント構成がより柔軟

まとめと今後の展望

MindSearchは、DAGベースの認知グラフ計画により、マルチソース検索の品質と速度を同時に改善しました。7Bモデルでも商用検索エンジンに匹敵する性能を達成しており、オンプレミスでのAgentic RAGシステム構築に実用的な選択肢を提供します。

今後の課題として、マルチモーダル検索（画像・動画）の統合、セッション間のメモリ持続、プライベートナレッジベースとの連携が挙げられています。

参考文献

blog paper arxiv
multi-agent search DAG RAG parallel-execution langgraph agentic
この投稿は CC BY 4.0 でライセンスされています。
最近の更新
目次

論文解説: Speculative RAG — 並列ドラフト生成によるRAG高速化と精度向上の同時達成

論文解説: MCTS-RAG — モンテカルロ木探索で小規模LMの検索拡張推論を飛躍的に強化