OpenRAG 生产级知识库架构实战

很多团队做 RAG，真正卡住的不是“大模型会不会回答”，而是知识库系统能不能长期稳定地接住企业文档、权限、并发、更新、审计与成本压力。
OpenRAG 的价值，不是又一个“聊天 Demo”，而是把文档处理、检索基础设施、工作流编排和对外接入整合成一套更接近生产现实的 RAG 平台底座。

一、先说结论：企业知识库的难点，从来不只是向量检索

在真实业务里，知识库系统通常会同时面对以下约束：

• 文档来源复杂：Confluence、GitLab Wiki、SharePoint、本地文件系统、对象存储、邮件附件、扫描 PDF 并存
• 文档质量参差：表格、代码块、页眉页脚、双栏排版、图片 OCR、版本历史混杂
• 权限模型严格：同一知识库中既有公开资料，也有部门文档、租户隔离文档、敏感运维手册
• 更新频率高：每天都有新增文档、旧文档修改、历史文档失效
• 查询模式多样：关键词检索、自然语言问答、版本号定位、工单号定位、跨文档聚合问答并存
• 稳定性要求高：一旦接入客服、研发支持、内部 Copilot，查询延迟、错误答案、权限穿透都会直接变成生产事故

所以，企业级 RAG 系统的目标从来不是“做出一个能回答问题的机器人”，而是：

1. 把异构知识持续、稳定、可审计地接入平台。
2. 在权限约束下完成高召回、低延迟、低幻觉的检索增强生成。
3. 在高并发场景下维持吞吐、成本、可扩展性和故障可恢复性。

从这个角度看，OpenRAG 的意义在于，它不是单独的检索库，而是一个更完整的 RAG 平台组合。

二、为什么选 OpenRAG：它解决的是“平台化”问题，而不是单点功能

根据 OpenRAG 官方文档，OpenRAG 是一个开源的 Agentic RAG 平台，核心上将 Langflow、OpenSearch、Docling 以及 OpenRAG backend 组合成容器化架构，用于构建可部署的知识库系统。

如果把它放进企业级知识库建设语境中，可以把它理解为四个层次：

• Docling ：负责把复杂文档解析成结构化内容，是知识进入系统的入口质量保障
• OpenSearch ：负责全文检索、向量检索、过滤、聚合，是检索和索引治理的核心基础设施
• Langflow ：负责把检索、重排、Prompt、工具调用、Agent 流程组织成可编排工作流
• OpenRAG backend ：负责对外 API、平台协调、连接各组件并承接产品化能力

这套组合优于“LangChain + 向量库 + 脚本拼装”的地方，不在于它一定更轻，而在于它更像一套平台底座：

• 它天然适合做统一知识接入与统一检索服务
• 它更容易承载工作流编排、MCP 接入、Agentic RAG 扩展
• 它更适合和企业已有 OpenSearch、Kubernetes、对象存储、监控体系衔接

但也要明确边界： OpenRAG 不会自动替你完成企业级治理设计 。多租户隔离、索引生命周期、权限模型、异步导入、容量规划、可观测性，这些仍然需要架构师自己补齐。

三、生产级 OpenRAG 的正确架构视角：控制面、数据面、检索面、生成面分离

很多文章介绍 RAG，只画“上传文档 → 向量化 → 检索 → LLM 回答”的流程图。这对生产系统远远不够。

更合理的架构拆法如下：

┌──────────────────────────────┐
                         │         Access Layer         │
                         │ Web / API / MCP / SDK / Bot  │
                         └──────────────┬───────────────┘
                                        │
                         ┌──────────────▼───────────────┐
                         │        Control Plane         │
                         │ Tenant / ACL / Config / Flow │
                         │ Quota / Audit / Release      │
                         └──────────────┬───────────────┘
                                        │
          ┌─────────────────────────────┼─────────────────────────────┐
          │                             │                             │
┌─────────▼─────────┐        ┌──────────▼──────────┐       ┌──────────▼──────────┐
│   Ingestion Plane │        │   Retrieval Plane   │       │   Generation Plane  │
│ Source Adapters   │        │ Query Rewrite       │       │ Prompt Assembly      │
│ Docling Parse     │        │ Hybrid Search       │       │ LLM / Tools / Agent  │
│ Chunk / Embed     │        │ Rerank / ACL Filter │       │ Guardrail / Citation │
│ Async Index       │        │ Cache / Fallback    │       │ Streaming / Memory   │
└─────────┬─────────┘        └──────────┬──────────┘       └──────────┬──────────┘
          │                             │                             │
          └──────────────┬──────────────┴──────────────┬──────────────┘
                         │                             │
                ┌────────▼────────┐           ┌────────▼────────┐
                │  OpenSearch     │           │  Object Storage │
                │ Text + Vector   │           │ Source / Snapshot│
                │ Metadata + ACL  │           │ Parsed Artifacts │
                └─────────────────┘           └─────────────────┘

3.1 控制面解决“治理”

控制面不是可选项，它决定系统能否进入生产：

• 租户管理：索引命名、查询隔离、配额和限流
• 权限治理：文档 ACL、用户组映射、字段级脱敏、审计追踪
• 工作流治理：Langflow 流程版本管理、灰度发布、回滚
• 成本治理：Embedding 配额、Reranker GPU 池限额、模型调用预算
• 运行治理：SLO、告警、故障演练、索引生命周期策略

