上一篇介绍了 ORIGIN AI Workspace 的整体功能，本文聚焦其中最复杂的子系统——RAG 知识库管道，逐层拆解它的架构设计、关键代码与工程权衡。

图表说明：文中包含 5 张 Mermaid 技术图表（Pipeline 架构图、分块策略示意、向量库选型决策树、检索时序图、批量嵌入性能对比）。CSDN 若不支持 Mermaid 渲染，可将代码块内容粘贴到 mermaid.live 导出为 PNG/SVG 后上传。

---

为什么 RAG 是自托管 AI 的「杀手功能」？

一个只能聊天的 AI 工具，和 ChatGPT 没有本质区别。真正让私有 AI 工作站产生壁垒的，是它能理解你的数据。

RAG（Retrieval-Augmented Generation，检索增强生成）让 AI 在回答前先从你的文档中检索相关信息，再结合检索结果生成回答。这意味着：

• 你的代码库、技术文档、笔记不再是"静态文件"，而是 AI 能查询的知识来源
• 不需要微调模型，上传文档即可让 AI 获得领域知识
• 数据始终在你自己的服务器上，不会上传到第三方

ORIGIN 的 RAG Pipeline 从 v0.4 开始引入，设计上追求务实——不引入额外的向量数据库，不依赖云服务，一切在 Docker 里自闭环。

---

一、全链路架构

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│  文件    │ →  │  文本    │ →  │  文档    │ →  │  向量    │ →  │ pgvector │ →  │  注入    │
│  上传    │    │  提取    │    │  分块    │    │  嵌入    │    │  存储    │    │ AI 对话  │
└──────────┘    └──────────┘    └──────────┘    └──────────┘    └──────────┘    └──────────┘

六个阶段各司其职，每一层都有明确的输入/输出契约。这种线性 Pipeline 的好处是：每个阶段可以独立测试、独立替换。比如你想换一个嵌入模型，只需改嵌入层，其他五层完全不感知。

上图展示了 RAG 管道的六个阶段及其数据流向。每个阶段的输出是下一个阶段的输入，形成了一个清晰的 ETL（Extract-Transform-Load）模式。图渲染工具：mermaid.live

---

二、文件上传与文本提取

2.1 上传安全

文件上传是所有后续处理的入口，也是安全攻击面最大的环节。ORIGIN 做了三层防护：

# 第一层：文件大小限制（默认 50MB）
MAX_UPLOAD_SIZE = 50 * 1024 * 1024  # 50MB

# 第二层：扩展名白名单
ALLOWED_EXTENSIONS = {
    ".txt", ".md", ".pdf", ".csv",
    ".py", ".ts", ".tsx", ".js", ".jsx",
    ".json", ".yaml", ".yml", ".toml",
    ".html", ".css", ".sql",
}

# 第三层：MIME 类型校验
# 不仅检查扩展名，还读取文件头魔数验证真实类型

为什么不用黑名单而用白名单？黑名单永远追不上攻击者。白名单的思想是"只放行你明确需要的"，攻击面从无限收敛到可控范围。

2.2 文本提取策略

不同文件类型走不同的提取路径：

文件类型	提取方式	说明
.txt, .md, 代码文件	直接读取 UTF-8	O(1) 开销
.pdf	PyMuPDF (fitz)	支持中文、表格、多栏布局
.csv	pandas 读取后格式化	保留列名和数据结构
图片（OCR 预留）	Tesseract / PaddleOCR	v0.4 适配中

PDF 提取选 PyMuPDF 而非 PyPDF2，是因为 PyMuPDF 对中文 PDF 的支持好得多，而且速度快 4-5 倍——这在对 100+ 页技术文档建立知识库时差异明显。

---

三、文档分块：RAG Pipeline 中最被低估的环节

3.1 为什么要分块？

嵌入模型（如 text-embedding-3-small）有输入长度限制（8191 tokens），你不能把一本 200 页的书直接丢进去。更重要的是，检索精度与块大小强相关：

