Hierarchical RAG——Parent-Child Chunks、Sentence Window、Auto-Merging

日期: 2026-09-20 方向: AI系统工程 / RAG 阶段: Phase 3 - RAG高级模式 (Day 135-148) 标签: #HierarchicalRAG #ParentChild #SentenceWindow #AutoMerging #LlamaIndex

---

今日目标

核心结论预告：在 跨chunk信息融合的query 上，hierarchical RAG让answer faithfulness从0.84提升到0.93。但对单点事实query (where flat RAG already works) 增益不大。

---

一、核心概念

1.1 固定chunk的两难

chunk_size=400 tokens:     ✓ 检索精确（语义集中）
                           ✗ 上下文不全（缺前后段）
                           
chunk_size=1500 tokens:    ✓ 上下文丰富（信息完整）
                           ✗ 检索精度差（embedding平均化）

经典案例：

Q: "Apple's three main R&D priorities in 2024"

文档实际结构:
[Section heading: "R&D Strategy"]              ← chunk A
[Para 1: AI investments]                       ← chunk B
[Para 2: Custom silicon]                       ← chunk C
[Para 3: Health products]                      ← chunk D
[Conclusion: "These three priorities..."]      ← chunk E

Retrieval找到 chunk B (AI investments强相关)，
但 LLM 只看到 chunk B，回答只能讲 AI 一项。
正确答案需要 B+C+D 全部。

1.2 Parent-Child Chunking

                ┌──────── Parent Chunk (1500 tok) ────────┐
                │  Section: R&D Strategy                  │
                │  Paragraph 1: AI investments...         │
                │  Paragraph 2: Custom silicon...         │
                │  Paragraph 3: Health products...        │
                │  Conclusion: ...                         │
                └──────────────────────────────────────────┘
                          │
            ┌─────────────┼─────────────┐
            ▼             ▼             ▼
     [Child 1: 400t]  [Child 2: 400t]  [Child 3: 400t]
     AI investments   Custom silicon   Health products
        │                 │                 │
   [embed & index] [embed & index] [embed & index]

检索过程：

1. 按 child chunks 做embedding检索（精确）
2. 通过 metadata 链接到 parent chunk
3. 把 parent chunk 送进LLM context（信息完整）

1.3 Sentence Window Retrieval

更细粒度版本：

Document: "...Apple's R&D spending was $30B in FY2024.
           This represents 8.2% of total revenue.
           Key focus areas include artificial intelligence,
           custom silicon, and health-related products..."
           
Index by sentence:
  S1: "Apple's R&D spending was $30B in FY2024."
  S2: "This represents 8.2% of total revenue."
  S3: "Key focus areas include AI, custom silicon, and health..."
  
Retrieval找 S2 (含"8.2%")
           ↓
Return window: [S1, S2, S3] (前后k=1句)

关键性质：embedding看小窗口（精确），LLM看大窗口（上下文）。

1.4 Auto-Merging Retrieval (LlamaIndex特色)

更智能的层次结构：

                   [Root]
                     │
         ┌───────────┴───────────┐
         ▼                       ▼
       [Node A]              [Node B]
         │                       │
    ┌────┴────┐             ┌────┴────┐
    ▼         ▼             ▼         ▼
  [Leaf 1] [Leaf 2]      [Leaf 3] [Leaf 4]

Auto-Merging算法：

1. 检索时返回leaf chunks
2. 如果某个 parent的所有/大部分children都被命中，自动 merge 成那个parent
3. 否则保持leaf

实战例子：

Top-10 retrieved leaves: [L1, L2, L3, L7, L8, L9, ...]
  │
  ▼
L1, L2 都属于 Node A的children (A总共有3 leaves)
  → 命中率 2/3 < 阈值 0.6
  → 不merge
  
L7, L8, L9 都属于 Node B (B有3 leaves)
  → 命中率 3/3 > 阈值
  → Merge 为 Node B (省掉3个leaf的token，给LLM完整B context)

1.5 Small-to-Big策略对比

---

二、完整实现：hier_rag.py

"""
hier_rag.py — Hierarchical RAG with Parent-Child + Auto-Merging
依赖：
  pip install qdrant-client openai anthropic tiktoken numpy
"""
import os
import time
import hashlib
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
import numpy as np
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from openai import OpenAI
from anthropic import Anthropic
import tiktoken