3.2 数据面解决“持续接入”

数据面要保证文档可持续进入系统，而不是导入一次就结束：

• 全量初始化导入
• 增量变更同步
• 删除与失效同步
• 重建索引
• 文档版本追踪
• 元数据补全与血缘追踪

3.3 检索面解决“召回质量与延迟”

检索面是 RAG 的核心竞争力，不仅要“找到内容”，还要“在权限内、低延迟、可解释地找到正确内容”。

它通常由以下阶段组成：

1. Query 理解与改写
2. 权限与租户过滤
3. 混合检索
4. 重排精排
5. 去重与上下文拼装
6. 引用片段构造

3.4 生成面解决“可回答、可约束、可追责”

生成面要解决的不是“把检索结果塞给模型”，而是：

• 如何限制模型只能基于检索证据作答
• 如何返回引用来源
• 如何在无结果时安全降级
• 如何支持工具调用、SQL、工单系统、CMDB、知识卡片拼装

四、OpenRAG 在生产中的价值，不在“能接文档”，而在“解析质量”

企业知识库的第一层门槛，不是向量库，而是文档解析质量。

Docling 的价值在于它不是简单抽文本，而是尽量保留文档结构语义。根据官方项目说明，Docling 对 PDF 等文档具备版面、阅读顺序、表格结构、代码、公式、OCR 等处理能力，并能导出 Markdown、JSON 等格式。这一点对 RAG 至关重要，因为决定召回质量上限的，往往不是模型，而是文档被切成什么样。

4.1 为什么“结构保真”比“纯文本提取”重要

如果文档在进入向量化前就被错误拆解，会直接造成以下问题：

• 代码块被截断，导致回答片段缺上下文
• 表格行列关系丢失，检索时语义被破坏
• 页眉页脚和正文混淆，召回大量噪声
• 双栏文档阅读顺序错误，句子前后颠倒
• 扫描件 OCR 错误被当作有效知识写入索引

因此，生产级知识库要先做“文档语义重建”，再做 Chunk。

4.2 文档接入的正确流水线

建议把接入链路拆为以下阶段：

Source Connector
  -> File Snapshot
  -> Dedup / Hash
  -> Docling Parse
  -> Structure Normalize
  -> Metadata Enrich
  -> Smart Chunk
  -> Embedding
  -> Bulk Index
  -> Validation / Audit

每一步都要能落库、能重试、能审计。

五、生产级文档导入链路设计：一定要异步化、幂等化、可恢复

5.1 反例：同步上传即解析

很多团队一开始会这么写：

• 用户上传文件
• API 同步调用解析器
• 同步分块
• 同步向量化
• 同步写入 OpenSearch
• 返回成功

这种链路在文档数量少时可行，但在生产里会很快崩溃：

• 大 PDF 或扫描件解析时间不可控
• Embedding 推理耗时高且成本敏感
• OpenSearch 批量写入存在背压
• 某一步失败会让整个请求超时
• 前台接口与后台重任务强耦合

5.2 正确模型：命令请求 + 异步流水线

Upload API
  -> persist raw file
  -> create ingestion task
  -> enqueue event
  -> return task_id

Worker
  -> parse
  -> normalize
  -> chunk
  -> embed
  -> bulk index
  -> publish completed event

5.3 生产级导入任务数据模型

CREATE TABLE kb_ingestion_task (
  id                BIGINT PRIMARY KEY,
  tenant_id         VARCHAR(64) NOT NULL,
  knowledge_base_id VARCHAR(64) NOT NULL,
  source_uri        VARCHAR(1024) NOT NULL,
  source_type       VARCHAR(32) NOT NULL,
  content_hash      CHAR(64) NOT NULL,
  doc_version       VARCHAR(128) NOT NULL,
  status            VARCHAR(32) NOT NULL,
  retry_count       INT NOT NULL DEFAULT 0,
  error_code        VARCHAR(64),
  error_message     TEXT,
  created_at        TIMESTAMP NOT NULL,
  updated_at        TIMESTAMP NOT NULL,
  UNIQUE (tenant_id, knowledge_base_id, source_uri, content_hash)
);

这个唯一键非常关键，它让“同一文档同一版本”天然幂等，避免重复导入、重复切块、重复向量化。

5.4 生产级导入编排代码骨架

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from hashlib import sha256
from typing import Iterable

@dataclass
class IngestionCommand:
    tenant_id: str
    knowledge_base_id: str
    source_uri: str
    source_type: str
    operator: str

class IngestionOrchestrator:
    def __init__(
        self,
        task_repo,
        file_store,
        event_bus,
        clock=datetime,
    ) -> None:
        self.task_repo = task_repo
        self.file_store = file_store
        self.event_bus = event_bus
        self.clock = clock

    def submit(self, cmd: IngestionCommand, content: bytes) -> dict:
        content_hash = sha256(content).hexdigest()
        snapshot_uri = self.file_store.put(
            tenant_id=cmd.tenant_id,
            knowledge_base_id=cmd.knowledge_base_id,
            content_hash=content_hash,
            content=content,
        )

        existing = self.task_repo.find_by_unique_key(
            tenant_id=cmd.tenant_id,
            knowledge_base_id=cmd.knowledge_base_id,
            source_uri=cmd.source_uri,
            content_hash=content_hash,
        )
        if existing:
            return {"task_id": existing.id, "deduplicated": True}

        task = self.task_repo.create(
            tenant_id=cmd.tenant_id,
            knowledge_base_id=cmd.knowledge_base_id,
            source_uri=cmd.source_uri,
            source_type=cmd.source_type,
            content_hash=content_hash,
            snapshot_uri=snapshot_uri,
            status="PENDING",
            created_at=self.clock.now(timezone.utc),
        )

        self.event_bus.publish(
            topic="kb.ingestion.submitted",
            key=f"{cmd.tenant_id}:{cmd.knowledge_base_id}",
            value={
                "task_id": task.id,
                "tenant_id": cmd.tenant_id,
                "knowledge_base_id": cmd.knowledge_base_id,
                "snapshot_uri": snapshot_uri,
            },
        )
        return {"task_id": task.id, "deduplicated": False}