• 块太大 → 包含太多无关信息，检索精度下降
• 块太小 → 语义碎片化，缺少上下文
• 没有重叠 → 关键信息可能正好落在两个块的边界上

重叠窗口机制示意：相邻 Chunk 共享 160 字符，确保跨 Chunk 边界的语义不会被切断。

3.2 ORIGIN 的分块策略

class TextChunker:
    def __init__(self, chunk_size: int = 1200, overlap: int = 160) -> None:
        self.chunk_size = chunk_size
        self.overlap = overlap

    def split(self, text: str) -> list[str]:
        # Step 1: 清洗——合并多余空行，统一换行符
        cleaned = self._clean_text(text)

        # Step 2: 按自然段落边界预分割
        paragraphs = cleaned.split("\n\n")

        # Step 3: 合并短段落直到接近 chunk_size
        chunks = []
        current = ""
        for para in paragraphs:
            if len(current) + len(para) < self.chunk_size:
                current += para + "\n\n"
            else:
                if current:
                    chunks.append(current.strip())
                current = para + "\n\n"
        if current:
            chunks.append(current.strip())

        # Step 4: 滑动窗口 overlap——相邻块共享 160 字符
        overlapped = []
        for i, chunk in enumerate(chunks):
            if i > 0:
                # 前一块的尾部作为当前块的前缀
                prefix = chunks[i-1][-self.overlap:]
                overlapped.append(prefix + chunk)
            else:
                overlapped.append(chunk)

        return overlapped

3.3 参数选择背后的考量

chunk_size = 1200：

这个数字不是拍脑袋定的。中英文混合的技术文档，1200 字符约等于 600-800 个 token（中文约 2 字符/token，英文约 4 字符/token）。这个大小刚好能容纳一个完整的技术概念——比如一段 API 文档（函数签名 + 参数说明 + 示例代码），而不至于塞进多个无关话题。

overlap = 160：

160 字符的重叠窗口能覆盖 2-3 句话。假设你问「FastAPI 的依赖注入怎么用？」，相关描述可能跨越两个 chunk 的边界。如果 overlap=0 或太小，前一个 chunk 的尾部和后一个 chunk 的头部都可能漏掉关键上下文。160 字符的冗余让检索系统有"二次机会"。

3.4 为什么不做语义分块？

你可能会问：为什么不按语义边界切分（比如用 NLP 模型判断句子边界）？

答案很务实：性价比。语义分块需要额外的模型推理（如 spaCy 句子分割器），在初次索引 100 篇文档时，这段等待时间用户能明显感知。ORIGIN 的策略是用段落边界 + 滑动窗口这种 O(1) 的方式逼近语义分块的效果，同时保持毫秒级的分块速度。

---

四、向量嵌入：模型选择与工程细节

4.1 为什么选 text-embedding-3-small

OpenAI 的嵌入模型家族目前有三个选择：

模型	维度	价格	适用场景
text-embedding-3-small	1536	$0.02/1M tokens	通用语义检索
text-embedding-3-large	3072	$0.13/1M tokens	高精度专业检索
text-embedding-ada-002	1536	$0.10/1M tokens	旧版，不推荐

ORIGIN 选 text-embedding-3-small 的理由：

1. 性价比极致：比 large 便宜 6.5 倍，比 ada-002 便宜 5 倍
2. 1536 维恰到好处：pgvector 的 IVFFlat 索引在 1536 维下表现最好，超过 2000 维索引构建时间会显著增加
3. 中文支持好：实际测试中，对小段中文技术文本的检索精度与 large 差距在 5% 以内

4.2 本地开发降级方案

生产环境走 OpenAI API 做嵌入，但本地开发时如果没配 API Key，总不能每次改一行代码都要调远程 API。ORIGIN 预留了本地 Hash 嵌入方案：

