版本: 3.1.0 大模型RAG与LangChain

LangChain项目实战

RAG项目实战

课程目标

掌握RAG 工作流程
掌握能够基于 langchain 开发出基础的 RAG 问答系统

RAG 介绍

概述

2023年以来，随着ChatGPT的火爆，使得LLM成为研究和应用的热点，但是市面上大部分LLM都存在一个共同的问题：模型都是基于过去的经验数据进行训练完成，无法获取最新的知识，以及各企业私有的知识。因此很多企业为了处理私有的知识，主要借助以下两种手段来实现

利用企业私有知识，基于开源大模型进行微调
基于LangChain集成向量数据库以及LLM搭建本地知识库的问答（RAG）

RAG（Retrieval-Augmented Generation）检索增强生成，在不改变模型权重的情况下，提升大模型生成能力。用户query 会从知识库中检索出相关文档，大模型依据文档生成用户的 query 回答。这样可以实现低成本提升大模型的回复能力。RAG 核心点在于知识库的构建和检索策略。

RAG 的工作流程图解

RAG 标准流程由索引（Indexing）、检索（Retriever）和生成（Generation）三个核心阶段组成。

索引阶段
- 加载文件
- 内容提取
- 文本分割，形成chunk
- 文本向量化
- 存向量数据库
检索阶段
- query向量化
- 在文本向量中匹配出与问句向量相似的top_k个
生成阶段
- 匹配出的文本作为上下文和问题一起添加到prompt中
- 提交给LLM生成答案：

环境准备

安装依赖

首先，确保你的机器安装了conda，创建虚拟环境，pycharm 选择 conda 环境
也可以使用python创建虚拟环境

# 终端查看python的版本
conda create -n rag python=3.12
# 推荐在pycharm中选择uv环境或者Virtualenv创建虚拟环境

紧接着安装项目的依赖

# 安装全部依赖
pip install -r requirements.txt

Qwen在线大模型调用准备

物流行业信息咨询智能问答系统（RAG）

项目需求

本次项目以"某物流行业"为例，基于物流信息构建RAG系统，测试问答效果。注意：除物流场景外，使用者可以自由切换其他行业类型知识，实现本地知识库问答的效果。

项目思路

离线部分
- 本地知识文件加载，读取
- 文本切分
- 向量化
- 存向量库
在线部分
- query 向量化
- 在文本向量中匹配出与问句向量相似的top_k个
- 匹配出的文本作为上下文和问题一起添加到prompt中
- 提交给LLM生成答案

项目代码

项目结构

├── chroma # 向量库
├── data # 数据集
│   └── 物流信息.pdf
├── db.py # 根据语料创建向量库
├── main.py # 运行主函数
├── model.py # 模型文件 包括大模型的和embedding模型
└── web_qa.py # 对话页面

db.py 创建离线知识库

PyMuPDFLoader加载pdf文件
对文本进行分割
向量化，得到向量库的对象
创建检索器并返回

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Author  : Vincent
@Time    : 2026/1/4 18:20
@File    : db.py
@Function :
"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyMuPDFLoader
# from langchain_chroma import Chroma
from langchain_community.vectorstores import Chroma


def create_db(file_path, embeddings, persist_directory='./chroma'):
    # 第一步：加载文档
    loader = PyMuPDFLoader(file_path)
    docs = loader.load()
    # 第二部：切分文本
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=20)
    split_texts = text_splitter.split_documents(docs)
    print(split_texts)
    # 第三步：将 document通过embeddings对象计算得到向量信息并永久存入Chroma向量数据库，用于后续匹配查询
    vectordb = Chroma.from_documents(
        documents=split_texts,
        embedding=embeddings,
        persist_directory=persist_directory
    )
    retriever = vectordb.as_retriever(search_kwargs={'k': 2})
    return retriever


if __name__ == '__main__':
    from model import embeddings

    retriever = create_db('./data/物流信息.pdf', embeddings)
    print(retriever.invoke("仓库"))

model.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Author  : Vincent
@Time    : 2026/1/4 18:20
@File    : model.py
@Function :
"""
import os

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_openai import ChatOpenAI

# 初始化 llm 对象
llm = ChatTongyi(
    api_key=os.getenv("API_KEY"),
    model="qwen-flash"
)

# llm = ChatOpenAI(
#     model="qwen-max",
#     api_key=os.getenv("API_KEY"),
#     base_url=os.getenv("BASE_URL"),
# )

# 初始化embeddings 对象
embeddings = DashScopeEmbeddings(
    model="text-embedding-v1",  # text-embedding-v3
    dashscope_api_key=os.getenv("API_KEY")
)

main.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Author  : Vincent
@Time    : 2026/1/4 18:20
@File    : main.py
@Function :
"""
from langchain_core.prompts import PromptTemplate
from db import create_db
from model import llm, embeddings

# 加载向量库
retriever = create_db('./data/物流信息.pdf', embeddings, persist_directory='./chroma')


def get_related_content(related_docs):
    related_content = []
    for doc in related_docs:
        related_content.append(doc.page_content.replace("\n\n", "\n"))
    return "\n".join(related_content)


