LangChain实现RAG

一、RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种结合信息检索与生成式 AI的技术框架，核心目标是让 AI 模型在生成回答时，能够基于外部知识库的准确信息，而非仅依赖训练时学到的内容。这一技术有效解决了LLM的“幻觉生成”（编造不存在的信息）、“知识过时”（无法获取训练数据之后的新信息）等问题，广泛应用于智能问答、文档助手、客服系统等场景。

RAG 的核心逻辑是“检索 + 生成”的过程：当用户提出问题时，先从外部知识库中搜索出与问题相关的信息，再将这些信息作为参考资料交给生成模型，让模型基于参考资料生成回答。

具体分为三个关键环节：

1.1 知识准备

将外部知识处理成适合检索的格式，存储到向量数据库中。首先通过工具读取各类格式的文档，然后将长文档拆分成短文本块，避免信息冗余，再用嵌入模型将文本块转换成向量，捕捉语义信息，最后将向量和对应文本块存入向量数据库，方便后续快速检索。

1.2 检索

当用户提问时，从向量数据库中找到与问题 “语义最相关” 的文本块。首先将用户问题转换成向量，在向量数据库中通过向量相似度计算找到与问题向量最接近的top k个文本块，这些文本块就是回答问题的参考资料。

1.3 增强与生成

增强就是将用户查询问题和检索到的参考资料填充到一个提示模板中，生成就是让大模型结合参考资料和问题，生成最终回答，模型会明确基于检索到的资料生成内容，避免编造信息，如果资料中没有答案，也能提示无法找到相关信息。

二、为什么需要RAG

大型语言模型普遍面临数据时效性的局限，即使像 deepseek 这类的模型，对于近期发生的事件也无法知晓，这是因为它们的认知完全基于训练数据所构建的世界，而训练数据集不可能覆盖特定领域的专属信息，也无法纳入最新产生的数据，这种特性使得需要依赖最新消息或特定数据的应用场景都遇到了问题。

比如下面这个最新热议的事件，LLM是无法知晓的，因为其认知是基于之前的数据训练的，该问题可以通过RAG解决，RAG可以给LLM 装一个实时信息接口，通过动态更新外部知识库和精准检索，让模型在回答时临时调用最新数据。

from langchain.prompts import ChatPromptTemplate
import os
from langchain_core.callbacks import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI
API_SECRET_KEY = "API_KEY"
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
os.environ["OPENAI_API_KEY"] = API_SECRET_KEY
os.environ["OPENAI_API_BASE"] = BASE_URL
chat = ChatOpenAI(model='deepseek-v3',api_key=API_SECRET_KEY,  base_url=BASE_URL,temperature=0,streaming=True,callbacks=[StreamingStdOutCallbackHandler()]
)
template = "根据以下内容回答问题，文本：{text}"
# 将template字符串转换为langchain模板
prompt_template = ChatPromptTemplate.from_template(template)
customer_text = '介绍一下2025年关于汪苏泷与张碧晨的《年轮》版权问题'customer_messages = prompt_template.format_messages(text=customer_text)
print(customer_messages[0])
customer_response = chat.invoke(customer_messages, temperature=0.0)

三、通过 LangChain 实现RAG

以下是一份蜡笔小新的pdf文档，现需要在pdf文档中搜索野原新之助的性格特征。

3.1 提取文档

首先通过PyPDFLoader加载需要检索的文档。

loaders = [PyPDFLoader('蜡笔小新.pdf')]
docs = []
for loader in loaders:try:pages = loader.load()# 输出第一页的前10个文字print(pages[0].page_content[:10])# 打印第一个元素的元数据print(pages[0].metadata)docs.extend(pages)except Exception as e:print("PDF加载失败：", e)exit()

3.2 分割文本块

为了更高效地处理文档，需要对其进行分块处理，加载后的原始文档往往很长，这时就需要将其分割成一系列较小的文本块，然后再进行存储，在后续的检索中，就能直接返回与查询相关的具体文档块，而不需要把整个文档都发送给大模型进行处理。

# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(chunk_size=520,  # 每个文本块的最大字符长度chunk_overlap=10,  # 相邻文本块之间的重叠字符数separators=["\n\n", "\n", " ", ""]
)
blocks = text_splitter.split_documents(docs)

3.3 向量化存储

将文本转换为向量形式后存储在本地环境中，之后通过向量计算，系统可以快速找到与用户查询相关的信息，并将这些信息返回给大模型。

