1. 项目简介

本项目是一个基于 RAG (Retrieval-Augmented Generation, 检索增强生成) 技术的智能客服/知识库问答系统。项目利用 LangChain 框架构建，结合 ChromaDB 向量数据库、DashScope (通义千问) 的 Embedding 和 LLM 模型，实现了私有知识库的构建与基于知识库的智能问答。

核心功能

• 知识库管理：支持上传文本文件，自动进行文本切分、向量化并存入 ChromaDB。内置 MD5 校验机制，防止重复上传。
• 智能问答：基于用户问题检索知识库相关内容，结合对话历史（Context-Aware），由大模型生成准确回答。
• 会话记忆：支持基于文件的长时会话历史存储，能够进行多轮对话。
• Web 交互界面：提供两个 Streamlit 界面，分别用于"知识库管理"和"在线问答"。

技术栈

技术	用途
Python 3.11	开发语言
LangChain	RAG 流程编排框架
ChromaDB	本地向量数据库
DashScope (通义千问)	Embedding 模型 + LLM 大语言模型
Streamlit	Web UI 快速开发框架

项目文件结构

rag/
├── app_file_uploader.py   # [入口] 知识库管理 Web 界面 (Streamlit)
├── app_qa.py              # [入口] 智能客服问答 Web 界面 (Streamlit)
├── config_data.py         # [配置] 项目全局配置文件 (模型、路径、参数等)
├── file_history_store.py  # [工具] 基于文件的会话历史存储服务
├── knowledge_base.py      # [核心] 知识库服务 (文档处理、向量化入库)
├── rag.py                 # [核心] RAG 核心逻辑 (检索链构建、Prompt 工程)
├── vector_stores.py       # [核心] 向量数据库服务 (ChromaDB 封装)
├── data/                  # [数据] 存放原始文本数据及 ChromaDB 数据库文件
│   ├── chroma_db/         # 向量数据库持久化目录
│   ├── md5_records.txt    # 已上传文件的 MD5 记录
│   ├── 尺码推荐.txt       # 示例知识：尺码推荐指南
│   └── 洗涤养护.txt       # 示例知识：洗涤养护说明
└── chat_history/          # [数据] 存放用户的会话历史记录 (JSON 格式)
    └── user_001           # 用户 user_001 的聊天记录文件

项目流程图解

离线构建流程（Knowledge Base Construction）：

上传 TXT → MD5 校验（去重）→ 文本切分 (Splitter) → 向量化 (Embedding) → 存入 ChromaDB

在线问答流程（RAG Pipeline）：

用户提问 → 加载历史记录 → 向量检索 (Retriever) → 构建 Prompt (Query + Context + History) → LLM 生成 → 流式返回

---

2. 前置要求

2.1 Python 环境

推荐安装 Python 3.11 版本。只要你有一个 Python 环境，后面的问题都会迎刃而解。

2.2 安装依赖

pip install streamlit langchain langchain-community langchain-chroma langchain-core langchain-text-splitters dashscope

注意：请确保已配置 DashScope API Key 环境变量 DASHSCOPE_API_KEY，或在代码中显式指定。

设置环境变量（Windows PowerShell）：

$env:DASHSCOPE_API_KEY = "你的API Key"

设置环境变量（Linux / macOS）：

export DASHSCOPE_API_KEY="你的API Key"

2.3 核心依赖说明

依赖包	作用
streamlit	构建 Web UI 界面
langchain	LangChain 核心框架
langchain-community	社区集成模型（ChatTongyi、DashScopeEmbeddings）
langchain-chroma	LangChain 的 ChromaDB 集成
langchain-core	LangChain 核心抽象（Prompt、Runnable、Messages 等）
langchain-text-splitters	文本切分工具
dashscope	阿里云 DashScope SDK（通义千问系列模型）

---

3. 功能实现

整个项目的开发顺序如下：

1. 全局配置 (config_data.py) — 定义所有可调参数
2. 知识库服务 (knowledge_base.py) — 实现文本上传、去重、切分、入库
3. 知识库上传页面 (app_file_uploader.py) — 提供 Web 界面上传知识文件
4. 向量检索服务 (vector_stores.py) — 封装 ChromaDB 检索器
5. 会话历史存储 (file_history_store.py) — 实现基于文件的聊天记录持久化
6. RAG 核心链 (rag.py) — 串联检索、Prompt、LLM、历史记录
7. 智能问答页面 (app_qa.py) — 提供聊天界面，流式输出回答

下面按照开发顺序，逐一讲解每个模块。

---

3.1 全局配置 — config_data.py

在开始编写业务逻辑之前，我们先把项目中需要用到的所有可配置参数集中管理，方便后续修改和维护。

创建文件 config_data.py，写入以下代码：

# Chroma数据库配置
from langchain_community.embeddings import DashScopeEmbeddings

# chroma数据库配置
collection_name = "knowledge_rag"
embedding_function = DashScopeEmbeddings( # 嵌入模型选择
    model="text-embedding-v4"
)
chat_model = "qwen3-max-preview" # 对话大模型选择
persist_directory = "./data/chroma_db" # chroma数据库文件存储路径

# RecursiveCharacterTextSplitter配置
chunk_size = 1000  # 每个文档分块的最大字符数
chunk_overlap = 100  # 分块之间的重叠字符数
separators = ["\n\n", "\n", ".", "！", "？", "!", "?", "。"," ",""]  # 用于分割文本的分隔符列表
split_char_len = 1000  # 超过这个长度的文本会被递归分割

# 知识库检索器配置
retriever_k = 1  # 每次检索返回的文档数量

# md5记录文件路径，用于记录已上传文档的md5值，避免重复上传
md5_path = "./data/md5_records.txt"

