【深度学习总结】使用PDF构建RAG：结合Langchain和通义千问

使用平台：趋动云，注册送算力

前言

在大型语言模型（LLMs）应用领域，我们面临着大量挑战，从特定领域知识的匮乏到信息准确性的窘境，以及可能生成虚假内容。检索增强生成（RAG）通过引入外部知识库等补充信息源，成为解决这些难题的有效策略。事实证明，在需要持续更新或特定领域应用的知识密集型场景中，RAG 尤其有效。与其他方法相比，RAG 的一个显著优势在于无需为特定任务重新培训 LLM。最近，RAG 因其在会话助手等应用中的成功应用而备受瞩目。

RAG构成了一种将输入与相关支持文档的语料库相结合的技术。这些文档被合并到输入提示中，并联合输入到文本生成器中，从而产生最终的输出。这种RAG的这种机制在需要适应不断变化的信息环境的场景中找到了特殊的效用，因为llm所依赖的参数化知识本质上是静态的。通过RAG，语言模型可以直接访问最新的信息，而不需要再训练，促进了生成可靠的、基于检索的输出。本质上，RAG通过检索证据提高了LLM响应的准确性、可控性和相关性，从而证明了在快速发展的环境中解决问题，并有效地缓解了错误信息生成和性能退化的问题。

RAG的一个典型应用程序如下图所示。

在这里，一个用户向ChatGPT提出了一个关于最近一个被广泛讨论的新闻的问题。鉴于ChatGPT依赖于训练前的数据，它最初缺乏提供最新发展的能力。RAG通过从外部数据库中获取和整合知识来弥补这一信息差距。在这种情况下，它会收集与用户查询相关的相关新闻文章。这些文章，结合最初的问题，形成了一个全面的提示，使llm能够生成一个知情的答案。典型的RAG遵循了一个传统的过程，包括索引、检索和生成，这也被描述为一个“检索“。

索引：首先以不同的格式清理和提取原始数据，如PDF、HTML、Word和标记，然后将其转换为统一的纯文本格式。为了适应语言模型的上下文限制，文本被分割成更小的、可理解的块。然后使用嵌入模型将块编码到向量表示中，并存储在向量数据库中。这一步对于在后续的检索阶段实现有效的相似性搜索至关重要。

检索：在收到用户查询后，RAG系统使用在索引阶段使用的相同的编码模型来将查询转换为向量表示。然后计算查询向量和索引语料库中的块向量之间的相似性得分。系统对与查询相似性最大的前k块进行优先排序和检索。这些数据块随后在提示符中被用作扩展的上下文。

生成：所提出的查询和所选择的文档被合成成一个连贯的提示，一个大型语言模型负责制定一个响应。模型的回答方法可能根据特定任务的标准而有所不同，允许它利用其固有的参数知识或限制其对所提供文档中包含的信息的响应。在正在进行的对话的情况下，任何现有的对话历史都可以集成到提示符中，使模型能够有效地参与多回合的对话交互

教程

PDF文件

使用的是民用飞机维修的PDF文件，可以私信我获取，你也可以用自己的。

准备

• 通义千问的API-Key
• 运行环境：将如下内容存入requirements.txt，然后运行：pip install -r requirements.txt

python-dotenv==1.0.1 # For reading environment variables stored in .env file
langchain==0.2.2
langchain-community==0.2.3
dashscope
unstructured==0.14.4 # Document loading
# onnxruntime==1.17.1 # chromadb dependency: on Mac use `conda install onnxruntime -c conda-forge`
# For Windows users, install Microsoft Visual C++ Build Tools first
# install onnxruntime before installing `chromadb`
chromadb==0.5.0 # Vector storage
tiktoken==0.7.0  # For embeddings 

构建RAG

先导入包：

from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings.huggingface import HuggingFaceEmbeddings
from dotenv import load_dotenv
import os
import shutil
import dashscope
from dashscope import Generation
from langchain.prompts import ChatPromptTemplate
from http import HTTPStatus

然后将PDF数据转换为向量存储起来：

# Load environment variables. Assumes that project contains .env file with API keys
load_dotenv()
# 设置镜像，便于下载后面的HuggingFaceEmbeddings
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
# huggingface下载地址
os.environ["HF_HOME"] = "/gemini/code/huggingface"
# huggingface下载地址
os.environ["TRANSFORMERS_CACHE"] = "/gemini/code/huggingface"

