From 649b631cacc61b04ca7d69fe406dff9f2eae3012 Mon Sep 17 00:00:00 2001 From: buua436 Date: Wed, 12 Aug 2026 19:12:39 +0800 Subject: [PATCH] fix: improve incremental wiki compilation (#18164) --- api/apps/services/dataset_api_service.py | 59 +- .../knowlege_compile/wiki_incremental.py | 740 +++++++++++++++--- .../dataset_wiki_generator.py | 80 +- .../task_executor_refactor/task_handler.py | 60 ++ .../test_dataset_api_service_list_datasets.py | 36 + .../advanced_rag/knowlege_compile/conftest.py | 10 + .../knowlege_compile/test_wiki_incremental.py | 364 ++++++++- 7 files changed, 1231 insertions(+), 118 deletions(-) diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py index 97c5eae325..ed80c4a430 100644 --- a/api/apps/services/dataset_api_service.py +++ b/api/apps/services/dataset_api_service.py @@ -2167,7 +2167,7 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw # Non-folded template kinds that make a doc eligible for each API kind. _ALTERATION_ELIGIBLE_TEMPLATE_KINDS = { - "wiki": {"artifacts"}, + "wiki": {"wiki"}, "graph": {"knowledge_graph"}, "mindmap": {"mind_map"}, "timeline": {"timeline"}, @@ -3992,7 +3992,7 @@ async def update_wiki_page( # :meth:`FileCommitService.get_page_commit_detail`. -# All seven row types the artifact pipeline writes. Listed in dependency +# All row types the artifact pipeline writes. Listed in dependency # order so partial failures of earlier deletes don't leave behind state # that downstream phases would silently reuse. ``wiki_page_graph`` # is the materialized canvas graph derived from the refined pages — @@ -4004,8 +4004,13 @@ _WIKI_COMPILE_KWDS = ( "wiki_page_draft", "wiki_page", "wiki_page_topic", + "wiki_canonical_entity", + "wiki_plan_group", + "wiki_doc_page_source", + "wiki_mode_meta", "wiki_entity", "wiki_relation", + "wiki_page_graph", ) # Tunables for the incremental graph loader. See ``get_wiki_graph``. @@ -4529,6 +4534,56 @@ async def clear_wiki(dataset_id: str, tenant_id: str): return True, {"deleted": {}} index_nm, _ = pack + # Repair rows damaged by the former doc-page-source upsert before deleting + # that bucket. It updated by ``doc_id`` and could stamp ordinary source + # chunks as ``wiki_doc_page_source``. Those rows retain chunk content; + # genuine tracking rows do not. Removing only the bad marker preserves the + # source chunks while allowing the real tracking rows to be cleared below. + try: + from common.doc_store.doc_store_base import OrderByExpr + + fields = ["id", "content_with_weight"] + offset = 0 + page_size = 1000 + damaged_row_ids: list[str] = [] + while True: + res = await thread_pool_exec( + settings.docStoreConn.search, + fields, + [], + {"compile_kwd": ["wiki_doc_page_source"]}, + [], + OrderByExpr(), + offset, + page_size, + index_nm, + [dataset_id], + ) + rows = settings.docStoreConn.get_fields(res, fields) or {} + for row_id, row in rows.items(): + if row.get("content_with_weight"): + damaged_row_ids.append(row_id) + if len(rows) < page_size: + break + offset += page_size + for row_id in damaged_row_ids: + await thread_pool_exec( + settings.docStoreConn.update, + {"id": row_id}, + {"remove": "compile_kwd"}, + index_nm, + dataset_id, + ) + if damaged_row_ids: + logging.warning( + "clear_wiki: repaired %d source chunk(s) mislabeled as wiki_doc_page_source kb=%s", + len(damaged_row_ids), + dataset_id, + ) + except Exception: + logging.exception("clear_wiki: failed to repair mislabeled source chunks kb=%s", dataset_id) + return False, "Failed to repair legacy Wiki state before clearing" + deleted: dict[str, object] = {} for kwd in _WIKI_COMPILE_KWDS: try: diff --git a/rag/advanced_rag/knowlege_compile/wiki_incremental.py b/rag/advanced_rag/knowlege_compile/wiki_incremental.py index dd50081c79..8b65852ab3 100644 --- a/rag/advanced_rag/knowlege_compile/wiki_incremental.py +++ b/rag/advanced_rag/knowlege_compile/wiki_incremental.py @@ -63,11 +63,9 @@ WIKI_TOPIC_FALLBACK = "General" # bucket for pages that match no topic WIKI_PAGE_TOPIC_CANDIDATE_LIMIT = 50 # Page Router thresholds (kept as code constants — not exposed in YAML) -PAGE_ROUTER_UPDATE_THRESHOLD = 0.80 PAGE_ROUTER_MAYBE_THRESHOLD = 0.50 -PAGE_ROUTER_DIRECT_THRESHOLD = 0.90 -PAGE_ROUTER_MIN_MARGIN = 0.03 PAGE_ROUTER_TOP_K = 5 +PAGE_ROUTER_MAX_CANDIDATES = 12 PAGE_CLUSTER_MIN_PAGES = 8 PAGE_CLUSTER_MAX_PAGES = 60 PAGE_CLUSTER_ITEMS_PER_PAGE = 3 @@ -91,6 +89,11 @@ WIKI_SOURCE_BUDGET_RUNES = 12_000 # per-chunk-batch budget (rune-based, mirrors # ----- helpers --------------------------------------------------------------- +def _wiki_log_stats(stage: str, event: str, **fields) -> None: + """Emit machine-readable compilation statistics for one pipeline stage.""" + logging.info("wiki stats %s", json.dumps({"stage": stage, "event": event, **fields}, ensure_ascii=False, sort_keys=True)) + + def _wiki_derive_page_id(term: str, prefix: str = "concept") -> str: """Derive a URL-safe page identifier from a concept/entity name. @@ -674,6 +677,7 @@ async def _wiki_match_entities( exact_flat[_normalize_key(alias)] = cname name_resolution: dict[str, str] = {} # raw_name → canonical_name + llm_merge_pairs: list[dict[str, str]] = [] unmatched: list[dict] = [] # entities not matched by exact for entry in raw_entities: raw_name = entry["name"] @@ -737,6 +741,7 @@ async def _wiki_match_entities( for raw_name, cname in confirmed: name_resolution[raw_name] = cname confirmed_set.add(raw_name) + llm_merge_pairs.append({"from": raw_name, "into": cname, "scope": "existing_canonical"}) for e, cname in maybe_pairs: if e["name"] not in confirmed_set: still_unmatched.append(e) @@ -824,8 +829,10 @@ async def _wiki_match_entities( if ri != rj: if unmatched[ri].get("claim_count", 0) >= unmatched[rj].get("claim_count", 0): merged_into[rj] = ri + llm_merge_pairs.append({"from": unmatched[rj]["name"], "into": unmatched[ri]["name"], "scope": "intra_build"}) else: merged_into[ri] = rj + llm_merge_pairs.append({"from": unmatched[ri]["name"], "into": unmatched[rj]["name"], "scope": "intra_build"}) # Apply merges merged_indices: dict[int, list[int]] = {} @@ -895,6 +902,10 @@ async def _wiki_match_entities( aliases.discard(cname) canonical_map[cname]["aliases"] = sorted(aliases) + for merge in llm_merge_pairs: + _wiki_log_stats("MATCH", "llm_merge", kb_id=kb_id, incremental=incremental, **merge) + _wiki_log_stats("MATCH", "llm_merge_summary", kb_id=kb_id, incremental=incremental, before=len(raw_entities), after=len(canonical_map), llm_merge_count=len(llm_merge_pairs)) + return canonical_map, name_resolution @@ -987,7 +998,7 @@ async def _search_existing_pages( return results -async def _load_map_relations(tenant_id: str, kb_id: str) -> list[dict]: +async def _load_map_relations(tenant_id: str, kb_id: str, excluded_doc_ids: set[str] | None = None) -> list[dict]: """Load all extracted (from, to, type) relations from wiki_map_extract rows. These are the semantic edges the LLM extracted during MAP. When both @@ -1006,7 +1017,7 @@ async def _load_map_relations(tenant_id: str, kb_id: str) -> list[dict]: try: res = await thread_pool_exec( settings.docStoreConn.search, - ["content_with_weight"], + ["content_with_weight", "doc_id"], [], {"compile_kwd": ["wiki_map_extract"]}, [], @@ -1016,11 +1027,14 @@ async def _load_map_relations(tenant_id: str, kb_id: str) -> list[dict]: index, [kb_id], ) - field_map = settings.docStoreConn.get_fields(res, ["content_with_weight"]) or {} + field_map = settings.docStoreConn.get_fields(res, ["content_with_weight", "doc_id"]) or {} except Exception: logging.exception("wiki: failed to load map relations for kb=%s", kb_id) return relations for row in field_map.values(): + row_doc_ids = _as_str_list(row.get("doc_id")) + if excluded_doc_ids and any(doc_id in excluded_doc_ids for doc_id in row_doc_ids): + continue raw = row.get("content_with_weight") if isinstance(raw, str): try: @@ -1041,7 +1055,11 @@ async def _load_map_relations(tenant_id: str, kb_id: str) -> list[dict]: return relations -async def _wiki_load_pages_for_graph(tenant_id: str, kb_id: str) -> list[dict]: +async def _wiki_load_pages_for_graph( + tenant_id: str, + kb_id: str, + excluded_doc_ids: set[str] | None = None, +) -> list[dict]: """Reload compiled wiki_page rows and project them onto the canvas-graph shape expected by ``dataset_wiki_generator.build_wiki_page_graph``. @@ -1121,6 +1139,37 @@ async def _wiki_load_pages_for_graph(tenant_id: str, kb_id: str) -> list[dict]: if len(field_map) < page_size: break offset += page_size + # A page may contain no generated wikilink even though MAP extracted a + # semantic relation. Rebuild the graph from those grounded MAP relations + # as a fallback; otherwise wiki_entity rows exist but wiki_relation stays + # empty. Page slugs remain the graph identities, while member names, + # titles, and slug suffixes are accepted as relation endpoints. + if pages: + name_to_slug: dict[str, str] = {} + for page in pages: + slug = page["slug"] + names = [slug.rsplit("/", 1)[-1], page.get("title", ""), *page.get("entity_names", [])] + for name in names: + if isinstance(name, str) and name.strip(): + name_to_slug.setdefault(name.strip(), slug) + try: + if excluded_doc_ids is None: + from api.db.services.document_service import DocumentService + + excluded_doc_ids = await thread_pool_exec(DocumentService.get_disabled_doc_ids_by_kb_id, kb_id) + map_relations = await _load_map_relations(tenant_id, kb_id, excluded_doc_ids=excluded_doc_ids) + except Exception: + logging.exception("wiki: failed to load MAP relations for graph fallback kb=%s", kb_id) + map_relations = [] + pages_by_slug = {page["slug"]: page for page in pages} + for relation in map_relations: + source = name_to_slug.get(str(relation.get("from") or "").strip()) + target = name_to_slug.get(str(relation.get("to") or "").strip()) + if not source or not target or source == target: + continue + outlinks = pages_by_slug[source].setdefault("outlinks", []) + if target not in outlinks: + outlinks.append(target) return pages @@ -1139,14 +1188,17 @@ def _wiki_extract_outlinks_from_content(content: str, kb_id: str = "") -> list[s seen: set[str] = set() outlinks: list[str] = [] for m in _WIKILINK_RE.finditer(content): - link = m.group(1).strip() + # ``[[page_slug|display text]]`` stores the page identity before the + # pipe. The display text is only presentation data and must never be + # used as the graph target. + link = m.group(1).split("|", 1)[0].strip() if link and link not in seen: seen.add(link) outlinks.append(link) if kb_id: kb_esc = re.escape(str(kb_id)) for m in re.finditer(rf"\]\(artifact/{kb_esc}/([^)]+)\)", content): - slug = m.group(1).strip() + slug = m.group(1).split("|", 1)[0].strip() if slug and slug not in seen: seen.add(slug) outlinks.append(slug) @@ -1313,7 +1365,11 @@ def _wiki_dedupe_claims(claims: list[dict]) -> list[dict]: return result -def _wiki_topics_for_docs(doc_ids: list[str] | set[str], doc_topics: dict[str, list[str]] | None) -> list[str]: +def _wiki_topics_for_docs( + doc_ids: list[str] | set[str], + doc_topics: dict[str, list[str]] | None, + topic_pool: dict[str, str] | None = None, +) -> list[str]: topics: list[str] = [] seen: set[str] = set() for doc_id in doc_ids: @@ -1321,16 +1377,120 @@ def _wiki_topics_for_docs(doc_ids: list[str] | set[str], doc_topics: dict[str, l if not isinstance(topic, str): continue topic = topic.strip() - key = topic.casefold() - if not topic or key == WIKI_TOPIC_FALLBACK.casefold() or key in seen: + key = _normalize_key(topic) + if not topic or key == _normalize_key(WIKI_TOPIC_FALLBACK) or key in seen: continue seen.add(key) topics.append(topic) - if len(topics) >= WIKI_PAGE_TOPIC_CANDIDATE_LIMIT: - return topics + for topic in (topic_pool or {}).values(): + key = _normalize_key(topic) + if topic and key not in seen: + seen.add(key) + topics.append(topic) return topics +async def _wiki_prepare_topic_embeddings( + doc_topics: dict[str, list[str]], + embd_mdl, + extra_topics: list[str] | None = None, +) -> dict[str, object]: + topics = sorted( + { + topic + for values in list(doc_topics.values()) + [extra_topics or []] + for topic in values + if isinstance(topic, str) and topic.strip() and _normalize_key(topic) != _normalize_key(WIKI_TOPIC_FALLBACK) + }, + key=lambda value: (value.casefold(), value), + ) + if not topics or embd_mdl is None: + return {} + embeddings, _ = await thread_pool_exec(embd_mdl.encode, topics) + return {topic: vector for topic, vector in zip(topics, embeddings, strict=True)} + + +def _wiki_topic_query_text( + page_title: str, + claims: list[dict] | None, + source_chunks: list[dict] | None, + existing_page: dict | None = None, +) -> str: + parts = [f"title={page_title}"] if page_title else [] + if existing_page: + summary = existing_page.get("summary_with_weight") or "" + if summary: + parts.append(f"summary={summary}") + evidence = [] + for claim in (claims or [])[:8]: + if isinstance(claim, dict): + text = claim.get("statement") or claim.get("text") + if text: + evidence.append(str(text)) + if evidence: + parts.append(f"evidence={' | '.join(evidence)}") + chunk_text = [] + for chunk in (source_chunks or [])[:4]: + if isinstance(chunk, dict): + text = chunk.get("text") or chunk.get("content_with_weight") + if text: + chunk_text.append(str(text)[:500]) + if chunk_text: + parts.append(f"source={' | '.join(chunk_text)}") + return "; ".join(parts) + + +async def _wiki_rank_topic_candidates( + page_title: str, + claims: list[dict] | None, + source_chunks: list[dict] | None, + existing_page: dict | None, + topic_candidates: list[str] | None, + topic_embeddings: dict[str, object] | None, + embd_mdl, +) -> list[str]: + """Use embedding only to recall topic candidates; LLM remains the selector.""" + candidates = [] + seen: set[str] = set() + for topic in topic_candidates or []: + if not isinstance(topic, str): + continue + topic = topic.strip() + key = _normalize_key(topic) + if topic and key not in seen: + seen.add(key) + candidates.append(topic) + if len(candidates) <= 1 or embd_mdl is None: + return candidates[:WIKI_PAGE_TOPIC_CANDIDATE_LIMIT] + + query_text = _wiki_topic_query_text(page_title, claims, source_chunks, existing_page) + query_embedding, _ = await thread_pool_exec(embd_mdl.encode, [query_text]) + query = np.asarray(query_embedding[0], dtype=np.float32) + query_norm = np.linalg.norm(query) + if query_norm <= 0: + return candidates[:WIKI_PAGE_TOPIC_CANDIDATE_LIMIT] + query = query / query_norm + + local_topic_embeddings = dict(topic_embeddings or {}) + missing_topics = [topic for topic in candidates if topic not in local_topic_embeddings] + if missing_topics: + encoded, _ = await thread_pool_exec(embd_mdl.encode, missing_topics) + local_topic_embeddings.update({topic: vector for topic, vector in zip(missing_topics, encoded, strict=True)}) + if topic_embeddings is not None: + topic_embeddings.update({topic: vector for topic, vector in zip(missing_topics, encoded, strict=True)}) + + ranked = [] + for topic in candidates: + vector = np.asarray(local_topic_embeddings.get(topic), dtype=np.float32) if topic in local_topic_embeddings else None + if vector is None or vector.size == 0: + continue + norm = np.linalg.norm(vector) + score = float(np.dot(query, vector / norm)) if norm > 0 else -1.0 + ranked.append((score, topic)) + ranked.sort(key=lambda item: (-item[0], item[1])) + return [topic for _, topic in ranked[:WIKI_PAGE_TOPIC_CANDIDATE_LIMIT]] + + def _wiki_decide_concept_pages(all_concepts: list[dict]) -> list[dict]: """Return every concept as a wiki page. @@ -1593,7 +1753,11 @@ async def _wiki_update_doc_page_source( if existing_map: await thread_pool_exec( settings.docStoreConn.update, - {"doc_id": doc_id}, + # ``doc_id`` is shared by source chunks, MAP resume rows, and this + # tracking row. Updating by doc_id rewrites every one of them into + # ``wiki_doc_page_source``. The tracking row has a stable unique + # id, so updates must always use that identity. + {"id": doc["id"]}, doc, index, kb_id, @@ -1682,8 +1846,14 @@ async def _wiki_refine_page( page_version: int, entity_names: list[str] | None = None, page_embedding=None, + embed_routing_context: bool = False, source_doc_ids: list[str] | None = None, topic_candidates: list[str] | None = None, + topic_selection_stats: dict[str, int] | None = None, + topic_embeddings: dict[str, object] | None = None, + topic_pool: dict[str, str] | None = None, + topic_pool_lock: asyncio.Lock | None = None, + member_evidence: list[dict] | None = None, ) -> dict | None: """Run a single Mode A REFINE action on one concept page. @@ -1695,6 +1865,16 @@ async def _wiki_refine_page( if not page_id or not str(page_id).strip(): return existing_page + topic_candidates = await _wiki_rank_topic_candidates( + page_title, + claims, + source_chunks, + existing_page, + topic_candidates, + topic_embeddings, + embd_mdl, + ) + if mode == "delete": deleted_count = await thread_pool_exec( settings.docStoreConn.delete, @@ -1736,6 +1916,7 @@ async def _wiki_refine_page( available_pages, contextual_hints, topic_candidates, + member_evidence, ) elif mode == "re-synthesize": system_prompt = _WIKI_MODE_A_MODIFY_SYSTEM @@ -1751,6 +1932,7 @@ async def _wiki_refine_page( contextual_hints, topic_candidates, force_full=True, + member_evidence=member_evidence, ) else: # modify system_prompt = _WIKI_MODE_A_MODIFY_SYSTEM @@ -1766,6 +1948,7 @@ async def _wiki_refine_page( contextual_hints, topic_candidates, force_full=False, + member_evidence=member_evidence, ) # Call LLM @@ -1783,16 +1966,21 @@ async def _wiki_refine_page( # knowledge-base-wide embedding nearest-neighbour pass. content_lines = response.strip().splitlines() summary = "" + title = "" topic = "" while content_lines: line = content_lines[0].strip() - if not line and (summary or topic): + if not line and (summary or title or topic): content_lines.pop(0) continue if line.upper().startswith("SUMMARY:") and not summary: summary = line.split(":", 1)[1].strip() content_lines.pop(0) continue + if line.upper().startswith("TITLE:") and not title: + title = line.split(":", 1)[1].strip() + content_lines.pop(0) + continue if line.upper().startswith("TOPIC:") and not topic: topic = line.split(":", 1)[1].strip() content_lines.pop(0) @@ -1802,13 +1990,52 @@ async def _wiki_refine_page( if not content: return existing_page - # Build the wiki_page dict + # Build the wiki_page dict. Single-member pages keep their original title; + # only grouped pages may receive a synthesized title from the LLM. existing = existing_page or {} + member_names = {str(name).strip() for name in (entity_names or []) if str(name).strip()} + if len(member_names) <= 1: + title = str(existing.get("title_kwd") or page_title).strip() + else: + title = title or str(existing.get("title_kwd") or page_title).strip() if not topic: existing_topic = existing.get("topic_kwd") if isinstance(existing_topic, (list, tuple)): existing_topic = existing_topic[0] if existing_topic else "" topic = str(existing_topic or WIKI_TOPIC_FALLBACK).strip() + topic_key = _normalize_key(topic) + candidate_keys = {_normalize_key(candidate) for candidate in topic_candidates or [] if candidate} + is_new_topic = bool(topic and topic_key not in candidate_keys) + added_to_candidates = False + if is_new_topic and topic_pool is not None: + added_to_candidates = topic_key not in topic_pool + if topic_pool_lock is not None: + async with topic_pool_lock: + added_to_candidates = topic_key not in topic_pool + topic_pool.setdefault(topic_key, topic) + else: + topic_pool.setdefault(topic_key, topic) + if topic_embeddings is not None and topic not in topic_embeddings: + encoded, _ = await thread_pool_exec(embd_mdl.encode, [topic]) + topic_embeddings[topic] = encoded[0] + if added_to_candidates and topic_selection_stats is not None: + topic_selection_stats["new_added"] = topic_selection_stats.get("new_added", 0) + 1 + normalized_topic_candidates = {_normalize_key(candidate) for candidate in topic_candidates or [] if candidate} + topic_in_candidates = _normalize_key(topic) in normalized_topic_candidates + _wiki_log_stats( + "TOPIC", + "page_selection", + page_id=page_id, + candidate_count=len(normalized_topic_candidates), + candidates=list((topic_candidates or [])[:WIKI_PAGE_TOPIC_CANDIDATE_LIMIT]), + selected=topic, + is_new=not topic_in_candidates, + added_to_candidates=added_to_candidates, + ) + if topic_selection_stats is not None: + topic_selection_stats["selected"] = topic_selection_stats.get("selected", 0) + 1 + if not topic_in_candidates: + topic_selection_stats["new"] = topic_selection_stats.get("new", 0) + 1 new_version = page_version + 1 raw_existing_claims = existing.get("claims", []) if isinstance(raw_existing_claims, str): @@ -1857,12 +2084,25 @@ async def _wiki_refine_page( if did and did not in doc_ids: doc_ids.append(did) - # Embed for search - from common.misc_utils import thread_pool_exec + # Mode B embeds the generated page subject for subsequent routing. A + # centroid of member vectors favors lexical similarity and loses thematic + # relations (for example a company and a technology it develops). from rag.nlp import rag_tokenizer if page_embedding is None: - embeddings, _ = await thread_pool_exec(embd_mdl.encode, [summary or content[:200]]) + embedding_text = summary or content[:200] + if embed_routing_context: + embedding_text = "; ".join( + part + for part in ( + f"title={page_title}" if page_title else "", + f"summary={summary}" if summary else "", + f"members={', '.join(sorted(set(entity_names or [])))}" if entity_names else "", + f"content={content[:500]}" if content else "", + ) + if part + ) + embeddings, _ = await thread_pool_exec(embd_mdl.encode, [embedding_text]) page_embedding = embeddings[0] # Derive vector dimension from the embedding shape @@ -1873,9 +2113,9 @@ async def _wiki_refine_page( page = { "id": _stable_row_id(WIKI_PAGE_COMPILE_KWD, kb_id, page_id), "slug_kwd": page_id, - "title_kwd": page_title, + "title_kwd": title, "md_with_weight": content, - "summary_with_weight": summary or page_title, + "summary_with_weight": summary or title, "entity_names_kwd": sorted(set(entity_names or [page_title])), "source_chunk_ids": sorted(source_chunk_ids), "source_doc_ids": doc_ids, @@ -1886,7 +2126,7 @@ async def _wiki_refine_page( "topic_kwd": topic, "compile_kwd": WIKI_PAGE_COMPILE_KWD, "knowledge_graph_kwd": WIKI_PAGE_COMPILE_KWD, - "title_tks": rag_tokenizer.tokenize(page_title), + "title_tks": rag_tokenizer.tokenize(title), "content_ltks": content_ltks, "content_sm_ltks": rag_tokenizer.fine_grained_tokenize(content_ltks), } @@ -1992,14 +2232,20 @@ def _build_mode_a_generate_prompt( available_pages: list[str], contextual_hints: str, topic_candidates: list[str] | None = None, + member_evidence: list[dict] | None = None, ) -> str: chunks_text = _build_source_chunks_block(source_chunks) claims_text = "\n".join(f"- {c.get('statement', c.get('text', ''))}" for c in claims) if claims else "(no claims)" + member_text = _build_member_evidence_block(member_evidence) + return f"""## Concept Page Identity - Page ID: {page_id} - Title: {page_title} +## Required Page Members +{member_text or "(single member page)"} + ## Source Chunks (verbatim source text — ground every fact in these) {chunks_text or "(no source chunks available)"} @@ -2028,12 +2274,14 @@ def _build_mode_a_modify_prompt( contextual_hints: str, topic_candidates: list[str] | None = None, force_full: bool = False, + member_evidence: list[dict] | None = None, ) -> str: existing_content = existing_page.get("md_with_weight", "") if existing_page else "" existing_topic = existing_page.get("topic_kwd", "") if existing_page else "" if isinstance(existing_topic, (list, tuple)): existing_topic = existing_topic[0] if existing_topic else "" topic_block = chr(10).join(f"- {topic}" for topic in (topic_candidates or [])[:WIKI_PAGE_TOPIC_CANDIDATE_LIMIT]) + member_text = _build_member_evidence_block(member_evidence) if not force_full: additions_text = "\n".join(f"- {c.get('statement', c.get('text', ''))}" for c in (additions or [])) if additions else "(none)" @@ -2044,6 +2292,9 @@ def _build_mode_a_modify_prompt( - Page ID: {page_id} - Title: {page_title} +## Required Page Members +{member_text or "(single member page)"} + ## Current Page {existing_content[:10000] if existing_content else "(empty)"} @@ -2076,6 +2327,9 @@ def _build_mode_a_modify_prompt( - Page ID: {page_id} - Title: {page_title} +## Required Page Members +{member_text or "(single member page)"} + ## All Source Chunks (for full re-synthesis — verbatim source text) {chunks_text or "(none)"} @@ -2095,6 +2349,25 @@ def _build_mode_a_modify_prompt( """ +def _build_member_evidence_block(member_evidence: list[dict] | None) -> str: + """Render per-member evidence so grouped pages cannot silently omit members.""" + if not member_evidence: + return "" + blocks: list[str] = [] + for member in member_evidence: + name = str(member.get("name") or "").strip() + if not name: + continue + claims = member.get("claims") or [] + claims_text = ( + "\n".join(f"- {c.get('statement', c.get('text', ''))}" for c in claims if isinstance(c, dict) and c.get("statement", c.get("text", ""))) + or "(no extracted claims; use the member's source evidence)" + ) + chunk_ids = ", ".join(str(cid) for cid in member.get("source_chunk_ids") or [] if cid) + blocks.append(f"### Member: {name}\nClaims:\n{claims_text}\nSource chunk IDs: {chunk_ids or '(none)'}") + return "\n\n".join(blocks) + + # System prompts for Mode A _WIKI_MODE_A_GENERATE_SYSTEM = """You are a wiki COMPILER. Generate a new wiki page for the given concept using the provided source chunks and extracted claims. @@ -2110,6 +2383,7 @@ Write the ENTIRE page in the SAME LANGUAGE as the source chunks. If the source c Markdown formatting is mandatory: put every heading on its own line and separate every paragraph with a blank line. 5. WIKILINKS: Use ONLY the exact page IDs listed in "Available Pages for [[wikilinks]]" (they already carry the entity/ or concept/ prefix). Insert [[EXACT_PAGE_ID]] on first mention of a related concept/entity. NEVER invent a link target, NEVER drop the prefix, NEVER write English names. 6. DICTIONARY PREVENTION: Do NOT group content by source document. Do NOT create one section per entity. Do NOT write flat bullet lists. +7. MEMBER COVERAGE: If "Required Page Members" lists multiple members, the page MUST contain grounded factual content about EVERY listed member. Do not silently omit or replace any member. If the members are unrelated, keep them in clearly separated subsections while preserving all supported facts. ## SOURCE GROUNDING (COMPILER, not writer) - The "Source Chunks" section contains VERBATIM source text. Stay close to the source wording — reuse the source's own sentences and facts where possible. @@ -2120,9 +2394,14 @@ Write the ENTIRE page in the SAME LANGUAGE as the source chunks. If the source c ## OUTPUT Return ONLY the complete markdown page. First line: SUMMARY: {one-sentence description, 15-40 words} -Second line: TOPIC: {the best short canonical topic for this page} +Second line: TITLE: {a concise title covering all required page members} +Third line: TOPIC: {the best short canonical topic for this page} Then the page content. +TITLE is the human-readable page title, not the page ID. When multiple +members are merged, synthesize a title covering the combined subject. For a +single-member page, keep the supplied title unchanged. + Choose TOPIC by understanding the page subject and evidence. Prefer a fitting item from Candidate Topics. If none fits, create a concise topic in the source language. Do not choose by superficial character or word overlap. @@ -2141,6 +2420,7 @@ Write the ENTIRE page in the SAME LANGUAGE as the source chunks. If the source c 5. For FULL RE-SYNTHESIS: Use all source chunks + all claims to rewrite from scratch. 6. For INCREMENTAL MODIFY: Integrate additions, remove retracted content, keep unchanged content. 7. MARKDOWN FORMATTING: Put every heading on its own line and separate every paragraph with a blank line. Do not return the whole page as one line. +8. MEMBER COVERAGE: If "Required Page Members" lists multiple members, the updated page MUST retain grounded factual content about EVERY listed member. Do not silently omit any member. ## DICTIONARY PREVENTION - Do NOT group content by source document. @@ -2156,9 +2436,14 @@ Write the ENTIRE page in the SAME LANGUAGE as the source chunks. If the source c ## OUTPUT Return ONLY the complete updated markdown page. First line: SUMMARY: {one-sentence description of what changed, 15-40 words} -Second line: TOPIC: {the best short canonical topic for the complete updated page} +Second line: TITLE: {a concise title covering all required page members} +Third line: TOPIC: {the best short canonical topic for the complete updated page} Then the updated page content. +TITLE is the human-readable page title, not the page ID. When multiple +members are merged, rewrite the title to cover the complete updated subject. +For a single-member page, keep the supplied title unchanged. + Choose TOPIC by understanding the complete page subject and evidence. Prefer a fitting item from Candidate Topics; retain Current Topic when it remains the best fit. If neither fits, create a concise topic in the source language. Do @@ -2270,29 +2555,52 @@ async def _wiki_llm_group_entities( embeddings: list, chat_mdl, semaphore: asyncio.Semaphore | None = None, + kb_id: str = "", ) -> list[list[dict]]: """Use embeddings for candidate communities and the LLM for final groups.""" if len(entities) <= 1: + _wiki_log_stats("PLAN", "group_summary", kb_id=kb_id, before=len(entities), after=len(entities), reduction_count=0, merged_group_count=0) return [entities] candidate_count = max(1, int(np.ceil(len(entities) / WIKI_GROUP_LLM_CANDIDATE_SIZE))) candidates = _wiki_cluster_entities(entities, embeddings, target_count=candidate_count) - embedding_by_entity_id = {id(entity): embedding for entity, embedding in zip(entities, embeddings, strict=True)} semaphore = semaphore or asyncio.Semaphore(WIKI_GROUP_LLM_MAX_CONCURRENT) async def _partition(candidate: list[dict]) -> list[list[dict]]: - async with semaphore: - try: - groups = await _wiki_llm_partition_candidate(candidate, chat_mdl) - except Exception: - logging.exception("wiki: LLM page grouping failed") - groups = None + groups = None + for attempt in range(2): + async with semaphore: + try: + groups = await _wiki_llm_partition_candidate(candidate, chat_mdl) + except Exception: + logging.exception("wiki: LLM page grouping failed (attempt %s)", attempt + 1) + groups = None + if groups is not None: + break if groups is not None: + merged_groups = [[str(entity.get("entity_name") or entity.get("term") or "") for entity in group] for group in groups if len(group) > 1] + for members in merged_groups: + _wiki_log_stats("PLAN", "llm_page_group", kb_id=kb_id, member_count=len(members), members=members) + _wiki_log_stats( + "PLAN", "llm_group_candidate", kb_id=kb_id, before=len(candidate), after=len(groups), reduction_count=sum(len(group) - 1 for group in groups), merged_group_count=len(merged_groups) + ) return groups - fallback_vectors = [embedding_by_entity_id[id(entity)] for entity in candidate] - return _wiki_cluster_entities(candidate, fallback_vectors) + # Embeddings only form the retrieval community. They must not decide + # the final page boundary when the LLM is unavailable or invalid. + _wiki_log_stats("PLAN", "llm_group_unresolved", kb_id=kb_id, before=len(candidate), after=len(candidate), retry_count=2) + return [[entity] for entity in candidate] grouped = await asyncio.gather(*(_partition(candidate) for candidate in candidates)) - return [group for candidate_groups in grouped for group in candidate_groups] + groups = [group for candidate_groups in grouped for group in candidate_groups] + _wiki_log_stats( + "PLAN", + "group_summary", + kb_id=kb_id, + before=len(entities), + after=len(groups), + reduction_count=sum(len(group) - 1 for group in groups), + merged_group_count=sum(1 for group in groups if len(group) > 1), + ) + return groups async def _wiki_llm_route_batches( @@ -2304,11 +2612,10 @@ async def _wiki_llm_route_batches( return {} semaphore = asyncio.Semaphore(WIKI_GROUP_LLM_MAX_CONCURRENT) - async def _route_batch(offset: int, batch: list[tuple[dict, list[dict]]]) -> dict[int, str]: + async def _route_batch(batch: list[tuple[int, dict, list[dict]]]) -> dict[int, str]: lines = [] allowed: dict[int, set[str]] = {} - for local_idx, (entity, candidates) in enumerate(batch): - item_id = offset + local_idx + for item_id, entity, candidates in batch: options = [] allowed[item_id] = {"NEW"} for candidate in candidates: @@ -2321,6 +2628,8 @@ async def _wiki_llm_route_batches( "summary": candidate.get("summary", ""), "members": candidate.get("members", []), "similarity": round(candidate.get("score", 0.0), 4), + "signals": candidate.get("signals", []), + "cooccurrence_count": candidate.get("cooccurrence_count", 0), } ) lines.append(json.dumps({"id": item_id, "entity": _wiki_entity_planning_text(entity), "options": options}, ensure_ascii=False)) @@ -2349,9 +2658,95 @@ Items: result[item_id] = page_id return result - batches = [route_items[i : i + WIKI_ROUTE_LLM_BATCH_SIZE] for i in range(0, len(route_items), WIKI_ROUTE_LLM_BATCH_SIZE)] - results = await asyncio.gather(*(_route_batch(i * WIKI_ROUTE_LLM_BATCH_SIZE, batch) for i, batch in enumerate(batches))) - return {item_id: page_id for result in results for item_id, page_id in result.items()} + indexed_items = [(item_id, entity, candidates) for item_id, (entity, candidates) in enumerate(route_items)] + + async def _run(items: list[tuple[int, dict, list[dict]]]) -> dict[int, str]: + batches = [items[i : i + WIKI_ROUTE_LLM_BATCH_SIZE] for i in range(0, len(items), WIKI_ROUTE_LLM_BATCH_SIZE)] + results = await asyncio.gather(*(_route_batch(batch) for batch in batches)) + return {item_id: page_id for result in results for item_id, page_id in result.items()} + + decisions = await _run(indexed_items) + missing = [item for item in indexed_items if item[0] not in decisions] + if missing: + # Retry only missing/invalid items. A malformed item in one batch must + # not make correctly routed entities pay for a full-batch retry. + decisions.update(await _run(missing)) + return decisions + + +def _wiki_route_page_candidate(page_id: str, page: dict, *, score: float = 0.0) -> dict: + title = page.get("title_kwd", "") + if isinstance(title, (list, tuple)): + title = title[0] if title else "" + return { + "score": float(score or 0.0), + "page_id": page_id, + "title": str(title or ""), + "summary": str(page.get("summary_with_weight") or ""), + "members": _as_str_list(page.get("entity_names_kwd"))[:12], + "signals": [], + "cooccurrence_count": 0, + } + + +def _wiki_expand_route_candidates( + entity: dict, + dense_candidates: list[dict], + existing_pages: dict[str, dict], + entity_pages: dict[str, set[str]], + chunk_pages: dict[str, set[str]], + *, + include_candidate_neighbors: bool = False, +) -> list[dict]: + """Merge semantic retrieval with authoritative ownership and graph evidence.""" + candidates = {candidate["page_id"]: dict(candidate) for candidate in dense_candidates if candidate.get("page_id") in existing_pages} + + def _add(page_id: str, signal: str, *, cooccurrence_count: int = 0) -> None: + page = existing_pages.get(page_id) + if not page: + return + candidate = candidates.setdefault(page_id, _wiki_route_page_candidate(page_id, page)) + signals = set(candidate.get("signals") or []) + signals.add(signal) + candidate["signals"] = sorted(signals) + candidate["cooccurrence_count"] = max(int(candidate.get("cooccurrence_count") or 0), cooccurrence_count) + + entity_name = str(entity.get("entity_name") or entity.get("term") or "").strip() + for page_id in entity_pages.get(_normalize_key(entity_name), set()): + _add(page_id, "current_owner") + + for relation in entity.get("relations") or []: + if not isinstance(relation, dict): + continue + counterpart = str(relation.get("entity") or relation.get("counterpart") or "").strip() + for page_id in entity_pages.get(_normalize_key(counterpart), set()): + _add(page_id, "relation") + + cooccurrence: dict[str, int] = {} + for chunk_id in _as_str_list(entity.get("source_chunk_ids")): + for page_id in chunk_pages.get(chunk_id, set()): + cooccurrence[page_id] = cooccurrence.get(page_id, 0) + 1 + for page_id, count in cooccurrence.items(): + _add(page_id, "cooccurrence", cooccurrence_count=count) + + if include_candidate_neighbors: + initial_page_ids = list(candidates) + for page_id in initial_page_ids: + page = existing_pages.get(page_id, {}) + for neighbor_ref in _as_str_list(page.get("outlinks_kwd")) + _as_str_list(page.get("related_kb_pages_kwd")): + if neighbor_ref in existing_pages: + _add(neighbor_ref, "candidate_neighbor") + continue + for neighbor_id in entity_pages.get(_normalize_key(neighbor_ref), set()): + _add(neighbor_id, "candidate_neighbor") + + priority = {"current_owner": 0, "relation": 1, "cooccurrence": 2, "embedding": 3, "candidate_neighbor": 4} + + def _rank(candidate: dict) -> tuple: + signal_rank = min((priority.get(signal, 4) for signal in candidate.get("signals") or []), default=4) + return (signal_rank, -int(candidate.get("cooccurrence_count") or 0), -float(candidate.get("score") or 0.0), candidate["page_id"]) + + return sorted(candidates.values(), key=_rank)[:PAGE_ROUTER_MAX_CANDIDATES] async def _wiki_page_router( @@ -2360,7 +2755,7 @@ async def _wiki_page_router( embd_mdl, tenant_id: str, kb_id: str, - existing_page_ids: set[str] | None = None, + existing_pages: dict[str, dict] | None = None, ) -> dict[str, list[dict]]: """Route entities using KNN candidates followed by an LLM decision. @@ -2368,8 +2763,8 @@ async def _wiki_page_router( - "_new_{page_id}" → new page to create - existing page_id → entities assigned to that page - ``existing_page_ids`` is supplied by Mode B from its already-loaded page - set. An explicitly empty set means this is a first build, so page-index + ``existing_pages`` is supplied by Mode B from its already-loaded page set. + An explicitly empty dict means this is a first build, so page-index candidate retrieval can be skipped and entities can go straight to grouping. """ from common.misc_utils import thread_pool_exec @@ -2388,11 +2783,21 @@ async def _wiki_page_router( orphans: list[dict] = [] embedding_by_entity_id = {id(entity): vec for entity, vec in zip(affected_entities, embeddings, strict=False)} - if existing_page_ids is not None and not existing_page_ids: + existing_pages = existing_pages or {} + entity_pages: dict[str, set[str]] = {} + chunk_pages: dict[str, set[str]] = {} + for page_id, page in existing_pages.items(): + for entity_name in _as_str_list(page.get("entity_names_kwd")): + entity_pages.setdefault(_normalize_key(entity_name), set()).add(page_id) + for chunk_id in _as_str_list(page.get("source_chunk_ids")): + chunk_pages.setdefault(chunk_id, set()).add(page_id) + + if not existing_pages: # On a first build there are no pages that can accept a routed entity. # Querying the page index once per entity only produces orphans, so go # directly to clustering and reuse the embeddings already computed. orphans = list(affected_entities) + _wiki_log_stats("ROUTE", "summary", affected=len(affected_entities), llm_existing=0, llm_new=0, llm_missing=0, new_confirmed_existing=0, final_new=len(orphans)) else: router_sem = asyncio.Semaphore(PAGE_ROUTER_KNN_CONCURRENT) @@ -2426,12 +2831,8 @@ async def _wiki_page_router( if entity.get("action") == "delete": assignments.setdefault("_deleted", []).append(entity) continue - if not field_map: - orphans.append(entity) - continue - candidates = [] - for row in field_map.values(): + for row in (field_map or {}).values(): score = float(row.get("_score", 0.0) or 0.0) page_id = row.get("slug_kwd", "") if isinstance(page_id, (list, tuple)): @@ -2441,16 +2842,10 @@ async def _wiki_page_router( title = row.get("title_kwd", "") if isinstance(title, (list, tuple)): title = title[0] if title else "" - candidates.append( - { - "score": score, - "page_id": page_id, - "title": str(title or ""), - "summary": str(row.get("summary_with_weight") or ""), - "members": _as_str_list(row.get("entity_names_kwd"))[:12], - } - ) - candidates.sort(key=lambda item: (-item["score"], item["page_id"])) + candidate = _wiki_route_page_candidate(page_id, existing_pages.get(page_id, row), score=score) + candidate["signals"] = ["embedding"] + candidates.append(candidate) + candidates = _wiki_expand_route_candidates(entity, candidates, existing_pages, entity_pages, chunk_pages) if not candidates: orphans.append(entity) continue @@ -2461,29 +2856,78 @@ async def _wiki_page_router( except Exception: logging.exception("wiki: LLM page routing failed") decisions = {} + first_new_count = sum(1 for page_id in decisions.values() if page_id == "NEW") + first_existing_count = sum(1 for page_id in decisions.values() if page_id != "NEW") + missing_count = len(route_items) - len(decisions) + second_pass_items: list[tuple[int, dict, list[dict]]] = [] + confirmed_existing_count = 0 for item_id, (entity, candidates) in enumerate(route_items): page_id = decisions.get(item_id) if page_id and page_id != "NEW": assignments.setdefault(page_id, []).append(entity) continue if page_id == "NEW": - orphans.append(entity) + expanded = _wiki_expand_route_candidates( + entity, + candidates, + existing_pages, + entity_pages, + chunk_pages, + include_candidate_neighbors=True, + ) + original_ids = {candidate["page_id"] for candidate in candidates} + added_ids = [candidate["page_id"] for candidate in expanded if candidate["page_id"] not in original_ids] + _wiki_log_stats( + "ROUTE", + "new_confirmation_candidates", + entity=str(entity.get("entity_name") or entity.get("term") or ""), + initial_candidate_count=len(candidates), + added_candidate_count=len(added_ids), + added_to_candidates=bool(added_ids), + added_page_ids=added_ids, + confirmation_candidate_count=len(expanded), + ) + second_pass_items.append((item_id, entity, expanded)) continue - best = candidates[0] - second_score = candidates[1]["score"] if len(candidates) > 1 else 0.0 - if best["score"] >= PAGE_ROUTER_DIRECT_THRESHOLD or (best["score"] >= PAGE_ROUTER_UPDATE_THRESHOLD and best["score"] - second_score >= PAGE_ROUTER_MIN_MARGIN): - assignments.setdefault(best["page_id"], []).append(entity) + owner = next((candidate for candidate in candidates if "current_owner" in candidate.get("signals", [])), None) + if owner: + assignments.setdefault(owner["page_id"], []).append(entity) else: orphans.append(entity) + if second_pass_items: + confirmation_items = [(entity, candidates) for _, entity, candidates in second_pass_items] + confirmations = await _wiki_llm_route_batches(confirmation_items, chat_mdl) + for confirmation_id, (_, entity, candidates) in enumerate(second_pass_items): + page_id = confirmations.get(confirmation_id) + if page_id and page_id != "NEW": + assignments.setdefault(page_id, []).append(entity) + confirmed_existing_count += 1 + continue + owner = next((candidate for candidate in candidates if "current_owner" in candidate.get("signals", [])), None) + if owner and page_id is None: + assignments.setdefault(owner["page_id"], []).append(entity) + else: + orphans.append(entity) + _wiki_log_stats( + "ROUTE", + "summary", + affected=len(affected_entities), + llm_existing=first_existing_count, + llm_new=first_new_count, + llm_missing=missing_count, + new_confirmed_existing=confirmed_existing_count, + final_new=len(orphans), + ) + # Orphans: cluster by similarity, create grouped pages # A deletion that cannot be routed to an existing page must not create a # new page merely so the downstream delete action can remove it again. orphans = [entity for entity in orphans if entity.get("action") != "delete"] if orphans: orphan_embs = [embedding_by_entity_id[id(entity)] for entity in orphans] - clusters = await _wiki_llm_group_entities(orphans, orphan_embs, chat_mdl) - used_page_ids = set(existing_page_ids or ()) | {key[5:] for key in assignments if key.startswith("_new_")} + clusters = await _wiki_llm_group_entities(orphans, orphan_embs, chat_mdl, kb_id=kb_id) + used_page_ids = set(existing_pages) | {key[5:] for key in assignments if key.startswith("_new_")} for cluster in clusters: representative = min( cluster, @@ -2667,6 +3111,7 @@ async def _wiki_finalize( relation_map: dict[str, list[dict]] = {} outlink_map: dict[str, list[str]] = {} # pid → [valid target slugs] dead_links: dict[str, list[str]] = {} # pid → [dead links to remove] + index = search.index_name(tenant_id) # name → page slug map for AUTO-LINKING. Built from every page's plain name # (slug suffix) + title, longest names first so multi-word / multi-char @@ -2694,7 +3139,10 @@ async def _wiki_finalize( # MAP to connect pages. A page-to-page edge is created whenever both # endpoints resolve to compiled wiki pages. This is the primary source of # graph connections — far more reliable than prose [[wikilinks]]. - map_relations = await _load_map_relations(tenant_id, kb_id) + from api.db.services.document_service import DocumentService + + disabled_doc_ids = await thread_pool_exec(DocumentService.get_disabled_doc_ids_by_kb_id, kb_id) + map_relations = await _load_map_relations(tenant_id, kb_id, excluded_doc_ids=disabled_doc_ids) relation_edges: dict[str, set[str]] = {} # pid → {target slug} if map_relations: for rel in map_relations: @@ -2710,31 +3158,39 @@ async def _wiki_finalize( for match in wikilink_re.finditer(content): link = match.group(1).strip() - if link in valid_ids and link != pid: + # Keep an explicit display label when a page contains several + # merged entities. The page slug identifies the destination; + # it must not replace the entity name shown in the prose. + target, separator, display_text = link.partition("|") + target = target.strip() + display_text = display_text.strip() if separator else "" + if target in valid_ids and target != pid: # Valid wikilink → record for cross-reference + outlink relation_map.setdefault(pid, []).append( { - "entity_name": link.split("/")[-1] if "/" in link else link, + "entity_name": display_text or (target.split("/")[-1] if "/" in target else target), "relation": "see_also", } ) - outlink_map.setdefault(pid, []).append(link) - elif link in canonical_names: + outlink_map.setdefault(pid, []).append(target) + elif target in canonical_names: # Entity reference (Mode A): remove [[]] keep plain text - content = content.replace(f"[[{link}]]", link, 1) + replacement = display_text or target + content = content.replace(match.group(0), replacement, 1) else: # Dead link — try WeKnora-style fuzzy resolution to a similar # existing page before giving up. If a close slug exists, retarget # the link (cross-link survives); otherwise degrade to plain text. - resolved = _wiki_resolve_dead_slug(link, valid_ids, name_slug) + resolved = _wiki_resolve_dead_slug(target, valid_ids, name_slug) if resolved: - content = content.replace(f"[[{link}]]", f"[[{resolved}]]", 1) - relation_map.setdefault(pid, []).append({"entity_name": resolved.split("/")[-1] if "/" in resolved else resolved, "relation": "see_also"}) + resolved_link = f"[[{resolved}|{display_text}]]" if display_text else f"[[{resolved}]]" + content = content.replace(match.group(0), resolved_link, 1) + relation_map.setdefault(pid, []).append({"entity_name": display_text or (resolved.split("/")[-1] if "/" in resolved else resolved), "relation": "see_also"}) if resolved not in outlink_map.setdefault(pid, []): outlink_map[pid].append(resolved) else: - content = content.replace(f"[[{link}]]", link, 1) - dead_links.setdefault(pid, []).append(link) + content = content.replace(match.group(0), display_text or target, 1) + dead_links.setdefault(pid, []).append(target) # AUTO-LINK: guarantee cross-page connections even when the LLM omits # [[...]]. Scan for standalone mentions of other pages' plain names and @@ -2742,7 +3198,7 @@ async def _wiki_finalize( # existing_links covers both the raw [[slug]] form and the rendered # `[text](artifact/{kb_id}/slug)` form (so re-runs stay idempotent even # after links were rendered on a previous run). - existing_links = {m.group(1).strip() for m in wikilink_re.finditer(content)} + existing_links = {m.group(1).split("|", 1)[0].strip() for m in wikilink_re.finditer(content)} existing_links |= {m.group(1) for m in re.finditer(rf"\]\(artifact/{re.escape(str(kb_id))}/([^)]+)\)", content)} for name in ordered_names: target = name_slug[name] @@ -2756,7 +3212,11 @@ async def _wiki_finalize( # Skip if the occurrence is already inside a [[...]] link span if _inside_wikilink(content, idx): continue - content = content[:idx] + f"[[{target}]]" + content[idx + len(name) :] + # Preserve the matched prose as the link label. This matters when + # several entities share one page: e.g. the page slug may be + # ``entity/五色棒`` while the text mentions + # ``治世之能臣，乱世之奸雄``. + content = content[:idx] + f"[[{target}|{name}]]" + content[idx + len(name) :] existing_links.add(target) if target not in outlink_map.setdefault(pid, []): outlink_map[pid].append(target) @@ -2826,7 +3286,6 @@ async def _wiki_finalize( update["outlinks_kwd"] = list(outlinks) update["outlinks_int"] = len(outlinks) - index = search.index_name(tenant_id) await thread_pool_exec( settings.docStoreConn.update, {"id": page["id"]}, @@ -2835,6 +3294,15 @@ async def _wiki_finalize( kb_id, ) + # FINALIZE updates every page without forcing a refresh per write. Make the + # complete batch searchable once here, before the caller reloads these rows + # to materialize wiki_entity/wiki_relation. Without this barrier, the page + # API can expose the new outlinks while the graph is built from the previous + # search snapshot (for example, a linked page still gets weight=0/no edges). + refresh_idx = getattr(settings.docStoreConn, "refresh_idx", None) + if callable(refresh_idx): + await thread_pool_exec(refresh_idx, index) + def _wiki_normalize_rows(matrix): """L2-normalize the rows of a 2-D float matrix (safe on zero rows).""" @@ -2967,6 +3435,7 @@ async def wiki_compile_incremental( # extracted from that page's own source documents, rather than from an # unrelated knowledge-base-wide label pool. doc_topics: dict[str, list[str]] = {} + raw_topic_count = 0 raw_relations: list[dict] = [] for _mr in map_results: _doc_id = str(_mr.get("doc_id") or "").strip() @@ -2975,6 +3444,7 @@ async def wiki_compile_incremental( _seen_topics: set[str] = set() for _t in _mr.get("topics") or []: if isinstance(_t, str): + raw_topic_count += 1 _t = _t.strip() _k = _t.casefold() if _t and _k not in _seen_topics: @@ -2993,6 +3463,16 @@ async def wiki_compile_incremental( if isinstance(_from, str) and isinstance(_to, str) and _from and _to: raw_relations.append({"from": _from, "to": _to, "type": _relation.get("type") or "related"}) + unique_topics = sorted({_t for _topics in doc_topics.values() for _t in _topics}, key=lambda value: (value.casefold(), value)) + _wiki_log_stats( + "TOPIC", + "map_summary", + document_count=len(doc_topics), + raw_count=raw_topic_count, + unique_count=len(unique_topics), + topics=unique_topics, + ) + # Release the heavy raw MAP payload as early as possible. All metadata is # now in raw_entities and full claim text in claim_index; keeping # map_results alive would pin the raw MAP structures in memory. @@ -3009,11 +3489,27 @@ async def wiki_compile_incremental( kb_id=kb_id, incremental=incremental, ) + if plan and incremental: + try: + from api.db.services.document_service import DocumentService + + disabled_doc_ids = await thread_pool_exec(DocumentService.get_disabled_doc_ids_by_kb_id, kb_id) + historical_relations = await _load_map_relations(tenant_id, kb_id, excluded_doc_ids=disabled_doc_ids) + raw_relations.extend(historical_relations) + except Exception: + logging.exception("wiki: failed to load historical relations for incremental routing") + + canonical_resolution = dict(name_resolution) + for canonical_name, canonical_entry in canonical_entities.items(): + canonical_resolution.setdefault(canonical_name, canonical_name) + for alias in canonical_entry.get("aliases") or []: + if isinstance(alias, str) and alias: + canonical_resolution.setdefault(alias, canonical_name) entity_relations: dict[str, list[dict]] = {} seen_relations: set[tuple[str, str, str]] = set() for relation in raw_relations: - source = name_resolution.get(relation["from"], relation["from"]) - target = name_resolution.get(relation["to"], relation["to"]) + source = canonical_resolution.get(relation["from"], relation["from"]) + target = canonical_resolution.get(relation["to"], relation["to"]) relation_type = str(relation.get("type") or "related") if not source or not target or source == target: continue @@ -3024,6 +3520,7 @@ async def wiki_compile_incremental( seen_relations.add(key) entity_relations.setdefault(owner, []).append({"entity": counterpart, "type": relation_type}) del raw_relations + del canonical_resolution # raw_entities (lightweight) no longer needed after matching. del raw_entities @@ -3171,11 +3668,15 @@ async def wiki_compile_incremental( "page_version_int", "synthesis_version_int", "entity_names_kwd", + "outlinks_kwd", "related_kb_pages_kwd", "page_type_kwd", "topic_kwd", ], ) + topic_pool = { + _normalize_key(topic): topic for page in existing_pages.values() for topic in _as_str_list(page.get("topic_kwd")) if topic and _normalize_key(topic) != _normalize_key(WIKI_TOPIC_FALLBACK) + } if plan and existing_pages: plan_members = await _wiki_load_plan_group_members(tenant_id, kb_id) for page_id, names in plan_members.items(): @@ -3225,6 +3726,8 @@ async def wiki_compile_incremental( doc_to_entities.setdefault(did, []).append(cname) del canonical_map + topic_embeddings = await _wiki_prepare_topic_embeddings(doc_topics, embd_mdl, list(topic_pool.values())) + topic_pool_lock = asyncio.Lock() if plan: summary = await _wiki_mode_b_run( deltas=deltas, @@ -3238,6 +3741,9 @@ async def wiki_compile_incremental( entity_evidence=entity_evidence, entity_relations=entity_relations, doc_topics=doc_topics, + topic_embeddings=topic_embeddings, + topic_pool=topic_pool, + topic_pool_lock=topic_pool_lock, ) else: # Mode A: every entity AND concept becomes a page (no PLAN grouping). @@ -3254,6 +3760,9 @@ async def wiki_compile_incremental( canonical_claims=canonical_claims, doc_to_entities=doc_to_entities, doc_topics=doc_topics, + topic_embeddings=topic_embeddings, + topic_pool=topic_pool, + topic_pool_lock=topic_pool_lock, ) del deltas del canonical_claims @@ -3285,6 +3794,9 @@ async def _wiki_mode_a_run( canonical_claims: dict[str, list[dict]] | None = None, doc_to_entities: dict[str, list[str]] | None = None, doc_topics: dict[str, list[str]] | None = None, + topic_embeddings: dict[str, object] | None = None, + topic_pool: dict[str, str] | None = None, + topic_pool_lock: asyncio.Lock | None = None, ) -> dict: """Mode A: every grounded entity and concept compiles to its own page. @@ -3408,6 +3920,7 @@ async def _wiki_mode_a_run( all_page_ids = list(existing_pages.keys()) doc_updates: dict[str, list[str]] = {} + topic_selection_stats = {"selected": 0, "new": 0, "new_added": 0} # Do not use a 20-slot semaphore around the whole page worker. The worker # also performs source loading, embedding, and persistence after the LLM # returns; limiting that whole region would artificially starve the LLM @@ -3468,7 +3981,11 @@ async def _wiki_mode_a_run( kb_id=kb_id, page_version=existing.get("page_version_int", 0) if existing else 0, source_doc_ids=sorted(entry["source_doc_ids"]), - topic_candidates=_wiki_topics_for_docs(entry["source_doc_ids"], doc_topics), + topic_candidates=_wiki_topics_for_docs(entry["source_doc_ids"], doc_topics, topic_pool), + topic_selection_stats=topic_selection_stats, + topic_embeddings=topic_embeddings, + topic_pool=topic_pool, + topic_pool_lock=topic_pool_lock, ) if refine_mode == "generate": summary["pages_created"] += 1 @@ -3487,6 +4004,7 @@ async def _wiki_mode_a_run( if tasks: _progress(f"REFINE A: {len(tasks)} pages (LLM pool max {WIKI_REFINE_MAX_CONCURRENT}) ...") await asyncio.gather(*tasks) + _wiki_log_stats("TOPIC", "selection_summary", mode="A", **topic_selection_stats) for did, pids in doc_updates.items(): try: @@ -3583,6 +4101,9 @@ async def _wiki_mode_b_run( entity_evidence: dict[str, dict[str, list[str]]] | None = None, entity_relations: dict[str, list[dict]] | None = None, doc_topics: dict[str, list[str]] | None = None, + topic_embeddings: dict[str, object] | None = None, + topic_pool: dict[str, str] | None = None, + topic_pool_lock: asyncio.Lock | None = None, ) -> dict: """Mode B: Page Router + per-page REFINE.""" @@ -3624,7 +4145,7 @@ async def _wiki_mode_b_run( embd_mdl=embd_mdl, tenant_id=tenant_id, kb_id=kb_id, - existing_page_ids=set(existing_pages), + existing_pages=existing_pages, ) assignments = _wiki_reconcile_page_moves(assignments, existing_pages) @@ -3633,6 +4154,7 @@ async def _wiki_mode_b_run( existing_pages=existing_pages, chat_mdl=chat_mdl, embd_mdl=embd_mdl, + kb_id=kb_id, ) if not assignments: @@ -3664,6 +4186,7 @@ async def _wiki_mode_b_run( doc_updates: dict[str, list[str]] = {} # doc_id → [page_ids] doc_removals: dict[str, list[str]] = {} # doc_id → [page_ids] + topic_selection_stats = {"selected": 0, "new": 0, "new_added": 0} # The shared LLMCallPool limits chat calls. This semaphore must not cap # the complete worker because embedding and page persistence happen after # the chat call and should not consume an LLM concurrency slot. @@ -3700,11 +4223,20 @@ async def _wiki_mode_b_run( additions = [] retractions = [] page_source_doc_ids: set[str] = set() + member_evidence: list[dict] = [] action = "create" if is_new else "update" for ent in entities: - additions.extend(ent.get("claims", [])) + ent_claims = list(ent.get("claims", [])) + additions.extend(ent_claims) retractions.extend(ent.get("retractions", [])) page_source_doc_ids.update(ent.get("source_doc_ids", [])) + member_evidence.append( + { + "name": ent.get("entity_name", ""), + "claims": ent_claims, + "source_chunk_ids": ent.get("source_chunk_ids", []), + } + ) existing_names = _as_str_list(existing.get("entity_names_kwd")) if existing else [] added_names = [ent.get("entity_name", "") for ent in entities if ent.get("action") != "delete" and ent.get("entity_name")] @@ -3718,6 +4250,14 @@ async def _wiki_mode_b_run( evidence = (entity_evidence or {}).get(member_name, {}) page_source_doc_ids.update(evidence.get("source_doc_ids", [])) member_source_chunks.extend({"id": cid, "text": ""} for cid in evidence.get("source_chunk_ids", [])) + if not any(item.get("name") == member_name for item in member_evidence): + member_evidence.append( + { + "name": member_name, + "claims": _wiki_claims_for_entity(existing, member_name) if existing else [], + "source_chunk_ids": evidence.get("source_chunk_ids", []), + } + ) if action == "delete": deleted_page = await _wiki_refine_page( @@ -3753,21 +4293,6 @@ async def _wiki_mode_b_run( ): refine_mode = "re-synthesize" - incoming_vectors = { - ent.get("entity_name"): np.asarray(ent["_embedding"], dtype=np.float32) - for ent in entities - if ent.get("action") != "delete" and ent.get("entity_name") and ent.get("_embedding") is not None - } - missing_names = [name for name in member_names if name not in incoming_vectors] - if missing_names: - missing_vectors, _ = await thread_pool_exec(embd_mdl.encode, missing_names) - incoming_vectors.update(zip(missing_names, np.asarray(missing_vectors, dtype=np.float32), strict=True)) - member_matrix = _wiki_normalize_rows(np.asarray([incoming_vectors[name] for name in member_names], dtype=np.float32)) - page_centroid = np.mean(member_matrix, axis=0) - centroid_norm = np.linalg.norm(page_centroid) - if centroid_norm > 0: - page_centroid = page_centroid / centroid_norm - result = await _wiki_refine_page( mode=refine_mode, page_id=page_key, @@ -3786,9 +4311,14 @@ async def _wiki_mode_b_run( kb_id=kb_id, page_version=existing.get("page_version_int", 0) if existing else 0, entity_names=member_names, - page_embedding=page_centroid, + embed_routing_context=True, source_doc_ids=sorted(page_source_doc_ids), - topic_candidates=_wiki_topics_for_docs(page_source_doc_ids, doc_topics), + topic_candidates=_wiki_topics_for_docs(page_source_doc_ids, doc_topics, topic_pool), + topic_selection_stats=topic_selection_stats, + topic_embeddings=topic_embeddings, + topic_pool=topic_pool, + topic_pool_lock=topic_pool_lock, + member_evidence=member_evidence, ) if result: if is_new: @@ -3824,6 +4354,7 @@ async def _wiki_mode_b_run( if tasks: _progress(f"REFINE B: {len(tasks)} pages (LLM pool max {WIKI_REFINE_MAX_CONCURRENT}) ...") await asyncio.gather(*tasks) + _wiki_log_stats("TOPIC", "selection_summary", mode="B", **topic_selection_stats) # Apply doc_page_source updates serially (no race), preserving metadata for did in set(doc_updates) | set(doc_removals): @@ -3876,6 +4407,7 @@ async def _wiki_split_unstable_page_assignments( existing_pages: dict[str, dict], chat_mdl, embd_mdl, + kb_id: str = "", ) -> dict[str, list[dict]]: """Let the LLM reconsider affected pages whose embedding cohesion degrades.""" if not assignments: @@ -3936,7 +4468,7 @@ async def _wiki_split_unstable_page_assignments( degraded = len(old_member_indices) >= 2 and combined_cohesion < old_cohesion - 0.05 if not over_capacity and not degraded: return None - return await _wiki_llm_group_entities(record["members"], member_matrix, chat_mdl, semaphore=group_semaphore) + return await _wiki_llm_group_entities(record["members"], member_matrix, chat_mdl, semaphore=group_semaphore, kb_id=kb_id) reconsidered = await asyncio.gather(*(_reconsider(record) for record in candidates.values())) for record, clusters in zip(candidates.values(), reconsidered, strict=True): diff --git a/rag/svr/task_executor_refactor/dataset_wiki_generator.py b/rag/svr/task_executor_refactor/dataset_wiki_generator.py index 63abfcdd5b..a9649c2689 100644 --- a/rag/svr/task_executor_refactor/dataset_wiki_generator.py +++ b/rag/svr/task_executor_refactor/dataset_wiki_generator.py @@ -243,6 +243,12 @@ def _wiki_eligible_docs(all_docs, tenant_id: str, skip_doc_ids=None) -> list[tup for d in all_docs or []: if str(d.get("id")) in skip_doc_ids: continue + # Disabled documents remain in the document table and still retain + # their compilation-template configuration, but their source chunks + # have ``available_int=0``. They must not make the KB look buildable: + # after a Wiki clear there is intentionally no MAP input for them. + if str(d.get("status", "1")) != "1": + continue pc = d.get("parser_config") or {} template_ids: list[str] = [] seen_template_ids: set[str] = set() @@ -312,7 +318,34 @@ async def _wiki_existing_map_doc_ids(tenant_id: str, kb_id: str) -> set[str]: return doc_ids -async def _wiki_has_compiled_pages(tenant_id: str, kb_id: str) -> bool: +async def _wiki_delete_map_rows_for_docs( + tenant_id: str, + kb_id: str, + doc_ids: set[str], +) -> None: + """Remove MAP resume rows so the next run must re-extract the documents.""" + if not doc_ids: + return + try: + await thread_pool_exec( + settings.docStoreConn.delete, + { + "compile_kwd": [WIKI_MAP_COMPILE_KWD], + "doc_id": sorted(str(doc_id) for doc_id in doc_ids), + }, + search.index_name(tenant_id), + kb_id, + ) + except Exception: + logging.exception( + "wiki: failed to invalidate MAP resume rows for kb=%s docs=%s", + kb_id, + sorted(doc_ids), + ) + raise + + +async def _wiki_has_compiled_pages(tenant_id: str, kb_id: str) -> bool | None: """True when at least one compiled wiki page already exists for the KB. Used to tell "nothing changed and pages already exist" (a genuine no-op) @@ -341,7 +374,7 @@ async def _wiki_has_compiled_pages(tenant_id: str, kb_id: str) -> bool: return bool(settings.docStoreConn.get_total(res)) except Exception: logging.exception("wiki: page existence probe failed for kb=%s", kb_id) - return False + return None async def _wiki_delete_deleted_doc_state( @@ -1589,6 +1622,7 @@ async def run_wiki_incremental( embedding_model, load_chunks_for_doc: Callable[..., AsyncIterator[list[dict]]], plan: bool = False, + _map_rebuild_retry: bool = False, ) -> None: """Dual-mode wiki compilation with incremental support. @@ -1663,7 +1697,7 @@ async def run_wiki_incremental( eligible = _wiki_eligible_docs(all_docs, ctx.tenant_id, skip_doc_ids=deleted_doc_ids) if not eligible and not is_incremental: - progress(1.0, "No documents configured for wiki compilation.") + progress(1.0, "No enabled documents are configured for wiki compilation.") return # Re-resolve plan (Mode B) from the ELIGIBLE docs' templates. Each eligible @@ -1823,12 +1857,15 @@ async def run_wiki_incremental( if not all_map_results and not deleted_doc_ids: # Nothing fresh, changed, or deleted this run. Skip only when there is - # genuinely nothing to build: no MAP rows at all, or a compiled baseline - # already exists. When MAP rows exist but no pages were ever produced - # (e.g. a prior run persisted MAP then failed before REDUCE, so every - # chunk now looks "unchanged"), fall through — ``map_results=None`` below - # makes wiki_compile_incremental rebuild pages from the stored extracts. - if not existing_map_doc_ids or await _wiki_has_compiled_pages(ctx.tenant_id, ctx.kb_id): + # genuinely nothing to build: an existing MAP baseline already has + # compiled pages. When the Wiki was explicitly cleared, both + # ``existing_map_doc_ids`` and the pages are gone; eligible documents + # must go through the first full MAP/REDUCE/REFINE run again. Likewise, + # when MAP rows exist but no pages were ever produced (for example a + # prior run stopped after MAP), fall through so the compiler can rebuild + # pages from the stored extracts. + has_compiled_pages = await _wiki_has_compiled_pages(ctx.tenant_id, ctx.kb_id) if existing_map_doc_ids else None + if existing_map_doc_ids and has_compiled_pages is True: from rag.advanced_rag.knowlege_compile.wiki_incremental import ( _wiki_finalize, _wiki_load_pages_for_graph, @@ -1855,6 +1892,31 @@ async def run_wiki_incremental( progress(1.0, "Wiki is up to date.") return + if existing_map_doc_ids and has_compiled_pages is False and not _map_rebuild_retry: + # A first build can be interrupted after MAP rows are written. If + # those rows are subsequently removed or become unreadable, the + # resume check still suppresses MAP and the restore phase sees no + # payload. Invalidate only the affected documents and retry once; + # the second invocation then treats them as a clean MAP build. + stale_doc_ids = {str(doc.get("id")) for doc, _ in eligible if doc.get("id")} + logging.warning( + "wiki: MAP resume state is unusable with no compiled pages; forcing one MAP rebuild kb=%s docs=%s", + ctx.kb_id, + sorted(stale_doc_ids), + ) + try: + await _wiki_delete_map_rows_for_docs(ctx.tenant_id, ctx.kb_id, stale_doc_ids) + except Exception: + progress(-1, "Failed to reset stale MAP state for wiki rebuild.") + return + await run_wiki_incremental( + ctx=ctx, + embedding_model=embedding_model, + load_chunks_for_doc=load_chunks_for_doc, + plan=plan, + _map_rebuild_retry=True, + ) + return logging.info("wiki: MAP rows exist but no pages found for kb=%s; rebuilding from stored extracts.", ctx.kb_id) # 5. Run incremental wiki compilation (Mode A or Mode B) diff --git a/rag/svr/task_executor_refactor/task_handler.py b/rag/svr/task_executor_refactor/task_handler.py index 4416eafc6c..88034de97c 100644 --- a/rag/svr/task_executor_refactor/task_handler.py +++ b/rag/svr/task_executor_refactor/task_handler.py @@ -798,6 +798,66 @@ class TaskHandler: logging.exception("load_chunks_for_doc: failed to load chunks for doc=%s", doc_id) return if not field_map: + # Recover rows damaged by the old doc-page-source upsert, which + # updated every row sharing ``doc_id`` and stamped source chunks + # with ``compile_kwd=wiki_doc_page_source``. Genuine tracking + # rows have no chunk body; MAP rows are unavailable. Source + # rows remain available and retain their content, so they can + # be identified without guessing from ids. + try: + recovery_fields = [*select_fields, "available_int"] + recovered_batch: List[Dict] = [] + recovery_offset = 0 + recovery_page_size = 1000 + while True: + recovery_res = await thread_pool_exec( + settings.docStoreConn.search, + recovery_fields, + [], + {"doc_id": [doc_id], "available_int": 1}, + [], + order_by, + recovery_offset, + recovery_page_size, + index_nm, + [kb_id], + ) + recovery_rows = settings.docStoreConn.get_fields(recovery_res, recovery_fields) or {} + for row_id, recovery_row in recovery_rows.items(): + marker = recovery_row.get("compile_kwd") + if isinstance(marker, (list, tuple)): + marker = marker[0] if marker else "" + content = recovery_row.get("content_with_weight") or "" + if marker != "wiki_doc_page_source" or not content: + continue + await thread_pool_exec( + settings.docStoreConn.update, + {"id": row_id}, + {"remove": "compile_kwd"}, + index_nm, + kb_id, + ) + recovered_batch.append( + { + "id": row_id, + "doc_id": recovery_row.get("doc_id") or doc_id, + "content_with_weight": content, + "page_num_int": recovery_row.get("page_num_int", 0), + "top_int": recovery_row.get("top_int", 0), + } + ) + if len(recovery_rows) < recovery_page_size: + break + recovery_offset += recovery_page_size + if recovered_batch: + logging.warning( + "load_chunks_for_doc: recovered %d source chunk(s) mislabeled as wiki_doc_page_source doc=%s", + len(recovered_batch), + doc_id, + ) + yield recovered_batch + except Exception: + logging.exception("load_chunks_for_doc: recovery query failed for doc=%s", doc_id) return batch: List[Dict] = [] diff --git a/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py b/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py index 6f31421b4f..c350893c22 100644 --- a/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py +++ b/test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py @@ -381,3 +381,39 @@ def test_string_list_decodes_legacy_json_and_native_arrays(monkeypatch): assert module._string_list('["doc_1", "doc_2"]') == ["doc_1", "doc_2"] assert module._string_list(["doc_1", "doc_2", "doc_1"]) == ["doc_1", "doc_2"] assert module._string_list("doc_1###doc_2") == ["doc_1", "doc_2"] + + +def test_wiki_alteration_treats_wiki_template_as_eligible(monkeypatch): + module, _, _ = _load_list_datasets_module( + monkeypatch, + kbs=[], + parsing_status_by_kb={}, + ) + _stub( + monkeypatch, + "api.db.services.compilation_template_service", + CompilationTemplateService=SimpleNamespace( + get_saved=lambda template_id, tenant_id: { + "id": template_id, + "kind": "wiki", + "config": {"kind": "wiki"}, + } + ), + ) + _stub(monkeypatch, "rag.svr", __path__=[]) + _stub(monkeypatch, "rag.svr.task_executor_refactor", __path__=[]) + _stub( + monkeypatch, + "rag.svr.task_executor_refactor.chunk_post_processor", + _parser_config_compilation_template_ids=lambda parser_config, tenant_id: parser_config.get("compilation_template_ids", []), + ) + + docs = [ + { + "id": "doc-wiki", + "status": "1", + "parser_config": {"compilation_template_ids": ["template-wiki"]}, + } + ] + + assert module._eligible_doc_ids_for_kind(docs, "tenant-1", "wiki") == {"doc-wiki"} diff --git a/test/unit_test/rag/advanced_rag/knowlege_compile/conftest.py b/test/unit_test/rag/advanced_rag/knowlege_compile/conftest.py index aa9cd2b408..fe05e83f9d 100644 --- a/test/unit_test/rag/advanced_rag/knowlege_compile/conftest.py +++ b/test/unit_test/rag/advanced_rag/knowlege_compile/conftest.py @@ -13,6 +13,16 @@ import sys import types from unittest.mock import MagicMock +import pytest + + +@pytest.fixture(autouse=True) +def _mock_disabled_document_lookup(monkeypatch): + """Keep knowledge-compile unit tests independent of the MySQL database.""" + from api.db.services.document_service import DocumentService + + monkeypatch.setattr(DocumentService, "get_disabled_doc_ids_by_kb_id", MagicMock(return_value=set())) + async def _fake_thread_pool_exec(fn, *args, **kwargs): """Execute the function directly (no actual thread pool).""" diff --git a/test/unit_test/rag/advanced_rag/knowlege_compile/test_wiki_incremental.py b/test/unit_test/rag/advanced_rag/knowlege_compile/test_wiki_incremental.py index b27a4f6e3b..a7474c75c4 100644 --- a/test/unit_test/rag/advanced_rag/knowlege_compile/test_wiki_incremental.py +++ b/test/unit_test/rag/advanced_rag/knowlege_compile/test_wiki_incremental.py @@ -69,7 +69,7 @@ class MockChatModel: self.max_length = 4096 self._canned = canned - async def async_chat(self, system_prompt, messages, **kwargs): + async def async_chat(self, system_prompt, messages, request_conf=None, **kwargs): return self._canned def __enter__(self): @@ -87,6 +87,7 @@ def make_doc_store(search_results: list[dict] | None = None): conn.insert = AsyncMock(return_value=None) conn.update = AsyncMock(return_value=None) conn.delete = AsyncMock(return_value=None) + conn.refresh_idx = MagicMock(return_value=True) def _get_fields(res, fields): hits = res.get("hits", {}).get("hits", []) @@ -831,6 +832,40 @@ async def test_doc_page_source_entity_names(): assert len(dps.get("page_ids", [])) == 2 +@pytest.mark.asyncio +async def test_doc_page_source_update_targets_only_tracking_row(): + """Updating tracking state must never rewrite source chunks sharing doc_id.""" + tracking_id = _wiki._stable_row_id(_wiki.WIKI_DOC_PAGE_SOURCE_COMPILE_KWD, "kb1", "doc_1") + doc_store = make_doc_store( + [ + { + "_source": { + "id": tracking_id, + "doc_id": "doc_1", + "compile_kwd": _wiki.WIKI_DOC_PAGE_SOURCE_COMPILE_KWD, + "page_ids": '["concept/A"]', + "entity_names": '["Apple"]', + "source_chunk_hashes": "{}", + "map_checksum": "old", + } + } + ] + ) + + with patch("common.settings.docStoreConn", doc_store): + await _wiki._wiki_update_doc_page_source( + "t1", + "kb1", + "doc_1", + ["concept/B"], + entity_names=["Apple"], + map_checksum="new", + ) + + condition = doc_store.update.call_args.args[0] + assert condition == {"id": tracking_id} + + # ---- End-to-end: Entity Matching → REDUCE --------------------------------- @@ -1015,6 +1050,10 @@ async def test_finalize_writes_outlinks(): else: pytest.fail("No update for concept/A found") + # Page updates stay cheap (no per-row refresh), then one visibility barrier + # makes the finalized outlinks searchable before the canvas graph reloads. + doc_store.refresh_idx.assert_called_once_with(_wiki.search.index_name("t1")) + async def test_finalize_auto_links_mentions(): """_wiki_finalize auto-links standalone mentions of other pages' names.""" @@ -1059,6 +1098,49 @@ async def test_finalize_auto_links_mentions(): assert google_upd["outlinks_int"] == 1 +@pytest.mark.asyncio +async def test_finalize_preserves_merged_entity_name_as_link_label(): + """A merged page must not replace a member mention with its page title.""" + search_results = [ + { + "_source": { + "slug_kwd": "entity/五色棒", + "title_kwd": "五色棒", + "entity_names_kwd": ["五色棒", "治世之能臣，乱世之奸雄"], + "md_with_weight": "五色棒是曹操早年的刑具。", + "page_type_kwd": "entity", + } + }, + { + "_source": { + "slug_kwd": "entity/许劭", + "title_kwd": "许劭", + "entity_names_kwd": ["许劭"], + "md_with_weight": "许劭评价曹操。", + "page_type_kwd": "entity", + } + }, + { + "_source": { + "slug_kwd": "entity/曹操", + "title_kwd": "曹操", + "entity_names_kwd": ["曹操"], + "md_with_weight": "曹操闻言大喜。曹操提到治世之能臣，乱世之奸雄。", + "page_type_kwd": "entity", + } + }, + ] + doc_store = make_doc_store(search_results) + with ( + patch("common.settings.docStoreConn", doc_store), + patch(f"{_wiki.__name__}._load_canonical_entities", new_callable=AsyncMock, return_value={}), + ): + await _wiki._wiki_finalize(tenant_id="t1", kb_id="kb1", embd_mdl=None) + + update = next(args[0][1] for args in doc_store.update.call_args_list if args[0][0].get("id") == "entity/曹操") + assert "[治世之能臣，乱世之奸雄](artifact/kb1/entity/五色棒)" in update["md_with_weight"] + + def test_inside_wikilink(): content = "before [[entity/X]] after" assert _wiki._inside_wikilink(content, content.index("entity")) @@ -1077,6 +1159,10 @@ def test_extract_outlinks_from_content(): assert _wiki._wiki_extract_outlinks_from_content("no links here") == [] +def test_extract_outlinks_uses_page_id_for_piped_links(): + assert _wiki._wiki_extract_outlinks_from_content("[[entity/五色棒|治世之能臣，乱世之奸雄]]") == ["entity/五色棒"] + + @pytest.mark.asyncio async def test_load_pages_for_graph_outlink_fallback(): """Pages without outlinks_kwd get edges derived from content wikilinks.""" @@ -1346,6 +1432,76 @@ def test_topics_for_docs_only_returns_source_scoped_candidates(): assert _wiki._wiki_topics_for_docs(["doc_1", "doc_2"], doc_topics) == ["董卓被杀", "曹操刺董"] +@pytest.mark.asyncio +async def test_topic_candidates_are_ranked_by_page_embedding(): + class TopicEmbedding: + def encode(self, texts): + vectors = [] + for text in texts: + vectors.append([1.0, 0.0] if "target" in text or "page" in text else [0.0, 1.0]) + return np.asarray(vectors, dtype=np.float32), 0 + + model = TopicEmbedding() + topic_embeddings = await _wiki._wiki_prepare_topic_embeddings({"doc": ["unrelated", "target"]}, model) + ranked = await _wiki._wiki_rank_topic_candidates( + "page", + [{"statement": "page evidence"}], + [], + None, + ["unrelated", "target"], + topic_embeddings, + model, + ) + + assert ranked == ["target", "unrelated"] + + ranked_without_cache = await _wiki._wiki_rank_topic_candidates( + "page", + [{"statement": "page evidence"}], + [], + None, + ["unrelated", "target"], + None, + model, + ) + + assert ranked_without_cache == ["target", "unrelated"] + + +def test_topics_for_docs_includes_global_topic_pool(): + assert _wiki._wiki_topics_for_docs( + ["doc_1"], + {"doc_1": ["MAP topic"]}, + {_wiki._normalize_key("Generated topic"): "Generated topic"}, + ) == ["MAP topic", "Generated topic"] + + +@pytest.mark.asyncio +async def test_plan_does_not_use_embedding_as_final_fallback(): + entities = [{"entity_name": f"entity-{idx}", "claims": []} for idx in range(2)] + vectors = np.eye(2, dtype=np.float32) + + with patch(f"{_wiki.__name__}._wiki_llm_partition_candidate", return_value=None), patch(f"{_wiki.__name__}._wiki_cluster_entities", return_value=[entities]) as cluster: + groups = await _wiki._wiki_llm_group_entities(entities, vectors, MockChatModel(), kb_id="kb1") + + assert groups == [[entities[0]], [entities[1]]] + cluster.assert_called_once_with(entities, vectors, target_count=1) + + +def test_wiki_log_stats_emits_structured_json(): + with patch(f"{_wiki.__name__}.logging.info") as info: + _wiki._wiki_log_stats("ROUTE", "summary", llm_new=2, final_new=1) + + prefix, payload = info.call_args.args + assert prefix == "wiki stats %s" + assert json.loads(payload) == { + "event": "summary", + "final_new": 1, + "llm_new": 2, + "stage": "ROUTE", + } + + @pytest.mark.asyncio async def test_page_router_skips_knn_when_no_existing_pages(): """A first Mode B build should cluster directly without page-index searches.""" @@ -1366,11 +1522,11 @@ async def test_page_router_skips_knn_when_no_existing_pages(): ): assignments = await _wiki._wiki_page_router( affected_entities=entities, - chat_mdl=MockChatModel(), + chat_mdl=MockChatModel(canned="[[0, 1]]"), embd_mdl=embd_mdl, tenant_id="t1", kb_id="kb1", - existing_page_ids=set(), + existing_pages={}, ) assert assignments == {"_new_entity/apple": entities} @@ -1433,6 +1589,208 @@ async def test_llm_router_can_choose_existing_page_or_new(): assert decisions == {0: "entity/a", 1: "NEW"} +@pytest.mark.asyncio +async def test_llm_router_retries_only_missing_decisions(): + route_items = [ + ({"entity_name": "Alpha"}, [{"page_id": "entity/a", "title": "A"}]), + ({"entity_name": "Beta"}, [{"page_id": "entity/b", "title": "B"}]), + ] + responses = [ + '[{"id": 0, "page": "entity/a"}]', + '[{"id": 1, "page": "entity/b"}]', + ] + + with patch(f"{_wiki.__name__}._chat_mdl_ask", new_callable=AsyncMock, side_effect=responses) as ask: + decisions = await _wiki._wiki_llm_route_batches(route_items, MockChatModel()) + + assert decisions == {0: "entity/a", 1: "entity/b"} + assert ask.call_count == 2 + retry_items = ask.call_args_list[1].args[2].split("Items:\n", 1)[1] + assert '"id": 1' in retry_items + assert '"id": 0' not in retry_items + + +def test_route_candidates_include_owner_relations_and_cooccurrence(): + existing_pages = { + "entity/owner": {"title_kwd": "Owner", "entity_names_kwd": ["Alpha"], "source_chunk_ids": []}, + "entity/relation": {"title_kwd": "Relation", "entity_names_kwd": ["Beta"], "source_chunk_ids": []}, + "entity/chunk": {"title_kwd": "Chunk", "entity_names_kwd": ["Gamma"], "source_chunk_ids": ["c1"]}, + } + entity_pages = {"alpha": {"entity/owner"}, "beta": {"entity/relation"}, "gamma": {"entity/chunk"}} + chunk_pages = {"c1": {"entity/chunk"}} + entity = { + "entity_name": "Alpha", + "relations": [{"entity": "Beta", "type": "uses"}], + "source_chunk_ids": ["c1"], + } + + candidates = _wiki._wiki_expand_route_candidates(entity, [], existing_pages, entity_pages, chunk_pages) + + by_page = {candidate["page_id"]: candidate for candidate in candidates} + assert by_page["entity/owner"]["signals"] == ["current_owner"] + assert by_page["entity/relation"]["signals"] == ["relation"] + assert by_page["entity/chunk"]["signals"] == ["cooccurrence"] + assert by_page["entity/chunk"]["cooccurrence_count"] == 1 + + +@pytest.mark.asyncio +async def test_page_router_keeps_current_owner_when_llm_fails(): + entity = {"entity_name": "Alpha", "entity_type": "org", "claims": [], "action": "update"} + existing_pages = { + "entity/alpha": { + "title_kwd": "Alpha", + "entity_names_kwd": ["Alpha"], + "source_chunk_ids": [], + } + } + doc_store = make_doc_store() + doc_store.search = MagicMock(return_value={"hits": {"total": {"value": 0}, "hits": []}}) + + with ( + patch("common.settings.docStoreConn", doc_store), + patch(f"{_wiki.__name__}._chat_mdl_ask", new_callable=AsyncMock, return_value="not json") as ask, + ): + assignments = await _wiki._wiki_page_router( + affected_entities=[entity], + chat_mdl=MockChatModel(), + embd_mdl=MockEmbeddingModel(), + tenant_id="t1", + kb_id="kb1", + existing_pages=existing_pages, + ) + + assert assignments == {"entity/alpha": [entity]} + assert ask.call_count == 2 + + +@pytest.mark.asyncio +async def test_page_router_confirms_new_with_candidate_neighbors(): + entity = {"entity_name": "5G", "entity_type": "technology", "claims": [], "action": "create"} + existing_pages = { + "entity/huawei": { + "title_kwd": "Huawei", + "summary_with_weight": "A technology company", + "entity_names_kwd": ["Huawei"], + "source_chunk_ids": [], + "outlinks_kwd": ["entity/telecom"], + }, + "entity/telecom": { + "title_kwd": "Telecommunications", + "summary_with_weight": "Mobile network technologies", + "entity_names_kwd": ["Telecommunications"], + "source_chunk_ids": [], + }, + } + doc_store = make_doc_store( + [ + { + "_source": { + "slug_kwd": "entity/huawei", + "title_kwd": "Huawei", + "summary_with_weight": "A technology company", + "entity_names_kwd": ["Huawei"], + "_score": 0.7, + } + } + ] + ) + doc_store.search = MagicMock( + return_value={ + "hits": { + "total": {"value": 1}, + "hits": [ + { + "_source": { + "slug_kwd": "entity/huawei", + "title_kwd": "Huawei", + "summary_with_weight": "A technology company", + "entity_names_kwd": ["Huawei"], + "_score": 0.7, + } + } + ], + } + } + ) + responses = ['[{"id": 0, "page": "NEW"}]', '[{"id": 0, "page": "entity/telecom"}]'] + + with ( + patch("common.settings.docStoreConn", doc_store), + patch(f"{_wiki.__name__}._chat_mdl_ask", new_callable=AsyncMock, side_effect=responses), + ): + assignments = await _wiki._wiki_page_router( + affected_entities=[entity], + chat_mdl=MockChatModel(), + embd_mdl=MockEmbeddingModel(), + tenant_id="t1", + kb_id="kb1", + existing_pages=existing_pages, + ) + + assert assignments == {"entity/telecom": [entity]} + + +@pytest.mark.asyncio +async def test_page_router_confirms_new_even_without_extra_neighbors(): + entity = {"entity_name": "Alpha product", "entity_type": "product", "claims": [], "action": "create"} + existing_pages = { + "entity/alpha": { + "title_kwd": "Alpha", + "summary_with_weight": "Alpha products", + "entity_names_kwd": ["Alpha"], + "source_chunk_ids": [], + } + } + doc_store = make_doc_store( + [ + { + "_source": { + "slug_kwd": "entity/alpha", + "title_kwd": "Alpha", + "summary_with_weight": "Alpha products", + "entity_names_kwd": ["Alpha"], + "_score": 0.7, + } + } + ] + ) + doc_store.search = MagicMock( + return_value={ + "hits": { + "total": {"value": 1}, + "hits": [ + { + "_source": { + "slug_kwd": "entity/alpha", + "title_kwd": "Alpha", + "summary_with_weight": "Alpha products", + "entity_names_kwd": ["Alpha"], + "_score": 0.7, + } + } + ], + } + } + ) + responses = ['[{"id": 0, "page": "NEW"}]', '[{"id": 0, "page": "entity/alpha"}]'] + + with ( + patch("common.settings.docStoreConn", doc_store), + patch(f"{_wiki.__name__}._chat_mdl_ask", new_callable=AsyncMock, side_effect=responses) as ask, + ): + assignments = await _wiki._wiki_page_router( + affected_entities=[entity], + chat_mdl=MockChatModel(), + embd_mdl=MockEmbeddingModel(), + tenant_id="t1", + kb_id="kb1", + existing_pages=existing_pages, + ) + + assert assignments == {"entity/alpha": [entity]} + assert ask.call_count == 2 + + def test_spherical_clustering_is_input_order_independent(): entities = [{"entity_name": f"entity-{idx}", "claims": [{"statement": str(idx)}]} for idx in range(12)] vectors = np.asarray(