soarli

从零手搓一个“有记忆”的专属 RAG 知识库：DeepSeek + ChromaDB + Streamlit 全栈实战
在日常的学习和工作中，我们经常会被海量的文档（学生手册、规章制度、报销指南、产品说明书等）淹没。每次遇到问题，都要...
扫描右侧二维码阅读全文
16
2026/03

从零手搓一个“有记忆”的专属 RAG 知识库：DeepSeek + ChromaDB + Streamlit 全栈实战

  1.  首页
  2. 正文  
    3. 分享到：

在日常的学习和工作中，我们经常会被海量的文档（学生手册、规章制度、报销指南、产品说明书等）淹没。每次遇到问题，都要在几十个 Word 和 PDF 中痛苦地使用 Ctrl+F 检索。

既然现在大模型这么火，我们能不能让 AI 把这些文档全部“读”一遍，然后变成一个随问随答、绝不胡编乱造、还拥有上下文记忆的专属问答机器人呢？

答案是肯定的！这就是目前 AI 领域最火的技术落地范式——RAG（检索增强生成）。

今天，我将带你从零开始，用纯 Python 打造一个属于你自己的 RAG 知识库系统。我们将使用 DeepSeek V3 作为智慧大脑，ChromaDB 作为本地向量记忆体，并用 Streamlit 构建一个高颜值的 Web 聊天界面。

🎯 我们的系统架构与核心流程

构建一个 RAG 系统，就像是给大模型外挂一个“图书馆”。主要分为以下四个阶段：

  1. 数据炼丹炉（Data Extraction）：读取各种格式的文档（TXT、MD、Word、PDF），甚至用 OCR 识别扫描件，并将长文本智能分段，交给大模型提取出高质量的 Q&A 问答对。
  2. 向量记忆体（Vectorization & Storage）：把提取出的文字转化为计算机能看懂的“多维向量”，并存入 ChromaDB 向量数据库。
  3. 推理中枢（CLI Chat & Memory）：构建对话逻辑，引入滑动窗口记忆机制，让 AI 不仅能检索知识，还能记住你们的上下文。
  4. 可视化界面（Web UI）：用 Streamlit 为这个系统披上一层漂亮的外衣，实现开箱即用的 Web 交互。

废话不多说，我们直接开始！

🛠️ 第一阶段：打造“数据炼丹炉”

要把文档喂给 AI，首先得解决两个痛点：格式兼容和暴力截断。 很多开源项目直接把文本按 1000 字一刀切，导致一句话被劈成两半，AI 提取时疯狂产生幻觉。为此，我们引入了智能语义分段策略，并加入了针对扫描版 PDF 的 OCR（光学字符识别） 支持。

1. 安装依赖

pip install pymupdf rapidocr-onnxruntime python-docx openai

2. 核心代码：智能文档解析与 QA 提取

我们编写一个脚本，它能自动遍历目录下的文件，智能分块，并调用 DeepSeek V3 将非结构化文本转化为标准的 JSON Q&A 资产。

import os
import json
import time
import glob
import re
from openai import OpenAI
import docx 
import fitz  # PyMuPDF
from rapidocr_onnxruntime import RapidOCR

# 配置你的 API (这里以硅基流动平台为例)
API_KEY = "你的_API_KEY" 
BASE_URL = "[https://api.siliconflow.cn/v1](https://api.siliconflow.cn/v1)"
MODEL_NAME = "deepseek-ai/DeepSeek-V3.2" 
MAX_CHUNK_SIZE = 1500

client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
ocr_engine = RapidOCR()

def read_file_content(filepath):
    """提取纯文本，包含针对扫描版 PDF 的 OCR 降级策略"""
    ext = os.path.splitext(filepath)[1].lower()
    text = ""
    try:
        if ext in ['.txt', '.md']:
            with open(filepath, 'r', encoding='utf-8') as f: text = f.read()
        elif ext == '.docx':
            doc = docx.Document(filepath)
            text = "\n".join([para.text for para in doc.paragraphs])
        elif ext == '.pdf':
            doc = fitz.open(filepath)
            for page_num, page in enumerate(doc):
                page_text = page.get_text().strip()
                if page_text:
                    text += page_text + "\n"
                else: # 纯图片扫描件，启动 OCR
                    print(f"  🔍 第 {page_num + 1} 页启动 OCR 识别...")
                    pix = page.get_pixmap()
                    ocr_result, _ = ocr_engine(pix.tobytes("png"))
                    if ocr_result:
                        for line in ocr_result: text += line[1] + "\n"
    except Exception as e:
        print(f"⚠️ 读取失败: {e}")
    return text