os.environ["SENTENCE_TRANSFORMERS_HOME"] = "/gemini/code/huggingface/bce-embedding-base_v1"

# 向量存放位置
CHROMA_PATH = "chroma"
# 存放数据
DATA_PATH = "data/books"
dashscope.api_key = "你的通义千问api key"
print(os.getenv('DASHSCOPE_API_KEY'))
def prepare_db():
    # 处理读个pdf
    pdf_paths = ["data/books/M1.pdf",
                 "data/books/M2-航空器维修R1.pdf",
                 "data/books/M3-飞机结构和系统R1.pdf",
                 "data/books/M4-直升机结构和系统.pdf",
                 "data/books/M5-航空涡轮发动机R1.pdf",
                 "data/books/M6-活塞发动机及其维修.pdf",
                 "data/books/M7-航空器维修基本技能.pdf",
                 "data/books/M8-航空器维修实践R1.pdf"]
    documents = []
    count = 0
    for pdf_path in pdf_paths:

        loader = PyPDFLoader(pdf_path)
        doc = loader.load()
        documents.extend(doc)
        count += 1
        print(f"处理第{count}本")

    print(len(documents))

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=300,
        chunk_overlap=100,
        length_function=len,
        add_start_index=True,
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Split {len(documents)} documents into {len(chunks)} chunks.")

    document = chunks[10]
    print(document.page_content)
    print(document.metadata)

    if os.path.exists(CHROMA_PATH):
        shutil.rmtree(CHROMA_PATH)
    # 将文本保存为向量存储
    model_name = "maidalun1020/bce-embedding-base_v1"
    model_kwargs = {'device': 'cuda'}
    encode_kwargs = {'normalize_embeddings': False}
    embeddings = HuggingFaceEmbeddings(
        model_name=model_name,
        model_kwargs=model_kwargs,
        encode_kwargs=encode_kwargs,
        cache_folder="/gemini/code/huggingface/",
    )
    # Create a new DB from the documents.
    db = Chroma.from_documents(
        chunks, embeddings, persist_directory=CHROMA_PATH
    )
    db.persist()
    print(f"Saved {len(chunks)} chunks to {CHROMA_PATH}.")

然后调用通义千问的API，用户输入问题，然后根据问题从向量库中查找相关的内容，跟问题结合起来，一起喂给通义千问：

def query():
    model_name = "maidalun1020/bce-embedding-base_v1"
    model_kwargs = {'device': 'cuda'}
    encode_kwargs = {'normalize_embeddings': False}
    embeddings = HuggingFaceEmbeddings(
        model_name=model_name,
        model_kwargs=model_kwargs,
        encode_kwargs=encode_kwargs,
        # 模型缓存路径
        cache_folder="/gemini/code/huggingface/",
    )
    # 改成你的保存的路径
    db = Chroma(persist_directory="./chroma/aae4fae7-3477-4094-8a7d-c5df8be2223a", embedding_function=embeddings)
	# 提示模板
    PROMPT_TEMPLATE = """
    仅根据下列文本回答问题：
    {context}
    """
    while True:
        query = input('请输入问题:')
        results = db.similarity_search_with_relevance_scores(query, k=5)
        if len(results) == 0:
            print(f"Unable to find matching results.")
            return
        # 拼接成输入给大模型的内容
        context_text = "\n\n---\n\n".join([doc.page_content for doc, _score in results])
        messages = [
            {'role': 'system', 'content': PROMPT_TEMPLATE.format(context=context_text)},
            {'role': 'user', 'content': f"请回答如下问题:{query}"}
        ]
        print(messages)
        responses = Generation.call(Generation.Models.qwen_max, 
                                    api_key=os.getenv('DASHSCOPE_API_KEY'),
                                    messages=messages, 
                                    result_format='message')
        # 如果你不确定responses的结果，可以打印处理
        # print(responses.output)
        sources = [doc.metadata.get("source", None) for doc, _score in results]
        if responses.status_code == HTTPStatus.OK:
            whole_message = responses.output["choices"][0]["message"]["content"]
        else:
            whole_message = "error"
            print('Failed request_id: %s, status_code: %s, code: %s, message:%s' %
                (responses.request_id, responses.status_code, responses.code,
                responses.message))
        formatted_response = f"Response: {whole_message}\nSources: {sources}"
        print(formatted_response)

参考链接

langchain-rag-tutorial