session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

代码详解

Embedding 模型配置

from langchain_community.embeddings import DashScopeEmbeddings

embedding_function = DashScopeEmbeddings(
    model="text-embedding-v4"
)

• DashScopeEmbeddings：LangChain 社区提供的阿里云 DashScope Embedding 封装类，能将文本转化为高维向量。
• model="text-embedding-v4"：使用通义千问的 text-embedding-v4 模型，这是阿里云提供的最新文本向量化模型，支持中英文混合文本。

什么是 Embedding？

Embedding（嵌入/向量化）是将文本转化为固定维度的浮点数向量的过程。相似的文本在向量空间中距离较近，这是实现语义检索的基础。例如"我想买件衣服"和"购买服装"虽然用词不同，但 Embedding 后的向量非常相似。

ChromaDB 配置

collection_name = "knowledge_rag"
persist_directory = "./data/chroma_db"

参数	类型	说明
collection_name	str	ChromaDB 中的集合（Collection）名称，类似数据库中的"表名"。所有知识文档都存储在这个集合中。
persist_directory	str	ChromaDB 的本地持久化目录。数据会以文件形式保存在此路径下，重启程序后数据不会丢失。

LLM 模型配置

chat_model = "qwen3-max-preview"

指定用于最终回答生成的大语言模型。qwen3-max-preview 是通义千问3系列的高性能模型。

文本切分配置

chunk_size = 1000
chunk_overlap = 100
separators = ["\n\n", "\n", ".", "！", "？", "!", "?", "。"," ",""]
split_char_len = 1000

参数	类型	说明
chunk_size	int	每个文本分块的最大字符数。过大会导致检索不精准，过小会丢失上下文语义。1000 是一个合理的默认值。
chunk_overlap	int	相邻分块之间的重叠字符数。重叠可以保证分块边界处的语义连续性，避免关键信息被截断。
separators	list[str]	递归分割器使用的分隔符优先级列表。会优先按 \n\n（段落）分割，如果仍超长则按 \n（行），依次递归直到满足 chunk_size 要求。
split_char_len	int	只有文本长度超过此阈值时才会触发分割，短文本直接入库。

检索器配置

retriever_k = 1

每次向量检索返回的最相关文档数量。k=1 表示只返回最相关的 1 个文档片段。可根据需要增大。

会话配置

session_config = {
    "configurable": {
        "session_id": "user_001"
    }
}

定义了会话标识配置，session_id 用于区分不同用户的聊天历史记录。在实际生产环境中，这个 ID 应该由用户登录系统动态生成。

---

3.2 知识库服务 — knowledge_base.py

知识库服务是整个项目的基础。它负责接收上传的文本内容，进行 MD5 去重校验、文本切分，并将切分后的文本块向量化存入 ChromaDB。

创建文件 knowledge_base.py，写入以下代码：

import os
from pathlib import Path

from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter

import config_data
import hashlib
import datetime

3.2.1 MD5 工具函数

首先实现一组 MD5 相关的工具函数，用于防止重复上传相同内容到知识库：

def _ensure_md5_dir():
    Path(config_data.md5_path).parent.mkdir(parents=True, exist_ok=True)


def check_md5(md5_str: str) -> bool:
    if not os.path.exists(config_data.md5_path):
        return False

    with open(config_data.md5_path, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip() == md5_str:
                return True
    return False


def save_md5(md5_str: str) -> bool:
    if check_md5(md5_str):
        return False
    _ensure_md5_dir()
    with open(config_data.md5_path, 'a', encoding='utf-8') as f:
        f.write(md5_str + '\n')
    return True


def get_string_md5(input_str: str, encoding='utf-8') -> str:
    """
    将传入的字符串转为md5
    """
    str_bytes = input_str.encode(encoding)  # 将字符串转换为字节序列
    md5_hash = hashlib.md5()  # 创建md5哈希对象
    md5_hash.update(str_bytes)  # 更新哈希对象的状态，将字节序列加入计算
    return md5_hash.hexdigest()  # 返回md5哈希值的十六进制表示

函数说明：

函数	参数	返回值	作用
_ensure_md5_dir()	无	无	确保 MD5 记录文件的父目录存在，不存在则自动创建
check_md5(md5_str)	md5_str: str — MD5 哈希值	bool — 是否已存在	逐行读取 MD5 记录文件，检查给定的 MD5 值是否已经记录过
save_md5(md5_str)	md5_str: str — MD5 哈希值	bool — 是否保存成功	先调用 check_md5 校验是否重复，不重复则追加写入文件
get_string_md5(input_str, encoding)	input_str: str — 原文本；encoding: str — 编码方式，默认 'utf-8'	str — 32 位十六进制 MD5 字符串	计算文本内容的 MD5 哈希值

为什么用 MD5 做去重？

MD5 是一种哈希算法，能将任意长度的文本映射为固定 32 位的十六进制字符串。相同内容的 MD5 值一定相同，因此可以用来快速判断文本是否已经上传过，避免知识库中出现重复内容。

hashlib.md5() 核心方法说明：

方法	说明
hashlib.md5()	创建一个 MD5 哈希对象
.update(data)	将 bytes 数据加入哈希计算。可多次调用以分块处理大文件
.hexdigest()	返回最终的哈希值，格式为 32 位十六进制字符串（如 "d41d8cd98f00b204e9800998ecf8427e"）

3.2.2 KnowledgeBaseService 类

class KnowledgeBaseService:
    """知识库服务类，用于管理知识库的上传"""

    def __init__(self) -> None:
        Path(config_data.persist_directory).mkdir(parents=True, exist_ok=True)
        self.chroma = Chroma(
            collection_name=config_data.collection_name,
            embedding_function=config_data.embedding_function,
            persist_directory=config_data.persist_directory,
        )
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=config_data.chunk_size,
            chunk_overlap=config_data.chunk_overlap,
            separators=config_data.separators,
            length_function=len
        )

    def upload_by_str(self, data: str, filename: str) -> str:
        """
        上传字符串到知识库

        :param data: 要上传的文本数据
        :param filename: 文件名
        """
        if not data or not data.strip():
            return "[失败]上传内容为空，无法入库"
        md5_hex = get_string_md5(data)
        if check_md5(md5_hex):
            return "[跳过]内容已经存在在知识库中"
        # 对文本进行分块
        if len(data) > config_data.split_char_len:
            knowledge_chunks = self.splitter.split_text(data)
        else:
            knowledge_chunks = [data]
        # 元数据
        metadata = {
            "source": filename,
            "datetime": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "operator": "admin",
        }
        self.chroma.add_texts(
            texts=knowledge_chunks,
            metadatas=[metadata for _ in knowledge_chunks]
        )
        save_md5(md5_hex)
        return "[成功]知识库上传成功"

