From 2ba7ccecafc3a8cfccfe4a67f3cb114304d2ec5f Mon Sep 17 00:00:00 2001 From: buua436 Date: Fri, 24 Jul 2026 17:48:27 +0800 Subject: [PATCH] fix: stabilize knowledge compilation navigation updates (#17345) --- api/apps/restful_apis/agent_api.py | 3 + api/apps/restful_apis/document_api.py | 6 + api/apps/services/dataset_api_service.py | 32 +++-- api/db/services/document_service.py | 28 ++-- common/doc_store/infinity_conn_base.py | 12 +- .../knowlege_compile/dataset_nav.py | 124 ++++++++++++------ rag/advanced_rag/knowlege_compile/runner.py | 17 ++- rag/flow/compiler/compiler.py | 9 +- .../chunk_post_processor.py | 2 + rag/utils/infinity_conn.py | 16 +++ web/src/components/ui/tree-view.tsx | 6 +- .../dataset/compilation/utils/nav-tree.ts | 2 + 12 files changed, 181 insertions(+), 76 deletions(-) diff --git a/api/apps/restful_apis/agent_api.py b/api/apps/restful_apis/agent_api.py index e16f42d385..ded7a5b37c 100644 --- a/api/apps/restful_apis/agent_api.py +++ b/api/apps/restful_apis/agent_api.py @@ -1135,6 +1135,9 @@ async def rerun_agent(tenant_id): if 0 < doc["progress"] < 1: return get_data_error_result(message=f"`{doc['name']}` is processing...") + from rag.advanced_rag.knowlege_compile.dataset_nav import remove_dataset_nav_doc_sync + + remove_dataset_nav_doc_sync(tenant_id, doc["kb_id"], doc["id"]) if settings.docStoreConn.index_exist(search.index_name(tenant_id), doc["kb_id"]): settings.docStoreConn.delete({"doc_id": doc["id"]}, search.index_name(tenant_id), doc["kb_id"]) doc["progress_msg"] = "" diff --git a/api/apps/restful_apis/document_api.py b/api/apps/restful_apis/document_api.py index a17cfbafdc..c258ebc7d8 100644 --- a/api/apps/restful_apis/document_api.py +++ b/api/apps/restful_apis/document_api.py @@ -1470,6 +1470,9 @@ def _run_sync(user_id: str, req): DocumentService.update_by_id(doc_id, info) if req.get("delete", False): TaskService.filter_delete([Task.doc_id == doc_id]) + from rag.advanced_rag.knowlege_compile.dataset_nav import remove_dataset_nav_doc_sync + + remove_dataset_nav_doc_sync(doc_tenant_id, doc.kb_id, doc.id) if settings.docStoreConn.index_exist(search.index_name(doc_tenant_id), doc.kb_id): settings.docStoreConn.delete({"doc_id": doc_id}, search.index_name(doc_tenant_id), doc.kb_id) @@ -1580,6 +1583,9 @@ async def parse_documents(tenant_id, dataset_id): DocumentService.update_by_id(doc_id, info) TaskService.filter_delete([Task.doc_id == doc_id]) + from rag.advanced_rag.knowlege_compile.dataset_nav import remove_dataset_nav_doc_sync + + remove_dataset_nav_doc_sync(tenant_id, doc.kb_id, doc.id) if settings.docStoreConn.index_exist(search.index_name(tenant_id), doc.kb_id): settings.docStoreConn.delete({"doc_id": doc_id}, search.index_name(tenant_id), doc.kb_id) diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py index 54eeeb5f41..4f780a3ccd 100644 --- a/api/apps/services/dataset_api_service.py +++ b/api/apps/services/dataset_api_service.py @@ -2452,7 +2452,10 @@ def _nav_item(row: dict) -> dict: payload = json.loads(row.get("content_with_weight") or "{}") except Exception: payload = {} - is_cluster = (row.get("type_kwd") or payload.get("type")) == "nav_cluster" + row_type = row.get("type_kwd") + if isinstance(row_type, (list, tuple, set)): + row_type = next(iter(row_type), None) + is_cluster = (row_type or payload.get("type")) == "nav_cluster" return { "name": row.get("name") or "", "description": payload.get("description") or "", @@ -2551,7 +2554,8 @@ async def delete_nav(dataset_id: str, tenant_id: str): index_nm, _ = pack try: - deleted = settings.docStoreConn.delete( + deleted = await thread_pool_exec( + settings.docStoreConn.delete, {"compile_kwd": [_NAV_COMPILE_KWD]}, index_nm, dataset_id, @@ -2592,16 +2596,17 @@ async def delete_nav_node(dataset_id: str, tenant_id: str, name: str): if not frontier: break try: - res = settings.docStoreConn.search( - select_fields=["name"], - highlight_fields=[], - condition={"compile_kwd": [_NAV_COMPILE_KWD], "parent_kwd": frontier}, - match_expressions=[], - order_by=OrderByExpr(), - offset=0, - limit=10000, - index_names=index_nm, - knowledgebase_ids=[dataset_id], + res = await thread_pool_exec( + settings.docStoreConn.search, + ["name"], + [], + {"compile_kwd": [_NAV_COMPILE_KWD], "parent_kwd": frontier}, + [], + OrderByExpr(), + 0, + 10000, + index_nm, + [dataset_id], ) rows = settings.docStoreConn.get_fields(res, ["name"]) or {} except Exception: @@ -2616,7 +2621,8 @@ async def delete_nav_node(dataset_id: str, tenant_id: str, name: str): frontier = nxt try: - deleted = settings.docStoreConn.delete( + deleted = await thread_pool_exec( + settings.docStoreConn.delete, {"compile_kwd": [_NAV_COMPILE_KWD], "name": list(names)}, index_nm, dataset_id, diff --git a/api/db/services/document_service.py b/api/db/services/document_service.py index bb4c9acf1c..3f75f94b7d 100644 --- a/api/db/services/document_service.py +++ b/api/db/services/document_service.py @@ -492,6 +492,20 @@ class DocumentService(CommonService): except Exception as e: logging.warning(f"Failed to delete thumbnail for document {doc.id}: {e}") + # Prune this doc's line from the KB's tree-kind navigation before the + # broad doc_id delete below removes the nav_doc row needed to locate + # and update its parent cluster. + try: + from rag.advanced_rag.knowlege_compile.dataset_nav import ( + remove_dataset_nav_doc_sync, + ) + + remove_dataset_nav_doc_sync(tenant_id, doc.kb_id, doc.id) + except Exception as e: + logging.warning( + f"Failed to prune dataset_nav for document {doc.id}: {e}", + ) + # Delete chunks from doc store - this is critical, log errors try: settings.docStoreConn.delete({"doc_id": doc.id}, chunk_index_name, doc.kb_id) @@ -507,20 +521,6 @@ class DocumentService(CommonService): except Exception as e: logging.warning(f"Failed to clean up artifact products for document {doc.id}: {e}") - # Prune this doc's line from the KB's tree-kind navigation - # markdown (best-effort — the markdown is a downstream artifact, - # and failure here must not block the document delete). - try: - from rag.advanced_rag.knowlege_compile.dataset_nav import ( - remove_dataset_nav_doc_sync, - ) - - remove_dataset_nav_doc_sync(tenant_id, doc.kb_id, doc.id) - except Exception as e: - logging.warning( - f"Failed to prune dataset_nav for document {doc.id}: {e}", - ) - # Delete document metadata (non-critical, log and continue) try: DocMetadataService.delete_document_metadata(doc.id, doc.kb_id, tenant_id) diff --git a/common/doc_store/infinity_conn_base.py b/common/doc_store/infinity_conn_base.py index c89d38e64e..0484468860 100644 --- a/common/doc_store/infinity_conn_base.py +++ b/common/doc_store/infinity_conn_base.py @@ -33,7 +33,6 @@ from rag.nlp import is_english from common import settings from common.doc_store.doc_store_base import DocStoreConnection, MatchExpr, OrderByExpr - # Concurrent CREATE/DROP TABLE on the same Infinity instance can race on # Infinity's RocksDB-backed catalog counters (e.g. ``db|1|next_table_id``). # When two writers touch the counter at the same instant, Infinity surfaces @@ -321,6 +320,17 @@ class InfinityConnectionBase(DocStoreConnection): json_conditions.append(f"json_contains({k}, '{literal}')") if json_conditions: cond.append("(" + " or ".join(json_conditions) + ")") + elif k in {"compile_kwd", "type_kwd", "parent_kwd"}: + values = v if isinstance(v, list) else [v] + exact_conditions = [] + for item in values: + if isinstance(item, str): + item = item.replace("'", "''") + exact_conditions.append(f"{k}='{item}'") + else: + exact_conditions.append(f"{k}={item}") + if exact_conditions: + cond.append("(" + " or ".join(exact_conditions) + ")") elif self.field_keyword(k): if isinstance(v, list): inCond = list() diff --git a/rag/advanced_rag/knowlege_compile/dataset_nav.py b/rag/advanced_rag/knowlege_compile/dataset_nav.py index 90f4d1c6d8..1d910a802a 100644 --- a/rag/advanced_rag/knowlege_compile/dataset_nav.py +++ b/rag/advanced_rag/knowlege_compile/dataset_nav.py @@ -185,6 +185,7 @@ async def _store_knn( "content_with_weight", "name", "doc_id", + "compile_kwd", "type_kwd", "parent_kwd", "depth_int", @@ -213,7 +214,13 @@ async def _store_knn( [kb_id], ) results = settings.docStoreConn.get_fields(res, fields) if res else {} - return list(results.values()) + rows = list(results.values()) + if filter_condition and any(not _matches_condition(row, filter_condition) for row in rows): + scanned = await _store_search(tenant_id, kb_id, filter_condition, fields, limit=10000) + rows = [row for row in scanned if _vector_len(row.get(vf)) == vec_dim and _matches_condition(row, filter_condition)] + rows.sort(key=lambda row: _cosine_sim(vec, row.get(vf)), reverse=True) + rows = rows[:top_k] + return rows async def _store_upsert(tenant_id: str, kb_id: str, doc: dict) -> None: @@ -388,6 +395,19 @@ def _fine_tokenize(text: str) -> str: return rag_tokenizer.fine_grained_tokenize(text) +def _matches_condition(row: dict, condition: dict) -> bool: + """Check the simple equality filters used by dataset navigation.""" + for field, expected in condition.items(): + if not expected or field == "kb_id": + continue + actual = row.get(field) + actual_values = actual if isinstance(actual, (list, tuple, set)) else [actual] + expected_values = expected if isinstance(expected, (list, tuple, set)) else [expected] + if not any(str(value) == str(item) for value in actual_values for item in expected_values): + return False + return True + + # --------------------------------------------------------------------------- # Incremental clustering core # --------------------------------------------------------------------------- @@ -426,6 +446,7 @@ async def _find_best_cluster( # compute actual similarity to root stored = best.get(_vec_field(vec_dim)) sim = _cosine_sim(doc_embedding, stored) + visited_names = {best_name} # Step 2: recursively descend into children while sim >= _RECURSE_THRESHOLD: @@ -439,14 +460,18 @@ async def _find_best_cluster( if not children: break child = children[0] + child_name = child.get("name", "") + if not child_name or child_name in visited_names: + break stored = child.get(_vec_field(vec_dim)) child_sim = _cosine_sim(doc_embedding, stored) if child_sim < _RECURSE_THRESHOLD: break - best_name = child.get("name", best_name) + best_name = child_name best_parent = best.get("parent_kwd", best_parent) sim = child_sim best = child + visited_names.add(best_name) return best_name, best_parent, sim @@ -568,15 +593,9 @@ async def upsert_dataset_nav_doc( logging.info("dataset_nav: skipping doc=%s (kb=%s) — no summary", doc_id, kb_id) return - # 2. Check if this doc already has a nav_doc row - existing_doc = await _store_get(tenant_id, kb_id, _nav_doc_id(doc_id)) - if existing_doc: - old_payload = json.loads(existing_doc.get("content_with_weight") or "{}") - if old_payload.get("description") == summary: - logging.info("dataset_nav: doc=%s unchanged, skipping", doc_id) - return - - # 3. Embed doc summary + # 2. Embed doc summary before taking the KB lock. The result is + # independent of the nav tree; all tree reads, deletes, and writes below + # are serialized by the same lock. doc_embedding = await _embed(embd_mdl, summary) if embd_mdl else [] vec_dim = len(doc_embedding) @@ -592,8 +611,18 @@ async def upsert_dataset_nav_doc( return try: + # 3. Check and replace the existing nav_doc while holding the same + # lock as placement. Otherwise concurrent updates can both observe + # the old row and race while deleting/rebuilding its parent cluster. + existing_doc = await _store_get(tenant_id, kb_id, _nav_doc_id(doc_id)) + if existing_doc: + old_payload = json.loads(existing_doc.get("content_with_weight") or "{}") + if old_payload.get("description") == summary: + return + await _remove_dataset_nav_doc_locked(tenant_id, kb_id, doc_id) + # 4. Layered KNN search for nearest cluster - if doc_embedding: + if _vector_len(doc_embedding) > 0: best_name, best_parent, sim = await _find_best_cluster( tenant_id, kb_id, @@ -726,6 +755,45 @@ async def upsert_dataset_nav_doc( logging.exception("dataset_nav: lock release failed for kb=%s", kb_id) +async def _remove_dataset_nav_doc_locked( + tenant_id: str, + kb_id: str, + doc_id: str, +) -> None: + """Remove a document nav row; the caller must hold the KB nav lock.""" + # 1. Find and delete the nav_doc row + doc_row_id = _nav_doc_id(doc_id) + doc_row = await _store_get(tenant_id, kb_id, doc_row_id) + if not doc_row: + return + parent_name = doc_row.get("parent_kwd", "") + await _store_delete(tenant_id, kb_id, doc_row_id) + + # 2. Remove doc_id from the parent cluster's doc_ids_kwd + if parent_name and parent_name != "root": + cluster_id = _nav_cluster_id(kb_id, parent_name) + cluster_row = await _store_get(tenant_id, kb_id, cluster_id) + if cluster_row: + doc_ids = cluster_row.get("doc_ids_kwd") or [] + if doc_id in doc_ids: + doc_ids.remove(doc_id) + if not doc_ids: + # Cluster is empty — delete it + await _store_delete(tenant_id, kb_id, cluster_id) + # Recurse: check grandparent + grandparent = cluster_row.get("parent_kwd", "") + if grandparent and grandparent != "root": + await _cleanup_empty_cluster( + tenant_id, + kb_id, + grandparent, + ) + else: + cluster_row["doc_ids_kwd"] = doc_ids + cluster_row["doc_count_int"] = len(doc_ids) + await _store_upsert(tenant_id, kb_id, cluster_row) + + async def remove_dataset_nav_doc( tenant_id: str, kb_id: str, @@ -751,37 +819,7 @@ async def remove_dataset_nav_doc( return try: - # 1. Find and delete the nav_doc row - doc_row_id = _nav_doc_id(doc_id) - doc_row = await _store_get(tenant_id, kb_id, doc_row_id) - if not doc_row: - return - parent_name = doc_row.get("parent_kwd", "") - await _store_delete(tenant_id, kb_id, doc_row_id) - - # 2. Remove doc_id from the parent cluster's doc_ids_kwd - if parent_name and parent_name != "root": - cluster_id = _nav_cluster_id(kb_id, parent_name) - cluster_row = await _store_get(tenant_id, kb_id, cluster_id) - if cluster_row: - doc_ids = cluster_row.get("doc_ids_kwd") or [] - if doc_id in doc_ids: - doc_ids.remove(doc_id) - if not doc_ids: - # Cluster is empty — delete it - await _store_delete(tenant_id, kb_id, cluster_id) - # Recurse: check grandparent - grandparent = cluster_row.get("parent_kwd", "") - if grandparent and grandparent != "root": - await _cleanup_empty_cluster( - tenant_id, - kb_id, - grandparent, - ) - else: - cluster_row["doc_ids_kwd"] = doc_ids - cluster_row["doc_count_int"] = len(doc_ids) - await _store_upsert(tenant_id, kb_id, cluster_row) + await _remove_dataset_nav_doc_locked(tenant_id, kb_id, doc_id) except Exception: logging.exception( "dataset_nav: remove failed for kb=%s doc=%s", diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py index 58c04bbc78..8747fd99d3 100644 --- a/rag/advanced_rag/knowlege_compile/runner.py +++ b/rag/advanced_rag/knowlege_compile/runner.py @@ -197,13 +197,27 @@ async def _upsert_dataset_nav_from_page_index( if cancel_check(): raise TaskCanceledException("Task was cancelled before dataset navigation update") try: + # PageIndex rows use their preserved template kind as the + # compile keyword. Older rows were written as ``timeline`` + # before compilation kinds were kept distinct, so fall back to + # the legacy keyword when rebuilding an existing graph. graph = await rebuild_structure_graph_json( tenant_id, kb_id, doc_id, - "timeline", + "page_index", compilation_template_id=template_id, ) + summary = _page_index_graph_summary(graph) + if not summary: + graph = await rebuild_structure_graph_json( + tenant_id, + kb_id, + doc_id, + "timeline", + compilation_template_id=template_id, + ) + summary = _page_index_graph_summary(graph) except Exception: logging.exception( "page_index: failed to rebuild graph summary for dataset_nav doc %s template %s", @@ -212,7 +226,6 @@ async def _upsert_dataset_nav_from_page_index( ) continue - summary = _page_index_graph_summary(graph) if summary: summaries.append(summary) chat_mdl = chat_mdl or chat_mdl_by_tid.get(template_id) diff --git a/rag/flow/compiler/compiler.py b/rag/flow/compiler/compiler.py index 5ccd750f27..ccbea4e02f 100644 --- a/rag/flow/compiler/compiler.py +++ b/rag/flow/compiler/compiler.py @@ -163,7 +163,14 @@ class Compiler(ProcessBase, LLM): try: from rag.advanced_rag.knowlege_compile.dataset_nav import upsert_dataset_nav_doc - await upsert_dataset_nav_doc(tenant_id, kb_id, doc_id, tree) + await upsert_dataset_nav_doc( + tenant_id, + kb_id, + doc_id, + tree, + embd_mdl=embedding_model, + chat_mdl=chat_mdl_by_tid[template_id], + ) except Exception: logging.exception("Compiler: tree-template %s dataset navigation upsert failed for doc %s", template_id, doc_id) diff --git a/rag/svr/task_executor_refactor/chunk_post_processor.py b/rag/svr/task_executor_refactor/chunk_post_processor.py index 61956ee841..3e6cb4b300 100644 --- a/rag/svr/task_executor_refactor/chunk_post_processor.py +++ b/rag/svr/task_executor_refactor/chunk_post_processor.py @@ -1009,6 +1009,8 @@ async def run_tree_templates( ctx.kb_id, doc_id, tree, + embd_mdl=embedding_model, + chat_mdl=chat_mdl_by_tid[template_id], ) except Exception: logging.exception( diff --git a/rag/utils/infinity_conn.py b/rag/utils/infinity_conn.py index b57acb5d41..2d257fac96 100644 --- a/rag/utils/infinity_conn.py +++ b/rag/utils/infinity_conn.py @@ -649,6 +649,22 @@ class InfinityConnection(InfinityConnectionBase): if k in new_value: del new_value[k] + # The Infinity Python client inspects value[0] for list values + # while building an update expression. An empty list therefore + # raises IndexError before the request reaches Infinity. Keep + # JSON and keyword-list columns clearable, but do not send empty + # values for other columns (for example, an empty vector returned + # by a partial row read). + for k, v in list(new_value.items()): + if not isinstance(v, list) or v: + continue + if k in _JSON_LIST_FIELDS: + new_value[k] = json.dumps([], ensure_ascii=False) + elif self.field_keyword(k): + new_value[k] = "" + else: + del new_value[k] + remove_opt = {} # "[k,new_value]": [id_to_update, ...] if removeValue: col_to_remove = list(removeValue.keys()) diff --git a/web/src/components/ui/tree-view.tsx b/web/src/components/ui/tree-view.tsx index 661694e521..02d6312e99 100644 --- a/web/src/components/ui/tree-view.tsx +++ b/web/src/components/ui/tree-view.tsx @@ -23,6 +23,7 @@ export interface TreeDataItem { selectedIcon?: any; openIcon?: any; children?: TreeDataItem[]; + hasChildren?: boolean; actions?: React.ReactNode; onClick?: () => void; } @@ -142,7 +143,8 @@ const TreeItem = React.forwardRef(