查询转换:让提问精准匹配知识库

多查询重写:针对用户表述模糊

思路:将原始输入转化为多个变体的问题查询

假设文档法( HyDE ) :用户问题和知识库差异大

思路:先生成一个理想的、假设的答案,然后将这个答案向量化后与知识库匹配,弥补了原始问题和真实答案之间的语义鸿沟

问题与应对:如果生成的假设文档和真实文档的匹配相似度低(比如:<0.7),系统会触发“二次检索”,换一种方式再找一遍。

问题分解和回退:针对复杂问题和无直接答案

思路:如果原始查询是一个太复杂或者知识库中没有具体答案的问题,系统不会放弃回答,而是“尝试拆解”大问题到小问题,或者“退一步”找个更宽泛的答案。

“退一步”找个更宽泛的答案,是因为前边的所有的提问都没法命中,一种兜底的策略。

分解策略选择表:

问题类型 策略 案例说明
多步骤推理 串行分解 “RAG流程?” -> 分解成“检索阶段”和“生成阶段”
跨领域问题 并行分解 “AI医疗应用?” -> 分解成 “AI技术”和“医疗应用”
细节查询失败 抽象回退 “特斯拉Q3财报?” -> 回退为“查找新能源企业财报?”

伪代码:

1
2
3
4
5
6
7
8
9
def decompose_query(query):
if is_multi_step(query):
return split_into_steps(query)
elif is_cross_domain(query):
return split_into_domains(query)
elif is_too_specific(query):
return generalize_query(query)
else:
return [query]

路由优化:构建智能分发网络

构建一个问答系统时,不能把所有的问题都交给同一个模型来处理。针对不同的问题要交给不同的“领域专家模型”来处理,这就引出来“路由优化”方法,同时针对每个“领域专家模型”的回答也不能用同一个Prompt模板来回复,所以引出来“Prompt路由”方法。

元数据路由引擎

思路:类似于一个问题的“智能分拣中心”,这里的元数据指的是问题的元数据,比如问题的分类标签、问题类型等。

举例说明:用户问题是一个“快递”,元数据路由引擎会根据问题的内容,去分析Meta-Data元数据(比如问题类型),判断问题属于哪个领域(比如:医疗、教育、编程等),然后把问题交给特定的“领域专家模型”来进行处理。

image-20251020111549831

伪代码:

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
# 模拟一个分类模型
def llm_classify(query):
if "病" in query or "医生" in query:
return "medical"
elif "代码" in query or "函数" in query:
return "code"
else:
return "default"
# 不同领域的检索器
def medical_retriever(query):
return f"[医疗检索] 正在查找与 '{query}' 相关的医学资料..."
def code_retriever(query):
return f"[代码检索] 正在查找与 '{query}' 相关的编程资料..."
def default_retriever(query):
return f"[通用检索] 正在查找与 '{query}' 相关的一般资料..."
# 路由决策函数
def route_query(query):
topic = llm_classify(query)
if topic == "medical":
return medical_retriever(query)
elif topic == "code":
return code_retriever(query)
else:
return default_retriever(query)
# 测试
print(route_query("我最近总是头痛,可能是什么原因?"))
# 输出:[医疗检索] 正在查找与 '我最近总是头痛,可能是什么原因?' 相关的医学资料...

关键参数说明

  • 路由准确率 > 85%

    如果系统判断错误太多,就启用“多路并行检索”,也就是同时让多个检索器一起运行,确保不漏掉正确答案。

  • 响应延迟 < 200ms

动态Prompt路由

思路:Prompt是给模型的“答题模板”,不同的答案需要不同的回复模板。

  • 技术人员喜欢看代码和架构图
  • 学生希望有通俗易懂的解释
  • 商务人士希望简洁明了的总结

实现流程

image-20251020122919954

