Day8：RAG项目实战（3）（完结）

惬鹤

652人浏览 · 2026-06-04 19:12:34

惬鹤 · 2026-06-04 19:12:34 发布

前言

嗨嗨嗨，这里是惬鹤频道！
今天打算早一点把剩下的代码一起发出来，毕竟除了那几个还没和大家见面的文件，还有几个老面孔需要重新和大家见面。
这次的代码文件全部放出后，我会在文末附上GitHub的链接，大家可以去网站下载，按照README的指示就可以在本地运行了！
感谢大家的支持，下面先把新代码文件和大家分享。

代码文件：file_history_store.py

# 导入依赖
import os,json
from langchain_core.chat_history import BaseChatMessageHistory
from typing import Sequence
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict

def get_history(session_id):
    # 调用刚刚创建的类，输入会话序号和历史消息的路径
    return FileChatMessageHistory(session_id, "./chat_history")

# 定义最重要的类：FileChatMessageHistory，其中要定义两个方法（添加历史消息，清空历史消息），一个属性（转换消息）
class FileChatMessageHistory(BaseChatMessageHistory):
    # 初始化方法，后续的方法可以直接调用其中的变量
    def __init__(self, session_id, storage_path):
        # 会话ID
        self.session_id = session_id
        # 不同会话ID的存储文件
        self.storage_path = storage_path
        # 完整的路径（os.path.join 用于跨平台拼接文件/目录路径）
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件存在
        # os.path.dirname(self.file_path)用于获取一个路径的目录部分。
        # os.makedirs(path, exist_ok=True)的作用：创建目录树（多级目录）。
        # exist_ok=True 确保即使目录已存在也不会报错，程序继续运行。
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    # 定义方法：添加消息
    # messages: Sequence[BaseMessage]是一个类型注解，表示messages应该是这样的类型。
    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # 调用类自己的属性messages：从历史中读取一条历史消息（BaseMessage），并转换成列表list
        all_messages = list(self.messages)
        # 新的和旧的融合，all_messages变成一个新的消息（这里的messages（形参）和用self读取的messages（属性）不是一个东西）
        all_messages.extend(messages)
        # 用循环把all_messages洗成字典类型，赋给new_messages
        new_messages = [message_to_dict(msg) for msg in all_messages]
        # 写入文件，使用w模式，表示覆写
        with open(self.file_path, 'w',encoding="UTF-8") as f:
            # 把已经转为字典格式的new_messages写进文件中（注意不是dumps，dumps用来返回JSON字符串而不是写进去）
            json.dump(new_messages, f)

    # 使用property修饰，定义成成员属性
    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="UTF-8") as f:
                messages_data = json.load(f)    # 返回值就是：list[字典]
                # 把字典列表转成消息
                return messages_from_dict(messages_data)
        # 要是找不到文件，返回空
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="UTF-8") as f:
            # 要清除记录很简单，往文件夹内覆写一个空即可。
            json.dump([], f)

这是所有代码中最关键的文件之一：历史消息存储。
看这一段：

def get_history(session_id):
    # 调用刚刚创建的类，输入会话序号和历史消息的路径
    return FileChatMessageHistory(session_id, "./chat_history")

这个get_history，看过上期的朋友应该觉得眼熟：这就是增强链中作为“升级素材”的历史消息模块，这个函数会返回一个历史消息记录类，用于后续记录历史消息。
代码的注释很详细，因为这段代码当时花了我好大力气才理解，所以没有什么要补充的，大家到代码中看看注释，不出意外应该都可以理解。

代码文件：config_md5.py

"""
文件介绍
文件名：config_md5.py
作用：配置文件，用来指定路径
"""
md5_path = "./data/md5"

# 向量数据库的配置
collection_name = "chroma_db"
persist_directory = "./chroma_db"
k_num = 2

# 文本分割器的配置
chunk_size = 1000
chunk_overlap = 100
separator = ["\n\n", "\n", "!", "?", ".", "！", "？", "。", " ", ""]
max_split_char_number = 1000

# 模型名称
embedding_model_name = "text-embedding-v4"
chat_model_name = "qwen3-max"

# 会话轮次设置
session_config = {
        "configurable": {
            "session_id": "user_001",
        }
    }

这是整个项目的“设置项”。大多数重要且可能修改的变量值都在这里聚集。
所有代码文件中都有至少一两个变量可以在这里调节。
注意：如果你下载了这个项目的源码，请按照提示在这个文件添加你的DashScope API Key，或者在系统的环境变量里配置也可以。
具体每个变量有什么作用，可以在遇到时查看填入目标的说明或者查查资料，一般问AI很轻松就可以知道，这里不做过多注释。

代码文件：app_qa.py

import time
from rag import RagService
import streamlit as st
import config_md5 as config

# 标题
st.title("智能客服")
st.divider()

if "message" not in st.session_state:
    st.session_state["message"] = [{"role":"assistant","content":"你好，有什么可以帮你的？"}]

if "rag" not in st.session_state:
    st.session_state["rag"] = RagService()

for message in st.session_state["message"]:
    st.chat_message(message["role"]).write(message["content"])

