革命性突破：langchain与deepseek-r1结合rag技术重新定义ai应用

Rifx.Online
Large Language Models , Reinforcement Learning , AI Applications
23 Feb, 2025

上周，我制作了一个关于深度搜索-V3的视频，这在全球AI社区引起了巨大轰动。

两天前，一款中国的深度搜索发布了基于推理的大规模语言模型“深度搜索-R1”，并作为开源发布！

据说它的表现与开放AI的最准确推理模型“o1”一样出色。更令人印象深刻的是，它是一款价格极具竞争力的模型，应用程序接口的价格不到开放AI o1的1/25。此外，它在高度灵活的MIT许可证下开源，任何人都可以下载和使用这个模型。

这次R1模型一发布，不仅驳斥了之前关于提炼开放AI o1的说法，官方还直接表示：“我们可以与o1的开源版本相提并论。”

值得一提的是，R1突破了以往的模型训练方法，完全不使用任何监督微调数据。它仅通过纯强化学习来训练模型。这表明R1已经学会了独立思考问题——这实际上更符合人类的思维方式。

在我想开发RAG聊天机器人的时候，我发现语言链在我使用它开发RAG聊天机器人之前进行了巨大的更新。不幸的是，它不像聊天生成预训练变换器那样记住对话内容，也无法像客服聊天机器人那样输入新数据或进行特定问题的训练。

所以，让我给你快速演示一个实时聊天机器人，向你展示我的意思。

我将上传一个包含图像的PDF，然后问聊天机器人一个问题：“总结一下这个PDF。”请随意提问。如果你看看聊天机器人是如何生成输出的，你会看到PDF文件将其内容存储在一个临时文件中，使用PDFPlumber加载器处理它，以提取复杂数据如表格，并清理临时文件。

它使用语义分块器将文档分割成语义块，并使用高效相似性搜索进行基于相似性的高效搜索。检索器为查询找到最相似的前三个块，并使用历史感知检索器增强代理将整个对话历史纳入检索过程的能力。

它还集成了上下文链用于外部输入数据，以及文档链用于对话记录，使用创建检索链形成完整的检索过程。通过验证文档是否已上传、检索上下文并使用RAG链生成响应来处理上下文感知的响应。

在这个视频结束时，你将了解深度搜索R1是如何训练的，以及如何使用深度搜索-R1与RAG。

在我们开始之前！

如果你喜欢这个主题并且想要支持我：

为我的文章点赞 50 次；这将对我有很大帮助。
关注我在 Medium 上，并订阅以免费获取我的最新文章。
加入这个大家庭 — 订阅我的 YouTube 频道。

深度搜索-R1学习方法概述

深度搜索-R1的特点是使用强化学习（RL）进行后训练。一般来说，大规模语言模型的开发包括以下步骤：

预训练：一个大型语料库创建一个“预测下一个单词”的模型。
监督微调 (SFT)：使用高质量的人类创建的指令-响应对对模型进行特定任务的微调。
带人类反馈的强化学习 (RLHF)：人类评估模型的输出，并使用评分作为奖励来更新模型。

深度搜索-R1据报道在大规模上进行了强化学习，特别是第3步。此外，他们探索了一种直接应用RL的方法，而不使用传统的SFT（称为深度搜索-R1-零），然后再添加SFT以完成深度搜索-R1。这种方法被认为促进了模型的内部思维过程（思维链）和自我验证。

通常，有效处理思维链需要一些SFT数据和人类标签。然而，深度搜索-R1-零声称通过RL单独实现了推理能力，而无需经过SFT。这种方法提供了以下优势：

降低大规模数据收集的成本
使未知任务和复杂情况能够自我修正

然而，未经过SFT的模型仍然存在一些问题，例如文本的可读性较差和意外的多语言内容。

让我们构建应用程序

现在让我们一步一步地探索，揭示如何创建 RAG 应用程序的答案。我们将安装支持该模型的库。为此，我们将进行 pip install requirements。

pip install -r requirements.txt

下一步是常规步骤，我们将导入相关库，其重要性在后续过程中将变得明显。

PDFPlumberLoader 是一个非常强大的工具，用于处理来自 PDF 文件的复杂结构化数据，例如表格。它可以提取文本、图像、表格、字段等。

SemanticChunker 根据语义相似性将文本分块，确保相关内容保持在同一块中。

导入必要的库：

import os
import streamlit as st
from langchain_community.document_loaders import PDFPlumberLoader
from langchain_experimental.text_splitter import SemanticChunker
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.retrieval import create_retrieval_chain
from langchain_core.messages import AIMessage, HumanMessage
import tempfile
from langchain_openai.chat_models.base import BaseChatOpenAI

我创建一个函数来处理上传的文件，使用临时文件来存储其内容。然后，我使用 PDFPlumberLoader 处理 PDF，以处理来自 PDF 文件的复杂结构化数据，确保在之后清理临时文件，以保持系统整洁。最后，我返回提取的数据以供进一步使用。

def process_uploaded_file(uploaded_file):
    with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:
        tmp_file.write(uploaded_file.getvalue())
        loader = PDFPlumberLoader(tmp_file.name)
        documents = loader.load()
    os.unlink(tmp_file.name)

    return documents

我创建一个函数来处理文档列表，并构建一个用于基于相似性的搜索的检索器。首先，我使用 SemanticChunker 和 OpenAIEmbeddings 将文档分割成更小的块，同时保持语义关系，确保相关内容保持在一起。

在确认分割过程后，我使用 FAISS 开发了一个向量存储，它将块组织起来，以便进行高效的相似性搜索。

最后，我返回一个检索器，可以找到给定查询的前三个最相似的块，使文档检索过程既准确又高效。