def get_embedding_provider(settings: Settings) -> EmbeddingProvider:
    if settings.openai_api_key:
        return OpenAIEmbeddingProvider(
            api_key=settings.openai_api_key,
            model="text-embedding-3-small",
        )
    else:
        logger.warning("No OpenAI API key found, using local hash embeddings (dev only)")
        return LocalHashEmbeddingProvider(dimensions=1536)

Hash 嵌入是一个确定性函数：输入相同的文本，永远输出相同的向量。它不携带语义信息——"猫"和"小猫"的向量完全不相关——但足够验证 Pipeline 的连通性。这种设计让前后端联调不被 API Key 阻塞。

---

五、pgvector：为什么不需要独立的向量数据库？

5.1 选型分析

向量数据库市场已经很拥挤了：

方案	类型	优势	劣势
Pinecone	云服务	全托管、性能好	贵、数据不在本地
Milvus	独立部署	十亿级向量、GPU 加速	运维重、吃资源
Weaviate	独立部署	多模态、GraphQL	学习成本高
Chroma	嵌入式	轻量、适合原型	生产稳定性待验证
pgvector	PG 扩展	零额外运维、ACID	十亿级以上有瓶颈

ORIGIN 的目标用户是个人或小团队，文档量在几百到几千篇的级别。pgvector 可以在同一个 PostgreSQL 实例里同时管理业务数据和向量数据，不需要额外维护一个向量数据库服务。

向量数据库选型决策树。ORIGIN 的场景命中 pgvector 的最优区间：已有 PG + 万级数据 + 生产部署。

这意味着：

• Docker Compose 里少一个容器
• 备份恢复一个 pg_dump 搞定
• 事务一致性天然保证——文档删了，对应的向量也一起删，不会出现孤儿向量

5.2 索引选择

pgvector 支持两种近似索引：

-- IVFFlat：基于 IVF（倒排文件）的近似检索
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- HNSW：基于分层导航小世界图的近似检索
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

ORIGIN 默认使用 IVFFlat。HNSW 虽然查询更快，但构建索引时内存占用大，不适合 ORIGIN 以 NAS/低配服务器为主的目标场景。对于万级别向量，IVFFlat 的查询延迟通常在 10-50ms，完全够用。

---

六、语义检索与结果注入

6.1 检索流程

async def retrieve_context(
    query: str,
    knowledge_base_id: int,
    top_k: int = 5,
    similarity_threshold: float = 0.7,
) -> list[RetrievedChunk]:
    # Step 1: 将用户查询转为向量
    query_embedding = await embedding_provider.embed(query)

    # Step 2: pgvector 余弦相似度检索
    results = await db.execute(
        select(DocumentChunk, cosine_distance(DocumentChunk.embedding, query_embedding))
        .where(DocumentChunk.knowledge_base_id == knowledge_base_id)
        .order_by(cosine_distance(DocumentChunk.embedding, query_embedding))
        .limit(top_k)
    )

    # Step 3: 过滤低相关度结果
    chunks = []
    for chunk, distance in results:
        similarity = 1 - distance
        if similarity >= similarity_threshold:
            chunks.append(RetrievedChunk(
                content=chunk.content,
                source=chunk.filename,
                similarity=similarity,
            ))

    return chunks

检索与注入的完整时序图。注意 similarity_threshold 过滤掉了低相关度的 chunk3，避免噪音干扰 AI 回答。

这里有一个容易被忽略的细节：similarity_threshold = 0.7。

如果用户问「RAG Pipeline 的分块策略是什么？」，但知识库里只有 Python 入门教程，检索结果的相关度可能只有 0.3-0.5。如果不设阈值，系统会把这些不相关的 chunk 强行塞进 Prompt，AI 就会被噪音干扰，生成"幻觉"答案。阈值过滤相当于在说「找不到就别硬编」。

6.2 上下文注入 Prompt

检索到的 chunk 不会原样丢给 AI，而是按「来源 + 内容」格式化后注入：

