diff --git a/api/apps/restful_apis/chunk_api.py b/api/apps/restful_apis/chunk_api.py index 3de5463629..151ba19f66 100644 --- a/api/apps/restful_apis/chunk_api.py +++ b/api/apps/restful_apis/chunk_api.py @@ -166,10 +166,7 @@ def _get_dataset_tenant_id(dataset_id): def _compilation_template_kind(kind) -> str: if not isinstance(kind, str): return "" - normalized = kind.strip().lower().replace("-", "_") - if normalized in {"pageindex", "page_index", "knowledge_graph"}: - return "timeline" - return normalized + return kind.strip().lower().replace("-", "_") def _resolve_reference_metadata(req: dict, search_config: dict | None = None): @@ -581,6 +578,7 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id): """ from rag.nlp import search from api.db.services.compilation_template_group_service import CompilationTemplateGroupService + from api.db.services.compilation_template_service import CompilationTemplateService if not KnowledgebaseService.accessible(kb_id=dataset_id, user_id=tenant_id): return get_error_data_result(message=f"You don't own the dataset {dataset_id}.") @@ -624,6 +622,7 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id): seen_configured_ids: set[str] = set() template_meta: dict[str, dict] = {} template_meta_by_kind: dict[str, list[dict]] = {} + template_meta_by_id: dict[str, dict] = {} for group_id in group_ids: group = CompilationTemplateGroupService.get_saved(group_id, tenant_id) if not group: @@ -648,6 +647,7 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id): "kind_norm": kind_norm, } template_meta[template_id] = meta + template_meta_by_id[template_id] = meta template_meta_by_kind.setdefault(kind_norm, []).append(meta) # Load every graph row for this doc in one shot. Each row corresponds @@ -740,6 +740,24 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id): bucket_id = tid row_kind_norm = _compilation_template_kind(kind_val) meta = template_meta.get(bucket_id) + if not meta: + # Pipeline Compiler receives template groups as component + # parameters and does not persist those group ids in the + # document parser_config. Resolve the exact template id + # directly so Pipeline-produced rows do not fall back to + # exposing the opaque id as the display name. + if bucket_id not in template_meta_by_id: + saved_template = CompilationTemplateService.get_saved(bucket_id, tenant_id) + if saved_template: + template_meta_by_id[bucket_id] = { + "template_id": bucket_id, + "template_name": saved_template.get("name") or bucket_id, + "kind": saved_template.get("kind") or kind_val, + "kind_norm": _compilation_template_kind(saved_template.get("kind")), + } + else: + template_meta_by_id[bucket_id] = {} + meta = template_meta_by_id[bucket_id] or None if not meta: kind_matches = template_meta_by_kind.get(row_kind_norm) or [] if len(kind_matches) == 1: diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py index 0bfd4450a6..c2ba8bb127 100644 --- a/rag/advanced_rag/knowlege_compile/runner.py +++ b/rag/advanced_rag/knowlege_compile/runner.py @@ -45,6 +45,7 @@ from common.exceptions import TaskCanceledException from rag.advanced_rag.knowlege_compile.structure import ( LLMCallPool, compile_structure_from_text, + cleanup_timeline_isolated_entities, merge_compiled_structures, ) @@ -372,6 +373,25 @@ async def run_structure_compile_over_batches( finally: flush_tasks.clear() + # Timeline entity cleanup must happen after every flush has completed; + # otherwise an entity can look isolated in one flush and be referenced by + # a relation from a later flush. Keep this scoped to timeline templates. + for template_id, _ in active_templates: + if template_kinds.get(template_id) != "timeline": + continue + try: + await cleanup_timeline_isolated_entities( + tenant_id, + kb_id, + doc_id, + compilation_template_id=template_id, + ) + except Exception: + logging.exception( + "document_structure_compile: timeline isolated-entity cleanup failed for template=%s", + template_id, + ) + for idx, (template_id, parser_cfg) in enumerate(active_templates): if cancel_check(): raise TaskCanceledException("Task was cancelled during document knowledge compilation") diff --git a/rag/advanced_rag/knowlege_compile/structure.py b/rag/advanced_rag/knowlege_compile/structure.py index 32f62ba4a6..e3b437df46 100644 --- a/rag/advanced_rag/knowlege_compile/structure.py +++ b/rag/advanced_rag/knowlege_compile/structure.py @@ -110,10 +110,7 @@ class PooledChatModel: def _struct_normalize_kind(kind) -> str: if not isinstance(kind, str): return "" - normalized = kind.strip().lower().replace("-", "_") - if normalized in {"pageindex", "page_index", "knowledge_graph"}: - return "timeline" - return normalized + return kind.strip().lower().replace("-", "_") def _struct_localize(value, language: str = "en") -> str: @@ -1869,6 +1866,7 @@ async def _struct_rebuild_graph_json( [kb_id], ) rows = settings.docStoreConn.get_fields(res, fields) + entities: list[dict] = [] relations: list[dict] = [] for row in rows.values(): @@ -1888,6 +1886,91 @@ async def _struct_rebuild_graph_json( } +async def cleanup_timeline_isolated_entities( + tenant_id: str, + kb_id: str, + doc_id: str, + compilation_template_id: str | None = None, +) -> int: + """Remove timeline entity rows that are not used by any relation. + + This runs after all structure flushes for the document have completed; + otherwise an entity can look isolated in one flush and be referenced by a + relation from a later flush. The cleanup is intentionally limited to the + ``timeline`` compile kind. + """ + from common import settings + from common.doc_store.doc_store_base import OrderByExpr + from rag.nlp import search as _rag_search + + index = _rag_search.index_name(tenant_id) + fields = [ + "content_with_weight", + "knowledge_graph_kwd", + "from_entity_kwd", + "to_entity_kwd", + ] + condition: dict = { + "doc_id": [doc_id], + "compile_kwd": ["timeline"], + "knowledge_graph_kwd": ["entity", "relation"], + } + if compilation_template_id: + condition["compilation_template_ids"] = [compilation_template_id] + + res = await thread_pool_exec( + settings.docStoreConn.search, + fields, + [], + condition, + [], + OrderByExpr(), + 0, + 10000, + index, + [kb_id], + ) + rows = settings.docStoreConn.get_fields(res, fields) or {} + connected_names: set[str] = set() + for row in rows.values(): + if row.get("knowledge_graph_kwd") != "relation": + continue + edge = _chain_extract_edge(row) + if edge is not None: + connected_names.update(name.casefold() for name in edge if name) + + orphan_ids = [ + row_id + for row_id, row in rows.items() + if row.get("knowledge_graph_kwd") == "entity" + and _struct_entity_name(row).casefold() not in connected_names + ] + if orphan_ids: + await thread_pool_exec( + settings.docStoreConn.delete, + {"id": orphan_ids}, + index, + kb_id, + ) + logging.info( + "structure graph: removed %d isolated timeline entity row(s) for doc=%s template=%s", + len(orphan_ids), + doc_id, + compilation_template_id or "legacy", + ) + + # Refresh the compact graph after source-row cleanup. This also handles + # the no-relation case, where every timeline entity is isolated. + await rebuild_structure_graph_json( + tenant_id, + kb_id, + doc_id, + "timeline", + compilation_template_id, + ) + return len(orphan_ids) + + async def _struct_upsert_graph_json( graph: dict, tenant_id: str, @@ -2423,5 +2506,6 @@ async def merge_compiled_structures( __all__ = [ "compile_structure_from_text", "merge_compiled_structures", + "cleanup_timeline_isolated_entities", "rebuild_structure_graph_json", ]