__init__ 构造方法详解：

构造方法初始化了两个核心组件：

1. Chroma 向量数据库实例

self.chroma = Chroma(
    collection_name=config_data.collection_name,
    embedding_function=config_data.embedding_function,
    persist_directory=config_data.persist_directory,
)

Chroma 是 LangChain 对 ChromaDB 的封装类，来自 langchain_chroma 包。

参数	类型	说明
collection_name	str	集合名称，相当于数据库中的表名。同一个持久化目录下可以有多个集合。
embedding_function	Embeddings	嵌入模型实例，用于将文本转为向量。这里使用的是 DashScopeEmbeddings。
persist_directory	str	数据库持久化目录路径，数据以文件形式保存在磁盘上。

2. RecursiveCharacterTextSplitter 文本切分器

self.splitter = RecursiveCharacterTextSplitter(
    chunk_size=config_data.chunk_size,
    chunk_overlap=config_data.chunk_overlap,
    separators=config_data.separators,
    length_function=len
)

RecursiveCharacterTextSplitter 是 LangChain 提供的递归字符文本切分器，它的工作原理是：

1. 按照 separators 列表的顺序依次尝试分割文本
2. 优先使用排在前面的分隔符（如 \n\n 段落分隔）
3. 如果分割后的块仍然超过 chunk_size，则用下一个分隔符继续递归分割
4. 相邻块之间保留 chunk_overlap 个字符的重叠

参数	类型	说明
chunk_size	int	每个分块的最大字符数
chunk_overlap	int	相邻分块的重叠字符数，保证语义连续性
separators	list[str]	分隔符优先级列表，从左到右依次尝试
length_function	callable	计算文本长度的函数，默认 len（按字符数）

为什么需要文本切分？

• Embedding 模型通常有 token 长度限制
• 短文本的向量表示更精准，检索相关性更高
• 避免将大量无关信息一起塞入 LLM 的 Prompt

upload_by_str 方法详解：

该方法是知识库上传的核心入口，完整流程如下：

接收文本 → 空值校验 → 计算 MD5 → 去重检查 → 文本切分 → 构建元数据 → 写入向量库 → 记录 MD5

关键步骤分析：

① 空值校验

if not data or not data.strip():
    return "[失败]上传内容为空，无法入库"

防止空字符串或纯空白字符串入库。

② MD5 去重

md5_hex = get_string_md5(data)
if check_md5(md5_hex):
    return "[跳过]内容已经存在在知识库中"

计算上传文本的 MD5 值，与已记录的 MD5 比对，如果相同则跳过。

③ 条件切分

if len(data) > config_data.split_char_len:
    knowledge_chunks = self.splitter.split_text(data)
else:
    knowledge_chunks = [data]

只有文本长度超过阈值（1000 字符）时才调用切分器，短文本直接作为单个块处理。

④ 构建元数据并入库

metadata = {
    "source": filename,
    "datetime": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    "operator": "admin",
}
self.chroma.add_texts(
    texts=knowledge_chunks,
    metadatas=[metadata for _ in knowledge_chunks]
)

chroma.add_texts() 方法说明：

参数	类型	说明
texts	list[str]	要存入的文本列表，每个元素是一个文本块
metadatas	list[dict]	与 texts 一一对应的元数据列表。元数据会随向量一起存储，检索时可返回

元数据（metadata）记录了每个文本块的来源信息：

字段	说明
source	原始文件名
datetime	上传时间
operator	操作者

---

3.3 知识库上传页面 — app_file_uploader.py

有了 KnowledgeBaseService 之后，我们需要一个 Web 界面让用户方便地上传知识文件。这里使用 Streamlit 快速实现。

3.3.1 Streamlit 是什么？

Streamlit 是一个开源的 Python 库，用于快速构建和共享数据科学与机器学习的交互式 Web 应用。它的核心理念是：用纯 Python 编写应用，无需前端开发经验（如 HTML、CSS 或 JavaScript）。

主要特点：

1. 简单易用 — 只需几行 Python 代码就能创建一个带有滑块、按钮、图表等交互组件的 Web 界面。
2. 专为数据科学设计 — 无缝集成 Pandas、NumPy、Matplotlib、Plotly 等常用库。
3. 自动重载（Hot Reload） — 修改代码后，浏览器中的应用会自动刷新，便于快速迭代开发。
4. 部署方便 — 可通过 Streamlit Community Cloud（免费）、Heroku、AWS、Docker 等平台一键部署。
5. 支持丰富的 UI 组件 — 包括文本输入、文件上传、下拉菜单、复选框、表格、图表、进度条、侧边栏等。