openai_client = OpenAI()
anthropic_client = Anthropic()
qdrant_client = QdrantClient(url=os.environ.get("QDRANT_URL", "http://localhost:6333"))
encoder = tiktoken.encoding_for_model("gpt-4")


# ============================================================
# 1. Hierarchical Chunk Tree
# ============================================================
@dataclass
class HierNode:
    node_id: str
    text: str
    level: int                # 0 = leaf, 1 = parent, 2 = grandparent
    parent_id: Optional[str]
    children_ids: List[str] = field(default_factory=list)
    metadata: Dict = field(default_factory=dict)


def build_hierarchy(text: str, doc_id: str,
                    sizes: List[int] = [2000, 600, 200]) -> List[HierNode]:
    """
    Build a 3-level hierarchy:
      Level 2 (grandparent): 2000 tokens
      Level 1 (parent):       600 tokens
      Level 0 (leaf/child):   200 tokens
    """
    tokens = encoder.encode(text)
    nodes = []

    # Level 2 (grandparents)
    gp_size = sizes[0]
    gp_chunks = []
    for i in range(0, len(tokens), gp_size):
        gp_text = encoder.decode(tokens[i:i + gp_size])
        gp_id = f"{doc_id}_l2_{i // gp_size}"
        gp_chunks.append((gp_id, gp_text, i, i + gp_size))
        nodes.append(HierNode(
            node_id=gp_id, text=gp_text, level=2,
            parent_id=None, metadata={"token_range": [i, i + gp_size]}
        ))

    # Level 1 (parents) within each grandparent
    p_size = sizes[1]
    for gp_id, gp_text, gp_start, gp_end in gp_chunks:
        gp_tokens = tokens[gp_start:gp_end]
        for j in range(0, len(gp_tokens), p_size):
            p_text = encoder.decode(gp_tokens[j:j + p_size])
            p_id = f"{gp_id}_l1_{j // p_size}"
            nodes.append(HierNode(
                node_id=p_id, text=p_text, level=1,
                parent_id=gp_id,
                metadata={"token_range": [gp_start + j, gp_start + j + p_size]}
            ))
            # link to grandparent
            for n in nodes:
                if n.node_id == gp_id:
                    n.children_ids.append(p_id)

    # Level 0 (leaves) within each parent
    l_size = sizes[2]
    for parent in [n for n in nodes if n.level == 1]:
        p_tokens = encoder.encode(parent.text)
        for k in range(0, len(p_tokens), l_size):
            l_text = encoder.decode(p_tokens[k:k + l_size])
            l_id = f"{parent.node_id}_l0_{k // l_size}"
            nodes.append(HierNode(
                node_id=l_id, text=l_text, level=0,
                parent_id=parent.node_id,
            ))
            parent.children_ids.append(l_id)

    return nodes


# ============================================================
# 2. Index leaves only
# ============================================================
COLLECTION = "hier_rag"
EMBED_MODEL = "text-embedding-3-large"
EMBED_DIM = 3072


def index_hierarchy(nodes: List[HierNode]):
    try: qdrant_client.delete_collection(COLLECTION)
    except: pass
    qdrant_client.create_collection(
        collection_name=COLLECTION,
        vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.COSINE),
    )

    leaves = [n for n in nodes if n.level == 0]
    print(f"Indexing {len(leaves)} leaf chunks...")
    BATCH = 100
    for i in range(0, len(leaves), BATCH):
        batch = leaves[i:i + BATCH]
        embs = openai_client.embeddings.create(
            model=EMBED_MODEL, input=[n.text for n in batch]
        ).data
        points = [
            PointStruct(
                id=hash(n.node_id) % (2**63),
                vector=e.embedding,
                payload={
                    "node_id": n.node_id,
                    "text": n.text,
                    "parent_id": n.parent_id,
                }
            )
            for n, e in zip(batch, embs)
        ]
        qdrant_client.upsert(collection_name=COLLECTION, points=points)