这个实现有几个生产含义：

• 原始文件先落对象存储，保证后续可重试
• API 只做轻事务，不执行重计算
• 事件按租户或知识库键路由，便于做有序消费
• 重复提交天然去重

5.5 Worker 处理阶段的职责拆分

class IngestionWorker:
    def __init__(
        self,
        parser,
        chunker,
        embedding_gateway,
        index_writer,
        task_repo,
    ) -> None:
        self.parser = parser
        self.chunker = chunker
        self.embedding_gateway = embedding_gateway
        self.index_writer = index_writer
        self.task_repo = task_repo

    def handle(self, task: dict) -> None:
        self.task_repo.mark_running(task["task_id"])

        parsed_doc = self.parser.parse(task["snapshot_uri"])
        chunks = self.chunker.split(parsed_doc)
        vectors = self.embedding_gateway.embed_batch(
            [chunk.text for chunk in chunks]
        )
        self.index_writer.bulk_upsert(
            tenant_id=task["tenant_id"],
            knowledge_base_id=task["knowledge_base_id"],
            chunks=chunks,
            vectors=vectors,
        )

        self.task_repo.mark_succeeded(task["task_id"], chunk_count=len(chunks))

真正落地时，这里还要继续补上：

• 重试上限与指数退避
• Poison Message 隔离
• 失败任务转人工复核队列
• 导入超时终止
• 文档页数、大小、类型白名单控制

六、Chunk 不是切字符串，而是做“知识单元建模”

RAG 中最容易被低估的工作之一，就是 Chunk 策略。

错误的 Chunk 策略通常会导致两类问题：

• 召回差：一个知识点被切散，检索时命不中
• 生成差：命中多个碎片，模型拼上下文时发生语义丢失

6.1 生产级 Chunk 设计原则

1. 以语义结构为边界，而不是固定字符数。
2. 保留章节路径、标题层级、表格标题、代码语言等元数据。
3. 对表格、代码块、命令清单使用特殊分块策略。
4. 为每个 Chunk 建立稳定主键和文档版本关系。
5. 分离“检索文本”和“展示文本”。

6.2 推荐的 Chunk 元数据模型

{
  "chunk_id": "kb1-docA-v17-sec03-chunk02",
  "tenant_id": "t1",
  "knowledge_base_id": "kb1",
  "doc_id": "docA",
  "doc_version": "v17",
  "title_path": ["OpenRAG 生产部署", "检索服务扩缩容"],
  "chunk_type": "paragraph",
  "language": "zh-CN",
  "token_count": 286,
  "page_range": [12, 13],
  "acl_tags": ["team-search", "p2-internal"],
  "source_uri": "s3://kb/raw/docA.pdf",
  "content": "......",
  "display_content": "......",
  "embedding_model": "bge-m3",
  "created_at": "2026-06-04T08:00:00Z"
}

6.3 智能分块实现骨架

class StructuredChunker:
    def split(self, parsed_document) -> list:
        chunks = []
        for section in parsed_document.sections:
            if section.kind == "table":
                chunks.extend(self._split_table(section))
            elif section.kind == "code":
                chunks.extend(self._split_code(section))
            else:
                chunks.extend(self._split_paragraph(section))
        return chunks

    def _split_paragraph(self, section):
        # 以标题层级和语义段落为主，必要时再按 token 预算切分
        ...

    def _split_table(self, section):
        # 表格优先整体保留，超大表格再按行组拆分，并保留列头
        ...

    def _split_code(self, section):
        # 代码块不按字符硬切，必要时按函数、类、配置块切分
        ...

真正影响效果的，不是 _split_* 的代码行数，而是你有没有把“什么是一个可检索知识单元”定义清楚。

七、OpenSearch 在知识库中的正确用法：不是“向量库替代品”，而是检索控制中枢

OpenSearch 对企业级知识库的价值，不仅是能存向量，更重要的是：

• 它同时支持全文检索与向量检索
• 它支持丰富过滤条件、聚合、排序、生命周期管理
• 它适合与日志、监控、审计、权限元数据一起形成统一检索基础设施

7.1 索引建模建议：文档索引与 Chunk 索引分离

生产实践里，不建议只保留一个“大而全”的索引。更推荐至少拆成两类：

• kb_document_index
• 保存文档级元数据：来源、版本、权限、更新时间、摘要、状态
• kb_chunk_index
• 保存检索单元：Chunk 内容、向量、标题路径、页码、ACL、版本信息

这样做的价值在于：

• 文档级查询与片段级查询职责分离
• 删除、重建、失效控制更容易
• 版本治理与展示拼装更清晰

7.2 生产级 Chunk 索引 Mapping 示例

