RAGFlow架构详解与性能优化方案

数据处理模块

• 数据加载：读取文件内容，支持多种格式的输入。
• 解析分块：根据文件类型提取文本和相关信息，并按既定策略将内容切分成多个片段。
• 向量化嵌入：对每个文本片段生成语义向量，使其具备可检索性。
• 持久化存储：将片段及其对应的向量写入向量索引或数据库中。

FACTORY = {
        ParserType.PRESENTATION.value: presentation,
        ParserType.PICTURE.value: picture,
        ParserType.AUDIO.value: audio,
        ParserType.EMAIL.value: email
    }
    parser_config = {“chunk_token_num”: 4096, “delimiter”: “n!?;。；！？”, “layout_recognize”: “Plain Text”}
    exe = ThreadPoolExecutor(max_workers=12)
    threads = []
...
for d, blob in files:
        kwargs = {
            “callback”: dummy,
            “parser_config”: parser_config,
            “from_page”: 0,
            “to_page”: 100000,
            “tenant_id”: kb.tenant_id,
            “lang”: kb.language
        }
        threads.append(exe.submit(FACTORY.get(d[“parser_id”], naive).chunk, d["name"], blob, **kwargs))

if re.search(r“.pptx?$”, filename, re.IGNORECASE):
        ppt_parser = Ppt()
        for pn, (txt, img) in enumerate(ppt_parser(
                filename if not binary else binary, from_page, 1000000, callback)):
            d = copy.deepcopy(doc)
            pn += from_page
            d["image"] = img
            d["page_num_int"] = [pn + 1]
            d["top_int"] = [0]
            d["position_int"] = [(pn + 1, 0, img.size[0], 0, img.size[1])]
            tokenize(d, txt, eng)
            res.append(d)
        return res
elif re.search(r".pdf$", filename, re.IGNORECASE):
        pdf_parser = Pdf()
        if kwargs.get("layout_recognize", "DeepDOC") == "Plain Text":
            pdf_parser = PlainParser()
        for pn, (txt, img) in enumerate(pdf_parser(filename, binary,
                                                       from_page=from_page, to_page=to_page, callback=callback)):
            d = copy.deepcopy(doc)
            pn += from_page
            if img:
                d["image"] = img
            d["page_num_int"] = [pn + 1]
            d["top_int"] = [0]
            d["position_int"] = [(pn + 1, 0, img.size[0] if img else 0, 0, img.size[1] if img else 0)]
            tokenize(d, txt, eng)
            res.append(d)
        return res

for (docinfo, _), th in zip(files, threads):
        docs = []
        doc = {
            "doc_id": docinfo["id"],
            "kb_id": [kb.id]
        }
        for ck in th.result():
            d = deepcopy(doc)
            d.update(ck)
            d["id"] = xxhash.xxh64((ck["content_with_weight"] + str(d["doc_id"])).encode("utf-8")).hexdigest()
            d["create_time"] = str(datetime.now()).replace("T", " ")[:19]
            d["create_timestamp_flt"] = datetime.now().timestamp()
            if not d.get("image"):
                docs.append(d)
                continue
            output_buffer = BytesIO()
            if isinstance(d["image"], bytes):
                output_buffer = BytesIO(d["image"])
            else:
                d["image"].sa ve(output_buffer, format='JPEG')
            STORAGE_IMPL.put(kb.id, d["id"], output_buffer.getvalue())
            d["img_id"] = "{}-{}".format(kb.id, d["id"])
            d.pop("image", None)
            docs.append(d)

class StorageFactory:
    storage_mapping = {
        Storage.MINIO: RAGFlowMinio,
        Storage.AZURE_SPN: RAGFlowAzureSpnBlob,
        Storage.AZURE_SAS: RAGFlowAzureSasBlob,
        Storage.AWS_S3: RAGFlowS3,
        Storage.OSS: RAGFlowOSS,
    }
    @classmethod
    def create(cls, storage: Storage):
        return cls.storage_mapping[storage]()

