本地 RAG 智能助手 - 技术方案文档

场景 ：技术方案 | 受众 ：后端工程师、架构师、开发者 | 字数 ：约5000字

背景与目标

业务需求

本项目是一个基于 Streamlit 的本地知识库 RAG（Retrieval-Augmented Generation）智能助手，核心业务需求如下：

技术指标

核心价值

• 数据隐私保护 ：文档在本地处理，不传输到外部服务器
• 离线可用 ：支持本地哈希向量和 echo 模式，无需网络也能运行演示
• 配置灵活 ：所有关键参数可通过配置文件调整
• 易于扩展 ：模块化设计，便于替换组件

现状与约束

技术栈现状

约束条件

已有系统分析

项目采用模块化架构，各组件职责清晰：

• 数据层 ： data/knowledge_base/ 存储原始文档， data/vector_store/ 存储向量索引
• 核心层 ： src/rag_assistant/ 包含所有业务逻辑
• 展示层 ： app.py 提供 Streamlit 交互界面
• 配置层 ： config/app.yaml 集中管理所有参数

.
├── app.py                    # Streamlit 交互界面
├── config/
│   └── app.yaml              # 应用配置文件
├── data/
│   ├── knowledge_base/       # 原始文档存储
│   └── vector_store/         # 向量索引存储
├── src/
│   └── rag_assistant/        # 核心业务逻辑
│       ├── config.py         # 配置加载
│       ├── document_loader.py # 文档解析
│       ├── embedding.py      # 向量化模型
│       ├── llm.py            # 大模型客户端
│       ├── pipeline.py       # RAG 管线编排
│       ├── schema.py         # 数据模型定义
│       ├── text_splitter.py  # 文本切分
│       └── vector_store.py   # 向量存储与检索
├── requirements.txt          # 核心依赖
├── requirements-optional.txt # 可选依赖
└── README.md                 # 项目说明

方案概述

核心架构

┌─────────────────────────────────────────────────────────────────┐
│                        Streamlit UI                             │
│  （文件上传、问题输入、结果展示）                                  │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│                       RAGPipeline                               │
│  (编排索引构建、检索、回答生成流程)                                │
└────────────────────────────┬────────────────────────────────────┘
        │           │           │           │           │
        ▼           ▼           ▼           ▼           ▼
┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐
│Document   │ │Text       │ │Embedding  │ │Vector     │ │LLM        │
│Loader     │ │Splitter   │ │Model      │ │Store      │ │Client     │
│(文档解析)  │ │(文本切分) │ │(向量化)    │ │(向量检索)  │ │(回答生成)  │
└───────────┘ └───────────┘ └───────────┘ └───────────┘ └───────────┘
        │           │           │           │           │
        └───────────┴───────────┴───────────┴───────────┘
                             │
                     ┌───────▼───────┐
                     │   data/       │
                     │ (知识库+向量)  │
                     └───────────────┘

核心思路

索引构建流程

1. 用户上传文档 ——> 保存到 knowledge_base 文件夹
2. DocumentLoader 解析文档 ——> 生成 Document 对象列表
3. TextSplitter 切分文本 ——> 生成带重叠的文本片段
4. EmbeddingModel 向量化 ——> 生成向量矩阵
5. VectorStore 构建索引 ——> 持久化到 vector_store 文件夹

关键设计原则

详细设计

数据模型设计

配置类（Schema）

# 定义位置：src/rag_assistant/schema.py

@dataclass(slots=True)
class Settings:
    app: AppSettings          # 应用配置
    documents: DocumentSettings  # 文档配置
    splitter: SplitterSettings   # 切分配置
    embedding: EmbeddingSettings # 向量化配置
    retrieval: RetrievalSettings # 检索配置
    llm: LLMSettings          # 大模型配置

字段说明 ：

业务数据类

# 定义位置：src/rag_assistant/schema.
py

@dataclass(slots=True)
class Document:
    text: str              # 文档内容
    metadata: dict[str, Any] = field (default_factory=dict)  # 元信息（来源、页码等）

