Feat: Add knowledge compilation workflows (#16515)

## Summary
- Add knowledge compilation template APIs, services, and builtin
template seed data
- Add advanced knowledge compile structure/artifact/RAPTOR workflow
support
- Update parsing, dataset/document APIs, and supporting services for
compilation workflows

rag/svr/task_executor_refactor/chunk_post_processor.py

@@ -74,8 +74,7 @@ async def extract_keywords(docs: List[Dict], ctx: TaskContext) -> None:
 
         tasks = []
         for doc in docs:
-            tasks.append(
-                asyncio.create_task(doc_keyword_extraction(chat_model, doc, ctx.parser_config["auto_keywords"])))
+            tasks.append(asyncio.create_task(doc_keyword_extraction(chat_model, doc, ctx.parser_config["auto_keywords"])))
         try:
             await asyncio.gather(*tasks, return_exceptions=False)
         except Exception as e:
@@ -116,8 +115,7 @@ async def generate_questions(docs: List[Dict], ctx: TaskContext) -> None:
 
         tasks = []
         for doc in docs:
-            tasks.append(
-                asyncio.create_task(doc_question_proposal(chat_model, doc, ctx.parser_config["auto_questions"])))
+            tasks.append(asyncio.create_task(doc_question_proposal(chat_model, doc, ctx.parser_config["auto_questions"])))
         try:
             await asyncio.gather(*tasks, return_exceptions=False)
         except Exception as e:
@@ -184,18 +182,14 @@ async def generate_metadata(docs: List[Dict], ctx: TaskContext) -> None:
         metadata_conf = build_metadata_config(ctx.parser_config)
 
         async def gen_metadata_task(chat_mdl, d):
-            cached = get_llm_cache(chat_mdl.llm_name, d["content_with_weight"], "metadata",
-                                   metadata_conf)
+            cached = get_llm_cache(chat_mdl.llm_name, d["content_with_weight"], "metadata", metadata_conf)
             if not cached:
                 if ctx.has_canceled_func(ctx.id):
                     ctx.progress_cb(-1, msg="Task has been canceled.")
                     return
                 async with chat_limiter:
-                    cached = await gen_metadata(chat_mdl,
-                                                turn2jsonschema(metadata_conf),
-                                                d["content_with_weight"])
-                set_llm_cache(chat_mdl.llm_name, d["content_with_weight"], cached, "metadata",
-                              metadata_conf)
+                    cached = await gen_metadata(chat_mdl, turn2jsonschema(metadata_conf), d["content_with_weight"])
+                set_llm_cache(chat_mdl.llm_name, d["content_with_weight"], cached, "metadata", metadata_conf)
             if cached:
                 d["metadata_obj"] = cached
 
@@ -256,8 +250,7 @@ async def apply_tags(docs: List[Dict], ctx: TaskContext) -> None:
             if ctx.has_canceled_func(ctx.id):
                 ctx.progress_cb(-1, msg="Task has been canceled.")
                 return
-            if settings.retriever.tag_content(tenant_id, kb_ids, doc, all_tags, topn_tags=topn_tags, S=S) and len(
-                    doc.get(TAG_FLD, [])) > 0:
+            if settings.retriever.tag_content(tenant_id, kb_ids, doc, all_tags, topn_tags=topn_tags, S=S) and len(doc.get(TAG_FLD, [])) > 0:
                 examples.append({"content": doc["content_with_weight"], TAG_FLD: doc[TAG_FLD]})
             else:
                 docs_to_tag.append(doc)
@@ -270,7 +263,7 @@ async def apply_tags(docs: List[Dict], ctx: TaskContext) -> None:
                     return
                 picked_examples = random.choices(examples, k=2) if len(examples) > 2 else examples
                 if not picked_examples:
-                    picked_examples.append({"content": "This is an example", TAG_FLD: {'example': 1}})
+                    picked_examples.append({"content": "This is an example", TAG_FLD: {"example": 1}})
                 async with chat_limiter:
                     cached = await content_tagging(
                         chat_mdl,
@@ -310,3 +303,883 @@ def count_with_key(docs: List[Dict], key: str) -> int:
         Count of docs that have the key.
     """
     return sum(1 for d in docs if d.get(key))
+
+
+# =====================================================================
+# Document post-chunking pipeline
+# ---------------------------------------------------------------------
+# Extracted from ``task_handler`` to keep the handler class small.
+# The public entry point is :func:`run_document_post_chunking_if_last`;
+# everything below is called (transitively) from there:
+#   run_document_post_chunking_if_last
+#     ├─ run_document_structure_compile
+#     │    ├─ run_tree_templates
+#     │    │    ├─ load_chunks_with_vec
+#     │    │    ├─ rechunk_doc_by_tree
+#     │    │    └─ raptor_tree_to_graph
+#     │    └─ (streaming compile via chat models per template)
+#     └─ handler._run_raptor       ← stays on the handler
+#
+# All entries take ``handler`` (``TaskHandler``) as their first arg so
+# they can reach the handler's ``_task_context``, ``_run_raptor``, and
+# ``_load_chunks_for_doc`` without a circular import.
+# =====================================================================
+
+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.apps.restful_apis.chunk_api import _compilation_template_kind  # 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,
+)
+from api.db.services.task_service import (  # noqa: E402
+    abort_doc_chunking_counter,
+    clear_doc_chunking_counter,
+    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
+
+
+# ----- parser_config helpers -----------------------------------------
+# Duplicated from ``task_handler`` so this module stays free of a
+# reverse import (task_handler → this module via dispatch; the other
+# direction would be circular).
+
+
+def _parser_config_compilation_template_group_ids(parser_config) -> list[str]:
+    def _normalize(raw) -> list[str]:
+        if isinstance(raw, str):
+            raw = [raw]
+        if not isinstance(raw, list):
+            return []
+        ids: list[str] = []
+        seen: set[str] = set()
+        for gid in raw:
+            if not isinstance(gid, str):
+                continue
+            gid = gid.strip()
+            if gid and gid not in seen:
+                seen.add(gid)
+                ids.append(gid)
+        return ids
+
+    if not isinstance(parser_config, dict):
+        return []
+    if "compilation_template_group_id" in parser_config:
+        return _normalize(parser_config.get("compilation_template_group_id"))
+    ext = parser_config.get("ext")
+    if isinstance(ext, dict):
+        return _normalize(ext.get("compilation_template_group_id"))
+    return []
+
+
+def _parser_config_compilation_template_ids(parser_config, tenant_id: str) -> list[str]:
+    template_ids: list[str] = []
+    seen: set[str] = set()
+    for group_id in _parser_config_compilation_template_group_ids(parser_config):
+        for template_id in CompilationTemplateGroupService.resolve_template_ids(
+            group_id,
+            tenant_id,
+        ):
+            if template_id in seen:
+                continue
+            seen.add(template_id)
+            template_ids.append(template_id)
+    return template_ids
+
+
+def _resolve_template_chat_llm_id(parser_cfg: dict, ctx) -> str:
+    """Pick the chat model id for a knowledge-compilation template.
+
+    Resolution order: template ``llm_id`` → doc ``parser_config.llm_id``
+    → ``ctx.llm_id`` (the chunking task's default).
+    """
+    if isinstance(parser_cfg, dict):
+        tid = parser_cfg.get("llm_id")
+        if isinstance(tid, str) and tid.strip():
+            return tid.strip()
+    doc_cfg = getattr(ctx, "parser_config", None) or {}
+    if isinstance(doc_cfg, dict):
+        did = doc_cfg.get("llm_id")
+        if isinstance(did, str) and did.strip():
+            return did.strip()
+    return ctx.llm_id
+
+
+# ----- progress helper -----------------------------------------------
+
+
+def cap_done_progress(progress_cb: Callable) -> Callable:
+    """Wrap a progress callback so any ``prog >= 1`` gets clamped to
+    ``0.99`` — the final ``1.0`` is reserved for the caller who owns
+    the task's terminal state."""
+
+    def capped_progress(*args, **kwargs):
+        args = list(args)
+        if args:
+            prog = args[0]
+            if isinstance(prog, (int, float)) and not isinstance(prog, bool) and prog >= 1:
+                args[0] = 0.99
+        if "prog" in kwargs:
+            prog = kwargs["prog"]
+            if isinstance(prog, (int, float)) and not isinstance(prog, bool) and prog >= 1:
+                kwargs["prog"] = 0.99
+        return progress_cb(*args, **kwargs)
+
+    return capped_progress
+
+
+# ----- tree helpers --------------------------------------------------
+
+
+def raptor_tree_to_graph(tree: Dict) -> Dict:
+    """Project a RAPTOR tree dict (from ``Raptor(is_tree=True)``) onto
+    the ``{entities, relations}`` shape the document-structure graph
+    endpoint already serves for ``page_index``-kind rows."""
+    entities: list[dict] = []
+    relations: list[dict] = []
+
+    def _walk(node: dict, parent_id: Optional[str]) -> None:
+        if not isinstance(node, dict):
+            return
+        title = node.get("title") or ""
+        node_id = title
+        ent: dict = {
+            "name": node_id,
+            "type": "tree_node",
+            "description": node.get("description", title),
+            "mention_count": 1,
+        }
+        src_ids = node.get("source_chunk_ids")
+        if isinstance(src_ids, list) and src_ids:
+            ent["source_chunk_ids"] = [s for s in src_ids if isinstance(s, str) and s]
+        entities.append(ent)
+        if parent_id is not None:
+            relations.append({"from": parent_id, "to": node_id, "type": "child"})
+        for child in node.get("children") or []:
+            _walk(child, node_id)
+
+    _walk(tree, None)
+    return {"entities": entities, "relations": relations}
+
+
+async def load_chunks_with_vec(
+    tenant_id: str,
+    kb_id: str,
+    doc_id: str,
+    vctr_nm: str,
+) -> list[tuple[str, "np.ndarray", str]]:
+    """Page through this doc's chunks pulling content + vector +
+    chunk_id, in the shape ``RaptorService.build_doc_tree`` expects.
+    Mirrors the streaming ``_load_chunks_for_doc`` loader but with the
+    vector field pre-selected."""
+    from common.doc_store.doc_store_base import OrderByExpr
+
+    index_nm = search.index_name(tenant_id)
+    if not settings.docStoreConn.index_exist(index_nm, kb_id):
+        return []
+    select_fields = ["id", "doc_id", "content_with_weight", vctr_nm]
+    order_by = OrderByExpr()
+    order_by.asc("page_num_int")
+    order_by.asc("top_int")
+
+    out: list[tuple[str, "np.ndarray", str]] = []
+    offset = 0
+    PAGE = 500
+    while True:
+        try:
+            res = await thread_pool_exec(
+                settings.docStoreConn.search,
+                select_fields,
+                [],
+                {"doc_id": [doc_id], "available_int": 1},
+                [],
+                order_by,
+                offset,
+                PAGE,
+                index_nm,
+                [kb_id],
+            )
+            field_map = settings.docStoreConn.get_fields(res, select_fields)
+        except Exception:
+            logging.exception(
+                "tree-template: failed to load chunks for doc=%s",
+                doc_id,
+            )
+            break
+        if not field_map:
+            break
+        for row_id, row in field_map.items():
+            if row.get("compile_kwd"):
+                continue
+            text = row.get("content_with_weight") or ""
+            vec = row.get(vctr_nm)
+            if not text or vec is None:
+                continue
+            try:
+                arr = np.asarray(vec, dtype=np.float32)
+            except Exception:
+                continue
+            if arr.size == 0:
+                continue
+            out.append((text, arr, str(row_id)))
+        if len(field_map) < PAGE:
+            break
+        offset += PAGE
+    return out
+
+
+async def rechunk_doc_by_tree(
+    handler,
+    tree: dict,
+    template_id: str,
+    embedding_model,
+) -> None:
+    """Merge each leaf cluster's source chunks into a single
+    replacement chunk and rewrite the tree's leaf-cluster
+    ``source_chunk_ids`` in-place. Original chunks are soft-deleted
+    via ``available_int=0`` and stamped with ``superseded_by_chunk_id``.
+    """
+    from datetime import datetime
+    from common.misc_utils import get_uuid
+
+    ctx = handler._task_context
+
+    cluster_id_map: dict[int, tuple[dict, list[str]]] = {}
+
+    def _is_terminal(node: object) -> bool:
+        return isinstance(node, dict) and not (node.get("children") or [])
+
+    def _walk(node: object) -> None:
+        if not isinstance(node, dict):
+            return
+        children = node.get("children") or []
+        if children and all(_is_terminal(c) for c in children):
+            src_ids: list[str] = []
+            seen: set[str] = set()
+            for c in children:
+                for cid in c.get("source_chunk_ids") or []:
+                    if isinstance(cid, str) and cid and cid not in seen:
+                        seen.add(cid)
+                        src_ids.append(cid)
+            for cid in node.get("source_chunk_ids") or []:
+                if isinstance(cid, str) and cid and cid not in seen:
+                    seen.add(cid)
+                    src_ids.append(cid)
+            if src_ids:
+                cluster_id_map[id(node)] = (node, src_ids)
+        else:
+            for c in children:
+                _walk(c)
+
+    _walk(tree)
+    if not cluster_id_map:
+        return
+
+    all_source_ids = sorted({sid for _, ids in cluster_id_map.values() for sid in ids})
+
+    from common.doc_store.doc_store_base import OrderByExpr
+
+    index_nm = search.index_name(ctx.tenant_id)
+    if not settings.docStoreConn.index_exist(index_nm, ctx.kb_id):
+        return
+
+    vctr_nm = "q_%d_vec" % len(embedding_model.encode(["x"])[0][0])
+    select_fields = [
+        "id",
+        "doc_id",
+        "kb_id",
+        "content_with_weight",
+        "page_num_int",
+        "top_int",
+        "position_int",
+        "docnm_kwd",
+        "title_tks",
+        "title_sm_tks",
+        "available_int",
+    ]
+    try:
+        res = await thread_pool_exec(
+            settings.docStoreConn.search,
+            select_fields,
+            [],
+            {"id": all_source_ids, "available_int": 1},
+            [],
+            OrderByExpr(),
+            0,
+            len(all_source_ids) + 16,
+            index_nm,
+            [ctx.kb_id],
+        )
+        field_map = settings.docStoreConn.get_fields(res, select_fields)
+    except Exception:
+        logging.exception(
+            "rechunk: failed to load source chunks for doc=%s template=%s",
+            ctx.doc_id,
+            template_id,
+        )
+        return
+    if not field_map:
+        return
+
+    chunks_by_id: dict[str, dict] = {str(rid): {**row, "id": str(rid)} for rid, row in field_map.items()}
+
+    merged_rows: list[dict] = []
+    cluster_new_id: dict[int, str] = {}
+
+    for node_id_int, (node, src_ids) in cluster_id_map.items():
+        cluster_chunks = [chunks_by_id[c] for c in src_ids if c in chunks_by_id]
+        if not cluster_chunks:
+            continue
+
+        def _sort_key(c: dict) -> tuple:
+            pages = c.get("page_num_int") or [0]
+            tops = c.get("top_int") or [0]
+            return (
+                min(pages) if pages else 0,
+                min(tops) if tops else 0,
+                c.get("id") or "",
+            )
+
+        cluster_chunks.sort(key=_sort_key)
+
+        merged_content = "\n\n".join((c.get("content_with_weight") or "") for c in cluster_chunks).strip()
+        if not merged_content:
+            continue
+        page_union = sorted({p for c in cluster_chunks for p in (c.get("page_num_int") or [])})
+        top_union = sorted({t for c in cluster_chunks for t in (c.get("top_int") or [])})
+
+        base = dict(cluster_chunks[0])
+        new_id = get_uuid()
+        cluster_new_id[node_id_int] = new_id
+
+        base.update(
+            {
+                "id": new_id,
+                "content_with_weight": merged_content,
+                "content_ltks": rag_tokenizer.tokenize(merged_content),
+                "page_num_int": page_union,
+                "top_int": top_union,
+                "available_int": 1,
+                "rechunk_kwd": "tree",
+                "rechunked_from_template_id": template_id,
+                "rechunked_from_chunk_ids": [c.get("id") for c in cluster_chunks if c.get("id")],
+                "token_num": num_tokens_from_string(merged_content),
+                "create_time": str(datetime.now()).replace("T", " ")[:19],
+                "create_timestamp_flt": datetime.now().timestamp(),
+            }
+        )
+        base["content_sm_ltks"] = rag_tokenizer.fine_grained_tokenize(base["content_ltks"])
+        merged_rows.append(base)
+
+    if not merged_rows:
+        return
+
+    contents = [r["content_with_weight"] for r in merged_rows]
+    try:
+        vectors, _ = embedding_model.encode(contents)
+    except Exception:
+        logging.exception(
+            "rechunk: embedding failed for doc=%s template=%s",
+            ctx.doc_id,
+            template_id,
+        )
+        return
+    for row, vec in zip(merged_rows, vectors):
+        try:
+            row[vctr_nm] = np.asarray(vec, dtype=np.float32).tolist()
+        except Exception:
+            logging.exception(
+                "rechunk: vector cast failed; skipping row %s",
+                row.get("id"),
+            )
+            row[vctr_nm] = None
+    merged_rows = [r for r in merged_rows if r.get(vctr_nm) is not None]
+    if not merged_rows:
+        return
+
+    try:
+        await thread_pool_exec(
+            settings.docStoreConn.insert,
+            merged_rows,
+            index_nm,
+            ctx.kb_id,
+        )
+    except Exception:
+        logging.exception(
+            "rechunk: insert failed for doc=%s template=%s",
+            ctx.doc_id,
+            template_id,
+        )
+        return
+
+    for node_id_int, new_chunk_id in cluster_new_id.items():
+        node, _ = cluster_id_map[node_id_int]
+        node["source_chunk_ids"] = [new_chunk_id]
+        for child in node.get("children") or []:
+            if isinstance(child, dict):
+                child["source_chunk_ids"] = [new_chunk_id]
+
+    for node_id_int, new_chunk_id in cluster_new_id.items():
+        _, src_ids = cluster_id_map[node_id_int]
+        for cid in src_ids:
+            try:
+                await thread_pool_exec(
+                    settings.docStoreConn.update,
+                    {"id": cid},
+                    {
+                        "available_int": 0,
+                        "superseded_by_chunk_id": new_chunk_id,
+                    },
+                    index_nm,
+                    ctx.kb_id,
+                )
+            except Exception:
+                logging.exception(
+                    "rechunk: soft-delete failed for chunk=%s (merged=%s)",
+                    cid,
+                    new_chunk_id,
+                )
+
+
+async def run_tree_templates(
+    handler,
+    templates: list[tuple[str, dict]],
+    chat_mdl_by_tid: dict[str, "LLMBundle"],
+    embedding_model,
+) -> None:
+    """Run the ``tree``-kind compilation templates for the current
+    doc. Each pair runs RAPTOR with ``is_tree=True`` via
+    ``RaptorService.build_doc_tree`` and persists a single graph row
+    via ``_struct_upsert_graph_json``."""
+    from rag.svr.task_executor_refactor.raptor_service import RaptorService
+    from rag.advanced_rag.knowlege_compile.structure import _struct_upsert_graph_json
+
+    ctx = handler._task_context
+    progress_cb = ctx.progress_cb
+
+    try:
+        doc_id = ctx.doc_id
+    except Exception:
+        doc_id = getattr(ctx, "_task", {}).get("doc_id") if hasattr(ctx, "_task") else None
+    if not doc_id:
+        logging.warning("tree-template: no doc_id on task context; skipping")
+        return
+
+    vctr_nm = "q_%d_vec" % len(embedding_model.encode(["x"])[0][0])
+    chunks = await load_chunks_with_vec(
+        ctx.tenant_id,
+        ctx.kb_id,
+        doc_id,
+        vctr_nm,
+    )
+    if not chunks:
+        progress_cb(msg=f"tree-template: doc {doc_id} has no chunks; skipping")
+        return
+
+    raptor_service = RaptorService(ctx)
+
+    for idx, (template_id, parser_cfg) in enumerate(templates):
+        raptor_cfg = (parser_cfg or {}).get("raptor") or {}
+        raptor_config = {
+            "prompt": raptor_cfg.get("prompt") or "Please write a concise summary of the following texts:\n{cluster_content}",
+            "max_token": int(raptor_cfg.get("max_token") or 512),
+            "threshold": float(raptor_cfg.get("threshold") or 0.1),
+            "random_seed": int(raptor_cfg.get("random_seed") or 0),
+            "max_cluster": int(raptor_cfg.get("max_cluster") or 64),
+            "ext": raptor_cfg.get("ext") or {},
+        }
+        progress_cb(
+            msg=f"tree-template ({idx + 1}/{len(templates)}): building tree for doc={doc_id}",
+        )
+        try:
+            tree = await raptor_service.build_doc_tree(
+                chunks=chunks,
+                raptor_config=raptor_config,
+                chat_mdl=chat_mdl_by_tid[template_id],
+                embd_mdl=embedding_model,
+                tree_builder="raptor",
+                clustering_method="gmm",
+                max_errors=3,
+            )
+        except Exception:
+            logging.exception(
+                "tree-template %s: RAPTOR build failed for doc %s",
+                template_id,
+                doc_id,
+            )
+            continue
+        if tree is None:
+            logging.info(
+                "tree-template %s: no tree produced for doc %s",
+                template_id,
+                doc_id,
+            )
+            continue
+
+        if bool((raptor_cfg or {}).get("rechunk")):
+            try:
+                await rechunk_doc_by_tree(
+                    handler=handler,
+                    tree=tree,
+                    template_id=template_id,
+                    embedding_model=embedding_model,
+                )
+            except Exception:
+                logging.exception(
+                    "tree-template %s: re-chunking failed for doc %s; persisting tree with original chunk ids",
+                    template_id,
+                    doc_id,
+                )
+
+        graph = raptor_tree_to_graph(tree)
+        try:
+            await _struct_upsert_graph_json(
+                graph,
+                ctx.tenant_id,
+                ctx.kb_id,
+                doc_id,
+                compile_kwd="tree",
+                compilation_template_id=template_id,
+            )
+        except Exception:
+            logging.exception(
+                "tree-template %s: graph upsert failed for doc %s",
+                template_id,
+                doc_id,
+            )
+            continue
+
+        try:
+            from rag.advanced_rag.knowlege_compile.dataset_nav import (
+                upsert_dataset_nav_doc,
+            )
+
+            await upsert_dataset_nav_doc(
+                ctx.tenant_id,
+                ctx.kb_id,
+                doc_id,
+                tree,
+            )
+        except Exception:
+            logging.exception(
+                "tree-template %s: dataset_nav upsert failed for doc %s",
+                template_id,
+                doc_id,
+            )
+
+        progress_cb(
+            msg=f"tree-template ({idx + 1}/{len(templates)}): persisted {len(graph['entities'])} node(s), {len(graph['relations'])} edge(s) for doc {doc_id}",
+        )
+
+
+async def run_document_structure_compile(handler, embedding_model: LLMBundle) -> None:
+    """Run document-scoped knowledge compilation for non-artifact
+    templates. Streams the doc's chunks (via
+    ``handler._load_chunks_for_doc``) and fans each batch out to every
+    configured non-artifact template, flushing accumulators through
+    ``merge_compiled_structures`` at :data:`DOC_STRUCTURE_MERGE_MAX_DOCS`.
+    """
+    ctx = handler._task_context
+    template_ids = _parser_config_compilation_template_ids(ctx.parser_config, ctx.tenant_id)
+    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))
+
+    if not active_templates:
+        return
+
+    llm_bundle_cache: dict[str, LLMBundle] = {}
+    chat_mdl_by_tid: dict[str, LLMBundle] = {}
+    filtered_templates: list[tuple[str, dict]] = []
+    for template_id, parser_cfg in active_templates:
+        chat_llm_id = _resolve_template_chat_llm_id(parser_cfg, ctx)
+        if chat_llm_id not in llm_bundle_cache:
+            try:
+                cfg = get_model_config_from_provider_instance(
+                    ctx.tenant_id,
+                    LLMType.CHAT,
+                    chat_llm_id,
+                )
+                llm_bundle_cache[chat_llm_id] = LLMBundle(
+                    ctx.tenant_id,
+                    cfg,
+                    lang=ctx.language,
+                )
+            except Exception:
+                logging.exception(
+                    "document_structure_compile: 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]
+        filtered_templates.append((template_id, parser_cfg))
+
+    if not filtered_templates:
+        return
+    active_templates = filtered_templates
+
+    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))
+
+    if tree_templates:
+        await run_tree_templates(
+            handler,
+            tree_templates,
+            chat_mdl_by_tid,
+            embedding_model,
+        )
+
+    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,
+            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)
+
+    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}")
+
+
+async def run_document_post_chunking_if_last(
+    handler,
+    embedding_model: LLMBundle,
+    vector_size: int,
+    task_start_ts: float,
+    chunks_len: int,
+    token_count: int,
+) -> bool:
+    """Gate: only the last chunking task for a doc runs post-processing.
+    Returns ``True`` if the caller may proceed to its own terminal
+    progress update, ``False`` if the task was cancelled.
+
+    The pass runs :func:`run_document_structure_compile` and
+    ``handler._run_raptor`` concurrently — they read the same chunks
+    but write disjoint ES rows.
+    """
+    ctx = handler._task_context
+    task_id = ctx.id
+    task_doc_id = ctx.doc_id
+
+    if ctx.has_canceled_func(task_id):
+        abort_doc_chunking_counter(task_doc_id)
+        ctx.progress_cb(-1, msg="Task has been canceled.")
+        return False
+
+    chunking_aborted = is_doc_chunking_aborted(task_doc_id)
+    remaining_chunking_tasks = (
+        0 if ctx.write_interceptor
+        else credit_doc_chunking_task(task_doc_id, task_id)
+    )
+    if remaining_chunking_tasks != 0:
+        if chunking_aborted:
+            logging.info(
+                "Chunking for doc %s was aborted before task %s reached post-processing; "
+                "skip document finalizers.",
+                task_doc_id,
+                task_id,
+            )
+        elif remaining_chunking_tasks is not None and remaining_chunking_tasks < 0:
+            logging.warning(
+                "Chunking counter for doc %s is missing or expired after task %s; skip post-processing to avoid duplicate finalizers.",
+                task_doc_id,
+                task_id,
+            )
+        else:
+            logging.info(
+                "Chunk doc(%s), page(%s-%s), chunks(%s), token(%s), elapsed:%.2f; waiting for %s chunking task(s) before post-processing",
+                ctx.name,
+                ctx.from_page,
+                ctx.to_page,
+                chunks_len,
+                token_count,
+                timer() - task_start_ts,
+                remaining_chunking_tasks,
+            )
+        return True
+
+    async def _maybe_run_raptor():
+        raptor_cfg = (ctx.parser_config or {}).get("raptor") or {}
+        if not raptor_cfg.get("use_raptor"):
+            return
+        try:
+            ok_doc, doc_obj = DocumentService.get_by_id(task_doc_id)
+            if ok_doc and doc_obj is not None:
+                ctx.progress_cb(msg="Starting RAPTOR task.")
+                await handler._run_raptor(embedding_model, vector_size, mark_done=False)
+            else:
+                logging.warning(
+                    "raptor: cannot resolve doc %s to queue per-doc task",
+                    task_doc_id,
+                )
+        except Exception:
+            logging.exception(
+                "raptor: failed to queue per-doc task for doc %s",
+                task_doc_id,
+            )
+
+    original_progress_cb = getattr(ctx, "_progress_cb", None)
+    if original_progress_cb is not None:
+        ctx._progress_cb = cap_done_progress(original_progress_cb)
+    try:
+        await asyncio.gather(
+            run_document_structure_compile(handler, embedding_model),
+            _maybe_run_raptor(),
+        )
+    finally:
+        if original_progress_cb is not None:
+            ctx._progress_cb = original_progress_cb
+        clear_doc_chunking_counter(task_doc_id)
+
+    if ctx.has_canceled_func(task_id):
+        abort_doc_chunking_counter(task_doc_id)
+        ctx.progress_cb(-1, msg="Task has been canceled.")
+        return False
+    return True