STORAGE_IMPL_TYPE = os.getenv('STORAGE_IMPL', 'MINIO')
STORAGE_IMPL = StorageFactory.create(Storage[STORAGE_IMPL_TYPE])

• 问：如何理解解析输出的内容？ 答：解析器的 chunk 方法输出一个列表，列表中的每个元素都是一个字典，代表一个文本片段。这个字典包含了文本内容（content_with_weight）、所属文档 ID、知识库 ID、自身片段 ID、页面号、位置信息等元数据。图片数据会被单独存储（如 MINIO），只在字典中保留 img_id 作为引用，从而优化向量库的存储。
• 问：为什么要使用线程池并行解析文档？ 答：由于不同文档的解析过程相互独立，使用线程池可以同时处理多个文件，显著缩短批量处理时间。默认的 12 个线程设置是一个平衡点，但可以根据服务器的 CPU 核心数量进行调整。

• 大文件处理开销高： 对于超大文件（如含成千上万页的 PDF），单个解析线程可能长时间占用，并一次性产生大量片段数据存入内存。代码中会等待解析完成后再批量处理嵌入和存储，这可能导致内存占用峰值较高。

  • 优化建议： 可引入流式处理或分段加载机制，分批解析和处理文档。例如，将超长文档按章节分页解析，每生成一定数量片段就立刻进行嵌入和写入，减少单批内存占用。也可考虑在解析阶段对文本进行摘要或过滤无关信息，减小后续处理量。

• 缺乏解析缓存和增量更新： 每次新增或修改文档都会重新解析全文。DocumentService 目前通过在入库前删除原有片段实现更新，没有细粒度的增量更新机制。

  • 优化建议： 对于未变化的文档内容可跳过重复解析和嵌入。可以维护文件内容的哈希或指纹，如果文档再次上传且内容未变则复用已有向量。对于部分更新的文档，可对比新旧内容的差异，仅解析变化部分并更新对应片段。

  @classmethod
  @DB.connection_context()
  def remove_document(cls, doc, tenant_id):
          cls.clear_chunk_num(doc.id)
          try:
              settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
              graph_source = settings.docStoreConn.getFields(
                  settings.docStoreConn.search(["source_id"], [], {"kb_id": doc.kb_id, "knowledge_graph_kwd": ["graph"]}, [], OrderByExpr(), 0, 1, search.index_name(tenant_id), [doc.kb_id]), ["source_id"]
              )
              if len(graph_source) > 0 and doc.id in list(graph_source.values())[0]["source_id"]:
                  settings.docStoreConn.update({"kb_id": doc.kb_id, "knowledge_graph_kwd": ["entity", "relation", "graph", "subgraph", "community_report"], "source_id": doc.id},
                                              {"remove": {"source_id": doc.id}},
                                              search.index_name(tenant_id), doc.kb_id)
                  settings.docStoreConn.update({"kb_id": doc.kb_id, "knowledge_graph_kwd": ["graph"]},
                                              {"removed_kwd": "Y"},
                                              search.index_name(tenant_id), doc.kb_id)
                  settings.docStoreConn.delete({"kb_id": doc.kb_id, "knowledge_graph_kwd": ["entity", "relation", "graph", "subgraph", "community_report"], "must_not": {"exists": "source_id"}},
                                              search.index_name(tenant_id), doc.kb_id)
          except Exception:
              pass
          return cls.delete_by_id(doc.id)