def define_prompt(query):
    # 1. 检索出query的相关文档
    docs = retriever.invoke(query)
    print("检索出上下文：", f"[{len(docs)}]", docs)
    # 2. 类型转换，把多篇上下文转为字符串
    related_content = get_related_content(docs)

    PROMPT_TEMPLATE = """
        基于以下已知信息，简洁和专业的来回答用户的问题。不允许在答案中添加编造内容。
        已知内容:
        {context}
        问题:
        {question}
        """
    # 3. 提示词模板组装
    prompt = PromptTemplate(
        input_variables=["context", "question"],
        template=PROMPT_TEMPLATE, )
    my_pmt = prompt.format(context=related_content,
                           question=query)
    print(my_pmt)
    return my_pmt


def qa(question):
    my_pmt = define_prompt(question)  # 返回提示词
    result = llm.invoke(my_pmt)
    return result.content


if __name__ == '__main__':
    question = '我们用的什么快递公司？'
    result = qa(question)
    print(result)

运行结果：

web_qa.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Author  : Vincent
@Time    : 2026/1/4 18:21
@File    : web_qa.py
@Function :
"""
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from model import llm, embeddings
from db import create_db
import streamlit as st

# 设置标题
st.set_page_config(page_title="物流行业信息咨询系统")
st.title("物流行业信息咨询RAG系统")

# 初始化全局变量
retriever = None


def get_retriever():
    """
    获取或创建检索器（单例模式）
    """
    global retriever
    if retriever is None:
        retriever = create_db('data/物流信息.pdf', embeddings, persist_directory='./chroma')
    return retriever


def format_docs(docs):
    """
    格式化检索到的文档
    """
    return "\n\n".join(doc.page_content for doc in docs)


def create_chain(retriever):
    """
    创建基于 RAG 的问答链（使用 LangChain Expression Language）
    """
    # 定义提示词模板
    template = """基于以下已知信息回答用户的问题。如果你不知道答案，就说你不知道，不要编造信息。

已知信息:
{context}

对话历史:
{chat_history}

问题：{question}

回答："""

    prompt = ChatPromptTemplate.from_template(template)

    # 构建 RAG 链
    rag_chain = (
            RunnablePassthrough.assign(
                context=lambda x: format_docs(retriever.invoke(x["question"])),
            )
            | prompt
            | llm
            | StrOutputParser()
    )

    return rag_chain


# 主逻辑
def main():
    """
    Streamlit 主页面的交互逻辑。
    """
    # print(f'st.session_state-->{st.session_state}')
    # 初始化会话状态
    if "messages" not in st.session_state:
        st.session_state.messages = []  # 用于保存聊天记录
    # print(f'st.session_state-->{st.session_state}')
    # 展示历史聊天记录
    for message in st.session_state.messages:
        # print(f'message["role"]-->{message["role"]}')
        with st.chat_message(message["role"]):
            st.markdown(message["content"])  # 显示消息内容

    # 接受用户输入
    if prompt := st.chat_input("请输入你的问题:"):
        # 保存用户消息到会话状态
        print(f'prompt--》{prompt}')
        st.session_state.messages.append({"role": "user", "content": prompt})

        # 显示用户输入
        with st.chat_message("user"):
            st.markdown(prompt)

        # 调用模型获取回答
        with st.chat_message("assistant"):
            # 占位符用于显示逐字生成的回答
            message_placeholder = st.empty()
            full_response = ""

            # 获取检索器并创建链
            retriever = get_retriever()
            chain = create_chain(retriever)

            # 准备对话历史
            chat_history_str = ""
            if len(st.session_state.messages) > 1:
                # 构建对话历史字符串
                history_messages = st.session_state.messages[:-1]  # 排除当前最新消息
                for i in range(0, min(len(history_messages), 6), 2):  # 最近 3 轮对话
                    if i + 1 < len(history_messages):
                        user_msg = history_messages[i]
                        ai_msg = history_messages[i + 1]
                        if user_msg["role"] == "user" and ai_msg["role"] == "assistant":
                            chat_history_str += f"用户：{user_msg['content']}\nAI: {ai_msg['content']}\n"

            # 调用链获取答案
            result = chain.invoke({
                "question": prompt,
                "chat_history": chat_history_str
            })
            print(f'result--->{result}')

            assistant_response = result
            message_placeholder.markdown(assistant_response)

            # 保存回答到会话状态
            st.session_state.messages.append({"role": "assistant", "content": assistant_response})


# 运行主逻辑
if __name__ == "__main__":
    main()

streamlit run web_qa.py

小结

本文主要讲述了RAG的流程，并且实现了基础的RAG系统。

需要思考优化的方向：

如果用户的query很随意，或者query的描述不清楚，应该如何处理？
如果检索出的内容过长，或者检索的内容相关性不高。相当于引入了大量噪音，可能会降低大模型的回答能力。如何优化检索出的内容相关性？
RAG主要是解决大模型幻觉问题，知识缺乏问题，有什么办法能够从根本上缓解？