def get_vs_retriever_from_docs(doc_list):
    text_splitter = SemanticChunker(HuggingFaceEmbeddings())
    documents = text_splitter.split_documents(doc_list)
    st.write('文档分割完成')

    embedder = HuggingFaceEmbeddings()
    vector = FAISS.from_documents(documents, embedder)

    return vector.as_retriever(search_type="similarity", search_kwargs={"k": 3})

我创建 init_ui() 函数，以设置一个用户友好的界面，用于上传和分析 PDF 文档，使用 Streamlit。首先，我配置页面的标题和描述性标题。

然后，我初始化会话状态变量，以管理检索器、聊天记录和上传状态。接下来，我提供一个文件上传小部件，供用户上传 PDF 文件。

一旦文件上传，我处理它以提取其内容，使用 get_vs_retriever_from_docs 创建一个检索器，并将其存储在会话状态中以供进一步使用。最后，我通过成功消息通知用户文档处理成功。

def init_ui():
    st.set_page_config(page_title='文档上传器')
    st.markdown('#### :books:🧙高达烈: 您的文档摘要工具')
    st.markdown("<h8 style='text-align: right; color: green;'>*分享您想阅读的书籍的 PDF*</h8>", unsafe_allow_html=True)

    if "vector_store" not in st.session_state:
        st.session_state.vector_store = None
    if "chat_history" not in st.session_state:
        st.session_state.chat_history = []  
    if "doc_upload" not in st.session_state:
        st.session_state.doc_upload = False

    uploaded_file = st.file_uploader("上传 PDF", type=["pdf"])
    if uploaded_file:
        docs = process_uploaded_file(uploaded_file)
        if docs:
            retriever = get_vs_retriever_from_docs(docs)
            st.session_state.vector_store = retriever
            st.session_state.doc_upload = True
            st.success("文档处理成功")

我创建 get_related_context() 函数，以改善在对话系统中检索相关信息的能力。首先，我初始化 Ollama 模型（深度搜索-R1）以处理用户输入。

然后，我定义一个提示模板，将聊天记录和用户输入结合在一起，指示模型根据正在进行的对话生成搜索查询。

最后，我使用 create_history_aware_retriever 来增强代理在检索过程中整合整个对话历史的能力。

def get_related_context(vector_store):

llm = Ollama(model=“深度搜索-R1”)

llm = BaseChatOpenAI(
        model='深度搜索-推理器', 
        openai_api_key='sk-68f459660c7b4d179e074cbedce962c0', 
        openai_api_base='https://api.deepseek.com',
        max_tokens=1024
    )
prompt = ChatPromptTemplate.from_messages([
        ("system", "根据对话生成搜索查询。"),
        ("user", "{input}")
    ])
return create_history_aware_retriever(llm, vector_store, prompt)

在这个函数中，我设计了一个集成两个核心链的函数：

context_chain 负责查找外部输入数据
负责查找对话记录

最后，使用 retrieval_chain = create_retrieval_chain(context_chain, docs_chain) 创建一个可以检索对话和外部输入数据的链，完成一个可以实现自然对话的链。

对话记录需要自我创建一个列表存储，即 chat_history 部分。

def get_context_aware_prompt(context_chain):
    # llm_! = Ollama(model="深度搜索-R1")

llm = BaseChatOpenAI(
        model='深度搜索-推理器', 
        openai_api_key='sk-68f459660c7b4d179e074cbedce962c0', 
        openai_api_base='https://api.deepseek.com',
        max_tokens=1024
    )
prompt = ChatPromptTemplate.from_messages([
       ("system", "使用提供的上下文回答问题：\n\n{context}"),
       ("user", "{input}")
   ])
    # st.write(docs_chain)
return create_retrieval_chain(context_chain, docs_chain)

在这个函数中，我处理为给定查询生成上下文感知响应的过程。我开始检查必要的文档是否已上传。

如果没有上传文档，我返回一个错误。一旦文档被验证，我检索一个 上下文链 来访问上传文档中的相关信息。如果检索上下文失败，我返回一个错误。

然后，我创建一个 RAG链，将上下文与对话历史和用户的查询结合起来以生成响应。最后，我调用 RAG 链，处理响应并返回生成的答案。

def get_response(query: str) -> str:
    if not st.session_state.vector_store:
        return "错误：请先上传文档"
    try:
        context_chain = get_related_context(st.session_state.vector_store)
        # st.write(context_chain)
        if not context_chain:
            return "错误：处理上下文失败"

        rag_chain = get_context_aware_prompt(context_chain)
        # st.write(rag_chain)
        current_history = st.session_state.chat_history[-2:] if len(st.session_state.chat_history) > 1 else []
        return rag_chain.invoke({
            "chat_history": current_history,
            "input": query
        })["answer"]
    except Exception as e:
        return f"错误：{str(e)}"

在这个函数中，我设置并显示一个聊天界面供用户互动。首先，我循环遍历存储的 聊天记录，并在聊天中显示用户和AI的消息。

然后，我提供一个输入框，用户可以在其中输入新问题。如果用户提交查询，我调用 get_response 函数生成答案，然后用新消息和响应更新聊天记录。

def init_chat_interface():
    for message in st.session_state.chat_history:
        with st.chat_message("user" if isinstance(message, HumanMessage) else "assistant"):
            st.write(message.content)

if prompt := st.chat_input("问一个问题...", disabled=not st.session_state.doc_upload):
        st.session_state.chat_history.append(HumanMessage(content=prompt))
        response = get_response(prompt)
        st.session_state.chat_history.append(AIMessage(content=response))

if __name__ == "__main__":
    init_ui()
    if st.session_state.vector_store:
        init_chat_interface()