一个简单的例子：

import streamlit as st

st.title("你好，Streamlit！")
name = st.text_input("请输入你的名字")
if name:
    st.write(f"欢迎你，{name}！")

将上述代码保存为 app.py，然后在终端运行：

streamlit run app.py

浏览器会自动打开一个本地 Web 应用，用户可以输入名字并看到欢迎信息。

Streamlit 的执行模型：Streamlit 的一个重要特性是——每当页面上的任何元素发生交互（如按钮点击、文件上传），整个 Python 脚本会从头到尾重新执行一遍。因此需要使用 st.session_state 来保存跨交互的状态。

3.3.2 上传页面代码实现

创建文件 app_file_uploader.py，写入以下代码：

import streamlit as st
from time import sleep
from knowledge_base import KnowledgeBaseService

# 添加网页标题
st.title("知识库更新服务")

# 文件上传组件
uploader_file = st.file_uploader(
    "上传文件 (TXT)",
    type=["txt"],
    accept_multiple_files=False,  # 是否允许上传多个文件
)

# 初始化计数器，记录上传文件的数量
if "service" not in st.session_state:
    st.session_state["service"] = KnowledgeBaseService()

if uploader_file is not None:
    # 获取文件信息
    file_name = uploader_file.name
    file_size = uploader_file.size / 1024  # 转换为KB
    file_type = uploader_file.type
    st.subheader(f"文件名: {file_name}")
    st.write(f"文件类型: {file_type} | 文件大小: {file_size:.2f} KB")

    text = uploader_file.read().decode("utf-8")

    with st.spinner("正在上传..."):
        sleep(1)
        res = st.session_state["service"].upload_by_str(text, file_name)
        st.write(res)

代码详解

导入部分

import streamlit as st
from time import sleep
from knowledge_base import KnowledgeBaseService

• streamlit as st：导入 Streamlit 主模块，用于构建 Web UI。
• sleep：来自标准库 time，用于模拟耗时操作（如网络请求、文件处理），让界面显示"加载中"效果。
• KnowledgeBaseService：我们在上一步实现的知识库服务类。

页面标题

st.title("知识库更新服务")

st.title() 在页面顶部渲染一个大号标题（对应 HTML 的 <h1> 标签）。

文件上传组件

uploader_file = st.file_uploader(
    "上传文件 (TXT)",
    type=["txt"],
    accept_multiple_files=False,
)

st.file_uploader() 是 Streamlit 提供的文件上传控件，返回一个 UploadedFile 对象（或 None）。

参数	类型	默认值	含义
label ("上传文件 (TXT)")	str	必填	显示在上传按钮上方的提示文本
type (["txt"])	list[str] 或 str	None	限制允许上传的文件类型。这里只允许 .txt 文件。若设为 None，则不限制
accept_multiple_files	bool	False	是否允许多选文件上传。False → 返回单个 UploadedFile 对象或 None；True → 返回 list[UploadedFile]

uploader_file 的值：

• 若用户未上传：None
• 若已上传：UploadedFile 对象，可像文件对象一样操作（如 .read()、.name、.size）

使用 Session State 初始化服务实例

if "service" not in st.session_state:
    st.session_state["service"] = KnowledgeBaseService()

st.session_state 是 Streamlit 的会话状态管理器，类似 Web 开发中的 “session”。用于在用户多次交互中持久化数据（否则每次操作都会重新运行整个脚本，变量重置）。

为什么需要它？

如果不在 session_state 中保存 KnowledgeBaseService() 实例，每次上传文件都会创建新实例，导致无法累积知识或保持内部状态（如数据库连接、缓存等）。

读取文件内容

text = uploader_file.read().decode("utf-8")

• .read()：返回文件内容的 bytes 对象
• .decode("utf-8")：将 bytes 转为字符串（假设文件是 UTF-8 编码）

⚠️ 如果文件不是 UTF-8 编码（如 GBK），会报错。生产环境应加异常处理。

显示加载动画并调用服务

with st.spinner("正在上传..."):
    sleep(1)
    res = st.session_state["service"].upload_by_str(text, file_name)
    st.write(res)

st.spinner() 是一个上下文管理器（with 语句），在执行内部代码时显示旋转加载图标 + 提示文字。适用于任何耗时操作（如 API 调用、模型推理、数据库写入）。

参数	说明
"正在上传..."	显示在 spinner 下方的提示文本

sleep(1) 用于模拟耗时操作（实际项目中可以去掉）。

启动方式：

streamlit run app_file_uploader.py

启动后应该能看到如下页面

---

3.4 向量检索服务 — vector_stores.py

向量检索服务封装了 ChromaDB 的检索功能，为 RAG 核心链提供检索器（Retriever）。

创建文件 vector_stores.py，写入以下代码：

from langchain_chroma import Chroma

import config_data


class VectorStoreService:
    """向量存储服务类，用于根据知识库提问"""

    def __init__(self, embedding_function=config_data.embedding_function):
        self.vector_store = Chroma(
            collection_name=config_data.collection_name,
            embedding_function=embedding_function,
            persist_directory=config_data.persist_directory,
        )

    def get_retriever(self):
        """获取向量存储的检索器"""
        return self.vector_store.as_retriever(
            search_kwargs={"k": config_data.retriever_k},
        )

代码详解

__init__ 构造方法

def __init__(self, embedding_function=config_data.embedding_function):
    self.vector_store = Chroma(
        collection_name=config_data.collection_name,
        embedding_function=embedding_function,
        persist_directory=config_data.persist_directory,
    )

