Add comprehensive document metadata support in Knowledge Base workflow node

api/core/workflow/nodes/knowledge_index/entities.py

@@ -150,6 +150,15 @@ class ParentChildStructureChunk(BaseModel):
     data_source_info: Union[FileInfo, OnlineDocumentInfo, WebsiteInfo]
 
 
+class DocMetadata(BaseModel):
+    """
+    Doc Metadata.
+    """
+
+    metadata_id: str
+    value: str | int | float | list[str]
+
+
 class KnowledgeIndexNodeData(BaseNodeData):
     """
     Knowledge index Node Data.
@@ -158,5 +167,6 @@ class KnowledgeIndexNodeData(BaseNodeData):
     type: str = "knowledge-index"
     chunk_structure: str
     index_chunk_variable_selector: list[str]
+    doc_metadata: list[DocMetadata] | None = None
     indexing_technique: str | None = None
     summary_index_setting: dict | None = None

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -2,11 +2,12 @@
 import datetime
 import logging
 import time
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from typing import Any
 
 from flask import current_app
 from sqlalchemy import func, select
+from sqlalchemy.orm import attributes
 
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.rag.index_processor.index_processor_factory import IndexProcessorFactory
@@ -18,7 +19,14 @@
 from core.workflow.nodes.base.template import Template
 from core.workflow.runtime import VariablePool
 from extensions.ext_database import db
-from models.dataset import Dataset, Document, DocumentSegment, DocumentSegmentSummary
+from models.dataset import (
+    Dataset,
+    DatasetMetadata,
+    DatasetMetadataBinding,
+    Document,
+    DocumentSegment,
+    DocumentSegmentSummary,
+)
 from services.summary_index_service import SummaryIndexService
 from tasks.generate_summary_index_task import generate_summary_index_task
 
@@ -29,6 +37,9 @@
 
 logger = logging.getLogger(__name__)
 
+# Constant for built-in metadata identifier
+BUILT_IN_METADATA_ID = "built-in"
+
 default_retrieval_model = {
     "search_method": RetrievalMethod.SEMANTIC_SEARCH,
     "reranking_enable": False,
@@ -192,6 +203,86 @@ def _invoke_knowledge_index(
             }
         )
 
+        # Process doc_metadata before commit to ensure it's saved with the same document object
+        if node_data.doc_metadata:
+            try:
+                # Fetch metadata definitions for name mapping
+                metadata_name_map: dict[str, str] = {}
+                dataset_metadatas = db.session.scalars(
+                    select(DatasetMetadata).where(DatasetMetadata.dataset_id == dataset.id)
+                ).all()
+                for md in dataset_metadatas:
+                    metadata_name_map[md.id] = md.name
+
+                # Collect valid metadata IDs (excluding built-in)
+                valid_metadata_ids = [
+                    item.metadata_id
+                    for item in node_data.doc_metadata
+                    if item.metadata_id != BUILT_IN_METADATA_ID and item.metadata_id in metadata_name_map
+                ]
+
+                # Batch fetch existing bindings to avoid N+1 query
+                existing_binding_ids: set[str] = set()
+                if valid_metadata_ids:
+                    existing_bindings = db.session.scalars(
+                        select(DatasetMetadataBinding.metadata_id).where(
+                            DatasetMetadataBinding.dataset_id == dataset.id,
+                            DatasetMetadataBinding.document_id == doc_id_value,
+                            DatasetMetadataBinding.metadata_id.in_(valid_metadata_ids),
+                        )
+                    ).all()
+                    existing_binding_ids = set(existing_bindings)
+
+                doc_metadata_dict = document.doc_metadata or {}
+
+                for item in node_data.doc_metadata:
+                    # Skip built-in fields
+                    if item.metadata_id == BUILT_IN_METADATA_ID:
+                        continue
+
+                    # Resolve Name
+                    md_name = metadata_name_map.get(item.metadata_id)
+                    if not md_name:
+                        logger.warning("[KnowledgeIndexNode] metadata_id %s not found, skipping", item.metadata_id)
+                        continue
+
+                    # Resolve Value
+                    value = item.value
+                    if isinstance(value, list):
+                        var_obj = variable_pool.get(value)
+                        if var_obj:
+                            value = var_obj.to_object()
+                        else:
+                            # Variable not found - raise error to notify user of configuration issue
+                            variable_path = ".".join(value)
+                            raise KnowledgeIndexNodeError(
+                                f"Variable '{variable_path}' not found for metadata '{md_name}'. "
+                                f"Please check your variable configuration."
+                            )
+
+                    if value is not None:
+                        doc_metadata_dict[md_name] = value
+
+                    # Create DatasetMetadataBinding if not exists
+                    if item.metadata_id not in existing_binding_ids:
+                        binding = DatasetMetadataBinding(
+                            tenant_id=dataset.tenant_id,
+                            dataset_id=dataset.id,
+                            metadata_id=item.metadata_id,
+                            document_id=doc_id_value,
+                            created_by=self.user_id,
+                        )
+                        db.session.add(binding)
+                        existing_binding_ids.add(item.metadata_id)  # Prevent duplicate in same batch
+
+                document.doc_metadata = doc_metadata_dict
+                # Force SQLAlchemy to recognize the change to the JSON field
+                attributes.flag_modified(document, "doc_metadata")
+
+            except Exception as e:
+                logger.exception("[KnowledgeIndexNode] Failed to process doc_metadata")
+                raise KnowledgeIndexNodeError(f"Failed to process document metadata: {e}") from e
+
         db.session.commit()
 
         # Generate summary index if enabled
@@ -522,3 +613,28 @@ def get_streaming_template(self) -> Template:
             Template instance for this knowledge index node
         """
         return Template(segments=[])
