用 GraphRAG 做多跳證據檢索：一個完整的實戰案例

• 14 January 2026
• LLM,
• RAG,
• GraphRAG,
• Evidence Selection,
• IR,
• Rust,
• Petgraph

用 GraphRAG 做多跳證據檢索：一個完整的實戰案例

問題：向量搜尋抓不到跨概念的完整答案

你有一個技術文件庫，用戶問「整個 RAG 流程是什麼？」

用純向量搜尋，你會拿到：

• 一段講 embedding 的
• 一段講 retrieval 的
• 一段講 generation 的

但它們各自獨立，沒有串起完整 pipeline。用戶只能自己拼湊，或期望 LLM 幫忙——這正是幻覺的來源。

GraphRAG 解決這個問題：用知識圖譜的關係邊，沿著「索引 → 檢索 → 生成」的路徑走訪，把跨概念的多跳上下文一次抓齊。

本文用一個完整案例帶你實作。

案例背景

輸入：

• 知識庫：技術文件、專案筆記
• 查詢：「整個 RAG 流程是什麼？」

輸出：

• 多跳證據子圖（涵蓋 embedding → retrieval → generation）
• 每個節點可追溯到原始文件
• 結構化的回答

Step 1：建構知識圖譜

從文件中抽取實體和關係，存入圖資料庫：

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from my_graph_rs import RustGraph  # Petgraph + PyO3 編譯的模組

# OpenRouter 設定
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
OPENROUTER_API_KEY = os.environ["OPENROUTER_API_KEY"]

class Entity(BaseModel):
    name: str
    type: str  # concept, tool, process, etc.

class Relation(BaseModel):
    source: str
    target: str
    relation: str  # depends_on, part_of, followed_by, etc.

class ExtractionResult(BaseModel):
    entities: list[Entity]
    relations: list[Relation]

EXTRACTION_PROMPT = """從以下文本中抽取實體和關係。

文本：
{text}

實體類型：concept, tool, process, component
關係類型：depends_on, part_of, followed_by, uses, produces
"""

def extract_graph(text: str) -> ExtractionResult:
    """從文本抽取知識圖譜"""
    llm = ChatOpenAI(
        model="xiaomi/mimo-v2-flash",
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
        temperature=0,
    )
    structured_llm = llm.with_structured_output(ExtractionResult)
    
    prompt = ChatPromptTemplate.from_template(EXTRACTION_PROMPT)
    chain = prompt | structured_llm
    
    return chain.invoke({"text": text})


def build_graph(documents: list[str]) -> RustGraph:
    """從多份文件建構圖譜（用 Rust Petgraph）"""
    graph = RustGraph()  # Rust 側建立空圖
    
    for doc in documents:
        result = extract_graph(doc)
        
        # 加入節點（Rust 側處理）
        for entity in result.entities:
            graph.add_node(entity.name, entity.type)
        
        # 加入邊（Rust 側處理）
        for rel in result.relations:
            graph.add_edge(rel.source, rel.target, rel.relation)
    
    return graph

Step 2：Seed Retrieval（種子檢索）

用向量搜尋找到與查詢相關的起始節點：

from openai import OpenAI
import numpy as np

# Embedding client (qwen3-embedding-8b via OpenRouter)
embed_client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.environ["OPENROUTER_API_KEY"],
)

def get_embedding(text: str) -> list[float]:
    """用 qwen3-embedding-8b 產生 768 維向量"""
    response = embed_client.embeddings.create(
        model="qwen/qwen3-embedding-8b",
        input=text,
        dimensions=768,
    )
    return response.data[0].embedding

def seed_retrieve(query: str, graph: RustGraph, top_k: int = 3) -> list[str]:
    """找到與查詢最相關的圖譜節點"""
    query_vec = get_embedding(query)
    nodes = graph.get_nodes()  # Rust 側取得節點列表
    
    # 計算每個節點與查詢的相似度
    scores = []
    for node in nodes:
        node_vec = get_embedding(node)
        similarity = np.dot(query_vec, node_vec) / (
            np.linalg.norm(query_vec) * np.linalg.norm(node_vec)
        )
        scores.append((node, similarity))
    
    # 取 top-k
    scores.sort(key=lambda x: x[1], reverse=True)
    return [node for node, _ in scores[:top_k]]

