fix: improve pipeline compilation, template updates, and document resets (#16815)

### What problem does this PR solve?

- Clear stale pipeline IDs and generated data when updating documents
without `pipeline_id`.
- Support tree compilation results in pipeline workflows.
- Update compilation templates in place while preserving existing
template IDs.
- Improve duplicate-template validation messages.

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)
- [x] New Feature (non-breaking change which adds functionality)

Co-authored-by: Jin Hai <haijin.chn@gmail.com>

api/apps/restful_apis/document_api.py

@@ -272,19 +272,25 @@ async def update_document(tenant_id, dataset_id, document_id):
         parser_config_template_group_changed = parser_config_template_group_touched and _compilation_template_group_id_changed(old_parser_config, req["parser_config"])
         DocumentService.update_parser_config(doc.id, req["parser_config"])
 
-    # pipeline_id provided - reset document for reparse
-    if update_doc_req.pipeline_id:
-        if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=update_doc_req.pipeline_id):
+    # A non-empty pipeline_id selects pipeline parsing; an explicitly empty
+    # value clears it and switches back to the direct parser path.
+    if "pipeline_id" in req:
+        if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=update_doc_req.pipeline_id or ""):
             return error
     # chunk method provided - the update method will check if it's different with existing one
     elif update_doc_req.chunk_method:
         if error := update_chunk_method(req, doc, tenant_id):
             return error
         if parser_config_template_group_changed and doc.parser_id.lower() == req["chunk_method"].lower():
-            if error := reset_document_for_reparse(doc, tenant_id):
+            if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=""):
                 return error
     elif parser_config_template_group_changed:
-        if error := reset_document_for_reparse(doc, tenant_id):
+        if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=""):
+            return error
+    else:
+        # Direct-parser updates do not carry a pipeline_id. Clear any stale
+        # pipeline selection and its generated document-store data.
+        if error := reset_document_for_reparse(doc, tenant_id, pipeline_id=""):
             return error
 
     if "enabled" in req:  # already checked in UpdateDocumentReq - it's int if present

api/apps/services/document_api_service.py

