RAGFlow中Embeddings模型选择与向量数据库选型实现分析

承接之前对 RAGFlow 的探讨，这篇来重点拆解一下它在 Embeddings 模型选型与配置，以及向量数据库实现方面的技术细节。这部分内容主要基于源码 `rag/llm/embedding_model.py` 和 `rag/utils/doc_store_conn.py`，以及官方文档，信息量不小，值得仔细过一遍。

1. Embeddings 模型选择与配置

RAGFlow 在 Embedding 模型这一层做得相当灵活。它没有把某个模型写死，而是通过 `rag/llm/embedding_model.py` 构建了一套通用的接口和配置体系。

1.1 模型架构设计

它采用了一个抽象基类的设计模式，核心就是那个 `Base` 类。这个基类定义了所有 Embedding 模型需要实现的基本接口，比如 `encode`（把文本列表转成向量）、`encode_queries`（专门编码查询语句）、以及计算 token 消耗的方法。来看一下骨架：

class Base(ABC):
    def __init__(self, key, model_name):
        pass

    def encode(self, texts: list):
        raise NotImplementedError("Please implement encode method!")

    def encode_queries(self, text: str):
        raise NotImplementedError("Please implement encode method!")

    def total_token_count(self, resp):
        try:
            return resp.usage.total_tokens
        except Exception:
            pass
        try:
            return resp["usage"]["total_tokens"]
        except Exception:
            pass
        return 0

有了这个顶层设计，后续无论对接哪个厂家的模型，都只需要继承 `Base` 并实现那几个核心方法就行。扩展起来非常清爽。

1.2 支持的 Embedding 模型

从源码里梳理，RAGFlow 目前支持九种 Embedding 模型，覆盖了主流厂商和开源方案：

DefaultEmbedding：这是默认的开箱方案，基于 FlagEmbedding 框架，底层用的是 BAAI/bge-large-zh-v1.5。代码里可以看到它做了不少细节处理，比如用线程锁保证全局单例、自动下载模型到缓存目录、开启 FP16 加速（如果有 GPU 的话）。

class DefaultEmbedding(Base):
    os.environ['CUDA_VISIBLE_DEVICES'] = '0'
    _model = None
    _model_name = ""
    _model_lock = threading.Lock()

    def __init__(self, key, model_name, **kwargs):
        if not settings.LIGHTEN:
            with DefaultEmbedding._model_lock:
                from FlagEmbedding import FlagModel
                import torch
                if not DefaultEmbedding._model or model_name != DefaultEmbedding._model_name:
                    try:
                        DefaultEmbedding._model = FlagModel(
                            os.path.join(get_home_cache_dir(), re.sub(r"^[a-zA-Z0-9]+/", "", model_name)),
                            query_instruction_for_retrieval="为这个句子生成表示以用于检索相关文章：",
                            use_fp16=torch.cuda.is_a vailable())
                        DefaultEmbedding._model_name = model_name
                    except Exception:
                        model_dir = snapshot_download(
                            repo_id="BAAI/bge-large-zh-v1.5",
                            local_dir=os.path.join(get_home_cache_dir(), re.sub(r"^[a-zA-Z0-9]+/", "", model_name)),
                            local_dir_use_symlinks=False)
                        DefaultEmbedding._model = FlagModel(model_dir, ...)
                self._model = DefaultEmbedding._model
                self._model_name = DefaultEmbedding._model_name

OpenAIEmbed：这个很直接，就是用 OpenAI 官方 API。默认模型是 text-embedding-ada-002，也支持自定义 base_url 来对接袋里或其他兼容接口。

class OpenAIEmbed(Base):
    def __init__(self, key, model_name="text-embedding-ada-002", base_url="https://api.openai.com/v1"):
        if not base_url:
            base_url = "https://api.openai.com/v1"
        self.client = OpenAI(api_key=key, base_url=base_url)
        self.model_name = model_name

LocalAIEmbed：适合本地部署的场景，通过 OpenAI 兼容接口来调用本地模型服务。注意它强制要求传入 base_url，并且 API key 直接写死为 "empty"。

AzureEmbed：专门为 Azure OpenAI 服务做的封装，继承自 OpenAIEmbed，但内部换用了 AzureOpenAI 客户端。API key 和版本号通过一个 JSON 字符串传递。

BaiChuanEmbed、QWenEmbed、ZhipuEmbed、OllamaEmbed、GoogleEmbed：这几家分别对应百川、通义千问、智谱、Ollama 和 Google 的 Embedding 服务。其中 Ollama 实现比较有意思，它支持通过 header 传递 Bearer Token 来做鉴权。

可以说，从国产大模型到国际巨头再到本地开源，RAGFlow 几乎覆盖了目前市面上所有主流的 Embedding 服务。

1.3 批处理优化

在实际生产中，Embedding 计算往往需要处理大量文本片段。RAGFlow 在各个模型实现里都加入了批处理逻辑，拿默认模型的 `encode` 方法为例：