rag/svr/task_executor_refactor/dataset_skill_generator.py

@@ -0,0 +1,588 @@
+#
+#  Copyright 2026 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.
+#
+
+"""Corpus → Skill tree generator.
+
+Extracted from ``rag.svr.task_executor_refactor.task_handler`` where the
+same pipeline previously lived as a set of ``_skill_*`` methods and one
+``_corpus2skill`` orchestrator. The public entry point is
+:func:`run_corpus2skill`; the module-level helpers are internal but
+kept accessible so tests can exercise the individual phases.
+
+Design notes:
+
+* The pipeline is per-KB: given every parsed doc in a KB it produces a
+  hierarchical "skill" tree by summarizing each doc, RAPTOR-clustering
+  the summaries, summarizing each cluster, then repeating until the top
+  fan-out is at or below :data:`SKILL_MAX_TOP_CLUSTERS`.
+* Each node lands in ES twice: one per-node row under
+  ``compile_kwd="skill"`` carrying markdown metadata + a leaf-vs-branch
+  contents section, and one aggregate row under
+  ``compile_kwd="skill_all"`` holding the whole recursive tree as JSON
+  for cheap sidebar reads.
+* The layout mirrors Corpus2Skill's ``SKILL.md`` / ``INDEX.md`` naming
+  so a future on-disk export is a straight projection.
+
+The extraction keeps the callable surface minimal:
+
+    run_corpus2skill(ctx, embedding_model, load_chunks_for_doc)
+
+``load_chunks_for_doc`` is injected rather than imported to keep the
+module decoupled from ``TaskHandler``'s streaming chunk loader — any
+async iterator that yields batches of ``{content_with_weight: str, ...}``
+dicts will do.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import AsyncIterator, Callable, Dict, Optional
+
+import numpy as np
+import xxhash
+
+from common import settings
+from common.constants import LLMType
+from common.misc_utils import thread_pool_exec
+from common.token_utils import num_tokens_from_string
+from rag.nlp import search
+from rag.svr.task_executor_refactor.task_context import TaskContext
+
+
+# ----- tunables ------------------------------------------------------
+# Stop folding clusters once we've boiled the KB down to ≤ this many
+# top-level nodes. Mirrors Corpus2Skill's default top-of-tree fan-out.
+SKILL_MAX_TOP_CLUSTERS = 8
+# Per-doc summary is built from a budget of this fraction of the chat
+# model's context window. Stops adding chunks once cumulative tokens
+# hit the cap.
+SKILL_DOC_BUDGET_FRACTION = 0.5
+# Concurrency caps for the two LLM-bound stages.
+SKILL_DOC_SUMMARY_CONCURRENCY = 8
+SKILL_LABEL_CONCURRENCY = 10
+# Defensive cap on the clustering loop — a degenerate clustering that
+# keeps returning N clusters for N inputs would otherwise loop forever.
+SKILL_MAX_TREE_ITERATIONS = 12
+# Page size for the streaming chunk reader used during per-doc
+# summarization.
+SKILL_CHUNK_BATCH = 64
+
+
+# A ``load_chunks_for_doc`` callable takes ``(tenant_id, kb_id, doc_id,
+# batch_size)`` and returns an async iterator of chunk-batch lists.
+ChunkLoader = Callable[
+    [str, str, str],  # positional: tenant_id, kb_id, doc_id
+    AsyncIterator[list[dict]],
+]
+
+
+@dataclass
+class SkillNode:
+    """One node in the corpus → skill hierarchy.
+
+    Leaves wrap a single doc; branch nodes wrap a cluster of child
+    nodes plus the LLM summary of their summaries. ``doc_ids`` is the
+    flattened leaf set under this node, so any branch can quickly
+    report ``num_documents``. ``doc_texts`` carries first-page
+    previews keyed by ``doc_id`` only for leaves whose markdown nav
+    file needs the first-line title — branches inherit it as a no-op
+    union so the dict is non-empty across the tree.
+    """
+
+    level: int
+    label: str
+    summary: str
+    vec: np.ndarray
+    doc_ids: list[str]
+    doc_texts: dict[str, str] = field(default_factory=dict)
+    children: list["SkillNode"] = field(default_factory=list)
+    folder_name: str = ""
+
+
+# ----- helpers -------------------------------------------------------
+
+
+def skill_safe_name(text: str, max_len: int = 50) -> str:
+    """Lowercase, hyphen-only, max-len-clamped slug. Mirrors
+    Corpus2Skill ``_safe_name`` for cross-system stability of folder
+    names."""
+    name = (text or "").lower().strip()
+    name = re.sub(r"[^a-z0-9\s-]", "", name)
+    name = re.sub(r"\s+", "-", name)
+    name = name.strip("-")[:max_len]
+    return name
+
+
+async def label_skill_node_one(
+    summary: str,
+    chat_mdl,
+    semaphore: asyncio.Semaphore,
+) -> str:
+    """Generate a single fs-safe label: 2–5 word lowercase
+    hyphenated label, max_tokens=20, sanitized to [a-z0-9-], capped
+    at 50 chars, falls back to "cluster" on any failure.
+    """
+    async with semaphore:
+        try:
+            cnt = await chat_mdl.async_chat(
+                "You generate short filesystem-safe cluster labels. Reply with the label only.",
+                [
+                    {
+                        "role": "user",
+                        "content": (
+                            "Generate a short (2-5 word) filesystem-safe label for this cluster. "
+                            "Use lowercase(MUST be in the same language as 'Summary'), hyphens instead of spaces. No quotes.\n\n"
+                            f"Summary: {(summary or '')[:500]}"
+                        ),
+                    }
+                ],
+                {"max_tokens": 20, "temperature": 0.0},
+            )
+            raw = (cnt or "").strip().lower()
+            label = re.sub(r"[^a-z0-9-]", "-", raw)
+            label = re.sub(r"-+", "-", label).strip("-")[:50]
+            return label or "cluster"
+        except Exception:
+            logging.exception("skill: label generation failed; using fallback")
+            return "cluster"
+
+
+async def doc_summary_for_skill(
+    doc_id: str,
+    raptor,
+    chat_mdl,
+    ctx: TaskContext,
+    load_chunks_for_doc: Callable[..., AsyncIterator[list[dict]]],
+) -> Optional["SkillNode"]:
+    """Concatenate chunks up to half the chat model's context budget,
+    then summarize via RAPTOR's ``_summarize_texts`` (which also
+    returns the embedding). Returns a leaf-shaped :class:`SkillNode`
+    or ``None`` if the doc has no usable chunks."""
+    max_ctx = int(getattr(chat_mdl, "max_length", 4096) or 4096)
+    budget = max(512, int(max_ctx * SKILL_DOC_BUDGET_FRACTION))
+
+    accumulated: list[str] = []
+    running = 0
+    async for batch in load_chunks_for_doc(
+        ctx.tenant_id,
+        ctx.kb_id,
+        doc_id,
+        batch_size=SKILL_CHUNK_BATCH,
+    ):
+        for chunk in batch:
+            text = chunk.get("content_with_weight") or ""
+            if not isinstance(text, str) or not text:
+                continue
+            t_tokens = num_tokens_from_string(text)
+            if running + t_tokens > budget and accumulated:
+                break
+            accumulated.append(text)
+            running += t_tokens
+        else:
+            continue
+        break
+
+    if not accumulated:
+        return None
+
+    result = await raptor._summarize_texts(accumulated, callback=None, task_id="")
+    if result is None:
+        return None
+    title, summary_text, vec = result
+
+    doc_preview = accumulated[0][:600] if accumulated else ""
+    return SkillNode(
+        level=0,
+        label="",  # filled in phase 5
+        summary=summary_text or title,
+        vec=np.asarray(vec),
+        doc_ids=[doc_id],
+        doc_texts={doc_id: doc_preview},
+        children=[],
+        folder_name="",  # filled in phase 5
+    )
+
+
+def build_skill_md(node: "SkillNode") -> str:
+    """SKILL.md (depth 0) / INDEX.md (deeper) text. Mirrors
+    Corpus2Skill's ``_format_skill_md`` (skill_builder.py:193): YAML
+    frontmatter (name / description / level / num_documents), then
+    ``## Overview`` with the full summary, then ``## Contents`` —
+    sub-groups for branches, ``- `doc_id`: <first 120 chars>`` for
+    leaves.
+    """
+    depth = node.level
+    name = node.folder_name or node.label or f"cluster-{depth}"
+    desc = (node.summary or "")[:300].replace("\n", " ").strip()
+    lines: list[str] = [
+        "---",
+        f"name: {name}",
+        "description: >",
+        f"  {desc}",
+        f"level: {depth}",
+        f"num_documents: {len(node.doc_ids)}",
+        "---",
+        "",
+        "## Overview",
+        "",
+        (node.summary or "").strip() or "(no summary)",
+        "",
+        "## Contents",
+        "",
+    ]
+    if node.children:
+        lines.append("### Sub-groups (directories)")
+        lines.append("")
+        for child in node.children:
+            child_name = child.folder_name or child.label or "cluster"
+            summary_snip = (child.summary or "")[:200].replace("\n", " ").strip()
+            lines.append(f"- **{child_name}/** ({len(child.doc_ids)} docs): {summary_snip}")
+        lines.append("")
+    else:
+        lines.append(f"### Documents ({len(node.doc_ids)} items)")
+        lines.append("")
+        for doc_id in node.doc_ids:
+            preview = node.doc_texts.get(doc_id, "")
+            first_line = (preview.split("\n", 1)[0] if preview else "").strip()[:120]
+            lines.append(f"- `{doc_id}`: {first_line}")
+        lines.append("")
+    return "\n".join(lines)
+
+
+def skill_node_es_row(ctx: TaskContext, node: "SkillNode") -> Dict:
+    """Build the ES row for one tree node. Stable id from
+    (kb_id, folder_name) so re-runs upsert cleanly."""
+    kb_id_str = str(ctx.kb_id)
+    row_id = xxhash.xxh64(
+        f"skill:{kb_id_str}:{node.folder_name}".encode("utf-8", "surrogatepass"),
+    ).hexdigest()
+    return {
+        "id": row_id,
+        "kb_id": kb_id_str,
+        "doc_id": kb_id_str,  # KB-scoped sentinel
+        "compile_kwd": "skill",
+        "skill_kwd": node.folder_name,
+        "depth_int": int(node.level),
+        "children_kwd": [c.folder_name for c in node.children],
+        "source_doc_ids": list(node.doc_ids),
+        "md_with_weight": build_skill_md(node),
+        "available_int": 1,
+    }
+
+
+def skill_tree_md_snippet(node: "SkillNode") -> str:
+    """Return only the frontmatter/preamble before the Overview body.
+
+    The one-shot tree browser needs enough metadata to render the skill
+    directory without loading every full node body up front.
+    """
+    md = build_skill_md(node)
+    return md.split("\n## Overview", 1)[0].strip()
+
+
+def skill_tree_node(node: "SkillNode") -> Dict:
+    return {
+        "skill_kwd": node.folder_name,
+        "md_with_weight": skill_tree_md_snippet(node),
+        "children_kwd": [skill_tree_node(child) for child in node.children],
+    }
+
+
+def skill_all_es_row(ctx: TaskContext, roots: list["SkillNode"]) -> Dict:
+    """Build the aggregate tree row loaded by the Skills sidebar."""
+    kb_id_str = str(ctx.kb_id)
+    row_id = xxhash.xxh64(
+        f"skill_all:{kb_id_str}".encode("utf-8", "surrogatepass"),
+    ).hexdigest()
+    return {
+        "id": row_id,
+        "kb_id": kb_id_str,
+        "doc_id": kb_id_str,
+        "compile_kwd": "skill_all",
+        "skill_with_weight": json.dumps(
+            [skill_tree_node(root) for root in roots],
+            ensure_ascii=False,
+            indent=2,
+        ),
+        "available_int": 1,
+    }
+
+
+# ----- main entry ----------------------------------------------------
+
+
+async def run_corpus2skill(
+    ctx: TaskContext,
+    embedding_model,
+    load_chunks_for_doc: Callable[..., AsyncIterator[list[dict]]],
+) -> None:
+    """Build a hierarchical skill tree for the current KB and persist
+    one ES row per node under ``compile_kwd="skill"`` plus a full
+    recursive aggregate row under ``compile_kwd="skill_all"``.
+
+    Always-rebuild semantics for v1: every parsed doc in the KB is
+    re-summarized on each call. (Incremental "only changed docs" is a
+    TODO — needs a per-doc content-hash similar to MAP's
+    ``chunk_hash_kwd``.)
+    """
+    # Local imports so the module doesn't drag in the API service layer
+    # at import time — that's a source of circular-import risk given how
+    # much lives under ``api.db.services``.
+    from api.db.services.document_service import DocumentService
+    from api.db.services.llm_service import LLMBundle
+    from api.db.joint_services.tenant_model_service import (
+        get_tenant_default_model_by_type,
+    )
+    from rag.advanced_rag.knowlege_compile.raptor import (
+        RecursiveAbstractiveProcessing4TreeOrganizedRetrieval as Raptor,
+    )
+
+    progress = ctx.progress_cb
+    progress(0.0, "skill: loading documents")
+
+    # ---- Phase 0: chat model + RAPTOR instance for summarization/clustering.
+    chat_model_config = get_tenant_default_model_by_type(ctx.tenant_id, LLMType.CHAT)
+    chat_mdl = LLMBundle(ctx.tenant_id, chat_model_config, lang=ctx.language)
+
+    raptor = Raptor(
+        max_cluster=128,
+        llm_model=chat_mdl,
+        embd_model=embedding_model,
+        prompt="Please write a concise summary of the following texts:\n{cluster_content}",
+        max_token=256,
+        threshold=0.1,
+        max_errors=3,
+    )
+
+    # ---- Phase 1: per-doc summaries.
+    all_docs, _ = await thread_pool_exec(
+        DocumentService.get_by_kb_id,
+        kb_id=ctx.kb_id,
+        page_number=0,
+        items_per_page=0,
+        orderby="create_time",
+        desc=False,
+        keywords="",
+        run_status=[],
+        types=[],
+        suffix=[],
+    )
+    eligible_docs = [d for d in (all_docs or []) if d.get("id")]
+    if not eligible_docs:
+        progress(1.0, "skill: no documents in KB")
+        return
+
+    # Phase-1 gate: bail before spinning up N per-doc summarizations.
+    if ctx.has_canceled_func(ctx.id):
+        progress(-1, "skill: task has been canceled")
+        return
+
+    n_docs = len(eligible_docs)
+    progress(0.05, f"skill: summarizing {n_docs} document(s)")
+    doc_sem = asyncio.Semaphore(SKILL_DOC_SUMMARY_CONCURRENCY)
+
+    async def _summarize_doc(d: Dict) -> Optional[SkillNode]:
+        async with doc_sem:
+            try:
+                return await doc_summary_for_skill(
+                    d["id"],
+                    raptor,
+                    chat_mdl,
+                    ctx,
+                    load_chunks_for_doc,
+                )
+            except Exception:
+                logging.exception(
+                    "skill: doc summary failed for doc=%s",
+                    d.get("id"),
+                )
+                return None
+
+    leaf_results = await asyncio.gather(
+        *(_summarize_doc(d) for d in eligible_docs),
+        return_exceptions=False,
+    )
+    leaves: list[SkillNode] = [n for n in leaf_results if n is not None]
+    if not leaves:
+        progress(1.0, "skill: no doc summaries produced")
+        return
+
+    # Post-Phase-1 gate: bail before starting the iterative clustering.
+    if ctx.has_canceled_func(ctx.id):
+        progress(-1, "skill: task has been canceled")
+        return
+
+    # ---- Phase 2-4: iterative clustering until ≤ MAX_TOP.
+    current_layer = leaves
+    level = 0
+    for iteration in range(SKILL_MAX_TREE_ITERATIONS):
+        # Per-iteration gate: caps the wasted LLM cost when the task is
+        # canceled mid-way through a many-layer clustering run.
+        if ctx.has_canceled_func(ctx.id):
+            progress(-1, "skill: task has been canceled")
+            return
+        if len(current_layer) <= SKILL_MAX_TOP_CLUSTERS:
+            break
+        progress(
+            0.3 + 0.4 * iteration / SKILL_MAX_TREE_ITERATIONS,
+            f"skill: clustering layer {level} ({len(current_layer)} nodes)",
+        )
+        try:
+            embeddings = np.asarray([n.vec for n in current_layer])
+            n_clusters, labels = raptor.clustering(
+                embeddings,
+                random_state=0,
+                task_id="",
+            )
+        except Exception:
+            logging.exception("skill: clustering failed at level %d", level)
+            break
+        if n_clusters <= 0 or n_clusters >= len(current_layer):
+            # No reduction → stop to avoid an infinite loop.
+            logging.warning(
+                "skill: clustering did not reduce node count (%d → %d); stopping",
+                len(current_layer),
+                n_clusters,
+            )
+            break
+
+        cluster_buckets: dict[int, list[SkillNode]] = {}
+        for idx, lbl in enumerate(labels):
+            cluster_buckets.setdefault(int(lbl), []).append(current_layer[idx])
+
+        async def _summarize_cluster(children: list[SkillNode]) -> Optional[SkillNode]:
+            texts = [c.summary for c in children if c.summary]
+            if not texts:
+                return None
+            try:
+                res = await raptor._summarize_texts(texts, callback=None, task_id="")
+            except Exception:
+                logging.exception("skill: cluster summary failed")
+                return None
+            if res is None:
+                return None
+            title, summary_text, vec = res
+            merged_doc_ids: list[str] = []
+            seen: set[str] = set()
+            merged_doc_texts: dict[str, str] = {}
+            for c in children:
+                for did in c.doc_ids:
+                    if did and did not in seen:
+                        seen.add(did)
+                        merged_doc_ids.append(did)
+                merged_doc_texts.update(c.doc_texts)
+            return SkillNode(
+                level=level + 1,
+                label="",
+                summary=summary_text or title,
+                vec=np.asarray(vec),
+                doc_ids=merged_doc_ids,
+                doc_texts=merged_doc_texts,
+                children=list(children),
+                folder_name="",
+            )
+
+        parent_results = await asyncio.gather(
+            *(_summarize_cluster(children) for children in cluster_buckets.values()),
+            return_exceptions=False,
+        )
+        next_layer = [p for p in parent_results if p is not None]
+        if not next_layer:
+            logging.warning("skill: no cluster summaries produced at level %d", level)
+            break
+        current_layer = next_layer
+        level += 1
+
+    roots: list[SkillNode] = current_layer
+
+    # ---- Phase 5: label every node (concurrent), then assign folders.
+    all_nodes: list[SkillNode] = []
+
+    def _collect(node: SkillNode) -> None:
+        all_nodes.append(node)
+        for c in node.children:
+            _collect(c)
+
+    for r in roots:
+        _collect(r)
+
+    progress(0.75, f"skill: labelling {len(all_nodes)} cluster node(s)")
+    label_sem = asyncio.Semaphore(SKILL_LABEL_CONCURRENCY)
+    labels_out = await asyncio.gather(
+        *(label_skill_node_one(n.summary, chat_mdl, label_sem) for n in all_nodes),
+        return_exceptions=False,
+    )
+    for n, lbl in zip(all_nodes, labels_out):
+        n.label = lbl or "cluster"
+
+    # Folder naming: roots get ``skill-NN-<label>``; deeper nodes get
+    # ``group-NN-<label>`` keyed by their position in the parent's
+    # children list (mirrors Corpus2Skill's skill_builder).
+    def _assign_folders(node: SkillNode, idx: int, is_root: bool) -> None:
+        slug = skill_safe_name(node.label)
+        prefix = f"skill-{idx:02d}" if is_root else f"group-{idx:02d}"
+        node.folder_name = f"{prefix}-{slug}" if slug else prefix
+        for ci, child in enumerate(node.children):
+            _assign_folders(child, ci, is_root=False)
+
+    for ri, root in enumerate(roots):
+        _assign_folders(root, ri, is_root=True)
+
+    # Final gate before the destructive delete + bulk insert. This is
+    # the most important check — without it a late cancel would still
+    # wipe the KB's existing ``skill``/``skill_all`` rows AND spend a
+    # full bulk-insert round-trip on data the caller no longer wants.
+    if ctx.has_canceled_func(ctx.id):
+        progress(-1, "skill: task has been canceled")
+        return
+
+    # ---- Phase 6: clean + bulk insert.
+    index = search.index_name(ctx.tenant_id)
+    try:
+        await thread_pool_exec(
+            settings.docStoreConn.delete,
+            {"compile_kwd": ["skill", "skill_all"]},
+            index,
+            ctx.kb_id,
+        )
+    except Exception:
+        logging.debug("skill: prior delete failed; relying on id-upsert")
+
+    rows = [skill_node_es_row(ctx, n) for n in all_nodes]
+    rows.append(skill_all_es_row(ctx, roots))
+    if not rows:
+        progress(1.0, "skill: nothing to persist")
+        return
+
+    try:
+        await thread_pool_exec(
+            settings.docStoreConn.insert,
+            rows,
+            index,
+            ctx.kb_id,
+        )
+    except Exception:
+        logging.exception("skill: bulk insert failed (rows=%d)", len(rows))
+        return
+
+    progress(
+        1.0,
+        f"skill: built {len(roots)} top-level skill(s), {len(all_nodes)} total node(s), {len(leaves)} doc(s)",
+    )

rag/svr/task_executor_refactor/dataset_wiki_generator.py

@@ -0,0 +1,805 @@
+#
+#  Copyright 2026 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.
+#
+
+"""KB-wide wiki / artifact compilation.
+
+Extracted from ``rag.svr.task_executor_refactor.task_handler`` where the
+same pipeline previously lived as a set of ``_wiki_*`` / ``_persist_wiki_*``
+methods and one ``_run_wiki`` orchestrator. The public entry point is
+:func:`run_wiki`.
+
+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.
+
+Design notes:
+
+* ``load_chunks_for_doc`` is injected rather than imported to keep the
+  module decoupled from ``TaskHandler``'s streaming chunk loader.
+* The eligibility loop resolves each doc's
+  ``parser_config.compilation_template_group_id`` to a template list
+  via ``CompilationTemplateGroupService.resolve_template_ids``. That
+  helper is duplicated as a small private function here so the module
+  stays free of a task_handler import (which would be circular).
+* The persistence helpers (``persist_wiki_pages_to_es`` etc.) are
+  exposed at module level for testing but are only called from
+  :func:`run_wiki` in production.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import AsyncIterator, Callable, Dict, List, Optional
+
+import xxhash
+
+from common import settings
+from common.constants import LLMType
+from common.misc_utils import thread_pool_exec
+from rag.nlp import search
+from rag.advanced_rag.knowlege_compile.wiki import (
+    wiki_map_from_chunks,
+    wiki_plan_from_reduction,
+    wiki_reduce_from_extracts,
+    wiki_refine_from_plan,
+)
+from rag.svr.task_executor_refactor.task_context import TaskContext
+
+
+# ----- tunables ------------------------------------------------------
+# Artifact-MAP tuning: how many chunks to feed per ``wiki_map_from_chunks``
+# invocation. The function does its own per-call resume-set load + ES
+# persist, so smaller batches mean more (small) ES round-trips but a flat
+# memory footprint. 64 keeps the resume-set re-reads cheap while leaving
+# room for the function's internal split_chunks packing to do real work.
+WIKI_MAP_BATCH_CHUNKS = 64
+
+# Per-node cap on ``source_chunk_ids`` carried by the canvas graph blob.
+# Pages can accumulate hundreds of source chunks; the graph response is
+# meant for fast canvas rendering, not full provenance audit, so we trim
+# each node's list. The full per-page list is still available on the
+# ``artifact_page`` row the UI deep-links into.
+WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE = 64
+
+# Title + comments stamped on every ``artifact_commit`` row produced by
+# :func:`run_wiki`'s regeneration path (as opposed to the dialog-edit
+# 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})"
+
+
+# ----- helpers -------------------------------------------------------
+
+
+def _parser_config_compilation_template_group_id(parser_config) -> str:
+    """Read the single template-group id from a doc's ``parser_config``.
+
+    Duplicated from ``task_handler`` so this module doesn't import from
+    it (avoids a circular import — ``task_handler`` imports this module
+    from its ``task_type == "artifact"`` dispatch branch).
+    """
+    if not isinstance(parser_config, dict):
+        return ""
+    gid = parser_config.get("compilation_template_group_id")
+    if isinstance(gid, str) and gid.strip():
+        return gid.strip()
+    ext = parser_config.get("ext")
+    if isinstance(ext, dict):
+        gid = ext.get("compilation_template_group_id")
+        if isinstance(gid, str) and gid.strip():
+            return gid.strip()
+    return ""
+
+
+def _parser_config_compilation_template_ids(parser_config, tenant_id: str) -> list[str]:
+    """Resolve a doc's ``parser_config`` to its compile-template ids by
+    looking up the configured group. Returns ``[]`` if the doc has no
+    group set or the group cannot be resolved."""
+    from api.db.services.compilation_template_group_service import (
+        CompilationTemplateGroupService,
+    )
+
+    group_id = _parser_config_compilation_template_group_id(parser_config)
+    if not group_id:
+        return []
+    return CompilationTemplateGroupService.resolve_template_ids(group_id, tenant_id)
+
+
+# ----- persistence ---------------------------------------------------
+
+
+async def persist_wiki_pages_to_es(
+    ctx: TaskContext,
+    pages: List[Dict],
+    embd_mdl,
+) -> None:
+    """Insert one ES row per generated artifact page using the
+    knowledge-compilation schema:
+
+      id                  xxh64(kb_id + ":" + slug)
+      compile_kwd         "artifact_page"
+      slug_kwd            page.slug
+      title_kwd           page.title
+      page_type_kwd       page.page_type
+      entity_names_kwd    page.entity_names
+      outlinks_kwd        page.outlinks
+      related_kb_pages_kwd page.related_kb_pages
+      source_chunk_ids    page.source_chunk_ids
+      source_doc_ids      page.source_doc_ids
+      kb_id               ctx.kb_id
+      content_with_weight rendered markdown
+      content_ltks /
+      content_sm_ltks     tokenize(content_md + summary)
+      q_<dim>_vec         embed(summary)
+
+    ``action`` is intentionally not stored — it's a planner artifact
+    and has no meaning post-write.
+    """
+    if not pages:
+        return
+
+    from rag.nlp import rag_tokenizer
+    from common.doc_store.doc_store_base import OrderByExpr
+    from api.db.services.file_commit_service import (
+        FileCommitService as WikiCommitService,
+    )
+
+    index = search.index_name(ctx.tenant_id)
+    kb_id_str = str(ctx.kb_id)
+
+    # Capture the prior rendered content for every slug we're about to
+    # overwrite, so the per-page commit row downstream has a real diff
+    # baseline. Single batch read by slug_kwd IN [...] — one round-trip
+    # regardless of page count. Failures here degrade gracefully.
+    target_slugs: list[str] = [(p.get("slug") or "").strip() for p in pages if isinstance(p.get("slug"), str) and p.get("slug")]
+    prior_by_slug: dict[str, str] = {}
+    if target_slugs:
+        try:
+            res = await thread_pool_exec(
+                settings.docStoreConn.search,
+                ["id", "slug_kwd", "content_with_weight"],
+                [],
+                {"compile_kwd": ["artifact_page"], "slug_kwd": list(target_slugs)},
+                [],
+                OrderByExpr(),
+                0,
+                max(len(target_slugs), 1),
+                index,
+                [ctx.kb_id],
+            )
+            field_map = settings.docStoreConn.get_fields(
+                res,
+                ["id", "slug_kwd", "content_with_weight"],
+            )
+            for row in (field_map or {}).values():
+                s = row.get("slug_kwd")
+                c = row.get("content_with_weight")
+                if isinstance(s, str) and isinstance(c, str):
+                    prior_by_slug[s] = c
+        except Exception:
+            logging.exception(
+                "wiki_persist: prior-content read failed for kb=%s; commit audit will treat all pages as creations",
+                kb_id_str,
+            )
+
+    # Batch the summary embeddings in one model call. Empty strings are
+    # swapped for a single space so the encoder doesn't reject them —
+    # they still yield a vector but contribute nothing meaningful.
+    summaries = [(p.get("summary") or "").strip() for p in pages]
+    embed_inputs = [s if s else " " for s in summaries]
+    try:
+        embeddings, _ = await thread_pool_exec(embd_mdl.encode, embed_inputs)
+    except Exception:
+        logging.exception(
+            "wiki_persist: summary embedding batch failed for kb=%s",
+            kb_id_str,
+        )
+        return
+    try:
+        n_emb = len(embeddings) if embeddings is not None else 0
+    except TypeError:
+        n_emb = 0
+    if n_emb != len(pages):
+        logging.warning(
+            "artifact_persist: embedding count %d != pages %d for kb=%s; aborting",
+            n_emb,
+            len(pages),
+            kb_id_str,
+        )
+        return
+
+    rows: List[Dict] = []
+    for page, vec in zip(pages, embeddings):
+        slug = page.get("slug") or ""
+        if not slug:
+            continue
+        title = page.get("title") or slug
+        summary = page.get("summary") or ""
+        content_md = page.get("content_md_rendered") or page.get("content_md") or page.get("content_md_raw") or ""
+
+        vec_list = vec.tolist() if hasattr(vec, "tolist") else list(vec)
+        if not vec_list:
+            logging.warning(
+                "artifact_persist: empty embedding for slug=%s; skipping",
+                slug,
+            )
+            continue
+
+        text_for_search = (content_md + "\n\n" + summary).strip()
+        content_ltks = rag_tokenizer.tokenize(text_for_search) if text_for_search else ""
+        content_sm_ltks = rag_tokenizer.fine_grained_tokenize(content_ltks) if content_ltks else ""
+
+        row_id = xxhash.xxh64(
+            f"{kb_id_str}:{slug}".encode("utf-8", "surrogatepass"),
+        ).hexdigest()
+
+        rows.append(
+            {
+                "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",
+                "slug_kwd": slug,
+                "title_kwd": title,
+                "page_type_kwd": page.get("page_type") or "concept",
+                "entity_names_kwd": list(page.get("entity_names") or []),
+                "outlinks_kwd": list(page.get("outlinks") or []),
+                "outlinks_int": len(list(page.get("outlinks") or [])),
+                "related_kb_pages_kwd": list(page.get("related_kb_pages") or []),
+                "source_chunk_ids": list(page.get("source_chunk_ids") or []),
+                "source_doc_ids": list(page.get("source_doc_ids") or []),
+                "content_with_weight": content_md,
+                # Summary kept verbatim alongside the rendered body so the
+                # viewer can render it as a distinct (smaller) block above
+                # the main content.
+                "summary_with_weight": summary,
+                "content_ltks": content_ltks,
+                "content_sm_ltks": content_sm_ltks,
+                f"q_{len(vec_list)}_vec": vec_list,
+                "available_int": 1,
+            }
+        )
+
+    if not rows:
+        return
+
+    try:
+        await thread_pool_exec(settings.docStoreConn.insert, rows, index, ctx.kb_id)
+    except Exception:
+        logging.exception(
+            "wiki_persist: bulk insert failed for kb=%s (rows=%d)",
+            kb_id_str,
+            len(rows),
+        )
+        return
+
+    # 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
+    # compile.
+    for page in pages:
+        slug = (page.get("slug") or "").strip()
+        if not slug:
+            continue
+        if not prior_by_slug.get(slug, ""):
+            continue
+        content_md = page.get("content_md_rendered") or page.get("content_md") or page.get("content_md_raw") or ""
+        action = (page.get("action") or "CREATE").upper()
+        try:
+            WikiCommitService.record_page_edit(
+                tenant_id=ctx.tenant_id,
+                kb_id=ctx.kb_id,
+                page_type=page.get("page_type") or "concept",
+                slug=slug,
+                content_before=prior_by_slug.get(slug, ""),
+                content_after=content_md,
+                title=WIKI_REGEN_COMMIT_TITLE,
+                comments=WIKI_REGEN_COMMIT_COMMENTS_TEMPLATE.format(action=action),
+                user_id=None,  # system commit — no human author
+            )
+        except Exception:
+            logging.exception(
+                "wiki_persist: commit record failed for kb=%s slug=%s",
+                kb_id_str,
+                slug,
+            )
+
+
+def build_wiki_page_graph(
+    pages: List[Dict],
+    kb_id: str,
+) -> tuple[List[Dict], List[Dict]]:
+    """Project the REFINE-emitted page list onto per-entity and
+    per-relation ES rows.
+
+    Returns:
+        (entity_rows, relation_rows)
+
+    Both lists are ES-ready docs (one per node / per surviving edge)
+    using the standard artifact envelope. They are BM25-only (no
+    ``q_<dim>_vec``) — entities carry ``content_ltks`` derived from
+    ``slug + " " + summary`` so name/summary lookups hit the lexical
+    index.
+
+    ``by_slug`` is kept internally only to filter dangling outlinks
+    (a target slug not present as a node in this KB).
+    """
+    from rag.nlp import rag_tokenizer
+
+    by_slug: Dict[str, Dict] = {}
+    entity_rows: List[Dict] = []
+    for p in pages or []:
+        slug = (p.get("slug") or "").strip()
+        if not slug:
+            continue
+        outlinks_raw = p.get("outlinks") or []
+        # ``weight`` is the page's outlink count. Drives node size on
+        # the canvas. Computed on the raw outlink list before dangling-
+        # target filtering so visual weight reflects what the writer
+        # actually emitted.
+        weight = len(outlinks_raw) if isinstance(outlinks_raw, list) else 0
+
+        # Per-node provenance: union of source chunks REFINE attributed
+        # to this page. Dedup preserves first-seen order; cap the list
+        # to keep the graph blob small.
+        raw_chunk_ids = p.get("source_chunk_ids") or []
+        seen_chunk_ids: dict[str, None] = {}
+        for cid in raw_chunk_ids:
+            if isinstance(cid, str) and cid and cid not in seen_chunk_ids:
+                seen_chunk_ids[cid] = None
+                if len(seen_chunk_ids) >= WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE:
+                    break
+        capped_chunk_ids = list(seen_chunk_ids.keys())
+
+        page_type = p.get("page_type") or "concept"
+        description = p.get("summary") or ""
+        name = p.get("title") or slug
+        aliases = list(p.get("entity_names") or [])
+
+        by_slug[slug] = {
+            "slug": slug,
+            "name": name,
+            "aliases": aliases,
+            "description": description,
+            "type": page_type,
+            "weight": weight,
+            "source_chunk_ids": capped_chunk_ids,
+        }
+
+        # Per-entity ES row. content_ltks is built from slug + summary
+        # so BM25 hits both the deep-link key and human prose.
+        content_text = (slug + " " + description).strip()
+        entity_payload = {
+            "slug": slug,
+            "name": name,
+            "aliases": aliases,
+            "description": description,
+            "type": page_type,
+            "weight": weight,
+        }
+        entity_rows.append(
+            {
+                "id": xxhash.xxh64(
+                    f"artifact_entity:{kb_id}:{slug}".encode("utf-8", "surrogatepass"),
+                ).hexdigest(),
+                "kb_id": kb_id,
+                "doc_id": kb_id,  # KB-scoped sentinel
+                "available_int": 1,
+                "compile_kwd": "artifact_entity",
+                "type_kwd": "artifact_" + page_type,
+                "slug_kwd": slug,
+                "weight_int": int(weight),
+                "source_chunk_ids": capped_chunk_ids,
+                "content_ltks": rag_tokenizer.tokenize(content_text) if content_text else "",
+                "content_with_weight": json.dumps(entity_payload, ensure_ascii=False),
+            }
+        )
+
+    relation_rows: List[Dict] = []
+    for p in pages or []:
+        src = (p.get("slug") or "").strip()
+        if not src or src not in by_slug:
+            continue
+        for raw_target in p.get("outlinks") or []:
+            if isinstance(raw_target, str):
+                tgt = raw_target.strip()
+            elif isinstance(raw_target, dict):
+                tgt = str(raw_target.get("slug") or "").strip()
+            else:
+                tgt = ""
+            if not tgt or tgt == src or tgt not in by_slug:
+                continue
+            relation_payload = {"from": src, "to": tgt}
+            relation_rows.append(
+                {
+                    "id": xxhash.xxh64(
+                        f"artifact_relation:{kb_id}:{src}:{tgt}".encode("utf-8", "surrogatepass"),
+                    ).hexdigest(),
+                    "kb_id": kb_id,
+                    "doc_id": kb_id,
+                    "available_int": 1,
+                    "compile_kwd": "artifact_relation",
+                    "type_kwd": "artifact_relation",
+                    "from_kwd": src,
+                    "to_kwd": tgt,
+                    "content_with_weight": json.dumps(relation_payload, ensure_ascii=False),
+                }
+            )
+
+    return entity_rows, relation_rows
+
+
+async def persist_wiki_page_graph_to_es(
+    ctx: TaskContext,
+    pages: List[Dict],
+) -> None:
+    """Materialize and store the per-entity / per-relation ES rows
+    derived from artifact pages.
+
+    Writes two row types — both delete-then-insert for idempotent
+    re-runs:
+
+    1. ``compile_kwd="artifact_entity"`` — one row per page node,
+       BM25-only via ``content_ltks``.
+    2. ``compile_kwd="artifact_relation"`` — one row per surviving
+       edge (dangling outlinks dropped by the builder).
+
+    Also sweeps any leftover legacy ``artifact_page_graph`` blob so
+    the index doesn't accumulate stale state.
+    """
+    kb_id_str = str(ctx.kb_id)
+    entity_rows, relation_rows = build_wiki_page_graph(pages or [], kb_id_str)
+
+    index = search.index_name(ctx.tenant_id)
+
+    async def _replace_bucket(kwd: str, rows: List[Dict]) -> None:
+        try:
+            await thread_pool_exec(
+                settings.docStoreConn.delete,
+                {"compile_kwd": kwd},
+                index,
+                ctx.kb_id,
+            )
+        except Exception:
+            logging.debug(
+                "%s: prior delete failed; relying on id-upsert",
+                kwd,
+            )
+        if not rows:
+            return
+        try:
+            await thread_pool_exec(
+                settings.docStoreConn.insert,
+                rows,
+                index,
+                ctx.kb_id,
+            )
+        except Exception:
+            logging.exception(
+                "%s: insert failed for kb=%s (%d rows)",
+                kwd,
+                kb_id_str,
+                len(rows),
+            )
+
+    async def _sweep_legacy_blob() -> None:
+        try:
+            await thread_pool_exec(
+                settings.docStoreConn.delete,
+                {"compile_kwd": "artifact_page_graph"},
+                index,
+                ctx.kb_id,
+            )
+        except Exception:
+            logging.debug(
+                "artifact_page_graph: legacy blob sweep failed for kb=%s",
+                kb_id_str,
+            )
+
+    await asyncio.gather(
+        _replace_bucket("artifact_entity", entity_rows),
+        _replace_bucket("artifact_relation", relation_rows),
+        _sweep_legacy_blob(),
+    )
+
+
+# ----- main entry ----------------------------------------------------
+
+
+async def run_wiki(
+    ctx: TaskContext,
+    embedding_model,
+    load_chunks_for_doc: Callable[..., AsyncIterator[list[dict]]],
+) -> None:
+    """KB-wide artifact compilation task.
+
+    Runs after the user clicks the "Artifact" button in the dataset
+    generate menu. Iterates every doc in the KB whose parser_config
+    has a compilation template group resolving to an artifacts-kind
+    child, runs MAP per-doc (which uses ES-stored resume rows to skip
+    chunks already processed in a previous run), then runs REDUCE /
+    PLAN / REFINE KB-wide and persists pages.
+
+    Batching: each MAP call uses ``batch_size_cap=8`` and
+    ``window_fraction=0.5`` — i.e. roll over to a new batch when the
+    current batch reaches 8 chunks OR its accumulated token count
+    exceeds 50% of the chat model's ``max_length``.
+    """
+    # Local imports so this module doesn't drag in the API service
+    # layer at import time — that's a source of circular-import risk
+    # given how much lives under ``api.db.services``.
+    from api.db.services.document_service import DocumentService
+    from api.db.services.knowledgebase_service import KnowledgebaseService
+    from api.db.services.compilation_template_service import CompilationTemplateService
+    from api.db.services.llm_service import LLMBundle
+    from api.db.joint_services.tenant_model_service import (
+        get_tenant_default_model_by_type,
+        get_model_config_from_provider_instance,
+    )
+    from api.apps.restful_apis.chunk_api import _compilation_template_kind
+
+    progress = ctx.progress_cb
+    progress(0.0, "Loading documents for wiki compilation...")
+
+    # 1. Resolve KB metadata for PLAN.
+    ok, kb = KnowledgebaseService.get_by_id(ctx.kb_id)
+    if not ok:
+        progress(-1, f"KB {ctx.kb_id} not found.")
+        return
+    kb_name = kb.name
+    kb_description = kb.description
+
+    # 2. Pick docs eligible for artifact compilation (those whose
+    # configured template group resolves to at least one artifacts-kind
+    # child). The frontend Artifact button targets the KB, but the
+    # per-doc opt-in is what gates inclusion.
+    all_docs, _ = await thread_pool_exec(
+        DocumentService.get_by_kb_id,
+        kb_id=ctx.kb_id,
+        page_number=0,
+        items_per_page=0,
+        orderby="create_time",
+        desc=False,
+        keywords="",
+        run_status=[],
+        types=[],
+        suffix=[],
+    )
+    eligible = []
+    for d in all_docs or []:
+        pc = d.get("parser_config") or {}
+        for template_id in _parser_config_compilation_template_ids(pc, ctx.tenant_id):
+            template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id)
+            config = template.get("config") if template else {}
+            kind = _compilation_template_kind(config.get("kind") if isinstance(config, dict) else "")
+            if kind == "artifacts":
+                eligible.append((d, template_id))
+                break
+    if not eligible:
+        progress(1.0, "No documents are configured for wiki compilation.")
+        return
+
+    # 3. Resolve chat models. MAP is per-(doc, template) so each pair
+    # uses its template's own ``llm_id``. REDUCE / PLAN / REFINE are
+    # KB-wide and need exactly one model — we pick the first eligible
+    # template's ``llm_id`` as the canonical KB chat model.
+    llm_bundle_cache: dict[str, LLMBundle] = {}
+
+    def _bundle_for(llm_id: str | None) -> LLMBundle:
+        key = (llm_id or "").strip() or "__tenant_default__"
+        cached = llm_bundle_cache.get(key)
+        if cached is not None:
+            return cached
+        try:
+            if key == "__tenant_default__":
+                cfg = get_tenant_default_model_by_type(ctx.tenant_id, LLMType.CHAT)
+            else:
+                cfg = get_model_config_from_provider_instance(
+                    ctx.tenant_id,
+                    LLMType.CHAT,
+                    key,
+                )
+        except Exception:
+            logging.exception(
+                "wiki: chat model resolution failed for llm_id=%s (kb=%s); falling back to tenant default",
+                key,
+                ctx.kb_id,
+            )
+            cfg = get_tenant_default_model_by_type(ctx.tenant_id, LLMType.CHAT)
+            key = "__tenant_default__"
+            cached = llm_bundle_cache.get(key)
+            if cached is not None:
+                return cached
+        bundle = LLMBundle(ctx.tenant_id, cfg, lang=ctx.language)
+        llm_bundle_cache[key] = bundle
+        return bundle
+
+    def _stage_cb(prefix: str):
+        def _cb(*args, **kwargs):
+            try:
+                if args and isinstance(args[0], (int, float)):
+                    msg = args[1] if len(args) > 1 else kwargs.get("msg", "")
+                    progress(msg=f"{prefix} {msg}")
+                else:
+                    msg = kwargs.get("msg") or (args[0] if args else "")
+                    progress(msg=f"{prefix} {msg}")
+            except Exception:
+                logging.exception("wiki: progress callback failed")
+
+        return _cb
+
+    # 4. MAP per eligible doc. Each MAP call's own resume mechanism
+    # (artifact_map_extract rows keyed by chunk_id) skips chunks that
+    # were already processed in a prior run — this is the incremental
+    # behavior the user asked for.
+    #
+    # ``kb_chat_llm_id`` is captured from the first eligible template
+    # and used as the canonical chat model for REDUCE/PLAN/REFINE.
+    # ``kb_writer_example`` follows the same first-template-wins rule:
+    # the REFINE writer's page-structure section is pulled from the
+    # first eligible artifact template's ``parser_config.example``
+    # override (None falls back to the built-in ``WIKI_TEMPLATE_EXAMPLE``).
+    kb_chat_llm_id: Optional[str] = None
+    kb_writer_example: Optional[str] = None
+    n_docs = len(eligible)
+    for i, (doc, template_id) in enumerate(eligible):
+        doc_id = doc["id"]
+        progress(
+            0.05 + 0.6 * (i / n_docs),
+            f"MAP {i + 1}/{n_docs}: {doc.get('name', doc_id)}",
+        )
+
+        template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id)
+        if not template:
+            logging.warning(
+                "artifact: template %s not found for doc %s; skipping",
+                template_id,
+                doc_id,
+            )
+            continue
+        parser_cfg = template.get("config") or {}
+
+        map_llm_id = (parser_cfg.get("llm_id") or "").strip() if isinstance(parser_cfg, dict) else ""
+        map_chat_mdl = _bundle_for(map_llm_id)
+        if kb_chat_llm_id is None:
+            # First eligible template wins — canonical for KB-wide
+            # REDUCE/PLAN/REFINE further down.
+            kb_chat_llm_id = map_llm_id or None
+            tmpl_example = parser_cfg.get("example") if isinstance(parser_cfg, dict) else None
+            if isinstance(tmpl_example, str) and tmpl_example.strip():
+                kb_writer_example = tmpl_example
+
+        # Stream the doc's chunks in batches and call MAP per batch so
+        # peak memory stays bounded for long docs.
+        agg = {"entities": 0, "concepts": 0, "claims": 0, "relations": 0}
+        agg_delta = {"new": 0, "changed": 0, "unchanged": 0, "deleted": 0}
+        doc_had_delta = False
+        saw_any = False
+        batch_no = 0
+        async for batch in load_chunks_for_doc(
+            ctx.tenant_id,
+            ctx.kb_id,
+            doc_id,
+            batch_size=WIKI_MAP_BATCH_CHUNKS,
+        ):
+            saw_any = True
+            batch_no += 1
+            try:
+                phase1 = await wiki_map_from_chunks(
+                    chunks=batch,
+                    chat_mdl=map_chat_mdl,
+                    embd_mdl=embedding_model,
+                    doc_id=doc_id,
+                    tenant_id=ctx.tenant_id,
+                    kb_id=ctx.kb_id,
+                    language=ctx.language,
+                    callback=_stage_cb(f"[wiki MAP {i + 1}/{n_docs} b{batch_no}]"),
+                    parser_config=parser_cfg,
+                    batch_size_cap=8,
+                    window_fraction=0.5,
+                )
+            except Exception:
+                logging.exception(
+                    "wiki: MAP failed for doc %s batch %d",
+                    doc_id,
+                    batch_no,
+                )
+                continue
+            for k in agg.keys():
+                agg[k] += len(phase1.get(k) or [])
+            meta = phase1.get("_meta") or {}
+            if isinstance(meta, dict):
+                for k in agg_delta.keys():
+                    agg_delta[k] += int(meta.get(k, 0) or 0)
+                if meta.get("had_delta"):
+                    doc_had_delta = True
+
+        if not saw_any:
+            logging.info("wiki: no chunks for doc %s; skipping", doc_id)
+            continue
+        logging.info(
+            "wiki: MAP doc=%s entities=%d concepts=%d claims=%d relations=%d (batches=%d, new=%d changed=%d unchanged=%d deleted=%d, delta=%s)",
+            doc_id,
+            agg["entities"],
+            agg["concepts"],
+            agg["claims"],
+            agg["relations"],
+            batch_no,
+            agg_delta["new"],
+            agg_delta["changed"],
+            agg_delta["unchanged"],
+            agg_delta["deleted"],
+            doc_had_delta,
+        )
+
+    # 5. REDUCE / PLAN / REFINE KB-wide. Each phase has its own
+    # input_hash gate (REDUCE keys off the MAP-state hash, PLAN off
+    # REDUCE's hash, REFINE off PLAN's hash) so re-runs without an
+    # upstream delta short-circuit at the cache layer.
+    kb_chat_mdl = _bundle_for(kb_chat_llm_id)
+    try:
+        progress(0.65, "Reducing extracts KB-wide...")
+        await wiki_reduce_from_extracts(
+            chat_mdl=kb_chat_mdl,
+            embd_mdl=embedding_model,
+            tenant_id=ctx.tenant_id,
+            kb_id=ctx.kb_id,
+            callback=_stage_cb("[wiki REDUCE]"),
+        )
+
+        progress(0.75, "Planning wiki pages...")
+        await wiki_plan_from_reduction(
+            chat_mdl=kb_chat_mdl,
+            embd_mdl=embedding_model,
+            tenant_id=ctx.tenant_id,
+            kb_id=ctx.kb_id,
+            kb_name=kb_name,
+            kb_description=kb_description,
+            callback=_stage_cb("[wiki PLAN]"),
+        )
+
+        progress(0.85, "Refining pages...")
+        pages = await wiki_refine_from_plan(
+            chat_mdl=kb_chat_mdl,
+            embd_mdl=embedding_model,
+            tenant_id=ctx.tenant_id,
+            kb_id=ctx.kb_id,
+            callback=_stage_cb("[wiki REFINE]"),
+            example=kb_writer_example,
+        )
+    except Exception:
+        logging.exception("wiki: REDUCE/PLAN/REFINE failed for kb %s", ctx.kb_id)
+        progress(-1, "Wiki pipeline failed during REDUCE/PLAN/REFINE.")
+        return
+
+    # 6. Persist searchable artifact_page rows.
+    try:
+        await persist_wiki_pages_to_es(ctx, pages or [], embedding_model)
+    except Exception:
+        logging.exception("wiki: ES persist failed for kb %s", ctx.kb_id)
+
+    # 7. Materialize the canvas graph from the refined pages.
+    try:
+        await persist_wiki_page_graph_to_es(ctx, pages or [])
+    except Exception:
+        logging.exception("wiki: page-graph persist failed for kb %s", ctx.kb_id)
+
+    progress(1.0, f"Wiki compiled {len(pages or [])} page(s).")