@dataclass(slots=True)
class SearchResult:
    text: str              # 检索到的文本片段
    metadata: dict[str, Any]  # 来源元信息
    score: float           # 相似度分数

@dataclass(slots=True)
class IndexBuildResult:
    document_count: int    # 处理的文档数
    chunk_count: int       # 生成的片段数

@dataclass(slots=True)
class Answer:
    answer: str            # 生成的回答
    sources: list[SearchResult]  # 引用来源列表

模块设计

DocumentLoader（文档加载器）

职责 ：将不同格式的文档解析为统一的 Document 对象

支持格式 ：

核心方法 ：

def load_directory(self, directory: Path) -> list[Document]
# 输入：目录路径
# 输出：Document 列表
# 遍历目录下所有允许的文件，逐一解析

def load_file(self, path: Path) -> list[Document]
# 输入：文件路径
# 输出：Document 列表（可能多个，如多页PDF）
# 根据扩展名选择对应的解析方法

元信息生成 （ _meta 函数）：

代码位置 ： src/rag_assistant/document_loader.py

TextSplitter（文本切分器）

职责 ：按字符长度切分文本，保留重叠区域减少语义断裂

核心参数 ：

约束条件 ： chunk_overlap < chunk_size （在 init 中检查）

切分算法 ：

原始文本：[===============================] (长度 N)
切分结果：
块0: [===========]           位置: 0 ~ chunk_size
块1:      [===========]       位置: chunk_size-overlap ~ 2*chunk_size-overlap
块2:            [===========]   位置: 2*chunk_size-2*overlap ~ 3*chunk_size-2*overlap

文本归一化 （ _normalize_text 函数）：

• 压缩多余空白
• 移除空行
• 每行前后去空格

代码位置 ： src/rag_assistant/text_splitter.py

EmbeddingModel（向量化模型）

职责 ：将文本转换为向量表示，支持真实模型和降级方案

设计模式 ：延迟加载 + 降级机制

核心参数 ：

属性 ：

@property
    def dimension(self) -> int:
        self._ensure_model()
        if self._model is not None:
            return int(self._model.get_sentence_embedding_dimension())
        return self.fallback_dimension

核心方法 ：

    def encode(self, texts: list[str]) -> np.ndarray:
        self._ensure_model()
        if not texts:
            return np.empty((0, self.dimension), dtype="float32")

        if self._model is not None:
            vectors = self._model.encode(texts, normalize_embeddings=True)
            return np.asarray(vectors, dtype="float32")

        vectors = np.vstack([self._hash_embedding(text) for text in texts]).astype("float32")
        norms = np.linalg.norm(vectors, axis=1, keepdims=True)
        return vectors / np.clip(norms, 1e-12, None)

降级方案 （ _hash_embedding ）：

当无法加载真实模型时，使用 SHA-256 哈希生成伪向量：

    def _hash_embedding(self, text: str) -> np.ndarray:
        """无需下载模型的兜底方案，便于项目首次运行和离线演示。"""
        vector = np.zeros(self.fallback_dimension, dtype="float32")
        for token in _tokenize(text):
            digest = hashlib.sha256(token.encode("utf-8")).digest()
            index = int.from_bytes(digest[:4], "little") % self.fallback_dimension
            sign = 1.0 if digest[4] % 2 == 0 else -1.0
            vector[index] += sign
        return vector

分词策略 （ _tokenize ）：

• 中文：按字分词
• 英文：按词分词（字母数字序列）
• 统一转为小写

代码位置 ： src/rag_assistant/embedding.py

VectorStore（向量存储）

职责 ：负责向量索引的内存检索和本地持久化

存储结构 ：

核心方法 ：

检索算法 ：使用点积计算余弦相似度（向量已归一化）

代码位置 ： src/rag_assistant/vector_store.py

LLMClient（大模型客户端）

职责 ：封装大模型调用，支持多种提供者和降级模式

支持的提供者 ：

核心方法 ：