Step 3：Graph Expansion（圖擴散）

從種子節點沿著關係邊擴散，收集多跳上下文：

def expand_graph(
    graph: RustGraph, 
    seeds: list[str], 
    hops: int = 2
) -> RustGraph:
    """從種子節點擴散 n 跳，回傳子圖（Rust 側 BFS，比 Python loop 快 10-100x）"""
    return graph.expand_from_seeds(seeds, hops)

Step 4：Evidence Selection（證據選擇）

從子圖中選出支持回答的最小節點集：

from pydantic import BaseModel

class EvidenceNode(BaseModel):
    node: str
    relevance: str  # 為什麼這個節點重要
    source_doc: str  # 原始文件定位

class EvidenceSubgraph(BaseModel):
    nodes: list[EvidenceNode]
    reasoning_path: list[str]  # 推理路徑

SELECTION_PROMPT = """從以下子圖中選出回答問題所需的最小節點集。

問題：{query}

子圖節點：
{nodes}

子圖邊：
{edges}

要求：
1. 只選必要節點（最小充分）
2. 說明推理路徑
"""

def select_evidence(
    query: str, 
    subgraph: RustGraph
) -> EvidenceSubgraph:
    """選出最小證據子圖"""
    llm = ChatOpenAI(
        model="xiaomi/mimo-v2-flash",
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
        temperature=0,
    )
    structured_llm = llm.with_structured_output(EvidenceSubgraph)
    
    # 格式化子圖資訊（從 Rust 側取得）
    nodes_str = "\n".join(
        f"- {n['name']} ({n.get('type', 'unknown')})" 
        for n in subgraph.get_nodes_with_attrs()
    )
    edges_str = "\n".join(
        f"- {e['source']} --[{e['relation']}]--> {e['target']}" 
        for e in subgraph.get_edges_with_attrs()
    )
    
    prompt = ChatPromptTemplate.from_template(SELECTION_PROMPT)
    chain = prompt | structured_llm
    
    return chain.invoke({
        "query": query,
        "nodes": nodes_str,
        "edges": edges_str
    })

Step 5：Context Assembly + Generation

把證據子圖組裝成 prompt，生成回答：

def generate_answer(query: str, evidence: EvidenceSubgraph) -> str:
    """根據證據子圖生成回答"""
    llm = ChatOpenAI(
        model="xiaomi/mimo-v2-flash",
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
        temperature=0.3,
    )
    
    # 組裝上下文
    context = "## 推理路徑\n"
    context += " → ".join(evidence.reasoning_path)
    context += "\n\n## 證據節點\n"
    for node in evidence.nodes:
        context += f"- **{node.node}**: {node.relevance}\n"
    
    prompt = f"""根據以下證據回答問題。每個論點都要標注來源。

問題：{query}

{context}

請用結構化方式回答，標注每個論點的證據來源。"""
    
    return llm.invoke(prompt).content

Step 6：完整流程

def graphrag_query(query: str, documents: list[str]) -> str:
    """GraphRAG 完整查詢流程"""
    print(f"查詢：{query}\n")
    
    # 1. 建構圖譜
    print("Step 1: 建構知識圖譜...")
    graph = build_graph(documents)
    print(f"  節點數：{graph.node_count()}")
    print(f"  邊數：{graph.edge_count()}")
    
    # 2. 種子檢索
    print("\nStep 2: 種子檢索...")
    seeds = seed_retrieve(query, graph)
    print(f"  種子節點：{seeds}")
    
    # 3. 圖擴散
    print("\nStep 3: 圖擴散 (2 跳)...")
    subgraph = expand_graph(graph, seeds, hops=2)
    print(f"  子圖節點：{subgraph.get_nodes()}")
    
    # 4. 證據選擇
    print("\nStep 4: 證據選擇...")
    evidence = select_evidence(query, subgraph)
    print(f"  選中節點：{[n.node for n in evidence.nodes]}")
    print(f"  推理路徑：{' → '.join(evidence.reasoning_path)}")
    
    # 5. 生成回答
    print("\nStep 5: 生成回答...")
    answer = generate_answer(query, evidence)
    
    return answer