• 并行度与异步处理： 代码使用线程池并行解析，但嵌入计算仍在主线程同步执行：线程池收集所有文档片段后，依次对每个文档调用嵌入模型编码。若解析速度远快于嵌入，则 CPU 可能空闲等待 GPU 或外部 API 响应。

  • 优化建议： 引入异步任务队列或多线程/多进程并发嵌入。例如，可在解析线程完成一个文档的切片后立即异步提交嵌入任务，而不是等所有解析结束；或者利用多个 GPU/进程并行处理不同文档的向量生成，从而提高整体吞吐。还可以考虑批量嵌入的异步 I/O，例如使用 asyncio 批发调用 OpenAI 等服务接口，加速远程嵌入 API 调用。

  def embedding(doc_id, cnts, batch_size=16):
          nonlocal embd_mdl, chunk_counts, token_counts
          vects = []
          for i in range(0, len(cnts), batch_size):
              vts, c = embd_mdl.encode(cnts[i: i + batch_size])
              vects.extend(vts.tolist())
              chunk_counts[doc_id] += len(cnts[i:i + batch_size])
              token_counts[doc_id] += c
          return vects

• 解析器性能与定制： RagFlow 支持丰富格式，但解析依赖第三方库（如 Aspose 用于 PPT, PyPDF 用于 PDF, OCR 库用于图片）可能成为瓶颈。例如，大 PDF 的 OCR 耗时很长。

  • 优化建议： 针对常用格式引入更高效的解析工具或并行 OCR 方案；提供解析参数调优（如是否提取图片、OCR 精度等）以在性能和精度间平衡。对于企业场景，可允许用户预先转换好部分数据格式，减少在线解析工作量。高精度场景可以提供多种方法及模型等配置选项，如 MinerU 等。

for i, d in enumerate(cks):
    v = vects[i]
    d["q_%d_vec" % len(v)] = v
for b in range(0, len(cks), es_bulk_size=64):
    if try_create_idx:
        if not settings.docStoreConn.indexExist(idxnm, kb_id):
            settings.docStoreConn.createIdx(idxnm, kb_id, len(vects[0]))
        try_create_idx = False
    settings.docStoreConn.insert(cks[b:b + es_bulk_size], idxnm, kb_id)

Retriever 模块

global DOC_ENGINE, docStoreConn, retrievaler, kg_retrievaler
DOC_ENGINE = os.environ.get("DOC_ENGINE", "elasticsearch")

lower_case_doc_engine = DOC_ENGINE.lower()
if lower_case_doc_engine == "elasticsearch":
        docStoreConn = rag.utils.es_conn.ESConnection()
elif lower_case_doc_engine == "infinity":
        docStoreConn = rag.utils.infinity_conn.InfinityConnection()
elif lower_case_doc_engine == "opensearch":
        docStoreConn = rag.utils.opensearch_coon.OSConnection()
else:
        raise Exception(f"Not supported doc engine: {DOC_ENGINE}")

retrievaler = search.Dealer(docStoreConn)
kg_retrievaler = kg_search.KGSearch(docStoreConn)

• 问：如何增加新的向量库类型？ 答：尽管社区中还有 FAISS、Wea viate、Milvus 等向量库，当前源码主要直接支持上述三种后端。但 RagFlow 架构上具备扩展能力：开发者可以参考现有 DocStoreConnection 接口封装新的后端（例如利用 FAISS 在本地内存建立索引，或通过 Wea viate/Milvus 的 Python 客户端发送查询）。引入这些向量库有望在纯向量检索性能和分布式扩展上提供优势。

def get_vector(self, txt, emb_mdl, topk=10, similarity=0.1):
        qv, _ = emb_mdl.encode_queries(txt)
        shape = np.array(qv).shape
        if len(shape) > 1:
            raise Exception(
                f"Dealer.get_vector returned array's shape {shape} doesn't match expectation(exact one dimension).")
        embedding_data = [get_float(v) for v in qv]
        vector_column_name = f"q_{len(embedding_data)}_vec"
        return MatchDenseExpr(vector_column_name, embedding_data, 'float', 'cosine', topk, {"similarity": similarity})