rag/svr/task_executor_refactor/raptor_service.py

@@ -21,12 +21,14 @@ Provides [`RaptorService`](rag/svr/task_executor_refactor/raptor_service.py:48)
 """
 
 import copy
+import json
 import logging
 import os
 from datetime import datetime
 from typing import Dict, List, Optional, Set, Tuple
 
 import numpy as np
+import xxhash
 
 from api.db.services.document_service import DocumentService
 from api.db.services.task_service import GRAPH_RAPTOR_FAKE_DOC_ID
@@ -48,6 +50,31 @@ from rag.utils.raptor_utils import (
 from rag.svr.task_executor_refactor.task_context import TaskContext
 
 
+def _sum_tree_text_tokens(tree) -> int:
+    """Count tokens across every ``title`` string in the RAPTOR tree.
+
+    Mirrors the legacy ``tk_count`` semantic (sum over summary texts)
+    so the orchestrator's downstream logging / billing keeps working
+    when the tree path replaces the per-summary rows. Walks the dict
+    iteratively to avoid recursion-limit issues on deep trees.
+    """
+    if not isinstance(tree, dict):
+        return 0
+    total = 0
+    stack = [tree]
+    while stack:
+        node = stack.pop()
+        if not isinstance(node, dict):
+            continue
+        title = node.get("title")
+        if isinstance(title, str) and title:
+            total += num_tokens_from_string(title)
+        children = node.get("children")
+        if isinstance(children, list):
+            stack.extend(children)
+    return total
+
+
 class RaptorService:
     """Service for RAPTOR summary generation.
 