# 使用範例
if __name__ == "__main__":
    documents = [
        "RAG 由三個階段組成：索引、檢索、生成。",
        "索引階段將文件切 chunk 並用 embedding 模型轉成向量。",
        "檢索階段用查詢向量找相似的 chunk。",
        "生成階段將檢索結果餵給 LLM 產生回答。",
    ]
    
    answer = graphrag_query("整個 RAG 流程是什麼？", documents)
    print("\n" + "="*50)
    print(answer)

完整流程圖

flowchart TD
    Q[Query] --> S[Seed Retrieval]
    S --> X[Graph Expansion]
    X --> E[Evidence Selection]
    E --> A[Context Assembly]
    A --> G[Generate Answer]
    G --> V{Verify}
    V -->|incomplete| S
    V -->|ok| O[Answer with Citations]

實際輸出範例

查詢：整個 RAG 流程是什麼？

Step 1: 建構知識圖譜...
  節點數：8
  邊數：6

Step 2: 種子檢索...
  種子節點：['RAG', 'retrieval', 'generation']

Step 3: 圖擴散 (2 跳)...
  子圖節點：['RAG', 'indexing', 'retrieval', 'generation', 'embedding', 'LLM']

Step 4: 證據選擇...
  選中節點：['indexing', 'retrieval', 'generation']
  推理路徑：indexing → retrieval → generation

Step 5: 生成回答...

==================================================
## RAG 流程

RAG（Retrieval-Augmented Generation）由三個階段組成：

1. **索引（Indexing）**：將文件切 chunk 並用 embedding 模型轉成向量 [來源：indexing 節點]

2. **檢索（Retrieval）**：用查詢向量找相似的 chunk [來源：retrieval 節點]

3. **生成（Generation）**：將檢索結果餵給 LLM 產生回答 [來源：generation 節點]

附錄

A. 為什麼 GraphRAG 更像 AST？

關鍵是 IR + selection + verification 這種「可控推理流程」的結構。

B. 什麼時候用 GraphRAG？

C. 實現挑戰

D. 本文方案的適用規模

本文使用 Petgraph（Rust 記憶體圖 + PyO3 橋接）實作，適合小到中規模場景：

為什麼 Petgraph 適合更大規模？

• Rust 原生效能，記憶體布局緊湊，比 NetworkX 省 5-10x RAM
• BFS/DFS 擴散用 Rust 實作，比 Python loop 快 10-100x
• PyO3 橋接 overhead 極低（~20ns/call），幾乎保留原生 Rust 速度
• 預先索引 embedding，避免每次查詢都重算

E. 從 Petgraph 到生產的升級路徑

Petgraph 效能參考

[^memgraph]: Memgraph vs NetworkX PageRank
[^petgraph]: Petgraph on lib.rs
[^pyo3]: PyO3 Performance Discussion

注意：效能差距 10-100x 主要在大圖演算法（如 PageRank），小圖 < 1k nodes 時差距較小。PyO3 橋接的 overhead 極低（~20ns/call，佔 < 1%），Petgraph 在 Python 中幾乎保留 Rust 速度。

升級到 Petgraph 的範例（本文已使用）：

# 用 maturin 建立 Rust 擴充模組
# Cargo.toml: petgraph = "0.6", pyo3 = { version = "0.21", features = ["extension-module"] }

from my_graph_rs import RustGraph  # 編譯後的 Rust 模組