PUT kb_chunk_index_v1
{
  "settings": {
    "index": {
      "knn": true,
      "number_of_shards": 6,
      "number_of_replicas": 1,
      "refresh_interval": "30s"
    }
  },
  "mappings": {
    "properties": {
      "tenant_id": { "type": "keyword" },
      "knowledge_base_id": { "type": "keyword" },
      "doc_id": { "type": "keyword" },
      "doc_version": { "type": "keyword" },
      "chunk_id": { "type": "keyword" },
      "chunk_type": { "type": "keyword" },
      "title_path": {
        "type": "text",
        "fields": {
          "keyword": { "type": "keyword", "ignore_above": 256 }
        }
      },
      "content": {
        "type": "text"
      },
      "display_content": {
        "type": "text",
        "index": false
      },
      "acl_tags": { "type": "keyword" },
      "page_start": { "type": "integer" },
      "page_end": { "type": "integer" },
      "updated_at": { "type": "date" },
      "embedding": {
        "type": "knn_vector",
        "dimension": 1024,
        "method": {
          "name": "hnsw",
          "space_type": "cosinesimil",
          "engine": "faiss",
          "parameters": {
            "m": 16,
            "ef_construction": 256
          }
        }
      }
    }
  }
}

7.3 检索过滤的一个关键认知：不要把旧经验写死

过去很多文章会强调“为了支持过滤条件，必须用 script_score 自己做向量检索”。这个结论在今天已经不能简单照搬。

根据 OpenSearch 当前官方文档：

• k-NN query 已支持在特定引擎下直接使用 filter
• script_score 更适合做精确的 pre-filter exact search，适用于过滤后数据量较小、追求高精度的场景

因此，生产实践中的正确策略应该是：

• 大规模常规查询：优先使用 k-NN + filter 做近似向量检索
• 高选择性过滤或强精度场景：再考虑 script_score
• 精确编号类查询：直接走 BM25、term 或 keyword 检索，不要强行向量化

这比“永远 script_score”更符合当前 OpenSearch 能力，也更符合性能现实。

八、混合检索的核心，不是把 BM25 和向量硬拼，而是构造多路召回

8.1 为什么企业知识库必须做 Hybrid Search

纯向量检索对语义类问题很有效，但对以下内容通常不稳定：

• 版本号，如 v3.2.17
• 工单号，如 INC-20260601-0091
• 配置项名，如 spring.kafka.consumer.max-poll-records
• 精确报错，如 ClassNotFoundException: org...

纯关键词检索又很难理解语义问法，例如：

• “我们知识库的导入链路怎么做幂等”
• “为什么检索命中了但回答还是不准”

所以，企业级查询一定是多路召回：

1. 语义向量召回
2. BM25 关键词召回
3. 精确字段召回
4. 规则增强召回

8.2 生产级混合检索实现骨架

class RetrievalPlanBuilder:
    def build(self, query: str, tenant_id: str, acl_tags: list[str]) -> dict:
        filter_clause = [
            {"term": {"tenant_id": tenant_id}},
            {"terms": {"acl_tags": acl_tags}}
        ]

        return {
            "vector_plan": {
                "k": 80,
                "num_candidates": 300,
                "filter": {"bool": {"must": filter_clause}}
            },
            "bm25_plan": {
                "size": 40,
                "query": {
                    "bool": {
                        "must": [
                            {"match": {"content": {"query": query}}}
                        ],
                        "filter": filter_clause
                    }
                }
            },
            "exact_plan": {
                "size": 10,
                "query": {
                    "bool": {
                        "should": [
                            {"term": {"doc_id": query}},
                            {"term": {"chunk_id": query}},
                            {"term": {"title_path.keyword": query}}
                        ],
                        "filter": filter_clause,
                        "minimum_should_match": 1
                    }
                }
            }
        }

8.3 多路召回后一定要做融合

融合不是简单拼接结果，而是做以下工作：

• 分数归一化
• 去重
• 文档级去偏置
• 标题命中加权
• 新鲜度加权
• 权限再次校验

class ResultFusion:
    def fuse(self, vector_hits, bm25_hits, exact_hits):
        merged = {}
        for source, hits, weight in [
            ("vector", vector_hits, 0.45),
            ("bm25", bm25_hits, 0.35),
            ("exact", exact_hits, 0.20),
        ]:
            for hit in hits:
                key = hit["chunk_id"]
                merged.setdefault(key, {**hit, "score": 0.0, "reasons": []})
                merged[key]["score"] += self._normalize(hit["_score"]) * weight
                merged[key]["reasons"].append(source)

        results = list(merged.values())
        results.sort(key=lambda x: x["score"], reverse=True)
        return results

    def _normalize(self, score: float) -> float:
        return min(max(score / 10.0, 0.0), 1.0)

九、Reranker 才是把 Top-K 变成可用答案的关键一跳

大多数生产系统里，向量召回的职责只是“别漏掉”，不是“最终排序绝对准确”。

最终可用性，通常由 Reranker 决定。

9.1 Reranker 的工程意义

Reranker 是 Cross-Encoder，它直接对 (query, candidate) 成对打分，通常比双塔向量模型更准，但代价也更高。因此它的正确位置是：

• 放在初召回之后
• 只处理 Top 30 ~ Top 100 候选
• 独立部署成 GPU 推理服务

9.2 精排服务实现骨架

