RAG高阶技巧-如何实现窗口上下文检索

在RAG（检索增强生成）模型的落地实践中，检索效果往往是决定最终回答质量的关键瓶颈。很多团队明明用了语义向量、调了相似度阈值，但生成的回答依然不够精准——问题出在哪里？大概率出在检索环节的“粒度”上。今天分享一个高阶技巧：窗口上下文检索。它能同时兼顾检索精度和上下文的完整性，让RAG模型的输出更贴近真实场景。

基础RAG存在的问题及解决方案

基础RAG检索流程

RAG融合了检索与生成，是当前AI落地的热门方案。其核心流程很清晰：从外部知识库（比如维基百科）中检索出与问题相关的文档，再将这些文档喂给大模型生成回答。具体拆解下来，大致包括这几个环节：

• 加载文档
  ：把PDF、表格等不同格式的文件转为原始文本。

• 文档拆分
  ：将长文本切分成适合向量存储的小单元，比如把“我是kxc。我喜欢唱跳，rap，和篮球。”拆成“我是kxc。”和“我喜欢唱跳，rap，和篮球。”两个数据块（通常称为chunk）。

• 向量化
  ：用BERT或TF-IDF等模型将每个chunk转为向量表示。

• 存储
  ：把向量化的chunk存入向量数据库。

• 检索
  ：计算问题向量与文档向量的相似度（如余弦相似度），选出最相关的几个chunk。

• 生成回答
  ：将检索到的chunk拼接后作为上下文，输入大模型生成最终回答。

基础RAG的痛点

流程看起来简单，但实际跑起来就会发现问题集中在

和这两个环节。拆得太大或太小，都会导致检索结果“水土不服”。

• 拆得太大的情况
  ：比如把一整个维基百科页面作为一个文档，里面可能混杂了N个主题。检索时虽然命中了一个“相关”的页面，但页面里大量无关内容会稀释信息密度，大模型容易受干扰，回答自然跑偏。

• 拆得太小的情况
  ：比如切成一个个短句，检索精度上去了，匹配到的句子确实相关。但到了生成环节，模型只看到一句孤零零的“我喜欢唱跳”，却不知道这句话的前因后果——缺少上下文支撑，回答自然不完整。

解法：窗口上下文检索

要解决这个两难困境，思路其实很朴素：

。具体做法分三步：

• 第一步，拆分时切到最小的语义单元（句子或短段落），并给每个单元一个编号，比如按原文顺序标记为1、2、3……
• 第二步，检索时根据向量相似度找出最相关的chunk，同时记录下它们的编号。
• 第三步，根据编号找到这些chunk前后若干个相邻单元（窗口），把窗口内的内容拼成一个完整的文档片段，再交给大模型。

这样一来，检索到的数据既精准（命中核心语义），又完整（带了上下文），大模型自然能给出更靠谱的回答。

窗口上下文检索实践

实现思路

要实现上述方案，核心就是建立每个chunk与其前后邻居的关联关系。怎么做？极其简单——在拆分时给每个chunk按顺序编号，并把这个编号作为元数据存进向量数据库。检索时，通过当前chunk的编号就能定位到它的邻居chunk，从而拉取上下文。

基于Chroma的代码示例

下面用一段实际的代码来演示整个流程。假设我们从一篇博客中加载文档，使用LangChain和Chroma向量库。

import bs4, uuid
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
doc = loader.load()

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
docs = text_splitter.split_documents(doc)
# 给每个chunk注入file_id和chunk_id
file_id = uuid.uuid4().hex
chunk_id_counter = 0
for doc in docs:
    doc.metadata["file_id"] = file_id
    doc.metadata["chunk_id"] = f'{file_id}_{chunk_id_counter}'
    chunk_id_counter += 1
    for key, value in doc.metadata.items():
        if not isinstance(value, (str, int, float, bool)):
            doc.metadata[key] = str(value)

vectorstore = Chroma.from_documents(documents=docs, embedding=OpenAIEmbeddings())

def expand_doc(group):
    new_cands = []
    group.sort(key=lambda x: int(x.metadata['chunk_id'].split('_')[-1]))
    id_set = set()
    file_id = group[0].metadata['file_id']
    group_scores_map = {}
    cand_chunks = []
    for cand_doc in group:
        current_chunk_id = int(cand_doc.metadata['chunk_id'].split('_')[-1])
        group_scores_map[current_chunk_id] = cand_doc.metadata['score']
        for i in range(current_chunk_id - 200, current_chunk_id + 200):
            need_search_id = file_id + '_' + str(i)
            if need_search_id not in cand_chunks:
                cand_chunks.append(need_search_id)
    where = {"chunk_id": {"$in": cand_chunks}}
    ids, group_relative_chunks = get(where)
    group_chunk_map = {int(item.metadata['chunk_id'].split('_')[-1]): item.page_content for item in group_relative_chunks}
    group_file_chunk_num = list(group_chunk_map.keys())
    for cand_doc in group:
        current_chunk_id = int(cand_doc.metadata['chunk_id'].split('_')[-1])
        doc = copy.deepcopy(cand_doc)
        id_set.add(current_chunk_id)
        docs_len = len(doc.page_content)
        for k in range(1, 200):
            break_flag = False
            for expand_index in [current_chunk_id + k, current_chunk_id - k]:
                if expand_index in group_file_chunk_num:
                    merge_content = group_chunk_map[expand_index]
                    if docs_len + len(merge_content) > CHUNK_SIZE:
                        break_flag = True
                        break
                    else:
                        docs_len += len(merge_content)
                        id_set.add(expand_index)
            if break_flag:
                break
    id_list = sorted(list(id_set))
    id_lists = seperate_list(id_list)
    for id_seq in id_lists:
        for id in id_seq:
            if id == id_seq[0]:
                doc = Document(page_content=group_chunk_map[id],
                               metadata={"score": 0, "file_id": file_id})
            else:
                doc.page_content += " " + group_chunk_map[id]
        doc_score = min([group_scores_map[id] for id in id_seq if id in group_scores_map])
        doc.metadata["score"] = doc_score
        new_cands.append(doc)
    return new_cands

总结

窗口上下文检索的核心思路其实就一句话：拆细点，但检索时带亲戚。它既保留了小粒度带来的高匹配精度，又通过邻居窗口弥补了上下文的缺失。从实践来看，这个方法对RAG系统的检索质量提升非常明显，尤其是在处理长文档或复杂问答时。

希望这篇文章能帮助你理解并落地这个技巧。如果你在实践中遇到过其他检索难题，或者有更好的优化思路，欢迎在评论区交流讨论。谢谢阅读。