伪代码

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
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer
# 加载语义模型
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
# 定义Prompt模板池
prompt_templates = {
"technical": "请用数学公式和架构图解释{query}", # 模版一
"educational": "请用通俗易懂的语言解释{query}", # 模版2
"business": "请用简洁明了的方式总结{query}" # 模版3
}
# 用户问题
user_query = "解释Transformer注意力机制"
# 计算语义相似度
query_embedding = model.encode([user_query])
template_embeddings = model.encode(list(prompt_templates.values()))
# 计算相似度
similarities = cosine_similarity(query_embedding, template_embeddings)
# 找出最匹配的模板
best_index = similarities.argmax()
best_template = list(prompt_templates.keys())[best_index]
best_score = similarities[0][best_index]
if best_score > 0.8:
print(f"匹配成功:使用 {best_template} 模板,相似度 {best_score:.2f}")
final_prompt = prompt_templates[best_template].format(query=user_query)
print("生成的Prompt:", final_prompt)
else:
print("没有找到合适的模板")

输出示例

1
2
匹配成功:使用 technical 模板,相似度 0.87
生成的Prompt:请用数学公式和架构图解释解释Transformer注意力机制

索引优化:知识库的智能重构

多表征索引 ( Multi-Representation )

什么是表征(Representation)?

简单来说,表征就是将现实世界中的信息(如:文字、语音、图片等)转换成机器可以理解和计算的数据形式(即向量或者张量)

本质:一种信息的编码,计算机只能处理数字。它将原始、复杂、高维的数据(例如一段文字、一张图片)映射到一个低纬、连续的向量空间中。

思路:给知识库的每一段内容,从不同的角度“拍照”,每张照片都是一种理解的方式。用户搜索时就会有更高的几率去命中匹配。

多表征索引流程图

image-20251020145116099

多表征索引策略

嵌入类型 拍照方式 作用说明
原始分块 把文本按固定长度切开 保留原始信息,就像原图一样清晰
摘要向量 用模型生成摘要 抓住整体意思,像看图说故事一样
问题向量 模型模拟用户可能问的问题 提高匹配度,像提前猜到你想问啥

举例说明

假设你有一段关于“人工智能”的内容,多表征索引生成三重嵌入策略如下:

  • 原始分块就是原文本。
  • 摘要向量是大模型生成的摘要“人工智能是模拟人类智能的技术”类似的内容向量。
  • 问题向量是大模型生成的“什么是人工智能?”、“AI有哪些应用?”这些用户可能问的问题。

存储优化:三重嵌入策略的向量存储方式

FAISS是一个高效的向量数据库。

嵌入类型 生成方式 存储优化
原始分块 文本分割 FAISS-IVF
摘要向量 T5生成摘要 + 嵌入 量化索引(PQ)
问题向量 GPT生成“可能被问的问题” HNSW图索引

注意,表中IVF_PQ是它的一种压缩索引方式。

  • 类比:就像把高清图片压缩成小图,既节省空间又不影响查找。
  • 效果:内存占用减少4倍,相当于把100G的数据压缩到25G。

IVF_PQ 和 HNSW 的性能收益:

  • 内存占用降低4倍(IVF_PQ量化)
  • 检索速度提升3倍(HNSW近邻搜索)

RAPTOR分层索引 ( 处理超长文档 )

RAPTOR:Recursive Abstractive Processing for Tree-Organized Retrieval,通过递归抽象化构建一个树结构的索引。

RAPTOR是将大模型的文本生成能力(用户总结)和检索系统的多层结构(用于索引)完美结合的表现。

RAG的工作方式和痛点?

  • 传统RAG:将长文档切分为固定大小的扁平化(Flat)文档块(Chunk),对每个文档块embedding并存储到向量数据库。
  • 局限性:
    • 丢失高层次语义:只能检索到具体的、低层次的细节。当用户询问涉及到文档的主题、跨章节的概念等高层次问题时,传统的RAG往往无法准确查询。
    • 上下文窗口限制:检索出的文档块数量有限,难以将整个文档的背景信息纳入到大模型的上下文(Context)。

RAPTOR思路:构建一个多阶段、递归式的索引过程,最终形成一个金字塔式的知识结构。

RAPTOR流程图

image-20251020162742917

