From 0c5732108a0fcd4ddc6f09e0afabdd039fa4cd65 Mon Sep 17 00:00:00 2001 From: buua436 Date: Thu, 23 Jul 2026 20:48:39 +0800 Subject: [PATCH] fix: support Infinity knowledge compilation (#17288) ### What problem does this PR solve? Fix Infinity compatibility issues in knowledge compilation. This change: - Stores compilation source ID lists as JSON arrays in Infinity. - Parses JSON array fields when reading compiled documents. - Uses `json_contains` for filtering JSON array fields. - Adds the missing `name` column to the Infinity mapping. - Updates dataset navigation KNN search to use the unified `MatchDenseExpr` interface. - Handles unavailable embeddings without querying an invalid `q_0_vec` field. ### Type of change - [x] Bug Fix (non-breaking change which fixes an issue) --- common/doc_store/infinity_conn_base.py | 19 ++++- conf/infinity_mapping.json | 17 ++--- .../knowlege_compile/dataset_nav.py | 72 +++++++++---------- rag/utils/infinity_conn.py | 35 ++++++++- web/src/hooks/use-document-request.ts | 33 ++++++++- 5 files changed, 125 insertions(+), 51 deletions(-) diff --git a/common/doc_store/infinity_conn_base.py b/common/doc_store/infinity_conn_base.py index 1a97e2a833..c89d38e64e 100644 --- a/common/doc_store/infinity_conn_base.py +++ b/common/doc_store/infinity_conn_base.py @@ -304,7 +304,24 @@ class InfinityConnectionBase(DocStoreConnection): continue if not v: continue - if self.field_keyword(k): + if k in { + "source_chunk_ids", + "source_doc_ids", + "compilation_template_ids", + "doc_ids_kwd", + "entity_names_kwd", + "outlinks_kwd", + "related_kb_pages_kwd", + "rechunked_from_chunk_ids", + }: + values = v if isinstance(v, list) else [v] + json_conditions = [] + for item in values: + literal = json.dumps(item, ensure_ascii=False).replace("'", "''") + json_conditions.append(f"json_contains({k}, '{literal}')") + if json_conditions: + cond.append("(" + " or ".join(json_conditions) + ")") + elif self.field_keyword(k): if isinstance(v, list): inCond = list() for item in v: diff --git a/conf/infinity_mapping.json b/conf/infinity_mapping.json index f09925c84d..d46af006ec 100644 --- a/conf/infinity_mapping.json +++ b/conf/infinity_mapping.json @@ -8,6 +8,7 @@ "create_timestamp_flt": {"type": "float", "default": 0.0}, "img_id": {"type": "varchar", "default": ""}, "docnm": {"type": "varchar", "default": "", "analyzer": ["rag-coarse", "rag-fine"], "comment": "docnm_kwd, title_tks, title_sm_tks"}, + "name": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "name_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "tag_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "important_kwd_empty_count": {"type": "integer", "default": 0}, @@ -43,9 +44,9 @@ "extra": {"type": "varchar", "default": ""}, "compile_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "source_chunk_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "source_doc_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "compilation_template_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, + "source_chunk_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, + "source_doc_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, + "compilation_template_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, "compilation_template_kind_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "chunk_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "input_hash_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, @@ -55,20 +56,20 @@ "skill_with_weight": {"type": "varchar", "default": ""}, "skill_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "children_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "doc_ids_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, + "doc_ids_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, "slug_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "title_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "topic_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "page_type_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "entity_names_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "outlinks_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, - "related_kb_pages_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, + "entity_names_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, + "outlinks_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, + "related_kb_pages_kwd": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, "type_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "from_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "to_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "rechunk_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "rechunked_from_template_id": {"type": "varchar", "default": ""}, - "rechunked_from_chunk_ids": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, + "rechunked_from_chunk_ids": {"type": "json", "default": "[]", "index_type": {"type": "secondary", "cardinality": "low"}}, "superseded_by_chunk_id": {"type": "varchar", "default": ""}, "doc_count_int": {"type": "integer", "default": 0}, "depth_int": {"type": "integer", "default": 0}, diff --git a/rag/advanced_rag/knowlege_compile/dataset_nav.py b/rag/advanced_rag/knowlege_compile/dataset_nav.py index d618c8ad27..90f4d1c6d8 100644 --- a/rag/advanced_rag/knowlege_compile/dataset_nav.py +++ b/rag/advanced_rag/knowlege_compile/dataset_nav.py @@ -177,6 +177,7 @@ async def _store_knn( ) -> list[dict]: """KNN search with dense vector and filter, returning top_k hits.""" from common import settings + from common.doc_store.doc_store_base import MatchDenseExpr, OrderByExpr index = _index_name(tenant_id) vf = _vec_field(vec_dim) @@ -191,39 +192,26 @@ async def _store_knn( "doc_ids_kwd", vf, ] - try: - res = await thread_pool_exec( - settings.docStoreConn.search, - fields, - [], - filter_condition, - [], - None, - 0, - top_k, - index, - [kb_id], - knn_vector=vec, - knn_vector_field=vf, - ) - except TypeError: - # Fallback: some doc store connectors don't accept knn_* kwargs. - # Perform a plain search and lambda-rank in Python (slow-path). - rows = await _store_search( - tenant_id, - kb_id, - filter_condition, - fields, - limit=top_k * 10, - ) - scoring = [] - for r in rows: - stored = r.get(vf) - if _vector_len(stored) > 0 and _vector_len(stored) == _vector_len(vec): - sim = sum(a * b for a, b in zip(stored, vec)) - scoring.append((sim, r)) - scoring.sort(key=lambda x: -x[0]) - return [r for _, r in scoring[:top_k]] + match_expr = MatchDenseExpr( + vector_column_name=vf, + embedding_data=list(vec), + embedding_data_type="float", + distance_type="cosine", + topn=top_k, + extra_options={}, + ) + res = await thread_pool_exec( + settings.docStoreConn.search, + fields, + [], + filter_condition, + [match_expr], + OrderByExpr(), + 0, + top_k, + index, + [kb_id], + ) results = settings.docStoreConn.get_fields(res, fields) if res else {} return list(results.values()) @@ -590,7 +578,7 @@ async def upsert_dataset_nav_doc( # 3. Embed doc summary doc_embedding = await _embed(embd_mdl, summary) if embd_mdl else [] - vec_dim = _EMBED_DIM or 0 + vec_dim = len(doc_embedding) lock = RedisDistributedLock( _nav_lock_key(kb_id), @@ -605,12 +593,16 @@ async def upsert_dataset_nav_doc( try: # 4. Layered KNN search for nearest cluster - best_name, best_parent, sim = await _find_best_cluster( - tenant_id, - kb_id, - doc_embedding, - vec_dim, - ) + if doc_embedding: + best_name, best_parent, sim = await _find_best_cluster( + tenant_id, + kb_id, + doc_embedding, + vec_dim, + ) + else: + logging.warning("dataset_nav: embedding unavailable for doc=%s, skipping KNN placement", doc_id) + best_name, best_parent, sim = None, None, 0.0 if best_name and sim >= _MERGE_THRESHOLD: # ── Merge into best cluster ── diff --git a/rag/utils/infinity_conn.py b/rag/utils/infinity_conn.py index aac207e608..a8e68d682d 100644 --- a/rag/utils/infinity_conn.py +++ b/rag/utils/infinity_conn.py @@ -26,6 +26,20 @@ from common.doc_store.doc_store_base import MatchExpr, MatchTextExpr, MatchDense from common.doc_store.infinity_conn_base import InfinityConnectionBase +_JSON_LIST_FIELDS = frozenset( + ( + "source_chunk_ids", + "source_doc_ids", + "compilation_template_ids", + "doc_ids_kwd", + "entity_names_kwd", + "outlinks_kwd", + "related_kb_pages_kwd", + "rechunked_from_chunk_ids", + ) +) + + @singleton class InfinityConnection(InfinityConnectionBase): """ @@ -440,6 +454,8 @@ class InfinityConnection(InfinityConnectionBase): elif k == "question_tks": if not d.get("question_kwd"): d["questions"] = self.list2str(v) + elif k in _JSON_LIST_FIELDS: + d[k] = json.dumps(list(v) if isinstance(v, (list, tuple, set)) else [], ensure_ascii=False) elif self.field_keyword(k): if isinstance(v, list): d[k] = "###".join(v) @@ -579,6 +595,8 @@ class InfinityConnection(InfinityConnectionBase): elif k == "question_tks": if not new_value.get("question_kwd"): new_value["questions"] = self.list2str(v) + elif k in _JSON_LIST_FIELDS: + new_value[k] = json.dumps(list(v) if isinstance(v, (list, tuple, set)) else [], ensure_ascii=False) elif self.field_keyword(k): if isinstance(v, list): new_value[k] = "###".join(v) @@ -794,7 +812,22 @@ class InfinityConnection(InfinityConnectionBase): for column in list(res2.columns): k = column.lower() - if self.field_keyword(k): + if k in _JSON_LIST_FIELDS: + + def parse_json_list(value): + if isinstance(value, list): + return value + if not value: + return [] + try: + parsed = json.loads(value) + return parsed if isinstance(parsed, list) else [parsed] + except (TypeError, json.JSONDecodeError): + # Read rows written by the previous varchar encoding. + return [item for item in str(value).split("###") if item] + + res2[column] = res2[column].apply(parse_json_list) + elif self.field_keyword(k): res2[column] = res2[column].apply(lambda v: [kwd for kwd in v.split("###") if kwd]) elif re.search(r"_feas$", k): res2[column] = res2[column].apply(lambda v: json.loads(v) if v else {}) diff --git a/web/src/hooks/use-document-request.ts b/web/src/hooks/use-document-request.ts index c73724b4b9..316975585a 100644 --- a/web/src/hooks/use-document-request.ts +++ b/web/src/hooks/use-document-request.ts @@ -74,6 +74,8 @@ const DocumentKeys = { [DocumentApiAction.FetchDocumentList, 'byIds', ids] as const, }; +const documentIngestInFlight = new Map>(); + export const DocumentStructureKeys = { graph: (datasetId: string, documentId: string) => [ @@ -389,7 +391,36 @@ export const useRunDocument = () => { }, }); - return { runDocumentByIds: mutateAsync, loading, data }; + const runDocumentByIds = useCallback( + (params: { + documentIds: string[]; + run: number; + option?: { delete: boolean; apply_kb: boolean }; + }) => { + const key = JSON.stringify({ + documentIds: [...params.documentIds].sort(), + run: params.run, + option: params.option || null, + }); + const existingRequest = documentIngestInFlight.get(key); + if (existingRequest) { + return existingRequest; + } + + const request = mutateAsync(params); + documentIngestInFlight.set(key, request); + const clearRequest = () => { + if (documentIngestInFlight.get(key) === request) { + documentIngestInFlight.delete(key); + } + }; + void request.then(clearRequest, clearRequest); + return request; + }, + [mutateAsync], + ); + + return { runDocumentByIds, loading, data }; }; export const useRemoveDocument = () => {