@@ -106,15 +133,11 @@ class RaptorService:
         # Determine scope
         if raptor_config.get("scope", "file") == "file":
             res, tk_count = await self._run_file_level_raptor(
-                raptor_config, tree_builder, clustering_method,
-                chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id,
-                max_errors, res, tk_count, cleanup_raptor_chunks
+                raptor_config, tree_builder, clustering_method, chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id, max_errors, res, tk_count, cleanup_raptor_chunks
             )
         else:
             res, tk_count = await self._run_dataset_level_raptor(
-                raptor_config, tree_builder, clustering_method,
-                chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id,
-                max_errors, res, tk_count, cleanup_raptor_chunks
+                raptor_config, tree_builder, clustering_method, chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id, max_errors, res, tk_count, cleanup_raptor_chunks
             )
 
         return res, tk_count, cleanup_raptor_chunks
@@ -135,15 +158,11 @@ class RaptorService:
             }
         return doc_info_by_id
 
-    async def _run_file_level_raptor(
-        self, raptor_config, tree_builder, clustering_method,
-        chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id,
-        max_errors, res, tk_count, cleanup_raptor_chunks
-    ):
+    async def _run_file_level_raptor(self, raptor_config, tree_builder, clustering_method, chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id, max_errors, res, tk_count, cleanup_raptor_chunks):
         """Run RAPTOR at file level (per document)."""
         ctx = self._task_context
         fake_doc_id = GRAPH_RAPTOR_FAKE_DOC_ID