+
+    @classmethod
+    def _extract_variable_selector_to_variable_mapping(
+        cls, *, graph_config: Mapping[str, Any], node_id: str, node_data: Mapping[str, Any]
+    ) -> Mapping[str, Sequence[str]]:
+        """
+        Extract variable selector to variable mapping
+        :param graph_config: graph config
+        :param node_id: node id
+        :param node_data: node data
+        :return:
+        """
+        variable_mapping = {}
+        node_data_obj = KnowledgeIndexNodeData(**node_data)
+
+        # index chunk variable
+        variable_mapping[node_id + ".index_chunk_variable_selector"] = node_data_obj.index_chunk_variable_selector
+
+        # doc_metadata variables
+        if node_data_obj.doc_metadata:
+            for item in node_data_obj.doc_metadata:
+                if isinstance(item.value, list):
+                    variable_mapping[node_id + "." + item.metadata_id] = item.value
+
+        return variable_mapping

api/services/dataset_service.py

@@ -41,6 +41,8 @@
     Dataset,
     DatasetAutoDisableLog,
     DatasetCollectionBinding,
+    DatasetMetadata,
+    DatasetMetadataBinding,
     DatasetPermission,
     DatasetPermissionEnum,
     DatasetProcessRule,
@@ -1902,6 +1904,36 @@ def save_document_with_dataset_id(
                         else default_retrieval_model
                     )
 