class RerankService:
    def __init__(self, model_client, max_batch_size: int = 16):
        self.model_client = model_client
        self.max_batch_size = max_batch_size

    def rerank(self, query: str, candidates: list[dict], top_k: int = 8):
        pairs = [
            {
                "query": query,
                "document": item["content"],
                "chunk_id": item["chunk_id"],
            }
            for item in candidates
        ]

        scores = []
        for batch in self._batched(pairs, self.max_batch_size):
            scores.extend(self.model_client.score(batch))

        for item, score in zip(candidates, scores):
            item["rerank_score"] = score

        candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
        return candidates[:top_k]

    def _batched(self, items, size):
        for i in range(0, len(items), size):
            yield items[i:i + size]

9.3 高并发下的 Reranker 设计原则

• 模型服务独立伸缩，不和 API Pod 混布
• 批处理合并请求，提升 GPU 利用率
• 设置最大候选数，避免长尾请求拖垮系统
• 有明确降级策略：Reranker 故障时退回融合结果，而不是全链路失败

十、权限模型是企业知识库的生死线：检索前过滤，生成前复核

很多 Demo 系统没有 ACL，这在企业里是绝对不能上线的。

如果没有权限模型，最常见的事故就是：

• 向量召回命中了本不该看的文档
• LLM 在回答里拼接了敏感片段
• 用户通过模糊提问绕过了目录权限

10.1 正确的权限控制原则

1. 权限必须进入索引，而不是只在前端页面判断。
2. 权限过滤必须发生在检索阶段，而不是答案生成后再删。
3. 检索后拼装上下文前要再做一次权限复核。
4. 文档权限变更要能增量传播到索引。

10.2 ACL 字段设计建议

{
  "tenant_id": "t1",
  "acl_tags": ["department-rd", "service-search", "internal"],
  "acl_mode": "ALLOW",
  "owner_uid": "u1001",
  "sensitivity_level": "L2"
}

10.3 检索前权限收敛

不要把用户所有权限一股脑透传给搜索层，应该先在鉴权层归一成可检索标签：

class AccessScopeResolver:
    def resolve(self, user) -> dict:
        return {
            "tenant_id": user.tenant_id,
            "acl_tags": sorted(set(user.roles + user.groups + user.scopes)),
            "sensitivity_ceiling": user.sensitivity_ceiling,
        }

10.4 生成前复核

即使检索阶段已经过滤，生成前仍建议做一次上下文复核：

• 防止缓存污染
• 防止跨请求上下文混入
• 防止多路召回融合时出现脏数据

十一、问答链路设计：把检索增强生成拆成可治理的服务拓扑

11.1 典型查询链路

User Query
  -> AuthN/AuthZ
  -> Query Rewrite
  -> Retrieval Plan
  -> Hybrid Recall
  -> Rerank
  -> Context Assemble
  -> Prompt Build
  -> LLM Generate
  -> Citation Attach
  -> Stream Response

11.2 生产级问答服务骨架

class KnowledgeQaService:
    def __init__(
        self,
        access_scope_resolver,
        query_rewriter,
        retriever,
        reranker,
        prompt_builder,
        llm_client,
        answer_guard,
    ) -> None:
        self.access_scope_resolver = access_scope_resolver
        self.query_rewriter = query_rewriter
        self.retriever = retriever
        self.reranker = reranker
        self.prompt_builder = prompt_builder
        self.llm_client = llm_client
        self.answer_guard = answer_guard

    async def ask(self, user, question: str) -> dict:
        scope = self.access_scope_resolver.resolve(user)
        rewritten = await self.query_rewriter.rewrite(question)
        recalls = await self.retriever.retrieve(rewritten, scope)
        ranked = self.reranker.rerank(question, recalls, top_k=6)
        prompt = self.prompt_builder.build(question, ranked)
        raw_answer = await self.llm_client.chat(prompt)
        safe_answer = self.answer_guard.validate(raw_answer, ranked)
        return {
            "answer": safe_answer.text,
            "citations": safe_answer.citations,
            "used_chunks": [item["chunk_id"] for item in ranked],
        }

11.3 Answer Guard 应该做什么

• 检查回答是否引用到了未授权上下文
• 检查是否出现与证据不一致的断言
• 检查在无证据时是否执行拒答或降级回复
• 检查是否含敏感字段、密钥、账号、手机号等脱敏规则

十二、高并发架构升级：把“RAG 调用”拆成资源隔离的多个池

很多团队一到 30~50 QPS 就开始抖动，根因通常不是某个组件单独慢，而是所有重任务都挤在同一条资源池里。

生产级 RAG 至少要拆出以下资源池：

• API 接入池
• Query Rewrite / Prompt 编排池
• Retrieval 检索池
• Reranker GPU 池
• LLM 推理池
• 文档导入池

12.1 为什么必须资源池拆分

因为导入链路和查询链路的资源画像完全不同：

• 导入链路偏 CPU、批处理、异步、长任务
• 查询链路偏低延迟、在线、突发流量
• Reranker 偏 GPU 批量吞吐
• LLM 推理偏令牌吞吐和上下文长度成本

如果不隔离，常见后果就是：

• 大文档导入把 API CPU 打满
• Reranker 拖慢检索链路
• LLM 长上下文请求挤占普通问答资源

12.2 推荐的 Kubernetes 部署拆分

apiVersion: apps/v1
kind: Deployment
metadata:
  name: openrag-api