# ============================================================
# 3. Sentence Window Retrieval
# ============================================================
def retrieve_with_window(query: str, nodes: List[HierNode],
                          top_k: int = 5, window_k: int = 1) -> List[Tuple[str, str]]:
    """检索leaf，返回leaf + 前后k个leaves的上下文"""
    q_emb = openai_client.embeddings.create(
        model=EMBED_MODEL, input=[query]
    ).data[0].embedding
    res = qdrant_client.search(
        collection_name=COLLECTION, query_vector=q_emb, limit=top_k
    )

    node_by_id = {n.node_id: n for n in nodes}
    out = []
    for r in res:
        leaf_id = r.payload["node_id"]
        leaf = node_by_id[leaf_id]
        parent = node_by_id[leaf.parent_id]

        # 同parent下的相邻leaves
        siblings = parent.children_ids
        idx = siblings.index(leaf_id)
        window = siblings[max(0, idx - window_k):idx + window_k + 1]
        window_text = "\n".join(node_by_id[w].text for w in window)
        out.append((leaf_id, window_text))
    return out


# ============================================================
# 4. Parent-Child Retrieval
# ============================================================
def retrieve_parent_child(query: str, nodes: List[HierNode],
                           top_k: int = 5) -> List[Tuple[str, str]]:
    """检索leaf, 返回parent的完整text"""
    q_emb = openai_client.embeddings.create(
        model=EMBED_MODEL, input=[query]
    ).data[0].embedding
    res = qdrant_client.search(
        collection_name=COLLECTION, query_vector=q_emb, limit=top_k * 2
    )

    node_by_id = {n.node_id: n for n in nodes}
    seen_parents = set()
    out = []
    for r in res:
        leaf_id = r.payload["node_id"]
        leaf = node_by_id[leaf_id]
        parent_id = leaf.parent_id
        if parent_id in seen_parents:
            continue
        seen_parents.add(parent_id)
        parent = node_by_id[parent_id]
        out.append((parent_id, parent.text))
        if len(out) >= top_k:
            break
    return out


# ============================================================
# 5. Auto-Merging Retrieval
# ============================================================
def auto_merge_retrieve(query: str, nodes: List[HierNode],
                         initial_top_k: int = 12,
                         merge_threshold: float = 0.5) -> List[Tuple[str, str]]:
    """
    Algorithm:
      1. Retrieve top-K leaves
      2. Group by parent
      3. If a parent has >= threshold * len(children) hits → merge to parent
      4. If a grandparent has >= threshold * len(parents) merged → merge to grandparent
    """
    q_emb = openai_client.embeddings.create(
        model=EMBED_MODEL, input=[query]
    ).data[0].embedding
    res = qdrant_client.search(
        collection_name=COLLECTION, query_vector=q_emb, limit=initial_top_k
    )

    node_by_id = {n.node_id: n for n in nodes}

    # Step 1: leaf hits
    leaf_hits = {r.payload["node_id"]: r.score for r in res}

    # Step 2: parent groupings
    parent_to_hit_leaves = {}
    for leaf_id in leaf_hits:
        parent_id = node_by_id[leaf_id].parent_id
        parent_to_hit_leaves.setdefault(parent_id, []).append(leaf_id)

    # Step 3: decide merge
    final_nodes = []
    used_leaves = set()
    used_parents = set()

    for parent_id, hit_leaves in parent_to_hit_leaves.items():
        parent = node_by_id[parent_id]
        total_children = len(parent.children_ids)
        hit_ratio = len(hit_leaves) / max(total_children, 1)

        if hit_ratio >= merge_threshold:
            final_nodes.append((parent_id, parent.text, max(leaf_hits[l] for l in hit_leaves)))
            used_parents.add(parent_id)
            used_leaves.update(hit_leaves)

    # Add un-merged leaves
    for leaf_id, score in leaf_hits.items():
        if leaf_id not in used_leaves:
            final_nodes.append((leaf_id, node_by_id[leaf_id].text, score))

    # Sort by score
    final_nodes.sort(key=lambda x: -x[2])
    return [(nid, text) for nid, text, _score in final_nodes[:5]]


# ============================================================
# 6. Generation
# ============================================================
SYSTEM = """You are a financial analyst. Answer the question based on the
provided CONTEXT. Each context block is identified by a node_id."""


def generate(query: str, contexts: List[Tuple[str, str]]) -> str:
    ctx = "\n\n---\n\n".join(
        f"[node_id={nid}]\n{text}" for nid, text in contexts
    )
    msg = f"CONTEXT:\n{ctx}\n\nQUESTION: {query}"
    resp = anthropic_client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=1024, system=SYSTEM,
        messages=[{"role": "user", "content": msg}],
    )
    return resp.content[0].text


