02 RAG技术与应用

RAG（Retrieval-Augmented Generation）技术与应用

RAG（Retrieval-Augmented Generation），即检索增强生成，是一种结合了信息检索和文本生成的技术。它通过实时检索外部知识并将其作为上下文输入到大型语言模型（LLM）中，以提高生成结果的时效性和准确性。

大模型应用开发模式

大模型应用开发主要有三种模式：

提示工程 (Prompt Engineering): 侧重于通过优化用户输入（Prompt）来引导LLM生成期望的输出。
RAG (Retrieval-Augmented Generation): 解决LLM缺乏实时或专业领域知识的问题。
微调 (Fine-tuning): 通过在特定数据集上对LLM进行额外训练，使其更好地适应特定任务或领域。

在LLM出现错误回复时，RAG主要用于解决“缺乏背景知识”的问题。

RAG的优势

解决知识时效性问题: LLM训练数据通常是静态的，RAG能够检索外部知识库以实时更新信息。
减少模型幻觉 (Hallucination): 引入外部知识可以减少模型生成虚假或不准确内容的可能性。
提升专业领域回答质量: RAG能结合垂直领域的专业知识库，生成更具专业深度的回答。

RAG的核心原理与流程

RAG流程通常分为三个主要阶段：

Step1：数据预处理 (Indexing)

知识库构建: 收集并整理文档、网页、数据库等多源数据。
文档分块 (Chunking): 将文档切分为适当大小的片段（chunks），以平衡语义完整性与检索效率。
向量化处理: 使用嵌入模型（Embedding Model）将文本块转换为向量，并存储在向量数据库中。

Step2：检索阶段 (Retrieval)

查询处理: 将用户输入的问题转换为向量。
相似度检索: 在向量数据库中进行相似度检索，找到最相关的文本片段。
重排序 (Re-ranking): 对检索结果进行相关性排序，选择最相关的片段作为生成阶段的输入。

Step3：生成阶段 (Generation)

上下文组装: 将检索到的文本片段与用户问题结合，形成增强的上下文输入。
生成回答: 大语言模型基于增强的上下文生成最终回答。

NativeRAG

NativeRAG指的是RAG的三个核心步骤：Indexing、Retrieval、Generation。虽然概念简单，但在实际构建和落地过程中涉及较多复杂的工作内容，如如何更好地存储知识、如何在海量知识中找到有用部分，以及如何结合提问和知识生成有用答案。

Embedding模型选择

选择合适的Embedding模型对RAG系统的性能至关重要。Hugging Face的MTEB排行榜提供了1000多种语言中100多种文本嵌入模型的比较。

常见的Embedding模型分类及示例：

通用文本嵌入模型

Code
_BGE-M3 (智源研究院)_*: 支持100+语言，输入长度达8192 tokens，融合密集、稀疏、多向量混合检索，适用于跨语言长文档检索、高精度RAG应用。 _text-embedding-3-large (OpenAI)_*: 向量维度3072，长文本语义捕捉能力强，英文表现优秀。 _Jina-embeddings-v2-small (Jina AI)_*: 参数量仅35M，支持实时推理（RT<50ms），适合轻量化部署。
中文嵌入模型

Code
_xiaobu-embedding-v2_*: 针对中文语义优化，语义理解能力强，适用于中文文本分类、语义检索。 _M3E-Base_*: 针对中文优化的轻量模型，适合本地私有化部署，适用于中文法律、医疗领域检索任务。 _stella-mrl-large-zh-v3.5-1792_*: 处理大规模中文数据能力强，捕捉细微语义关系，适用于中文文本高级语义分析、自然语言处理任务。
指令驱动与复杂任务模型

Code
_gte-Qwen2-7B-instruct (阿里巴巴)_*: 基于Qwen大模型微调，支持代码与文本跨模态检索，适用于复杂指令驱动任务、智能问答系统。 _E5-mistral-7B (Microsoft)_*: 基于Mistral架构，Zero-shot任务表现优异，适用于动态调整语义密度的复杂系统。

代码示例：BGE-M3 使用

Python

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True) # Setting use_fp16 to True speeds up computation with a slight performance degradation
sentences_1 = ["What is BGE M3?", "Defination of BM25"]
sentences_2 = ["BGE M3 is an embedding model supporting dense retrieval, lexical matching and multi-vector interaction.", 
               "BM25 is a bag-of-words retrieval function that ranks a set of documents based on the query terms appearing in each document"]

