otter_ai1. Small-to-Big分块检索策略解析
1.1 传统分块策略的局限性
在信息检索领域，文档分块策略直接影响着检索系统的性能表现。传统单粒度分块方法通常面临一个两难选择：选择小分块还是大分块？
小分块（100-200 token）的优势在于检索精度高，能够准确定位到与查询最相关的文本片段。然而，这种方法的致命缺陷是上下文信息的缺失。当我们将文档切分得过细时，每个文本块只能提供局部的信息片段，大语言模型难以理解这些片段之间的逻辑关联，容易产生"幻觉"回答。
大分块（500-1000 token）虽然能提供更完整的上下文信息，但会引入大量噪声内容。更严重的是，在向量检索过程中，大文本块的嵌入表示往往是其内部多个语义单元的平均，这会导致检索时漏掉关键细节。特别是在处理事实性问答时，这种"信息稀释"效应尤为明显。
1.2 Small-to-Big的创新设计
Small-to-Big策略通过引入两级分块机制，巧妙地解决了上述矛盾。其核心思想是：
-
子块（Child Chunks）：尺寸较小（通常100-200 token），专门用于向量相似度检索。这些小块的语义表示更加集中，能够精准匹配用户查询意图。
-
父块（Parent Chunks）：尺寸较大（通常500-1000 token或完整段落），包含子块所在的完整上下文。这些大块在检索阶段不直接参与向量匹配，而是在获取相关子块后作为生成阶段的上下文来源。
这种分层设计的关键在于建立了子块与父块之间的映射关系。每个子块都记录其所属父块的标识符，使得系统能够：
- 先用小分块实现精准检索
- 再通过映射关系获取完整上下文
- 最终实现"精准定位+完整理解"的双重目标
1.3 技术实现要点
在实际工程实现中，有几个关键技术点需要特别注意：
分块重叠设计：相邻子块之间应保持适当重叠（通常10-20%），避免关键信息恰好落在分块边界而被切断。这种重叠虽然会增加存储开销，但能显著提高检索召回率。
元数据管理：每个子块需要存储丰富的元数据，包括但不限于：
- 父块ID（建立映射关系的关键）
- 文档来源信息
- 位置信息（在原文中的起止位置）
- 其他业务相关属性
索引策略：父块通常只需要建立简单的键值存储（通过ID快速查找），而子块则需要构建高效的向量索引。对于大规模应用，建议采用专业向量数据库如Milvus或Qdrant。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. LangGraph框架深度适配
2.1 为什么选择LangGraph
LangGraph作为LangChain团队推出的新一代框架，特别适合实现复杂的检索增强生成(RAG)流程。与传统的链式(Chain)结构相比，LangGraph提供了更强大的功能：
- 状态化管理：整个流程的状态可以持久化和传递，方便调试和监控
- 灵活拓扑：支持循环、分支等复杂流程，而不仅限于线性链条
- 模块化设计：每个处理步骤可以独立开发和测试，再组合成完整流程
- 可视化支持：能够直观展示整个处理流程的拓扑结构
这些特性使得LangGraph成为实现Small-to-Big策略的理想选择，特别是当我们需要在标准流程中加入重排序、查询扩展等增强功能时。
2.2 系统架构设计
基于LangGraph的Small-to-Big系统通常包含以下核心组件：
-
离线处理管道：
- 文档加载与解析
- 两级分块处理
- 向量嵌入计算
- 存储索引构建
-
在线查询管道：
mermaid复制graph LR A[用户查询] --> B[查询理解] B --> C[子块检索] C --> D[父块获取] D --> E[结果重排序] E --> F[答案生成] F --> G[响应返回] -
增强模块（可选）：
- 查询扩展
- 混合检索（向量+关键词）
- 结果去重
- 相关性验证
2.3 状态机建模
在LangGraph中，我们将整个检索生成流程建模为一个状态机。以下是典型的状态定义：
python复制from typing import TypedDict, List, Dict, Optional
class RAGState(TypedDict):
# 输入相关
original_query: str
processed_query: str
# 检索相关
child_chunks: List[Dict] # 检索到的子块列表
parent_chunks: List[str] # 获取的父块内容
# 增强相关
expanded_queries: List[str] # 查询扩展结果
hybrid_results: List[Dict] # 混合检索结果
# 生成相关
ranked_contexts: List[str] # 最终上下文
generated_answer: str # 最终答案
# 元信息
timestamps: Dict[str, float] # 各阶段耗时
debug_info: Dict[str, Any] # 调试信息
这种细粒度的状态划分使得我们可以：
- 精确控制每个节点的输入输出
- 方便地添加监控指标
- 支持断点调试和错误恢复
- 实现流程的灵活组合
3. 核心实现细节
3.1 分块与索引实现
父子块生成算法
实现高质量的两级分块需要考虑多种因素：
- 语义完整性：分块边界应该尽可能落在自然段落或语义单元边界
- 长度控制：确保子块和父块在预设的token范围内
- 重叠处理：智能处理重叠区域，避免信息重复
以下是改进后的分块算法实现：
python复制from langchain.text_splitter import TextSplitter
import re
class HierarchicalTextSplitter:
def __init__(self, child_size=200, child_overlap=20, parent_size=800):
self.child_splitter = TextSplitter(
chunk_size=child_size,
chunk_overlap=child_overlap
)
self.parent_size = parent_size
def split_text(self, text):
# 先按段落粗分
paragraphs = re.split(r'\n\n+', text)
parents = []
current_parent = []
current_size = 0
for para in paragraphs:
para_size = len(para.split())
if current_size + para_size > self.parent_size and current_parent:
parents.append("\n\n".join(current_parent))
current_parent = []
current_size = 0
current_parent.append(para)
current_size += para_size
if current_parent:
parents.append("\n\n".join(current_parent))
# 对每个父块生成子块
parent_child_map = {}
for i, parent in enumerate(parents):
parent_id = f"parent_{i}"
child_chunks = self.child_splitter.split_text(parent)
# 为每个子块添加父块引用
for j, chunk in enumerate(child_chunks):
chunk_id = f"{parent_id}_child_{j}"
parent_child_map[chunk_id] = {
"text": chunk,
"parent_id": parent_id,
"position": j
}
return {
"parents": parents,
"children": parent_child_map
}
索引优化策略
为了提高检索效率，我们需要对索引结构进行特别优化：
-
向量索引：
- 使用量化技术减少存储空间
- 采用HNSW等近似最近邻算法加速检索
- 对高频查询建立缓存
-
文档存储：
- 父块采用压缩存储
- 建立内存缓存层
- 实现批量预取机制
3.2 检索流程实现
子块检索节点
python复制def retrieve_child_chunks(state: RAGState):
query = state["processed_query"]
# 获取嵌入向量
query_embedding = embed_text(query)
# 执行向量检索
raw_results = vector_index.search(
query_embedding,
top_k=10,
filters={"status": "active"} # 可添加业务过滤条件
)
# 处理结果
child_chunks = []
for result in raw_results:
child_chunks.append({
"id": result.id,
"text": result.text,
"parent_id": result.metadata["parent_id"],
"score": result.score,
"metadata": result.metadata
})
# 更新状态
return {"child_chunks": child_chunks}
父块获取节点
python复制async def fetch_parent_chunks(state: RAGState):
child_chunks = state["child_chunks"]
# 收集唯一父块ID
parent_ids = {c["parent_id"] for c in child_chunks}
# 批量获取父块
parent_texts = await docstore.batch_get(parent_ids)
# 保持原始顺序
ordered_parents = [
parent_texts[pid] for pid in parent_ids
if pid in parent_texts
]
return {
"parent_chunks": ordered_parents,
"timestamps.fetch_parent": time.time()
}
3.3 增强模块实现
混合检索增强
python复制def hybrid_search(state: RAGState):
query = state["processed_query"]
# 并行执行向量和关键词检索
vector_results = vector_index.search(query, top_k=10)
keyword_results = bm25_index.search(query, top_k=10)
# 使用RRF算法融合结果
combined = reciprocal_rank_fusion(
vector_results,
keyword_results,
k=60
)
return {"hybrid_results": combined[:15]}
动态重排序
python复制def dynamic_rerank(state: RAGState):
query = state["processed_query"]
candidates = state["parent_chunks"]
# 使用轻量级模型初筛
if len(candidates) > 10:
scores = fast_ranker.score(query, candidates)
candidates = [c for c, s in zip(candidates, scores) if s > threshold]
# 使用强模型精排
pairs = [(query, text) for text in candidates]
detailed_scores = cross_encoder.predict(pairs)
# 组合排序
ranked = sorted(
zip(candidates, detailed_scores),
key=lambda x: x[1],
reverse=True
)
return {
"ranked_contexts": [r[0] for r in ranked[:5]],
"rerank_scores": [r[1] for r in ranked[:5]]
}
4. 生产环境优化策略
4.1 性能优化技巧
- 批量处理：对多个查询进行批量化处理，提高GPU利用率
- 缓存策略：
- 对频繁查询建立结果缓存
- 对嵌入向量建立缓存
- 对父块内容建立缓存
- 预计算：对静态内容预先计算可能需要的衍生数据
- 异步IO：对网络和磁盘IO密集型操作使用异步模式
4.2 质量提升方法
-
分块评估指标：
- 设计专门的测试集评估分块质量
- 监控检索准确率和召回率
- 跟踪生成答案的事实准确性
-
动态分块策略：
- 根据文档类型自动调整分块参数
- 对表格、代码等特殊内容采用专门处理
- 实现内容感知的分块算法
-
反馈循环：
- 收集用户对答案质量的反馈
- 记录系统各环节的中间结果
- 建立持续改进机制
4.3 监控与调试
建立完善的监控体系对生产环境至关重要：
-
性能监控：
- 各阶段耗时统计
- 资源使用情况
- 吞吐量指标
-
质量监控：
- 检索结果相关性
- 生成答案准确性
- 用户满意度指标
-
调试工具：
- 查询重放功能
- 中间结果检查
- 差异比较工具
python复制class MonitoringMiddleware:
def __init__(self, app):
self.app = app
self.metrics = defaultdict(list)
async def __call__(self, state):
start_time = time.time()
result = await self.app(state)
end_time = time.time()
# 记录执行时间
self.metrics["latency"].append(end_time - start_time)
# 记录各阶段状态
if "timestamps" in result:
stages = sorted(result["timestamps"].items(), key=lambda x: x[1])
for i in range(len(stages)-1):
stage_name = stages[i][0]
duration = stages[i+1][1] - stages[i][1]
self.metrics[f"stage_{stage_name}"].append(duration)
return result
5. 典型问题与解决方案
5.1 检索相关问题
问题1：检索结果不相关
- 可能原因：子块太小导致语义不完整
- 解决方案：调整子块大小，增加重叠区域
问题2：重要信息被切分
- 可能原因：分块边界不合理
- 解决方案：改进分块算法，考虑语义边界
问题3：检索速度慢
- 可能原因：向量索引效率低
- 解决方案：优化索引参数，使用量化技术
5.2 生成相关问题
问题1：答案缺乏上下文
- 可能原因：父块太小
- 解决方案：调整父块大小，确保包含足够背景
问题2：答案包含矛盾信息
- 可能原因：不同父块之间存在冲突
- 解决方案：增加一致性检查步骤
问题3：生成速度慢
- 可能原因：LLM响应时间长
- 解决方案：使用缓存，优化prompt
5.3 系统级问题
问题1：内存占用高
- 可能原因：缓存策略不当
- 解决方案：实现LRU缓存，控制内存使用
问题2：扩展性差
- 可能原因：架构设计限制
- 解决方案：采用微服务架构，分离组件
问题3：更新延迟高
- 可能原因：全量重建索引
- 解决方案：实现增量更新机制
6. 进阶应用场景
6.1 多文档关联检索
Small-to-Big策略可以扩展到跨文档检索场景：
- 建立文档间的引用关系图
- 检索时沿着引用关系扩展上下文
- 实现真正意义上的多跳推理
python复制def multi_hop_retrieval(state: RAGState):
initial_chunks = state["child_chunks"]
related_docs = set()
# 第一跳：初始检索
for chunk in initial_chunks:
related_docs.add(chunk["doc_id"])
# 第二跳：基于文档关系扩展
for doc_id in list(related_docs):
for related_id in doc_graph.get_related(doc_id):
related_docs.add(related_id)
# 获取扩展后的内容
additional_parents = docstore.batch_get(related_docs)
return {
"parent_chunks": state["parent_chunks"] + additional_parents
}
6.2 时序感知检索
对于包含时序信息的内容（如新闻、日志），可以增强时序感知能力：
- 在元数据中记录时间信息
- 检索时考虑时间相关性
- 按时间线组织上下文
6.3 多模态扩展
将Small-to-Big理念扩展到多模态场景：
- 对文本、图像、表格等内容分别处理
- 建立跨模态的关联关系
- 实现统一检索和生成
在实际业务场景中，我们还需要考虑如何平衡检索质量与系统性能。经过多次实验，我发现以下参数组合在大多数情况下表现良好：
- 子块大小：256-384 token
- 子块重叠：15-25%
- 父块大小：1024-1536 token
- 初始检索数量：8-12个子块
- 最终上下文数量：3-5个父块
这些参数需要根据具体业务需求和文档特性进行调整。建议建立自动化测试流程，通过A/B测试确定最优配置。