texts = [block.page_content for block in filtered_blocks]
# 创建向量数据库
embedding_model = DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key=API_SECRET_KEY
)
# 使用FAISS向量数据库
vector_db = FAISS.from_texts(texts=texts,embedding=embedding_model
)

3.4 检索

定义检索器retriever，其功能是根据用户查询内容与向量化文本块之间的相似度，选出相关的上下文信息。

# 创建检索器，限制返回的文档数量
retriever = vector_db.as_retriever(search_kwargs={"k": 2})

3.5 增强

创建一个提示词模板，当用户基于这个模板提出问题时，大模型就会通过检索器自动查找与问题相关的上下文信息，让生成的回答更加准确。

template = '''请严格按照提供的上下文回答问题，如果上下文未包含答案信息，请回复"相关信息未提供"：上下文：{context}问题：{question}'''
qa_prompt = PromptTemplate.from_template(template)
print(qa_prompt)

3.6 生成

最后为 RAG 系统创建一个完整的链式结构，该结构会将大语言模型、检索器、提示词模板依次连接起来，之后就可以直接调用 RAG 流程完成各种问答或内容生成任务。

qa_chain = RetrievalQA.from_chain_type(llm,retriever=retriever,return_source_documents=True,chain_type="stuff",chain_type_kwargs={'prompt': qa_prompt}
)
try:quest = '请详细介绍一下野原新之助的性格特点'print('请详细介绍一下野原新之助的性格特点')print('==== 模型响应 ====')response = qa_chain.invoke({"query": quest})
except Exception as e:print(f"处理请求时出错: {e}")

完整实现代码

from langchain_community.document_loaders.pdf import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain_core.callbacks import BaseCallbackHandler, CallbackManagerclass StreamingResultCallbackHandler(BaseCallbackHandler):def __init__(self):super().__init__()self.result_buffer = ""def on_llm_new_token(self, token: str, **kwargs) -> None:print(token, end="", flush=True)self.result_buffer += tokendef get_result(self) -> str:return self.result_bufferAPI_SECRET_KEY = "API_KEY"
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"# 1. 加载PDF
loaders = [PyPDFLoader('蜡笔小新.pdf')]
docs = []
for loader in loaders:try:pages = loader.load()# 输出第一页的前10个文字print(pages[0].page_content[:10])# 打印第一个元素的元数据print(pages[0].metadata)docs.extend(pages)except Exception as e:print("PDF加载失败：", e)exit()# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(chunk_size=520,  # 每个文本块的最大字符长度chunk_overlap=10,  # 相邻文本块之间的重叠字符数。separators=["\n\n", "\n", " ", ""]
)
blocks = text_splitter.split_documents(docs)# 3. 过滤空文本块
filtered_blocks = []
for block in blocks:content = str(block.page_content).strip()if content:filtered_blocks.append(block)
if not filtered_blocks:print("错误：所有文本块均为空")exit()# 4. 使用原始文本
texts = [block.page_content for block in filtered_blocks]
# 5. 创建向量数据库
embedding_model = DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key=API_SECRET_KEY
)# 使用FAISS向量数据库
vector_db = FAISS.from_texts(texts=texts,embedding=embedding_model
)# 6. 创建流式回调处理器
streaming_callback = StreamingResultCallbackHandler()
callback_manager = CallbackManager([streaming_callback])
llm = ChatOpenAI(model_name='qwen-turbo',temperature=0.0,openai_api_key=API_SECRET_KEY,openai_api_base=BASE_URL,max_retries=2,streaming=True, callback_manager=callback_manager  
)# 创建检索器，限制返回的文档数量
retriever = vector_db.as_retriever(search_kwargs={"k": 2})template = '''请严格按照提供的上下文回答问题，如果上下文未包含答案信息，请回复"相关信息未提供"：上下文：{context}问题：{question}'''
qa_prompt = PromptTemplate.from_template(template)
print(qa_prompt)qa_chain = RetrievalQA.from_chain_type(llm,retriever=retriever,return_source_documents=True,chain_type="stuff",chain_type_kwargs={'prompt': qa_prompt}
)try:quest = '请详细介绍一下野原新之助的性格特点'print('请详细介绍一下野原新之助的性格特点')print('==== 模型响应 ====')response = qa_chain.invoke({"query": quest})
except Exception as e:print(f"处理请求时出错: {e}")