def expand_graph_petgraph(graph: RustGraph, seeds: list[str], hops: int = 2):
    """用 Rust Petgraph 做圖擴散，比 Python loop 快 10-100x"""
    # Rust 側實作 BFS，直接回傳子圖節點
    return graph.expand_from_seeds(seeds, hops)

升級到 Neo4j 的範例：

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687")

def expand_graph_neo4j(seeds: list[str], hops: int = 2):
    """用 Cypher 做圖擴散，適合持久化與複雜查詢"""
    query = """
    MATCH path = (start)-[*1..{hops}]-(end)
    WHERE start.name IN $seeds
    RETURN path
    """
    with driver.session() as session:
        return session.run(query, seeds=seeds, hops=hops)

F. 中規模的務實方案：分區 Petgraph

中規模場景不一定需要 Neo4j——如果你的文件有清晰的分類邊界（按年份、按主題、按部門），可以用多個獨立的 Petgraph 圖來處理，兼顧效能與開發彈性：

flowchart TD
    subgraph Shards["分區 Petgraph 圖"]
        G1["graph_2024
~100 docs"]
        G2["graph_2025
~100 docs"]
        G3["graph_tech
~100 docs"]
    end
    
    G1 --> Q1[獨立查詢]
    G2 --> Q2[獨立查詢]
    G3 --> Q3[獨立查詢]
    
    Q1 --> R[路由層合併結果]
    Q2 --> R
    Q3 --> R

實作範例：

from my_graph_rs import RustGraph  # Petgraph + PyO3 編譯的模組

class ShardedGraphRAG:
    def __init__(self):
        self.shards: dict[str, RustGraph] = {}
        self.entity_index: dict[str, list[str]] = {}  # 實體 → 所屬分區
    
    def add_shard(self, name: str, documents: list[str]):
        """新增一個分區（用 Rust Petgraph 建圖）"""
        self.shards[name] = build_graph_rust(documents)
        # 建立實體索引
        for node in self.shards[name].get_nodes():
            self.entity_index.setdefault(node, []).append(name)
    
    def query(self, question: str, target_shards: list[str] = None):
        """查詢指定分區，或自動判斷相關分區"""
        if target_shards is None:
            # 用 embedding 或 LLM 判斷相關分區
            target_shards = self._route_query(question)
        
        # 各分區獨立處理，合併結果（Rust 擴散快 10-100x）
        all_evidence = []
        for shard_name in target_shards:
            graph = self.shards[shard_name]
            seeds = seed_retrieve(question, graph)
            subgraph = graph.expand_from_seeds(seeds, hops=2)  # Rust BFS
            evidence = select_evidence(question, subgraph)
            all_evidence.append(evidence)
        
        return self._merge_evidence(all_evidence)

這種方式的優勢：

適用條件：

• ✅ 分類邊界清晰（按時間/主題/來源）
• ✅ 很少需要跨類別關聯查詢
• ✅ 願意維護一個簡單的路由/索引層
• ✅ 願意用 Rust + PyO3 建立擴充模組

何時升級到 Neo4j：

• 跨分區查詢變成常態（> 50% 的查詢需要跨區）
• 分區數量爆炸（> 20 個分區管理困難）
• 需要複雜的圖查詢模式（如：路徑分析、社群偵測）
• 需要持久化、ACID 事務、多用戶並發

G. 開源框架

• Microsoft GraphRAG：完整腳本，一次運行
• LlamaIndex：步驟式建置，整合 Neo4j
• Petgraph：Rust 高效能圖處理，搭配 PyO3 + Maturin 可無縫整合 Python

本文基於 Career Knowledge Base 專案的 GraphRAG 實驗，使用 Rust Petgraph + PyO3 + Python + LangChain 建構。

• ← Previous
  用 LLM 建立職涯技能圖譜：做出類似 GraphRAG 的能力
• Next →
  2026 年主流 RAG 架構：從 Hybrid 到 Agentic / Graph 的工程化落地