embeddings_1 = model.encode(sentences_1, 
                            batch_size=12, 
                            max_length=8192 # If you don't need such a long length, you can set a smaller value to speed up the encoding process.
                           )['dense_vecs']
embeddings_2 = model.encode(sentences_2)['dense_vecs']

similarity = embeddings_1 @ embeddings_2.T
print(similarity)
# 输出示例:
# [[0.626  0.3477]
#  [0.3499 0.678 ]]

similarity = embeddings_1 @ embeddings_2.T 通过矩阵乘法计算了两组句子嵌入向量之间的余弦相似度矩阵。

代码示例：gte-Qwen2 使用 (SentenceTransformer 封装)

Python

from sentence_transformers import SentenceTransformer

model_dir = "/root/autodl-tmp/models/iic/gte_Qwen2-1___5B-instruct" # 模型路径
model = SentenceTransformer(model_dir, trust_remote_code=True)
model.max_seq_length = 8192

queries = [
    "how much protein should a female eat",
    "summit define",
]
documents = [
    "As a general guideline, the CDC's average requirement of protein for women ages 19 to 70 is 46 grams per day. But, as you can see from this chart, you'll need to increase that if you're expecting or training for a marathon. Check out the chart below to see how much protein you should be eating each day.",
    "Definition of summit for English Language Learners. : 1 the highest point of a mountain : the top of a mountain. : 2 the highest level. : 3 a meeting or series of meetings between the leaders of two or more governments.",
]

query_embeddings = model.encode(queries, prompt_name="query")
document_embeddings = model.encode(documents)

scores = (query_embeddings @ document_embeddings.T) * 100
print(scores.tolist())
# 输出示例:
# [[78.49691772460938, 17.04286003112793],
#  [14.924489974975586, 75.37960815429688]]

CASE：DeepSeek + Faiss 搭建本地知识库检索

本项目旨在构建一个基于RAG的本地知识库检索系统，回答关于“客户经理考核办法”PDF文件的问题。

1. RAG架构:

检索 (Retrieval): 使用向量相似度搜索从PDF文档中检索相关内容。
增强 (Augmentation): 将检索到的文档片段作为上下文。
生成 (Generation): 基于上下文和用户问题生成答案。

2. 技术栈选择:

向量数据库: Faiss (高效向量检索)。
嵌入模型: 阿里云DashScope的text-embedding-v1。
大语言模型: deepseek-v3。
文档处理: PyPDF2 (PDF文本提取)。
框架: LangChain (问答链)。

3. 程序的逻辑结构:

Step1：文档预处理

PDF文本提取:
- 逐页提取文本内容。
- 记录每行文本对应的页码信息。
- 处理空页和异常情况。
文本分割策略:
- 使用RecursiveCharacterTextSplitter。
- 分割参数：chunk_size=1000, chunk_overlap=200。
- 分割符优先级：段落→句子→空格→字符。
页码映射处理:
- 基于字符位置计算每个文本块的页码。
- 使用众数统计确定文本块的主要来源页码。
- 建立文本块与页码的映射关系。

Python

from PyPDF2 import PdfReader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from typing import List, Tuple
# from loguru import Logger # 假设有Logger

def extract_text_with_page_numbers(pdf) -> Tuple[str, List[int]]:
    """
    从PDF中提取文本并记录每行文本对应的页码
    参数: pdf: PDF文件对象
    返回: text: 提取的文本内容, page_numbers: 每行文本对应的页码列表
    """
    text = ""
    page_numbers = []
    for page_number, page in enumerate(pdf.pages, start=1):
        extracted_text = page.extract_text()
        if extracted_text:
            text += extracted_text
            # 简化处理：将整页的文本都标记为当前页码
            # 更精确的映射需要更复杂的逻辑，例如按行或字符范围
            page_numbers.extend([page_number] * len(extracted_text.split("\n")))
        # else:
            # Logger.warning(f"No text found on page {page_number}.")
    return text, page_numbers