与 KnowledgeBaseService 中的 Chroma 初始化类似，连接到同一个持久化数据库和集合。embedding_function 作为参数传入，方便在不同场景下替换 Embedding 模型。

get_retriever 方法

def get_retriever(self):
    return self.vector_store.as_retriever(
        search_kwargs={"k": config_data.retriever_k},
    )

as_retriever() 是 LangChain 向量存储的标准方法，将 VectorStore 转为 Retriever 对象。

参数	类型	说明
search_kwargs	dict	传递给底层搜索方法的关键字参数
search_kwargs["k"]	int	返回最相关的 Top-K 个文档。这里 k=1 表示只返回最匹配的 1 个文档片段

什么是 Retriever？

Retriever（检索器）是 LangChain 中的一个核心抽象。它接收一个查询字符串，返回一组相关的 Document 对象。在 RAG 流程中，Retriever 负责从知识库中找到与用户问题最相关的文档片段，作为 LLM 回答的参考资料。

检索过程：

1. 将用户问题通过 Embedding 模型转为向量
2. 在 ChromaDB 中做近似最近邻搜索（ANN）
3. 返回距离最近的 K 个文档

测试代码：

if __name__ == "__main__":
    vector_store_service = VectorStoreService()
    retriever = vector_store_service.get_retriever()
    res = retriever.invoke("我的身高是180cm，体重是85kg，应该选什么尺码")
    for i, doc in enumerate(res):
        print(f"Document {i + 1}:")
        print(f"Content: {doc.page_content}")
        print(f"Metadata: {doc.metadata}")
        print("-" * 20)

retriever.invoke(query) 接收一个查询字符串，返回 List[Document]。每个 Document 包含：

• page_content：文档文本内容
• metadata：元数据字典（如来源文件名、上传时间等）

---

3.5 会话历史存储 — file_history_store.py

为了支持多轮对话，我们需要一种机制来存储和读取用户的聊天历史。LangChain 提供了 BaseChatMessageHistory 抽象基类，我们基于文件系统实现一个自定义的历史存储。

创建文件 file_history_store.py，写入以下代码：

import json
import os
from typing import Sequence

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict


def get_history(session_id: str):
    """获取会话历史记录"""
    return FileChatMessageHistory(session_id, store_path="./chat_history")


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id: str, store_path: str = "./chat_history"):
        self.session_id = session_id
        self.store_path = store_path
        self.file_path = os.path.join(self.store_path, self.session_id)
        os.makedirs(self.store_path, exist_ok=True)

    def add_message(self, message: BaseMessage) -> None:
        self.add_messages([message])

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        all_messages = list(self.messages)
        all_messages.extend(messages)
        new_messages = [message_to_dict(msg) for msg in all_messages]
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="utf-8") as f:
                messages = json.load(f)
                return messages_from_dict(messages)
        except json.JSONDecodeError:
            return []
        except FileNotFoundError:
            return []

    def clear(self):
        if os.path.exists(self.file_path):
            with open(self.file_path, "w", encoding="utf-8") as f:
                json.dump([], f)

代码详解

工厂函数 get_history

def get_history(session_id: str):
    return FileChatMessageHistory(session_id, store_path="./chat_history")

这是一个简单的工厂函数，接收 session_id，返回对应的历史存储实例。后续在 RAG 链中会用到它——RunnableWithMessageHistory 需要一个"根据 session_id 获取历史记录对象"的函数。

FileChatMessageHistory 类

该类继承自 LangChain 的 BaseChatMessageHistory 抽象基类，实现了基于本地文件的消息持久化存储。

BaseChatMessageHistory 是什么？

它是 LangChain 定义的聊天历史存储接口，任何实现了这个接口的类都可以被 LangChain 的会话管理组件使用。需要实现以下方法/属性：

• add_message() / add_messages() — 添加消息
• messages (property) — 获取所有消息
• clear() — 清空消息

__init__ 构造方法

def __init__(self, session_id: str, store_path: str = "./chat_history"):
    self.session_id = session_id
    self.store_path = store_path
    self.file_path = os.path.join(self.store_path, self.session_id)
    os.makedirs(self.store_path, exist_ok=True)

参数	类型	说明
session_id	str	会话唯一标识，用作文件名（如 user_001）
store_path	str	存储目录路径，默认 ./chat_history

每个用户的聊天记录保存为一个独立文件，文件路径为 ./chat_history/{session_id}。

add_messages 方法

def add_messages(self, messages: Sequence[BaseMessage]) -> None:
    all_messages = list(self.messages)  # 读取现有消息
    all_messages.extend(messages)       # 追加新消息
    new_messages = [message_to_dict(msg) for msg in all_messages]  # 序列化
    with open(self.file_path, "w", encoding="utf-8") as f:
        json.dump(new_messages, f)      # 写入文件

核心逻辑是"读取全部 → 追加新消息 → 全量覆写"。这里用到了两个 LangChain 工具函数：

函数	说明
message_to_dict(msg)	将 BaseMessage 对象序列化为字典，便于 JSON 存储
messages_from_dict(dicts)	将字典列表反序列化为 BaseMessage 对象列表

messages 属性（Property）

@property
def messages(self) -> list[BaseMessage]:
    try:
        with open(self.file_path, "r", encoding="utf-8") as f:
            messages = json.load(f)
            return messages_from_dict(messages)
    except json.JSONDecodeError:
        return []
    except FileNotFoundError:
        return []

使用 @property 装饰器，使得可以像访问属性一样调用 self.messages。异常处理覆盖了两种情况：

• json.JSONDecodeError：文件存在但内容不是合法 JSON（如被手动篡改）
• FileNotFoundError：文件不存在（新用户首次对话）

