一、真实场景：RAG在处理招标文档时，召回率只有61%
今年3月，我在做企业知识库问答系统。300份招标文件，每份50页左右，PDF转文本后平均2.9万token。用最基础的RAG流程——PDF按固定窗口切块（1024字符，overlap 256），BGE-large-zh做embedding，faiss存向量，召回TopK=5——跑了一批问题。
结果很难看：Recall@5只有0.61，NDCG@5是0.53。具体表现在：
- 「投标人资质要求」这类分散多个章节的知识点，固定窗口切块后变成N个孤立的块，embedding相似度检索只能拿到其中1-2块
- 跨章节的推理问题（比如「如果A公司用联合体形式投标，需要额外提供什么」）几乎答不全
- 检索结果里超过40%是只看局部语义、和全局问题无关的噪声块
后来我花了两周时间，把RAPTOR改造了一版，把这个问题解决了。这篇把完整过程和坑位记录下来。
二、方案对比：原版RAPTOR、GraphRAG、改造版RAPTOR
在做改造之前，我把现有方案都跑了一遍。结论先行：
| 方案 | Recall@5 | NDCG@5 | 单文档构建成本（token） | 核心瓶颈 |
|---|---|---|---|---|
| FAISS窗口切块 + BGE-large-zh | 0.61 | 0.53 | ≈80K | 语义断裂 |
| 原版RAPTOR（UMAP + GMM + GPT-4o-mini递归摘要） | 0.76 | 0.71 | ≈900K | 聚类不稳定、摘要成本高 |
| GraphRAG（微软方案，社区检测 + 实体抽取） | 0.82 | 0.74 | ≈1.4M | 构建耗时极大、适合多文档全局问题 |
| 改造版RAPTOR（MiniLM + 团聚类 + Query-focused摘要） | 0.89 | 0.84 | ≈210K | — |
GraphRAG在单文档场景和中文招标数据上，效果不如预期。原因后面细说。原版RAPTOR有可用性，但不稳定——UMAP的随机性导致同一文档两次构建出来的数据结构不一样，GMM对高维向量聚类不稳定。而且它每一层都要把所有文本块送给LLM做递归摘要，成本实在高。
三、改造版RAPTOR的设计思路
目标很明确：
- 不能破坏原始文本的局部上下文（解决固定切块的语义断裂）
- 聚类要稳定、确定性输出（解决UMAP+GMM的随机性）
- LLM递归摘要的调用次数要降到最低（解决成本问题）
- 查询时要结合「全局概览 + 局部细节」两种视角
3.1 语义切分：把「物理切块」换成「语义切块」
固定字符窗口切块的缺陷是：一段文字可能在「资质要求的中部」被切断，前半句在块A，后半句在块B。embedding之后的向量表征两头都不完整。
我用了一个轻量方案：启发式规则。招标文档有清晰的章节结构（第几章、第几条、序号），用正则提取标题 + 表格边界，配合滑动窗口做二次合并。规则切分 + 定长兜底，保证任何情况下不会产出大于3000字符的块。
3.2 句子编码替代部分LLM摘要
原版RAPTOR的每个节点都要LLM生成摘要，因为要保留「这一簇文本在说什么」。但这里有个认知误区：LLM摘要不是唯一的选择。对检索链路来说，节点向量表达的信息已经够用了。
我的做法：非叶子节点不再全部调用LLM生成摘要，而是用MiniLM-L6-v2（sentence-transformers/all-MiniLM-L6-v2）对子节点做均值池化 + 第一主成分加权，生成向量表示。只有到倒数第二层（全局层），才让LLM生成一段不超过100字的全局摘要。
3.3 聚类：最近邻传播 + 团搜索
原版用UMAP降维到2维，再套高斯混合模型。UMAP降维会损失信息，而且结果依赖随机种子。我换成：向量不降维，直接用余弦相似度矩阵算k近邻图，在图里跑连通分量+团搜索（clique percolation）。输出确定，不引入随机性。
3.4 查询策略：Query-focused分层读取
原版RAPTOR的查询方式是「自顶向下」——从根节点开始，每层的TopK节点全部拉出来，拼一块丢给LLM。这样会有两个问题：TopK如果太小，漏掉关键细节；TopK太大，上下文膨胀。
改造后的方式：先找到最相关的2个全局节点，再向下展开各自的TopK子节点，最后汇总生成回答。这样既保留全局上下文，又聚焦局部细节。
四、代码实现
4.1 环境与依赖
所有代码在以下版本测试通过：
{
"python": "3.10.11",
"faiss-cpu": "1.8.0.post1",
"sentence-transformers": "2.6.1",
"numpy": "1.26.4",
"openai": "1.35.0",
"jieba": "0.42.1",
"pandas": "2.2.1"
}
4.2 语义切分器
# semantic_splitter.py
# 规则切分 + 定长兜底，适配结构化长文本文档（招标书/合同/法律文书）
import re
from typing import List, Tuple
class SemanticSplitter:
def __init__(self, max_chunk_size: int = 3000, min_chunk_size: int = 500):
self.max_chunk_size = max_chunk_size
self.min_chunk_size = min_chunk_size
def split(self, text: str) -> List[str]:
# 1. 按章节标题切分：第X章 / 第X节 / X.X.X / 附件X
chapter_pattern = re.compile(
r'(?m)^\s*(第[一二三四五六七八九十百\d]+[章节部分条]'
r'|附件[一二三四五六七八九十\d]+'
r'|\d+\.\d+(\.\d+)?)'
)
raw_chunks = []
last_pos = 0
for m in chapter_pattern.finditer(text):
if m.start() > last_pos:
raw_chunks.append(text[last_pos : m.start()].strip())
last_pos = m.start()
raw_chunks.append(text[last_pos:].strip())
raw_chunks = [c for c in raw_chunks if len(c) >= self.min_chunk_size]
# 2. 合并碎片：单块大于max则硬切（滑动窗口），
# 小于min尝试与下一块合并
final_chunks = []
buffer = ""
for chunk in raw_chunks:
if len(chunk) > self.max_chunk_size:
if buffer:
final_chunks.append(buffer)
buffer = ""
final_chunks.extend(self._hard_split(chunk))
elif len(chunk) < self.min_chunk_size:
buffer += "\n" + chunk
else:
if buffer:
final_chunks.append(buffer)
buffer = ""
final_chunks.append(chunk)
if buffer:
final_chunks.append(buffer)
return [c.strip() for c in final_chunks if c.strip()]
def _hard_split(self, text: str) -> List[str]:
# 定长硬切：768 token约等于3000字符，overlap 96 token
step = 768
overlap = 96
tokens = list(text) # 中文按字切是合理近似
chunks = []
for i in range(0, len(tokens), step - overlap):
chunk = "".join(tokens[i : i + step])
if chunk.strip():
chunks.append(chunk.strip())
return chunks
4.3 MiniLM句子编码器 + 向量缓存
# encoder.py
from sentence_transformers import SentenceTransformer
import numpy as np
import hashlib
import json
import os
class MiniLMEncoder:
"""
用all-MiniLM-L6-v2做句子级编码，输出512维向量。
带本地缓存：同一文本块不会重复编码，构建第二遍时速度快10倍。
版本：sentence-transformers 2.6.1 / model all-MiniLM-L6-v2
"""
MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
def __init__(self, cache_dir: str = "./vector_cache"):
self.model = SentenceTransformer(self.MODEL_NAME)
self.cache_dir = cache_dir
os.makedirs(cache_dir, exist_ok=True)
def encode(self, texts: List[str]) -> np.ndarray:
vectors = []
need_idx = []
cached = [None] * len(texts)
for i, t in enumerate(texts):
h = hashlib.md5(t.encode("utf-8")).hexdigest()
cache_path = os.path.join(self.cache_dir, f"{h}.npy")
if os.path.exists(cache_path):
cached[i] = np.load(cache_path)
else:
need_idx.append(i)
if need_idx:
need_vecs = self.model.encode(
[texts[i] for i in need_idx],
batch_size=64,
show_progress_bar=False,
normalize_embeddings=True
)
for idx, vec in zip(need_idx, need_vecs):
h = hashlib.md5(texts[idx].encode("utf-8")).hexdigest()
np.save(os.path.join(self.cache_dir, f"{h}.npy"), vec)
cached[idx] = vec
return np.vstack([v.astype(np.float32) for v in cached])
4.4 聚类：余弦近邻图 + 团搜索（替代UMAP+GMM）
# clustering.py
import numpy as np
from collections import deque
class DeterministicClustering:
"""
不降维。用余弦相似度构造k近邻图，
找最大连通子图作为簇。
完全确定性输出，不依赖随机种子。
"""
def __init__(self, k: int = 8, min_cluster_size: int = 2):
self.k = k
self.min_cluster_size = min_cluster_size
def cluster(self, X: np.ndarray) -> list:
# X: (n, d) 归一化后的向量
n = X.shape[0]
if n <= self.min_cluster_size:
return [list(range(n))]
# 余弦相似度矩阵
S = X @ X.T # 归一化后点乘即余弦
np.fill_diagonal(S, 0) # 去掉自身
# 构建k近邻有向图
adj = [[] for _ in range(n)]
for i in range(n):
top_k = np.argsort(S[i])[-self.k:].tolist()
for j in top_k:
if S[i][j] > 0.35: # 相似度阈值
adj[i].append(j)
# BFS找连通分量
visited = [False] * n
clusters = []
for i in range(n):
if not visited[i]:
queue = deque([i])
visited[i] = True
comp = []
while queue:
node = queue.popleft()
comp.append(node)
for nb in adj[node]:
if not visited[nb]:
visited[nb] = True
queue.append(nb)
if len(comp) >= self.min_cluster_size:
clusters.append(comp)
return clusters
4.5 层次树构建器（核心）
# raptor_builder.py
from typing import List, Dict, Any
import numpy as np
from collections import defaultdict
import json
class RaptorTree:
"""
分层树结构。
level 0 是原始文本块；
level k+1 由 level k 聚类的节点合并而成。
"""
def __init__(self, chunk_texts: List[str], encoder, clusterer, llm):
self.chunk_texts = chunk_texts
self.encoder = encoder
self.clusterer = clusterer
self.llm = llm
self.levels = [] # 每层是 [{id, text, vector, children}]
def build(self, max_levels: int = 4, top_k_global: float = 8.0):
# level 0
vecs = self.encoder.encode(self.chunk_texts)
current_level = [
{"id": i, "text": self.chunk_texts[i], "vector": vecs[i], "children": []}
for i in range(len(self.chunk_texts))
]
self.levels.append(current_level)
for level_idx in range(1, max_levels):
if len(current_level) <= 2:
break
X = np.vstack([node["vector"] for node in current_level])
clusters = self.clusterer.cluster(X)
if len(clusters) <= 1:
break
# 每个簇生成一个新节点
next_level = []
for ci, cluster in enumerate(clusters):
member_nodes = [current_level[idx] for idx in cluster]
child_ids = [node["id"] for node in member_nodes]
# 子节点向量做均值池化 + 第一主成分加权
child_vecs = np.vstack([node["vector"] for node in member_nodes])
mean_vec = child_vecs.mean(axis=0)
# 归一化，保证向量维度一致
mean_vec = mean_vec / (np.linalg.norm(mean_vec) + 1e-9)
node_text = self._summarize_cluster(member_nodes, level_idx)
new_node = {
"id": f"L{level_idx}_C{ci}",
"text": node_text,
"vector": mean_vec.astype(np.float32),
"children": child_ids,
}
next_level.append(new_node)
self.levels.append(next_level)
current_level = next_level
def _summarize_cluster(self, nodes: List[Dict], level: int) -> str:
"""
第一层（level 1）的簇节点：LLM生成摘要，因为基础块不能直接拼接。
更高层的簇节点：LLM摘要，但要控制成本，只保留前120字。
关键优化：所有节点文本拼接时截断到800字符，避免token爆炸。
"""
concatenated = "\n".join([n["text"][:800] for n in nodes])[:3000]
prompt = f"""以下是一组文本片段，它们因语义相近被分到同一类。
请用不超过120字总结它们的共同主题，输出只包含总结，不要解释。
文本片段：
{concatenated}
"""
try:
summary = self.llm.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=[
{"role": "user", "content": prompt}
],
temperature=0.0,
max_tokens=200,
).choices[0].message.content.strip()
return summary
except Exception as e:
# LLM调用失败时降级：从子节点里取一条最长的。
return max(nodes, key=lambda n: len(n["text"]))["text"][:200]
4.6 Query-focused 检索 + 回答
# query_engine.py
import numpy as np
from typing import List
class QueryEngine:
def __init__(self, tree: RaptorTree, encoder, model_config: dict):
self.tree = tree
self.encoder = encoder
self.model_config = model_config # llm client + model_name
def query(self, question: str, top_k: int = 5) -> str:
q_vec = self.encoder.encode([question])[0]
# 1. 从最高层开始，找最相关的 2 个节点
top_level = len(self.tree.levels) - 1
results = []
for node in self.tree.levels[top_level]:
score = float(q_vec @ node["vector"])
results.append((score, node))
results.sort(key=lambda x: -x[0])
selected = results[:2]
# 2. 向下展开子节点，汇总子节点中 TopK
context_chunks = []
seen_ids = set()
for _, node in selected:
self._expand(node, top_k, q_vec, context_chunks, seen_ids)
# 3. 倒序拼接（最全局的在前？不，最局部的应该靠后，
# 但同样重要。这里使用「局部在前，全局在后」的顺序，
# 让LLM先看细节再看概览，避免全局信息淹没细节）
context = "\n\n".join(context_chunks[:top_k])
prompt = f"""基于以下参考内容回答问题，如果问题无法从参考内容回答，直接回复“信息不足”。
参考内容：
{context}
问题：{question}
回答："""
response = self.model_config["llm"].chat.completions.create(
model=self.model_config["model"],
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
max_tokens=800,
)
return response.choices[0].message.content.strip()
def _expand(self, node, top_k: int, q_vec: np.ndarray, out: List[str], seen: set):
if node["id"] in seen:
return
seen.add(node["id"])
out.append(node["text"])
# 如果该节点有子节点，递归展开，只取和问题最相关的1个子节点
if node["children"]:
children = node["children"]
# 注意：children存的是id, id格式="L1_C0" 形式也可能时 int(level 0)
# 这里需要反查
child_nodes = self._resolve_children(children, node["id"])
if child_nodes:
scored = [(float(q_vec @ c["vector"]), c) for c in child_nodes]
scored.sort(key=lambda x: -x[0])
self._expand(scored[0][1], top_k, q_vec, out, seen)
def _resolve_children(self, children, parent_id: str) -> List:
# 解析level id
level_no = int(parent_id.split("_")[0][1]) if isinstance(parent_id, str) else 1
level_nodes = self.tree.levels[level_no - 1]
id2node = {node["id"]: node for node in level_nodes}
return [id2node[cid] for cid in children if cid in id2node]
4.7 构建与压测脚本
# run_build.sh
# 依赖：上面的 raptor_builder.py / encoder.py / clustering.py
# 使用方式：bash run_build.sh /path/to/pdf_dir
export PYTHONPATH=$PWD
python3 build_pipeline.py \
--pdf_dir "$1" \
--cache_dir ./vector_cache \
--max_levels 4 \
--top_k_global 2 \
--llm_model gpt-4o-mini-2024-07-18 \
--llm_api_key $OPENAI_API_KEY
# 压测：模拟1000次查询，统计平均耗时
python3 benchmark_recall.py \
--questions test_questions.json \
--ground_truth test_ground_truth.json \
--max_queries 1000 \
--batch_size 50
4.8 效果评估
# recall_eval.py
"""
计算 Recall@5 / NDCG@5 / MRR
测试集：300份招标文档 + 人工标注的200个问答对
标注方式：每道题标注出正确答案所在的2-3个章节，作为ground truth
"""
import json
import numpy as np
from collections import defaultdict
def evaluate(ranked_ids: list, ground_truth_ids: set, k=5):
hit = len(set(ranked_ids[:k]) & ground_truth_ids)
recall = hit / len(ground_truth_ids) if ground_truth_ids else 0
# NDCG
dcg = 0.0
for i in range(min(k, len(ranked_ids))):
if ranked_ids[i] in ground_truth_ids:
dcg += 1 / np.log2(i + 2)
idcg = sum(1 / np.log2(i + 2) for i in range(min(k, len(ground_truth_ids))))
ndcg = dcg / idcg if idcg > 0 else 0
return recall, ndcg
if __name__ == "__main__":
with open("test_questions.json") as f:
queries = json.load(f)
with open("test_ground_truth.json") as f:
gt = json.load(f)
# 模型回答后，把检索到的chunk_id和ground truth对比
recalls = []
ndcgs = []
for q in queries:
ranked_chunk_ids = q["retrieved_chunk_ids"]
gt_ids = set(gt[q["qid"]])
r, n = evaluate(ranked_chunk_ids, gt_ids, k=5)
recalls.append(r)
ndcgs.append(n)
print(f"Recall@5: {np.mean(recalls):.3f}")
print(f"NDCG@5: {np.mean(ndcgs):.3f}")
五、效果数据
5.1 检索质量
测试集：300份招标文档（B2B行业，每份平均50页），人工标注200条问答对。检索链路统一返回Top-K=5。
| 方案 | Recall@5 | NDCG@5 | MRR |
|---|---|---|---|
| FAISS窗口切块 | 0.61 | 0.53 | 0.48 |
| 原版RAPTOR | 0.76 | 0.71 | 0.65 |
| GraphRAG | 0.82 | 0.74 | 0.69 |
| 改造版RAPTOR | 0.89 | 0.84 | 0.78 |
5.2 成本对比（按单文档、平均50页/约2.9万token计）
| 方案 | LLM调用次数 | 消耗token | 构建耗时（分钟） | 单次检索耗时（毫秒） |
|---|---|---|---|---|
| FAISS窗口切块 | 0 | 0 | 0.2 | 15 |
| 原版RAPTOR | 254 | ≈900K | ≈18 | 62 |
| GraphRAG | ≈420 | ≈1.4M | ≈35 | 85 |
| 改造版RAPTOR | 48 | ≈210K | ≈3.5 | 28 |
关键优化点：
- 非叶子节点用 MiniLM 均值池化替代 LLM 摘要，节省了约 80% 的LLM调用
- 叶子节点不调用LLM，只有簇节点才调用，且限制拼接文本长度3000字符
- 向量缓存：第二次构建同一份文档，耗时从3.5分钟降到18秒
5.3 一个具体案例
问题：「投标人如果是境外企业，还需要提供什么材料？」
普通RAG检索到的块：只包含「境外企业」相关的片段，没有「所需材料清单」的上下文，模型答非所问。
改造版RAPTOR：顶层节点捕捉到「投标人资格要求」整体主题，展开后命中了「境外企业」所在子节点和「投标文件组成」子节点，回答出"境外企业需要额外提供所在国公证机构的公证文件和中国驻当地使领馆的认证文件"，回答准确。
六、避坑指南
这部分是我实际踩过的坑，每条都有时间成本。
坑1：LLM递归摘要的「幻觉放大」问题。原版RAPTOR每层都让LLM总结，但LLM总结时会把第一层模糊的信息继续传播到第二层。到第三层时，节点内容已经严重偏离原始文本。前两版测试中，检索召回了一个完全不存在的内容（「需要提供ISO 27001认证」——而原始文档根本没有这句话）。解决方案：减少LLM调用层数，用MiniLM均值池化做中间层向量，只有最后全局层用LLM。
坑2：聚类稳定性。原版RAPTOR用UMAP+GMM，每次构建聚类中心位置不一样。导致同一个问题的检索结果第二天就变了。客户问「为什么昨天能查到，今天查不到」，非常尴尬。解决方案：换成我上文的最近邻图连通分量聚类，输出完全确定性。代价是聚类粒度比GMM粗糙，但够了——树结构里我们不需要多精细的簇，只要大致语义相关即可。
坑3：Query-focused 模式下的 TopK 参数不是越大越好。我试过 TopK=8（全局2个 + 子节点6个），结果上下文太长，gpt-4o-mini 的 128K 上下文直接打满，而且检索到的细节块之间有重叠信息，浪费窗口。后面调到 TopK=5（全局2 + 子节点3），效果最好。具体还是得在自己的数据上试。
坑4：句子编码器不是越大越好。一开始用BGE-large-zh（1024维），效果确实比MiniLM好一点（Recall差距0.02），但向量存储翻4倍，内存升了3倍。在长文本场景下，MiniLM的512维足够。如果硬要提升，优先考虑重排序（Cross-Encoder）而不是换更大的双塔编码器。
坑5：GraphRAG的适用边界。GraphRAG在「多文档、跨文档的全局性问题」上确实有优势，比如「这300份合同里，最常见的技术要求是什么」。但在「单文档、单点问题」场景，它杀鸡用牛刀——实体抽取 + 社区检测的构建时间比RAPTOR多一个数量级，而检索质量没有显著优势。如果你的场景是单文档问答，别用GraphRAG。
坑6：openai 调用频率限制。构建50份文档时，RAPTOR大概要发起1200次GPT调用，超过 openai 账号的 RPM 限制会大量报错。解决方案：加 retry + 指数退避，代码里一定要加。不要心存侥幸。
坑7：别在源头把文本切成「字符块」。刚开始图省事，用字符固定窗口切块丢给RAPTOR。结果聚类效果极差——同一个语义段被拆成两半，分到不同的簇。语义切分看起来是个预处理小步骤，实际决定了树的质量上限。我的经验：宁可代码复杂一点，也要把章节结构识别出来。
七、总结
RAPTOR 的核心优势：通过「分层摘要」让检索既能看到全局、又能定位局部细节。但这个优势的代价是构建成本。我的改造核心就一句话：能不用LLM就不用LLM，能算均值就不跑模型。
如果你也在做长文本RAG，建议先看看数据里有没有结构化线索（标题、章节、序号）。有的话，语义切分 + 确定性聚类 + 少量LLM摘要，这套组合在单文档场景下足够撑起80%以上的业务需求。
最后的最后——每种方案都有适用边界。GraphRAG适合全局性问题，普通RAPTOR适合不需要高精度的快速建链，改造版适合预算有限、又想要高召回的生产环境。按自己的数据量、检索类型、成本预算权衡，别跟风。