def process_text_with_splitter(text: str, page_numbers: List[int], DASHSCOPE_API_KEY: str) -> FAISS:
    """
    处理文本并创建向量存储
    参数: text: 提取的文本内容, page_numbers: 每行文本对应的页码列表
    返回: knowledgeBase: 基于FAISS的向量存储对象
    """
    text_splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", ".", " ", ""],
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
    )
    chunks = text_splitter.split_text(text)
    print(f"文本被分割成{len(chunks)} 个块。")

    embeddings = DashScopeEmbeddings(
        model="text-embedding-v1",
        dashscope_api_key=DASHSCOPE_API_KEY,
    )
    
    # 原始实现中page_info的创建可能存在问题，因为它简单地将chunk与page_numbers的索引对应
    # 正确的页码映射需要更复杂的逻辑，例如追踪每个chunk在原始文本中的起始和结束位置
    # 这里为了演示，我们假设每个chunk能大致对应一个页码（例如取第一个字符所在页码）
    # 实际应用中需要更精确的页码对应逻辑
    metadatas = []
    # 简化处理：对于每个块，尝试找到其在原始文本中的大致位置，并推断页码
    # 这是一个简化且可能不准确的页码映射方法，实际应用中需要更精确的逻辑
    chunk_start_idx = 0
    for i, chunk in enumerate(chunks):
        chunk_end_idx = chunk_start_idx + len(chunk)
        # 简单地取chunk的起始位置在原始文本中的行数，然后映射到页码

Step2：知识库构建

使用DashScope嵌入模型生成向量。
将向量存储到Faiss索引结构中。
数据持久化：保存Faiss索引文件（.faiss）、元数据信息（.pkl）和页码映射关系。

Step3：问答查询

相似度检索: 将用户问题转换为向量，在Faiss中搜索最相似的文档块，返回Top-K相关文档。
问答链处理: 使用LangChain的load_qa_chain，采用stuff策略组合文档，将组合后的上下文和问题发送给LLM。
答案生成与展示: LLM生成答案后，展示结果并记录来源页码。

Python

from langchain.chains.question_answering import load_qa_chain
from langchain_community.callbacks.manager import get_openai_callback
from langchain_community.llms import Tongyi # 或其他LLM封装

# # 假设已经有了knowledgeBase对象和DASHSCOPE_API_KEY
# llm = Tongyi(model_name="deepseek-v3", dashscope_api_key=DASHSCOPE_API_KEY)

# query = "客户经理被投诉了，投诉一次扣多少分"
# # query = "客户经理每年评聘申报时间是怎样的？" # 第二个测试问题

# if query:
#     docs = knowledgeBase.similarity_search(query) # 执行相似度搜索

#     chain = load_qa_chain(llm, chain_type="stuff") # 加载问答链

#     input_data = {"input_documents": docs, "question": query}

#     # with get_openai_callback() as cost: # 如果是OpenAI模型可以使用此回调
#     response = chain.invoke(input=input_data)
#     # print(f"查询已处理。成本: {cost}")

#     print(response["output_text"])
#     print("来源:")
#     unique_pages = set()
#     for doc in docs:
#         # 页码信息现在应该在doc.metadata中
#         source_page = doc.metadata.get("source_page", "未知")
#         if source_page not in unique_pages:
#             unique_pages.add(source_page)
#             print(f"文本块页码: {source_page}")

页码映射问题说明: 原始材料中页码映射的逻辑knowledgeBase.page_info = {chunk: page_numbers[i] for i, chunk in enumerate(chunks)}可能存在问题，因为chunks是文本分割后的结果，直接与原始的page_numbers列表按索引对应并不精确。更准确的页码映射需要根据chunk在原始text中的起始字符位置来确定其所属的页码。在LangChain中，通常通过metadata来存储每个Document（即这里的chunk）的来源信息。

LangChain中的问答链 (chain_type)

LangChain问答链提供了四种chain_type策略来处理检索到的文档：

stuff (Stuffing):

Code
_原理_*: 直接将所有文档作为prompt输入给LLM。 _适用场景_*: 文档拆分得比较小，一次获取文档较少，且总token数不超过LLM上下文窗口限制的情况。 _优势_*: 调用LLM次数最少，效率高，文档间上下文保留完整。 _局限_*: 容易超出LLM的上下文窗口限制。
map_reduce:

Code
_原理_*: 对于每个文档块单独做一个prompt（回答或摘要），然后将所有结果再合并（reduce）生成最终答案。 _适用场景_*: 文档数量多，单个文档块处理无上下文依赖，或需要独立摘要。 _优势_*: 可以并发处理每个文档，避免超出上下文窗口。 _局限_*: 每个文档之间缺少直接上下文，合并阶段可能丢失细微信息，多次调用LLM成本较高。
refine:

Code
_原理_*: 在第一个文档块上做prompt得到初始结果，然后将该结果与下一个文档合并作为新的prompt，迭代地“细化”最终答案。 _适用场景_*: 需要逐步积累和细化答案，或文档之间有较强顺序依赖。 _优势_*: 能部分保留上下文，token使用可在一定范围。 _局限_*: 顺序处理，效率相对较低，可能受早期文档偏见影响。
map_rerank:

Code
_原理_*: 对每个文档块做prompt，并要求LLM为每个结果打分，然后根据分数返回最好的文档中的结果。 _适用场景_*: 需要从多个文档中选出“最佳”答案，对答案质量有较高要求。 _优势_*: 能有效识别最相关的文档，并提供置信度。 _局限_*: 会大量调用LLM，每个文档独立处理，成本最高。

LLM处理无限上下文与RAG的意义

即使LLM未来可以处理“无限上下文”，RAG仍然具有重要意义：

效率与成本: LLM处理超长上下文时计算资源消耗大，响应时间增加。RAG通过检索相关片段，减少输入长度，降低成本。
知识更新: LLM的知识截止于训练数据，无法实时更新。RAG可以连接外部知识库，增强时效性。
可解释性: RAG的检索过程透明，用户可查看来源，增强信任。LLM的生成过程则较难追溯。
定制化: RAG可针对特定领域定制检索系统，提供更精准的结果，而LLM的通用性可能无法满足特定需求。
数据隐私: RAG允许在本地或私有数据源上检索，避免敏感数据上传云端，适合隐私要求高的场景。

Query改写

RAG的核心在于“检索-生成”，如果“检索”走偏，后续“生成”质量也会降低。用户提出的问题往往是口语化、承接上下文、模糊的，而知识库文本通常是陈述性、客观的。因此，Query改写作为“翻译官”，将用户的口语化查询转换成书面化、精确的检索语句。

通过精心设计的Prompt来引导LLM完成Query改写任务。

1. 上下文依赖型 Query 改写

用于将依赖前序对话的模糊查询改写为独立的、完整的查询。

Python

instruction = """
你是一个智能的查询优化助手。请分析用户的当前问题以及前序对话历史，判断当前问题是否依赖于上下文。
如果依赖，请将当前问题改写成一个独立的、包含所有必要上下文信息的完整问题。
如果不依赖，直接返回原问题。
"""        
prompt = f"""
### 指令###
{instruction}
### 对话历史###
{conversation_history}
### 当前问题###
{current_query}
### 改写后的问题###
"""
# 示例：
# 对话历史: 用户: "我想了解一下上海迪士尼乐园的最新项目。" AI: "上海迪士尼乐园最新推出了'疯狂动物城'主题园区..."
# 当前查询: 还有其他设施吗？
# 改写结果: 除了疯狂动物城警察局、朱迪警官训练营和尼克狐的冰淇淋店之外，'疯狂动物城'园区还有其他设施吗？

2. 对比型 Query 改写

用于识别问题中需要进行比较的对象，改写成更明确的对比性查询。

Python

instruction = """
你是一个查询分析专家。请分析用户的输入和相关的对话上下文，识别出问题中需要进行比较的多个对象。
然后，将原始问题改写成一个更明确、更适合在知识库中检索的对比性查询。
"""        
prompt = f"""
### 指令###
{instruction}
### 对话历史/上下文信息###
{context_info}
### 原始问题###
{query}
### 改写后的查询###
"""
# 示例：
# 对话历史: 用户: "我想了解一下上海迪士尼乐园的最新项目。" AI: "上海迪士尼乐园最新推出了疯狂动物城主题园区，还有蜘蛛侠主题园区"
# 当前查询: 哪个游玩的时间比较长，比较有趣
# 改写结果: 哪个游玩时间更长、更有趣：上海迪士尼乐园的疯狂动物城主题园区和蜘蛛侠主题园区？

3. 模糊指代型 Query 改写

用于消除“都”、“它”、“这个”等模糊指代词，替换为明确的对象名称。

Python