索引构建过程:从下到上的递归抽象

  • 初始层(叶子结点):

    • 将原始文档切分为常规的小文档块(chunk)
    • 将每个chunk embedding并存储到向量数据库
  • 聚类(Clustering):

    使用聚类算法(例如UMAP、HDBScan等),根据初始层生成chunk的向量相似度,将语义相似性高的文档聚集在一块,形成簇(Cluster)

  • 抽象与总结(Abstractive / Summarization):

    • 将聚类后的同一个簇中所有的文档块文本输入到大模型中。
    • 大模型为这个簇生成抽象的、更高层次的总结(Summary)。这个总结代表这一个簇的核心主题。
  • 递归创建层级(Recursive):

    • 将新生成的总结文本视为新的一层。
    • 重复步骤2和步骤3:对这些总结进行聚类,再生成更高层次的、抽象的总结。
    • 这个过程一直重复,知道打到预设的层级深度,最终在数的根部生成一个高度概括的总结。

检索过程:从上到下的分层检索和融合

当用户查询时,RAPTOR可以根据查询的性质,灵活的在树状结构中进行检索:

  • 查询扁平化:将树结构中所有层级(包括底层的原始文本chunk和高层的抽象总结)的嵌入向量都放到同一个向量数据库中进行检索。

  • 具体查询:

    • 对于细节性问题(例如:“某年某月某日发生了什么?”),查询会匹配底层的原始文档。

    • 对于概括性问题(例如:“这个项目的主题思想是什么?”),查询会匹配高层的总结。

    • 最终,模型可以将同时检索到的细节信息高层总结信息一起送到LLM的上下文窗口中,让LLM获得更全面的背景知识。

伪代码

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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
# 假设已经导入了必要的库:
# from LLM_Framework import LLM, EmbeddingModel
# from Clustering_Library import cluster_texts

# --- 1. 配置和初始化 ---
# 定义常量
MAX_CHUNK_SIZE = 512 # 初始文本块的最大Token数
CLUSTERING_METHOD = "HDBSCAN" # 假设使用的聚类算法
EMBED_MODEL = EmbeddingModel("BGE-M3") # 假设的嵌入模型
LLM_SUMMARIZER = LLM("GPT-4o-mini") # 假设的大模型用于总结
MAX_RECURSION_DEPTH = 3 # 递归的最大深度
# 存储所有层级的节点(用于最终检索)
ALL_INDEX_NODES = []

# --- 2. 核心函数:递归构建树 ---
def build_raptor_tree_recursive(current_texts: list[str], current_depth: int):
"""
递归地将文本列表聚类、总结,并构建树的下一层。

Args:
current_texts (list[str]): 当前层级的文本列表 (可能是原始Chunk或上一层的Summary)。
current_depth (int): 当前递归深度。
"""

if not current_texts or current_depth > MAX_RECURSION_DEPTH:
return []
print(f"\n--- 正在处理第 {current_depth} 层级 ({len(current_texts)} 个节点) ---")

# 步骤 2.1: 嵌入 (Representation Learning)
# 将当前层的所有文本转换为向量
current_embeddings = EMBED_MODEL.encode(current_texts)

# 步骤 2.2: 聚类 (Clustering)
# 找到语义相近的文本簇
clusters = cluster_texts(current_embeddings, method=CLUSTERING_METHOD)

next_layer_summaries = []

# 步骤 2.3: 抽象总结 (Abstraction/Summarization)
for cluster_id, text_indices in clusters.items():
# 获取属于当前簇的文本
clustered_texts = [current_texts[i] for i in text_indices]

# 将簇内所有文本合并,并输入给 LLM 进行总结
prompt = f"请对以下相关文本进行高度抽象的总结和概括:\n\n{clustered_texts}"

# 假设 LLM.generate 返回总结文本
summary_text = LLM_SUMMARIZER.generate(prompt)

# 创建新的总结节点
summary_node = {
"text": summary_text,
"level": current_depth + 1,
"source_nodes": [current_texts[i] for i in text_indices] # 记录来源
}

# 将总结节点添加到下一层级的输入列表
next_layer_summaries.append(summary_text)