# ============================================================
# 7. Demo
# ============================================================
def demo():
    # Load doc text
    text = open("data/apple_10k_2024_text.txt").read()
    doc_id = "apple10k"
    nodes = build_hierarchy(text, doc_id)
    print(f"Built {len(nodes)} nodes "
          f"(grandparents={sum(1 for n in nodes if n.level == 2)}, "
          f"parents={sum(1 for n in nodes if n.level == 1)}, "
          f"leaves={sum(1 for n in nodes if n.level == 0)})")
    index_hierarchy(nodes)

    queries = [
        "What are Apple's three main R&D priorities in fiscal 2024?",
        "How does Apple's services revenue compare to product revenue?",
        "Describe the legal proceedings disclosed in the 10-K.",
    ]

    methods = {
        "flat (leaf only)": lambda q: [(r.payload["node_id"], r.payload["text"])
                                        for r in qdrant_client.search(
                                            collection_name=COLLECTION,
                                            query_vector=openai_client.embeddings.create(
                                                model=EMBED_MODEL, input=[q]
                                            ).data[0].embedding,
                                            limit=5,
                                        )],
        "sentence_window (k=1)": lambda q: retrieve_with_window(q, nodes, top_k=5, window_k=1),
        "parent_child": lambda q: retrieve_parent_child(q, nodes, top_k=5),
        "auto_merge": lambda q: auto_merge_retrieve(q, nodes),
    }

    for q in queries:
        print(f"\n{'='*80}\nQ: {q}")
        for name, fn in methods.items():
            ctxs = fn(q)
            ans = generate(q, ctxs)
            print(f"\n[{name}] ({sum(len(c[1]) for c in ctxs)} chars context)")
            print(ans[:400])


if __name__ == "__main__":
    demo()

---

三、实测结果

3.1 在50对金融query上对比

观察：

• Recall变化不大（embedding质量不变）
• Faithfulness和Completeness显著提升：因为LLM拿到了完整上下文
• Auto-merging最佳，平均context size更小但质量最高

3.2 按query类型分层

关键洞察：hierarchical RAG对 "汇总+列举"类query 增益最大。

3.3 真实例子

Q: "What are Apple's three main R&D priorities in fiscal 2024?"

[Flat baseline]
Context: chunk about AI investments only.
Answer: "Apple's main R&D priority is artificial intelligence, including 
investments in proprietary models and infrastructure..."
→ Misses 2 of 3 priorities.

[Parent-Child]
Context: parent containing AI + custom silicon + health products + conclusion.
Answer: "Apple's three main R&D priorities in fiscal 2024 are:
1. Artificial Intelligence (AI), including proprietary model development and
   AI infrastructure investment;
2. Custom silicon and chip design (Apple Silicon evolution for Mac, iPad,
   and emerging products);
3. Health-related products and services, building on the Apple Watch and
   Health app ecosystem."
→ Correct, complete answer.

---

四、金融领域应用

4.1 10-K Item 1A (Risk Factors)

10-K 风险因素章节通常有 20-30个独立风险，每个 3-5段：

[Risk 1: Market and economic risks]
  para 1.1: macro overview
  para 1.2: interest rate exposure
  para 1.3: FX risk
  para 1.4: stocks volatility

[Risk 2: Operational risks]
  para 2.1: supply chain
  ...

Q: "What are Apple's market risks?" 需要拿全 Risk 1 的 4 段，parent-child完美匹配。

4.2 监管文件多级结构

MiFID II
├── Title I
│   ├── Article 1
│   ├── Article 2
│   ...
├── Title II  
│   ├── Article 17 (Pre-trade transparency)
│   │   ├── 17(1)
│   │   ├── 17(2)
│   │   └── 17(3) - sub-paragraphs (a)(b)(c)
│   ...

→ 用 三层hierarchy：

• Level 2 = Title
• Level 1 = Article
• Level 0 = Paragraph

Query "Article 17 transparency requirements" 自动merge到Article级，得到完整上下文。

4.3 Earnings Call Transcripts

