收藏这篇就够了！LangChain+ RAG从0到1搭建智能客服系统

溏心富贵虾

469人浏览 · 2026-04-27 18:27:25

溏心富贵虾 · 2026-04-27 18:27:25 发布

文章目录

概要

学会本文，你将完整掌握 RAG 项目全链路开发，独立实现私有知识库构建、向量检索、多轮会话、流式问答与混合回答模式,已整理了完整可直接运行的全套源码.

对比维度	直接使用通用大模型（豆包 / Deepseek等）	LangChain + RAG 检索增强方案
私有业务适配	无本地知识库，无法读取企业内部、行业专属资料，只能依靠训练通识数据	支持私有文档向量化入库，定制专属知识库，精准匹配业务场景
回答真实性	易产生AI 幻觉，无依据编造内容，答案不可控、不可溯源	回答依赖检索到的真实文档内容，有据可依，从源头抑制幻觉
内容精准度	回答泛化笼统，无法贴合细分业务规则、产品资料	定向检索关联片段，答案针对性强，贴合实际业务需求
知识时效性	模型训练数据固定，无法同步最新业务文件、更新资料	支持文档随时新增、替换、更新，知识库可动态迭代
开发落地成本	仅能单轮简单问答，无法私有化部署，无定制化能力	模块化全流程编排，易私有化部署、二次开发与功能拓展
上下文与工程化	缺少定制会话管理、文档处理、向量存储等配套能力	集成文本切割、MD5 去重、会话记忆、流式输出等完整工程能力

系统实现两大核心流程：

离线流程：文档上传 → MD5 去重 → 文本分块 → 向量化 → 存入向量库
在线流程：用户提问 → 向量检索 → 提示词组装 → 大模型生成 → 流式输出

系统支持混合问答（知识库内外自动切换）、流式打字机效果、会话历史保存，可直接用于企业级私有知识库、智能客服等场景。

整体架构流程

总架构

离线流程（知识库构建）

在线流程（智能问答）

技术名词解释

LangChain：RAG 全流程编排工具，负责文档加载、分块、检索、Prompt 组装、大模型调用等全链路管理。
RAG：检索增强生成（项目核心架构），通过 “检索私有知识库 + 大模型生成”，解决大模型幻觉问题。
Chroma：轻量级向量数据库，用于存储文档向量化后的向量数据，支持快速相似度检索。
Streamlit：Web 交互界面框架，用于快速搭建项目可视化界面，实现文档上传、问答交互等功能。
Embedding 模型：负责将文本转化为向量，为向量检索提供支撑。
大模型：Qwen/Deepseek/GPT 等（本章所使用大模型为Qwen3-max,相关的 API_KEY 已配置环境变量，代码中不体现）。
MD5：用于文档内容去重，避免重复文档多次存入向量数据库，提升数据库效率。
流式输出：实现 “打字机效果”，让大模型回答逐字逐句显示，优化用户交互体验。

技术细节

文件名	核心作用
`app_file_upload.py`	知识库更新主程序（Streamlit 界面），负责文档上传与入库流程
`app_qa.py`	项目对话主程序（Streamlit 界面），启动问答 Web 页面，处理用户交互
`config_data.py`	全局配置文件，统一管理项目参数
`file_history_store.py`	长期会话记忆存储服务，负责用户对话历史的读写
`knowledge_base.py`	知识库更新服务，实现文档去重、分块、元数据处理
`rag.py`	RAG 核心服务，负责检索、提示词组装与大模型调用
`vector_stores.py`	向量存储服务，封装向量数据库的写入、检索等底层操作

MD5 去重

为什么要去重?

防止相同 / 重复的文档多次存入向量数据库，造成数据库冗余、检索效率降低

（核心原理：只要文档内容一致，其 MD5 字符串就完全相同）

def check_md5(md5_str: str):
    """
    检查传入的md5字符串是否已经被处理过了
    :param md5_str:
    :return: False:未处理过 True:处理过
    """
    if not os.path.exists(config.md5_path):
        #if进入,表示文件不存在,那肯定没有处理过这个Md5
        open(config.md5_path,'w',encoding='utf-8').close()
        return False
    else:
        for line in open(config.md5_path,'r',encoding='utf-8').readlines():
            line = line.strip()  #处理字符串前后的空格和回车
            if line == md5_str:
                return True  #已处理过
        return False

MD5去重代码实现

st.session_state

st.session_state 是 Streamlit 框架专用的会话状态存储器，可理解为：每个用户打开网页后，独有的一块 “临时内存 / 小仓库”。