spec:
  replicas: 4
  template:
    spec:
      containers:
      - name: api
        image: your-registry/openrag-api:1.0.0
        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2"
            memory: "4Gi"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: reranker-service
spec:
  replicas: 2
  template:
    spec:
      nodeSelector:
        workload: gpu
      containers:
      - name: reranker
        image: your-registry/reranker-service:1.0.0
        resources:
          requests:
            nvidia.com/gpu: "1"
            cpu: "2"
            memory: "8Gi"
          limits:
            nvidia.com/gpu: "1"
            cpu: "4"
            memory: "16Gi"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ingestion-worker
spec:
  replicas: 6
  template:
    spec:
      containers:
      - name: worker
        image: your-registry/openrag-ingestion:1.0.0
        resources:
          requests:
            cpu: "1"
            memory: "2Gi"
          limits:
            cpu: "4"
            memory: "8Gi"

12.3 扩缩容不要只看 CPU，要看业务指标

RAG 系统最不应该只用 CPU 做 HPA。更合理的指标是：

• 检索 P95 延迟
• Kafka 导入堆积量
• Reranker 队列长度
• LLM token 吞吐
• 活跃查询数

如果使用 KEDA，可以直接基于 Kafka Lag、Prometheus 指标做弹性伸缩。

十三、缓存体系设计：缓存的不是“答案”本身，而是链路中的稳定中间结果

很多系统上来就缓存最终答案，这通常收益有限且风险不小。更推荐缓存以下内容：

• Query Embedding
• 权限范围解析结果
• 多路召回结果
• Reranker 中间结果
• Prompt 模板片段

13.1 推荐缓存分层

L1: 进程内短 TTL Cache
L2: Redis 分布式缓存
L3: OpenSearch Query Cache
L4: 对象存储中的解析产物缓存

13.2 Embedding 缓存示例

import hashlib
from datetime import timedelta

class EmbeddingCache:
    def __init__(self, redis_client, embedding_client):
        self.redis = redis_client
        self.embedding_client = embedding_client

    async def embed(self, text: str, model: str) -> list[float]:
        digest = hashlib.sha256(f"{model}:{text}".encode()).hexdigest()
        cache_key = f"embed:{digest}"
        cached = await self.redis.get(cache_key)
        if cached:
            return self._decode(cached)

        vector = await self.embedding_client.embed(text=text, model=model)
        await self.redis.set(cache_key, self._encode(vector), ex=86400 * 7)
        return vector

    def _encode(self, vector):
        ...

    def _decode(self, payload):
        ...

13.3 缓存使用原则

• 带权限差异的结果必须把权限范围纳入 Key
• Prompt 结果缓存要考虑模型版本和模板版本
• 热门查询要设置抖动 TTL，避免同时失效造成缓存雪崩

十四、可观测性不是锦上添花，而是知识库能不能运营的前提

RAG 平台上线后，最怕的不是“完全不可用”，而是“表面可用但答案质量在持续下滑而没人知道”。

所以监控必须覆盖两层：

• 系统可用性
• 检索与回答质量

14.1 核心 SLI / SLO

建议至少定义以下指标：

指标	含义	建议目标
`qa_request_success_rate`	问答请求成功率	`>= 99.9%`
`qa_p95_latency_ms`	端到端问答 P95	`< 3000ms`
`retrieval_p95_latency_ms`	检索阶段 P95	`< 400ms`
`rerank_p95_latency_ms`	精排阶段 P95	`< 300ms`
`ingestion_success_rate`	导入成功率	`>= 99.5%`
`citation_coverage_ratio`	回答附带引用比例	`>= 95%`
`no_answer_safe_ratio`	无证据时安全拒答比例	`>= 99%`

14.2 指标埋点建议

class MetricsAwareRetriever:
    async def retrieve(self, query: str, scope: dict):
        started = time.perf_counter()
        try:
            result = await self._do_retrieve(query, scope)
            metrics.histogram("retrieval_latency_ms").record(
                (time.perf_counter() - started) * 1000
            )
            metrics.counter("retrieval_success_total").add(1)
            metrics.histogram("retrieval_hit_count").record(len(result))
            return result
        except Exception:
            metrics.counter("retrieval_failure_total").add(1)
            raise

14.3 日志与 Trace 要打到哪一层

• 上传请求
• 导入任务
• 解析耗时
• Chunk 数量
• Embedding 批次数
• OpenSearch 查询体摘要
• Reranker 候选数
• LLM 调用 token
• 最终引用片段 ID

推荐全链路 Trace 贯穿 request_id 、 tenant_id 、 knowledge_base_id 、 task_id 。

十五、从 0 到 100 万文档的演进路线

15.1 阶段一：验证期

适用规模：

• 文档量 < 1 万
• 使用者 < 50
• 目标是验证解析质量、召回质量和知识组织方式

建议架构：

• OpenRAG 单实例
• OpenSearch 单节点
• Docling 本地服务
• 单租户
• 手工导入

这个阶段不要过早做复杂分布式设计，但要把后续要用的数据模型、任务状态、ACL 字段一次设计正确。

15.2 阶段二：生产初期

适用规模：

• 文档量 1 万 ~ 20 万
• 多部门使用
• 日常增量导入

建议架构：

• OpenRAG API 多副本
• OpenSearch 3 节点
• Kafka 异步导入
• Redis 缓存
• 对象存储保存原文与解析产物
• 简化 ACL 模型

