From d291e4641d0e3638357c8c852066546eb7cf1c5a Mon Sep 17 00:00:00 2001 From: buua436 Date: Fri, 10 Jul 2026 21:17:19 +0800 Subject: [PATCH] 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 --- api/apps/restful_apis/document_api.py | 16 +- api/apps/services/document_api_service.py | 14 +- .../compilation_template_group_service.py | 112 +++++++++++-- rag/flow/compiler/compiler.py | 157 +++++++++++++++--- .../request/compilation-template.ts | 1 + .../edit-template/schema.ts | 1 + .../edit-template/utils.ts | 4 + 7 files changed, 266 insertions(+), 39 deletions(-) diff --git a/api/apps/restful_apis/document_api.py b/api/apps/restful_apis/document_api.py index f37d4e9cd7..c5aae42899 100644 --- a/api/apps/restful_apis/document_api.py +++ b/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 diff --git a/api/apps/services/document_api_service.py b/api/apps/services/document_api_service.py index d7b45e6218..6292f92044 100644 --- a/api/apps/services/document_api_service.py +++ b/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: diff --git a/api/db/services/compilation_template_group_service.py b/api/db/services/compilation_template_group_service.py index cd5d5c7efd..81c6f8b423 100644 --- a/api/db/services/compilation_template_group_service.py +++ b/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 diff --git a/rag/flow/compiler/compiler.py b/rag/flow/compiler/compiler.py index d8f427f45e..081ac53345 100644 --- a/rag/flow/compiler/compiler.py +++ b/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, diff --git a/web/src/interfaces/request/compilation-template.ts b/web/src/interfaces/request/compilation-template.ts index fde2677aa9..fe64c8ecd0 100644 --- a/web/src/interfaces/request/compilation-template.ts +++ b/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; diff --git a/web/src/pages/user-setting/compilation-templates/edit-template/schema.ts b/web/src/pages/user-setting/compilation-templates/edit-template/schema.ts index d17f80705a..8c01d58cce 100644 --- a/web/src/pages/user-setting/compilation-templates/edit-template/schema.ts +++ b/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')), diff --git a/web/src/pages/user-setting/compilation-templates/edit-template/utils.ts b/web/src/pages/user-setting/compilation-templates/edit-template/utils.ts index d896a5d48d..13c295345d 100644 --- a/web/src/pages/user-setting/compilation-templates/edit-template/utils.ts +++ b/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,