clear 方法

def clear(self):
    if os.path.exists(self.file_path):
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)

将文件内容重置为空 JSON 数组 []，实现清空历史记录。

---

3.6 RAG 核心链 — rag.py

这是整个项目最核心的模块，将 向量检索、Prompt 模板、LLM、会话历史 串联成一条完整的 LangChain 处理链（Chain）。

创建文件 rag.py，写入以下代码：

from langchain_community.chat_models import ChatTongyi
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory, RunnableLambda
from typing import List

import config_data
from file_history_store import get_history
from vector_stores import VectorStoreService

3.6.1 调试辅助函数

def print_prompt(prompt):
    print("=" * 50)
    print(prompt)
    print("=" * 50)
    return prompt

这是一个简单的调试函数，用于在控制台打印最终构建好的 Prompt 内容，方便开发时检查 Prompt 是否正确。它作为 RunnableLambda 插入链中，不修改数据，只打印后原样返回。

3.6.2 RagService 类

class RagService:
    def __init__(self):
        self.vector_service = VectorStoreService(
            embedding_function=config_data.embedding_function
        )
        self.prompt_template = ChatPromptTemplate.from_messages(
            [
                ("system", "以我提供已知的参考资料为主，简洁的回答用户问题，参考资料:{context}"),
                ("system", "根据我提供的对话历史记录，回答用户问题"),
                MessagesPlaceholder(variable_name="history"),
                ("user", "{question}"),
            ]
        )
        self.chat_model = ChatTongyi(
            model=config_data.chat_model
        )
        self.chain = self.__get_chain()

构造方法详解：

1. 初始化向量检索服务

self.vector_service = VectorStoreService(
    embedding_function=config_data.embedding_function
)

复用之前实现的 VectorStoreService，用于获取 Retriever。

2. 构建 Prompt 模板

self.prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "以我提供已知的参考资料为主，简洁的回答用户问题，参考资料:{context}"),
        ("system", "根据我提供的对话历史记录，回答用户问题"),
        MessagesPlaceholder(variable_name="history"),
        ("user", "{question}"),
    ]
)

ChatPromptTemplate.from_messages() 用于创建多消息 Prompt 模板。每个元素是一个 (role, content) 元组或特殊占位符。

消息	角色	说明
第 1 条	system	系统指令，告诉 LLM 优先使用参考资料回答。{context} 是检索到的知识库内容占位符
第 2 条	system	系统指令，告诉 LLM 结合对话历史回答
第 3 条	MessagesPlaceholder	历史消息占位符，运行时会被替换为之前的对话记录列表
第 4 条	user	用户当前的问题。{question} 是用户输入占位符

MessagesPlaceholder 是什么？

它是 LangChain 提供的特殊占位符，用于在 Prompt 模板中插入一组动态的消息列表（如聊天历史）。与普通的 {variable} 不同，它可以插入多条不同角色的消息。

参数	说明
variable_name	变量名，运行时通过这个 key 传入消息列表

3. 初始化 LLM 模型

self.chat_model = ChatTongyi(
    model=config_data.chat_model
)

ChatTongyi 是 LangChain 社区提供的通义千问聊天模型封装。model="qwen3-max-preview" 指定使用的具体模型。

3.6.3 构建 RAG 链 — __get_chain 方法(RagService类下)

这是整个项目最复杂的部分，用 LangChain Expression Language (LCEL) 构建了一条完整的 RAG 处理管道：

def __get_chain(self):
    """获取RAG链"""
    retriever = self.vector_service.get_retriever()

    def format_docs(docs: List[Document]):
        """格式化文档，将文档内容合并为一个字符串"""
        if not docs:
            print(">>> [参考资料] 没有检索到相关文档")
            return "没有相关文档"
        formatted_str = ""
        for doc in docs:
            formatted_str += f"文档片段：{doc.page_content}\n文档元数据：{doc.metadata}\n\n"
        print("=" * 50)
        print(">>> [参考资料]")
        print(formatted_str)
        print("=" * 50)
        return formatted_str

    def print_history(x):
        history = x.get("history", [])
        print("=" * 50)
        print(">>> [历史聊天记录]")
        if not history:
            print("（暂无历史记录）")
        else:
            for msg in history:
                role = getattr(msg, "type", "unknown")
                content = getattr(msg, "content", str(msg))
                print(f"  [{role}]: {content}")
        print("=" * 50)
        return x

    chain = (
            RunnableLambda(print_history)
            | {
                "context": (lambda x: x["question"]) | retriever | format_docs,
                "question": lambda x: x["question"],
                "history": lambda x: x.get("history", []),
            }
            | self.prompt_template
            | RunnableLambda(print_prompt)
            | self.chat_model
            | StrOutputParser()
    )

    # 包装RAG链，添加会话历史记录功能
    conversation_chain = RunnableWithMessageHistory(
        chain,
        get_history,
        input_messages_key="question",
        history_messages_key="history"
    )

    return conversation_chain

内部辅助函数：

format_docs(docs) — 将检索到的 Document 列表格式化为字符串，作为 Prompt 的 {context} 内容。

def format_docs(docs: List[Document]):
    if not docs:
        return "没有相关文档"
    formatted_str = ""
    for doc in docs:
        formatted_str += f"文档片段：{doc.page_content}\n文档元数据：{doc.metadata}\n\n"
    return formatted_str

print_history(x) — 调试函数，打印当前的历史聊天记录到控制台。

LCEL 链的构成（核心重点）：