def encode(self, texts: list):
    batch_size = 16
    texts = [truncate(t, 2048) for t in texts]
    token_count = 0
    for t in texts:
        token_count += num_tokens_from_string(t)
    ress = []
    for i in range(0, len(texts), batch_size):
        ress.extend(self._model.encode(texts[i:i + batch_size]).tolist())
    return np.array(ress), token_count

不同模型的批大小略有差异：DefaultEmbedding 和 OpenAIEmbed 是 16，QWenEmbed 则保守一些设为 4。这么做的好处很明显——减少 API 的调用频率，提升整体吞吐。

1.4 文本长度处理

并不是所有模型的“胃口”都一样大。比如 OpenAI 的模型最多能消化 8191 个 token，而通义千问和智谱的 Embedding-2 模型则分别限制在 2048 和 512 个 token。RAGFlow 在调用前做了统一的截断处理，避免因为文本过长而报错。

# OpenAIEmbed中
texts = [truncate(t, 8191) for t in texts]

# QWenEmbed中
texts = [truncate(t, 2048) for t in texts]

# ZhipuEmbed中
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]

这一点看似简单，但在实际部署中非常关键——不少线上故障就是忘了做这个处理导致的。

1.5 默认模型选择

根据官方文档，RAGFlow 的 Docker 镜像（非 slim 版）预装了两种优化过的 Embedding 模型：

• BAAI/bge-large-zh-v1.5
• maidalun1020/bce-embedding-base_v1

如果不做任何配置，系统默认会使用 BAAI/bge-large-zh-v1.5。这两个模型在中英文场景下表现都相当不错，对多语言支持也做了针对性优化。

2. 向量数据库选型与实现

说完了 Embedding 模型，再看向量数据库这一层。RAGFlow 同样没有绑定某个具体产品，而是通过一个抽象的接口层来对接不同的向量存储方案。

2.1 数据库抽象接口

核心接口定义在 `rag/utils/doc_store_conn.py` 中，叫作 `DocStoreConnection`。它规定了每个向量数据库必须实现的几大类操作：

• 数据库类型与健康检查：`dbType()` 和 `health()`
• 索引管理：`createIdx()`, `deleteIdx()`, `indexExist()`
• CRUD 与搜索：`search()`, `insert()`, `update()`, `delete()`

看一段搜索方法的签名就能感受到设计的通用性：

@abstractmethod
def search(
    self, selectFields: list[str], highlightFields: list[str],
    condition: dict, matchExprs: list[MatchExpr],
    orderBy: OrderByExpr, offset: int, limit: int,
    indexNames: str|list[str], knowledgebaseIds: list[str],
    aggFields: list[str] = [], rank_feature: dict | None = None
):
    raise NotImplementedError("Not implemented")

这种抽象层次使得未来想切换到 Milvus、Pinecone 或其他向量库时，只需要新增一个实现类即可，对上层业务逻辑完全透明。

2.2 OpenSearch 实现

当前默认的实现是 OpenSearch，代码在 `rag/utils/opensearch_coon.py`。`OSConnection` 类用单例模式包装，启动时会自动尝试连接，并最多重试若干次。

@singleton
class OSConnection(DocStoreConnection):
    def __init__(self):
        self.info = {}
        logger.info(f"Use OpenSearch {settings.OS['hosts']} as the doc engine.")
        for _ in range(ATTEMPT_TIME):
            try:
                self.os = OpenSearch(
                    settings.OS["hosts"].split(","),
                    http_auth=(settings.OS["username"], settings.OS["password"])
                        if "username" in settings.OS and "password" in settings.OS else None,
                    verify_certs=False,
                    timeout=600
                )
                if self.os:
                    self.info = self.os.info()
                    break
            except Exception as e:
                logger.warning(f"{str(e)}. Waiting OpenSearch {settings.OS['hosts']} to be healthy.")
                time.sleep(5)

连接参数（如 hosts、用户名、密码）均通过配置文件或环境变量注入，灵活性很高。

2.3 索引设计

RAGFlow 为每个知识库单独创建一个索引（Index），命名规则是 `ragflow_{uid}`，其中 `uid` 是知识库的唯一标识。创建索引时使用的 mapping 结构定义在 `conf/os_mapping.json` 文件中，里面包含了文本字段、向量字段以及各种检索所需的元数据字段。

def createIdx(self, indexName: str, knowledgebaseId: str, vectorSize: int):
    if self.indexExist(indexName, knowledgebaseId):
        return True
    try:
        from opensearchpy.client import IndicesClient
        return IndicesClient(self.os).create(index=indexName, body=self.mapping)
    except Exception:
        logger.exception("OSConnection.createIndex error %s" % (indexName))

这种每个知识库独立索引的设计，天然就支持了数据隔离和多租户场景。

2.4 混合检索策略

这是 RAGFlow 检索能力的一个亮点。它没有简单只做向量相似度匹配，而是把关键词搜索（全文检索）和向量搜索融合在了一起。

在 `search` 方法中，系统会同时处理 `MatchTextExpr`（文本匹配）和 `MatchDenseExpr`（稠密向量匹配），然后把两者的结果通过加权求和的方式合并。代码中可以看到，系统首先解析出 `FusionExpr` 中配置的权重，然后分别构建关键词查询（query_string）和 KNN 向量查询：