@@ -80,7 +80,13 @@ def update_chunk_method(req, doc, tenant_id):
     """
     if doc.parser_id.lower() != req["chunk_method"].lower():
         # if chunk method changed, reset document for reparse
-        result = reset_document_for_reparse(doc, tenant_id, parser_id=req["chunk_method"])
+        result = reset_document_for_reparse(doc, tenant_id, parser_id=req["chunk_method"], pipeline_id="")
+        if result:
+            return result
+    elif doc.pipeline_id:
+        # An explicit chunk method selects the direct parser path. Clear the
+        # previous pipeline even when the parser method itself is unchanged.
+        result = reset_document_for_reparse(doc, tenant_id, pipeline_id="")
         if result:
             return result
     if not req.get("parser_config"):
@@ -122,7 +128,9 @@ def reset_document_for_reparse(doc, tenant_id, parser_id=None, pipeline_id=None)
     if not e:
         return get_error_data_result(message="Document not found!")
 
-    # Delete chunks from document store
+    # Update document statistics before deleting all document rows. Pipeline
+    # compilation rows may exist even when token_num is zero, so the doc-store
+    # cleanup must not be gated by the document counters.
     if doc.token_num > 0:
         try:
             e = DocumentService.increment_chunk_num(
@@ -136,7 +144,7 @@ def reset_document_for_reparse(doc, tenant_id, parser_id=None, pipeline_id=None)
             return get_error_data_result(message="Document not found!")
         if not e:
             return get_error_data_result(message="Document not found!")
-        settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
+    settings.docStoreConn.delete({"doc_id": doc.id}, search.index_name(tenant_id), doc.kb_id)
 
     # Delete chunk images
     try:

api/db/services/compilation_template_group_service.py

@@ -296,16 +296,84 @@ class CompilationTemplateGroupService(CommonService):
                 cls.model.update(**updates).where(cls.model.id == group_id).execute()
 
             if templates is not None:
-                # Soft-delete previous children (en-bloc replace). Simpler than
-                # diffing and acceptable given small N — child IDs are not
-                # referenced externally (parser_config keys the group, not its
-                # children).
-                CompilationTemplate.update(status=StatusEnum.INVALID.value).where(
-                    CompilationTemplate.group_id == group_id,
-                    CompilationTemplate.status == StatusEnum.VALID.value,
-                ).execute()
-                for i, child in enumerate(templates):
-                    cls._insert_child(group_id, tenant_id, child, index=i)
+                current_children = list(
+                    CompilationTemplate.select()
+                    .where(
+                        CompilationTemplate.group_id == group_id,
+                        CompilationTemplate.status == StatusEnum.VALID.value,
+                    )
+                    .order_by(CompilationTemplate.create_time.asc())
+                )
+                current_by_id = {child.id: child for child in current_children}
+                retained_ids: set[str] = set()
+                seen_names: set[str] = set()
+
+                for index, child in enumerate(templates):
+                    child_id = str((child or {}).get("id") or "").strip()
+                    target = current_by_id.get(child_id) if child_id else None
+                    if child_id and target is None:
+                        raise GroupValidationError(f"Template {child_id} does not belong to this group.")
+                    # Older clients did not send child ids. Preserve their
+                    # existing ids by matching the submitted order.
+                    if target is None and not child_id and index < len(current_children):
+                        target = current_children[index]
+
+                    kind = str((child or {}).get("kind") or "").strip()
+                    name = str((child or {}).get("name") or "").strip()
+                    config = (child or {}).get("config") or {}
+                    if not kind or not name or not isinstance(config, dict):
+                        raise GroupValidationError("Each template must include a name, kind, and config object.")
+                    if name in seen_names:
+                        raise GroupValidationError(f"Template name '{name}' is duplicated in this group.")
+                    seen_names.add(name)
+
+                    from api.db.services.compilation_template_service import CompilationTemplateService
+
+                    config = CompilationTemplateService.fill_config_default_llm(config, tenant_id)
+                    duplicate_query = CompilationTemplate.select().where(
+                        CompilationTemplate.tenant_id == tenant_id,
+                        CompilationTemplate.name == name,
+                        ~CompilationTemplate.is_builtin,
+                        CompilationTemplate.status == StatusEnum.VALID.value,
+                    )
+                    if target is not None:
+                        duplicate_query = duplicate_query.where(CompilationTemplate.id != target.id)
+                    if duplicate_query.exists():
+                        raise GroupValidationError(f"Template name '{name}' already exists. Please choose another name.")
+
+                    if target is None:
+                        new_id = cls._insert_child(
+                            group_id,
+                            tenant_id,
+                            {
+                                "name": name,
+                                "description": (child or {}).get("description") or "",
+                                "kind": kind,
+                                "config": config,
+                            },
+                            index=index,
+                        )
+                        retained_ids.add(new_id)
+                        continue
+
+                    CompilationTemplate.update(
+                        name=name,
+                        description=str((child or {}).get("description") or ""),
+                        kind=kind,
+                        config=config,
+                    ).where(CompilationTemplate.id == target.id).execute()
+                    retained_ids.add(target.id)
+
+                removed_children = [child for child in current_children if child.id not in retained_ids]
+                if removed_children:
+                    removed_names = [child.name for child in removed_children]
+                    removed_ids = [child.id for child in removed_children]
+                    cls._purge_stale_invalid_children(tenant_id, removed_names)
+                    CompilationTemplate.update(status=StatusEnum.INVALID.value).where(
+                        CompilationTemplate.group_id == group_id,
+                        CompilationTemplate.id.in_(removed_ids),
+                        CompilationTemplate.status == StatusEnum.VALID.value,
+                    ).execute()
 
         return cls.get_saved(group_id, tenant_id)
 
@@ -336,7 +404,7 @@ class CompilationTemplateGroupService(CommonService):
         child: dict,
         *,
         index: int,
-    ) -> None:
+    ) -> str:
         kind = str((child or {}).get("kind") or "").strip()
         name = str((child or {}).get("name") or "").strip()
         config = (child or {}).get("config") or {}
@@ -345,6 +413,15 @@ class CompilationTemplateGroupService(CommonService):
         from api.db.services.compilation_template_service import CompilationTemplateService
 
         config = CompilationTemplateService.fill_config_default_llm(config, tenant_id)
+        duplicate = CompilationTemplate.select().where(
+            CompilationTemplate.tenant_id == tenant_id,
+            CompilationTemplate.name == name,
+            ~CompilationTemplate.is_builtin,
+            CompilationTemplate.status == StatusEnum.VALID.value,
+        )
+        if duplicate.exists():
+            raise GroupValidationError(f"A compilation template named '{name}' already exists. Please choose a different name.")
+
         template_id = get_uuid()
         CompilationTemplate.create(
             id=template_id,
@@ -357,6 +434,19 @@ class CompilationTemplateGroupService(CommonService):
             is_builtin=False,
             status=StatusEnum.VALID.value,
         )
+        return template_id
+
+    @classmethod
+    def _purge_stale_invalid_children(cls, tenant_id: str, names: list[str]) -> None:
+        cleaned_names = [name for name in {str(name).strip() for name in names} if name]
+        if not cleaned_names:
+            return
+        CompilationTemplate.delete().where(
+            CompilationTemplate.tenant_id == tenant_id,
+            CompilationTemplate.name.in_(cleaned_names),
+            ~CompilationTemplate.is_builtin,
+            CompilationTemplate.status == StatusEnum.INVALID.value,
+        ).execute()
 
     # ------------------------------------------------------------------
     # Lookup helpers used by the orchestrator

rag/flow/compiler/compiler.py

@@ -15,12 +15,14 @@
 import logging
 import random
 from copy import deepcopy
+from types import SimpleNamespace
 
 import xxhash
 
 from agent.component.llm import LLMParam, LLM
-from api.db.joint_services.tenant_model_service import get_model_config_from_provider_instance
+from api.db.joint_services.tenant_model_service import get_model_config_by_id, get_tenant_default_model_by_type, resolve_model_config
 from api.db.services.document_service import DocumentService
+from api.db.services.knowledgebase_service import KnowledgebaseService
 from api.db.services.llm_service import LLMBundle
 from api.db.services.task_service import has_canceled
 from common.constants import LLMType
@@ -62,7 +64,108 @@ class Compiler(ProcessBase, LLM):
         ``callback(prog, msg)`` (positional) or ``callback(msg=...)``; the
         flow's ``self.callback`` expects ``(progress, message)``.
         """