chain = (
        RunnableLambda(print_history)           # ① 打印历史记录（调试）
        | {                                      # ② 并行处理，构建 Prompt 变量
            "context": (lambda x: x["question"]) | retriever | format_docs,
            "question": lambda x: x["question"],
            "history": lambda x: x.get("history", []),
        }
        | self.prompt_template                   # ③ 填充 Prompt 模板
        | RunnableLambda(print_prompt)           # ④ 打印 Prompt（调试）
        | self.chat_model                        # ⑤ LLM 生成回答
        | StrOutputParser()                      # ⑥ 提取纯文本回答
)

什么是 LCEL (LangChain Expression Language)？

LCEL 是 LangChain 提供的声明式链组合语法，使用 | 管道运算符将多个 Runnable 组件串联起来，数据从左到右依次流过每个组件。类似 Unix 的管道 cmd1 | cmd2 | cmd3。

链的每一步详解：

步骤	组件	输入	输出	说明
①	RunnableLambda(print_history)	{"question": "...", "history": [...]}	原样返回	调试用，打印历史记录
②	字典（并行 Runnable）	同上	{"context": "...", "question": "...", "history": [...]}	并行执行三个子任务，构建 Prompt 所需的变量
③	prompt_template	上一步的字典	ChatPromptValue	将变量填入 Prompt 模板，生成完整的 Prompt
④	RunnableLambda(print_prompt)	ChatPromptValue	原样返回	调试用，打印最终 Prompt
⑤	self.chat_model	ChatPromptValue	AIMessage	调用通义千问 LLM 生成回答
⑥	StrOutputParser()	AIMessage	str	从 LLM 返回的消息中提取纯文本

步骤②的并行处理详解：

{
    "context": (lambda x: x["question"]) | retriever | format_docs,
    "question": lambda x: x["question"],
    "history": lambda x: x.get("history", []),
}

当 LCEL 中使用字典时，LangChain 会并行执行字典中每个 key 的 Runnable：

• "context"：提取问题 → 向量检索 → 格式化文档。这是一条子链：
  1. lambda x: x["question"] — 从输入中提取用户问题字符串
  2. | retriever — 将问题传给检索器，返回相关文档列表
  3. | format_docs — 将文档列表格式化为字符串
• "question"：直接透传用户问题
• "history"：透传历史消息列表

RunnableLambda 是什么？

RunnableLambda 将普通 Python 函数包装为 LangChain 的 Runnable 对象，使其可以参与 LCEL 链的组合。

RunnableLambda(my_function)

等价于一个 Runnable，调用时执行 my_function(input) 并返回结果。

StrOutputParser 是什么？

StrOutputParser 是 LangChain 的输出解析器，它从 LLM 返回的 AIMessage 对象中提取 .content 属性（即纯文本回答字符串）。

会话历史包装：

conversation_chain = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="question",
    history_messages_key="history"
)

RunnableWithMessageHistory 是 LangChain 提供的高阶组件，它自动完成以下工作：

1. 调用前：根据 session_id 调用 get_history 函数获取历史消息，注入到链的输入中（key 为 history）
2. 调用后：将用户的问题和 LLM 的回答自动追加到历史记录中

参数	类型	说明
runnable	Runnable	要包装的链（即上面构建的 chain）
get_session_history	Callable[[str], BaseChatMessageHistory]	根据 session_id 获取历史记录对象的函数
input_messages_key	str	输入字典中"用户消息"对应的 key
history_messages_key	str	输入字典中"历史消息"对应的 key

测试代码：

if __name__ == "__main__":
    rag_service = RagService()
    res = rag_service.chain.invoke(
        {
            "question": "你知道我的身高和体重吗？我应该选什么尺码的衣服？"
        },
        config=config_data.session_config
    )
    print(res)

调用时需要传入 config 参数，其中包含 session_id，用于 RunnableWithMessageHistory 识别当前会话。

假设你之前告诉了大模型你的身高，当你再次执行时，大模型能够读取历史聊天记录，然后根据知识库来回答你的问题

---

3.7 智能问答页面 — app_qa.py

最后，我们构建一个类似 ChatGPT 的聊天界面，用户可以在此与智能客服进行多轮对话，系统支持流式输出（逐字显示回答）。

创建文件 app_qa.py，写入以下代码：

import streamlit as st

import config_data
from rag import RagService

st.set_page_config(page_title="智能客服", page_icon="🤖")
st.title("🤖 智能客服")
st.divider()

# ── 初始化 session_state ──
if "messages" not in st.session_state:
    st.session_state["messages"] = [
        {"role": "assistant", "content": "你好，我是智能客服，有什么问题我可以帮助你吗？"}
    ]

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

# ── 渲染历史消息 ──
for message in st.session_state["messages"]:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# ── 用户输入 ──
if prompt := st.chat_input("请输入你的问题..."):
    # 显示用户消息并追加到历史
    st.session_state["messages"].append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # 流式输出助手回复
    with st.chat_message("assistant"):
        response = st.write_stream(
            st.session_state["rag_service"].chain.stream(
                {"question": prompt},
                config=config_data.session_config,
            )
        )

    # 将完整回复追加到历史
    st.session_state["messages"].append({"role": "assistant", "content": response})

代码详解

页面配置

st.set_page_config(page_title="智能客服", page_icon="🤖")
st.title("🤖 智能客服")
st.divider()

函数	说明
st.set_page_config()	设置浏览器标签页的标题和图标。必须是脚本中调用的第一个 Streamlit 命令
st.title()	页面大标题
st.divider()	渲染一条水平分隔线

st.set_page_config() 参数说明：

参数	类型	说明
page_title	str	浏览器标签页标题
page_icon	str	浏览器标签页图标，支持 emoji 或图片路径

初始化 Session State