def search(...):
    use_knn = False
    # ... 构建布尔查询条件 ...
    s = Search()
    vector_similarity_weight = 0.5

    for m in matchExprs:
        if isinstance(m, FusionExpr) and m.method == "weighted_sum" and "weights" in m.fusion_params:
            # 解析权重，例如 "0.7,0.3"
            weights = m.fusion_params["weights"]
            vector_similarity_weight = float(weights.split(",")[1])

    knn_query = {}
    for m in matchExprs:
        if isinstance(m, MatchTextExpr):
            # 关键词搜索
            bqry.must.append(Q("query_string", fields=m.fields,
                type="best_fields", query=m.matching_text,
                minimum_should_match=..., boost=1))
            bqry.boost = 1.0 - vector_similarity_weight
        elif isinstance(m, MatchDenseExpr):
            # 向量相似度搜索
            use_knn = True
            knn_query[vector_column_name] = {
                "vector": list(m.embedding_data),
                "k": m.topn,
                "filter": bqry.to_dict(),
                "boost": similarity
            }

也就是说，最终的检索结果是关键词精确匹配和语义相似度搜索的结合体，两者的权重默认是五五开，但用户可以自行调整。

2.5 权重配置

前面已经提到了权重的解析逻辑。系统通过一个 `FusionExpr` 来承载权重配置，默认向量相似度权重是 0.5。用户可以在创建对话或调整检索策略时，根据业务需要修改这个值。比如在需要高精度匹配的场景下，调高关键词搜索的权重；而在需要更广泛语义理解时，加大向量搜索的占比。

2.6 批量操作优化

在高吞吐场景下，单条插入效率很低。RAGFlow 的 `insert` 方法利用 OpenSearch 的 bulk API 来实现批量写入。代码会收集一批文档，构建成 bulk action 列表，然后一次性提交：

def insert(self, rows: list[dict], indexName: str, knowledgebaseId: str = None) -> list[str]:
    if len(rows) == 0:
        return []
    actions = []
    ids = []
    for row in rows:
        # 处理_id字段
        action = {"_index": indexName, "_source": row}
        if _id:
            action["_id"] = _id
        actions.append(action)
    try:
        from opensearchpy.helpers import bulk
        success, failed = bulk(self.os, actions, stats_only=True)
        return ids
    except Exception:
        logger.exception("OSConnection.insert error")
        return []

这种做法能大幅减少网络往返次数，在文档量较大的知识库构建阶段尤其显著。

3. 配置与集成

3.1 Embedding 模型配置

RAGFlow 允许为每个知识库独立选择 Embedding 模型。官方的说明很直白：一旦知识库中有了 Chunk，就不能再更换 Embedding 模型。如果要换，必须先把现有 Chunk 全部删掉。理由也很简单——所有文档必须用同一个模型生成向量，才能保证它们在同一个向量空间下做比较。这是确保检索准确性的基本前提。

3.2 向量数据库配置

通过 `settings.OS` 这个配置对象，可以指定 OpenSearch 的 host 列表、认证信息等。这些配置既可以在环境变量中定义，也可以通过配置文件加载。比如常见的 Docker 部署中，会通过环境变量 `OPENSEARCH_HOSTS` 来传入地址。

3.3 检索参数配置

在配置对话或检索参数时，有两个关键指标值得关注：

• 相似度阈值（Similarity threshold）：低于该阈值的 Chunk 会被过滤掉，默认是 0.2。
• 向量相似度权重（Vector similarity weight）：向量相似度在最终得分中的占比，默认是 0.3（注意这个默认值和代码里 FusionExpr 的初始值 0.5 可能不同，具体以实际配置为准）。

这两个参数直接影响检索结果的质量和覆盖范围，值得根据业务场景反复调优。

4. 总结

梳理下来看，RAGFlow 在这两个核心技术点上确实有不少可圈可点之处：

• 多模型支持：从 OpenAI 到智谱、通义千问，再到本地开源的 FlagEmbedding，几乎覆盖了所有主流方案。
• 批处理与长度控制：内置了合理的批处理策略和文本截断逻辑，避免了许多常见的 API 调用陷阱。
• 一致性保证：一个知识库内部强制使用同一种 Embedding 模型，确保向量空间的一致性。
• 灵活的配置体系：支持为不同知识库选择不同模型，且所有参数均可通过配置或环境变量调整。
• 默认 OpenSearch 引擎：作为默认的向量数据库，性能表现不错，并且扩展性良好。
• 混合检索策略：关键词 + 向量的加权融合方案，兼顾了精确匹配和语义理解。
• 批量操作优化：利用 bulk API 大幅提升了写入和检索的效率。

这些设计让 RAGFlow 在面对不同数据规模、不同业务需求的场景时，都能保持较好的适应性和检索质量。当然，实际生产环境中效果如何，还需要结合具体的模型选择、数据量和检索权重来做仔细的测试和调优。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：RAGFlow中Embeddings模型选择与向量数据库选型实现分析要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题