15.3 阶段三：平台化阶段

适用规模：

• 文档量 20 万 ~ 100 万+
• 多租户
• 接入多个业务系统或 AI 助手

建议架构：

• 接入层 API Gateway
• 控制面独立服务
• 检索服务、Reranker 服务、导入服务分池部署
• OpenSearch 热温冷分层
• Langflow 工作流版本管理
• 审计与计费体系接入

十六、真实案例：企业内部研发知识助手如何落地

下面给出一个更贴近生产的场景。

16.1 场景背景

某企业内部研发平台需要建设统一知识助手，覆盖：

• API 设计规范
• 架构设计文档
• 运行手册
• 故障复盘
• SDK 使用说明
• 发布流程文档

数据规模：

• 存量文档约 65 万份
• 日新增文档约 3000 份
• 活跃用户 4000+
• 工作日问答峰值 120 QPS

16.2 关键挑战

• 文档格式高度异构，扫描 PDF 占比高
• 权限隔离严格，研发、测试、SRE、客服可见范围不同
• 问题类型复杂，既有语义问答，也有精确配置定位
• 高峰期更新与查询并发发生

16.3 最终落地架构

Confluence / Git / SharePoint / S3
    -> Connector
    -> Snapshot Store
    -> Kafka Ingestion
    -> Docling Parse Cluster
    -> Structured Chunker
    -> Embedding Gateway
    -> OpenSearch Chunk Index
    -> Retrieval Service
    -> Reranker GPU Service
    -> LLM Gateway
    -> Internal Copilot / MCP / Web Chat

16.4 核心收益

• 文档导入从“全量重跑”改为“按版本增量更新”
• 权限隔离从前端控制升级为检索层 ACL 过滤
• 平均问答延迟稳定在秒级
• 无结果场景实现了安全拒答与人工跳转
• 故障定位从“猜哪一步慢”变为可 Trace、可归因

这类项目真正的价值，不是把问答做得更炫，而是把组织知识流动效率从“靠人记忆”升级成“靠平台供给”。

十七、Agentic RAG 在 OpenRAG 中怎么用，才不会把系统复杂度拉爆

OpenRAG 天然适合向 Agentic RAG 扩展，因为它本身具备工作流编排能力。但要注意，Agentic RAG 不是默认更好。

17.1 什么时候值得引入 Agentic RAG

• 用户问题本身是多跳问题
• 需要多知识域组合
• 需要调用外部系统确认事实
• 需要先检索、再规划、再二次检索

17.2 不适合 Agentic 的场景

• FAQ 类简单问答
• 强实时、极低延迟场景
• 权限边界非常敏感且工具调用链复杂的场景

17.3 推荐的引入方式

不要一开始就把所有请求都走 Agent。

更合理的策略是：

1. 先做 Query Complexity 判定
2. 简单问题走普通 RAG
3. 复杂问题走 Agentic Flow
4. Agent 流程有最大步数、最大工具数、最大总耗时限制

这样才能兼顾效果与成本。

十八、MCP 接入的真正意义：知识库从“网页应用”变成“组织能力”

OpenRAG 提供 MCP 方向的接入能力，这意味着知识库不只是一个网页聊天系统，还可以成为 IDE、桌面 Agent、企业 Copilot 的统一知识后端。

从架构角度看，这件事的意义是：

• Web Chat 不再是唯一入口
• IDE、机器人、自动化流程都可以复用同一知识库
• 检索、权限、审计、引用能力实现平台复用

但这里同样要补齐治理：

• MCP Token 鉴权
• 客户端限流
• 工具权限边界
• 审计日志
• 多客户端版本兼容

否则“接得越多，风险越大”。

十九、生产排坑清单：这些问题几乎一定会遇到

19.1 召回看起来很多，但答案还是不准

常见根因：

• Chunk 粒度不对
• 标题路径丢失
• 表格被打散
• Reranker 缺失
• Prompt 未明确要求基于证据回答

19.2 导入吞吐很低

常见根因：

• Docling 解析和 Embedding 串行执行
• 批量写入参数过小
• OpenSearch refresh_interval 太短
• 大量重复文档未做去重

19.3 并发一高就抖动

常见根因：

• 导入与查询未分池
• Reranker 与 API 服务混部
• 缺少请求限流与队列保护
• 查询走了超长上下文 LLM

19.4 权限偶发穿透

常见根因：

• 缓存 Key 未包含权限范围
• 多路召回融合时漏掉 ACL 复核
• 文档权限变更未同步刷新索引

二十、上线前 Checklist：没有这份清单，不建议直接进生产

架构侧

• 是否拆分了控制面、导入面、检索面、生成面职责
• 是否明确了租户边界、权限边界、配额边界
• 是否定义了索引版本、别名切换、重建策略

数据侧

• 是否有文档去重与幂等主键
• 是否支持增量更新、删除同步、版本追踪
• 是否保存原始文件快照和解析产物

检索侧

• 是否实现多路召回而非单一向量检索
• 是否做了 Reranker
• 是否对编号类查询保留精确检索路径

工程侧

• 是否异步化导入
• 是否分离资源池
• 是否有缓存策略、限流策略、降级策略

治理侧

• 是否实现检索前权限过滤与生成前复核
• 是否有审计日志、Trace、指标告警
• 是否对无证据回答做了安全拒答

二十一、总结：OpenRAG 适合作为企业级知识平台底座，但前提是你把工程化补全