if "messages" not in st.session_state:
    st.session_state["messages"] = [
        {"role": "assistant", "content": "你好，我是智能客服，有什么问题我可以帮助你吗？"}
    ]

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

• messages：存储聊天消息的列表，每条消息是一个字典，包含 role（"user" 或 "assistant"）和 content。初始包含一条欢迎语。
• rag_service：RAG 服务实例，保存在 session_state 中避免重复创建。

渲染历史消息

for message in st.session_state["messages"]:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

st.chat_message(role) 是 Streamlit 提供的聊天消息容器组件：

参数	说明
role	消息角色。"user" 会显示用户头像和靠右对齐的消息气泡；"assistant" 会显示机器人头像和靠左对齐的消息气泡

每次脚本重新运行时，都会遍历 messages 列表重新渲染所有历史消息，这就是 Streamlit 的"重新运行"执行模型。

用户输入

if prompt := st.chat_input("请输入你的问题..."):

st.chat_input() 是 Streamlit 提供的聊天输入框组件，固定显示在页面底部。

参数	说明
placeholder	输入框的占位提示文字
返回值	用户按下回车后返回输入的文本字符串；未输入时返回 None

这里使用了 Python 3.8 引入的海象运算符 :=（赋值表达式），在 if 条件中同时完成赋值和判空。等价于：

prompt = st.chat_input("请输入你的问题...")
if prompt:
    ...

流式输出助手回复

with st.chat_message("assistant"):
    response = st.write_stream(
        st.session_state["rag_service"].chain.stream(
            {"question": prompt},
            config=config_data.session_config,
        )
    )

这里是整个问答页面的核心——流式输出（Streaming）。

chain.stream() — LangChain Runnable 的流式调用方法。与 invoke() 一次性返回完整结果不同，stream() 返回一个生成器（Generator），LLM 每生成一小段文本（token）就立即 yield 出来。

参数	说明
input	输入字典，包含 question
config	配置字典，包含 session_id 用于历史记录管理

st.write_stream() — Streamlit 的流式写入函数，接收一个生成器，逐步将内容渲染到页面上（实现逐字显示效果）。返回值是完整的拼接字符串，可用于保存到历史记录。

invoke() vs stream() 的区别：

方法	返回值	特点
chain.invoke(input)	完整结果字符串	需要等待 LLM 完全生成完毕才返回，用户看到的是"加载中…然后一次性显示全部"
chain.stream(input)	生成器 Generator[str]	每生成一个 token 就立即返回，用户看到的是"逐字逐字地出现"，体验更好

保存回复到历史

st.session_state["messages"].append({"role": "assistant", "content": response})

将 st.write_stream() 返回的完整回复追加到消息列表，下次脚本重新运行时会渲染出来。

启动方式：

streamlit run app_qa.py

---

4. 运行指南

4.1 启动知识库管理界面

用于上传本地 TXT 文件到知识库：

streamlit run app_file_uploader.py

打开浏览器访问 http://localhost:8501，上传知识文件（如 尺码推荐.txt、洗涤养护.txt）。

4.2 启动智能客服界面

streamlit run app_qa.py

打开浏览器访问 http://localhost:8501，即可开始与智能客服对话。

注意：如果两个界面需要同时运行，需要指定不同端口：

streamlit run app_file_uploader.py --server.port 8501
streamlit run app_qa.py --server.port 8502

---

5. 示例知识库数据

项目 data/ 目录下包含两个示例知识文件：

5.1 尺码推荐.txt

尺码推荐指南 
 
S码：适合身高160-165cm，体重45-55kg 
M码：适合身高165-170cm，体重55-65kg 
L码：适合身高170-175cm，体重65-75kg 
XL码：适合身高175-180cm，体重75-85kg 
XXL码：适合身高180-185cm，体重85-95kg 

注意事项： 
1. 如果身材偏胖，建议选择大一码 
2. 如果身材偏瘦，建议选择小一码 
3. 具体尺码请参考商品详情页的尺码表

5.2 洗涤养护.txt

洗涤养护说明

洗涤方法：
1. 建议手洗或机洗轻柔模式
2. 水温不超过30℃
3. 使用中性洗涤剂，避免漂白剂
4. 深浅色衣物分开洗涤

晾晒方法：
1. 避免阳光直射，阴干为宜
2. 悬挂晾晒时注意整理衣型
3. 避免使用衣架直接挂在衣领处

熨烫方法：
1. 建议低温熨烫（不超过150℃）
2. 熨烫前可在衣物上垫一块湿毛巾
3. 避免直接熨烫印花或装饰部位

保存方法：
1. 洗净晾干后折叠存放
2. 可放置干燥剂防潮
3. 避免与尖锐物品放在一起

上传这些文件后，用户即可在智能客服中询问"我身高 180cm 体重 85kg 应该穿什么尺码？"或"衣服应该怎么洗？"等问题，系统会基于知识库内容给出准确回答。

---

6. 总结

本项目通过 7 个模块的协作，实现了一个完整的 RAG 智能客服系统：

模块	职责	关键技术
config_data.py	全局配置管理	DashScope Embedding、模型参数
knowledge_base.py	知识库入库	MD5 去重、RecursiveCharacterTextSplitter、ChromaDB
app_file_uploader.py	知识上传界面	Streamlit file_uploader、session_state
vector_stores.py	向量检索	ChromaDB Retriever
file_history_store.py	聊天记录持久化	BaseChatMessageHistory、JSON 文件存储
rag.py	RAG 核心链	LCEL、RunnableWithMessageHistory、ChatPromptTemplate
app_qa.py	问答聊天界面	Streamlit chat_message、write_stream、流式输出

整体架构遵循了关注点分离原则，每个模块职责单一，便于维护和扩展。