各位同仁，各位对信息检索与自然语言处理充满热情的专家们：
欢迎来到今天的技术讲座。在当今这个信息爆炸的时代，我们每天都面临着海量数据的洪流。从企业内部的文档库、科研论文集，到互联网上的无数网页，如何高效地从这些庞大的非结构化数据中提取有价值的信息，并以易于理解的方式呈现，是我们面临的核心挑战。传统的关键词搜索、甚至是基于向量相似度的检索，在面对需要深层理解、概括和跨文档关联的复杂查询时，往往显得力不从心。
今天，我们将深入探讨一项革命性的技术框架，它旨在解决这一难题——那就是 RAPTOR，全称 Recursive Abstractive Processing for Hierarchical Summarization and Retrieval（递归抽象处理，用于层次化摘要与检索）。顾名思义，RAPTOR 的核心在于其“递归”和“抽象”的特性，它通过构建文档库的多层次语义表示，使得我们能够像剥洋葱一样，从宏观概览逐步深入到微观细节，实现更智能、更具上下文感知的检索。
1. 挑战与机遇：为什么我们需要RAPTOR？
想象一下，你面对的是一个包含数百万份技术文档、研究报告、客户案例和内部知识库的巨大档案库。你的任务是快速了解某个复杂项目的所有相关信息，或者找出某个特定技术在不同产品线中的应用情况。
传统检索方法的局限性：
- 关键词匹配的局限性： 关键词搜索容易错过同义词、近义词，更无法理解语义上的关联。它找到的是“词”，而不是“概念”。
- 扁平化向量检索的不足： 虽然基于嵌入（embeddings）的向量搜索能够捕捉语义相似性，但它通常将整个文档或固定大小的文本块视为一个独立的单元。当文档非常长时，一个单一的嵌入向量很难完整地代表其所有复杂的主题。此外，它缺乏对信息“层次”的感知，无法区分一篇文档的中心思想和某个细枝末节。
- 缺乏概括能力： 即使找到了相关文档，用户仍然需要阅读全文来理解其核心内容。对于大量结果，这会造成巨大的阅读负担。
- 上下文缺失： 当用户找到一个与查询相关的段落时，他们往往需要向上或向下追溯，以了解该段落的上下文。扁平的检索结果通常难以直接提供这种上下文。
- 处理复杂查询的困难： 像“总结过去五年公司在人工智能领域的主要进展和挑战”这样的查询，需要跨文档的聚合、概括和结构化呈现，这是传统方法难以实现的。
RAPTOR 带来的机遇：
RAPTOR 的出现，正是为了弥补这些不足。它不只停留在“找到”信息，更致力于“理解”和“组织”信息。通过其递归抽象处理机制，RAPTOR 能够：
- 构建层次化的语义索引： 将庞大的文档库分解成不同抽象层次的摘要，形成一个语义树或图结构。
- 提供多粒度检索： 用户可以从高层概览开始，逐步下钻到具体细节，或者从细节出发，追溯其所属的更广泛上下文。
- 生成抽象性摘要： 不仅仅是提取原文片段，而是利用大型语言模型（LLMs）生成全新的、连贯的、概括性的文本摘要。
- 增强上下文感知能力： 检索结果不再是孤立的文本块，而是带有明确上下文路径的摘要链，极大地提升了用户对信息的理解效率。
简而言之，RAPTOR 旨在将一个无序的、庞大的文档库，转化为一个可导航、可概括、可深度探索的知识图谱。
2. RAPTOR 的核心概念与架构
RAPTOR 的核心在于其“递归抽象”的思想。它将原始文档视为最底层的“叶子节点”，然后通过迭代地聚类、摘要和嵌入这些节点，逐步向上构建更高级别的“父节点”，直到形成一个能够代表整个文档库的层次结构。
2.1 核心组件
RAPTOR 框架主要依赖以下几种关键技术：
- 分块策略 (Chunking Strategy): 将原始文档分解成更小的、语义连贯的文本块。
- 嵌入模型 (Embedding Models): 将文本块或摘要转换为高维向量，捕捉其语义信息。
- 聚类算法 (Clustering Algorithms): 根据语义相似性将文本块或摘要分组。
- 摘要模型 (Summarization Models / LLMs): 对聚类后的文本组生成概括性摘要。
- 层次化数据结构 (Hierarchical Data Structure): 存储和管理不同抽象层次的摘要及其关系。
2.2 总体工作流程概览
RAPTOR 的工作流程可以分为两个主要阶段：索引构建阶段 和 检索阶段。
索引构建阶段：
- 原始文档预处理： 清洗、分块。
- 初始嵌入： 对所有原始文本块生成向量嵌入。
- 递归抽象循环：
- 聚类： 对当前层级的文本（原始块或前一层级的摘要）进行聚类。
- 摘要： 对每个聚类中的文本生成一个更高层级的抽象摘要。
- 嵌入： 对新生成的摘要进行嵌入。
- 层级连接： 记录新摘要与生成它的子文本之间的关系。
- 循环终止： 直到达到预设的抽象层数，或者所有文本聚合到一个根摘要。
- 构建层次化索引： 将所有生成的摘要及其嵌入、层级关系存储起来。
检索阶段：
- 用户查询嵌入： 将用户查询转换为向量。
- 多层次检索： 根据查询向量，在层次结构的不同层级进行相似性搜索。
- 结果聚合与提炼： 结合不同层级的检索结果，生成上下文丰富、多粒度的最终答案。这可以是一个高层摘要，也可以是带有其上下文路径的详细片段。
为了更深入理解，我们现在将详细分解 RAPTOR 的各个阶段。
3. RAPTOR 索引构建阶段：从原始数据到层次化知识
索引构建是 RAPTOR 的核心，它负责将海量原始文档转化为一个可查询的层次化语义结构。
3.1 阶段一：原始文档处理与初始嵌入
这一步是所有文本处理任务的基础，它将原始、非结构化的文档转化为可操作的单元，并赋予其初步的语义表示。
3.1.1 文档分块 (Chunking)
由于大型语言模型和嵌入模型通常有输入长度限制，并且为了更好地捕捉局部语义，我们需要将原始文档分解成更小的、语义连贯的文本块。分块的策略有很多种，例如：
- 固定大小分块 (Fixed-size Chunking): 简单地按字符数或单词数截断。
- 滑动窗口分块 (Sliding Window Chunking): 固定大小分块，但块之间有重叠，以保留上下文。
- 语义分块 (Semantic Chunking): 尝试根据文档结构（段落、标题、章节）或语义连贯性进行分块。这是更理想但实现更复杂的方法。
对于 RAPTOR，一个合适的起始分块大小至关重要，它需要足够小以捕捉具体细节，又足够大以提供一定的上下文供后续摘要。
import os
import re
from typing import List, Dict
def simple_chunking(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""
简单的滑动窗口分块函数。
"""
if not text:
return []
words = text.split()
chunks = []
current_pos = 0
while current_pos < len(words):
end_pos = min(current_pos + chunk_size, len(words))
chunk = " ".join(words[current_pos:end_pos])
chunks.append(chunk)
if end_pos == len(words):
break
current_pos += (chunk_size - overlap)
if current_pos < 0: # 避免负数步进
current_pos = 0
return chunks
def process_document(doc_path: str, chunk_size: int = 500, overlap: int = 50) -> List[Dict]:
"""
读取文档并进行分块。
"""
with open(doc_path, 'r', encoding='utf-8') as f:
content = f.read()
# 简单清洗，例如移除多余空白
content = re.sub(r's+', ' ', content).strip()
chunks = simple_chunking(content, chunk_size, overlap)
# 为每个块添加元数据，如原始文档ID、块ID
processed_chunks = []
doc_id = os.path.basename(doc_path)
for i, chunk_text in enumerate(chunks):
processed_chunks.append({
"id": f"{doc_id}_chunk_{i}",
"text": chunk_text,
"doc_id": doc_id,
"level": 0 # 原始块位于层级0
})
return processed_chunks
# 示例用法
# sample_doc_path = "path/to/your/document.txt"
# initial_chunks = process_document(sample_doc_path)
# print(f"Generated {len(initial_chunks)} initial chunks.")
# print(initial_chunks[0])3.1.2 初始嵌入 (Initial Embedding)
每个文本块都需要被转换成一个固定长度的数值向量，这个向量能够捕捉其语义信息。这通常通过预训练的嵌入模型（如 Sentence-BERT, OpenAI Embeddings, Cohere Embeddings等）来完成。选择一个高性能的嵌入模型对后续的聚类和检索效果至关重要。
from sentence_transformers import SentenceTransformer
import numpy as np
# 初始化嵌入模型
# 推荐使用性能较好的模型，例如 'all-MiniLM-L6-v2', 'BAAI/bge-small-en-v1.5'
# 或者通过API调用OpenAI/Cohere等
try:
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
except Exception as e:
print(f"Error loading SentenceTransformer: {e}. Please ensure you have internet access or the model is downloaded.")
print("Falling back to a dummy embedding model for demonstration.")
class DummyEmbeddingModel:
def encode(self, texts, **kwargs):
# 模拟生成随机嵌入向量
return np.random.rand(len(texts), 384) # all-MiniLM-L6-v2 维度是 384
embedding_model = DummyEmbeddingModel()
def generate_embeddings(texts: List[str]) -> np.ndarray:
"""
为给定文本列表生成嵌入向量。
"""
if not texts:
return np.array([])
embeddings = embedding_model.encode(texts, convert_to_numpy=True)
return embeddings
# 为初始块生成嵌入
# chunk_texts = [chunk['text'] for chunk in initial_chunks]
# initial_embeddings = generate_embeddings(chunk_texts)
# for i, chunk in enumerate(initial_chunks):
# chunk['embedding'] = initial_embeddings[i]
# print(f"Generated embeddings for {len(initial_chunks)} chunks.")
# print(initial_chunks[0]['embedding'].shape)3.2 阶段二：递归聚类与摘要
这是 RAPTOR 最核心的循环过程，它将重复执行，逐步将底层的细节信息抽象化为高层概览。
3.2.1 循环迭代逻辑
我们从最底层的原始文本块（Level 0）开始。
- 聚类 (Clustering): 在当前层级，我们根据文本块（或前一层级的摘要）的嵌入向量，使用聚类算法将语义相似的文本分组。常用的聚类算法包括 K-Means、Agglomerative Clustering、HDBSCAN 等。HDBSCAN 特别适合处理密度不均的数据，并且可以自动发现聚类的数量，还能将噪音点标识出来，这在文档聚类中很有用。
- 摘要 (Summarization): 对于每个聚类，我们将其包含的所有文本（原始块或前一层级的摘要）合并，然后使用一个强大的摘要模型（通常是大型语言模型 LLM）生成一个简洁、连贯、抽象的摘要。这个摘要应该能概括该聚类中所有文本的核心内容。
- 嵌入新摘要： 对新生成的每个摘要，我们再次使用嵌入模型生成其向量表示。
- 构建层级关系： 我们记录下新生成的摘要（作为父节点）与构成它的所有文本（作为子节点）之间的关系。
- 提升层级： 新生成的摘要及其嵌入成为下一个迭代的输入，它们构成了更高一级的抽象。
这个过程会重复进行，直到达到某个终止条件，例如：- 只剩下一个或少数几个根摘要。
- 达到预设的最大层级数。
- 聚类效果不再显著提升。
3.2.2 伪代码与概念实现
from sklearn.cluster import AgglomerativeClustering
from sklearn.metrics.pairwise import cosine_similarity
from collections import defaultdict
import uuid
import openai # 假设使用OpenAI API进行摘要
# 设置OpenAI API Key
# openai.api_key = os.getenv("OPENAI_API_KEY")
def get_llm_summary(texts: List[str], model: str = "gpt-3.5-turbo") -> str:
"""
使用LLM对一组文本进行摘要。
"""
if not texts:
return ""
# 组合文本，注意控制总长度以适应LLM的上下文窗口
combined_text = "n---n".join(texts)
# 实际应用中需要更复杂的提示工程来控制摘要质量和风格
prompt = f"请对以下文本进行简洁、抽象的概括，提取其核心思想和关键信息：nn{combined_text}nn摘要："
try:
# 模拟LLM调用
if openai.api_key:
response = openai.ChatCompletion.create(
model=model,
messages=[
{"role": "system", "content": "You are a helpful assistant that summarizes documents."},
{"role": "user", "content": prompt}
],
max_tokens=200, # 控制摘要长度
temperature=0.3
)
summary = response.choices[0].message.content.strip()
else:
# 模拟摘要生成，如果没有API Key
print("Warning: OpenAI API Key not found. Using dummy summary generator.")
summary = f"这是一个关于 {texts[0][:50]}... 等主题的概括性摘要，由 {len(texts)} 个子文本组成。"
except Exception as e:
print(f"Error calling LLM for summarization: {e}. Using dummy summary.")
summary = f"这是一个关于 {texts[0][:50]}... 等主题的概括性摘要，由 {len(texts)} 个子文本组成。"
return summary
def recursive_abstractive_processing(
items: List[Dict],
embedding_model: SentenceTransformer,
max_levels: int = 3,
clustering_threshold: float = 0.5, # 聚类相似度阈值
min_cluster_size: int = 2
) -> Dict:
"""
执行RAPTOR的递归抽象处理过程。
items: 当前层级的文本项列表，每个项包含 'id', 'text', 'embedding', 'level'。
返回一个包含所有层级摘要的字典结构。
"""
hierarchy = defaultdict(list)
current_level_items = items
level = 0
while True:
print(f"Processing Level {level} with {len(current_level_items)} items...")
if len(current_level_items) <= min_cluster_size or level >= max_levels:
# 终止条件：没有足够的项目进行聚类，或达到最大层级
for item in current_level_items:
hierarchy[item['level']].append(item)
break
# 提取当前层级所有项的文本和嵌入
current_texts = [item['text'] for item in current_level_items]
current_embeddings = np.array([item['embedding'] for item in current_level_items])
if len(current_embeddings) == 0:
break
# 聚类：使用AgglomerativeClustering，通过距离阈值来控制聚类数量
# distance_threshold = 1 - clustering_threshold (cosine distance = 1 - cosine similarity)
# affinity='cosine' 表示使用余弦相似度，linkage='average' 是平均链接
clustering_model = AgglomerativeClustering(
n_clusters=None,
distance_threshold=1 - clustering_threshold,
affinity='cosine',
linkage='average'
)
cluster_labels = clustering_model.fit_predict(current_embeddings)
# 将项目分组到聚类中
clusters = defaultdict(list)
for i, label in enumerate(cluster_labels):
if label != -1: # 忽略HDBSCAN可能产生的噪音点
clusters[label].append(current_level_items[i])
next_level_items = []
for cluster_id, cluster_items in clusters.items():
if len(cluster_items) < min_cluster_size:
# 小于最小聚类大小的，视为噪音或不适合聚合，将它们带到下一层
# 或者可以把它们直接添加到当前层级，不生成摘要
for item in cluster_items:
hierarchy[item['level']].append(item)
continue
# 提取聚类中所有子项的文本
sub_texts = [item['text'] for item in cluster_items]
# 生成摘要
summary_text = get_llm_summary(sub_texts)
summary_embedding = generate_embeddings([summary_text])[0] # 摘要的嵌入
# 创建新的父摘要项
new_summary_item = {
"id": str(uuid.uuid4()),
"text": summary_text,
"embedding": summary_embedding,
"level": level + 1,
"children_ids": [item['id'] for item in cluster_items] # 记录子节点ID
}
next_level_items.append(new_summary_item)
# 将当前层级的这些子项加入到层次结构中
for item in cluster_items:
hierarchy[item['level']].append(item)
current_level_items = next_level_items
level += 1
# 将最终的顶层项目也添加到层次结构中
for item in current_level_items:
hierarchy[item['level']].append(item)
return hierarchy
# --- 完整的索引构建流程 ---
def build_raptor_index(doc_paths: List[str], embedding_model: SentenceTransformer, **kwargs) -> Dict:
all_raw_chunks = []
for doc_path in doc_paths:
all_raw_chunks.extend(process_document(doc_path))
# 为所有原始块生成嵌入
chunk_texts = [chunk['text'] for chunk in all_raw_chunks]
if chunk_texts:
initial_embeddings = generate_embeddings(chunk_texts)
for i, chunk in enumerate(all_raw_chunks):
chunk['embedding'] = initial_embeddings[i]
else:
print("No chunks generated. Exiting.")
return {}
# 执行递归抽象处理
raptor_hierarchy = recursive_abstractive_processing(
all_raw_chunks,
embedding_model,
**kwargs
)
print("RAPTOR hierarchy built successfully.")
return raptor_hierarchy
# # 示例：创建一些虚拟文档
# if not os.path.exists("docs"):
# os.makedirs("docs")
# with open("docs/doc1.txt", "w", encoding="utf-8") as f:
# f.write("人工智能是计算机科学的一个分支，它试图使机器像人类一样思考和学习。机器学习是人工智能的一个子集，专注于从数据中学习模式。深度学习是机器学习的一个更小子集，利用神经网络进行学习，并在图像识别和自然语言处理等领域取得了巨大成功。")
# with open("docs/doc2.txt", "w", encoding="utf-8") as f:
# f.write("自然语言处理（NLP）是人工智能的一个领域，它关注计算机与人类语言之间的交互。文本分类、命名实体识别和机器翻译是NLP的常见任务。大型语言模型（LLMs）如GPT系列，是NLP领域的最新突破，能够生成连贯且上下文相关的文本。")
# with open("docs/doc3.txt", "w", encoding="utf-8") as f:
# f.write("计算机视觉是另一个AI领域，专注于使计算机能够“看”并理解数字图像和视频。对象检测、图像分割和人脸识别是CV的关键技术。卷积神经网络（CNNs）是CV领域的核心模型。")
# with open("docs/doc4.txt", "w", encoding="utf-8") as f:
# f.write("推荐系统在电子商务和内容平台中无处不在，它们根据用户的历史行为和偏好来推荐商品或内容。协同过滤和基于内容的推荐是两种主要方法。深度学习模型也被广泛应用于推荐系统。")
# with open("docs/doc5.txt", "w", encoding="utf-8") as f:
# f.write("数据科学是一个交叉学科，结合了统计学、计算机科学和领域知识，从数据中提取洞察和知识。数据清洗、特征工程、模型训练和评估是数据科学家的日常工作。")
# with open("docs/doc6.txt", "w", encoding="utf-8") as f:
# f.write("云计算提供按需计算资源，包括服务器、存储、数据库、网络、软件、分析和智能。它分为IaaS、PaaS和SaaS三种服务模式。无服务器计算是云计算的最新趋势之一。")
# doc_paths = [os.path.join("docs", f) for f in os.listdir("docs") if f.endswith(".txt")]
# raptor_index = build_raptor_index(doc_paths, embedding_model, max_levels=2, clustering_threshold=0.7, min_cluster_size=2)
# # 打印构建好的层级结构概览
# for level, items in raptor_index.items():
# print(f"nLevel {level} has {len(items)} items.")
# if items:
# print(f" Example text from Level {level}: {items[0]['text'][:100]}...")
# if 'children_ids' in items[0]:
# print(f" Example item from Level {level} has {len(items[0]['children_ids'])} children.")
3.2.3 层次化索引的存储
构建好的层次化结构需要以一种高效的方式存储，以便后续检索。这可以是一个复杂的图结构，也可以是嵌套的字典或者数据库表。每个节点（无论是原始块还是摘要）都应包含：
- ID: 唯一标识符。
- Text: 原始文本或摘要文本。
- Embedding: 文本的向量表示。
- Level: 所处的抽象层级。
- Parent IDs: 指向其父摘要的ID（如果存在）。
- Children IDs: 指向其子文本块或子摘要的ID（如果存在）。
- Original Doc ID: 原始文档的ID（对于原始块和其直接摘要）。
这种结构允许我们轻松地向上或向下遍历层次结构。
表格：RAPTOR 层次化索引节点结构示例
| 字段名称 | 数据类型 | 描述 | 示例值 |
|---|---|---|---|
id |
字符串 | 节点的唯一标识符 | doc1_chunk_0, uuid-abc-123 |
text |
字符串 | 原始文本块或摘要文本 | "人工智能是计算机科学的一个分支…" |
embedding |
浮点数数组 | 文本的向量表示 | [0.1, -0.5, 0.8, ...] |
level |
整型 | 节点所处的抽象层级 (0代表原始块) | 0, 1, 2 |
parent_ids |
字符串列表 | 父节点的ID列表 | ['uuid-xyz-456'] |
children_ids |
字符串列表 | 子节点的ID列表 | ['doc1_chunk_0', 'doc1_chunk_1'] |
doc_id |
字符串 | 原始文档ID (仅用于层级0和直接摘要) | doc1.txt |
metadata |
JSON对象 | 其他元数据，如创建时间、作者等 | {'author': 'Jane Doe', 'date': '2023-10-27'} |
4. RAPTOR 检索阶段：智能导航与信息提取
RAPTOR 的检索阶段利用其构建的层次化索引，提供比传统方法更强大、更具上下文感知的查询能力。
4.1 检索策略
RAPTOR 提供了多种检索策略，以适应不同的用户需求和查询类型。
4.1.1 顶层下钻检索 (Top-Down Retrieval)
这种策略适用于用户需要从宏观概览逐步深入细节的场景。
- 查询嵌入： 将用户查询转换为向量。
- 高层搜索： 在最高层级（最抽象的摘要）中搜索与查询最相似的摘要。
- 递归下钻： 对于每个选定的高层摘要，递归地向下遍历其子摘要或原始文本块，继续进行相似性搜索或上下文匹配。
- 结果聚合： 收集相关度高的摘要链或原始文本块，并根据其层级关系进行组织呈现。
这种方式的好处是能够快速聚焦到相关主题的宏观区域，避免遍历大量不相关的底层细节。
def find_similar_nodes(query_embedding: np.ndarray, nodes: List[Dict], top_k: int = 5) -> List[Dict]:
"""
在给定节点列表中查找与查询嵌入最相似的节点。
"""
if not nodes:
return []
node_embeddings = np.array([node['embedding'] for node in nodes])
if node_embeddings.size == 0:
return []
similarities = cosine_similarity(query_embedding.reshape(1, -1), node_embeddings)[0]
# 获取相似度最高的top_k个节点的索引
top_k_indices = np.argsort(similarities)[::-1][:top_k]
# 构造结果，包含相似度
results = []
for idx in top_k_indices:
node = nodes[idx].copy()
node['similarity'] = similarities[idx]
results.append(node)
return results
def recursive_top_down_retrieval(
query_embedding: np.ndarray,
hierarchy: Dict[int, List[Dict]],
max_results: int = 5,
top_k_per_level: int = 3,
min_similarity_threshold: float = 0.5
) -> List[Dict]:
"""
执行顶层下钻检索。
"""
retrieved_nodes = []
# 从最高层级开始
max_level = max(hierarchy.keys())
current_level_nodes = hierarchy[max_level]
# 在最高层级找到最相关的节点
top_level_matches = find_similar_nodes(query_embedding, current_level_nodes, top_k=top_k_per_level)
# 递归遍历子节点
queue = []
for match in top_level_matches:
if match['similarity'] >= min_similarity_threshold:
queue.append(match)
visited_ids = set() # 避免重复添加节点
while queue and len(retrieved_nodes) < max_results:
current_node = queue.pop(0) # 广度优先
if current_node['id'] in visited_ids:
continue
visited_ids.add(current_node['id'])
retrieved_nodes.append(current_node)
# 如果有子节点，继续下钻
if 'children_ids' in current_node and current_node['children_ids']:
child_level = current_node['level'] - 1
if child_level >= 0:
children_nodes_at_level = hierarchy[child_level]
# 筛选出当前节点的子节点
direct_children = [
node for node in children_nodes_at_level
if node['id'] in current_node['children_ids']
]
# 对子节点进行相似性搜索，确保它们也与原始查询相关
# 或者可以简单地将所有子节点加入队列，依赖于父节点的相似度
# 这里我们选择再次过滤，以确保子节点也直接相关
child_matches = find_similar_nodes(query_embedding, direct_children, top_k=top_k_per_level)
for child_match in child_matches:
if child_match['similarity'] >= min_similarity_threshold:
queue.append(child_match)
return retrieved_nodes[:max_results]
# # 示例检索
# query = "人工智能在自然语言处理中的应用"
# query_embedding = generate_embeddings([query])[0]
#
# if raptor_index:
# print(f"nPerforming top-down retrieval for query: '{query}'")
# results = recursive_top_down_retrieval(query_embedding, raptor_index, max_results=10, top_k_per_level=2)
#
# for i, res in enumerate(results):
# print(f"--- Result {i+1} (Level {res['level']}, Similarity: {res['similarity']:.2f}) ---")
# print(f"ID: {res['id']}")
# print(f"Text: {res['text'][:200]}...") # 打印前200字
# if 'children_ids' in res:
# print(f"Children count: {len(res['children_ids'])}")
# print("-" * 30)4.1.2 底层上溯检索 (Bottom-Up Retrieval)
这种策略适用于用户需要找到某个具体信息，并希望了解其更大上下文的场景。
- 查询嵌入： 将用户查询转换为向量。
- 底层搜索： 在最底层（原始文本块）中搜索与查询最相似的文本块。
- 递归上溯： 对于每个选定的底层文本块，向上遍历其父摘要，收集其所有祖先摘要，直到最高层级。
- 结果聚合： 将原始匹配的文本块及其上下文摘要链一起呈现。
这种方式能够确保用户不会错过任何具体细节，同时提供了理解这些细节所必需的宏观背景。
def get_node_by_id(node_id: str, hierarchy: Dict[int, List[Dict]]) -> Dict:
"""根据ID在整个层次结构中查找节点。"""
for level_nodes in hierarchy.values():
for node in level_nodes:
if node['id'] == node_id:
return node
return None
def find_parents(node_id: str, hierarchy: Dict[int, List[Dict]]) -> List[Dict]:
"""
给定一个节点ID，查找其所有父节点直到根节点。
"""
parents = []
current_node = get_node_by_id(node_id, hierarchy)
if not current_node:
return []
# 遍历所有层级，查找哪些节点将当前节点作为子节点
for level in sorted(hierarchy.keys(), reverse=True): # 从高层级开始找父节点
for parent_candidate in hierarchy[level]:
if 'children_ids' in parent_candidate and node_id in parent_candidate['children_ids']:
parents.append(parent_candidate)
# 递归查找父节点的父节点
parents.extend(find_parents(parent_candidate['id'], hierarchy)) # 注意：这里会重复，需要去重
return parents # 假设每个节点只有一个直接父节点（在简单的树结构中），或者我们只关心直接父链
return parents
def recursive_bottom_up_retrieval(
query_embedding: np.ndarray,
hierarchy: Dict[int, List[Dict]],
max_results: int = 5,
top_k_initial: int = 5,
min_similarity_threshold: float = 0.6
) -> List[Dict]:
"""
执行底层上溯检索。
"""
retrieved_paths = [] # 存储路径：[原始块, 父摘要1, 父摘要2, ...]
# 从最低层级（原始块）开始搜索
initial_chunks_level = 0
if initial_chunks_level not in hierarchy:
return []
raw_chunks = hierarchy[initial_chunks_level]
# 找到最相关的原始块
initial_matches = find_similar_nodes(query_embedding, raw_chunks, top_k=top_k_initial)
for match in initial_matches:
if match['similarity'] >= min_similarity_threshold:
path = [match]
# 找到其所有父节点
parent_nodes = find_parents(match['id'], hierarchy)
# 去重并按层级排序
unique_parents = {}
for p in parent_nodes:
unique_parents[p['id']] = p
sorted_parents = sorted(list(unique_parents.values()), key=lambda x: x['level'])
path.extend(sorted_parents)
retrieved_paths.append(path)
if len(retrieved_paths) >= max_results:
break
return retrieved_paths
# # 示例检索
# query_detail = "哪个模型用于图像识别和自然语言处理？"
# query_detail_embedding = generate_embeddings([query_detail])[0]
#
# if raptor_index:
# print(f"nPerforming bottom-up retrieval for query: '{query_detail}'")
# paths = recursive_bottom_up_retrieval(query_detail_embedding, raptor_index, max_results=3)
#
# for i, path in enumerate(paths):
# print(f"--- Path {i+1} ---")
# for node in path:
# print(f" Level {node['level']} ({'Chunk' if node['level']==0 else 'Summary'}): {node['text'][:150]}...")
# print("-" * 30)
4.1.3 混合检索与答案生成 (Hybrid Retrieval & Answer Generation)
最强大的 RAPTOR 检索通常是混合式的：
- 初始多层搜索： 同时在多个层级（例如，最高层级和中间层级）进行相似性搜索，以捕捉宏观和中观的相关性。
- 上下文扩展： 对于初步检索到的相关节点，向上扩展到其父摘要以获取更广泛的上下文，向下扩展到其子节点以获取更详细的信息。
- 重排序与精炼： 使用一个更强大的重排序模型（如交叉编码器
cross-encoder）或再次使用 LLM 对检索到的所有相关文本进行相关性排序。 - 答案生成： 将重排序后的顶级相关文本（包括摘要和原始块）作为上下文，提供给一个大型语言模型，由其生成一个连贯、全面的最终答案。这通常被称为 RAG (Retrieval Augmented Generation) 模式。
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
# 假设使用一个交叉编码器进行重排序
# tokenizer_reranker = AutoTokenizer.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
# model_reranker = AutoModelForSequenceClassification.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
def rerank_results(query: str, retrieved_texts: List[str], tokenizer, model) -> List[Dict]:
"""
使用交叉编码器对检索到的文本进行重排序。
"""
if not retrieved_texts:
return []
features = tokenizer([query] * len(retrieved_texts), retrieved_texts, padding=True, truncation=True, return_tensors="pt")
with torch.no_grad():
scores = model(**features).logits.squeeze().tolist()
reranked_results = []
for i, text in enumerate(retrieved_texts):
reranked_results.append({'text': text, 'score': scores[i]})
reranked_results.sort(key=lambda x: x['score'], reverse=True)
return reranked_results
def generate_answer_with_llm(query: str, context_texts: List[str], llm_model: str = "gpt-3.5-turbo") -> str:
"""
利用检索到的上下文，使用LLM生成最终答案。
"""
if not context_texts:
return "未能找到足够的信息来回答您的问题。"
combined_context = "n---n".join(context_texts)
prompt = f"根据以下提供的上下文信息，请详细回答问题：'{query}'。nn上下文信息：n{combined_context}nn答案："
try:
# 模拟LLM调用
if openai.api_key:
response = openai.ChatCompletion.create(
model=llm_model,
messages=[
{"role": "system", "content": "You are a helpful assistant that answers questions based on provided context."},
{"role": "user", "content": prompt}
],
max_tokens=500,
temperature=0.5
)
answer = response.choices[0].message.content.strip()
else:
print("Warning: OpenAI API Key not found. Using dummy answer generator.")
answer = f"（模拟答案）根据提供的 {len(context_texts)} 段上下文，与您的查询 '{query}' 相关的信息是：...nn（请配置OpenAI API Key以获取真实答案）"
except Exception as e:
print(f"Error calling LLM for answer generation: {e}. Using dummy answer.")
answer = f"（模拟答案）由于LLM调用失败，未能生成答案。但根据上下文，与查询 '{query}' 相关的信息如下：...nn（错误信息：{e}）"
return answer
def hybrid_raptor_retrieval_and_generation(
query: str,
query_embedding: np.ndarray,
hierarchy: Dict[int, List[Dict]],
embedding_model: SentenceTransformer,
reranker_tokenizer=None,
reranker_model=None,
llm_model: str = "gpt-3.5-turbo",
max_context_chunks: int = 5,
top_k_per_level: int = 3,
min_similarity_threshold: float = 0.5
) -> str:
"""
结合多层检索、重排序和LLM生成答案。
"""
all_retrieved_nodes = []
# 1. 在多个层级进行初始检索
# 例如，在最高层级和最低层级（原始块）都进行搜索
max_level = max(hierarchy.keys())
# 从最高层级获取概览
top_level_matches = find_similar_nodes(query_embedding, hierarchy[max_level], top_k=top_k_per_level)
for match in top_level_matches:
if match['similarity'] >= min_similarity_threshold:
all_retrieved_nodes.append(match)
# 从最低层级获取具体细节
if 0 in hierarchy:
bottom_level_matches = find_similar_nodes(query_embedding, hierarchy[0], top_k=top_k_per_level)
for match in bottom_level_matches:
if match['similarity'] >= min_similarity_threshold:
all_retrieved_nodes.append(match)
# 同时获取其直接父摘要，作为上下文
parent_nodes = find_parents(match['id'], hierarchy)
if parent_nodes:
all_retrieved_nodes.extend(parent_nodes)
# 去重
unique_nodes = {node['id']: node for node in all_retrieved_nodes if node is not None}
retrieved_texts_for_reranking = [node['text'] for node in unique_nodes.values()]
final_context_texts = []
if reranker_tokenizer and reranker_model:
# 2. 重排序
reranked = rerank_results(query, retrieved_texts_for_reranking, reranker_tokenizer, reranker_model)
final_context_texts = [item['text'] for item in reranked[:max_context_chunks]]
print(f"Reranked {len(reranked)} texts, selected top {len(final_context_texts)} for context.")
else:
# 如果没有重排序器，就取相似度最高的几个
# 注意：这里需要重新计算相似度或从原始节点中保留相似度信息
# 简单起见，这里直接取原始检索到的前几个
print("No reranker provided. Taking top initial retrieved texts as context.")
# 这里需要更精细的逻辑来选择哪些文本作为上下文，例如根据相似度从all_retrieved_nodes中筛选
sorted_nodes = sorted(unique_nodes.values(), key=lambda x: cosine_similarity(query_embedding.reshape(1, -1), x['embedding'].reshape(1, -1))[0][0], reverse=True)
final_context_texts = [node['text'] for node in sorted_nodes[:max_context_chunks]]
# 3. LLM生成答案
answer = generate_answer_with_llm(query, final_context_texts, llm_model)
return answer
# # 示例：混合检索与答案生成
# # 确保 raptor_index 和 embedding_model 已初始化
# # 假设 reranker 已经加载
# # from transformers import AutoTokenizer, AutoModelForSequenceClassification
# # tokenizer_reranker = AutoTokenizer.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
# # model_reranker = AutoModelForSequenceClassification.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
#
# # 这里为了演示，我们假设 reranker_tokenizer 和 reranker_model 存在
# # 如果没有，hybrid_raptor_retrieval_and_generation 函数会退化到不使用重排序
# try:
# tokenizer_reranker = AutoTokenizer.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
# model_reranker = AutoModelForSequenceClassification.from_pretrained('cross-encoder/ms-marco-MiniLM-L-6-v2')
# except Exception as e:
# print(f"Could not load reranker model: {e}. Proceeding without reranker.")
# tokenizer_reranker = None
# model_reranker = None
#
# if raptor_index:
# query_rag = "请总结人工智能、机器学习和深度学习之间的关系，并提及它们在哪些领域有应用。"
# query_rag_embedding = generate_embeddings([query_rag])[0]
#
# print(f"nPerforming hybrid retrieval and answer generation for query: '{query_rag}'")
# final_answer = hybrid_raptor_retrieval_and_generation(
# query_rag,
# query_rag_embedding,
# raptor_index,
# embedding_model,
# reranker_tokenizer=tokenizer_reranker,
# reranker_model=model_reranker,
# max_context_chunks=7 # 增加上下文块数量
# )
# print("n--- Final Generated Answer ---")
# print(final_answer)4.2 结果呈现与用户交互
RAPTOR 的最终输出不仅仅是一个简单的答案，更可以是一个可导航的知识图谱。用户界面可以：
- 显示摘要链： 对于每个检索结果，展示从原始文档块到最高层摘要的路径，让用户理解上下文。
- 交互式下钻/上溯： 用户可以点击摘要，查看其子内容，或点击原始块，查看其父摘要。
- 多角度呈现： 针对同一查询，可以同时呈现不同抽象层级的答案，例如一个高层概览和几个详细的段落。
5. 实际考量与挑战
虽然 RAPTOR 提供了强大的功能，但在实际部署和应用中，也面临一些挑战：
- 计算资源消耗：
- LLM 摘要成本： 每次摘要生成都需要调用 LLM，这在处理海量文档时会产生巨大的计算开销和 API 调用费用。
- 嵌入生成： 即使是相对较小的嵌入模型，处理数百万甚至数十亿个文本块也需要大量计算。
- 聚类计算： 对于大规模数据集，聚类算法的计算复杂度也很高。
- 摘要质量控制： LLM 可能会出现“幻觉”现象，生成不准确或与原文不符的摘要。需要精心的提示工程和后处理来确保摘要的准确性和连贯性。
- 分块与聚类参数调优： 最佳的分块大小、重叠量，以及聚类算法的参数（如聚类数量、距离阈值等）高度依赖于具体的数据集和应用场景，需要反复实验和调优。
- 层次结构管理： 随着文档库的增长和更新，如何高效地增量更新 RAPTOR 索引是一个复杂的问题。简单的重建整个索引是不可行的。
- 语义漂移： 在多层抽象过程中，信息可能会丢失，或者高层摘要可能无法完全代表其所有子内容。保持语义一致性是一个挑战。
- 模型选择： 嵌入模型和摘要模型的选择对最终效果至关重要。需要权衡模型性能、计算成本和部署复杂性。
6. 进阶思考与未来展望
RAPTOR 框架为智能信息检索开辟了新的道路，但其潜力远不止于此。
- 动态摘要与查询引导： 探索在检索时动态生成摘要，而不是预先生成所有摘要。根据用户查询的特定需求，只对相关路径上的信息进行摘要，以节省计算资源并提高相关性。
- 多模态 RAPTOR： 将 RAPTOR 扩展到处理图像、视频、音频等多模态数据。例如，对视频片段进行摘要，并构建其层次化索引。
- 与知识图谱融合： 将 RAPTOR 生成的层次化摘要与结构化知识图谱相结合，形成更丰富、更易于查询的知识表示。摘要可以作为知识图谱中节点的文本描述。
- Agentic RAPTOR： 将 RAPTOR 作为自主代理（Autonomous Agents）的记忆和感知层。代理可以使用 RAPTOR 检索相关信息，然后利用这些信息进行决策、规划和执行任务。
- 可解释性与透明度： 提升 RAPTOR 决策过程的可解释性，例如，在生成摘要时，高亮显示摘要中对应的原始文本片段。
- 边缘计算与隐私保护： 研究在资源受限的环境下或需要严格隐私保护的场景中部署 RAPTOR 的方法。
RAPTOR 代表了我们从“信息搜索”到“知识发现”的转变。它不仅仅是找到信息，更是理解信息、组织信息，并最终将其转化为可操作的洞察力。
7. 开启智能信息获取的新纪元
RAPTOR 框架以其独特的递归抽象处理能力，为我们解析海量文档库、实现层次化摘要与检索提供了强大的工具。它克服了传统检索方法的局限，使得用户能够以更自然、更高效的方式与庞大的信息集合进行交互。虽然在计算成本、模型选择和参数调优方面仍存在挑战，但通过持续的技术创新和优化，RAPTOR 无疑将成为未来智能信息管理和知识发现领域的核心技术之一。它将赋能企业、研究机构和个人，在信息洪流中精准捕捉价值，开启智能信息获取的新纪元。