-        self.callback(prog, msg)
+        self.callback(0 if prog is None else prog, msg)
+
+    async def _compile_tree_templates(
+        self,
+        templates: list[tuple[str, dict]],
+        chat_mdl_by_tid: dict[str, LLMBundle],
+        embedding_model: LLMBundle,
+        chunks: list[dict],
+        tenant_id: str,
+        kb_id: str,
+        doc_id: str,
+    ) -> None:
+        """Build and persist tree graphs from the pipeline's in-memory chunks.
+
+        The document post-chunking path can reload chunks from the doc store,
+        but a pipeline Compiler runs before DataflowService persists its final
+        chunks. Supply RAPTOR with the same ``(text, vector, chunk_id)`` shape
+        from the current pipeline output instead.
+        """
+        from rag.advanced_rag.knowlege_compile.structure import _struct_upsert_graph_json
+        from rag.svr.task_executor_refactor.chunk_post_processor import raptor_tree_to_graph
+        from rag.svr.task_executor_refactor.raptor_service import RaptorService
+
+        tree_inputs = []
+        texts = []
+        for chunk in chunks:
+            text = chunk.get("content_with_weight") or chunk.get("text") or ""
+            if not isinstance(text, str) or not text.strip():
+                continue
+            chunk_id = str(chunk.get("id") or "")
+            if not chunk_id:
+                continue
+            texts.append(text)
+            tree_inputs.append((text, chunk_id))
+        if not tree_inputs:
+            return
+
+        vectors, _ = embedding_model.encode(texts)
+        tree_chunks = [(text, vector, chunk_id) for (text, chunk_id), vector in zip(tree_inputs, vectors)]
+        if not tree_chunks:
+            return
+
+        tree_context = SimpleNamespace(
+            tenant_id=tenant_id,
+            kb_id=kb_id,
+            doc_id=doc_id,
+            id=getattr(self._canvas, "task_id", ""),
+            progress_cb=self._compile_progress,
+        )
+        raptor_service = RaptorService(tree_context)
+
+        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 {},
+            }
+            self._compile_progress(msg=f"tree-template ({idx + 1}/{len(templates)}): building tree for doc={doc_id}")
+            try:
+                tree = await raptor_service.build_doc_tree(
+                    chunks=tree_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("Compiler: tree-template %s build failed for doc %s", template_id, doc_id)
+                continue
+            if tree is None:
+                continue
+
+            if bool(raptor_cfg.get("rechunk")):
+                self._compile_progress(msg="Compiler: tree rechunking is not supported for in-memory pipeline chunks; keeping original chunks.")
+
+            try:
+                await _struct_upsert_graph_json(
+                    raptor_tree_to_graph(tree),
+                    tenant_id,
+                    kb_id,
+                    doc_id,
+                    compile_kwd="tree",
+                    compilation_template_id=template_id,
+                )
+            except Exception:
+                logging.exception("Compiler: 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(tenant_id, kb_id, doc_id, tree)
+            except Exception:
+                logging.exception("Compiler: tree-template %s dataset navigation upsert failed for doc %s", template_id, doc_id)
+
+            self._compile_progress(msg=f"tree-template ({idx + 1}/{len(templates)}): persisted tree graph for doc {doc_id}")
 
     def _compile_language(self, kwargs: dict) -> str:
         language = kwargs.get("language") or getattr(self._canvas, "_language", None)
@@ -79,13 +182,15 @@ class Compiler(ProcessBase, LLM):
         self.set_output("output_format", "chunks")
         self.callback(random.randint(1, 5) / 100.0, "Start knowledge compilation.")
 
-        # Collect the upstream chunk list (same contract as the Extractor).
-        inputs = self.get_input_elements()
-        chunks = []
-        for _, v in inputs.items():
-            val = v["value"]
-            if isinstance(val, list):
-                chunks = deepcopy(val)
+        # Pipeline components receive the previous component's output as
+        # kwargs. Do not call LLM.get_input_elements() here: it resolves the
+        # inherited prompt variables through Canvas.globals, while Pipeline
+        # is a Graph and has no globals.
+        chunks = deepcopy(kwargs.get("chunks") or [])
+        if not chunks:
+            for val in kwargs.values():
+                if isinstance(val, list):
+                    chunks = deepcopy(val)
 
         tenant_id = self._canvas.get_tenant_id()
         doc_id = self._canvas._doc_id
@@ -105,7 +210,7 @@ class Compiler(ProcessBase, LLM):
         template_ids = resolve_template_ids_from_groups(self._param.compilation_template_group_ids, tenant_id)
         active_templates = load_active_templates(template_ids, tenant_id)
         if not active_templates:
-            self.callback(msg="No active compilation templates resolved from the configured groups.")
+            self.callback(0, "No active compilation templates resolved from the configured groups.")
             self.set_output("chunks", chunks)
             return
 
@@ -121,7 +226,7 @@ class Compiler(ProcessBase, LLM):
                 chat_llm_id = tpl_llm_id.strip()
                 if chat_llm_id not in llm_bundle_cache:
                     try:
-                        cfg = get_model_config_from_provider_instance(tenant_id, LLMType.CHAT, chat_llm_id)
+                        cfg = resolve_model_config(tenant_id, LLMType.CHAT, chat_llm_id)
                         llm_bundle_cache[chat_llm_id] = LLMBundle(
                             tenant_id,
                             cfg,
@@ -154,9 +259,20 @@ class Compiler(ProcessBase, LLM):
             return
         active_templates = filtered_templates
 
+        if self._canvas._kb_id:
+            e, kb = KnowledgebaseService.get_by_id(self._canvas._kb_id)
+            if kb.tenant_embd_id:
+                try:
+                    embd_model_config = get_model_config_by_id(self._canvas._tenant_id, LLMType.EMBEDDING, kb.tenant_embd_id)
+                except LookupError:
+                    embd_model_config = resolve_model_config(self._canvas._tenant_id, LLMType.EMBEDDING, kb.embd_id)
+            else:
+                embd_model_config = resolve_model_config(self._canvas._tenant_id, LLMType.EMBEDDING, kb.embd_id)
+        else:
+            embd_model_config = get_tenant_default_model_by_type(self._canvas._tenant_id, LLMType.EMBEDDING)
         embedding_model = LLMBundle(
             tenant_id,
-            LLMType.EMBEDDING,
+            embd_model_config,
             lang=language,
             max_retries=self._param.max_retries,
             retry_interval=self._param.delay_after_error,
@@ -164,14 +280,15 @@ class Compiler(ProcessBase, LLM):
 
         tree_templates, non_tree_templates = split_tree_templates(active_templates)
         if tree_templates:
-            # ``tree`` templates run RAPTOR over the whole document by
-            # reloading vectors from the doc store; that path is owned by the
-            # chunking task executor and isn't available from the flow.
-            logging.warning(
-                "Compiler: %d tree-kind template(s) are not supported in the flow pipeline; skipping",
-                len(tree_templates),
+            await self._compile_tree_templates(
+                tree_templates,
+                chat_mdl_by_tid,
+                embedding_model,
+                chunks,
+                tenant_id,
+                kb_id,
+                doc_id,
             )
-            self.callback(msg=f"Skipping {len(tree_templates)} tree-kind template(s) (unsupported in flow).")
 
         if non_tree_templates:
             task_id = getattr(self._canvas, "task_id", None)
@@ -181,7 +298,7 @@ class Compiler(ProcessBase, LLM):
 
             async def _chunk_batches():
                 for i in range(0, len(chunks), DOC_STRUCTURE_COMPILE_BATCH_CHUNKS):
-                    yield chunks[i:i + DOC_STRUCTURE_COMPILE_BATCH_CHUNKS]
+                    yield chunks[i : i + DOC_STRUCTURE_COMPILE_BATCH_CHUNKS]
 
             await run_structure_compile_over_batches(
                 active_templates=non_tree_templates,

web/src/interfaces/request/compilation-template.ts

@@ -47,6 +47,7 @@ export interface ICreateCompilationTemplateGroupRequestBody {
   description?: string;
   avatar?: string;
   templates: Array<{
+    id?: string;
     name?: string;
     description?: string;
     kind: string;

web/src/pages/user-setting/compilation-templates/edit-template/schema.ts

@@ -18,6 +18,7 @@ export const buildRaptorConfigSchema = (t: (key: string) => string) =>
 
 export const buildTemplateSchema = (t: (key: string) => string) =>
   z.object({
+    id: z.string().optional(),
     name: z.string().optional(),
     description: z.string().optional(),
     llm_id: z.string().min(1, t('setting.llmForExtractionRequired')),

web/src/pages/user-setting/compilation-templates/edit-template/utils.ts

@@ -30,6 +30,7 @@ export const getFieldKeyOrder = (keys: string[]): string[] => {
 };
 
 export const DefaultTemplateValues: TemplateSchemaType = {
+  id: undefined,
   name: '',
   description: '',
   llm_id: '',
@@ -137,6 +138,7 @@ export const transformDetailToForm = (
   if (detail.kind === CompilationTemplateKind.Tree) {
     const raptor: ICompilationTemplateRaptorConfig = config.raptor ?? {};
     return {
+      id: detail.id,
       name: detail.name ?? '',
       description: detail.description ?? '',
       llm_id: config.llm_id ?? '',
@@ -161,6 +163,7 @@ export const transformDetailToForm = (
   });
 
   return {
+    id: detail.id,
     name: detail.name ?? '',
     description: detail.description ?? '',
     llm_id: config.llm_id ?? '',
@@ -207,6 +210,7 @@ export const transformTemplateToPayload = (template: TemplateSchemaType) => {
   }
 
   return {
+    id: template.id,
     name: template.name,
     description: template.description,
     kind: template.kind,