matchText, keywords = self.qryr.question(qst, min_match=0.3)
if emb_mdl is None:
    matchExprs = [matchText]
    res = self.dataStore.search(src, highlightFields, filters, matchExprs, orderBy, offset, limit,
                                idx_names, kb_ids, rank_feature=rank_feature)
    total = self.dataStore.getTotal(res)
    logging.debug("Dealer.search TOTAL: {}".format(total))
else:
    matchDense = self.get_vector(qst, emb_mdl, topk, req.get("similarity", 0.1))
    q_vec = matchDense.embedding_data
    src.append(f"q_{len(q_vec)}_vec")
    fusionExpr = FusionExpr("weighted_sum", topk, {"weights": "0.05, 0.95"})
    matchExprs = [matchText, matchDense, fusionExpr]
    res = self.dataStore.search(src, highlightFields, filters, matchExprs, orderBy, offset, limit,
                                idx_names, kb_ids, rank_feature=rank_feature)
    total = self.dataStore.getTotal(res)
    logging.debug("Dealer.search TOTAL: {}".format(total))
    # If result is empty, try again with lower min_match
    if total == 0:
        if filters.get("doc_id"):
            res = self.dataStore.search(src, [], filters, [], orderBy, offset, limit, idx_names, kb_ids)
            total = self.dataStore.getTotal(res)
        else:
            matchText, _ = self.qryr.question(qst, min_match=0.1)
            filters.pop("doc_id", None)
            matchDense.extra_options["similarity"] = 0.17
            res = self.dataStore.search(src, highlightFields, filters, [matchText, matchDense, fusionExpr],
                                        orderBy, offset, limit, idx_names, kb_ids, rank_feature=rank_feature)
            total = self.dataStore.getTotal(res)
            logging.debug("Dealer.search 2 TOTAL: {}".format(total))

class SearchResult:
        total: int
        ids: list[str]
        query_vector: list[float] | None = None
        field: dict | None = None
        highlight: dict | None = None
        aggregation: list | dict | None = None
        keywords: list[str] | None = None
        group_docs: list[list] | None = None

• 召回率不稳定： 稠密检索依赖向量质量，若嵌入模型未涵盖领域知识，可能漏掉相关内容；稀疏检索则受限于文本关键字匹配。融合后效果一般较单一路径好，但仍需调优参数（如向量与文本匹配的相对权重、min_match 阈值等）。目前代码里 similarity 下限等参数硬编码（如再次检索时将 matchDense 阈值设 0.17），这些可能需要根据实际数据调整。

  • 改进建议： 引入基于训练的数据驱动方法优化融合策略，例如学习一个小模型来判断向量匹配结果是否可靠，从而动态调整 dense vs. sparse 权重。此外可以增加对查询意图的判断：当用户问句包含具体实体/关键词时，加强 BM25 权重；当问句偏概念模糊时，更多依赖向量召回。

• 检索速度较慢： 使用 ElasticSearch/OpenSearch 进行大规模向量检索，在向量维度高和数据量大时性能可能不如专门的向量数据库（如 Milvus、FAISS）。Infinity 引擎采用 HNSW 提升了稠密检索速度，但如果知识库规模达到千万级，单机 HNSW 内存和查询延迟都会增长。

  • 优化方案： 引入更高性能的向量索引：例如将 Faiss 纳入后端，通过 GPU 加速近似最近邻搜索；或集成 Milvus 等支持分布式分片，以水平扩展大数据量下的查询。另外，可对现有 HNSW 参数调优（如 M、ef 值）或启用压缩降低内存占用。对于稀疏检索部分，可提前对文本做关键词过滤或索引精简（如移除停用词）提升匹配效率。

• 多 GPU/分布式支持不足： 当前检索流程主要在单实例完成，并未充分利用多 GPU 并行。Infinity 等向量库如果部署在多节点，需要协调多服务器的检索结果合并，这部分框架暂未体现。另外，在本地调用 Faiss 时，可以利用多 GPU 进行 IndexShard 或数据并行搜索，但目前架构未涵盖。

  • 改进建议： 提供分布式检索模式的支持，例如通过 gRPC 调用远程向量引擎集群；在应用层对大索引做分片并行检索，然后融合结果。此外，如上所述允许 embedding 环节利用多 GPU 生成查询向量，也能稍微减轻单卡压力。值得注意的是，多 GPU 更多地会用在嵌入阶段，检索阶段更多是 CPU/内存瓶颈，因此提升检索并发能力和索引优化更为重要。