def smart_chunk_text(text, max_length=MAX_CHUNK_SIZE):
    """智能文本分段：优先按段落切分，超长段落按句子切分，避免生硬截断"""
    paragraphs = re.split(r'\n\s*\n', text)
    chunks, current_chunk = [], ""

    for para in paragraphs:
        para = para.strip()
        if not para: continue
        if len(para) > max_length: # 超长段落按句子拆分
            sentences = re.split(r'([。！？.!?])', para)
            merged_sentences = [sentences[i] + sentences[i+1] for i in range(0, len(sentences)-1, 2)]
            if len(sentences) % 2 != 0 and sentences[-1]: merged_sentences.append(sentences[-1])
            for sentence in merged_sentences:
                if len(current_chunk) + len(sentence) <= max_length:
                    current_chunk += sentence
                else:
                    if current_chunk: chunks.append(current_chunk.strip())
                    current_chunk = sentence
        else: # 正常段落组装
            if len(current_chunk) + len(para) + 2 <= max_length:
                current_chunk += para + "\n\n"
            else:
                if current_chunk: chunks.append(current_chunk.strip())
                current_chunk = para + "\n\n"
    if current_chunk.strip(): chunks.append(current_chunk.strip())
    return chunks

# ... (此处省略调用大模型生成 JSON 的 extract_qa_with_retry 函数，详见完整项目)

亮点解析：这段代码不仅支持了离线 OCR 识别，其 smart_chunk_text 函数更是灵魂所在，它确保了交给大模型的每一段话都是语义完整的，大大提高了提取的准确率。

🗄️ 第二阶段：知识入库（向量数据库）

提取好的 Q&A 还是纯文本，我们需要把它们变成数学向量，存进数据库。这里我们选用极其轻量、对新手友好的 ChromaDB，配合 BAAI/bge-m3 中文向量模型。

1. 安装 ChromaDB

pip install chromadb

2. 一键入库脚本

创建一个脚本，将目录下的所有 _qa.json 自动灌入向量数据库中。

import json, glob, os, chromadb
from chromadb.utils import embedding_functions

API_KEY = "你的_API_KEY" 
BASE_URL = "[https://api.siliconflow.cn/v1](https://api.siliconflow.cn/v1)"
EMBEDDING_MODEL = "BAAI/bge-m3" 
DB_PATH = "./chroma_db" 
COLLECTION_NAME = "campus_knowledge_base"

def main():
    # 初始化 ChromaDB 和 Embedding 模型
    client = chromadb.PersistentClient(path=DB_PATH)
    custom_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=API_KEY, api_base=BASE_URL, model_name=EMBEDDING_MODEL
    )
    collection = client.get_or_create_collection(name=COLLECTION_NAME, embedding_function=custom_ef)
    
    # 遍历 JSON 入库
    json_files = glob.glob(os.path.join(".", "*_qa.json"))
    for filepath in json_files:
        with open(filepath, 'r', encoding='utf-8') as f:
            qa_data = json.load(f)
            documents, metadatas, ids = [], [], []
            
            for index, item in enumerate(qa_data):
                question, answer = item.get("question", ""), item.get("answer", "")
                if not question or not answer: continue
                
                documents.append(f"问题: {question}\n答案: {answer}")
                metadatas.append({"source": os.path.basename(filepath), "question": question, "answer": answer})
                ids.append(f"{os.path.basename(filepath)}_qa_{index}")
                
            if documents:
                collection.add(documents=documents, metadatas=metadatas, ids=ids)
    print("🎉 入库完成！知识已沉淀至 ChromaDB。")

运行完毕后，你的项目目录下会多出一个 chroma_db 文件夹，这就是我们外挂的“记忆芯片”。

💻 第三阶段：从终端对话到赋予“记忆”

很多基础的 RAG 教程到这里就结束了，导致做出来的机器人像个“失忆症患者”，每次只能问孤立的问题。 我们通过引入滑动窗口记忆机制（Sliding Window Memory），让它拥有多轮对话的能力！

核心思路是：每次提问时，不仅带着从数据库检索出来的 Context，还要把最近 N 轮的聊天记录一并打包发给大模型。

(注：终端版代码由于篇幅原因，直接整合进了最终的 Web 端代码中。核心逻辑完全一致。)

🚀 第四阶段：Streamlit Web 界面大放送

告别黑乎乎的命令行，我们只需 50 行左右的代码，就能用 Streamlit 渲染出一个媲美 ChatGPT 网页版的对话界面，支持流式打字输出、侧边栏管理和记忆清空。

1. 安装 Streamlit

pip install streamlit

2. 核心 UI 代码 (web_chat.py)