# 在页面下方提供用户输入栏
prompt = st.chat_input()

if prompt:

    # 输出用户的提问
    st.chat_message("user").write(prompt)
    st.session_state["message"].append({"role": "user", "content": prompt})

    ai_res_list = []
    with st.spinner("AI思考中。。。"):
        res_stream = st.session_state["rag"].chain.stream({"input":prompt}, config.session_config)

        def capture(generator, cache_list):
            for chunk in generator:
                cache_list.append(chunk)
                yield chunk

        st.chat_message("assistant").write_stream(capture(res_stream, ai_res_list))
        st.session_state["message"].append({"role": "assistant", "content": "".join(ai_res_list)})

最终，我们来到了项目的前端部分，也就是和用户离得最近的部分。
这部分的注释很少，主要是在企业中几乎不会使用这种前端，只会当作个人项目和测试用，所以大家理解个框架就可以，后面写前端时都可以按这个思路来。
如果想仔细看看的话，这里有一个我用大模型添加注释后的版本，但是不一定说的都对，谨慎对待。

注释版本

# 导入自定义的 RAG 服务类，用于处理检索增强生成逻辑
from rag import RagService
# 导入 Streamlit 库，用于快速构建 Web 前端界面
import streamlit as st
# 导入项目配置文件，其中包含会话配置等参数
import config_md5 as config

# ==================== 页面布局初始化 ====================

# 设置页面标题
st.title("智能客服")
# 添加一条水平分隔线，美化页面
st.divider()

# ==================== 会话状态初始化 ====================

# 检查 session_state 中是否已有 "message" 键（存储对话历史）
# 如果不存在，则初始化为一条助手欢迎消息
if "message" not in st.session_state:
    st.session_state["message"] = [{"role": "assistant", "content": "你好，有什么可以帮你的？"}]

# 检查 session_state 中是否已有 RAG 服务实例
# 如果不存在，则创建一个新的 RagService 对象并保存到 session_state 中
# 这样可以确保页面重新运行时不会重复初始化，保持对话上下文
if "rag" not in st.session_state:
    st.session_state["rag"] = RagService()

# ==================== 显示历史消息 ====================

# 遍历 session_state 中存储的所有历史消息
# 每条消息是一个字典，包含 "role"（角色：user/assistant）和 "content"（消息内容）
for message in st.session_state["message"]:
    # st.chat_message(role) 创建聊天消息容器，根据角色显示不同样式（左/右）
    # .write(content) 在容器内写入消息内容
    st.chat_message(message["role"]).write(message["content"])

# ==================== 用户输入与响应处理 ====================

# 在页面底部显示一个聊天输入框，用户输入后按回车触发
# prompt 接收用户输入的文本，如果没有输入则值为 None
prompt = st.chat_input()

# 如果用户输入了内容（即 prompt 不为 None）
if prompt:

    # 1. 在界面上立即显示用户发送的消息
    st.chat_message("user").write(prompt)
    # 2. 将用户消息追加到 session_state 的历史记录中，以便下次刷新页面后仍能显示
    st.session_state["message"].append({"role": "user", "content": prompt})

    # 3. 准备接收 AI 的流式响应
    ai_res_list = []  # 用于收集所有流式块，最终拼接成完整回答

    # 显示一个加载提示（旋转圆环），表示 AI 正在思考
    with st.spinner("AI思考中。。。"):
        # 调用 RAG 服务链的 stream 方法，传入用户输入和会话配置（如 session_id）
        # 该方法返回一个生成器，逐块产出 AI 的响应文本
        res_stream = st.session_state["rag"].chain.stream({"input": prompt}, config.session_config)

        # 定义一个生成器函数，用于同时完成两件事：
        #   - 将每个流式块缓存到 cache_list 中（供后续组装完整回答）
        #   - 将每个流式块原样 yield 出去（供 write_stream 实时显示）
        def capture(generator, cache_list):
            for chunk in generator:
                cache_list.append(chunk)   # 缓存块
                yield chunk                # 实时输出块

        # 4. 在助手消息容器中实时流式输出 AI 的回答
        # write_stream 方法会自动消费生成器，并在界面上逐字显示内容
        st.chat_message("assistant").write_stream(capture(res_stream, ai_res_list))

        # 5. 将完整的 AI 回答（拼接后的字符串）追加到 session_state 的历史记录中
        # 注意：cache_list 中收集的是原始块（可能是字符串或消息对象），需要用 "".join() 拼接
        st.session_state["message"].append({"role": "assistant", "content": "".join(ai_res_list)})

到这里，所有的代码文件都放出来了，共有七个，还有一些文件夹就是在运行时存储数据的，会自动生成。

展示

这里放两张运行的截图。
在这里插入图片描述

结尾

那么，这次的RAG项目就告一段落了，如果大家对这个项目感兴趣，可以到GitHub下载这个项目来看看，地址如下：
https://github.com/QHzzy035/RAG_program_easy_to_use
感谢大家的支持，接下来我要继续学习其他知识了，如果有可以分享的知识我会继续创作和大家分享的。
如果对这七个代码文件有问题的，可以在评论区提出来，我会尽可能解答！