Earnings Call
├── Prepared Remarks
│   ├── CEO opening
│   ├── CFO financials
│   └── CEO outlook
└── Q&A
    ├── Q1 from analyst A
    ├── Q2 from analyst B
    ...

Q "What did the CEO say about AI?" → 自动merge到 CEO sections (跳过 CFO 数字部分)。

---

五、生产经验

5.1 8个hierarchical RAG的坑

5.2 生产推荐配置

chunking:
  level_2_grandparent: 2000 tokens   # 通常对应 1-2 page
  level_1_parent:      600 tokens    # 通常对应 1 section
  level_0_leaf:        200 tokens    # 1-3 paragraphs

retrieval:
  initial_top_k: 12 leaves
  merge_threshold: 0.5
  final_top_k: 5
  
context_budget: 6000 tokens (after merging)

5.3 LlamaIndex的实现

直接用LlamaIndex会更省事：

from llama_index.core.node_parser import HierarchicalNodeParser
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.storage import StorageContext

node_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]
)
nodes = node_parser.get_nodes_from_documents(documents)

# 索引...
retriever = AutoMergingRetriever(
    vector_index.as_retriever(similarity_top_k=12),
    storage_context=storage_context,
    verbose=True
)

---

六、Cost & Latency

→ Latency差异小，主要是 storage和复杂度的trade-off。

---

七、关键速查表

7.1 决策矩阵

7.2 最佳chunk size组合

---

八、面试题

Q1: 解释Auto-Merging Retrieval的本质思想。

解决了 "小chunk检索准 vs 大chunk上下文足" 的矛盾。本质是 延迟决策的层次结构：建index时按leaf，但retrieval时根据 命中分布 动态决定层级。如果一个section内多数leaves被命中，说明这个section整体相关，merge回section给LLM完整上下文；如果只是零散命中，保持leaf精度。这是"information density"的统计推断。

Q2: 你的RAG用flat 800-token chunks，answer质量不行，何时切hierarchical？

看错例分析：(1) 如果错例是 "答案不完整" 或 "列举遗漏"，hierarchical能救——它给LLM完整parent context; (2) 如果错例是 "找错了文档"，先优化embedding/rerank，hierarchical帮不上; (3) 如果错例是 "幻觉"，hierarchical轻微帮助（更多context降低LLM编造概率）但不是主因。经验：flat → hierarchical切换前，先做 chunk size sweep（200, 400, 800, 1200），看是否flat小chunk + window就够了。

Q3: hierarchical RAG对rerank的影响？

Rerank应该作用在 leaf level（精确度最高），merge是rerank之后的步骤。具体流程：(1) retrieve top-50 leaves; (2) bge-rerank → top-10 leaves; (3) auto-merge → 5 final blocks (可能是parents或leaves混合); (4) 送LLM。注意：merge后的parent text没有reranker score，可以用其内部leaves的max score代替，或者重新rerank parent text。

Q4: Sentence Window vs Parent-Child，怎么选？

Sentence Window：实现简单（不需要存层级关系），适合 任意位置都可能有相关信息 的文档（小说、对话、转录）。Parent-Child：适合 结构化文档（10-K、法规、学术论文），parent对应自然边界（section、article），merge后保留语义完整性。Window的"k=1"是简单平均扩展，可能切到无关段落；parent-child的合并尊重作者的章节划分。

Q5: 生产中hierarchical RAG的最大风险是什么？

Context budget管理。假设parent是1500 tok，top_k=5，加上system prompt + question就 ~8000 tok。Claude Sonnet $3/M input → 单次$0.024，10K query/day = $7200/月。控制方法：(1) 限制max merge level (不让auto-merge到grandparent)；(2) 用parent summary代替full text（每个parent预先summarize成300 tok）；(3) 智能截断（保留parent的开头和结尾，省中间）；(4) prompt caching（system prompt和高频parents缓存，省80%）。

---

九、明日预告

Day 143: GraphRAG——hierarchical解决了"上下文完整性"，但还有一类问题flat/hierarchical都搞不定：多跳推理（"Apple收购的subsidiaries里CEO是X的有几个？"）。这需要 实体-关系图谱。明天我们用 Microsoft GraphRAG 处理金融报告，构建知识图谱 + LLM混合推理，看在多跳问题上是否能从0.45提升到0.85+。