def generate(self, question: str, contexts: list[SearchResult]) -> str
# 输入：问题、检索到的上下文片段
# 输出：生成的回答文本

Prompt 构建 ：

System Prompt （ _system_prompt ）：

"你是一个严谨的中文 RAG 智能助手。必须优
先依据给定的本地知识库片段回答。如果资料不
足，请明确说明不足，不要编造。回答要结构清
晰，并在适当位置提及来源。"

User Prompt （ _build_user_prompt ）：

问题：{question}

本地知识库片段：
[片段 1 | 来源：xxx | 相似度：0.85]
文本内容...

[片段 2 | 来源：yyy | 相似度：0.72]
文本内容...

降级模式逻辑 ：

if provider == "local_echo" 或 未配置 API Key:
    return _local_answer(question, 
    contexts)
else:
    调用 OpenAI API

代码位置 ： src/rag_assistant/llm.py

RAGPipeline（RAG 管线）

职责 ：编排各组件，提供统一的索引构建和问答接口

组件组合 ：

self.loader = DocumentLoader(settings.documents.allowed_extensions)
self.splitter = TextSplitter(settings.splitter.chunk_size, settings.splitter.chunk_overlap)
self.embedding_model = EmbeddingModel(...)
self.vector_store = VectorStore(settings.app.vector_store_dir)
self.llm = LLMClient(settings.llm)

核心方法 ：

def rebuild_index(self) -> IndexBuildResult
# 流程：加载文档 → 切分 → 向量化 → 构建索引 → 保存
# 输出：处理统计信息

def load_index(self) -> bool
# 从本地加载已构建的索引

def answer(self, question: str) -> Answer
# 流程：问题向量化 → 检索 → 生成回答
# 输出：Answer 对象（含引用来源）

代码位置 ： src/rag_assistant/pipeline.py

Streamlit 应用层

职责 ：提供 Web 交互界面，处理用户输入输出

页面结构 ：

会话状态管理 ：

if "messages" not in st.session_state:
    st.session_state.messages = []  
    # 存储对话历史

缓存策略 ：

@st.cache_resource
def get_pipeline(config_path: str) -> RAGPipeline:
# 缓存管线实例，避免重复初始化

代码位置 ： app.py

技术选型对比

Embedding 模型选型

大模型选型

向量存储选型

文档解析选型

风险与应对

风险识别与缓解

降级机制设计

Embedding 降级流程 ：

用户启动应用
    │
    ▼
尝试加载 SentenceTransformers 模型
    │
    ├─ 成功 → 使用真实向量
    │
    └─ 失败 → 使用哈希向量（提示用户质量受限）

LLM 降级流程 ：

用户提问
    │
    ▼
检查 provider 和 API Key
    │
    ├─ provider="local_echo" → 返回检索摘要
    │
    ├─ API Key 有效 → 调用大模型生成
    │
    └─ API Key 无效 → 返回检索摘要（提示未配置）

测试要点

单元测试用例

集成测试用例

性能测试

部署与运维建议

环境要求

安装步骤

# 1. 创建虚拟环境
python -m venv .venv

# 2. 激活虚拟环境（Windows）
.venv\Scripts\activate

# 3. 安装核心依赖
pip install -r requirements.txt

# 4. （可选）安装高质量向量模型
pip install -r requirements-optional.txt

# 5. 配置环境变量（如需使用大模型）
set OPENAI_API_KEY=your-api-key

# 6. 启动应用
streamlit run app.py

配置说明

配置文件位置 ： config/app.yaml

关键配置项说明 ：

运维建议

日志管理 ：

• Streamlit 默认输出到控制台

• 生产环境可配置日志文件 数据备份 ：

• 定期备份 data/knowledge_base/ 和 data/vector_store/

• 建议使用版本控制管理配置文件 性能优化 ：

• 对于大规模知识库，考虑引入 FAISS 索引

• 定期清理过期文档并重建索引 安全建议 ：

• API Key 通过环境变量传递，不要硬编码

• 限制知识库目录的访问权限

• 定期更新依赖版本