-        if self._task_context.write_interceptor: # dry run mode
+        if self._task_context.write_interceptor:  # dry run mode
             dataset_methods = set()
         else:
             dataset_methods = await self._get_raptor_chunk_methods(fake_doc_id, ctx.tenant_id, ctx.kb_id)
@@ -155,7 +174,7 @@ class RaptorService:
 
         for x, doc_id in enumerate(doc_ids):
             if self._should_skip_raptor(doc_id, doc_info_by_id, raptor_config):
-                self._task_context.progress_cb(prog=(x + 1.) / len(doc_ids))
+                self._task_context.progress_cb(prog=(x + 1.0) / len(doc_ids))
                 continue
             if self._task_context.write_interceptor:
                 existing_methods = set()
@@ -164,12 +183,10 @@ class RaptorService:
             if tree_builder in existing_methods:
                 has_file_level_target = True
                 if existing_methods != {tree_builder}:
-                    self._schedule_raptor_cleanup(
-                        doc_id, tree_builder, cleanup_raptor_chunks
-                    )
+                    self._schedule_raptor_cleanup(doc_id, tree_builder, cleanup_raptor_chunks)
                     self._task_context.progress_cb(msg=f"[RAPTOR] doc:{doc_id} will remove old RAPTOR summaries after insert.")
                 self._task_context.progress_cb(msg=f"[RAPTOR] doc:{doc_id} already has {tree_builder} RAPTOR chunks, skipping.")
-                self._task_context.progress_cb(prog=(x + 1.) / len(doc_ids))
+                self._task_context.progress_cb(prog=(x + 1.0) / len(doc_ids))
                 continue
 
             if existing_methods:
@@ -180,36 +197,25 @@ class RaptorService:
                 continue
 
             before_generate = len(res)
-            new_chunks, new_tk_count = await self._generate_raptor(
-                chunks, doc_id, raptor_config, chat_mdl, embd_mdl,
-                tree_builder, clustering_method, max_errors, doc_info_by_id
-            )
+            new_chunks, new_tk_count = await self._generate_raptor(chunks, doc_id, raptor_config, chat_mdl, embd_mdl, tree_builder, clustering_method, max_errors, doc_info_by_id)
             res.extend(new_chunks)
             tk_count += new_tk_count
 
             if len(res) > before_generate:
                 has_file_level_target = True
                 if existing_methods:
-                    self._schedule_raptor_cleanup(
-                        doc_id, tree_builder, cleanup_raptor_chunks
-                    )
-            self._task_context.progress_cb(prog=(x + 1.) / len(doc_ids))
+                    self._schedule_raptor_cleanup(doc_id, tree_builder, cleanup_raptor_chunks)
+            self._task_context.progress_cb(prog=(x + 1.0) / len(doc_ids))
 
         if remove_dataset_summaries:
             if has_file_level_target:
-                self._schedule_raptor_cleanup(
-                    fake_doc_id, None, cleanup_raptor_chunks
-                )
+                self._schedule_raptor_cleanup(fake_doc_id, None, cleanup_raptor_chunks)
             else:
                 self._task_context.progress_cb(msg="[RAPTOR] kept dataset-level summaries because no file-level summaries were built.")
 
         return res, tk_count
 
-    async def _run_dataset_level_raptor(
-        self, raptor_config, tree_builder, clustering_method,
-        chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id,
-        max_errors, res, tk_count, cleanup_raptor_chunks
-    ):
+    async def _run_dataset_level_raptor(self, raptor_config, tree_builder, clustering_method, chat_mdl, embd_mdl, vctr_nm, doc_ids, doc_info_by_id, max_errors, res, tk_count, cleanup_raptor_chunks):
         """Run RAPTOR at dataset level (all documents combined)."""
         ctx = self._task_context
         fake_doc_id = GRAPH_RAPTOR_FAKE_DOC_ID
@@ -230,9 +236,7 @@ class RaptorService:
                 migrated_file_docs += 1
 
         if migrated_file_docs:
-            self._task_context.progress_cb(
-                msg=f"[RAPTOR] will remove file-level summaries for {migrated_file_docs} docs after dataset-level build succeeds."
-            )
+            self._task_context.progress_cb(msg=f"[RAPTOR] will remove file-level summaries for {migrated_file_docs} docs after dataset-level build succeeds.")
 
         if self._task_context.write_interceptor:
             existing_methods = set()
@@ -240,9 +244,7 @@ class RaptorService:
             existing_methods = await self._get_raptor_chunk_methods(fake_doc_id, ctx.tenant_id, ctx.kb_id)
         if tree_builder in existing_methods:
             if existing_methods != {tree_builder}:
-                self._schedule_raptor_cleanup(
-                    fake_doc_id, tree_builder, cleanup_raptor_chunks
-                )
+                self._schedule_raptor_cleanup(fake_doc_id, tree_builder, cleanup_raptor_chunks)
                 self._task_context.progress_cb(msg="[RAPTOR] will remove old dataset-level RAPTOR summaries after insert.")
             for doc_id in file_cleanup_doc_ids:
                 self._schedule_raptor_cleanup(doc_id, None, cleanup_raptor_chunks)
@@ -262,10 +264,7 @@ class RaptorService:
             return res, tk_count
 
         before_generate = len(res)
-        new_chunks, new_tk_count = await self._generate_raptor(
-            chunks, fake_doc_id, raptor_config, chat_mdl, embd_mdl,
-            tree_builder, clustering_method, max_errors, doc_info_by_id
-        )
+        new_chunks, new_tk_count = await self._generate_raptor(chunks, fake_doc_id, raptor_config, chat_mdl, embd_mdl, tree_builder, clustering_method, max_errors, doc_info_by_id)
         res.extend(new_chunks)
         tk_count += new_tk_count
 
@@ -273,15 +272,11 @@ class RaptorService:
             for doc_id in file_cleanup_doc_ids:
                 self._schedule_raptor_cleanup(doc_id, None, cleanup_raptor_chunks)
             if migrate_dataset_summaries:
-                self._schedule_raptor_cleanup(
-                    fake_doc_id, tree_builder, cleanup_raptor_chunks
-                )
+                self._schedule_raptor_cleanup(fake_doc_id, tree_builder, cleanup_raptor_chunks)
 
         return res, tk_count
 
-    def _should_skip_raptor(
-        self, doc_id: str, doc_info_by_id: Dict, raptor_config: Dict
-    ) -> bool:
+    def _should_skip_raptor(self, doc_id: str, doc_info_by_id: Dict, raptor_config: Dict) -> bool:
         """Check if RAPTOR should be skipped for a document."""
         ctx = self._task_context
         doc_info = doc_info_by_id.get(doc_id, {})
@@ -297,67 +292,64 @@ class RaptorService:
             return True
         return False
 
-    def _load_doc_chunks(self, doc_id: str, vctr_nm: str) -> List[Tuple[str, np.ndarray]]:
-        """Load chunks for a single document."""
+    def _load_doc_chunks(self, doc_id: str, vctr_nm: str) -> List[Tuple[str, np.ndarray, str]]:
+        """Load chunks for a single document.
+
+        Returns ``(content, vector, chunk_id)`` triples so downstream
+        RAPTOR can attach ``source_chunk_ids`` provenance onto every
+        summary it produces. ``chunk_id`` may be an empty string if the
+        retriever didn't surface one — defensive against legacy rows.
+        """
         ctx = self._task_context
-        chunks = []
+        chunks: List[Tuple[str, np.ndarray, str]] = []
         skipped_chunks = 0
 
-        fields = ["content_with_weight", vctr_nm]
-        for d in settings.retriever.chunk_list(
-            doc_id, ctx.tenant_id, [str(ctx.kb_id)],
-            fields=fields,
-            sort_by_position=True
-        ):
+        # ``id`` is included so the source-chunk provenance survives
+        # through summarization; the retriever otherwise drops it when
+        # ``fields`` is provided.
+        fields = ["id", "content_with_weight", vctr_nm]
+        for d in settings.retriever.chunk_list(doc_id, ctx.tenant_id, [str(ctx.kb_id)], fields=fields, sort_by_position=True):
             if vctr_nm not in d or d[vctr_nm] is None:
                 skipped_chunks += 1
                 logging.warning(f"RAPTOR: Chunk missing vector field '{vctr_nm}' in doc {doc_id}, skipping")
                 continue