# 记录所有节点到全局索引中(用于扁平化检索)
ALL_INDEX_NODES.append(summary_node)

print(f" - 生成了 {cluster_id} 簇的高层总结。")

# 步骤 2.4: 递归调用
# 使用新生成的 Summary 作为下一层级的输入
if next_layer_summaries:
return build_raptor_tree_recursive(next_layer_summaries, current_depth + 1)
else:
return next_layer_summaries


# --- 3. 驱动主流程 ---
def raptor_indexing_pipeline(raw_document_text: str):
# 步骤 3.1: 初始分块
initial_chunks = split_text_into_chunks(raw_document_text, MAX_CHUNK_SIZE)

# 将初始 Chunk 加入全局索引
for chunk in initial_chunks:
ALL_INDEX_NODES.append({"text": chunk, "level": 0, "source_nodes": []})

print(f"原始文档切分为 {len(initial_chunks)} 个 Chunk。")

# 步骤 3.2: 启动递归构建
build_raptor_tree_recursive(initial_chunks, current_depth=0)

print("\n--- RAPTOR 索引构建完成 ---")
print(f"总共创建了 {len(ALL_INDEX_NODES)} 个可检索节点(包括原始 Chunk 和所有 Summary)。")

# 返回扁平化的索引(供检索使用)
return ALL_INDEX_NODES

# 假设的文本分块函数(简化)
def split_text_into_chunks(text: str, chunk_size: int) -> list[str]:
# 实际应用中需要更复杂的、基于 Token 的分块逻辑
return [f"Chunk {i+1}: {text[i*chunk_size:(i+1)*chunk_size]}" for i in range(len(text)//chunk_size + 1)]

# --- 4. 示例调用 ---
# 假设的输入文档
DOCUMENT = "大型语言模型(LLM)的兴起推动了检索增强生成(RAG)技术的发展。RAG通过引入外部知识库来解决LLM的幻觉问题。然而,传统的RAG面临着长文档的高层次语义丢失的挑战。RAPTOR方法通过递归地聚类和总结文档块,构建了一个分层的树状索引。这个索引在查询时能够同时提供细节和抽象信息,从而提高了复杂问答的准确性。RAPTOR的构建过程包括初始分块、基于嵌入向量的语义聚类,以及使用LLM进行抽象总结,这个过程会重复进行,直到达到预设的抽象层级。"

# 运行索引构建
raptor_index = raptor_indexing_pipeline(DOCUMENT)

# --- 5. 检索流程概念 (简化) ---
# 实际的检索会:
# 1. 对用户查询进行嵌入。
# 2. 在 raptor_index (ALL_INDEX_NODES) 中扁平化检索 Top-K 相似的节点。
# 3. 将检索到的节点文本送入 LLM 作为上下文,生成最终答案。

print("\n--- 检索流程(概念) ---")
QUERY = "RAG技术的局限性以及RAPTOR是如何解决的?"
# search_embedding = EMBED_MODEL.encode(QUERY)
# retrieved_nodes = find_top_k_nodes(raptor_index, search_embedding, k=5)
print(f"查询 '{QUERY}' 将会同时匹配到底层的 RAG 局限性细节和高层的 RAPTOR 解决总结。")

分块优化黄金法则

分块就像切蛋糕,切得太大吃起来不方便;切得太小,又容易丢失整体的味道。

分块大小建议

嵌入模型 最佳分块长度 重叠区间
text-embedding-ada-002 256 token 50 token
bge-large-zh 512 token 100 token

重叠部分:前后两部分重叠,避免断句导致信息丢失。

语义分块器 (Semantic Chunker)

1
2
from langchain_experimental.text_splitter import SemanticChunker
splitter = SemanticChunker(embeddings, breakpoint_threshold=0.7)
  • embeddings:模型对文本的理解来源
  • breakpoint_threshold:设定一个“语义断点”的阈值,超过这个值就切开。

流程图

image-20251020164042503

RAG策略优化:混合检索 + 重排序 + Self-RAG自检


生成控制:工业级部署关键


性能优化和基准测试


RAG优化落地检查表