# [S2-T3] 向量数据库选型、索引与检索策略实现详解 > 范围：`api/app/core/rag/vdb/elasticsearch/`、`api/app/core/rag/utils/es_conn.py`、`api/app/core/rag/utils/doc_store_conn.py`、`api/app/core/rag/nlp/{search.py, query.py}`、`api/app/core/rag/res/mapping.json` 以及调用方 `api/app/core/workflow/nodes/knowledge/node.py`、`api/app/services/memory_konwledges_server.py` 等。 > > 提示：MemoryBear 当前版本中存在**两套并行的 ES 实现路径**，本文会逐一拆开说明，并给出二者的边界与实际调用方。 --- ## 一、一句话定位 MemoryBear 使用 **Elasticsearch 8.x** 作为向量 + 全文一体化的检索引擎，通过 `dense_vector` (HNSW) 实现语义检索、Lucene + IK 分词器实现关键词检索，并在应用层与 ES DSL 层各自实现一套"混合搜索"策略（应用层为"双路 + 去重 + 可选 Rerank"，DSL 层为"weighted_sum 加权融合"）。 ## 二、设计目标与选型说明 ### 2.1 选型动机（为什么是 Elasticsearch 而非 Milvus / Qdrant / Pinecone？） README 中明确把 **"Hybrid Search: Keyword + Semantic Vector"** 列为产品级核心能力之一（README.md:62-66）。结合源码可以推出三条关键决策依据： 1. **关键词侧需要 Lucene 生态** — 既要中文分词（IK `ik_max_word`），又要 BM25 / 布尔过滤 / 高亮 / 同义词扩展 / 短语匹配 / 字段权重等成熟能力，Milvus / Qdrant / Pinecone 在这一侧几乎都需要外接 ES/OS。`api/app/core/rag/nlp/query.py:14-22` 的 `query_fields = ["title_tks^10", "important_kwd^30", "content_ltks^2", ...]` 就是典型 Lucene field-boost 写法，离开 ES 改造代价很高。 2. **一份索引同时承担多种载荷** — 一个 ES 索引同时存储 chunk 文本 (`page_content`)、向量 (`*_vec`)、稀疏 tokens (`*_tks` / `*_ltks`)、标签 rank_features (`tag_feas`)、PageRank-like 分数 (`pagerank_fea`)、地理 (`lat_lon`)、嵌套结构 (`*_nst`) 等异构字段（见 `api/app/core/rag/res/mapping.json:25-209`）。专用向量库无法承载这种混合 schema。 3. **运维与生态成本** — 团队仅运行 PostgreSQL / Neo4j / Redis / ES（README "Prerequisites"），引入第二套向量服务会显著抬高运维曲线。`@singleton` 的 `ESConnection` (`api/app/core/rag/utils/es_conn.py:26-56`) 与 `ElasticSearchVectorFactory._client` (`api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:666-732`) 共享连接，工程上已经按"单实例多用途"在使用 ES。 > 代价：ES 的 ANN 在百万-千万 chunk 时延迟会明显高于 Milvus/Qdrant；当未来 chunk 量级或 QPS 显著增长时，本架构需要拆出独立向量服务（详见 §6 优化建议）。 ### 2.2 ES 版本约束 启动期硬性校验 ES 必须 ≥ 8.0： ```python # api/app/core/rag/utils/es_conn.py:44-49 v = self.info.get("version", {"number": "8.0.0"}) v = v["number"].split(".")[0] if int(v) < 8: msg = f"Elasticsearch version must be greater than or equal to 8, current version: {v}" logger.error(msg) raise Exception(msg) ``` ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:716-722 if not cls._version_checked: info = client.info() version = info["version"]["number"] if parse_version(version) < parse_version("8.0.0"): raise ValueError(f"Elasticsearch version must be >= 8.0.0, got {version}") ``` > **why**：ES 8.0 才正式提供 `dense_vector` HNSW 索引、`knn` 顶层查询、以及 query_string + knn 的混合检索，本系统的 `s.knn(...)`、`type=dense_vector index=true similarity=cosine` 都依赖该版本。 --- ## 三、模块结构与两条实现路径 ``` api/app/core/rag/ ├── res/mapping.json ← graphrag_{workspace_id} 索引使用的全局 mapping ├── utils/ │ ├── doc_store_conn.py ← 抽象接口 DocStoreConnection + MatchExpr / FusionExpr │ └── es_conn.py ← @singleton 实现 DocStoreConnection（路径 A） ├── vdb/ │ ├── vector_base.py ← 抽象接口 BaseVector │ ├── field.py ← page_content / metadata / vector 等字段名常量 │ └── elasticsearch/ │ └── elasticsearch_vector.py ← BaseVector 的 ES 实现（路径 B） ├── nlp/ │ ├── search.py ← 同时承载两条路径：knowledge_retrieval（路径 B）+ Dealer（路径 A） │ └── query.py ← FulltextQueryer，构造 Lucene query_string └── common/ ├── settings.py ← 全局初始化 docStoreConn / retriever / kg_retriever └── constants.py ← PAGERANK_FLD / TAG_FLD 等常量 ``` ### 3.1 路径 A：`ESConnection`（DSL 抽象层，主要服务于 GraphRAG 与高级检索） - 抽象基类：`api/app/core/rag/utils/doc_store_conn.py:128-256` 定义 `DocStoreConnection` 接口（dbType / createIdx / search / insert / update / delete / sql 等）。 - 表达式族：同文件 43-126 行定义 `MatchTextExpr`、`MatchDenseExpr`、`MatchSparseExpr`、`MatchTensorExpr`、`FusionExpr`、`OrderByExpr` —— 这是上层与底层解耦的"查询 IR"。 - ES 实现：`@singleton class ESConnection(DocStoreConnection)`（`api/app/core/rag/utils/es_conn.py:26-634`）。 - 全局入口：`api/app/core/rag/common/settings.py:13-24` 在模块导入时即 `init_settings()`，把 `ESConnection()` 装进 `docStoreConn`，并注入 `Dealer` / `KGSearch`。 - 对应的检索门面：`api/app/core/rag/nlp/search.py: Dealer`（350-907 行），由 `kg_retriever`、`retriever` 全局共用。 ### 3.2 路径 B：`ElasticSearchVector`（应用层 BaseVector，主要服务于 KB 节点 / 工作流） - 抽象基类：`api/app/core/rag/vdb/vector_base.py:9-67` 定义 `BaseVector` 接口（create / add_texts / search_by_vector / search_by_full_text / delete 等）。 - 字段命名：`api/app/core/rag/vdb/field.py`：`page_content` / `metadata` / `vector` / `metadata.doc_id` 等。 - ES 实现：`class ElasticSearchVector(BaseVector)` + `class ElasticSearchVectorFactory`（`api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:29-732`）。 - 关键调用方： - `api/app/core/workflow/nodes/knowledge/node.py:195-298` 工作流知识节点，按 `RetrieveType` 分支调用 `search_by_vector / search_by_full_text`。 - `api/app/core/rag/nlp/search.py: knowledge_retrieval`（36-147 行）API/服务层入口。 - `api/app/services/memory_konwledges_server.py`、`api/app/controllers/{chunk,document,knowledge}_controller.py` 等。 ### 3.3 两条路径的边界 | 维度 | 路径 A（ESConnection / Dealer） | 路径 B（ElasticSearchVector） | | --- | --- | --- | | 索引名 | `graphrag_{workspace_id}` | `Vector_index_{kb_id}_Node`（小写）| | 索引粒度 | 一个 workspace 一个 ES index，多 KB 用 `kb_id` 字段过滤 | 一个 KB 一个 ES index | | Mapping | `res/mapping.json` 全局 dynamic_templates | 代码里 inline 的 `index_mapping` (`elasticsearch_vector.py:616-661`) | | 文本字段 | `content_ltks` / `content_sm_ltks` / `title_tks` / `important_kwd` / `*_tks` | `page_content` (`text` + `ik_max_word`) | | 向量字段 | 动态 `q_{dim}_vec` (`*_512_vec` / `*_768_vec` / `*_1024_vec` / `*_1536_vec`) | 固定 `vector`（dim 取首批 embeddings 长度）| | 关键词检索 | Lucene `query_string`（field-boost、同义词、短语）| `match` + `analyzer=ik_max_word`（BM25）| | 向量检索 | `s.knn(...)`（HNSW，ES 8 原生 ANN）| `script_score` + `cosineSimilarity`（暴力，但精度高）| | 混合融合 | `FusionExpr("weighted_sum", weights="0.05,0.95")` 应用层加权 + ES 内部混合 | 双路并发查 → metadata.doc_id 去重 → 可选 reranker | | 主要使用者 | GraphRAG、`Dealer.retrieval()`、tag/citation 等高级能力 | 工作流知识节点、KB CRUD、召回测试 | > **why 不合并**：路径 A 携带丰富 IR（同义词扩展、`tag_feas`、`pagerank_fea`、`question_tks` 等），是面向"知识图谱 + 复杂 RAG"的；路径 B 简单直接，是工作流/服务层的"够用就好"封装。代码上是渐进演化中的双轨，但**目前两条路径都在生产使用**。 --- ## 四、索引设计 ### 4.1 全局 mapping（路径 A，`api/app/core/rag/res/mapping.json`） #### 4.1.1 settings ```json // api/app/core/rag/res/mapping.json:2-15 "settings": { "index": { "number_of_shards": 2, "number_of_replicas": 0, "refresh_interval": "1000ms" }, "similarity": { "scripted_sim": { "type": "scripted", "script": { "source": "double idf = Math.log(1+(field.docCount-term.docFreq+0.5)/(term.docFreq + 0.5))/Math.log(1+((field.docCount-0.5)/1.5)); return query.boost * idf * Math.min(doc.freq, 1);" } } } } ``` | 项 | 值 | 说明 | | --- | --- | --- | | `number_of_shards` | 2 | 适合中小型部署；超过 50GB / 单 shard 时需重新规划 | | `number_of_replicas` | 0 | **生产风险点**：单副本意味着任一分片丢失即数据丢失，建议生产环境改为 ≥1 | | `refresh_interval` | 1000ms | 默认 1s 即可见，写入吞吐场景可调高至 30s 或写入期 `-1` | | `scripted_sim` | 自定义 BM25 变体 | 用 `Math.min(doc.freq, 1)` 把词频压成 0/1，等价于 binary BM25——抑制高 TF 的关键字"灌水"，对 token 字段更鲁棒 | #### 4.1.2 dynamic_templates（按字段名后缀决定字段类型） ```json // api/app/core/rag/res/mapping.json:25-209（节选） { "int": { "match": "*_int", "mapping": { "type": "integer", "store": "true" }}}, { "ulong": { "match": "*_ulong", "mapping": { "type": "unsigned_long" }}}, { "long": { "match": "*_long", "mapping": { "type": "long" }}}, { "numeric": { "match": "*_flt", "mapping": { "type": "float" }}}, { "tks": { "match": "*_tks", "mapping": { "type": "text", "similarity": "scripted_sim", "analyzer": "whitespace" }}}, { "ltks": { "match": "*_ltks", "mapping": { "type": "text", "analyzer": "whitespace" }}}, { "kwd": { "match_pattern": "regex", "match": "^(.*_(kwd|id|ids|uid|uids)|uid)$", "mapping": { "type": "keyword", "similarity": "boolean" }}}, { "dt": { "match_pattern": "regex", "match": "^.*(_dt|_time|_at)$", "mapping": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||yyyy-MM-dd_HH:mm:ss" }}}, { "rank_feature": { "match": "*_fea", "mapping": { "type": "rank_feature" }}}, { "rank_features": { "match": "*_feas", "mapping": { "type": "rank_features" }}}, { "dense_vector": { "match": "*_512_vec", "mapping": { "type": "dense_vector", "index": true, "similarity": "cosine", "dims": 512 }}}, { "dense_vector": { "match": "*_768_vec", "mapping": { "type": "dense_vector", "index": true, "similarity": "cosine", "dims": 768 }}}, { "dense_vector": { "match": "*_1024_vec", "mapping": { "type": "dense_vector", "index": true, "similarity": "cosine", "dims": 1024 }}}, { "dense_vector": { "match": "*_1536_vec", "mapping": { "type": "dense_vector", "index": true, "similarity": "cosine", "dims": 1536 }}}, { "nested": { "match": "*_nst", "mapping": { "type": "nested" }}}, { "binary": { "match": "*_bin", "mapping": { "type": "binary" }}} ``` **why dynamic 而不是 strict mapping**： - 不同 embedding 模型维度不同（512/768/1024/1536），通过字段名后缀让"模型即维度"，在 `nlp/search.py:372` 看到查询侧动态拼名 `f"q_{len(embedding_data)}_vec"`，写入侧也是同样命名，零配置切换 embedding。 - token 字段分 `*_tks` 与 `*_ltks`：前者使用 `scripted_sim`（去 TF），用于 important_kwd 这类"命中即可"字段；后者 BM25 默认，用于正文型 `content_ltks`。 - `*_fea` (rank_feature) 与 `*_feas` (rank_features) 用于 PageRank 与 tag 加权，详见检索章节的 `_rank_feature_scores`。 > **why analyzer 是 `whitespace` 而不是 IK**：路径 A 在写入前先用 `rag_tokenizer` 在应用层做完中文分词，写入 ES 时已经是空格分隔的 tokens。这样"分词逻辑"留在应用层，便于热更新词典与同义词，不用 reindex。 ### 4.2 路径 B 的 inline mapping（KB 索引） ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:609-663 def create_collection(self, embeddings, metadatas=None, index_params=None): if not self._client.indices.exists(index=self._collection_name): index_mapping = { "mappings": { "properties": { Field.CONTENT_KEY.value: { # "page_content" "type": "text", "analyzer": "ik_max_word" }, Field.METADATA_KEY.value: { # "metadata" "type": "object", "properties": { "doc_id": {"type": "keyword"}, "file_id": {"type": "keyword"}, "file_name": {"type": "keyword"}, "file_created_at": {"type": "date", "format": "epoch_millis"}, "document_id": {"type": "keyword"}, "knowledge_id": {"type": "keyword"}, "sort_id": {"type": "long"}, "status": {"type": "integer"} } }, Field.VECTOR.value: { # "vector" "type": "dense_vector", "dims": len(embeddings[0]), "index": True, "similarity": "cosine" } } } } self._client.indices.create(index=self._collection_name, body=index_mapping) ``` 要点： - 索引按 KB 隔离：`collection_name = f"Vector_index_{knowledge.id}_Node"`（同文件 738 行），ES 端要求小写，所以 `super().__init__(index_name.lower())`（32 行）。 - `dims = len(embeddings[0])` —— 维度由"第一批数据"决定，**一旦确定不可改**。换 embedding 模型必须重建索引（详见 §6 风险点）。 - `similarity = "cosine"` —— 写入向量不要求归一化，由 ES 内部计算余弦相似度。 - 没有显式 `number_of_shards`/`replicas` 设置，**走 ES 集群默认**（8.x 默认 1 shard 1 replica），可用性比路径 A 反而更好；但碎片化风险也更高（每个 KB 一个 index，KB 多了 cluster state 会膨胀）。 ### 4.3 索引命名与隔离 | 路径 | 索引模板 | 来源 | | --- | --- | --- | | A | `graphrag_{workspace_id}` | `nlp/search.py:346` `def index_name(uid): return f"graphrag_{uid}"` | | B | `Vector_index_{kb_id}_Node`（小写）| `elasticsearch_vector.py:738` | 路径 A 在删除知识库时**故意不删 ES 索引**，而是仅删 `kb_id` 维度的文档： ```python # api/app/core/rag/utils/es_conn.py:115-124 def deleteIdx(self, indexName: str, knowledgebaseId: str): if len(knowledgebaseId) > 0: # The index need to be alive after any kb deletion since all kb under this workspace are in one index. return try: self.es.indices.delete(index=indexName, allow_no_indices=True) except NotFoundError: pass ``` > **why**：一个 workspace 多 KB 共享同一个 index，单 KB 删除不能动 index；只能在 `delete()` 通过 `condition["kb_id"]=knowledgebaseId` 走 delete-by-query（同文件 424-471）。 --- ## 五、写入链路 ### 5.1 路径 B：高层封装（KB / 工作流场景） ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:55-87 def add_chunks(self, chunks: list[DocumentChunk], **kwargs): texts = [chunk.page_content for chunk in chunks] if self.is_multimodal_embedding: embeddings = self.embeddings.embed_batch(texts) else: embeddings = self.embeddings.embed_documents(list(texts)) self.create(chunks, embeddings, **kwargs) def create(self, chunks, embeddings, **kwargs): metadatas = [chunk.metadata or {} for chunk in chunks] if not self._client.indices.exists(index=self._collection_name): self.create_collection(embeddings, metadatas) # 懒建索引 self.add_texts(chunks, embeddings, **kwargs) def add_texts(self, chunks, embeddings, **kwargs): uuids = self._get_uuids(chunks) actions = [] for i, chunk in enumerate(chunks): actions.append({ "_index": self._collection_name, "_source": { Field.CONTENT_KEY.value: chunk.page_content, Field.METADATA_KEY.value: chunk.metadata or {}, Field.VECTOR.value: embeddings[i] or None, } }) result = helpers.bulk(self._client, actions) return uuids ``` 特性： - **懒建索引**：第一次写入时根据 `len(embeddings[0])` 建 mapping。 - **批量写**：`elasticsearch.helpers.bulk` 默认 chunk_size=500、max_chunk_bytes=100MB；这里不传 `_id`，ES 自动生成。 - **唯一性**：路径 B 把 chunk 唯一标识放在 `metadata.doc_id`（`vector_base.py:62-63 _get_uuids`），更新/删除走"先 search by metadata.doc_id 拿真正 _id 再 bulk delete"两步走（`elasticsearch_vector.py:148-174`）。 - **失败处理**：`helpers.bulk` 默认抛 `BulkIndexError`，调用方在 `delete_by_ids / delete_by_metadata_field` 中分桶捕获 404 与其它错误（同文件 137-147、164-174）。**`add_texts` 没有捕获**——一旦底层网络失败会向上抛，调用方需要保证幂等性或重试。 ### 5.2 路径 A：抽象层批量写 ```python # api/app/core/rag/utils/es_conn.py:294-330 def insert(self, documents, indexName, knowledgebaseId=None) -> list[str]: operations = [] for d in documents: assert "_id" not in d assert "id" in d d_copy = copy.deepcopy(d) d_copy["kb_id"] = knowledgebaseId meta_id = d_copy.pop("id", "") operations.append({"index": {"_index": indexName, "_id": meta_id}}) operations.append(d_copy) res = [] for _ in range(ATTEMPT_TIME): # 默认 2 次 try: r = self.es.bulk(index=indexName, operations=operations, refresh=False, timeout="60s") if re.search(r"False", str(r["errors"]), re.IGNORECASE): return res for item in r["items"]: for action in ["create", "delete", "index", "update"]: if action in item and "error" in item[action]: res.append(str(item[action]["_id"]) + ":" + str(item[action]["error"])) return res except ConnectionTimeout: time.sleep(3); self._connect(); continue except Exception as e: res.append(str(e)); break return res ``` 要点： - **显式 `_id = id`**：调用方自己保证 chunk_id 唯一（典型实现：`uuid4()` 或基于 `doc_id+chunk_idx` 的稳定 hash），重复写入即"覆盖式更新"，天然支持幂等重试。 - **强制注入 `kb_id`**：所有 chunk 都打上 `kb_id` 标签，作为多租户隔离与 delete-by-query 的依据。 - **refresh=False**：写入不等可见，吞吐优先；查询侧通过 1s 默认 refresh 间隔获得近实时性。 - **显式 timeout="60s"** + ATTEMPT_TIME=2 重连 —— 网络抖动会自动重试一次。 - **失败回滚？** 只返回失败列表，**不做事务回滚**。这是 ES 的典型用法：bulk 是 best-effort，调用方需要根据返回值决定是否补偿（如 chunker 重新生成失败 chunk）。 ### 5.3 增量更新（路径 A） ```python # api/app/core/rag/utils/es_conn.py:332-422 def update(self, condition, newValue, indexName, knowledgebaseId) -> bool: # 单文档 update if "id" in condition and isinstance(condition["id"], str): chunkId = condition["id"] # 删除字段（带 _feas 后缀的 rank_features 必须先 remove 再 set，否则旧 token 残留） for k in doc.keys(): if k.split("_")[-1] == "feas": self.es.update(index=indexName, id=chunkId, script=f"ctx._source.remove(\"{k}\");") self.es.update(index=indexName, id=chunkId, doc=doc) return True # 批量 update_by_query：构造 painless 脚本 bqry = Q("bool") # ... 把 condition 转成 filter scripts = []; params = {} for k, v in newValue.items(): if k == "remove": # remove 单个 list 元素 scripts.append(f"int i=ctx._source.{kk}.indexOf(params.p_{kk});ctx._source.{kk}.remove(i);") elif k == "add": # 向 list 追加 scripts.append(f"ctx._source.{kk}.add(params.pp_{kk});") elif isinstance(v, str): v = re.sub(r"(['\n\r]|\\.)", " ", v) # 防止脚本注入 scripts.append(f"ctx._source.{k}=params.pp_{k};") ... ubq = UpdateByQuery(index=indexName).using(self.es).query(bqry)\ .script(source="".join(scripts), params=params)\ .params(refresh=True, slices=5, conflicts="proceed") ubq.execute() ``` 亮点： - **slices=5** —— 并行 update-by-query，写吞吐放大 5 倍。 - **conflicts="proceed"** —— 遇到版本冲突跳过而不中止任务；适合"标签批量更新"这种最终一致场景。 - **rank_features 必须先 remove**：因为 `*_feas` 是"key→score"字典，新值无法覆盖旧 key（341-346 行的 patch）。 - **input sanitation**：对 string 值做 `re.sub(r"(['\n\r]|\\.)", " ", v)` 防止 painless 脚本注入。 ### 5.4 路径 B 的 update_by_query ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:299-342 def update_by_segment(self, chunk: DocumentChunk, **kwargs) -> str: if self.is_multimodal_embedding: chunk.vector = self.embeddings.embed_text(chunk.page_content) else: chunk.vector = self.embeddings.embed_query(chunk.page_content) body = { "script": { "source": """ ctx._source.page_content = params.new_content; ctx._source.vector = params.new_vector; """, "params": {"new_content": chunk.page_content, "new_vector": chunk.vector} }, "query": {"term": {Field.DOC_ID.value: chunk.metadata["doc_id"]}} } return self._client.update_by_query(index=indices, body=body)['updated'] ``` 注意：`metadata.doc_id`(关键字段) 一查多匹配 → 全部刷新内容与向量。这是路径 B 的"chunk 更新"语义，**没有版本控制**，并发更新会以最后写入为准；需要严格控制时应在调用方加锁或退化为先 `delete_by_ids` 再 `add_chunks`。 --- ## 六、检索链路 ### 6.1 三种检索类型（应用层枚举） ```python # api/app/schemas/chunk_schema.py:8-13 class RetrieveType(StrEnum): PARTICIPLE = "participle" # 关键词 / 分词检索（BM25） SEMANTIC = "semantic" # 语义 / 向量检索（cosine） HYBRID = "hybrid" # 混合检索：双路 + 去重 (+ rerank) Graph = "graph" # 在 hybrid 之上叠加 GraphRAG 检索 ``` 在 `api/app/core/workflow/nodes/knowledge/node.py:213-298` 与 `api/app/core/rag/nlp/search.py:220-281` 两处可以看到完全一致的三分支 + 默认走 hybrid 的派发逻辑。 ### 6.2 关键词检索（路径 B：BM25 + IK） ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:468-558（节选） def search_by_full_text(self, query: str, **kwargs) -> list[DocumentChunk]: top_k = kwargs.get("top_k", 1024) score_threshold = float(kwargs.get("score_threshold") or 0.2) file_names_filter = kwargs.get("file_names_filter") query_str = { "bool": { "must": { "match": { Field.CONTENT_KEY.value: { "query": query, "analyzer": "ik_max_word" # 与建索引时一致 } } }, "filter": {"term": {"metadata.status": 1}} # 只看启用状态 } } # 可选叠加 file_name 多选过滤 if file_names_filter: query_str["bool"]["filter"] = [ {"term": {"metadata.status": 1}}, {"terms": {"metadata.file_name": file_names_filter}} ] result = self._client.search(index=indices, from_=0, size=top_k, query=query_str) max_score = result["hits"]["max_score"] or 1.0 docs_and_scores = [] for res in result["hits"]["hits"]: normalized_score = res["_score"] / max_score # 归一化到 [0,1] ... return [doc for doc, score in docs_and_scores if score > score_threshold] ``` 要点： - BM25 默认相似度，`ik_max_word` 中文分词；写入与查询使用同一 analyzer，避免分词错位。 - **score 归一化**：BM25 score 是开放区间，除以 `max_score` 缩放到 [0,1]，便于与 `score_threshold` 比较，也便于和向量分数同尺度对齐。 - 默认 `score_threshold=0.2`、`top_k=1024`。 ### 6.3 关键词检索（路径 A：query_string + 同义词扩展） ```python # api/app/core/rag/nlp/query.py:69-201（节选） class FulltextQueryer: query_fields = [ "title_tks^10", "title_sm_tks^5", "important_kwd^30", "important_tks^20", "question_tks^20", "content_ltks^2", "content_sm_ltks", ] def question(self, txt, tbl="qa", min_match: float = 0.6): txt = self.add_space_between_eng_zh(txt) # 中英分词预处理 txt = self.rmWWW(txt) # 去问句词（怎么/吗/啥/what/how/...） ... # 中文分支：term_weight 权重 + synonym 同义词扩展 for tt in self.tw.split(txt)[:256]: twts = self.tw.weights([tt]) syns = self.syn.lookup(tt) tk_syns = [f"({tk} OR (%s)^0.2)" % " ".join(tk_syns), ...] # 同义词权重 0.2 tms.append((tk, w)) query = " OR ".join([f"({t})" for t in qs if t]) return MatchTextExpr(self.query_fields, query, 100, {"minimum_should_match": min_match}), keywords ``` ```python # api/app/core/rag/utils/es_conn.py:196-217 for m in matchExprs: if isinstance(m, MatchTextExpr): minimum_should_match = m.extra_options.get("minimum_should_match", 0.0) if isinstance(minimum_should_match, float): minimum_should_match = str(int(minimum_should_match * 100)) + "%" bqry.must.append(Q("query_string", fields=m.fields, type="best_fields", query=m.matching_text, minimum_should_match=minimum_should_match, boost=1)) bqry.boost = 1.0 - vector_similarity_weight ``` 亮点： - **多字段 field-boost**：`important_kwd^30` 表示标签字段权重远高于正文，符合"重要标签命中即高排名"的直觉。 - **同义词加权 0.2**：同义词召回但低权重，避免"同义词稀释"主体相关性。 - **minimum_should_match**：默认 0.3 / 0.6，控制 BM25 召回的"严苛度"。当 hybrid 总命中为 0 时会 fallback 到 0.1 重试（详见 6.7）。 - **`type="best_fields"`**：多字段场景取每字段最高分作为最终分，符合"标题命中比正文命中更重要"的语义。 ### 6.4 向量检索（路径 B：script_score + cosine） ```python # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:374-466（节选） def search_by_vector(self, query: str, **kwargs) -> list[DocumentChunk]: if self.is_multimodal_embedding: query_vector = self.embeddings.embed_text(query) else: query_vector = self.embeddings.embed_query(query) top_k = kwargs.get("top_k", 1024) score_threshold = float(kwargs.get("score_threshold") or 0.3) file_names_filter = kwargs.get("file_names_filter") query_str = { "bool": { "must": { "script_score": { "query": {"match_all": {}}, "script": { # cosineSimilarity 范围 [-1,1]，+1 后落到 [0,2] "source": f"cosineSimilarity(params.query_vector, '{Field.VECTOR.value}') + 1.0", "params": {"query_vector": query_vector} } } }, "filter": {"term": {"metadata.status": 1}} } } result = self._client.search(index=indices, from_=0, size=top_k, query=query_str) docs_and_scores = [] for res in result["hits"]["hits"]: score = res["_score"] / 2 # [0,2] -> [0,1] docs_and_scores.append((..., score)) return [doc for doc, score in docs_and_scores if score > score_threshold] ``` 特性与权衡： - **script_score 是暴力扫描**：会对 `match_all` 命中的所有文档（叠加 status=1 filter 后）逐一算 cosine，复杂度 O(N·dim)。优点是结果**精确**、无 ANN 召回率损失；缺点是延迟随 KB chunk 数线性增长，不适合 chunk 量级大的 KB。 - **score 归一化**：`(cos+1)/2 ∈ [0,1]`，与 BM25 归一化值同尺度。 - **过滤集成**：`metadata.status=1` 在 filter 上，先过滤再算分；`file_names_filter` 同理。 ### 6.5 向量检索（路径 A：knn + filter） ```python # api/app/core/rag/utils/es_conn.py:206-217 elif isinstance(m, MatchDenseExpr): similarity = m.extra_options.get("similarity", 0.0) s = s.knn( m.vector_column_name, m.topn, m.topn * 2, # num_candidates = 2 * k，控制召回率 query_vector=list(m.embedding_data), filter=bqry.to_dict(), # 与 BM25 同一份 bool filter # similarity=similarity # 已注释：未启用阈值剪枝 ) ``` ```python # api/app/core/rag/nlp/search.py:365-373 def get_vector(self, txt, emb_mdl, topk=10, similarity=0.1): qv, _ = emb_mdl.encode_queries(txt) 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}) ``` 要点： - **HNSW ANN**：路径 A 用的是 ES 8 原生 `knn` query，底层 HNSW 索引，毫秒级，但有近似召回率损失。 - **k vs num_candidates**：`topn * 2` 即 ANN 阶段先取 2k 候选再精排到 k，是召回率与延迟的折中。生产建议至少 `4 * topn`，更高召回。 - **filter 共享**：`filter=bqry.to_dict()`——把 BM25 那份 bool filter 同时挂在 knn 上，确保过滤条件在 ANN 内部应用（pre-filter）；这点对多租户 `kb_id` 隔离尤为关键，否则 ANN 先取 top-k 再过滤，可能完全不返回该 KB 的文档。 - **similarity 阈值已注释**：当前不启用 ES 内置阈值剪枝；需要按相似度阈值过滤的话，由应用层 (`Dealer.retrieval`) 在 rerank 阶段做。 ### 6.6 混合搜索 —— 这是本节最关键的"融合公式" #### 6.6.1 路径 A：`FusionExpr("weighted_sum")` + ES 内部混合（**核心融合点**） ```python # api/app/core/rag/nlp/search.py:435-445 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) ``` ```python # api/app/core/rag/utils/es_conn.py:186-218 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: # 必须按 [text, dense, fusion] 顺序传入 assert len(matchExprs) == 3 and isinstance(matchExprs[0], MatchTextExpr) \ and isinstance(matchExprs[1], MatchDenseExpr) and isinstance(matchExprs[2], FusionExpr) weights = m.fusion_params["weights"] vector_similarity_weight = get_float(weights.split(",")[1]) # "0.05,0.95" -> 0.95 for m in matchExprs: if isinstance(m, MatchTextExpr): ... bqry.must.append(Q("query_string", ..., boost=1)) bqry.boost = 1.0 - vector_similarity_weight # text 整体 boost = 0.05 elif isinstance(m, MatchDenseExpr): s = s.knn(m.vector_column_name, m.topn, m.topn * 2, query_vector=list(m.embedding_data), filter=bqry.to_dict()) if bqry: s = s.query(bqry) ``` **融合公式**（这是 [S2-T7] 评审要求"必须明确"的部分）： ``` final_score(doc) = (1 - w_vec) * BM25_query_string_score(doc) + w_vec * knn_cosine_score(doc) + Σ rank_feature_score(doc) ← PageRank + tag 加权（可选） ``` 其中： - `w_vec = 0.95`（来自 `FusionExpr` 的 `"weights": "0.05,0.95"` 第二个权重）。 - BM25 整体 `bqry.boost = 0.05`，即 `query_string` 的 BM25 分数被乘 0.05；knn 的分数没有显式 boost，相当于权重 1.0，但语义上由调用方约定 0.95（**即代码层面是"BM25 直接乘 0.05，knn 不缩放"，并未严格归一化到等比例**——这是一个已知近似，见 6.7 fallback）。 - 排序逻辑：ES 8 的 hybrid 行为是"bool query 命中集 ∪ knn top-k 候选集"，并集后用各自分数相加（未命中那侧分数为 0）。`elasticsearch-dsl Search` 的 `.query(...).knn(...)` 组合自动启用此模式。 - `rank_feature` 通过 `bqry.should.append(Q("rank_feature", field=fld, linear={}, boost=sc))`（es_conn.py:219-223）以**加性**方式融入最终分。 > 这种"应用层约定 + ES 端 boost 缩放"的混合不是教科书式的归一化加权，但工程上简单：BM25 与 cosine 在统计上不同尺度，0.05/0.95 的**极端偏向语义**是为了"以语义检索为主、关键词作为补强"。 #### 6.6.2 路径 B：双路 + 去重 + 可选 Rerank ```python # api/app/core/workflow/nodes/knowledge/node.py:236-271 case retrieve_type if retrieve_type in (RetrieveType.HYBRID, RetrieveType.Graph): rs1_task = asyncio.to_thread(vector_service.search_by_vector, **{ "query": query, "top_k": kb_config.top_k, "indices": indices, "score_threshold": kb_config.vector_similarity_weight }) rs2_task = asyncio.to_thread(vector_service.search_by_full_text, **{ "query": query, "top_k": kb_config.top_k, "indices": indices, "score_threshold": kb_config.similarity_threshold }) rs1, rs2 = await asyncio.gather(rs1_task, rs2_task) # 双路并发 unique_rs = self._deduplicate_docs(rs1, rs2) # 按 doc_id 去重 if not unique_rs: return [] if self.typed_config.reranker_id: rs.extend(await asyncio.to_thread( self.rerank, **{"query": query, "docs": unique_rs, "top_k": kb_config.top_k})) else: rs.extend(sorted(unique_rs, key=lambda d: d.metadata.get("score", 0), reverse=True)[:kb_config.top_k]) ``` ```python # api/app/core/rag/nlp/search.py:236-261（同等逻辑的同步版） case _: rs1 = vector_service.search_by_vector(...) rs2 = vector_service.search_by_full_text(...) seen_ids = set(); unique_rs = [] for doc in rs1 + rs2: if doc.metadata["doc_id"] not in seen_ids: seen_ids.add(doc.metadata["doc_id"]) unique_rs.append(doc) rs = unique_rs if unique_rs: rs = vector_service.rerank(query=..., docs=unique_rs, top_k=...) ``` **融合公式**（路径 B）： ``` candidates = vector_topk(q, w_v) ∪ bm25_topk(q, w_t) # 双路并发召回 deduped = unique_by(metadata.doc_id, candidates) # 后到的丢弃 if reranker: final = reranker(query, deduped)[:top_k] # 跨编码器重排 else: final = sort_by_score_desc(deduped)[:top_k] # 各自归一化分数直接比 ``` > **why 不在路径 B 做加权融合**：路径 B 双路分数已分别归一化到 [0,1]，但"BM25 归一化分"与"cosine 归一化分"之间**不可比**（一个是相对最大分，一个是绝对几何相似度）。直接把它们排序虽然不严谨，但通常依赖下游的 cross-encoder reranker 做最终排序，因此前置阶段以"召回多样性"为优先（vector 主召回 + BM25 补关键词），不再做权重融合。 ### 6.7 兜底：低召回 fallback ```python # api/app/core/rag/nlp/search.py:447-459 # If result is empty, try again with lower min_match if total == 0: if filters.get("document_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) # 0.3 -> 0.1 matchDense.extra_options["similarity"] = 0.17 # 0.1 -> 0.17（提高语义阈值） res = self.dataStore.search(src, highlightFields, filters, [matchText, matchDense, fusionExpr], orderBy, offset, limit, idx_names, kb_ids, rank_feature=rank_feature) ``` > **设计意图**：第一轮严格匹配（min_match=0.3）保证精度；命中为 0 时放宽 BM25 但提高向量阈值，等价于"换主导侧"，避免空结果。 ### 6.8 Rerank：模型重排 + 应用层混合相似度 `Dealer.rerank_by_model` 与 `Dealer.rerank` 是两套 reranker： ```python # api/app/core/rag/nlp/search.py:606-666 def rerank(self, sres, query, tkweight=0.3, vtweight=0.7, ...): sim, tksim, vtsim = self.qryr.hybrid_similarity( sres.query_vector, ins_embd, keywords, ins_tw, tkweight, vtweight) return sim + rank_fea, tksim, vtsim def rerank_by_model(self, rerank_mdl, sres, query, tkweight=0.3, vtweight=0.7, ...): tksim = self.qryr.token_similarity(keywords, ins_tw) vtsim, _ = rerank_mdl.similarity(query, [...]) return tkweight * (np.array(tksim) + rank_fea) + vtweight * vtsim, tksim, vtsim ``` ```python # api/app/core/rag/nlp/query.py:203-211 def hybrid_similarity(self, avec, bvecs, atks, btkss, tkweight=0.3, vtweight=0.7): sims = CosineSimilarity([avec], bvecs) tksim = self.token_similarity(atks, btkss) if np.sum(sims[0]) == 0: return np.array(tksim), tksim, sims[0] return np.array(sims[0]) * vtweight + np.array(tksim) * tkweight, tksim, sims[0] ``` 应用层重排公式： ``` final_score = vtweight * cosine(q_vec, c_vec) + tkweight * token_sim(q, c) + rank_feature_score ≈ 0.7 * vector_sim + 0.3 * keyword_sim + (PageRank + tag) ``` 注意 `Dealer.retrieval()`（674-768 行）调用时传入的是 `1 - vector_similarity_weight, vector_similarity_weight`，所以这两个权重由调用方（用户配置）决定，默认 0.3 / 0.7（`vector_similarity_weight=0.3` 见 678 行）。 ### 6.9 top_k / 召回率 / 延迟权衡 | 阶段 | 默认值 | 含义 | 调参建议 | | --- | --- | --- | --- | | `top_k` (KB 节点) | 工作流配置 | 单 KB 单路召回数 | hybrid 模式建议 ≥ 50；语义高质 KB 可 30 | | `topn` / `topk` (Dealer) | 1024 (ann fallback)，10 (默认) | knn 阶段 k | 与下游 RERANK_LIMIT 联动 | | `num_candidates` | `topn * 2` (es_conn.py:213) | HNSW 候选数，影响召回率 | 高召回场景改为 `4 * topn` | | `RERANK_LIMIT` | `ceil(64/page_size)*page_size` (search.py:683) | rerank 输入数 | 与显示页大小绑定，避免 rerank 过多 | | `score_threshold` (BM25) | 0.2 | 归一化后阈值 | 关键词强场景可调到 0.3 | | `score_threshold` (vector) | 0.3 | (cos+1)/2 后阈值 | 严苛去噪可到 0.5 | | `min_match` | 0.3，fallback 0.1 | BM25 词命中比 | 短查询调高，长查询调低 | | `request_timeout` | 30s | ES 客户端超时 | 高并发下 60s | | `search timeout` | "600s" (es_conn.py:257) | ES 服务端超时 | 超长 KB 才放宽 | --- ## 七、配置项与运维要点 ### 7.1 环境变量（连接 + 客户端调优） ```python # api/app/core/rag/utils/es_conn.py:60-80 # api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:685-710 ELASTICSEARCH_HOST # 默认 127.0.0.1，可填 http://es-1 / https://es-1 ELASTICSEARCH_PORT # 默认 9200 ELASTICSEARCH_USERNAME # 默认 elastic ELASTICSEARCH_PASSWORD # 默认 elastic ELASTICSEARCH_REQUEST_TIMEOUT # 默认 30 (秒) ELASTICSEARCH_RETRY_ON_TIMEOUT # 默认 True (es_conn 中是字符串比较，注意 bug 见下) ELASTICSEARCH_MAX_RETRIES # 默认 3 ELASTICSEARCH_VERIFY_CERTS # 默认 false ELASTICSEARCH_CA_CERTS # 自签证书路径 ELASTICSEARCH_CONNECTIONS_PER_NODE # 路径 B 独有，默认 10 ``` > **小坑**：`es_conn.py:72` 写的是 `os.getenv("ELASTICSEARCH_RETRY_ON_TIMEOUT", True) == "true"`——默认值是 bool `True`，但与字符串 `"true"` 比较恒为 `False`。所以**默认情况下其实没开启 retry_on_timeout**，需要显式设置 `ELASTICSEARCH_RETRY_ON_TIMEOUT=true`（小写）才生效。 ### 7.2 ES 集群规模建议 按 `mapping.json` 默认 2 shards、0 replicas，**不可直接用于生产**。建议： | 数据量 | 节点数 | shards | replicas | heap | 备注 | | --- | --- | --- | --- | --- | --- | | < 100w chunks | 1-3 | 2 | 1 | 8GB | 默认配置 + 1 副本 | | 100w-1000w | 3-5 | 4-8 | 1 | 16GB | 增加 shard 减少单 shard 体积 | | > 1000w | 5+ | 8-16 | 1-2 | 31GB(不超过 32) | shard 大小控制在 30-50GB | **核心准则**： - 单 shard 不超过 50GB； - replicas ≥ 1，至少容忍 1 节点宕机； - JVM heap 不超过 32GB（zero-based compressed oops）； - 留 50% RAM 给 OS file cache（lucene 依赖 mmap）。 ### 7.3 索引膨胀治理 观察点： - **路径 A 的 `graphrag_{workspace_id}` 索引**：随 workspace chunk 数增长，`number_of_shards=2` 容易超过 50GB/shard。需要按"workspace 容量分层"，对热门/大 workspace 单独 reindex 到更多 shards。 - **路径 B 的 `Vector_index_{kb_id}_Node` 索引**：每 KB 一个 index，KB 数 1000+ 时 cluster state 显著膨胀，可能拖慢所有索引创建/查询。建议引入"KB 共享索引 + kb_id 路由"模式（详见 §8 优化建议）。 ```python # api/app/core/rag/utils/es_conn.py:587-633 def get_cluster_stats(self): """ 暴露 store_size / docs / nodes_version / jvm_heap_used 等用于 dashboard """ raw_stats = self.es.cluster.stats() return {...} ``` > **建议**：在调度器里定时拉取 `get_cluster_stats()`，把 `store_size / docs / heap_used_percent` 接入告警。 ### 7.4 慢查询排查 ```python # api/app/core/rag/utils/es_conn.py:250-263 logger.debug(f"ESConnection.search {str(indexNames)} query: " + json.dumps(q)) res = self.es.search(index=indexNames, body=q, timeout="600s", # search_type="dfs_query_then_fetch", track_total_hits=True, _source=True) ``` 排查路径： 1. **打开 debug 日志**：`logger=rag.es_conn` 调到 DEBUG，可以看到完整 DSL。 2. **关闭 `track_total_hits=True`**：超过 10000 hits 时它会真正扫表，对大 KB 是常见慢点；如果不需要精确总数，改为 `track_total_hits=10000`。 3. **打开 `dfs_query_then_fetch`**：在多 shard 时让 IDF 全局计算，对相关性更准；代价是一次 RTT。 4. **限制 `num_candidates`**：HNSW 阶段候选数大幅影响延迟；已是 `topn * 2`，进一步压缩到 `topn` 可观察延迟下降。 5. **slow log**：在 ES 集群层面打开 `index.search.slowlog.threshold.query.warn: 1s`，定位单查询慢点。 ### 7.5 健康监控接口 ```python # api/app/core/rag/utils/es_conn.py:95-98 def health(self) -> dict: health_dict = dict(self.es.cluster.health()) health_dict["type"] = "elasticsearch" return health_dict ``` ```python # api/app/core/rag/utils/doc_store_conn.py:140-145 @abstractmethod def health(self) -> dict: """Return the health status of the database.""" ``` 接入业务监控的最简方法：起一个轻量 endpoint 调用 `docStoreConn.health()`，把 `status (green/yellow/red)`、`number_of_nodes`、`active_shards_percent_as_number` 上报。 --- ## 八、边界条件与已知限制 | 限制 | 影响 | 解决方向 | | --- | --- | --- | | 路径 B `dims = len(embeddings[0])` 锁定维度 | 换 embedding 模型必须重建索引 | 按维度后缀命名向量字段（参考路径 A 的 `q_{dim}_vec`）| | 路径 A 默认 0 副本 | 节点宕机即数据丢失 | 修改 `res/mapping.json` `number_of_replicas: 1` | | `ELASTICSEARCH_RETRY_ON_TIMEOUT` 默认未生效 | 网络抖动直接抛错 | bug：bool 与 "true" 字符串比较；需显式 `=true` | | `script_score` 暴力扫描 | 大 KB 延迟高 | 路径 B 升级到 `knn` query（ES 8 原生）| | 路径 B inline mapping 不带 metadata.kb_id | 多 KB 共享索引时无法过滤 | 与路径 A 对齐，引入 `kb_id` keyword | | `update_by_segment` 无并发控制 | 并发更新最后写入胜出 | 走 `delete_by_ids` + `add_chunks` 或显式版本号 | | `add_texts` 不捕获 `BulkIndexError` | 局部失败整批失败 | 增加 try/except + 失败重投队列 | | 一个 workspace 多 KB 共享路径 A 索引 | 单 KB 删除走 delete-by-query，不立即释放磁盘 | 定期 `_forcemerge?only_expunge_deletes=true` | | 路径 B 每 KB 一索引 | 大量 KB 时 cluster state 膨胀 | 改为共享索引 + `kb_id` routing | | `track_total_hits=True` | 大库 search 全表扫描慢 | 默认改为 10000，按需取 max | --- ## 九、监控指标与排错指引 ### 9.1 关键指标 | 指标 | 来源 | 告警阈值（参考）| | --- | --- | --- | | ES cluster status | `health()` | red 立即告警 | | `active_shards_percent_as_number` | `health()` | < 100% 持续 5min 告警 | | `jvm_heap_used_percent` | `get_cluster_stats()` | > 75% 警告，> 85% 紧急 | | `os_mem_used_percent` | `get_cluster_stats()` | > 90% 警告 | | 写入失败比例 | `ESConnection.insert` 返回的 `res` 列表长度 / 总 chunk 数 | > 1% 告警 | | 单次 search P95 延迟 | 调用方时序日志 | hybrid > 1s 告警 | | `track_total_hits` 命中超过 10k 比例 | search.py 总数 | 频繁触发即扩 shard | ### 9.2 典型故障与处理 | 现象 | 可能原因 | 处置 | | --- | --- | --- | | 写入超时 | bulk 太大 / refresh 阻塞 | 减小 batch（≤ 1000）/ 写入窗口 `refresh_interval=30s` | | 检索召回为 0 | min_match 过严 / kb_id 过滤不一致 | 看 search.py:447 fallback 是否触发；核对 kb_id | | HNSW 召回率低 | num_candidates 过小 | 增大到 `4 * topn` 或 `topn * 4` | | 维度不匹配报错 | 换 embedding 模型未 reindex | 按 §8 维度限制处理；或在路径 B 删 KB 重建 | | cluster state 过大 | KB 索引数过多 | §10 改造为共享索引 + kb_id routing | | 中文检索召回差 | 写入 analyzer 与查询 analyzer 不一致 | 路径 B 必须保持 `ik_max_word`（写入与查询）| --- ## 十、优化建议与未来扩展点 ### 10.1 架构改造（短期，1-2 个迭代） 1. **统一双路径**：保留路径 A 抽象 (`DocStoreConnection` + `Dealer`)，把路径 B 的 `ElasticSearchVector` 重构为 `DocStoreConnection` 的薄封装，删除重复的连接管理 (`ElasticSearchVectorFactory`)，全局只用 `@singleton ESConnection`。 2. **修复默认配置**： - `mapping.json` `number_of_replicas: 0 → 1`； - 修正 `ELASTICSEARCH_RETRY_ON_TIMEOUT` bool/str 比较； - 路径 B 的 `script_score` 切换为 `knn` query； - 路径 B mapping 加上 `kb_id` keyword 字段，为后续合并索引铺路。 3. **共享索引 + 路由**：把 `Vector_index_{kb_id}_Node` 改为 `kb_chunks_{workspace_id}` 共享索引，`kb_id` 字段做 routing key，索引数从 N(KB) 降到 N(workspace)。 ### 10.2 检索增强（中期） 1. **真正的 RRF**（reciprocal rank fusion）：当前 `weighted_sum` 对分数尺度敏感，引入 `rank_fusion` (ES 8.8+) 或在应用层实现 `rrf_score(d) = Σ 1/(k + rank_i(d))`，对尺度不敏感。 2. **稀疏向量（ELSER / SPLADE）**：路径 A 已在 `MatchSparseExpr` 接口预留位置，但 ES 实现未启用 `rank_features` 稀疏检索，引入后可在中文长尾查询上显著提升召回。 3. **多模态检索**：路径 B 已感知 `is_multimodal_embedding`（`elasticsearch_vector.py:41`），但只针对火山引擎；引入跨模态 BGE-M3 类模型后，可在同一 dense_vector 字段上做"图文混排"。 4. **HNSW 参数显式化**：`mapping.json` 没有指定 `index_options`（m / ef_construction）。在构建大索引时显式 `m=16, ef_construction=200` 可显著提升召回率。 ### 10.3 工程鲁棒性（中期） 1. **写入幂等保护**：路径 B `add_texts` 不传 `_id`，依赖 `metadata.doc_id` 后查；改为直接用 `doc_id` 作为 `_id`，写入即可幂等，省去后查。 2. **变更检测 reindex**：当 mapping 变化时，加一个 `migration_version` 字段触发 alias-swap reindex（`old_index → new_index`），避免线上停机重建。 3. **批量限流**：`helpers.bulk` 默认无背压，引入 `chunk_size=500, max_chunk_bytes=10MB` 显式限制，避免大 chunk 撑爆 ES heap。 4. **路径 A 的 `ATTEMPT_TIME=2`** 太少：网络抖动 2 次重试后丢错，建议升到 3-5 次，配合指数退避。 ### 10.4 长期扩展点 1. **冷热分离**：超过半年/低访问的 chunk 迁到冷节点（warm tier）+ rollover index，配合"记忆遗忘引擎" (Memory Forgetting Engine, README §4) 协同。 2. **跨集群联邦**：当多 workspace 数据量过大，引入 cross-cluster search（CCS）按 workspace 切集群。 3. **GraphRAG 与 VDB 联合检索**：当前 `kg_retriever.retrieval` 在路径 B 是后置 insert（node.py:286-298），可改为"先图谱召回相关实体 → 把实体名作为 `important_kwd^30` 注入 BM25"实现一次 ES 调用同时享受图谱与向量。 --- ## 十一、关键源码片段索引（评审检查点） | 主题 | 文件:行号 | 一句话说明 | | --- | --- | --- | | 抽象接口 | `api/app/core/rag/utils/doc_store_conn.py:128-256` | `DocStoreConnection` 14 个抽象方法 | | MatchExpr 族 | `api/app/core/rag/utils/doc_store_conn.py:43-114` | 文本/稠密/稀疏/张量/融合表达式 | | ES 连接管理 | `api/app/core/rag/utils/es_conn.py:26-86` | `@singleton` + 8.x 版本校验 | | 全局 mapping | `api/app/core/rag/res/mapping.json:1-211` | dynamic_templates + 自定义 BM25 | | ES 8 hybrid 核心 | `api/app/core/rag/utils/es_conn.py:186-218` | `query_string` + `s.knn(...)` 共享 filter | | 加权融合 | `api/app/core/rag/utils/es_conn.py:188-194` 与 `api/app/core/rag/nlp/search.py:439` | `FusionExpr("weighted_sum", weights="0.05,0.95")` | | 应用层 hybrid_similarity | `api/app/core/rag/nlp/query.py:203-211` | `0.7*cos + 0.3*token_sim` | | 双路 + 去重 + rerank | `api/app/core/workflow/nodes/knowledge/node.py:236-271` | 工作流默认混合策略 | | BaseVector 抽象 | `api/app/core/rag/vdb/vector_base.py:9-67` | 路径 B 的接口骨架 | | KB 索引 mapping | `api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:609-663` | inline 创建 + dims 锁定 | | 关键词检索（BM25+IK）| `api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:468-558` | match + ik_max_word + 归一化 | | 向量检索（cosine 暴力）| `api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:374-466` | script_score + filter | | 关键词构造（多字段 + 同义词）| `api/app/core/rag/nlp/query.py:14-22, 69-201` | query_fields field-boost + synonym | | Dealer.retrieval (主入口) | `api/app/core/rag/nlp/search.py:674-768` | 检索 + rerank + 分页 | | 低召回 fallback | `api/app/core/rag/nlp/search.py:447-459` | min_match 0.3→0.1，similarity 0.1→0.17 | | update_by_query | `api/app/core/rag/utils/es_conn.py:332-422` | painless + slices=5 + conflicts=proceed | | bulk 写 + 错误处理 | `api/app/core/rag/utils/es_conn.py:294-330` | refresh=False + 两次重试 + 错误聚合 | | 工厂单例 (路径 B) | `api/app/core/rag/vdb/elasticsearch/elasticsearch_vector.py:666-732` | 双重检查锁 + 版本校验一次性 | | 全局初始化 | `api/app/core/rag/common/settings.py:13-24` | docStoreConn / retriever / kg_retriever | | 检索类型枚举 | `api/app/schemas/chunk_schema.py:8-13` | participle / semantic / hybrid / graph | --- ## 十二、TL;DR（一段话总结） MemoryBear 用 Elasticsearch 8 同时承担**全文（Lucene + IK + 自定义 BM25）和向量（dense_vector + HNSW）**双引擎，所以选 ES 而不是专用向量库。代码里有**两套并行路径**：路径 A `ESConnection`（单例 `DocStoreConnection`，多字段动态模板，配 `Dealer` 做 `weighted_sum=0.05,0.95` 的应用层加权 + ES 原生 hybrid 与 rank_features，主要给 GraphRAG/复杂 RAG 用）；路径 B `ElasticSearchVector`（`BaseVector` 简化封装，`script_score+cosine` 与 `match+ik_max_word`，主要给工作流知识节点和 KB 服务用，hybrid 走"双路并发 → metadata.doc_id 去重 → 可选 reranker"）。索引按 workspace 或按 KB 隔离，`mapping.json` 默认 2 shards / 0 副本 / 1s refresh，向量字段按维度后缀（512/768/1024/1536）动态创建，文本字段以 `_tks/_ltks/_kwd` 后缀套用 dynamic_templates。生产化的主要风险点：路径 B 锁死 dims、默认 0 副本、`ELASTICSEARCH_RETRY_ON_TIMEOUT` 比较 bug、`script_score` 暴力扫描、KB 索引数膨胀；优化方向是合并双路径、改用 `knn` + RRF、共享索引 + `kb_id` routing、配合 GraphRAG 做联合检索。