-            chunks.append((d["content_with_weight"], np.array(d[vctr_nm])))
+            chunks.append((d["content_with_weight"], np.array(d[vctr_nm]), str(d.get("id") or "")))
 
         if skipped_chunks > 0:
-            self._task_context.progress_cb(
-                msg=f"[WARN] Skipped {skipped_chunks} chunks without vector field '{vctr_nm}' for doc {doc_id}."
-            )
+            self._task_context.progress_cb(msg=f"[WARN] Skipped {skipped_chunks} chunks without vector field '{vctr_nm}' for doc {doc_id}.")
         if not chunks:
             logging.warning(f"RAPTOR: No valid chunks with vectors found for doc {doc_id}")
             self._task_context.progress_cb(msg=f"[WARN] No valid chunks with vectors found for doc {doc_id}, skipping")
 
         return chunks
 
-    def _load_all_doc_chunks(
-        self, doc_ids: List[str], vctr_nm: str, skipped_doc_ids: Set[str]
-    ) -> List[Tuple[str, np.ndarray]]:
-        """Load chunks for all documents."""
+    def _load_all_doc_chunks(self, doc_ids: List[str], vctr_nm: str, skipped_doc_ids: Set[str]) -> List[Tuple[str, np.ndarray, str]]:
+        """Load chunks for all documents — returns provenance-carrying
+        ``(content, vector, chunk_id)`` triples. See ``_load_doc_chunks``
+        for the per-doc variant."""
         ctx = self._task_context
-        chunks = []
+        chunks: List[Tuple[str, np.ndarray, str]] = []
         skipped_chunks = 0
 
-        fields = ["content_with_weight", vctr_nm]
+        fields = ["id", "content_with_weight", vctr_nm]
         for doc_id in doc_ids:
             if doc_id in skipped_doc_ids:
                 continue
-            for d in settings.retriever.chunk_list(
-                doc_id, ctx.tenant_id, [str(ctx.kb_id)],
-                fields=fields,
-                sort_by_position=True
-            ):
+            for d in settings.retriever.chunk_list(doc_id, ctx.tenant_id, [str(ctx.kb_id)], fields=fields, sort_by_position=True):
                 if vctr_nm not in d or d[vctr_nm] is None:
                     skipped_chunks += 1
                     logging.warning(f"RAPTOR: Chunk missing vector field '{vctr_nm}' in doc {doc_id}, skipping")
                     continue
-                chunks.append((d["content_with_weight"], np.array(d[vctr_nm])))
+                chunks.append((d["content_with_weight"], np.array(d[vctr_nm]), str(d.get("id") or "")))
 
         if skipped_chunks > 0:
-            self._task_context.progress_cb(
-                msg=f"[WARN] Skipped {skipped_chunks} chunks without vector field '{vctr_nm}'."
-            )
+            self._task_context.progress_cb(msg=f"[WARN] Skipped {skipped_chunks} chunks without vector field '{vctr_nm}'.")
 
         return chunks
 
     async def _generate_raptor(
         self,
-        chunks: List[Tuple[str, np.ndarray]],
+        chunks: List[Tuple[str, np.ndarray, str]],
         doc_id: str,
         raptor_config: Dict,
         chat_mdl,
@@ -366,10 +358,19 @@ class RaptorService:
         clustering_method: str,
         max_errors: int,
         doc_info_by_id: Dict,
+        is_tree: bool = False,
     ) -> Tuple[List[Dict], int]:
-        """Run RAPTOR and generate summary chunks."""
+        """Run RAPTOR and generate summary chunks.
+
+        ``chunks`` is the provenance-carrying triple shape produced by
+        ``_load_doc_chunks`` / ``_load_all_doc_chunks``:
+        ``(content, vector, chunk_id)``. Each leaf is wrapped into the
+        ``(text, vec, [chunk_id])`` shape RAPTOR expects so every
+        summary it produces carries the order-preserving deduped union
+        of the leaf ids underneath it.
+        """
         ctx = self._task_context
-        from rag.raptor import RecursiveAbstractiveProcessing4TreeOrganizedRetrieval as Raptor
+        from rag.advanced_rag.knowlege_compile.raptor import RecursiveAbstractiveProcessing4TreeOrganizedRetrieval as Raptor
 
         raptor_ext_config = raptor_config.get("ext") or {}
         assert chunks, "_generate_raptor must not be called with empty chunks"
@@ -389,13 +390,173 @@ class RaptorService:
             psi_bucket_size=raptor_ext_config.get("psi_bucket_size", 1024),
         )
 
-        original_length = len(chunks)
-        processed_chunks, layers = await raptor(
-            chunks, raptor_config["random_seed"], self._task_context.progress_cb, ctx.id
-        )
+        # Seed each leaf with its own id as the start of its
+        # ``source_chunk_ids`` provenance trail. The id may be empty
+        # for malformed retriever rows; ``Raptor.__call__`` filters
+        # those out of the union on the inbound normalize step.
+        raptor_input = [(content, vctr, [chunk_id] if chunk_id else []) for content, vctr, chunk_id in chunks]
 
         effective_doc_name = ctx.name if doc_id == GRAPH_RAPTOR_FAKE_DOC_ID else doc_info_by_id.get(doc_id, {}).get("name") or ctx.name
 