• 问：如何调整稠密向量与稀疏文本匹配的融合权重？ 答：在 FusionExpr 中，weights 参数控制了融合比例。例如 {"weights": "0.05, 0.95"} 表示文本匹配权重为 0.05，向量匹配权重为 0.95。你可以根据知识库特性和业务需求调整这两个值，以达到最优的检索效果。通常建议在测试集上进行 A/B 测试来确定最佳权重。
• 问：检索结果为空时，系统会如何处理？ 答：如上文代码所示，当首次检索结果为空时，系统会尝试降低 min_match 阈值（从 0.3 降到 0.1），并放宽向量相似度门槛（从 0.1 降到 0.17），然后使用更低的标准重新发起检索，尽力找到相关片段。如果仍无结果，则可能是知识库中确实缺乏相关内容。

嵌入（Embedding）模块

• 本地向量模型： 默认使用开源的通用文本嵌入模型，代码中 DefaultEmbedding 会通过 FlagEmbedding 库加载指定名称的模型权重（默认 BAAI/bge-large-zh-v1.5），并将其缓存在本地。此外还有 FastEmbed 实现，使用轻量模型（如 bge-small-en-v1.5）及优化推理库以提升嵌入计算速度。另一个本地模型是 YoudaoEmbed，集成了网易有道的中英双语 Embedding 模型。这些本地模型在初始化时如果未缓存，会自动从 HuggingFace Hub 下载权重。
• OpenAI 等云嵌入服务： 模块提供 OpenAIEmbed 类封装 OpenAI 的文本嵌入 API（默认模型 text-embedding-ada-002）。AzureEmbed 支持 Azure OpenAI 服务。国内云服务也有：如 BaiChuanEmbed 对接百川智能的 Embedding API，QWenEmbed 通过阿里达摩院提供的 DashScope 接口获取其 Embedding 模型（如通义千问 embedding），ZhipuEmbed 对接智谱 AI 的 Embedding 接口。还有 JinaEmbed 用于调用 Jina AI 提供的向量服务接口。这些云服务类通常封装 HTTP 请求调用，并解析返回的 embedding 向量。例如 JinaEmbed 中直接以 requests.post 发送 JSON 请求并获取结果。
• 本地部署的 Embedding 服务： 模块也考虑了用户自托管的大模型服务，如 LocalAIEmbed 用于对接本地的开源 Embedding API（OpenAI 兼容接口），OllamaEmbed 用于调用本地的 Ollama 引擎生成 embedding（支持一些本地 LLM），XinferenceEmbed 对接 XInference 服务等。这些类通常复用 OpenAI 接口逻辑但将 base_url 指向本地服务器地址，从而利用本地算力生成向量。

BUILTIN_EMBEDDING_MODELS = ["BAAI/bge-large-zh-v1.5@BAAI", "maidalun1020/bce-embedding-base_v1@Youdao"]

class ZhipuEmbed(Base):
    def __init__(self, key, model_name="embedding-2", **kwargs):
        self.client = ZhipuAI(api_key=key)
        self.model_name = model_name

    def encode(self, texts: list):
        arr = []
        tks_num = 0
        MAX_LEN = -1
        if self.model_name.lower() == "embedding-2":
            MAX_LEN = 512
        if self.model_name.lower() == "embedding-3":
            MAX_LEN = 3072
        if MAX_LEN > 0:
            texts = [truncate(t, MAX_LEN) for t in texts]

        for txt in texts:
            res = self.client.embeddings.create(input=txt,
                                                model=self.model_name)
            arr.append(res.data[0].embedding)
            tks_num += self.total_token_count(res)
        return np.array(arr), tks_num

    def encode_queries(self, text):
        res = self.client.embeddings.create(input=text,
                                            model=self.model_name)
        return np.array(res.data[0].embedding), self.total_token_count(res)