def build_rag_prompt(query: str, chunks: list[RetrievedChunk]) -> str:
    context = "\n\n".join([
        f"[Source: {chunk.source} | Relevance: {chunk.similarity:.2f}]\n{chunk.content}"
        for chunk in chunks
    ])

    return f"""You are a helpful assistant with access to the following documents.

## Relevant Documents
{context}

## User Question
{query}

Answer the question based on the provided documents. If the documents don't contain relevant information, say so honestly."""

附带来源信息的好处是：AI 可以在回答中引用具体来源（“根据《xxx.md》文档…”），用户可以自行验证。这在技术问答场景中特别重要——当 AI 可能出错时，引用来源是唯一的信息可信度锚点。

---

七、性能优化与值得关注的细节

7.1 批量嵌入

单条嵌入 API 调用的网络往返延迟（RTT）通常在 50-200ms。如果上传一篇 50 个 chunk 的文档，串行调用就是 50 次 RTT，用户能喝完一杯咖啡。

ORIGIN 的解决方案是批量嵌入：OpenAI 的 Embedding API 支持单次请求发送最多 2048 条文本，嵌入层会将一个文档的所有 chunk 合并为一次 API 调用，50 个 chunk 的索引时间从 10 秒降到 1 秒以内。

串行嵌入 50 个 chunk ≈ 10 秒（50 × 200ms RTT）；批量嵌入 1 次请求 ≈ 0.8 秒。

7.2 嵌入缓存

同一个文件不会变，每次重启或重建索引时不应当重新嵌入。ORIGIN 基于文件内容的 SHA256 哈希做了嵌入缓存——只有文件内容变化时才重新计算向量，避免了重复的 API 调用和费用。

7.3 流式检索

用户提问 → 嵌入查询 → 检索 → 注入 Prompt → 流式生成。其中「嵌入查询 + 检索」的总延迟控制在 100ms 以内，不会让用户感觉到等待。真正的瓶颈在 LLM 生成阶段，而那是流式输出的，用户从打出问题到看到第一个 token，通常在 1 秒以内。

---

八、当前限制与演进方向

当前限制

1. 单知识库检索：目前一次查询只能在一个知识库内检索。跨知识库联合检索是 v0.5 的目标
2. 纯文本检索：不支持图片中的文字检索（OCR 适配中），也不支持表格的结构化查询
3. 无重排序（Re-rank）：检索结果的排序完全依赖向量相似度，没有经过 cross-encoder 重排序。这意味着最相关的 chunk 偶尔会排在第二名而不是第一名

演进方向

v0.5 路线图中提到了 AI Agent + 工作流构建器。在 RAG 的上下文里，这可能意味着：

• 多步检索：Agent 先检索一次，根据结果调整查询，再检索第二次
• Query 重写：用户问「那个怎么搞？」，Agent 先把它重写为「RAG 知识库的部署步骤是什么？」
• 混合检索：向量检索 + 关键词检索（BM25），互补各自的盲区

---

九、总结

ORIGIN 的 RAG Pipeline 设计贯穿了一个原则：最小化运维负担，最大化实用性。

• 不分块？检索精度直接掉一半
• 不做 overlap？20% 的跨边界信息会丢失
• 不加 similarity 阈值？AI 分分钟开始编造
• 不用 pgvector 而引入独立向量库？Docker Compose 里多一个容器，用户多一份运维成本

这些设计决策单独看都很微小，但串联成一条完整的 Pipeline 后，决定了用户是「上传文档 → 获得精准回答」还是「上传文档 → 得到一堆似是而非的废话」。

对于正在设计自己的 RAG 系统的开发者，ORIGIN 的分块策略、嵌入缓存、pgvector 选型和上下文注入模式都是值得借鉴的参考实现。

---

相关阅读：开源项目推荐：ORIGIN AI Workspace —— 一键部署你的私有 AI 工作站

项目地址：https://github.com/micklzhang/ORIGIN-AI-Workspace

镜像仓库：https://github.com/1304674612/-origin-ai-workspace

免责声明：本文为技术分析，部分代码为基于项目架构的示意性重构，非项目原始代码的直接复制。