import streamlit as st
from openai import OpenAI
import chromadb
from chromadb.utils import embedding_functions

# 配置与初始化
API_KEY = "你的_API_KEY" 
BASE_URL = "[https://api.siliconflow.cn/v1](https://api.siliconflow.cn/v1)"
MODEL_NAME = "deepseek-ai/DeepSeek-V3.2" 
EMBEDDING_MODEL = "BAAI/bge-m3" 
MAX_HISTORY_TURNS = 3 

st.set_page_config(page_title="RAG 知识助手", page_icon="🎓")
st.title("🎓 专属 RAG 知识库问答助手")

# 缓存资源，防止每次交互重新连接数据库
@st.cache_resource
def init_services():
    llm = OpenAI(api_key=API_KEY, base_url=BASE_URL)
    db_client = chromadb.PersistentClient(path="./chroma_db")
    custom_ef = embedding_functions.OpenAIEmbeddingFunction(
        api_key=API_KEY, api_base=BASE_URL, model_name=EMBEDDING_MODEL
    )
    collection = db_client.get_collection(name="campus_knowledge_base", embedding_function=custom_ef)
    return llm, collection

llm_client, collection = init_services()

# Session State 管理记忆
if "messages" not in st.session_state:
    st.session_state.messages = []

# 侧边栏
with st.sidebar:
    st.header("⚙️ 控制面板")
    if st.button("🧹 清空对话记忆"):
        st.session_state.messages = []
        st.rerun()

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

# 处理用户输入
if user_query := st.chat_input("请输入您的问题..."):
    with st.chat_message("user"): st.markdown(user_query)
    
    # 向量检索
    results = collection.query(query_texts=[user_query], n_results=3)
    context, sources = "", set()
    if results['documents'] and results['documents'][0]:
        for doc, meta in zip(results['documents'][0], results['metadatas'][0]):
            context += f"{doc}\n---\n"
            sources.add(meta.get('source', '未知文件'))

    # 构建 Prompt 与记忆
    system_prompt = f"""
    严格基于以下参考信息回答问题。如果找不到答案，请回答“抱歉，参考资料中未提及”，绝不编造。
    【参考信息】:\n{context}
    """
    messages_to_send = [{"role": "system", "content": system_prompt}]
    
    # 截取最近历史记忆（滑动窗口）并清理溯源干扰
    history_to_keep = st.session_state.messages[-(MAX_HISTORY_TURNS * 2):]
    cleaned_history = [{"role": m["role"], "content": m["content"].split("\n\n*📚")[0]} for m in history_to_keep]
    messages_to_send.extend(cleaned_history)
    messages_to_send.append({"role": "user", "content": user_query})

    # 流式输出
    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        response = llm_client.chat.completions.create(
            model=MODEL_NAME, messages=messages_to_send, temperature=0.3, stream=True
        )
        for chunk in response:
            if chunk.choices and chunk.choices[0].delta.content:
                full_response += chunk.choices[0].delta.content
                message_placeholder.markdown(full_response + "▌")
        
        if sources: full_response += f"\n\n*📚 参考来源: {', '.join(sources)}*"
        message_placeholder.markdown(full_response)
            
    # 存入记忆
    st.session_state.messages.append({"role": "user", "content": user_query})
    st.session_state.messages.append({"role": "assistant", "content": full_response})

3. 一键起飞

在终端输入以下命令：

streamlit run web_chat.py

浏览器会自动弹出页面。尽情向你的专属 AI 提问吧！它不仅能引经据典，还能在回答底部贴心地附上信息来源（Source），并能听懂你的上下文追问。

💡 总结与展望

到这里，我们就完整实现了一个包含多格式文档解析、智能分段、OCR识别、向量检索、多轮对话记忆以及流式 Web 界面的企业级 RAG 原型系统。

通过这个项目，你掌握了当下大语言模型应用开发中最核心的三板斧：Prompt Engineering（提示词工程）、Embedding（向量化）与 RAG（检索增强）。

如果你想更进一步，可以将这个 Streamlit 应用推送到 GitHub，并通过 Streamlit Community Cloud 免费部署到公网，只需一键发布，就能生成一个任何人都能访问的网页链接，作为你的硬核求职项目展示！

如果你在复现过程中遇到任何问题，欢迎在评论区留言讨论！如果觉得这篇教程对你有帮助，别忘了点赞和收藏哦~ ⭐

版权属于：soarli

本文链接：https://blog.soarli.top/archives/914.html

转载时须注明出处及本声明。

最后修改：2026 年 03 月 16 日 05 : 16 AM
© 著作权归作者所有

发表评论