其核心作用是保存页面数据、避免刷新丢失、实现跨模块数据共享，是保证 RAG 系统正常运行、不重复初始化、不丢失对话数据的核心机制。

if "message" not in st.session_state:
    st.session_state["message"] = [
        {
            "role":"assistant","context":"你好,有什么可以帮助你?"
        }
    ]

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["context"])

session_state代码实现

文档分割

作用:将长文档按规则切分成短文本片段（Chunk），让向量检索更精准、大模型理解更高效。

基础用法:

固定长度分块

     self.spliter = RecursiveCharacterTextSplitter(
            chunk_size=config.chunk_size,  #分割后的文本段最大长度
            chunk_overlap=config.chunk_overlap,  #分割后的文本段之间允许重叠的长度
            separators=config.separators,  #文本自然段落分隔的依据符号
            length_function=len,    #使用Python自带的len函数统计长度
        ) #文本分割器的对象

# spliter
chunk_size = 1000
chunk_overlap = 100
separators = ["\n\n", "\n", ".", "!", "?", "。", "！", "？", " ", ""]
max_split_char_number = 1000        # 文本分割的阈值

文本分割以及配置项

进阶用法:

1. 语义分块（推荐进阶方案）

按语义完整性切割，不破坏句子、段落逻辑，检索更准。

优先保证：一个块 = 一个完整意思
适用：商品说明、合同、文章、知识库

2. 结构分块

按文档自身结构切割：

标题 → 段落 → 子段落
目录层级切割
适用：PDF、说明书、论文、多章节文档

3. 递归分块（RecursiveCharacterTextSplitter）

LangChain 标准高级用法：

先按大分隔符（换行、段落）分
再按小分隔符（句子）切
最后保证块不超限最均衡、最常用。

4. 按专业场景优化

电商商品：按 SKU、属性、规格分块
客服知识库：按问题 - 答案分块
代码文档：按函数、类、注释分块

会话历史保存

实现多轮对话系统能记住上一轮对话内容，回答更连贯、更贴合上下文。
提升用户体验用户无需重复描述问题，交互更自然。
便于追溯与优化保存对话记录，可用于后续客服质检、问题分析、模型优化。
业务闭环支持对话导出、复盘、用户行为分析。

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

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)

会话历史保存功能实现

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

session_id可在配置项中配置

流式输出

什么是流式输出:
write_stream 是 Streamlit 框架提供的流式输出方法，配合大模型的流式返回，实现逐字、逐词、逐句的 “打字机效果”，让回答像真人一样逐步显示。
核心作用:
提升交互体验不用等大模型全部生成完才显示，用户看到文字逐字输出，等待感更低。
降低感知延迟大模型一返回内容就立刻展示，响应更快、更流畅。
贴近商用产品抖音、GPT、文心一言等产品都用这种流式输出。
配合会话保存一边流式输出，一边完整保存回答内容。
关键点:
必须使用yield关键字
注:以上问题与答案的模式就是类似流式输出,包括日常使用的豆包,deepseek等,默认都是采用流式输出

        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))

流式输出代码

如果不加yiled呢?

def capture(generator, cache_list):
    for chunk in generator:
        cache_list.append(chunk)
        # 没有 yield，这个函数就变成了普通函数
        # 它会把所有 chunk 都收完，最后返回 None

你可能会问：“我直接把 res_strem 传给 write_stream 不就行了吗？为什么要包一层 capture？”

这是因为有两个需求同时存在：

需求 A（给用户看）：需要实时流式显示（由 write_stream 处理）。
需求 B（给自己存）：需要把完整的回答存进 st.session_state 做历史记录（由 cache_list 处理）。

capture 函数的作用就是：

通过 yield chunk：满足需求 A，把数据吐给页面显示。
通过 cache_list.append(chunk)：满足需求 B，把数据悄悄存进列表。

混合模式

什么是混合模式?

私有知识库回答+大模型通用回答自动切换

系统自动判断问题是否在知识库范围内：

有答案 → 按私有资料专业回答

无答案 → 用大模型自身知识正常聊天

     self.prompt_template = ChatPromptTemplate.from_messages(
            [
                ("system", "你是一个智能助手。请根据以下【参考资料】回答用户问题。\n"
                           "如果【参考资料】中包含相关信息，请严格依据资料进行简洁、专业的回答。\n"
                           "如果【参考资料】为空或与问题无关，请利用你自身的通用知识来回答用户的问题。\n"
                           "参考资料: {context}"),
                ("user", "用户提问: {input}")
            ]

        )

   def format_document(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

混合模式代码实现