+        # Default path: ask RAPTOR for a single hierarchical tree dict
+        # and persist it as ONE non-searchable ES row. PSI's
+        # hyperedge-driven summarization can't form a strict
+        # parent-of relation, so __call__(is_tree=True) raises
+        # NotImplementedError there — catch and fall through to the
+        # legacy per-summary materialization below for that case.
+        original_length = len(chunks)
+        try:
+            processed_chunks, layers = await raptor(
+                raptor_input,
+                raptor_config["random_seed"],
+                self._task_context.progress_cb,
+                ctx.id,
+                is_tree=is_tree,
+            )
+        except NotImplementedError:
+            return await self._generate_raptor_legacy_rows(
+                raptor,
+                raptor_input,
+                raptor_config,
+                doc_id,
+                effective_doc_name,
+                tree_builder,
+                vctr_nm,
+            )
+
+        if processed_chunks is None:
+            return [], 0
+        doc = {
+            "doc_id": doc_id,
+            "kb_id": [str(ctx.kb_id)],
+            "docnm_kwd": effective_doc_name,
+            "title_tks": rag_tokenizer.tokenize(effective_doc_name),
+            "raptor_kwd": "raptor",
+            "extra": {"raptor_method": tree_builder},
+            "create_time": str(datetime.now()).replace("T", " ")[:19],
+            "create_timestamp_flt": datetime.now().timestamp(),
+        }
+        if ctx.pagerank:
+            doc[PAGERANK_FLD] = int(ctx.pagerank)
+
+        if not is_tree:
+            # Build index→layer mapping
+            chunk_layer = {}
+            for layer_idx, (layer_start, layer_end) in enumerate(layers):
+                if layer_idx == 0:
+                    continue
+                for ci in range(layer_start, layer_end):
+                    chunk_layer[ci] = layer_idx
+
+            res = []
+            tk_count = 0
+            for idx, (content, vctr, _, _) in enumerate(processed_chunks[original_length:], start=original_length):
+                d = copy.deepcopy(doc)
+                d["id"] = make_raptor_summary_chunk_id(content, doc_id)
+                d["create_time"] = str(datetime.now()).replace("T", " ")[:19]
+                d["create_timestamp_flt"] = datetime.now().timestamp()
+                d[vctr_nm] = vctr.tolist()
+                d["content_with_weight"] = content
+                d["content_ltks"] = rag_tokenizer.tokenize(content)
+                d["content_sm_ltks"] = rag_tokenizer.fine_grained_tokenize(d["content_ltks"])
+                d["raptor_layer_int"] = chunk_layer.get(idx, 1)
+                res.append(d)
+                tk_count += num_tokens_from_string(content)
+            return res, tk_count
+
+        row_id = xxhash.xxh64(
+            f"raptor_tree:{doc_id}:{tree_builder}".encode("utf-8", "surrogatepass"),
+        ).hexdigest()
+        row = {
+            **doc,
+            "id": row_id,
+            "raptor_kwd": "raptor_tree",
+            "content_with_weight": json.dumps(processed_chunks, ensure_ascii=False),
+            "available_int": 0,
+        }
+        return [row], _sum_tree_text_tokens(processed_chunks)
+
+    async def build_doc_tree(
+        self,
+        chunks: List[Tuple[str, np.ndarray, str]],
+        raptor_config: Dict,
+        chat_mdl,
+        embd_mdl,
+        tree_builder: str,
+        clustering_method: str,
+        max_errors: int,
+    ) -> Optional[Dict]:
+        """Build a RAPTOR tree dict for one document — no ES IO.
+
+        Used by the ``tree``-kind compilation template, which wraps the
+        returned tree into a per-template structure-graph row. Returns
+        None when the input has no chunks, the PSI builder is selected
+        (which can't form a strict tree), or RAPTOR itself fails.
+        """
+        if not chunks:
+            return None
+        from rag.advanced_rag.knowlege_compile.raptor import RecursiveAbstractiveProcessing4TreeOrganizedRetrieval as Raptor
+
+        raptor_ext_config = raptor_config.get("ext") or {}
+        raptor = Raptor(
+            raptor_config.get("max_cluster", 64),
+            chat_mdl,
+            embd_mdl,
+            raptor_config["prompt"],
+            raptor_config["max_token"],
+            raptor_config["threshold"],
+            max_errors=max_errors,
+            tree_builder=tree_builder,
+            clustering_method=clustering_method,
+            psi_exact_max_leaves=raptor_ext_config.get("psi_exact_max_leaves", 4096),
+            psi_bucket_size=raptor_ext_config.get("psi_bucket_size", 1024),
+        )
+
+        raptor_input = [(content, vctr, [chunk_id] if chunk_id else []) for content, vctr, chunk_id in chunks]
+        try:
+            tree, _ = await raptor(
+                raptor_input,
+                raptor_config["random_seed"],
+                self._task_context.progress_cb,
+                self._task_context.id,
+                is_tree=True,
+            )
+        except NotImplementedError:
+            # PSI builder — not supported in tree mode; surface as None
+            # so the compilation-template path can skip the doc cleanly.
+            logging.warning(
+                "build_doc_tree: PSI builder doesn't support is_tree; skipping",
+            )
+            return None
+        return tree if isinstance(tree, dict) else None
+
+    async def _generate_raptor_legacy_rows(
+        self,
+        raptor,
+        raptor_input,
+        raptor_config,
+        doc_id,
+        effective_doc_name,
+        tree_builder,
+        vctr_nm,
+    ) -> Tuple[List[Dict], int]:
+        """Legacy per-summary materialization, kept only for PSI builds.
+
+        PSI's hyperedge summaries don't map to a strict tree, so the
+        ``is_tree=True`` default in ``_generate_raptor`` raises and
+        falls through here. Same shape this function produced before
+        the tree migration — one ES row per appended summary, marked
+        ``raptor_kwd="raptor"``.
+        """
+        ctx = self._task_context
+        original_length = len(raptor_input)
+        processed_chunks, layers = await raptor(
+            raptor_input,
+            raptor_config["random_seed"],
+            self._task_context.progress_cb,
+            ctx.id,
+        )
+
         doc = {
             "doc_id": doc_id,
             "kb_id": [str(ctx.kb_id)],
@@ -407,7 +568,6 @@ class RaptorService:
         if ctx.pagerank:
             doc[PAGERANK_FLD] = int(ctx.pagerank)
 
-        # Build index→layer mapping
         chunk_layer = {}
         for layer_idx, (layer_start, layer_end) in enumerate(layers):
             if layer_idx == 0:
@@ -417,7 +577,12 @@ class RaptorService:
 
         res = []
         tk_count = 0
-        for idx, (content, vctr) in enumerate(processed_chunks[original_length:], start=original_length):
+        for idx, item in enumerate(processed_chunks[original_length:], start=original_length):
+            if len(item) >= 3:
+                content, vctr, source_chunk_ids = item[0], item[1], item[2] or []
+            else:
+                content, vctr = item[0], item[1]
+                source_chunk_ids = []
             d = copy.deepcopy(doc)
             d["id"] = make_raptor_summary_chunk_id(content, doc_id)
             d["create_time"] = str(datetime.now()).replace("T", " ")[:19]
@@ -427,6 +592,8 @@ class RaptorService:
             d["content_ltks"] = rag_tokenizer.tokenize(content)
             d["content_sm_ltks"] = rag_tokenizer.fine_grained_tokenize(d["content_ltks"])
             d["raptor_layer_int"] = chunk_layer.get(idx, 1)
+            if source_chunk_ids:
+                d["source_chunk_ids"] = list(source_chunk_ids)
             res.append(d)
             tk_count += num_tokens_from_string(content)
 
@@ -445,16 +612,17 @@ class RaptorService:
         from common.doc_store.doc_store_base import OrderByExpr
 
         async def search_fields(fields: list, condition: dict, order_by=None):
-            res = await thread_pool_exec(
-                settings.docStoreConn.search,
-                fields, [], condition, [], order_by or OrderByExpr(),
-                0, 10000, search.index_name(tenant_id), [kb_id]
-            )
+            res = await thread_pool_exec(settings.docStoreConn.search, fields, [], condition, [], order_by or OrderByExpr(), 0, 10000, search.index_name(tenant_id), [kb_id])
             return settings.docStoreConn.get_fields(res, fields)
 
         try:
+            # Accept both ``raptor`` (legacy per-summary rows, PSI
+            # builder still produces these) and ``raptor_tree`` (new
+            # single-row tree blob) so existing-method detection stays
+            # accurate across the migration.
             primary = await search_fields(
-                ["raptor_kwd", "extra"], {"doc_id": doc_id, "raptor_kwd": ["raptor"]}
+                ["raptor_kwd", "extra"],
+                {"doc_id": doc_id, "raptor_kwd": ["raptor", "raptor_tree"]},
             )
             if collect_raptor_chunk_ids(primary):
                 return collect_raptor_methods(primary)
@@ -469,3 +637,188 @@ class RaptorService:
         except Exception:
             logging.exception("Failed to check RAPTOR chunks for doc %s", doc_id)
             raise
+
+    @staticmethod
+    def _build_raptor_graph(rows: List[Dict]) -> Dict:
+        """Project loaded RAPTOR summary rows onto the canvas graph shape.
+
+        Each row contributes one entity::
+
+            {
+              "id":          xxh128(content)           # 32-char hex
+              "name":        first 16 whitespace tokens
+              "description": content_with_weight
+              "source_chunk_ids": row.source_chunk_ids
+            }
+
+        Relations: full bipartite layer-by-layer fan-out — every node at
+        layer K gets an edge to every node at layer K-1 (because we only
+        loaded ``content_with_weight`` + ``raptor_layer_int`` we don't
+        have the specific parent linkage). Self-edges and dangling
+        targets are dropped (the latter only matters if the layer-int
+        values are non-contiguous).
+        """
+        # Build entities. Dedup by id so two identical-content summaries
+        # collapse to one node — the canvas can't render multiple nodes
+        # at the same id anyway, and identical content is a defensible
+        # collapse.
+        by_id: Dict[str, Dict] = {}
+        by_layer: Dict[int, List[str]] = {}
+
+        for row in rows:
+            content = row.get("content_with_weight")
+            if not isinstance(content, str) or not content.strip():
+                continue
+            try:
+                layer = int(row.get("raptor_layer_int") or 0)
+            except (TypeError, ValueError):
+                layer = 0
+            if layer <= 0:
+                # Layer 0 would be the original leaf chunks; RAPTOR
+                # summaries start at layer 1. Anything claiming layer 0
+                # here is malformed; skip.
+                continue
+
+            name = " ".join(content.split()[:16])
+            nid = xxhash.xxh128(
+                content.encode("utf-8", "surrogatepass"),
+            ).hexdigest()  # 32-char hex
+            if nid in by_id:
+                continue
+            source_chunk_ids = row.get("source_chunk_ids") or []
+            if not isinstance(source_chunk_ids, list):
+                source_chunk_ids = []
+            by_id[nid] = {
+                "id": nid,
+                "name": name,
+                "description": content,
+                "source_chunk_ids": list(source_chunk_ids),
+            }
+            by_layer.setdefault(layer, []).append(nid)
+
+        # Layered fan-out from parent (higher layer) → child (lower layer).
+        relations: List[Dict] = []
+        layers_sorted = sorted(by_layer.keys())
+        for layer in layers_sorted:
+            child_layer = layer - 1
+            if child_layer not in by_layer:
+                continue
+            for parent in by_layer[layer]:
+                for child in by_layer[child_layer]:
+                    if parent == child:
+                        continue
+                    relations.append({"from": parent, "to": child})
+
+        return {"entities": list(by_id.values()), "relations": relations}
+
+    async def _persist_raptor_graph_to_es(self, doc_id: str) -> None:
+        """Load the just-inserted RAPTOR summaries for ``doc_id`` and
+        persist a single graph row that the dataset structure-graph
+        endpoint can surface as a tree.
+
+        Loads only ``content_with_weight`` + ``raptor_layer_int`` +
+        ``source_chunk_ids`` (per
+        the smallest-payload contract) and writes one row with::
+
+            compile_kwd:                  "raptor_graph"
+            compilation_template_kind_kwd:"raptor"
+            doc_id:                       <doc_id>
+
+        The row id is deterministic per ``(kb_id, doc_id)`` so re-runs
+        delete-and-replace cleanly through the same primary key.
+        ``knowledge_graph_kwd`` is intentionally NOT set — that field
+        belongs to the KG feature; this row is identified via
+        ``compile_kwd`` so the two paths stay semantically distinct.
+        """
+        from common.doc_store.doc_store_base import OrderByExpr
+
+        ctx = self._task_context
+        tenant_id = ctx.tenant_id
+        kb_id_str = str(ctx.kb_id)
+        index_nm = search.index_name(tenant_id)
+        select_fields = ["content_with_weight", "raptor_layer_int", "source_chunk_ids"]
+        try:
+            res = await thread_pool_exec(
+                settings.docStoreConn.search,
+                select_fields,
+                [],
+                {"raptor_kwd": ["raptor"], "doc_id": [doc_id]},
+                [],
+                OrderByExpr(),
+                0,
+                10000,
+                index_nm,
+                [kb_id_str],
+            )
+            field_map = settings.docStoreConn.get_fields(res, select_fields)
+        except Exception:
+            logging.exception(
+                "raptor_graph: load failed for kb=%s doc=%s",
+                kb_id_str,
+                doc_id,
+            )
+            return
+
+        rows = list((field_map or {}).values())
+        if not rows:
+            logging.info(
+                "raptor_graph: no summaries to render for kb=%s doc=%s",
+                kb_id_str,
+                doc_id,
+            )
+            return
+
+        graph = self._build_raptor_graph(rows)
+        if not graph["entities"]:
+            logging.info(
+                "raptor_graph: projection produced no entities for kb=%s doc=%s",
+                kb_id_str,
+                doc_id,
+            )
+            return
+
+        row_id = xxhash.xxh64(
+            f"raptor_graph:{kb_id_str}:{doc_id}".encode("utf-8", "surrogatepass"),
+        ).hexdigest()
+        row = {
+            "id": row_id,
+            "kb_id": kb_id_str,
+            "doc_id": doc_id,
+            "compile_kwd": "raptor_graph",
+            "compilation_template_kind_kwd": "raptor",
+            "content_with_weight": json.dumps(graph, ensure_ascii=False),
+            "available_int": 0,
+        }
+        try:
+            await thread_pool_exec(
+                settings.docStoreConn.delete,
+                {"compile_kwd": "raptor_graph", "doc_id": [doc_id]},
+                index_nm,
+                ctx.kb_id,
+            )
+        except Exception:
+            logging.debug(
+                "raptor_graph: prior delete failed for kb=%s doc=%s; relying on id-upsert",
+                kb_id_str,
+                doc_id,
+            )
+        try:
+            await thread_pool_exec(
+                settings.docStoreConn.insert,
+                [row],
+                index_nm,
+                ctx.kb_id,
+            )
+            logging.info(
+                "raptor_graph: stored %d entities / %d relations for kb=%s doc=%s",
+                len(graph["entities"]),
+                len(graph["relations"]),
+                kb_id_str,
+                doc_id,
+            )
+        except Exception:
+            logging.exception(
+                "raptor_graph: insert failed for kb=%s doc=%s",
+                kb_id_str,
+                doc_id,
+            )

rag/svr/task_executor_refactor/raptor_utils.py

@@ -38,11 +38,7 @@ async def get_raptor_chunk_field_map(doc_id: str, tenant_id: str, kb_id: str) ->
 
     async def search_fields(fields: list[str], condition: dict, order_by=None):
         """Search chunk fields in the current knowledge base."""
-        res = await thread_pool_exec(
-            settings.docStoreConn.search,
-            fields, [], condition, [], order_by or OrderByExpr(),
-            0, RAPTOR_METHOD_SEARCH_LIMIT, nlp_search.index_name(tenant_id), [kb_id]
-        )
+        res = await thread_pool_exec(settings.docStoreConn.search, fields, [], condition, [], order_by or OrderByExpr(), 0, RAPTOR_METHOD_SEARCH_LIMIT, nlp_search.index_name(tenant_id), [kb_id])
         return settings.docStoreConn.get_fields(res, fields)
 
     primary = await search_fields(["raptor_kwd", "extra"], {"doc_id": doc_id, "raptor_kwd": ["raptor"]})
@@ -65,11 +61,17 @@ async def delete_raptor_chunks(doc_id: str, tenant_id: str, kb_id: str, keep_met
     if keep_method is None:
         logging.info(
             "delete_raptor_chunks: removing all RAPTOR summaries (doc=%s tenant=%s kb=%s)",
-            doc_id, tenant_id, kb_id,
+            doc_id,
+            tenant_id,
+            kb_id,
         )
+        # Sweep both row types — legacy per-summary (``raptor``, still
+        # used by the PSI builder) and the new single tree blob
+        # (``raptor_tree``) — so re-runs always start from a clean
+        # slate regardless of which path produced the prior state.
         await thread_pool_exec(
             settings.docStoreConn.delete,
-            {"doc_id": doc_id, "raptor_kwd": ["raptor"]},
+            {"doc_id": doc_id, "raptor_kwd": ["raptor", "raptor_tree"]},
             nlp_search.index_name(tenant_id),
             kb_id,
         )
@@ -80,13 +82,20 @@ async def delete_raptor_chunks(doc_id: str, tenant_id: str, kb_id: str, keep_met
     if not chunk_ids:
         logging.debug(
             "delete_raptor_chunks: no stale RAPTOR chunks to remove (doc=%s tenant=%s kb=%s keep=%s)",
-            doc_id, tenant_id, kb_id, keep_method,
+            doc_id,
+            tenant_id,
+            kb_id,
+            keep_method,
         )
         return 0
 
     logging.info(
         "delete_raptor_chunks: removing %d stale RAPTOR chunks (doc=%s tenant=%s kb=%s keep=%s)",
-        len(chunk_ids), doc_id, tenant_id, kb_id, keep_method,
+        len(chunk_ids),
+        doc_id,
+        tenant_id,
+        kb_id,
+        keep_method,
     )
     await thread_pool_exec(
         settings.docStoreConn.delete,

rag/svr/task_executor_refactor/task_handler.py

@@ -23,20 +23,26 @@ for handling document processing tasks with refactored, testable methods.
 import asyncio
 import logging
 import json
+
+# Wiki / artifact compilation pipeline lives in
+# ``rag.svr.task_executor_refactor.dataset_wiki_generator`` — see the
+# ``task_type == "artifact"`` branch of ``TaskHandler.run`` for the
+# dispatch call.
+# Document-structure compilation helpers (CHAIN_KINDS,
+# compile_structure_from_text, merge_compiled_structures,
+# validate_and_correct_chain) moved to ``chunk_post_processor``.
 import xxhash
 
 from timeit import default_timer as timer
-from typing import Callable, Dict, List, Optional
+from typing import AsyncIterator, Callable, Dict, List, Optional
 
 from api.db.services.document_service import DocumentService
 from api.db.services.knowledgebase_service import KnowledgebaseService
+from api.db.services.compilation_template_group_service import CompilationTemplateGroupService
 from api.db.joint_services.memory_message_service import handle_save_to_memory_task
-from api.db.joint_services.tenant_model_service import (
-    get_tenant_default_model_by_type,
-    get_model_config_from_provider_instance
-)
+from api.db.joint_services.tenant_model_service import get_tenant_default_model_by_type, get_model_config_from_provider_instance
 from api.db.services.llm_service import LLMBundle
-from api.db.services.task_service import GRAPH_RAPTOR_FAKE_DOC_ID
+from api.db.services.task_service import GRAPH_RAPTOR_FAKE_DOC_ID, abort_doc_chunking_counter
 from common.constants import LLMType
 from common.exceptions import TaskCanceledException
 from common.connection_utils import timeout
@@ -57,6 +63,96 @@ from rag.prompts.generator import run_toc_from_text
 from common import settings
 
 
+def _parser_config_compilation_template_group_ids(parser_config) -> list[str]:
+    """Read template-group ids from a doc's parser_config.
+
+    Templates were previously referenced as a list
+    (``compilation_template_ids``); after the template-group refactor
+    a doc instead points at one or more groups, and the orchestrator
+    resolves each group's child templates at runtime. Old
+    ``compilation_template_ids`` data is intentionally ignored per
+    the migration spec.
+    """
+
+    def _normalize(raw) -> list[str]:
+        if isinstance(raw, str):
+            raw = [raw]
+        if not isinstance(raw, list):
+            return []
+        ids: list[str] = []
+        seen: set[str] = set()
+        for gid in raw:
+            if not isinstance(gid, str):
+                continue
+            gid = gid.strip()
+            if gid and gid not in seen:
+                seen.add(gid)
+                ids.append(gid)
+        return ids
+
+    if not isinstance(parser_config, dict):
+        return []
+    if "compilation_template_group_id" in parser_config:
+        return _normalize(parser_config.get("compilation_template_group_id"))
+    ext = parser_config.get("ext")
+    if isinstance(ext, dict):
+        return _normalize(ext.get("compilation_template_group_id"))
+    return []
+
+
+def _parser_config_compilation_template_ids(parser_config, tenant_id: str) -> list[str]:
+    """Resolve a doc's parser_config to compile-template ids by
+    looking up configured groups. Returns ``[]`` if the doc has no
+    group set or no group can be resolved.
+    """
+    template_ids: list[str] = []
+    seen: set[str] = set()
+    for group_id in _parser_config_compilation_template_group_ids(parser_config):
+        for template_id in CompilationTemplateGroupService.resolve_template_ids(group_id, tenant_id):
+            if template_id in seen:
+                continue
+            seen.add(template_id)
+            template_ids.append(template_id)
+    return template_ids
+
+
+def _resolve_template_chat_llm_id(parser_cfg: dict, ctx) -> str:
+    """Pick the chat model id for a knowledge-compilation template.
+
+    Resolution order:
+      1. The template's own ``llm_id`` (what the user picked in the
+         compilation-template panel).
+      2. The doc's ``parser_config.llm_id`` (the doc-level chunking
+         model).
+      3. ``ctx.llm_id`` (the chunking task's default).
+    """
+    if isinstance(parser_cfg, dict):
+        tid = parser_cfg.get("llm_id")
+        if isinstance(tid, str) and tid.strip():
+            return tid.strip()
+    doc_cfg = getattr(ctx, "parser_config", None) or {}
+    if isinstance(doc_cfg, dict):
+        did = doc_cfg.get("llm_id")
+        if isinstance(did, str) and did.strip():
+            return did.strip()
+    return ctx.llm_id
+
+
+# Document-structure compilation tunables
+# (DOC_STRUCTURE_COMPILE_BATCH_CHUNKS, DOC_STRUCTURE_MERGE_MAX_DOCS,
+# STRUCTURE_CHAIN_CORRECTION_TIMEOUT_S) moved to
+# ``chunk_post_processor``.
+
+# Wiki / artifact tunables (``WIKI_MAP_BATCH_CHUNKS``,
+# ``WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE``, commit-title / comments
+# templates) moved to ``dataset_wiki_generator``.
+
+# The corpus → skill compilation pipeline lives in
+# ``rag.svr.task_executor_refactor.dataset_skill_generator``. Its entry
+# point is :func:`run_corpus2skill`; this handler invokes it from the
+# ``task_type == "skill"`` branch of ``run`` below.
+
+
 class TaskHandler:
     """Main task handler for document processing.
 
@@ -85,32 +181,52 @@ class TaskHandler:
         self._task_context = ctx
         self._billing_hook = billing_hook
 
+    @staticmethod
+    def _is_standard_chunking_task(task_type: str) -> bool:
+        task_type = (task_type or "").lower()
+        return task_type not in {
+            "memory",
+            "raptor",
+            "graphrag",
+            "mindmap",
+            "artifact",
+            "skill",
+            "evaluation",
+            "reembedding",
+            "clone",
+        } and not task_type.startswith("dataflow")
+
     async def handle_task(self) -> None:
         try:
             await self.handle()
+        except Exception:
+            if self._is_standard_chunking_task(self._task_context.task_type):
+                abort_doc_chunking_counter(self._task_context.doc_id)
+            raise
         finally:
             task_id = self._task_context.id
             task_tenant_id = self._task_context.tenant_id
             task_dataset_id = self._task_context.kb_id
             task_doc_id = self._task_context.doc_id
             if self._task_context.has_canceled_func(task_id):
-                try:
-                    exists = await thread_pool_exec(
-                        settings.docStoreConn.index_exist,
-                        search.index_name(task_tenant_id),
-                        task_dataset_id,
-                    )
-                    if exists:
-                        ret = await thread_pool_exec(
-                            settings.docStoreConn.delete,
-                            {"doc_id": task_doc_id},
+                if self._is_standard_chunking_task(self._task_context.task_type):
+                    abort_doc_chunking_counter(task_doc_id)
+                    try:
+                        exists = await thread_pool_exec(
+                            settings.docStoreConn.index_exist,
                             search.index_name(task_tenant_id),
                             task_dataset_id,
                         )
-                        self._task_context.recording_context.save_func_return_value("docStoreConn.delete", ret)
-                except Exception as e:
-                    logging.exception(
-                        f"Remove doc({task_doc_id}) from docStore failed when task({task_id}) canceled, exception: {e}")
+                        if exists:
+                            ret = await thread_pool_exec(
+                                settings.docStoreConn.delete,
+                                {"doc_id": task_doc_id},
+                                search.index_name(task_tenant_id),
+                                task_dataset_id,
+                            )
+                            self._task_context.recording_context.save_func_return_value("docStoreConn.delete", ret)
+                    except Exception as e:
+                        logging.exception(f"Remove doc({task_doc_id}) from docStore failed when task({task_id}) canceled, exception: {e}")
 
     @timeout(60 * 60 * 3, 1)
     async def handle(self) -> None:
@@ -134,7 +250,7 @@ class TaskHandler:
             ctx.progress_cb(-1, msg="Task has been canceled.")
             return
 
-        # Language defaults to "Chinese" via TaskContext._DEFAULTS — safe to bind model directly.
+        # Language defaults to "Chinese" via TaskContext._DEFAULTS 鈥?safe to bind model directly.
         # Bind embedding model (matching original do_handle_task order: bind + init_kb before routing)
         result = await self._bind_embedding_model()
         if result is None:
@@ -154,12 +270,30 @@ class TaskHandler:
                 return
 
             # Route to appropriate handler
-            if task_type == "raptor":
-                await self._run_raptor(embedding_model, vector_size)
-            elif task_type == "graphrag":
+            if task_type == "graphrag":
                 await self._run_graphrag(embedding_model)
             elif task_type == "mindmap":
                 ctx.progress_cb(1, "place holder")
+            elif task_type == "artifact":
+                from rag.svr.task_executor_refactor.dataset_wiki_generator import (
+                    run_wiki,
+                )
+
+                await run_wiki(
+                    self._task_context,
+                    embedding_model,
+                    self._load_chunks_for_doc,
+                )
+            elif task_type == "skill":
+                from rag.svr.task_executor_refactor.dataset_skill_generator import (
+                    run_corpus2skill,
+                )
+
+                await run_corpus2skill(
+                    self._task_context,
+                    embedding_model,
+                    self._load_chunks_for_doc,
+                )
             elif task_type == "evaluation":
                 await self._run_evaluation()
             elif task_type == "reembedding":
@@ -167,8 +301,7 @@ class TaskHandler:
             elif task_type == "clone":
                 await self._run_clone()
             else:
-                await self._run_standard_chunking(embedding_model)
-
+                await self._run_standard_chunking(embedding_model, vector_size)
 
     def _init_kb(self, vector_size: int) -> None:
         """Initialize knowledge base index."""
@@ -214,18 +347,14 @@ class TaskHandler:
 
         try:
             if task_embedding_id:
-                embd_model_config = get_model_config_from_provider_instance(
-                    task_tenant_id, LLMType.EMBEDDING, task_embedding_id
-                )
+                embd_model_config = get_model_config_from_provider_instance(task_tenant_id, LLMType.EMBEDDING, task_embedding_id)
             else:
-                embd_model_config = get_tenant_default_model_by_type(
-                    task_tenant_id, LLMType.EMBEDDING
-                )
+                embd_model_config = get_tenant_default_model_by_type(task_tenant_id, LLMType.EMBEDDING)
             embedding_model = LLMBundle(task_tenant_id, embd_model_config, lang=task_language)
             vts, _ = embedding_model.encode(["ok"])
             return embedding_model, len(vts[0])
         except Exception as e:
-            error_message = f'Fail to bind embedding model: {str(e)}'
+            error_message = f"Fail to bind embedding model: {str(e)}"
             ctx.progress_cb(-1, msg=error_message)
             logging.exception(error_message)
             raise
@@ -234,6 +363,7 @@ class TaskHandler:
         self,
         embedding_model: LLMBundle,
         vector_size: int,
+        mark_done: bool = True,
     ) -> None:
         """Run RAPTOR summary generation."""
         ctx = self._task_context
@@ -248,19 +378,21 @@ class TaskHandler:
 
         kb_parser_config = kb.parser_config
         if not kb_parser_config.get("raptor", {}).get("use_raptor", False):
-            kb_parser_config.update({
-                "raptor": {
-                    "use_raptor": True,
-                    "prompt": "Please summarize the following paragraphs. Be careful with the numbers, do not make things up. Paragraphs as following:\n      {cluster_content}\nThe above is the content you need to summarize.",
-                    "max_token": 256,
-                    "threshold": 0.1,
-                    "max_cluster": 64,
-                    "random_seed": 0,
-                    "scope": "file",
-                    "clustering_method": "gmm",
-                    "tree_builder": "raptor",
-                },
-            })
+            kb_parser_config.update(
+                {
+                    "raptor": {
+                        "use_raptor": True,
+                        "prompt": "Please summarize the following paragraphs. Be careful with the numbers, do not make things up. Paragraphs as following:\n      {cluster_content}\nThe above is the content you need to summarize.",
+                        "max_token": 256,
+                        "threshold": 0.1,
+                        "max_cluster": 64,
+                        "random_seed": 0,
+                        "scope": "file",
+                        "clustering_method": "gmm",
+                        "tree_builder": "raptor",
+                    },
+                }
+            )
             if ctx.write_interceptor:
                 update_result = ctx.write_interceptor.intercept("KnowledgebaseService.update_by_id")
             else:
@@ -271,11 +403,8 @@ class TaskHandler:
                 return
 
         # Bind LLM for raptor
-        chat_model_config = get_model_config_from_provider_instance(
-            task_tenant_id, LLMType.CHAT, kb_task_llm_id
-        )
+        chat_model_config = get_model_config_from_provider_instance(task_tenant_id, LLMType.CHAT, kb_task_llm_id)
         with LLMBundle(task_tenant_id, chat_model_config, lang=ctx.language) as chat_model:
-
             # Run RAPTOR
             raptor_service = RaptorService(ctx=ctx)
 
@@ -285,7 +414,7 @@ class TaskHandler:
                     chat_mdl=chat_model,
                     embd_mdl=embedding_model,
                     vector_size=vector_size,
-                    doc_ids=ctx.doc_ids,
+                    doc_ids=ctx.doc_ids or [ctx.doc_id],
                 )
 
             ctx.recording_context.record("raptor_chunks", chunks)
@@ -293,7 +422,7 @@ class TaskHandler:
 
             # Insert RAPTOR chunks
             if chunks:
-                task_doc_id = (ctx.doc_ids or [GRAPH_RAPTOR_FAKE_DOC_ID])[0]
+                task_doc_id = (ctx.doc_ids or [ctx.doc_id] or [GRAPH_RAPTOR_FAKE_DOC_ID])[0]
                 chunk_service = ChunkService(ctx=ctx)
                 insert_result = await chunk_service.insert_chunks(ctx.id, task_tenant_id, task_dataset_id, chunks)
                 if insert_result:
@@ -304,27 +433,43 @@ class TaskHandler:
                 # Cleanup stale RAPTOR chunks
                 cleaned_chunks = 0
                 for cleanup_doc_id, keep_method in raptor_cleanup_chunks:
-                    ret = await self._delete_raptor_chunks(
-                        cleanup_doc_id, task_tenant_id, task_dataset_id, keep_method
-                    )
+                    ret = await self._delete_raptor_chunks(cleanup_doc_id, task_tenant_id, task_dataset_id, keep_method)
                     cleaned_chunks += ret
 
                 if cleaned_chunks:
                     ctx.progress_cb(msg=f"Cleaned up {cleaned_chunks} stale RAPTOR chunks.")
 
+                # Build the per-doc RAPTOR tree graph from the just-
+                # inserted summaries. Each chunk in ``chunks`` carries
+                # the doc_id it was written under (real doc id for
+                # scope="file"; GRAPH_RAPTOR_FAKE_DOC_ID for the
+                # dataset-scope path). We materialize one graph row per
+                # distinct doc_id so the dataset structure-graph
+                # endpoint can surface a RAPTOR tab per document.
+                # Failure here is best-effort — the summaries are
+                # already persisted; the tab just won't render.
+                raptor_doc_ids = {str(c.get("doc_id")) for c in chunks if c.get("doc_id")}
+                for raptor_doc_id in raptor_doc_ids:
+                    try:
+                        await raptor_service._persist_raptor_graph_to_es(raptor_doc_id)
+                    except Exception:
+                        logging.exception(
+                            "raptor_graph: build failed for kb=%s doc=%s",
+                            task_dataset_id,
+                            raptor_doc_id,
+                        )
+
                 # Update document stats
                 if ctx.write_interceptor:
                     ctx.write_interceptor.intercept("DocumentService.increment_chunk_num")
                 else:
                     DocumentService.increment_chunk_num(task_doc_id, task_dataset_id, token_count, len(chunks), 0)
 
-            ctx.recording_context.record("task_status", "completed")
-            ctx.progress_cb(prog=1.0, msg="RAPTOR done")
+            if mark_done:
+                ctx.recording_context.record("task_status", "completed")
+                ctx.progress_cb(prog=1.0, msg="RAPTOR done")
 
-    async def _run_graphrag(
-        self,
-        embedding_model: LLMBundle
-    ) -> None:
+    async def _run_graphrag(self, embedding_model: LLMBundle) -> None:
         """Run GraphRAG."""
         ctx = self._task_context
         task_tenant_id = ctx.tenant_id
@@ -339,29 +484,31 @@ class TaskHandler:
 
         kb_parser_config = kb.parser_config
         if not kb_parser_config.get("graphrag", {}).get("use_graphrag", False):
-            kb_parser_config.update({
-                "graphrag": {
-                    "use_graphrag": True,
-                    "entity_types": [
-                        "organization",
-                        "person",
-                        "geo",
-                        "event",
-                        "category",
-                    ],
-                    "method": "light",
-                    "batch_chunk_token_size": 4096,
-                    "retry_attempts": 2,
-                    "retry_backoff_seconds": 2.0,
-                    "retry_backoff_max_seconds": 60.0,
-                    "build_subgraph_timeout_per_chunk_seconds": 300,
-                    "build_subgraph_min_timeout_seconds": 600,
-                    "merge_timeout_seconds": 180,
-                    "resolution_timeout_seconds": 1800,
-                    "community_timeout_seconds": 1800,
-                    "lock_acquire_timeout_seconds": 600,
+            kb_parser_config.update(
+                {
+                    "graphrag": {
+                        "use_graphrag": True,
+                        "entity_types": [
+                            "organization",
+                            "person",
+                            "geo",
+                            "event",
+                            "category",
+                        ],
+                        "method": "light",
+                        "batch_chunk_token_size": 4096,
+                        "retry_attempts": 2,
+                        "retry_backoff_seconds": 2.0,
+                        "retry_backoff_max_seconds": 60.0,
+                        "build_subgraph_timeout_per_chunk_seconds": 300,
+                        "build_subgraph_min_timeout_seconds": 600,
+                        "merge_timeout_seconds": 180,
+                        "resolution_timeout_seconds": 1800,
+                        "community_timeout_seconds": 1800,
+                        "lock_acquire_timeout_seconds": 600,
+                    }
                 }
-            })
+            )
             if ctx.write_interceptor:
                 update_result = ctx.write_interceptor.intercept("KnowledgebaseService.update_by_id")
             else:
@@ -372,11 +519,8 @@ class TaskHandler:
 
         graphrag_conf = kb_parser_config.get("graphrag", {})
         start_ts = timer()
-        chat_model_config = get_model_config_from_provider_instance(
-            task_tenant_id, LLMType.CHAT, kb_task_llm_id
-        )
+        chat_model_config = get_model_config_from_provider_instance(task_tenant_id, LLMType.CHAT, kb_task_llm_id)
         with LLMBundle(task_tenant_id, chat_model_config, lang=task_language) as chat_model:
-
             with_resolution = graphrag_conf.get("resolution", False)
             with_community = graphrag_conf.get("community", False)
 
@@ -399,7 +543,20 @@ class TaskHandler:
 
     async def _run_standard_chunking(
         self,
-        embedding_model: LLMBundle
+        embedding_model: LLMBundle,
+        vector_size: int,
+    ) -> None:
+        ctx = self._task_context
+        try:
+            await self._run_standard_chunking_impl(embedding_model, vector_size)
+        except Exception:
+            abort_doc_chunking_counter(ctx.doc_id)
+            raise
+
+    async def _run_standard_chunking_impl(
+        self,
+        embedding_model: LLMBundle,
+        vector_size: int,
     ) -> None:
         """Run standard chunking pipeline."""
         ctx = self._task_context
@@ -409,7 +566,7 @@ class TaskHandler:
         task_doc_id = ctx.doc_id
         task_start_ts = timer()
         doc_task_llm_id = ctx.parser_config.get("llm_id") or ctx.llm_id
-        ctx.raw_task['llm_id'] = doc_task_llm_id
+        ctx.raw_task["llm_id"] = doc_task_llm_id
 
         # Build chunks
         start_ts = timer()
@@ -419,9 +576,7 @@ class TaskHandler:
         bucket, name = File2DocumentService.get_storage_address(doc_id=ctx.doc_id)
         binary = await self._get_storage_binary(bucket, name)
         if binary is None:
-            raise FileNotFoundError(
-                f"Can not find file <{ctx.name}> from minio. Could you try it again."
-            )
+            raise FileNotFoundError(f"Can not find file <{ctx.name}> from minio. Could you try it again.")
 
         chunks = await chunk_service.build_chunks(binary)
         ctx.recording_context.record("chunks", chunks)
@@ -431,7 +586,18 @@ class TaskHandler:
         logging.info("Build document {}: {:.2f}s".format(ctx.name, timer() - start_ts))
 
         if not chunks:
-            ctx.progress_cb(1., msg=f"No chunk built from {ctx.name}")
+            ctx.progress_cb(msg=f"No chunk built from {ctx.name}")
+            if not await self._run_document_post_chunking_if_last(
+                embedding_model,
+                vector_size,
+                task_start_ts,
+                0,
+                0,
+            ):
+                return
+            task_time_cost = timer() - task_start_ts
+            ctx.recording_context.record("task_status", "completed")
+            ctx.progress_cb(prog=1.0, msg="Task done ({:.2f}s)".format(task_time_cost))
             return
 
         ctx.progress_cb(msg="Generate {} chunks".format(len(chunks)))
@@ -440,9 +606,7 @@ class TaskHandler:
         start_ts = timer()
         embedding_service = EmbeddingService(ctx=ctx)
         try:
-            token_count, vector_size = await embedding_service.embed_chunks(
-                chunks, embedding_model, ctx.parser_config
-            )
+            token_count, vector_size = await embedding_service.embed_chunks(chunks, embedding_model, ctx.parser_config)
         except TaskCanceledException:
             raise
         except Exception as e:
@@ -457,11 +621,6 @@ class TaskHandler:
         logging.info(progress_message)
         ctx.progress_cb(msg=progress_message)
 
-        # Build TOC if needed
-        toc_thread = None
-        if ctx.parser_id.lower() == "naive" and ctx.parser_config.get("toc_extraction", False):
-            toc_thread = asyncio.create_task(asyncio.to_thread(self._build_toc, ctx, chunks, ctx.progress_cb))
-
         # Insert chunks
         chunk_count = len(set([chunk["id"] for chunk in chunks]))
         start_ts = timer()
@@ -469,15 +628,15 @@ class TaskHandler:
         chunk_service = ChunkService(ctx=ctx)
 
         if ctx.has_canceled_func(task_id):
+            abort_doc_chunking_counter(task_doc_id)
             ctx.progress_cb(-1, msg="Task has been canceled.")
             return
 
-        insert_result = await chunk_service.insert_chunks(
-            task_id, task_tenant_id, task_dataset_id, chunks
-        )
+        insert_result = await chunk_service.insert_chunks(task_id, task_tenant_id, task_dataset_id, chunks)
 
         if not insert_result:
             ctx.recording_context.record("insertion_result", "failed")
+            abort_doc_chunking_counter(task_doc_id)
             return
         ctx.recording_context.record("insertion_result", "success")
 
@@ -487,12 +646,8 @@ class TaskHandler:
 
         ctx.progress_cb(msg="Indexing done ({:.2f}s).".format(timer() - start_ts))
 
-        toc_chunk = await self._process_toc_thread(toc_thread)
-        if toc_chunk:
-            ctx.recording_context.record("toc_chunk", [toc_chunk])
-            await post_processor.insert_toc_chunk(toc_chunk, chunk_service)
-
         if ctx.has_canceled_func(task_id):
+            abort_doc_chunking_counter(task_doc_id)
             ctx.progress_cb(-1, msg="Task has been canceled.")
             return
 
@@ -502,17 +657,44 @@ class TaskHandler:
         else:
             DocumentService.increment_chunk_num(task_doc_id, task_dataset_id, token_count, chunk_count, 0)
 
+        if not await self._run_document_post_chunking_if_last(
+            embedding_model,
+            vector_size,
+            task_start_ts,
+            len(chunks),
+            token_count,
+        ):
+            return
+
         task_time_cost = timer() - task_start_ts
         ctx.recording_context.record("task_status", "completed")
         ctx.progress_cb(prog=1.0, msg="Task done ({:.2f}s)".format(task_time_cost))
 
-        logging.info(
-            "Chunk doc({}), page({}-{}), chunks({}), token({}), elapsed:{:.2f}".format(
-                ctx.name, ctx.from_page, ctx.to_page,
-                len(chunks), token_count, task_time_cost
-            )
+        logging.info("Chunk doc({}), page({}-{}), chunks({}), token({}), elapsed:{:.2f}".format(ctx.name, ctx.from_page, ctx.to_page, len(chunks), token_count, task_time_cost))
+
+    async def _run_document_post_chunking_if_last(
+        self,
+        embedding_model: LLMBundle,
+        vector_size: int,
+        task_start_ts: float,
+        chunks_len: int,
+        token_count: int,
+    ) -> bool:
+        """Thin delegator. The pipeline lives in
+        ``rag.svr.task_executor_refactor.chunk_post_processor``.
+        """
+        from rag.svr.task_executor_refactor.chunk_post_processor import (
+            run_document_post_chunking_if_last,
         )
 
+        return await run_document_post_chunking_if_last(
+            self,
+            embedding_model,
+            vector_size,
+            task_start_ts,
+            chunks_len,
+            token_count,
+        )
 
     async def _process_toc_thread(self, toc_thread):
         try:
@@ -527,30 +709,105 @@ class TaskHandler:
     @classmethod
     async def _get_storage_binary(cls, bucket: str, name: str) -> bytes:
         from common import settings
+
         """Get binary from storage."""
         return await thread_pool_exec(settings.STORAGE_IMPL.get, bucket, name)
 
+    @staticmethod
+    async def _load_chunks_for_doc(
+        tenant_id: str,
+        kb_id: str,
+        doc_id: str,
+        batch_size: int = 500,
+    ) -> AsyncIterator[List[Dict]]:
+        """Stream a document's chunks from the doc store one batch at a time.
+
+        Async generator that yields successive batches of up to ``batch_size``
+        chunks. Order is pushed to the doc store via
+        ``OrderByExpr().asc("page_num_int").asc("top_int")`` so callers do
+        not need to re-sort. Rows with a ``compile_kwd`` marker (artifact
+        pages, structure entities, etc.) are filtered out defensively.
+
+        Memory is bounded by ``batch_size``: at most one page is materialised
+        at a time, so long documents do not balloon the worker's heap.
+        """
+        from common.doc_store.doc_store_base import OrderByExpr
+
+        index_nm = search.index_name(tenant_id)
+        if not settings.docStoreConn.index_exist(index_nm, kb_id):
+            return
+
+        select_fields = [
+            "id",
+            "doc_id",
+            "content_with_weight",
+            "page_num_int",
+            "top_int",
+        ]
+        order_by = OrderByExpr()
+        order_by.asc("page_num_int")
+        order_by.asc("top_int")
+
+        offset = 0
+        while True:
+            try:
+                res = await thread_pool_exec(
+                    settings.docStoreConn.search,
+                    select_fields,
+                    [],
+                    {"doc_id": [doc_id], "available_int": 1},
+                    [],
+                    order_by,
+                    offset,
+                    batch_size,
+                    index_nm,
+                    [kb_id],
+                )
+                field_map = settings.docStoreConn.get_fields(res, select_fields)
+            except Exception:
+                logging.exception("load_chunks_for_doc: failed to load chunks for doc=%s", doc_id)
+                return
+            if not field_map:
+                return
+
+            batch: List[Dict] = []
+            for row_id, row in field_map.items():
+                if row.get("compile_kwd"):
+                    continue
+                batch.append(
+                    {
+                        "id": row_id,
+                        "doc_id": row.get("doc_id") or doc_id,
+                        "content_with_weight": row.get("content_with_weight") or "",
+                        "page_num_int": row.get("page_num_int", 0),
+                        "top_int": row.get("top_int", 0),
+                    }
+                )
+            if batch:
+                yield batch
+            if len(field_map) < batch_size:
+                return
+            offset += batch_size
+
     @classmethod
     def _build_toc(cls, ctx: TaskContext, docs: List[Dict], progress_cb: Callable) -> Optional[Dict]:
         """Build table of contents."""
         progress_cb(msg="Start to generate table of content ...")
-        chat_model_config = get_model_config_from_provider_instance(
-            ctx.tenant_id, LLMType.CHAT, ctx.llm_id
-        )
+        chat_model_config = get_model_config_from_provider_instance(ctx.tenant_id, LLMType.CHAT, ctx.llm_id)
         with LLMBundle(ctx.tenant_id, chat_model_config, lang=ctx.language) as chat_mdl:
-
-            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 = 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),
+                ),
+            )
 
             # NOTE: asyncio.run() creates a new event loop in the worker thread
             # (this method is called via asyncio.to_thread), which is the
             # intended pattern for bridging sync -> async in a thread context.
-            toc: list[dict] = asyncio.run(
-                run_toc_from_text([d["content_with_weight"] for d in docs], chat_mdl, progress_cb)
-            )
-            logging.info("------------ T O C -------------\n" + json.dumps(toc, ensure_ascii=False, indent='  '))
+            toc: list[dict] = asyncio.run(run_toc_from_text([d["content_with_weight"] for d in docs], chat_mdl, progress_cb))
+            logging.info("------------ T O C -------------\n" + json.dumps(toc, ensure_ascii=False, indent="  "))
 
             for ii, item in enumerate(toc):
                 try:
@@ -578,19 +835,17 @@ class TaskHandler:
 
             if toc:
                 import copy
+
                 d = copy.deepcopy(docs[-1])
                 d["content_with_weight"] = json.dumps(toc, ensure_ascii=False)
                 d["toc_kwd"] = "toc"
                 d["available_int"] = 0
                 d["page_num_int"] = [100000000]
-                d["id"] = xxhash.xxh64(
-                    (d["content_with_weight"] + str(d["doc_id"])).encode("utf-8", "surrogatepass")).hexdigest()
+                d["id"] = xxhash.xxh64((d["content_with_weight"] + str(d["doc_id"])).encode("utf-8", "surrogatepass")).hexdigest()
                 return d
             return None
 
-    async def _delete_raptor_chunks(
-        self, doc_id: str, tenant_id: str, kb_id: str, keep_method: Optional[str]
-    ) -> int:
+    async def _delete_raptor_chunks(self, doc_id: str, tenant_id: str, kb_id: str, keep_method: Optional[str]) -> int:
         """Delete RAPTOR chunks."""
         if self._task_context.write_interceptor:
             return self._task_context.write_interceptor.intercept("delete_raptor_chunks")