+        # Handle metadata configuration
+        # 1. Enable built-in metadata if requested
+        if knowledge_config.enable_built_in_metadata and not dataset.built_in_field_enabled:
+            dataset.built_in_field_enabled = True
+            db.session.add(dataset)
+
+        # 2. Process custom metadata - validate and build dict
+        custom_metadata: dict = {}
+        metadata_bindings_to_create: list[tuple[str, str]] = []  # (metadata_id, metadata_name)
+        if knowledge_config.doc_metadata:
+            # Batch fetch all metadata definitions to avoid N+1 query
+            metadata_ids = [item.metadata_id for item in knowledge_config.doc_metadata]
+            metadata_defs = (
+                db.session.query(DatasetMetadata)
+                .filter(
+                    DatasetMetadata.id.in_(metadata_ids),
+                    DatasetMetadata.dataset_id == dataset.id,
+                )
+                .all()
+            )
+            metadata_map = {md.id: md for md in metadata_defs}
+
+            for item in knowledge_config.doc_metadata:
+                # Validate metadata_id belongs to this dataset
+                metadata_def = metadata_map.get(item.metadata_id)
+                if not metadata_def:
+                    raise ValueError(f"Metadata with id '{item.metadata_id}' not found in this dataset")
+                custom_metadata[metadata_def.name] = item.value
+                metadata_bindings_to_create.append((item.metadata_id, metadata_def.name))
+
         documents = []
         if knowledge_config.original_document_id:
             document = DocumentService.update_document_with_dataset_id(dataset, knowledge_config, account)
@@ -2024,6 +2056,7 @@ def save_document_with_dataset_id(
                                     account,
                                     file.name,
                                     batch,
+                                    custom_metadata=custom_metadata or None,
                                 )
                                 db.session.add(document)
                                 db.session.flush()
@@ -2076,6 +2109,7 @@ def save_document_with_dataset_id(
                                         account,
                                         truncated_page_name,
                                         batch,
+                                        custom_metadata=custom_metadata or None,
                                     )
                                     db.session.add(document)
                                     db.session.flush()
@@ -2116,6 +2150,7 @@ def save_document_with_dataset_id(
                                 account,
                                 document_name,
                                 batch,
+                                custom_metadata=custom_metadata or None,
                             )
                             db.session.add(document)
                             db.session.flush()
@@ -2124,6 +2159,20 @@ def save_document_with_dataset_id(
                             position += 1
                     db.session.commit()
 
+                    # Create DatasetMetadataBinding records for custom metadata
+                    if metadata_bindings_to_create and document_ids:
+                        for doc_id in document_ids:
+                            for metadata_id, _ in metadata_bindings_to_create:
+                                binding = DatasetMetadataBinding(
+                                    tenant_id=dataset.tenant_id,
+                                    dataset_id=dataset.id,
+                                    document_id=doc_id,
+                                    metadata_id=metadata_id,
+                                    created_by=account.id,
+                                )
+                                db.session.add(binding)
+                        db.session.commit()
+
                     # trigger async task
                     if document_ids:
                         DocumentIndexingTaskProxy(dataset.tenant_id, dataset.id, document_ids).delay()
@@ -2436,6 +2485,7 @@ def build_document(
         account: Account,
         name: str,
         batch: str,
+        custom_metadata: dict | None = None,
     ):
         # Set need_summary based on dataset's summary_index_setting
         need_summary = False
@@ -2466,6 +2516,9 @@ def build_document(
                 BuiltInField.last_update_date: datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%d %H:%M:%S"),
                 BuiltInField.source: data_source_type,
             }
+        # Merge custom metadata if provided
+        if custom_metadata:
+            doc_metadata.update(custom_metadata)
         if doc_metadata:
             document.doc_metadata = doc_metadata
         return document

api/services/entities/knowledge_entities/knowledge_entities.py

@@ -112,6 +112,11 @@ class MetaDataConfig(BaseModel):
     doc_metadata: dict
 
 
+class DocumentMetadataInput(BaseModel):
+    metadata_id: str
+    value: str | int | float | None = None
+
+
 class KnowledgeConfig(BaseModel):
     original_document_id: str | None = None
     duplicate: bool = True
@@ -126,6 +131,8 @@ class KnowledgeConfig(BaseModel):
     embedding_model_provider: str | None = None
     name: str | None = None
     is_multimodal: bool = False
+    enable_built_in_metadata: bool = False
+    doc_metadata: list[DocumentMetadataInput] | None = None
 
 
 class SegmentCreateArgs(BaseModel):