OpenRAG 的优势，不在于“它能做 RAG”，而在于它把企业知识库最关键的几个基础能力拼到了同一平台上：

• Docling 负责高质量文档解析
• OpenSearch 负责混合检索与索引治理
• Langflow 负责流程编排与 Agent 扩展
• OpenRAG backend 负责平台化接入与能力整合

但一套能在生产稳定运行的知识库，真正决定成败的仍然是这些工程化能力：

• 导入链路异步化、幂等化、可恢复
• Chunk 语义建模而不是字符切分
• 混合检索、精排、权限过滤协同工作
• 高并发场景下的资源池隔离与弹性扩容
• 全链路可观测、可审计、可回滚

如果只是做 Demo，OpenRAG 当然可以很快跑起来。
如果是做企业知识中台，那么你要建设的不是“一个 RAG 应用”，而是一条长期稳定供给组织知识的生产系统。

参考资料

• OpenRAG 官方文档：https://docs.openr.ag/
• OpenRAG GitHub 仓库：https://github.com/langflow-ai/openrag
• Docling 官方项目：https://github.com/docling-project/docling
• Docling 官方文档：https://docling-project.github.io/docling/
• OpenSearch 向量过滤文档：https://docs.opensearch.org/latest/vector-search/filter-search-knn/
• OpenSearch k-NN 查询文档：https://docs.opensearch.org/latest/query-dsl/specialized/k-nn/

浏览主题

OpenRAG 生产级知识库架构实战：构建可治理、可扩展、可审计的企业级 RAG 平台

OpenRAG 生产级知识库架构实战

一、先说结论：企业知识库的难点，从来不只是向量检索

二、为什么选 OpenRAG：它解决的是“平台化”问题，而不是单点功能

三、生产级 OpenRAG 的正确架构视角：控制面、数据面、检索面、生成面分离

3.1 控制面解决“治理”

3.2 数据面解决“持续接入”

3.3 检索面解决“召回质量与延迟”

3.4 生成面解决“可回答、可约束、可追责”

四、OpenRAG 在生产中的价值，不在“能接文档”，而在“解析质量”

4.1 为什么“结构保真”比“纯文本提取”重要

4.2 文档接入的正确流水线

五、生产级文档导入链路设计：一定要异步化、幂等化、可恢复

5.1 反例：同步上传即解析

5.2 正确模型：命令请求 + 异步流水线

5.3 生产级导入任务数据模型

5.4 生产级导入编排代码骨架

5.5 Worker 处理阶段的职责拆分

六、Chunk 不是切字符串，而是做“知识单元建模”

6.1 生产级 Chunk 设计原则

6.2 推荐的 Chunk 元数据模型

6.3 智能分块实现骨架

七、OpenSearch 在知识库中的正确用法：不是“向量库替代品”，而是检索控制中枢

7.1 索引建模建议：文档索引与 Chunk 索引分离

7.2 生产级 Chunk 索引 Mapping 示例

7.3 检索过滤的一个关键认知：不要把旧经验写死

八、混合检索的核心，不是把 BM25 和向量硬拼，而是构造多路召回

8.1 为什么企业知识库必须做 Hybrid Search

8.2 生产级混合检索实现骨架

8.3 多路召回后一定要做融合

九、Reranker 才是把 Top-K 变成可用答案的关键一跳

9.1 Reranker 的工程意义

9.2 精排服务实现骨架

9.3 高并发下的 Reranker 设计原则

十、权限模型是企业知识库的生死线：检索前过滤，生成前复核

10.1 正确的权限控制原则

10.2 ACL 字段设计建议

10.3 检索前权限收敛

10.4 生成前复核

十一、问答链路设计：把检索增强生成拆成可治理的服务拓扑

11.1 典型查询链路

11.2 生产级问答服务骨架

11.3 Answer Guard 应该做什么

十二、高并发架构升级：把“RAG 调用”拆成资源隔离的多个池

12.1 为什么必须资源池拆分

12.2 推荐的 Kubernetes 部署拆分

12.3 扩缩容不要只看 CPU，要看业务指标

十三、缓存体系设计：缓存的不是“答案”本身，而是链路中的稳定中间结果

13.1 推荐缓存分层

13.2 Embedding 缓存示例

13.3 缓存使用原则

十四、可观测性不是锦上添花，而是知识库能不能运营的前提

14.1 核心 SLI / SLO

14.2 指标埋点建议

14.3 日志与 Trace 要打到哪一层

十五、从 0 到 100 万文档的演进路线

15.1 阶段一：验证期

15.2 阶段二：生产初期

15.3 阶段三：平台化阶段

十六、真实案例：企业内部研发知识助手如何落地

16.1 场景背景

16.2 关键挑战

16.3 最终落地架构

16.4 核心收益

十七、Agentic RAG 在 OpenRAG 中怎么用，才不会把系统复杂度拉爆

17.1 什么时候值得引入 Agentic RAG

17.2 不适合 Agentic 的场景

17.3 推荐的引入方式

十八、MCP 接入的真正意义：知识库从“网页应用”变成“组织能力”

十九、生产排坑清单：这些问题几乎一定会遇到

19.1 召回看起来很多，但答案还是不准

19.2 导入吞吐很低

19.3 并发一高就抖动

19.4 权限偶发穿透

二十、上线前 Checklist：没有这份清单，不建议直接进生产

架构侧

数据侧

检索侧

工程侧