instruction = """
你是一个消除语言歧义的专家。请分析用户的当前问题和对话历史，找出问题中"都"、"它"、"这个" 等模糊指代词具体指
向的对象。
然后，将这些指代词替换为明确的对象名称，生成一个清晰、无歧义的新问题。
"""        
prompt = f"""
### 指令###
{instruction}
### 对话历史###
{conversation_history}
### 当前问题###
{current_query}
### 改写后的问题###
"""
# 示例：
# 对话历史: 用户: "我想了解一下上海迪士尼乐园和香港迪士尼乐园的烟花表演。" AI: "好的，上海迪士尼乐园和香港迪士尼乐园都有精彩的烟花表演。"
# 当前查询: 都什么时候开始？
# 改写结果: 上海迪士尼乐园和香港迪士尼乐园的烟花表演都什么时候开始？

4. 多意图型 Query 改写 - 分解查询

用于将用户的复杂问题分解成多个独立的、可以单独回答的简单问题。

Python

instruction = """
你是一个任务分解机器人。请将用户的复杂问题分解成多个独立的、可以单独回答的简单问题。以JSON数组格式输出。
"""        
prompt = f"""
### 指令###
{instruction}
### 原始问题###
{query}
### 分解后的问题列表###
请以JSON数组格式输出，例如：["问题1", "问题2", "问题3"]
"""
# 示例：
# 原始查询: 门票多少钱？需要提前预约吗？停车费怎么收？
# 分解结果: ['门票多少钱？', '需要提前预约吗？', '停车费怎么收？']

5. 反问型 Query 改写

用于识别用户反问或带有情绪的陈述，改写成中立、客观、可以直接用于检索的问题。

Python

instruction = """
你是一个沟通理解大师。请分析用户的反问或带有情绪的陈述，识别其背后真实的意图和问题。
然后，将这个反问改写成一个中立、客观、可以直接用于知识库检索的问题。
"""        
prompt = f"""
### 指令###
{instruction}
### 对话历史###
{conversation_history}
### 当前问题###
{current_query}
### 改写后的问题###
"""
# 示例：
# 对话历史: 用户: "你好，我想预订下周六上海迪士尼乐园的门票。" AI: "查询到下周六的门票已经售罄。"
# 当前查询: 这不会也要提前一个月预订吧？
# 改写结果: 迪士尼乐园门票是否需要提前一个月预订？

6. 自动识别 Query 类型并进行改写

通过一个Prompt让LLM自动识别查询类型并进行相应改写。

Python

instruction = """
你是一个智能的查询分析专家。请分析用户的查询，识别其属于以下哪种类型：
1. 上下文依赖型- 包含"还有"、"其他"等需要上下文理解的词汇
2. 对比型- 包含"哪个"、"比较"、"更"、"哪个更好"、"哪个更"等比较词汇
3. 模糊指代型- 包含"它"、"他们"、"都"、"这个"等指代词
4. 多意图型- 包含多个独立问题，用"、"或"？"分隔
5. 反问型- 包含"不会"、"难道"等反问语气
说明：如果同时存在多意图型、模糊指代型，优先级为多意图型>模糊指代型
请返回JSON格式：
{
"query_type": "查询类型",
"rewritten_query": "改写后的查询",
"confidence": "置信度(0-1)"
}
"""        
prompt = f"""
### 指令###
{instruction}
### 对话历史###
{conversation_history}
### 上下文信息###
{context_info}
### 原始查询###
{query}
### 分析结果###
"""
# 示例输出见原始材料，展示了不同查询类型及其改写结果和置信度。

Query + 联网搜索

当RAG系统需要处理时效性强或实时变化的信息时，需要结合联网搜索。

1. 识别查询是否需要联网搜索

类型 | 关键词特征 | 示例查询 | 原因说明

时效性 | 最新、今天、现在、实时、当前 | 上海迪士尼乐园今天开放吗？ | 需获取当前时间的最新信息

价格信息 | 多少钱、价格、费用、票价 | 下周六的门票多少钱？ | 价格信息经常变动

营业信息 | 营业时间、开放时间、闭园、是否开放 | 迪士尼乐园现在开门吗？ | 营业状态可能调整

活动信息 | 活动、表演、演出、节日、庆典 | 最近有什么特别活动？ | 活动信息具有时效性

天气信息 | 天气、下雨、温度 | 明天去迪士尼天气怎么样？ | 天气信息需要实时获取

交通信息 | 怎么去、交通、地铁、公交 | 从浦东机场怎么去