🧠AI RAG

Chunk与召回策略

面试回答

常见问法

文档切分和召回策略为什么会直接影响 RAG 效果?

回答

因为模型能回答什么,首先取决于你给它喂了什么。切分太碎会丢语义,切分太大又会降低召回精度;召回策略如果只靠单一路径,也容易漏掉关键资料。Chunk 和召回质量决定了后面生成阶段的上限。

# 智能Chunk策略示例
def smart_chunking(document, chunk_size=512, overlap=128):
    """
    智能文档切分策略
    
    1. 按语义边界切分
    2. 保持关键信息完整
    3. 支持重叠上下文
    """
    import re
    
    chunks = []
    
    # 按标题分段
    sections = re.split(r'\n(?=#|\d+\.)', document)
    
    for section in sections:
        # 如果section太长,进一步切分
        if len(section) > chunk_size * 2:
            # 按句子切分
            sentences = re.split(r'(?<=[.!?])\s+', section)
            
            current_chunk = ""
            for sentence in sentences:
                if len(current_chunk) + len(sentence) < chunk_size:
                    current_chunk += sentence + " "
                else:
                    # 保存当前chunk,开始新的
                    if current_chunk:
                        chunks.append(current_chunk.strip())
                    
                    # 支持重叠
                    if overlap and chunks:
                        overlap_text = chunks[-1][-overlap:]
                        current_chunk = overlap_text + sentence + " "
                    else:
                        current_chunk = sentence + " "
            
            if current_chunk:
                chunks.append(current_chunk.strip())
        else:
            chunks.append(section.strip())
    
    return chunks

# 混合召回策略
class HybridRetriever:
    def __init__(self, vector_db, keyword_index, rerank_model):
        self.vector_db = vector_db
        self.keyword_index = keyword_index
        self.rerank_model = rerank_model
    
    def retrieve(self, query, top_k=10, alpha=0.7):
        """
        混合召回:向量检索 + 关键词检索 + 重排序
        
        alpha: 向量检索权重,(1-alpha): 关键词检索权重
        """
        # 向量检索
        vector_results = self.vector_db.similarity_search(
            query, k=top_k * 2
        )
        
        # 关键词检索
        keyword_results = self.keyword_index.search(
            query, limit=top_k * 2
        )
        
        # 合并结果
        all_results = vector_results + keyword_results
        
        # 去重
        unique_results = self.deduplicate(all_results)
        
        # 重排序
        reranked_results = self.rerank_model.rerank(
            query, unique_results, top_n=top_k
        )
        
        return reranked_results
    
    def deduplicate(self, results):
        """基于内容去重"""
        seen_content = set()
        unique_results = []
        
        for result in results:
            # 使用simhash或简单hash去重
            content_hash = hash(result['content'][:100])
            if content_hash not in seen_content:
                seen_content.add(content_hash)
                unique_results.append(result)
        
        return unique_results

追问

  • 固定长度切分和语义切分怎么选?(根据文档类型)
  • 为什么混合检索常比单一向量检索更稳?(召回多样性)
  • top-k 应该怎么调?(根据文档质量和数量)

原理展开

Chunk 的目标是在”语义完整”和”检索粒度”之间平衡。常见做法包括滑动窗口、按标题分段、按语义边界分块。召回层则可能结合关键词检索、向量检索和 metadata 过滤。

如果系统资料类型复杂,单一向量检索很容易漏召回,因此混合检索和后续 rerank 很常见。面试里最好把 chunk、召回、rerank 说成一条链路,而不是孤立优化点。

# 高级Chunk:基于语义边界的切分
def semantic_chunking(text, min_chunk_size=200, max_chunk_size=800):
    """
    基于语义的文档切分
    
    1. 使用句子嵌入计算相似度
    2. 在语义变化大的地方切分
    3. 保持语义完整性
    """
    from sentence_transformers import SentenceTransformer
    import numpy as np
    
    model = SentenceTransformer('all-MiniLM-L6-v2')
    
    # 先按句子切分
    sentences = re.split(r'(?<=[.!?])\s+', text)
    
    # 计算句子嵌入
    sentence_embeddings = model.encode(sentences)
    
    chunks = []
    current_chunk = [sentences[0]]
    current_embedding = sentence_embeddings[0]
    
    for i in range(1, len(sentences)):
        # 计算与当前chunk的相似度
        similarity = np.dot(current_embedding, sentence_embeddings[i]) / (
            np.linalg.norm(current_embedding) * np.linalg.norm(sentence_embeddings[i])
        )
        
        # 如果相似度低于阈值,或者超过最大长度,切分
        current_text = ' '.join(current_chunk)
        if (len(current_text) > min_chunk_size and similarity < 0.6) or \
           len(current_text) > max_chunk_size:
            
            chunks.append(current_text)
            current_chunk = [sentences[i]]
            current_embedding = sentence_embeddings[i]
        else:
            current_chunk.append(sentences[i])
            # 更新chunk的平均embedding
            current_embedding = (current_embedding + sentence_embeddings[i]) / 2
    
    # 添加最后一个chunk
    if current_chunk:
        chunks.append(' '.join(current_chunk))
    
    return chunks

# 召回策略调优
def adaptive_retrieval(query, documents, retriever, 
                      diversity_threshold=0.7, 
                      quality_threshold=0.5):
    """
    自适应召回策略
    
    1. 根据查询复杂度调整召回数量
    2. 保证召回多样性
    3. 过滤低质量文档
    """
    # 分析查询复杂度
    query_complexity = analyze_query_complexity(query)
    
    # 根据复杂度调整召回数量
    if query_complexity == 'high':
        initial_top_k = 15
    elif query_complexity == 'medium':
        initial_top_k = 10
    else:
        initial_top_k = 5
    
    # 初步召回
    candidates = retriever.retrieve(query, top_k=initial_top_k * 2)
    
    # 多样性过滤
    diverse_results = filter_by_diversity(
        candidates, diversity_threshold
    )
    
    # 质量过滤
    high_quality_results = filter_by_quality(
        diverse_results, quality_threshold
    )
    
    # 最终召回数量
    final_results = high_quality_results[:initial_top_k]
    
    return final_results

易错点

  • 一味追求更大 chunk(召回精度下降)
  • 不区分召回问题和排序问题(不同阶段优化目标)
  • 忽略不同类型文档的Chunk策略差异
  • 混合检索参数调优不当

记忆技巧

记住Chunk和召回的核心原则:

  1. Chunk平衡 = “语义完整 vs 检索精度”
  2. 混合召回 = “向量+关键词+重排序”
  3. 参数调优 = “根据数据特点调整”

典型应用场景:

  • 技术文档:按标题Chunk+精确召回
  • 小说文档:按段落Chunk+语义召回
  • 多源文档:混合Chunk+多样性召回
  • 实时系统:快速Chunk+增量召回