From 2c59d07bdbc857c6049f8a403b13eefa33a89dfd Mon Sep 17 00:00:00 2001 From: Kevin Hu Date: Wed, 8 Jul 2026 20:08:14 +0800 Subject: [PATCH] Feat: add wiki folder (#16749) ### Summary Add wiki folders. --- api/apps/restful_apis/dataset_api.py | 40 ++- api/apps/services/dataset_api_service.py | 93 ++++- conf/infinity_mapping.json | 1 + rag/advanced_rag/knowlege_compile/runner.py | 330 ++++++++++++++++++ rag/advanced_rag/knowlege_compile/wiki.py | 28 +- rag/flow/compiler/__init__.py | 18 + rag/flow/compiler/compiler.py | 199 +++++++++++ rag/flow/compiler/schema.py | 35 ++ rag/flow/extractor/extractor.py | 29 -- rag/flow/pipeline.py | 3 +- rag/svr/task_executor.py | 9 +- .../chunk_post_processor.py | 246 ++----------- .../dataflow_service.py | 9 +- .../dataset_wiki_generator.py | 117 ++++++- 14 files changed, 893 insertions(+), 264 deletions(-) create mode 100644 rag/advanced_rag/knowlege_compile/runner.py create mode 100644 rag/flow/compiler/__init__.py create mode 100644 rag/flow/compiler/compiler.py create mode 100644 rag/flow/compiler/schema.py diff --git a/api/apps/restful_apis/dataset_api.py b/api/apps/restful_apis/dataset_api.py index 0c4df2590..42b2a55d8 100644 --- a/api/apps/restful_apis/dataset_api.py +++ b/api/apps/restful_apis/dataset_api.py @@ -583,7 +583,7 @@ async def has_any_wiki(tenant_id, dataset_id): async def list_wiki_pages(tenant_id, dataset_id): """List artifact pages for the dataset Artifact tab. - GET /api/v1/datasets//artifacts?page=1&page_size=200&page_type=entity + GET /api/v1/datasets//artifacts?page=1&page_size=200&page_type=entity&topic=topic Success: {"code": 0, "data": {"total": int, "items": [{slug, title, page_type}]}} """ try: @@ -591,7 +591,8 @@ async def list_wiki_pages(tenant_id, dataset_id): page_size = int(request.args.get("page_size", 200) or 200) except (TypeError, ValueError): return get_error_argument_result("page and page_size must be integers") - page_type = request.args.get("page_type") or None + page_type = (request.args.get("page_type") or "").strip() or None + topic = (request.args.get("topic") or "").strip() or None try: success, result = await dataset_api_service.list_wiki_pages( @@ -600,6 +601,37 @@ async def list_wiki_pages(tenant_id, dataset_id): page=page, page_size=page_size, page_type=page_type, + topic=topic, + ) + if success: + return get_result(data=result) + return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR) + except Exception as e: + logging.exception(e) + return get_error_data_result(message="Internal server error") + + +@manager.route("/datasets//artifacts_topics", methods=["GET"]) # noqa: F821 +@login_required +@add_tenant_id_to_kwargs +async def list_wiki_topics(tenant_id, dataset_id): + """List wiki topics for the dataset Artifact tab. + + GET /api/v1/datasets//artifacts_topics?page=1&page_size=200 + Success: {"code": 0, "data": {"total": int, "items": [{topic, title, slug}]}} + """ + try: + page = int(request.args.get("page", 1) or 1) + page_size = int(request.args.get("page_size", 200) or 200) + except (TypeError, ValueError): + return get_error_argument_result("page and page_size must be integers") + + try: + success, result = await dataset_api_service.list_wiki_topics( + dataset_id, + tenant_id, + page=page, + page_size=page_size, ) if success: return get_result(data=result) @@ -647,8 +679,8 @@ async def clear_wiki(tenant_id, dataset_id): """Wipe every artifact-related row from ES for this KB. DELETE /api/v1/datasets//artifacts - Removes the five ``compile_kwd`` row types written by the artifact - pipeline (MAP extracts / REDUCE results / PLAN / page drafts / pages). + Removes the artifact ``compile_kwd`` row types written by the artifact + pipeline (MAP extracts / REDUCE results / PLAN / drafts / pages / topics / graph rows). Success: {"code": 0, "data": {"deleted": {kwd: result}}} """ try: diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py index f972e35c3..5932606c2 100644 --- a/api/apps/services/dataset_api_service.py +++ b/api/apps/services/dataset_api_service.py @@ -1496,11 +1496,12 @@ async def search_datasets(tenant_id: str, req: dict): # with ``compile_kwd="artifact_page"`` written by TaskHandler's # ``_persist_wiki_pages_to_es``. The schema fields they rely on are: # slug_kwd, title_kwd, page_type_kwd, content_with_weight, -# entity_names_kwd, outlinks_kwd, related_kb_pages_kwd, +# topic_kwd, entity_names_kwd, outlinks_kwd, related_kb_pages_kwd, # source_chunk_ids, source_doc_ids # --------------------------------------------------------------------------- _WIKI_COMPILE_KWD = "artifact_page" +_WIKI_TOPIC_COMPILE_KWD = "artifact_page_topic" _SKILL_COMPILE_KWD = "skill" _SKILL_ALL_COMPILE_KWD = "skill_all" @@ -1568,6 +1569,7 @@ async def list_wiki_pages( page: int = 1, page_size: int = 200, page_type: str | None = None, + topic: str | None = None, ): """List artifact pages for the left-hand 2-column list. @@ -1589,10 +1591,14 @@ async def list_wiki_pages( page = max(1, int(page or 1)) page_size = max(1, min(int(page_size or 200), 1000)) offset = (page - 1) * page_size + page_type = page_type.strip() if isinstance(page_type, str) else page_type + topic = topic.strip() if isinstance(topic, str) else topic condition: dict = {"compile_kwd": [_WIKI_COMPILE_KWD]} if page_type: condition["page_type_kwd"] = [page_type] + if topic: + condition["topic_kwd"] = [topic] order_by = OrderByExpr() try: @@ -1609,6 +1615,7 @@ async def list_wiki_pages( "slug_kwd", "title_kwd", "page_type_kwd", + "topic_kwd", "outlinks_int", "summary_with_weight", ] @@ -1640,6 +1647,7 @@ async def list_wiki_pages( "slug": slug, "title": row.get("title_kwd") or slug, "page_type": row.get("page_type_kwd") or "concept", + "topic": row.get("topic_kwd") or "", "summary": row.get("summary_with_weight") or "", } ) @@ -1647,6 +1655,74 @@ async def list_wiki_pages( return True, {"total": int(total or 0), "items": items} +async def list_wiki_topics( + dataset_id: str, + tenant_id: str, + page: int = 1, + page_size: int = 200, +): + """List wiki topics for the dataset Artifact tab.""" + if not KnowledgebaseService.accessible(dataset_id, tenant_id): + return False, "No authorization." + _, kb = KnowledgebaseService.get_by_id(dataset_id) + + pack = _wiki_index_or_none(kb.tenant_id, dataset_id) + if pack is None: + return True, {"total": 0, "items": []} + index_nm, _ = pack + + from common.doc_store.doc_store_base import OrderByExpr + + page = max(1, int(page or 1)) + page_size = max(1, min(int(page_size or 200), 1000)) + offset = (page - 1) * page_size + + order_by = OrderByExpr() + try: + order_by.asc("title_kwd") + except Exception: + order_by = OrderByExpr() + + select_fields = [ + "id", + "topic_kwd", + "title_kwd", + "slug_kwd", + ] + try: + res = settings.docStoreConn.search( + select_fields=select_fields, + highlight_fields=[], + condition={"compile_kwd": [_WIKI_TOPIC_COMPILE_KWD]}, + match_expressions=[], + order_by=order_by, + offset=offset, + limit=page_size, + index_names=index_nm, + knowledgebase_ids=[dataset_id], + ) + field_map = settings.docStoreConn.get_fields(res, select_fields) + except Exception: + logging.exception("list_wiki_topics: docStore search failed for kb=%s", dataset_id) + return True, {"total": 0, "items": []} + + total = settings.docStoreConn.get_total(res) + items = [] + for row in (field_map or {}).values(): + topic = row.get("topic_kwd") + if not isinstance(topic, str) or not topic: + continue + items.append( + { + "topic": topic, + "title": row.get("title_kwd") or topic, + "slug": row.get("slug_kwd") or topic, + } + ) + + return True, {"total": int(total or 0), "items": items} + + async def get_wiki_page( dataset_id: str, tenant_id: str, @@ -1679,6 +1755,7 @@ async def get_wiki_page( "slug_kwd", "title_kwd", "page_type_kwd", + "topic_kwd", "content_with_weight", "summary_with_weight", "entity_names_kwd", @@ -1722,6 +1799,7 @@ async def get_wiki_page( "slug": row.get("slug_kwd") or full_slug, "title": row.get("title_kwd") or full_slug, "page_type": row.get("page_type_kwd") or page_type, + "topic": row.get("topic_kwd") or "", "content_md_rendered": content_md, "summary": summary, "entity_names": row.get("entity_names_kwd") or [], @@ -2031,7 +2109,7 @@ async def update_wiki_page( # :meth:`FileCommitService.get_page_commit_detail`. -# All six row types the artifact pipeline writes. Listed in dependency +# All seven 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. ``artifact_page_graph`` # is the materialized canvas graph derived from the refined pages — @@ -2042,6 +2120,7 @@ _WIKI_COMPILE_KWDS = ( "artifact_compilation_plan", "artifact_page_draft", "artifact_page", + _WIKI_TOPIC_COMPILE_KWD, "artifact_entity", "artifact_relation", ) @@ -2470,11 +2549,11 @@ async def get_wiki_graph( async def clear_wiki(dataset_id: str, tenant_id: str): """Wipe every artifact-related row from ES for this KB. - Touches all five ``compile_kwd`` row types the artifact pipeline writes - (MAP extracts, REDUCE results, PLAN output, page drafts, and the - searchable artifact_page rows). After this completes the next "Artifact" - run starts from a clean slate — no resume cache to short-circuit MAP, no - prior pages to reconcile against in PLAN. + Touches all artifact ``compile_kwd`` row types the artifact pipeline writes + (MAP extracts, REDUCE results, PLAN output, drafts, pages, topics, and graph + rows). After this completes the next "Artifact" run starts from a clean + slate: no resume cache to short-circuit MAP, no prior pages to reconcile + against in PLAN. Returns ``(True, {"deleted": {kwd: count_or_True}})`` on success or ``(False, str)`` on auth failure. diff --git a/conf/infinity_mapping.json b/conf/infinity_mapping.json index dfc07d2e4..bb0401cb6 100644 --- a/conf/infinity_mapping.json +++ b/conf/infinity_mapping.json @@ -58,6 +58,7 @@ "doc_ids_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "slug_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "title_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, + "topic_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "page_type_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "entity_names_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, "outlinks_kwd": {"type": "varchar", "default": "", "analyzer": "whitespace-#"}, diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py new file mode 100644 index 000000000..b1146a72e --- /dev/null +++ b/rag/advanced_rag/knowlege_compile/runner.py @@ -0,0 +1,330 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Handler-free core for document-scoped knowledge (structure) compilation. + +Extracted from +``rag/svr/task_executor_refactor/chunk_post_processor.py`` so the same +template resolution, batching, accumulate / merge-flush and synthesis logic +can be driven from two places: + +* the chunking **task executor**, which streams a document's chunks out of + the doc store (``run_document_structure_compile``), and +* the ``rag.flow`` **Compiler** component, which receives chunks in-memory + from an upstream pipeline node. + +Only the non-``tree`` template kinds are handled here. ``tree`` templates run +RAPTOR over the whole document and are still driven from the task executor +(``run_tree_templates``), which owns the doc-store reload + ``RaptorService``. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import AsyncIterator, Callable + +from api.db.services.compilation_template_service import CompilationTemplateService +from api.db.services.compilation_template_group_service import ( + CompilationTemplateGroupService, +) +from api.db.services.llm_service import LLMBundle +from common.exceptions import TaskCanceledException +from rag.advanced_rag.knowlege_compile.structure import ( + CHAIN_KINDS, + compile_structure_from_text, + merge_compiled_structures, + validate_and_correct_chain, +) + + +# ----- tunables ------------------------------------------------------ +# Bound how many source chunks are handed to a single +# ``compile_structure_from_text`` invocation. The call fans them out +# across max_workers internally, so a moderate window keeps memory + +# LLM-context pressure predictable for long docs. +DOC_STRUCTURE_COMPILE_BATCH_CHUNKS = 4 + +# Bound how many compiled ES-ready docs may accumulate before we flush +# them through ``merge_compiled_structures``. The merger does pairwise +# cosine + LLM duplicate-judging, so it's the more expensive step; we +# cap the per-flush set to keep the local-dedup buckets tractable. +DOC_STRUCTURE_MERGE_MAX_DOCS = 512 + +# Hard wall on the chain-validator LLM correction step. ``list`` and +# ``timeline`` kinds run this just before each merge flush; anything +# longer than this is treated as a blocked LLM and the uncorrected +# docs are flushed instead. +STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S = 120.0 + + +# ----- template resolution ------------------------------------------- + + +def resolve_template_ids_from_groups(group_ids, tenant_id: str) -> list[str]: + """Resolve an ordered, de-duplicated list of compilation-template ids + from a list of template-*group* ids. + + Mirrors ``_parser_config_compilation_template_ids`` but takes the group + ids directly (the ``rag.flow`` Compiler carries them as a component + parameter rather than inside ``parser_config``). + """ + template_ids: list[str] = [] + seen: set[str] = set() + for group_id in group_ids or []: + if not isinstance(group_id, str) or not group_id.strip(): + continue + for template_id in CompilationTemplateGroupService.resolve_template_ids( + group_id.strip(), + tenant_id, + ): + if template_id in seen: + continue + seen.add(template_id) + template_ids.append(template_id) + return template_ids + + +def load_active_templates(template_ids, tenant_id: str) -> list[tuple[str, dict]]: + """Load each template's saved config and keep only the ones that drive a + real, non-``artifacts`` structure compilation. + + Returns ``[(template_id, parser_cfg), ...]`` — templates that are missing, + have an invalid config, or resolve to no/``artifacts`` kind are dropped + (with a warning for the missing/invalid cases). + """ + from api.apps.restful_apis.chunk_api import _compilation_template_kind + + active_templates: list[tuple[str, dict]] = [] + for template_id in template_ids: + template = CompilationTemplateService.get_saved(template_id, tenant_id) + if not template: + logging.warning("document_structure_compile: template %s not found", template_id) + continue + parser_cfg = template.get("config") or {} + if not isinstance(parser_cfg, dict): + logging.warning("document_structure_compile: template %s config is invalid", template_id) + continue + kind = _compilation_template_kind(parser_cfg.get("kind")) + if not kind or kind == "artifacts": + continue + active_templates.append((template_id, parser_cfg)) + return active_templates + + +def split_tree_templates( + active_templates: list[tuple[str, dict]], +) -> tuple[list[tuple[str, dict]], list[tuple[str, dict]]]: + """Partition templates into ``(tree, non_tree)`` by kind.""" + from api.apps.restful_apis.chunk_api import _compilation_template_kind + + tree_templates: list[tuple[str, dict]] = [] + non_tree_templates: list[tuple[str, dict]] = [] + for tid, cfg in active_templates: + if _compilation_template_kind((cfg or {}).get("kind")) == "tree": + tree_templates.append((tid, cfg)) + else: + non_tree_templates.append((tid, cfg)) + return tree_templates, non_tree_templates + + +# ----- non-tree compilation core ------------------------------------- + + +async def run_structure_compile_over_batches( + *, + active_templates: list[tuple[str, dict]], + chat_mdl_by_tid: dict[str, LLMBundle], + embedding_model: LLMBundle, + tenant_id: str, + kb_id: str, + doc_id: str, + language: str, + chunk_batches: AsyncIterator[list[dict]], + progress_cb: Callable[..., None], + cancel_check: Callable[[], bool] = lambda: False, + record: Callable[[str, dict], None] | None = None, +) -> dict[str, dict]: + """Extract + merge structures for every non-``tree`` template over an + async stream of chunk batches, then run the optional synthesis phase. + + ``active_templates`` must already be the non-tree subset with a resolved + chat model in ``chat_mdl_by_tid``. Chunks arrive as an async iterator of + batches so callers can stream them from the doc store or hand over an + in-memory list; each ``dict`` must expose ``id`` and text + (``content_with_weight`` / ``text``). + + Returns ``{template_id: {"inserted", "updated", "duplicates_dropped"}}``. + Raises :class:`TaskCanceledException` when ``cancel_check`` trips. + """ + from api.apps.restful_apis.chunk_api import _compilation_template_kind + + if not active_templates: + return {} + + total = len(active_templates) + + accumulators: dict[str, list[dict]] = {tid: [] for tid, _ in active_templates} + template_kinds: dict[str, str] = {tid: _compilation_template_kind((cfg or {}).get("kind")) for tid, cfg in active_templates} + agg_infos: dict[str, dict] = {tid: {"inserted": 0, "updated": 0, "duplicates_dropped": 0} for tid, _ in active_templates} + chunks_by_id: dict[str, str] = {} + + async def _flush(template_id: str) -> None: + acc = accumulators[template_id] + if not acc: + return + kind = template_kinds.get(template_id, "") + if kind in CHAIN_KINDS: + try: + acc = await asyncio.wait_for( + validate_and_correct_chain( + acc, + chunks_by_id, + chat_mdl_by_tid[template_id], + kind, + callback=progress_cb, + ), + timeout=STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S, + ) + accumulators[template_id] = acc + except asyncio.TimeoutError: + logging.warning( + "chain validate: timed out after %ss for template %s; using uncorrected docs", + STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S, + template_id, + ) + except Exception: + logging.exception( + "chain validate: unexpected failure for template %s; using uncorrected docs", + template_id, + ) + info = await merge_compiled_structures( + acc, + chat_mdl_by_tid[template_id], + embedding_model, + tenant_id, + kb_id, + compilation_template_id=template_id, + cancel_check=cancel_check, + ) + acc.clear() + if isinstance(info, dict): + agg = agg_infos[template_id] + for k in ("inserted", "updated", "duplicates_dropped"): + agg[k] = agg.get(k, 0) + int(info.get(k, 0) or 0) + + progress_cb(msg=f"Start document knowledge compilation ({total} template(s)) ...") + + batch_no = 0 + async for batch in chunk_batches: + batch_no += 1 + for chunk in batch: + cid = chunk.get("id") + if isinstance(cid, str) and cid not in chunks_by_id: + text = chunk.get("content_with_weight") or chunk.get("text") or "" + chunks_by_id[cid] = text if isinstance(text, str) else "" + for idx, (template_id, parser_cfg) in enumerate(active_templates): + progress_cb(msg=f" compile batch {batch_no} ({len(batch)} chunks) for template ({idx + 1}/{total})") + docs = await compile_structure_from_text( + batch, + parser_cfg, + chat_mdl_by_tid[template_id], + embedding_model, + doc_id, + language=language, + callback=progress_cb, + compilation_template_id=template_id, + ) + if docs: + accumulators[template_id].extend(docs) + if len(accumulators[template_id]) >= DOC_STRUCTURE_MERGE_MAX_DOCS: + progress_cb(msg=f" merge flush ({len(accumulators[template_id])} docs) for template ({idx + 1}/{total})") + await _flush(template_id) + + for idx, (template_id, parser_cfg) in enumerate(active_templates): + if cancel_check(): + raise TaskCanceledException("Task was cancelled during document knowledge compilation") + await _flush(template_id) + agg = agg_infos[template_id] + if record: + record(f"document_structure_compile:{template_id}", agg) + progress_cb(msg=f"Document knowledge compilation done ({idx + 1}/{total}): {agg}") + + # ── Synthesis phase ────────────────────────────────────────────── + # If the template has synthesis.enabled, run wiki PLAN+REFINE + # to generate output (wiki page, essence paragraph, etc.). + synthesis_cfg = (parser_cfg or {}).get("synthesis") or {} + if synthesis_cfg.get("enabled"): + example = synthesis_cfg.get("example") + compile_kwd = synthesis_cfg.get("compile_kwd", "artifact_page") + plan_cfg = synthesis_cfg.get("plan") or {} + + # Reserved for future wiki_plan_from_reduction extension: + # entity_type_filter, mention_count_threshold, top_n + if plan_cfg: + logging.debug( + "synthesis: template %s plan config %r reserved for future use", + template_id, plan_cfg, + ) + + if cancel_check(): + raise TaskCanceledException("Task was cancelled before synthesis PLAN") + + if not example: + logging.warning( + "synthesis: template %s has synthesis.enabled but no example; skipping", + template_id, + ) + else: + try: + from rag.advanced_rag.knowlege_compile.wiki import ( + wiki_plan_from_reduction, + wiki_refine_from_plan, + ) + + progress_cb(msg=f"Synthesis PLAN for template {template_id} (kind={compile_kwd}) ...") + plan = await wiki_plan_from_reduction( + chat_mdl=chat_mdl_by_tid[template_id], + embd_mdl=embedding_model, + tenant_id=tenant_id, + kb_id=kb_id, + callback=progress_cb, + ) + if cancel_check(): + raise TaskCanceledException("Task was cancelled after synthesis PLAN") + + if not plan or not plan.get("pages"): + progress_cb(msg=f"Synthesis: no pages planned for template {template_id}.") + else: + progress_cb(msg=f"Synthesis REFINE for template {template_id} ({len(plan['pages'])} page(s)) ...") + pages = await wiki_refine_from_plan( + chat_mdl=chat_mdl_by_tid[template_id], + embd_mdl=embedding_model, + tenant_id=tenant_id, + kb_id=kb_id, + callback=progress_cb, + example=example, + ) + # Overwrite compile_kwd on every output page so the + # synthesis type is tracked correctly in ES. + for p in pages or []: + p["compile_kwd"] = compile_kwd + progress_cb(msg=f"Synthesis done: {len(pages or [])} {compile_kwd} page(s) written.") + except TaskCanceledException: + raise + except Exception: + logging.exception("synthesis: failed for template %s", template_id) + + return agg_infos diff --git a/rag/advanced_rag/knowlege_compile/wiki.py b/rag/advanced_rag/knowlege_compile/wiki.py index 4ce272235..8522629e0 100644 --- a/rag/advanced_rag/knowlege_compile/wiki.py +++ b/rag/advanced_rag/knowlege_compile/wiki.py @@ -1630,6 +1630,9 @@ Description: {kb_description} ## Extracted concepts (with mention counts) {concepts_summary} +## Extracted topics +{topics_summary} + ## KB reconciliation results {kb_reconciliation} @@ -1642,6 +1645,7 @@ Produce a JSON compilation plan: "slug": "concept/example-name", "title": "Example Page Title", "page_type": "entity | concept | topic", + "topic": "short canonical topic name", "entity_names": ["entity or concept name covered by this page"], "related_kb_pages": ["existing-slug-1"], "priority": 1 @@ -1656,6 +1660,10 @@ Rules: - For UPDATE, slug MUST be an existing wiki page slug from the KB reconciliation list above. - page_type is one of: entity | concept | topic. Do NOT use "source". +- topic is required for every page. Prefer a topic from the extracted + topics implied by the entities/concepts. If none fits, create a short + canonical topic name in the user's language. For topic pages, topic should + usually match the page title. # Slug format (CRITICAL — every slug must follow this shape exactly) - The slug is ``/``. The separator @@ -1958,6 +1966,7 @@ async def _wiki_resolve_maybe_items( async def _wiki_planning_call( canonical_entities: list[dict], canonical_concepts: list[dict], + raw_topics: list, reconciliation: dict[str, dict], chat_mdl, kb_name: str | None, @@ -1981,6 +1990,9 @@ async def _wiki_planning_call( entities_summary = "\n".join(_wiki_format_entity_for_plan(e, reconciliation) for e in sorted_entities[:200]) or " (none)" concepts_summary = "\n".join(_wiki_format_concept_for_plan(c, reconciliation) for c in sorted_concepts[:200]) or " (none)" + topics_summary = "\n".join( + f" - {t.strip()}" for t in raw_topics[:200] if isinstance(t, str) and t.strip() + ) or " (none)" kb_lines: list[str] = [] for name, rec in reconciliation.items(): @@ -1993,6 +2005,7 @@ async def _wiki_planning_call( kb_description=kb_description or "(no description)", entities_summary=entities_summary, concepts_summary=concepts_summary, + topics_summary=topics_summary, kb_reconciliation=kb_reconciliation, target_page_count=target_page_count, ) @@ -2013,6 +2026,13 @@ async def _wiki_planning_call( return {"pages": [], "estimated_page_count": 0, "compilation_notes": "planner returned non-object"} if "pages" not in res or not isinstance(res.get("pages"), list): res["pages"] = [] + for page in res["pages"]: + if not isinstance(page, dict): + continue + topic = page.get("topic") + if not isinstance(topic, str) or not topic.strip(): + fallback = page.get("title") or page.get("slug") or page.get("page_type") or "General" + page["topic"] = str(fallback).strip() or "General" if "estimated_page_count" not in res: res["estimated_page_count"] = len(res["pages"]) res.setdefault("compilation_notes", "") @@ -2334,6 +2354,7 @@ async def wiki_plan_from_reduction( plan = await _wiki_planning_call( canonical_entities=canonical_entities, canonical_concepts=canonical_concepts, + raw_topics=raw_topics, reconciliation=reconciliation, chat_mdl=chat_mdl, kb_name=kb_name, @@ -3277,7 +3298,7 @@ async def wiki_refine_from_plan( callback: optional ``(progress: float, msg: str)`` callback. Returns the list of page dicts (one per planned entry). Each page dict - has ``slug, title, page_type, action, content_md, summary, + has ``slug, title, page_type, topic, action, content_md, summary, entity_names, related_kb_pages, source_chunk_ids``. """ # Defensive: some callers accidentally pass the result of @@ -3466,10 +3487,15 @@ async def wiki_refine_from_plan( source_doc_ids = await _wiki_collect_doc_ids(source_chunk_ids, tenant_id, kb_id) summary = _wiki_extract_summary(content_md_rendered) or title + topic = plan_item.get("topic") + if not isinstance(topic, str) or not topic.strip(): + topic = title or slug + page = { "slug": slug, "title": title, "page_type": page_type, + "topic": topic.strip(), "action": action, # Rendered content (with clickable artifact/{kb_id}/{slug} links) is # what callers and the UI consume; the raw [[slug]] form is diff --git a/rag/flow/compiler/__init__.py b/rag/flow/compiler/__init__.py new file mode 100644 index 000000000..c211f972f --- /dev/null +++ b/rag/flow/compiler/__init__.py @@ -0,0 +1,18 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from rag.flow.compiler.compiler import Compiler, CompilerParam + +__all__ = ["Compiler", "CompilerParam"] diff --git a/rag/flow/compiler/compiler.py b/rag/flow/compiler/compiler.py new file mode 100644 index 000000000..d8f427f45 --- /dev/null +++ b/rag/flow/compiler/compiler.py @@ -0,0 +1,199 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import logging +import random +from copy import deepcopy + +import xxhash + +from agent.component.llm import LLMParam, LLM +from api.db.joint_services.tenant_model_service import get_model_config_from_provider_instance +from api.db.services.document_service import DocumentService +from api.db.services.llm_service import LLMBundle +from api.db.services.task_service import has_canceled +from common.constants import LLMType +from rag.advanced_rag.knowlege_compile.runner import ( + DOC_STRUCTURE_COMPILE_BATCH_CHUNKS, + load_active_templates, + resolve_template_ids_from_groups, + run_structure_compile_over_batches, + split_tree_templates, +) +from rag.flow.base import ProcessBase, ProcessParamBase + + +class CompilerParam(ProcessParamBase, LLMParam): + """Parameters for the knowledge-Compiler flow component. + + Same LLM-backed shape as the Extractor, but instead of a single inline + ``knowledge_compilation`` config it drives compilation from one or more + saved **compilation-template groups** (``compilation_template_group_ids``). + Each group resolves to a set of templates, each of which carries its own + structure-compilation config (kind, fields, synthesis, ...). + """ + + def __init__(self): + super().__init__() + self.compilation_template_group_ids = [] + + def check(self): + super().check() + self.check_empty(self.compilation_template_group_ids, "Compilation Template Groups") + + +class Compiler(ProcessBase, LLM): + component_name = "Compiler" + + def _compile_progress(self, prog=None, msg=""): + """Adapt the knowledge-compile ``callback`` protocol to the flow + callback. Downstream compile helpers invoke the callback either as + ``callback(prog, msg)`` (positional) or ``callback(msg=...)``; the + flow's ``self.callback`` expects ``(progress, message)``. + """ + self.callback(prog, msg) + + def _compile_language(self, kwargs: dict) -> str: + language = kwargs.get("language") or getattr(self._canvas, "_language", None) + if isinstance(language, str): + language = language.strip() + if not language and getattr(self._canvas, "_doc_id", None): + config = DocumentService.get_chunking_config(self._canvas._doc_id) or {} + language = config.get("language") + if isinstance(language, str): + language = language.strip() + return language or "English" + + async def _invoke(self, **kwargs): + self.set_output("output_format", "chunks") + self.callback(random.randint(1, 5) / 100.0, "Start knowledge compilation.") + + # Collect the upstream chunk list (same contract as the Extractor). + inputs = self.get_input_elements() + chunks = [] + for _, v in inputs.items(): + val = v["value"] + if isinstance(val, list): + chunks = deepcopy(val) + + tenant_id = self._canvas.get_tenant_id() + doc_id = self._canvas._doc_id + kb_id = getattr(self._canvas, "_kb_id", None) or DocumentService.get_knowledgebase_id(doc_id) + language = self._compile_language(kwargs) + + if not chunks: + self.set_output("chunks", chunks) + return + + for ck in chunks: + ck["doc_id"] = doc_id + ck["id"] = xxhash.xxh64((ck["text"] + str(ck["doc_id"])).encode("utf-8")).hexdigest() + + # Resolve the configured template groups to concrete, active + # (non-artifact) structure-compilation templates. + template_ids = resolve_template_ids_from_groups(self._param.compilation_template_group_ids, tenant_id) + active_templates = load_active_templates(template_ids, tenant_id) + if not active_templates: + self.callback(msg="No active compilation templates resolved from the configured groups.") + self.set_output("chunks", chunks) + return + + # Per-template chat model: a template may pin its own ``llm_id``; + # otherwise fall back to this component's configured chat model. + llm_bundle_cache: dict[str, LLMBundle] = {} + chat_mdl_by_tid: dict[str, LLMBundle] = {} + filtered_templates: list[tuple[str, dict]] = [] + default_chat_mdl = None + for template_id, parser_cfg in active_templates: + tpl_llm_id = parser_cfg.get("llm_id") if isinstance(parser_cfg, dict) else None + if isinstance(tpl_llm_id, str) and tpl_llm_id.strip(): + chat_llm_id = tpl_llm_id.strip() + if chat_llm_id not in llm_bundle_cache: + try: + cfg = get_model_config_from_provider_instance(tenant_id, LLMType.CHAT, chat_llm_id) + llm_bundle_cache[chat_llm_id] = LLMBundle( + tenant_id, + cfg, + lang=language, + max_retries=self._param.max_retries, + retry_interval=self._param.delay_after_error, + ) + except Exception: + logging.exception( + "Compiler: cannot resolve chat model %s for template %s; skipping", + chat_llm_id, + template_id, + ) + continue + chat_mdl_by_tid[template_id] = llm_bundle_cache[chat_llm_id] + else: + if default_chat_mdl is None: + default_chat_mdl = LLMBundle( + tenant_id, + self.chat_mdl.model_config, + lang=language, + max_retries=self._param.max_retries, + retry_interval=self._param.delay_after_error, + ) + chat_mdl_by_tid[template_id] = default_chat_mdl + filtered_templates.append((template_id, parser_cfg)) + + if not filtered_templates: + self.set_output("chunks", chunks) + return + active_templates = filtered_templates + + embedding_model = LLMBundle( + tenant_id, + LLMType.EMBEDDING, + lang=language, + max_retries=self._param.max_retries, + retry_interval=self._param.delay_after_error, + ) + + tree_templates, non_tree_templates = split_tree_templates(active_templates) + if tree_templates: + # ``tree`` templates run RAPTOR over the whole document by + # reloading vectors from the doc store; that path is owned by the + # chunking task executor and isn't available from the flow. + logging.warning( + "Compiler: %d tree-kind template(s) are not supported in the flow pipeline; skipping", + len(tree_templates), + ) + self.callback(msg=f"Skipping {len(tree_templates)} tree-kind template(s) (unsupported in flow).") + + if non_tree_templates: + task_id = getattr(self._canvas, "task_id", None) + + def _cancelled() -> bool: + return bool(task_id) and has_canceled(task_id) + + async def _chunk_batches(): + for i in range(0, len(chunks), DOC_STRUCTURE_COMPILE_BATCH_CHUNKS): + yield chunks[i:i + DOC_STRUCTURE_COMPILE_BATCH_CHUNKS] + + await run_structure_compile_over_batches( + active_templates=non_tree_templates, + chat_mdl_by_tid=chat_mdl_by_tid, + embedding_model=embedding_model, + tenant_id=tenant_id, + kb_id=kb_id, + doc_id=doc_id, + language=language, + chunk_batches=_chunk_batches(), + progress_cb=self._compile_progress, + cancel_check=_cancelled, + ) + + self.set_output("chunks", chunks) diff --git a/rag/flow/compiler/schema.py b/rag/flow/compiler/schema.py new file mode 100644 index 000000000..484e706b8 --- /dev/null +++ b/rag/flow/compiler/schema.py @@ -0,0 +1,35 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class CompilerFromUpstream(BaseModel): + created_time: float | None = Field(default=None, alias="_created_time") + elapsed_time: float | None = Field(default=None, alias="_elapsed_time") + + name: str + file: dict | None = Field(default=None) + chunks: list[dict[str, Any]] | None = Field(default=None) + + output_format: Literal["json", "markdown", "text", "html", "chunks"] | None = Field(default=None) + + json_result: list[dict[str, Any]] | None = Field(default=None, alias="json") + markdown_result: str | None = Field(default=None, alias="markdown") + text_result: str | None = Field(default=None, alias="text") + html_result: str | None = Field(default=None, alias="html") + + model_config = ConfigDict(populate_by_name=True, extra="forbid") diff --git a/rag/flow/extractor/extractor.py b/rag/flow/extractor/extractor.py index 1e216e680..df50d8429 100644 --- a/rag/flow/extractor/extractor.py +++ b/rag/flow/extractor/extractor.py @@ -17,16 +17,9 @@ import logging import random from copy import deepcopy -from api.db.services.document_service import DocumentService -from api.db.services.llm_service import LLMBundle -from common.constants import LLMType import xxhash from agent.component.llm import LLMParam, LLM -from rag.advanced_rag.knowlege_compile.structure import ( - compile_structure_from_text, - merge_compiled_structures, -) from rag.flow.base import ProcessBase, ProcessParamBase from rag.prompts.generator import run_toc_from_text @@ -35,7 +28,6 @@ class ExtractorParam(ProcessParamBase, LLMParam): def __init__(self): super().__init__() self.field_name = "" - self.knowledge_compilation = {} def check(self): super().check() @@ -82,20 +74,6 @@ class Extractor(ProcessBase, LLM): return d return None - async def _knowledge_compile(self, docs): - embedding_model = LLMBundle(self._canvas.get_tenant_id(), LLMType.EMBEDDING, max_retries=self._param.max_retries, retry_interval=self._param.delay_after_error) - self.callback(0.2, message="Start to generate table of content ...") - docs = sorted( - docs, - key=lambda d: ( - d.get("page_num_int", 0)[0] if isinstance(d.get("page_num_int", 0), list) else d.get("page_num_int", 0), - d.get("top_int", 0)[0] if isinstance(d.get("top_int", 0), list) else d.get("top_int", 0), - ), - ) - docs = await compile_structure_from_text(docs, self._param.knowledge_compilation, self.chat_mdl, embedding_model, self._canvas._doc_id) - info = await merge_compiled_structures(docs, self.chat_mdl, embedding_model, self._canvas.get_tenant_id(), DocumentService.get_knowledgebase_id(self._canvas._doc_id)) - return info - async def _invoke(self, **kwargs): self.set_output("output_format", "chunks") self.callback(random.randint(1, 5) / 100.0, "Start to generate.") @@ -118,13 +96,6 @@ class Extractor(ProcessBase, LLM): chunks.append(toc) self.set_output("chunks", chunks) return - if self._param.field_name in ["set", "list", "graph"]: - for ck in chunks: - ck["doc_id"] = self._canvas._doc_id - ck["id"] = xxhash.xxh64((ck["text"] + str(ck["doc_id"])).encode("utf-8")).hexdigest() - await self._knowledge_compile(chunks) - self.set_output("chunks", chunks) - return prog = 0 for i, ck in enumerate(chunks): diff --git a/rag/flow/pipeline.py b/rag/flow/pipeline.py index 75951e4be..d5efa7b42 100644 --- a/rag/flow/pipeline.py +++ b/rag/flow/pipeline.py @@ -26,7 +26,7 @@ from rag.utils.redis_conn import REDIS_CONN class Pipeline(Graph): - def __init__(self, dsl: str | dict, tenant_id=None, doc_id=None, task_id=None, flow_id=None): + def __init__(self, dsl: str | dict, tenant_id=None, doc_id=None, task_id=None, flow_id=None, language=None): if isinstance(dsl, dict): dsl = json.dumps(dsl, ensure_ascii=False) super().__init__(dsl, tenant_id, task_id) @@ -34,6 +34,7 @@ class Pipeline(Graph): doc_id = None self._doc_id = doc_id self._flow_id = flow_id + self._language = language self._kb_id = None if self._doc_id: self._kb_id = DocumentService.get_knowledgebase_id(doc_id) diff --git a/rag/svr/task_executor.py b/rag/svr/task_executor.py index 2f367754e..5dd2ab98a 100644 --- a/rag/svr/task_executor.py +++ b/rag/svr/task_executor.py @@ -760,7 +760,14 @@ async def run_dataflow(task: dict): assert e, "Pipeline log not found." dsl = pipeline_log.dsl dataflow_id = pipeline_log.pipeline_id - pipeline = Pipeline(dsl, tenant_id=task["tenant_id"], doc_id=doc_id, task_id=task_id, flow_id=dataflow_id) + pipeline = Pipeline( + dsl, + tenant_id=task["tenant_id"], + doc_id=doc_id, + task_id=task_id, + flow_id=dataflow_id, + language=task.get("language"), + ) rag_tokenizer.tokenizer.set_language(task.get("language", "English")) chunks = await pipeline.run(file=task["file"]) if task.get("file") else await pipeline.run() if doc_id == CANVAS_DEBUG_DOC_ID: diff --git a/rag/svr/task_executor_refactor/chunk_post_processor.py b/rag/svr/task_executor_refactor/chunk_post_processor.py index fdd6a1623..c52e246c1 100644 --- a/rag/svr/task_executor_refactor/chunk_post_processor.py +++ b/rag/svr/task_executor_refactor/chunk_post_processor.py @@ -351,14 +351,10 @@ def count_with_key(docs: List[Dict], key: str) -> int: import numpy as np # noqa: E402 from typing import Callable, Optional # noqa: E402 -from common.exceptions import TaskCanceledException # noqa: E402 from common.misc_utils import thread_pool_exec # noqa: E402 from common.token_utils import num_tokens_from_string # noqa: E402 from rag.nlp import search # noqa: E402 from api.db.services.document_service import DocumentService # noqa: E402 -from api.db.services.compilation_template_service import ( # noqa: E402 - CompilationTemplateService, -) from api.db.services.compilation_template_group_service import ( # noqa: E402 CompilationTemplateGroupService, ) @@ -368,32 +364,20 @@ from api.db.services.task_service import ( # noqa: E402 credit_doc_chunking_task, is_doc_chunking_aborted, ) -from rag.advanced_rag.knowlege_compile.structure import ( # noqa: E402 - CHAIN_KINDS, - compile_structure_from_text, - merge_compiled_structures, - validate_and_correct_chain, -) # ----- tunables ------------------------------------------------------ -# Bound how many source chunks are handed to a single -# ``compile_structure_from_text`` invocation. The call fans them out -# across max_workers internally, so a moderate window keeps memory + -# LLM-context pressure predictable for long docs. -DOC_STRUCTURE_COMPILE_BATCH_CHUNKS = 4 - -# Bound how many compiled ES-ready docs may accumulate before we flush -# them through ``merge_compiled_structures``. The merger does pairwise -# cosine + LLM duplicate-judging, so it's the more expensive step; we -# cap the per-flush set to keep the local-dedup buckets tractable. -DOC_STRUCTURE_MERGE_MAX_DOCS = 512 - -# Hard wall on the chain-validator LLM correction step. ``list`` and -# ``timeline`` kinds run this just before each merge flush; anything -# longer than this is treated as a blocked LLM and the uncorrected -# docs are flushed instead. -STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S = 120.0 +# The structure-compile batching / merge-flush / chain-correction tunables +# and the non-tree compilation core moved to +# ``rag.advanced_rag.knowlege_compile.runner`` so the ``rag.flow`` Compiler +# component can share them. Re-exported here for backwards compatibility. +from rag.advanced_rag.knowlege_compile.runner import ( # noqa: E402 + DOC_STRUCTURE_COMPILE_BATCH_CHUNKS, + DOC_STRUCTURE_MERGE_MAX_DOCS, # noqa: F401 + STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S, # noqa: F401 + load_active_templates, + run_structure_compile_over_batches, +) # ----- parser_config helpers ----------------------------------------- @@ -945,27 +929,7 @@ async def run_document_structure_compile(handler, embedding_model: LLMBundle) -> if not template_ids: return - active_templates: list[tuple[str, dict]] = [] - for template_id in template_ids: - template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id) - if not template: - logging.warning( - "document_structure_compile: template %s not found", - template_id, - ) - continue - parser_cfg = template.get("config") or {} - if not isinstance(parser_cfg, dict): - logging.warning( - "document_structure_compile: template %s config is invalid", - template_id, - ) - continue - kind = _compilation_template_kind(parser_cfg.get("kind")) - if not kind or kind == "artifacts": - continue - active_templates.append((template_id, parser_cfg)) - + active_templates = load_active_templates(template_ids, ctx.tenant_id) if not active_templates: return @@ -1018,177 +982,29 @@ async def run_document_structure_compile(handler, embedding_model: LLMBundle) -> if not non_tree_templates: return - active_templates = non_tree_templates - progress_cb = ctx.progress_cb - total = len(active_templates) - - accumulators: dict[str, list[dict]] = {tid: [] for tid, _ in active_templates} - template_kinds: dict[str, str] = {tid: _compilation_template_kind((cfg or {}).get("kind")) for tid, cfg in active_templates} - agg_infos: dict[str, dict] = {tid: {"inserted": 0, "updated": 0, "duplicates_dropped": 0} for tid, _ in active_templates} - chunks_by_id: dict[str, str] = {} - - async def _flush(template_id: str) -> None: - acc = accumulators[template_id] - if not acc: - return - kind = template_kinds.get(template_id, "") - if kind in CHAIN_KINDS: - try: - acc = await asyncio.wait_for( - validate_and_correct_chain( - acc, - chunks_by_id, - chat_mdl_by_tid[template_id], - kind, - callback=progress_cb, - ), - timeout=STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S, - ) - accumulators[template_id] = acc - except asyncio.TimeoutError: - logging.warning( - "chain validate: timed out after %ss for template %s; using uncorrected docs", - STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S, - template_id, - ) - except Exception: - logging.exception( - "chain validate: unexpected failure for template %s; using uncorrected docs", - template_id, - ) - info = await merge_compiled_structures( - acc, - chat_mdl_by_tid[template_id], - embedding_model, + async def _stream_doc_batches(): + async for batch in handler._load_chunks_for_doc( ctx.tenant_id, ctx.kb_id, - compilation_template_id=template_id, - cancel_check=lambda: ctx.has_canceled_func(ctx.id), - ) - acc.clear() - if isinstance(info, dict): - agg = agg_infos[template_id] - for k in ("inserted", "updated", "duplicates_dropped"): - agg[k] = agg.get(k, 0) + int(info.get(k, 0) or 0) + ctx.doc_id, + batch_size=DOC_STRUCTURE_COMPILE_BATCH_CHUNKS, + ): + yield batch - progress_cb(msg=f"Start document knowledge compilation ({total} template(s)) ...") - - batch_no = 0 - async for batch in handler._load_chunks_for_doc( - ctx.tenant_id, - ctx.kb_id, - ctx.doc_id, - batch_size=DOC_STRUCTURE_COMPILE_BATCH_CHUNKS, - ): - batch_no += 1 - for chunk in batch: - cid = chunk.get("id") - if isinstance(cid, str) and cid not in chunks_by_id: - text = chunk.get("content_with_weight") or "" - chunks_by_id[cid] = text if isinstance(text, str) else "" - for idx, (template_id, parser_cfg) in enumerate(active_templates): - progress_cb(msg=f" compile batch {batch_no} ({len(batch)} chunks) for template ({idx + 1}/{total})") - docs = await compile_structure_from_text( - batch, - parser_cfg, - chat_mdl_by_tid[template_id], - embedding_model, - ctx.doc_id, - language=ctx.language, - callback=progress_cb, - compilation_template_id=template_id, - ) - if docs: - accumulators[template_id].extend(docs) - if len(accumulators[template_id]) >= DOC_STRUCTURE_MERGE_MAX_DOCS: - progress_cb(msg=f" merge flush ({len(accumulators[template_id])} docs) for template ({idx + 1}/{total})") - await _flush(template_id) - - for idx, (template_id, parser_cfg) in enumerate(active_templates): - if ctx.has_canceled_func(ctx.id): - raise TaskCanceledException(f"Task {ctx.id} was cancelled during document knowledge compilation") - await _flush(template_id) - agg = agg_infos[template_id] - ctx.recording_context.record(f"document_structure_compile:{template_id}", agg) - progress_cb(msg=f"Document knowledge compilation done ({idx + 1}/{total}): {agg}") - - # ── Synthesis phase ────────────────────────────────────────────── - # If the template has synthesis.enabled, run wiki PLAN+REFINE - # to generate output (wiki page, essence paragraph, etc.). - synthesis_cfg = (parser_cfg or {}).get("synthesis") or {} - if synthesis_cfg.get("enabled"): - example = synthesis_cfg.get("example") - compile_kwd = synthesis_cfg.get("compile_kwd", "artifact_page") - plan_cfg = synthesis_cfg.get("plan") or {} - - # Reserved for future wiki_plan_from_reduction extension: - # entity_type_filter, mention_count_threshold, top_n - if plan_cfg: - logging.debug( - "synthesis: template %s plan config %r reserved for future use", - template_id, plan_cfg, - ) - - if ctx.has_canceled_func(ctx.id): - raise TaskCanceledException( - f"Task {ctx.id} was cancelled before synthesis PLAN" - ) - - if not example: - logging.warning( - "synthesis: template %s has synthesis.enabled but no example; skipping", - template_id, - ) - else: - try: - from rag.advanced_rag.knowlege_compile.wiki import ( - wiki_plan_from_reduction, - wiki_refine_from_plan, - ) - - progress_cb( - msg=f"Synthesis PLAN for template {template_id} (kind={compile_kwd}) ..." - ) - plan = await wiki_plan_from_reduction( - chat_mdl=chat_mdl_by_tid[template_id], - embd_mdl=embedding_model, - tenant_id=ctx.tenant_id, - kb_id=ctx.kb_id, - callback=progress_cb, - ) - if ctx.has_canceled_func(ctx.id): - raise TaskCanceledException( - f"Task {ctx.id} was cancelled after synthesis PLAN" - ) - - if not plan or not plan.get("pages"): - progress_cb( - msg=f"Synthesis: no pages planned for template {template_id}." - ) - else: - progress_cb( - msg=f"Synthesis REFINE for template {template_id} ({len(plan['pages'])} page(s)) ..." - ) - pages = await wiki_refine_from_plan( - chat_mdl=chat_mdl_by_tid[template_id], - embd_mdl=embedding_model, - tenant_id=ctx.tenant_id, - kb_id=ctx.kb_id, - callback=progress_cb, - example=example, - ) - # Overwrite compile_kwd on every output page so the - # synthesis type is tracked correctly in ES. - for p in pages or []: - p["compile_kwd"] = compile_kwd - progress_cb( - msg=f"Synthesis done: {len(pages or [])} {compile_kwd} page(s) written." - ) - except Exception: - logging.exception( - "synthesis: failed for template %s", template_id, - ) + await run_structure_compile_over_batches( + active_templates=non_tree_templates, + chat_mdl_by_tid=chat_mdl_by_tid, + embedding_model=embedding_model, + tenant_id=ctx.tenant_id, + kb_id=ctx.kb_id, + doc_id=ctx.doc_id, + language=ctx.language, + chunk_batches=_stream_doc_batches(), + progress_cb=ctx.progress_cb, + cancel_check=lambda: ctx.has_canceled_func(ctx.id), + record=ctx.recording_context.record, + ) async def run_document_post_chunking_if_last( diff --git a/rag/svr/task_executor_refactor/dataflow_service.py b/rag/svr/task_executor_refactor/dataflow_service.py index 355fc02b5..c4912655f 100644 --- a/rag/svr/task_executor_refactor/dataflow_service.py +++ b/rag/svr/task_executor_refactor/dataflow_service.py @@ -109,7 +109,14 @@ class DataflowService: dataflow_id = corrected_id # Run pipeline - pipeline = Pipeline(dsl, tenant_id=ctx.tenant_id, doc_id=doc_id, task_id=task_id, flow_id=dataflow_id) + pipeline = Pipeline( + dsl, + tenant_id=ctx.tenant_id, + doc_id=doc_id, + task_id=task_id, + flow_id=dataflow_id, + language=ctx.language, + ) chunks = await pipeline.run(file=ctx.file) if ctx.file else await pipeline.run() if doc_id == CANVAS_DEBUG_DOC_ID: diff --git a/rag/svr/task_executor_refactor/dataset_wiki_generator.py b/rag/svr/task_executor_refactor/dataset_wiki_generator.py index 304dd2455..a747266ab 100644 --- a/rag/svr/task_executor_refactor/dataset_wiki_generator.py +++ b/rag/svr/task_executor_refactor/dataset_wiki_generator.py @@ -24,9 +24,9 @@ methods and one ``_run_wiki`` orchestrator. The public entry point is The pipeline runs MAP per (doc, template) — each MAP call resumes from its own ``artifact_map_extract`` ES rows — then REDUCE / PLAN / REFINE KB-wide via ``rag.advanced_rag.knowlege_compile.wiki``. Refined pages -land in ES twice: once as searchable ``artifact_page`` rows and once as -``artifact_entity`` / ``artifact_relation`` rows for the dataset -Artifact tab's canvas graph. +land in ES as searchable ``artifact_page`` rows, one +``artifact_page_topic`` row per topic, and ``artifact_entity`` / +``artifact_relation`` rows for the dataset Artifact tab's canvas graph. Design notes: @@ -46,6 +46,7 @@ from __future__ import annotations import asyncio import json import logging +import re from typing import AsyncIterator, Callable, Dict, List, Optional import xxhash @@ -83,6 +84,8 @@ WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE = 64 # path, which carries the user's own title/comments). WIKI_REGEN_COMMIT_TITLE = "Regenerated by artifact compilation" WIKI_REGEN_COMMIT_COMMENTS_TEMPLATE = "Auto-update via run_wiki (action={action})" +WIKI_PAGE_COMPILE_KWD = "artifact_page" +WIKI_PAGE_TOPIC_COMPILE_KWD = "artifact_page_topic" # ----- helpers ------------------------------------------------------- @@ -109,6 +112,95 @@ def _parser_config_compilation_template_ids(parser_config, tenant_id: str) -> li return template_ids +def _wiki_topic_from_page(page: Dict, fallback: str = "") -> str: + for key in ("topic", "title", "page_type"): + value = page.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return fallback.strip() + + +def _wiki_topic_slug(topic: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", topic.lower()).strip("-") + if not slug: + slug = xxhash.xxh64(topic.encode("utf-8", "surrogatepass")).hexdigest()[:12] + return f"topic/{slug[:80]}" + + +async def _ensure_wiki_topic_rows( + ctx: TaskContext, + index: str, + kb_id_str: str, + topics_by_name: dict[str, str], +) -> None: + if not topics_by_name: + return + + from common.doc_store.doc_store_base import OrderByExpr + from rag.nlp import rag_tokenizer + + topics = list(topics_by_name.keys()) + existing: set[str] = set() + try: + res = await thread_pool_exec( + settings.docStoreConn.search, + ["id", "topic_kwd"], + [], + { + "compile_kwd": [WIKI_PAGE_TOPIC_COMPILE_KWD], + "topic_kwd": topics, + }, + [], + OrderByExpr(), + 0, + max(len(topics), 1), + index, + [ctx.kb_id], + ) + field_map = settings.docStoreConn.get_fields(res, ["id", "topic_kwd"]) + for row in (field_map or {}).values(): + raw = row.get("topic_kwd") + if isinstance(raw, list): + existing.update(t for t in raw if isinstance(t, str) and t) + elif isinstance(raw, str) and raw: + existing.add(raw) + except Exception: + logging.exception( + "wiki_persist: topic existence read failed for kb=%s; inserting stable topic ids", + kb_id_str, + ) + + rows: list[dict] = [] + for topic, slug in topics_by_name.items(): + if topic in existing: + continue + topic_id = xxhash.xxh64( + f"{kb_id_str}:{WIKI_PAGE_TOPIC_COMPILE_KWD}:{topic}".encode( + "utf-8", + "surrogatepass", + ), + ).hexdigest() + content_ltks = rag_tokenizer.tokenize(topic) + rows.append( + { + "id": topic_id, + "kb_id": kb_id_str, + "doc_id": kb_id_str, + "compile_kwd": WIKI_PAGE_TOPIC_COMPILE_KWD, + "topic_kwd": topic, + "title_kwd": topic, + "slug_kwd": slug, + "content_with_weight": topic, + "content_ltks": content_ltks, + "content_sm_ltks": rag_tokenizer.fine_grained_tokenize(content_ltks), + "available_int": 1, + } + ) + + if rows: + await thread_pool_exec(settings.docStoreConn.insert, rows, index, ctx.kb_id) + + # ----- persistence --------------------------------------------------- @@ -125,6 +217,7 @@ async def persist_wiki_pages_to_es( slug_kwd page.slug title_kwd page.title page_type_kwd page.page_type + topic_kwd page.topic entity_names_kwd page.entity_names outlinks_kwd page.outlinks related_kb_pages_kwd page.related_kb_pages @@ -163,7 +256,7 @@ async def persist_wiki_pages_to_es( settings.docStoreConn.search, ["id", "slug_kwd", "content_with_weight"], [], - {"compile_kwd": ["artifact_page"], "slug_kwd": list(target_slugs)}, + {"compile_kwd": [WIKI_PAGE_COMPILE_KWD], "slug_kwd": list(target_slugs)}, [], OrderByExpr(), 0, @@ -213,11 +306,13 @@ async def persist_wiki_pages_to_es( return rows: List[Dict] = [] + topics_by_name: dict[str, str] = {} for page, vec in zip(pages, embeddings): slug = page.get("slug") or "" if not slug: continue title = page.get("title") or slug + topic = _wiki_topic_from_page(page, title) summary = page.get("summary") or "" content_md = page.get("content_md_rendered") or page.get("content_md") or page.get("content_md_raw") or "" @@ -242,10 +337,11 @@ async def persist_wiki_pages_to_es( "id": row_id, "kb_id": kb_id_str, "doc_id": kb_id_str, # sentinel; KB-scoped row, real provenance in source_doc_ids - "compile_kwd": "artifact_page", + "compile_kwd": WIKI_PAGE_COMPILE_KWD, "slug_kwd": slug, "title_kwd": title, "page_type_kwd": page.get("page_type") or "concept", + "topic_kwd": topic, "entity_names_kwd": list(page.get("entity_names") or []), "outlinks_kwd": list(page.get("outlinks") or []), "outlinks_int": len(list(page.get("outlinks") or [])), @@ -263,6 +359,8 @@ async def persist_wiki_pages_to_es( "available_int": 1, } ) + if topic: + topics_by_name.setdefault(topic, _wiki_topic_slug(topic)) if not rows: return @@ -277,6 +375,15 @@ async def persist_wiki_pages_to_es( ) return + if topics_by_name: + try: + await _ensure_wiki_topic_rows(ctx, index, kb_id_str, topics_by_name) + except Exception: + logging.exception( + "wiki_persist: topic row insert failed for kb=%s", + kb_id_str, + ) + # Audit trail: one ArtifactCommit row per page whose rendered # content actually changed (record_edit silently skips empty diffs). # Best-effort — commit failures log but don't fail the artifact