class LocalAIEmbed(Base):
    def __init__(self, key, model_name, base_url):
        if not base_url:
            raise ValueError("Local embedding model url cannot be None")
        if base_url.split("/")[-1] != "v1":
            base_url = os.path.join(base_url, "v1")
        self.client = OpenAI(api_key="empty", base_url=base_url)
        self.model_name = model_name.split("___")[0]

    def encode(self, texts: list):
        batch_size = 16
        ress = []
        for i in range(0, len(texts), batch_size):
            res = self.client.embeddings.create(input=texts[i:i + batch_size], model=self.model_name)
            ress.extend([d.embedding for d in res.data])
        # local embedding for LmStudio donot count tokens
        return np.array(ress), 1024

    def encode_queries(self, text):
        embds, cnt = self.encode([text])
        return np.array(embds[0]), cnt

• 批量并行： 对于上万段文本的嵌入，可将其划分给多个进程/线程分别处理，比如启动多个 EmbeddingWorker 进程，各自加载模型处理一部分数据，然后合并结果。这在多核 CPU 或多 GPU 场景下尤为有效。也可使用向量化工具（如 NumPy 并行、PyTorch DataLoader）对 CPU 计算部分提速。
• 模型压缩和加速： 针对大模型嵌入的性能问题，可引入蒸馏、小模型（如使用 small 版 BGE 或其他轻量模型）在对性能要求高时使用；或者使用 OpenVINO、TensorRT 等对 embedding 模型做推理加速。FastEmbed 类就是一个采用优化库的例子，其调用的 fastembed.TextEmbedding 内部可能利用 C++ 实现提升计算速度。进一步的优化可以考虑量化 embedding 模型的权重，用 FP16 甚至 INT8 以换取更高速度和更低内存占用。
• 流水线并行： 将文档解析->嵌入->存储三步解耦成流水线，不必串行等待。比如在一个文档解析出部分 chunk 后，就立刻交给嵌入线程处理，同时解析线程继续处理下一个部分，从而解析和嵌入两个阶段重叠执行，提高硬件利用率。需要在架构上实现任务调度和缓冲队列，防止某一步过慢拖累整体。

• 问：如何选择合适的嵌入模型？ 答：选择嵌入模型通常取决于几个关键因素：是否支持你的语言（如中文、英文或双语）、对向量维度的要求（高精度场景可能需要高维向量）、推理速度与成本（本地推理 vs. 云端 API 调用）。RagFlow 提供了丰富的选项，建议先用默认的 BGE-large-zh-v1.5 进行测试，如果性能和准确性满足要求则无需更换。如果需要更快的处理速度或更低的资源消耗，可以尝试 FastEmbed 或小模型。
• 问：遇到 Token 长度限制怎么办？ 答：代码已经对长文本做了截断处理，确保不会超出模型支持的最大长度。但如果截断导致关键信息丢失，可以考虑：1) 在解析阶段将文档切分成更短的片段；2) 使用支持更长上下文的嵌入模型（如 OpenAI 的 text-embedding-3-large 支持 8191 tokens）；3) 对超长段落进行分段嵌入后取均值作为该段落的最终向量。
• 问：如何加速本地嵌入模型的推理？ 答：可以通过多种方式加速：使用 FastEmbed 类（它内置了优化）；量化模型（如使用 FP16 或 INT8）；启用更大的批处理大小（batch_size）；如果有多张 GPU，可以考虑将模型复制到多卡上并行处理。另外，使用更小的模型（如 bge-small 而不是 bge-large）也能显著提升推理速度。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：RAGFlow架构详解与性能优化方案要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题