evaluations

Co-authored-by: jiasiqi <jiasiqi3@tal.com>

api/controllers/console/__init__.py

@@ -115,6 +115,12 @@ from .explore import (
     trial,
 )
 
+# Import evaluation controllers
+from .evaluation import evaluation
+
+# Import snippet controllers
+from .snippets import snippet_workflow
+
 # Import tag controllers
 from .tag import tags
 
@@ -128,6 +134,7 @@ from .workspace import (
     model_providers,
     models,
     plugin,
+    snippets,
     tool_providers,
     trigger_providers,
     workspace,
@@ -165,6 +172,7 @@ __all__ = [
     "datasource_content_preview",
     "email_register",
     "endpoint",
+    "evaluation",
     "extension",
     "external",
     "feature",
@@ -197,6 +205,8 @@ __all__ = [
     "saved_message",
     "setup",
     "site",
+    "snippet_workflow",
+    "snippets",
     "spec",
     "statistic",
     "tags",

api/controllers/console/datasets/datasets_document.py

@@ -1339,18 +1339,6 @@ class DocumentGenerateSummaryApi(Resource):
             missing_ids = set(document_list) - found_ids
             raise NotFound(f"Some documents not found: {list(missing_ids)}")
 
-        # Update need_summary to True for documents that don't have it set
-        # This handles the case where documents were created when summary_index_setting was disabled
-        documents_to_update = [doc for doc in documents if not doc.need_summary and doc.doc_form != "qa_model"]
-
-        if documents_to_update:
-            document_ids_to_update = [str(doc.id) for doc in documents_to_update]
-            DocumentService.update_documents_need_summary(
-                dataset_id=dataset_id,
-                document_ids=document_ids_to_update,
-                need_summary=True,
-            )
-
         # Dispatch async tasks for each document
         for document in documents:
             # Skip qa_model documents as they don't generate summaries

api/controllers/console/evaluation/__init__.py

@@ -0,0 +1 @@
+# Evaluation controller module

api/controllers/console/evaluation/evaluation.py

@@ -0,0 +1,288 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar, Union
+
+from flask import request
+from flask_restx import Resource, fields
+from pydantic import BaseModel, Field
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models import App
+from models.snippet import CustomizedSnippet
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Valid evaluation target types
+EVALUATE_TARGET_TYPES = {"app", "snippets"}
+
+
+class VersionQuery(BaseModel):
+    """Query parameters for version endpoint."""
+
+    version: str
+
+
+register_schema_models(
+    console_ns,
+    VersionQuery,
+)
+
+
+# Response field definitions
+file_info_fields = {
+    "id": fields.String,
+    "name": fields.String,
+}
+
+evaluation_log_fields = {
+    "created_at": TimestampField,
+    "created_by": fields.String,
+    "test_file": fields.Nested(
+        console_ns.model(
+            "EvaluationTestFile",
+            file_info_fields,
+        )
+    ),
+    "result_file": fields.Nested(
+        console_ns.model(
+            "EvaluationResultFile",
+            file_info_fields,
+        ),
+        allow_null=True,
+    ),
+    "version": fields.String,
+}
+
+evaluation_log_list_model = console_ns.model(
+    "EvaluationLogList",
+    {
+        "data": fields.List(fields.Nested(console_ns.model("EvaluationLog", evaluation_log_fields))),
+    },
+)
+
+customized_matrix_fields = {
+    "evaluation_workflow_id": fields.String,
+    "input_fields": fields.Raw,
+    "output_fields": fields.Raw,
+}
+
+condition_fields = {
+    "name": fields.List(fields.String),
+    "comparison_operator": fields.String,
+    "value": fields.String,
+}
+
+judgement_conditions_fields = {
+    "logical_operator": fields.String,
+    "conditions": fields.List(fields.Nested(console_ns.model("EvaluationCondition", condition_fields))),
+}
+
+evaluation_detail_fields = {
+    "evaluation_model": fields.String,
+    "evaluation_model_provider": fields.String,
+    "customized_matrix": fields.Nested(
+        console_ns.model("EvaluationCustomizedMatrix", customized_matrix_fields),
+        allow_null=True,
+    ),
+    "judgement_conditions": fields.Nested(
+        console_ns.model("EvaluationJudgementConditions", judgement_conditions_fields),
+        allow_null=True,
+    ),
+}
+
+evaluation_detail_model = console_ns.model("EvaluationDetail", evaluation_detail_fields)
+
+
+def get_evaluation_target(view_func: Callable[P, R]):
+    """
+    Decorator to resolve polymorphic evaluation target (app or snippet).
+
+    Validates the target_type parameter and fetches the corresponding
+    model (App or CustomizedSnippet) with tenant isolation.
+    """
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        target_type = kwargs.get("evaluate_target_type")
+        target_id = kwargs.get("evaluate_target_id")
+
+        if target_type not in EVALUATE_TARGET_TYPES:
+            raise NotFound(f"Invalid evaluation target type: {target_type}")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        target_id = str(target_id)
+
+        # Remove path parameters
+        del kwargs["evaluate_target_type"]
+        del kwargs["evaluate_target_id"]
+
+        target: Union[App, CustomizedSnippet] | None = None
+
+        if target_type == "app":
+            target = (
+                db.session.query(App).where(App.id == target_id, App.tenant_id == current_tenant_id).first()
+            )
+        elif target_type == "snippets":
+            target = (
+                db.session.query(CustomizedSnippet)
+                .where(CustomizedSnippet.id == target_id, CustomizedSnippet.tenant_id == current_tenant_id)
+                .first()
+            )
+
+        if not target:
+            raise NotFound(f"{str(target_type)} not found")
+
+        kwargs["target"] = target
+        kwargs["target_type"] = target_type
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/dataset-template/download")
+class EvaluationDatasetTemplateDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_dataset_template")
+    @console_ns.response(200, "Template download URL generated successfully")
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    @edit_permission_required
+    def post(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Download evaluation dataset template.
+
+        Generates a download URL for the evaluation dataset template
+        based on the target type (app or snippets).
+        """
+        # TODO: Implement actual template generation logic
+        # This is a placeholder implementation
+        return {
+            "download_url": f"/api/evaluation/{target_type}/{target.id}/template.csv",
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation")
+class EvaluationDetailApi(Resource):
+    @console_ns.doc("get_evaluation_detail")
+    @console_ns.response(200, "Evaluation details retrieved successfully", evaluation_detail_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation details for the target.
+
+        Returns evaluation configuration including model settings,
+        customized matrix, and judgement conditions.
+        """
+        # TODO: Implement actual evaluation detail retrieval
+        # This is a placeholder implementation
+        return {
+            "evaluation_model": None,
+            "evaluation_model_provider": None,
+            "customized_matrix": None,
+            "judgement_conditions": None,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/logs")
+class EvaluationLogsApi(Resource):
+    @console_ns.doc("get_evaluation_logs")
+    @console_ns.response(200, "Evaluation logs retrieved successfully", evaluation_log_list_model)
+    @console_ns.response(404, "Target not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get offline evaluation logs for the target.
+
+        Returns a list of evaluation runs with test files,
+        result files, and version information.
+        """
+        # TODO: Implement actual evaluation logs retrieval
+        # This is a placeholder implementation
+        return {
+            "data": [],
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/files/<uuid:file_id>")
+class EvaluationFileDownloadApi(Resource):
+    @console_ns.doc("download_evaluation_file")
+    @console_ns.response(200, "File download URL generated successfully")
+    @console_ns.response(404, "Target or file not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str, file_id: str):
+        """
+        Download evaluation test file or result file.
+
+        Returns file information and download URL for the specified file.
+        """
+        file_id = str(file_id)
+
+        # TODO: Implement actual file download logic
+        # This is a placeholder implementation
+        return {
+            "created_at": None,
+            "created_by": None,
+            "test_file": None,
+            "result_file": None,
+            "version": None,
+        }
+
+
+@console_ns.route("/<string:evaluate_target_type>/<uuid:evaluate_target_id>/evaluation/version")
+class EvaluationVersionApi(Resource):
+    @console_ns.doc("get_evaluation_version_detail")
+    @console_ns.expect(console_ns.models.get(VersionQuery.__name__))
+    @console_ns.response(200, "Version details retrieved successfully")
+    @console_ns.response(404, "Target or version not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_evaluation_target
+    def get(self, target: Union[App, CustomizedSnippet], target_type: str):
+        """
+        Get evaluation target version details.
+
+        Returns the workflow graph for the specified version.
+        """
+        version = request.args.get("version")
+
+        if not version:
+            return {"message": "version parameter is required"}, 400
+
+        # TODO: Implement actual version detail retrieval
+        # For now, return the current graph if available
+        graph = {}
+        if target_type == "snippets" and isinstance(target, CustomizedSnippet):
+            graph = target.graph_dict
+
+        return {
+            "graph": graph,
+        }

api/controllers/console/snippets/payloads.py

@@ -0,0 +1,75 @@
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class SnippetListQuery(BaseModel):
+    """Query parameters for listing snippets."""
+
+    page: int = Field(default=1, ge=1, le=99999)
+    limit: int = Field(default=20, ge=1, le=100)
+    keyword: str | None = None
+
+
+class IconInfo(BaseModel):
+    """Icon information model."""
+
+    icon: str | None = None
+    icon_type: Literal["emoji", "image"] | None = None
+    icon_background: str | None = None
+    icon_url: str | None = None
+
+
+class InputFieldDefinition(BaseModel):
+    """Input field definition for snippet parameters."""
+
+    default: str | None = None
+    hint: bool | None = None
+    label: str | None = None
+    max_length: int | None = None
+    options: list[str] | None = None
+    placeholder: str | None = None
+    required: bool | None = None
+    type: str | None = None  # e.g., "text-input"
+
+
+class CreateSnippetPayload(BaseModel):
+    """Payload for creating a new snippet."""
+
+    name: str = Field(..., min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    type: Literal["node", "group"] = "node"
+    icon_info: IconInfo | None = None
+    graph: dict[str, Any] | None = None
+    input_fields: list[InputFieldDefinition] | None = Field(default_factory=list)
+
+
+class UpdateSnippetPayload(BaseModel):
+    """Payload for updating a snippet."""
+
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=2000)
+    icon_info: IconInfo | None = None
+
+
+class SnippetDraftSyncPayload(BaseModel):
+    """Payload for syncing snippet draft workflow."""
+
+    graph: dict[str, Any]
+    hash: str | None = None
+    environment_variables: list[dict[str, Any]] | None = None
+    conversation_variables: list[dict[str, Any]] | None = None
+    input_variables: list[dict[str, Any]] | None = None
+
+
+class WorkflowRunQuery(BaseModel):
+    """Query parameters for workflow runs."""
+
+    last_id: str | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class PublishWorkflowPayload(BaseModel):
+    """Payload for publishing snippet workflow."""
+
+    knowledge_base_setting: dict[str, Any] | None = None

api/controllers/console/snippets/snippet_workflow.py

@@ -0,0 +1,306 @@
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import ParamSpec, TypeVar
+
+from flask import request
+from flask_restx import Resource, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.app.error import DraftWorkflowNotExist, DraftWorkflowNotSync
+from controllers.console.app.workflow import workflow_model
+from controllers.console.app.workflow_run import (
+    workflow_run_detail_model,
+    workflow_run_node_execution_list_model,
+    workflow_run_pagination_model,
+)
+from controllers.console.snippets.payloads import (
+    PublishWorkflowPayload,
+    SnippetDraftSyncPayload,
+    WorkflowRunQuery,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from factories import variable_factory
+from libs.helper import TimestampField
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import CustomizedSnippet
+from services.errors.app import WorkflowHashNotEqualError
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetDraftSyncPayload,
+    WorkflowRunQuery,
+    PublishWorkflowPayload,
+)
+
+
+class SnippetNotFoundError(Exception):
+    """Snippet not found error."""
+
+    pass
+
+
+def get_snippet(view_func: Callable[P, R]):
+    """Decorator to fetch and validate snippet access."""
+
+    @wraps(view_func)
+    def decorated_view(*args: P.args, **kwargs: P.kwargs):
+        if not kwargs.get("snippet_id"):
+            raise ValueError("missing snippet_id in path parameters")
+
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet_id = str(kwargs.get("snippet_id"))
+        del kwargs["snippet_id"]
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=snippet_id,
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        kwargs["snippet"] = snippet
+
+        return view_func(*args, **kwargs)
+
+    return decorated_view
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft")
+class SnippetDraftWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_draft_workflow")
+    @console_ns.response(200, "Draft workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet or draft workflow not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get draft workflow for snippet."""
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_draft_workflow(snippet=snippet)
+
+        if not workflow:
+            raise DraftWorkflowNotExist()
+
+        return workflow
+
+    @console_ns.doc("sync_snippet_draft_workflow")
+    @console_ns.expect(console_ns.models.get(SnippetDraftSyncPayload.__name__))
+    @console_ns.response(200, "Draft workflow synced successfully")
+    @console_ns.response(400, "Hash mismatch")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Sync draft workflow for snippet."""
+        current_user, _ = current_account_with_tenant()
+
+        payload = SnippetDraftSyncPayload.model_validate(console_ns.payload or {})
+
+        try:
+            environment_variables_list = payload.environment_variables or []
+            environment_variables = [
+                variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
+            ]
+            conversation_variables_list = payload.conversation_variables or []
+            conversation_variables = [
+                variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
+            ]
+            snippet_service = SnippetService()
+            workflow = snippet_service.sync_draft_workflow(
+                snippet=snippet,
+                graph=payload.graph,
+                unique_hash=payload.hash,
+                account=current_user,
+                environment_variables=environment_variables,
+                conversation_variables=conversation_variables,
+                input_variables=payload.input_variables,
+            )
+        except WorkflowHashNotEqualError:
+            raise DraftWorkflowNotSync()
+
+        return {
+            "result": "success",
+            "hash": workflow.unique_hash,
+            "updated_at": TimestampField().format(workflow.updated_at or workflow.created_at),
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/draft/config")
+class SnippetDraftConfigApi(Resource):
+    @console_ns.doc("get_snippet_draft_config")
+    @console_ns.response(200, "Draft config retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get snippet draft workflow configuration limits."""
+        return {
+            "parallel_depth_limit": 3,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/publish")
+class SnippetPublishedWorkflowApi(Resource):
+    @console_ns.doc("get_snippet_published_workflow")
+    @console_ns.response(200, "Published workflow retrieved successfully", workflow_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    @marshal_with(workflow_model)
+    def get(self, snippet: CustomizedSnippet):
+        """Get published workflow for snippet."""
+        if not snippet.is_published:
+            return None
+
+        snippet_service = SnippetService()
+        workflow = snippet_service.get_published_workflow(snippet=snippet)
+
+        return workflow
+
+    @console_ns.doc("publish_snippet_workflow")
+    @console_ns.expect(console_ns.models.get(PublishWorkflowPayload.__name__))
+    @console_ns.response(200, "Workflow published successfully")
+    @console_ns.response(400, "No draft workflow found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def post(self, snippet: CustomizedSnippet):
+        """Publish snippet workflow."""
+        current_user, _ = current_account_with_tenant()
+        snippet_service = SnippetService()
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            try:
+                workflow = snippet_service.publish_workflow(
+                    session=session,
+                    snippet=snippet,
+                    account=current_user,
+                )
+                workflow_created_at = TimestampField().format(workflow.created_at)
+                session.commit()
+            except ValueError as e:
+                return {"message": str(e)}, 400
+
+        return {
+            "result": "success",
+            "created_at": workflow_created_at,
+        }
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflows/default-workflow-block-configs")
+class SnippetDefaultBlockConfigsApi(Resource):
+    @console_ns.doc("get_snippet_default_block_configs")
+    @console_ns.response(200, "Default block configs retrieved successfully")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @edit_permission_required
+    def get(self, snippet: CustomizedSnippet):
+        """Get default block configurations for snippet workflow."""
+        snippet_service = SnippetService()
+        return snippet_service.get_default_block_configs()
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs")
+class SnippetWorkflowRunsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_runs")
+    @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_pagination_model)
+    def get(self, snippet: CustomizedSnippet):
+        """List workflow runs for snippet."""
+        query = WorkflowRunQuery.model_validate(
+            {
+                "last_id": request.args.get("last_id"),
+                "limit": request.args.get("limit", type=int, default=20),
+            }
+        )
+        args = {
+            "last_id": query.last_id,
+            "limit": query.limit,
+        }
+
+        snippet_service = SnippetService()
+        result = snippet_service.get_snippet_workflow_runs(snippet=snippet, args=args)
+
+        return result
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>")
+class SnippetWorkflowRunDetailApi(Resource):
+    @console_ns.doc("get_snippet_workflow_run_detail")
+    @console_ns.response(200, "Workflow run detail retrieved successfully", workflow_run_detail_model)
+    @console_ns.response(404, "Workflow run not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_detail_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """Get workflow run detail for snippet."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        workflow_run = snippet_service.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+
+        if not workflow_run:
+            raise NotFound("Workflow run not found")
+
+        return workflow_run
+
+
+@console_ns.route("/snippets/<uuid:snippet_id>/workflow-runs/<uuid:run_id>/node-executions")
+class SnippetWorkflowRunNodeExecutionsApi(Resource):
+    @console_ns.doc("list_snippet_workflow_run_node_executions")
+    @console_ns.response(200, "Node executions retrieved successfully", workflow_run_node_execution_list_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_snippet
+    @marshal_with(workflow_run_node_execution_list_model)
+    def get(self, snippet: CustomizedSnippet, run_id):
+        """List node executions for a workflow run."""
+        run_id = str(run_id)
+
+        snippet_service = SnippetService()
+        node_executions = snippet_service.get_snippet_workflow_run_node_executions(
+            snippet=snippet,
+            run_id=run_id,
+        )
+
+        return {"data": node_executions}

api/controllers/console/workspace/snippets.py

@@ -0,0 +1,202 @@
+import logging
+
+from flask import request
+from flask_restx import Resource, marshal, marshal_with
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import NotFound
+
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
+from controllers.console.snippets.payloads import (
+    CreateSnippetPayload,
+    SnippetListQuery,
+    UpdateSnippetPayload,
+)
+from controllers.console.wraps import (
+    account_initialization_required,
+    edit_permission_required,
+    setup_required,
+)
+from extensions.ext_database import db
+from fields.snippet_fields import snippet_fields, snippet_list_fields, snippet_pagination_fields
+from libs.login import current_account_with_tenant, login_required
+from models.snippet import SnippetType
+from services.snippet_service import SnippetService
+
+logger = logging.getLogger(__name__)
+
+# Register Pydantic models with Swagger
+register_schema_models(
+    console_ns,
+    SnippetListQuery,
+    CreateSnippetPayload,
+    UpdateSnippetPayload,
+)
+
+# Create namespace models for marshaling
+snippet_model = console_ns.model("Snippet", snippet_fields)
+snippet_list_model = console_ns.model("SnippetList", snippet_list_fields)
+snippet_pagination_model = console_ns.model("SnippetPagination", snippet_pagination_fields)
+
+
+@console_ns.route("/workspaces/current/customized-snippets")
+class CustomizedSnippetsApi(Resource):
+    @console_ns.doc("list_customized_snippets")
+    @console_ns.expect(console_ns.models.get(SnippetListQuery.__name__))
+    @console_ns.response(200, "Snippets retrieved successfully", snippet_pagination_model)
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List customized snippets with pagination and search."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        query_params = request.args.to_dict()
+        query = SnippetListQuery.model_validate(query_params)
+
+        snippets, total, has_more = SnippetService.get_snippets(
+            tenant_id=current_tenant_id,
+            page=query.page,
+            limit=query.limit,
+            keyword=query.keyword,
+        )
+
+        return {
+            "data": marshal(snippets, snippet_list_fields),
+            "page": query.page,
+            "limit": query.limit,
+            "total": total,
+            "has_more": has_more,
+        }, 200
+
+    @console_ns.doc("create_customized_snippet")
+    @console_ns.expect(console_ns.models.get(CreateSnippetPayload.__name__))
+    @console_ns.response(201, "Snippet created successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def post(self):
+        """Create a new customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        payload = CreateSnippetPayload.model_validate(console_ns.payload or {})
+
+        try:
+            snippet_type = SnippetType(payload.type)
+        except ValueError:
+            snippet_type = SnippetType.NODE
+
+        try:
+            snippet = SnippetService.create_snippet(
+                tenant_id=current_tenant_id,
+                name=payload.name,
+                description=payload.description,
+                snippet_type=snippet_type,
+                icon_info=payload.icon_info.model_dump() if payload.icon_info else None,
+                graph=payload.graph,
+                input_fields=[f.model_dump() for f in payload.input_fields] if payload.input_fields else None,
+                account=current_user,
+            )
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 201
+
+
+@console_ns.route("/workspaces/current/customized-snippets/<uuid:snippet_id>")
+class CustomizedSnippetDetailApi(Resource):
+    @console_ns.doc("get_customized_snippet")
+    @console_ns.response(200, "Snippet retrieved successfully", snippet_model)
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, snippet_id: str):
+        """Get customized snippet details."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("update_customized_snippet")
+    @console_ns.expect(console_ns.models.get(UpdateSnippetPayload.__name__))
+    @console_ns.response(200, "Snippet updated successfully", snippet_model)
+    @console_ns.response(400, "Invalid request or name already exists")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def patch(self, snippet_id: str):
+        """Update customized snippet."""
+        current_user, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        payload = UpdateSnippetPayload.model_validate(console_ns.payload or {})
+        update_data = payload.model_dump(exclude_unset=True)
+
+        if "icon_info" in update_data and update_data["icon_info"] is not None:
+            update_data["icon_info"] = payload.icon_info.model_dump() if payload.icon_info else None
+
+        if not update_data:
+            return {"message": "No valid fields to update"}, 400
+
+        try:
+            with Session(db.engine, expire_on_commit=False) as session:
+                snippet = session.merge(snippet)
+                snippet = SnippetService.update_snippet(
+                    session=session,
+                    snippet=snippet,
+                    account_id=current_user.id,
+                    data=update_data,
+                )
+                session.commit()
+        except ValueError as e:
+            return {"message": str(e)}, 400
+
+        return marshal(snippet, snippet_fields), 200
+
+    @console_ns.doc("delete_customized_snippet")
+    @console_ns.response(204, "Snippet deleted successfully")
+    @console_ns.response(404, "Snippet not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, snippet_id: str):
+        """Delete customized snippet."""
+        _, current_tenant_id = current_account_with_tenant()
+
+        snippet = SnippetService.get_snippet_by_id(
+            snippet_id=str(snippet_id),
+            tenant_id=current_tenant_id,
+        )
+
+        if not snippet:
+            raise NotFound("Snippet not found")
+
+        with Session(db.engine) as session:
+            snippet = session.merge(snippet)
+            SnippetService.delete_snippet(
+                session=session,
+                snippet=snippet,
+            )
+            session.commit()
+
+        return "", 204

api/core/indexing_runner.py

@@ -369,9 +369,7 @@ class IndexingRunner:
         # Generate summary preview
         summary_index_setting = tmp_processing_rule.get("summary_index_setting")
         if summary_index_setting and summary_index_setting.get("enable") and preview_texts:
-            preview_texts = index_processor.generate_summary_preview(
-                tenant_id, preview_texts, summary_index_setting, doc_language
-            )
+            preview_texts = index_processor.generate_summary_preview(tenant_id, preview_texts, summary_index_setting)
 
         return IndexingEstimate(total_segments=total_segments, preview=preview_texts)
 

api/core/llm_generator/prompts.py

@@ -441,13 +441,11 @@ DEFAULT_GENERATOR_SUMMARY_PROMPT = (
 
 Requirements:
 1. Write a concise summary in plain text
-2. You must write in {language}. No language other than {language} should be used.
+2. Use the same language as the input content
 3. Focus on important facts, concepts, and details
 4. If images are included, describe their key information
 5. Do not use words like "好的", "ok", "I understand", "This text discusses", "The content mentions"
 6. Write directly without extra words
-7. If there is not enough content to generate a meaningful summary, 
-   return an empty string without any explanation or prompt
 
 Output only the summary text. Start summarizing now:
 

api/core/rag/index_processor/index_processor_base.py

@@ -48,22 +48,12 @@ class BaseIndexProcessor(ABC):
 
     @abstractmethod
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each segment in preview_texts, generate a summary using LLM and attach it to the segment.
         The summary can be stored in a new attribute, e.g., summary.
         This method should be implemented by subclasses.
-
-        Args:
-            tenant_id: Tenant ID
-            preview_texts: List of preview details to generate summaries for
-            summary_index_setting: Summary index configuration
-            doc_language: Optional document language to ensure summary is generated in the correct language
         """
         raise NotImplementedError
 

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -275,11 +275,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("Chunks is not a list")
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each segment, concurrently call generate_summary to generate a summary
@@ -302,15 +298,11 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             if flask_app:
                 # Ensure Flask app context in worker thread
                 with flask_app.app_context():
-                    summary, _ = self.generate_summary(
-                        tenant_id, preview.content, summary_index_setting, document_language=doc_language
-                    )
+                    summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
                     preview.summary = summary
             else:
                 # Fallback: try without app context (may fail)
-                summary, _ = self.generate_summary(
-                    tenant_id, preview.content, summary_index_setting, document_language=doc_language
-                )
+                summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
                 preview.summary = summary
 
         # Generate summaries concurrently using ThreadPoolExecutor
@@ -364,7 +356,6 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         text: str,
         summary_index_setting: dict | None = None,
         segment_id: str | None = None,
-        document_language: str | None = None,
     ) -> tuple[str, LLMUsage]:
         """
         Generate summary for the given text using ModelInstance.invoke_llm and the default or custom summary prompt,
@@ -375,8 +366,6 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             text: Text content to summarize
             summary_index_setting: Summary index configuration
             segment_id: Optional segment ID to fetch attachments from SegmentAttachmentBinding table
-            document_language: Optional document language (e.g., "Chinese", "English")
-                to ensure summary is generated in the correct language
 
         Returns:
             Tuple of (summary_content, llm_usage) where llm_usage is LLMUsage object
@@ -392,22 +381,8 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             raise ValueError("model_name and model_provider_name are required in summary_index_setting")
 
         # Import default summary prompt
-        is_default_prompt = False
         if not summary_prompt:
             summary_prompt = DEFAULT_GENERATOR_SUMMARY_PROMPT
-            is_default_prompt = True
-
-        # Format prompt with document language only for default prompt
-        # Custom prompts are used as-is to avoid interfering with user-defined templates
-        # If document_language is provided, use it; otherwise, use "the same language as the input content"
-        # This is especially important for image-only chunks where text is empty or minimal
-        if is_default_prompt:
-            language_for_prompt = document_language or "the same language as the input content"
-            try:
-                summary_prompt = summary_prompt.format(language=language_for_prompt)
-            except KeyError:
-                # If default prompt doesn't have {language} placeholder, use it as-is
-                pass
 
         provider_manager = ProviderManager()
         provider_model_bundle = provider_manager.get_provider_model_bundle(

api/core/rag/index_processor/processor/parent_child_index_processor.py

@@ -358,11 +358,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         For each parent chunk in preview_texts, concurrently call generate_summary to generate a summary
@@ -393,7 +389,6 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                         tenant_id=tenant_id,
                         text=preview.content,
                         summary_index_setting=summary_index_setting,
-                        document_language=doc_language,
                     )
                     preview.summary = summary
             else:
@@ -402,7 +397,6 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     tenant_id=tenant_id,
                     text=preview.content,
                     summary_index_setting=summary_index_setting,
-                    document_language=doc_language,
                 )
                 preview.summary = summary
 

api/core/rag/index_processor/processor/qa_index_processor.py

@@ -241,11 +241,7 @@ class QAIndexProcessor(BaseIndexProcessor):
         }
 
     def generate_summary_preview(
-        self,
-        tenant_id: str,
-        preview_texts: list[PreviewDetail],
-        summary_index_setting: dict,
-        doc_language: str | None = None,
+        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
     ) -> list[PreviewDetail]:
         """
         QA model doesn't generate summaries, so this method returns preview_texts unchanged.

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

@@ -78,21 +78,12 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
                 indexing_technique = node_data.indexing_technique or dataset.indexing_technique
                 summary_index_setting = node_data.summary_index_setting or dataset.summary_index_setting
 
-                # Try to get document language if document_id is available
-                doc_language = None
-                document_id = variable_pool.get(["sys", SystemVariableKey.DOCUMENT_ID])
-                if document_id:
-                    document = db.session.query(Document).filter_by(id=document_id.value).first()
-                    if document and document.doc_language:
-                        doc_language = document.doc_language
-
                 outputs = self._get_preview_output_with_summaries(
                     node_data.chunk_structure,
                     chunks,
                     dataset=dataset,
                     indexing_technique=indexing_technique,
                     summary_index_setting=summary_index_setting,
-                    doc_language=doc_language,
                 )
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
@@ -324,7 +315,6 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
         dataset: Dataset,
         indexing_technique: str | None = None,
         summary_index_setting: dict | None = None,
-        doc_language: str | None = None,
     ) -> Mapping[str, Any]:
         """
         Generate preview output with summaries for chunks in preview mode.
@@ -336,7 +326,6 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
             dataset: Dataset object (for tenant_id)
             indexing_technique: Indexing technique from node config or dataset
             summary_index_setting: Summary index setting from node config or dataset
-            doc_language: Optional document language to ensure summary is generated in the correct language
         """
         index_processor = IndexProcessorFactory(chunk_structure).init_index_processor()
         preview_output = index_processor.format_preview(chunks)
@@ -376,7 +365,6 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
                                 tenant_id=dataset.tenant_id,
                                 text=preview_item["content"],
                                 summary_index_setting=summary_index_setting,
-                                document_language=doc_language,
                             )
                             if summary:
                                 preview_item["summary"] = summary
@@ -386,7 +374,6 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
                             tenant_id=dataset.tenant_id,
                             text=preview_item["content"],
                             summary_index_setting=summary_index_setting,
-                            document_language=doc_language,
                         )
                         if summary:
                             preview_item["summary"] = summary

api/fields/snippet_fields.py

@@ -0,0 +1,45 @@
+from flask_restx import fields
+
+from fields.member_fields import simple_account_fields
+from libs.helper import TimestampField
+
+# Snippet list item fields (lightweight for list display)
+snippet_list_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "description": fields.String,
+    "type": fields.String,
+    "version": fields.Integer,
+    "use_count": fields.Integer,
+    "is_published": fields.Boolean,
+    "icon_info": fields.Raw,
+    "created_at": TimestampField,
+    "updated_at": TimestampField,
+}
+
+# Full snippet fields (includes creator info and graph data)
+snippet_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "description": fields.String,
+    "type": fields.String,
+    "version": fields.Integer,
+    "use_count": fields.Integer,
+    "is_published": fields.Boolean,
+    "icon_info": fields.Raw,
+    "graph": fields.Raw(attribute="graph_dict"),
+    "input_fields": fields.Raw(attribute="input_fields_list"),
+    "created_by": fields.Nested(simple_account_fields, attribute="created_by_account", allow_null=True),
+    "created_at": TimestampField,
+    "updated_by": fields.Nested(simple_account_fields, attribute="updated_by_account", allow_null=True),
+    "updated_at": TimestampField,
+}
+
+# Pagination response fields
+snippet_pagination_fields = {
+    "data": fields.List(fields.Nested(snippet_list_fields)),
+    "page": fields.Integer,
+    "limit": fields.Integer,
+    "total": fields.Integer,
+    "has_more": fields.Boolean,
+}

api/migrations/versions/2026_01_27_1815-788d3099ae3a_add_summary_index_feature.py

@@ -51,7 +51,7 @@ def upgrade():
             batch_op.add_column(sa.Column('summary_index_setting', models.types.AdjustedJSON(), nullable=True))
 
         with op.batch_alter_table('documents', schema=None) as batch_op:
-            batch_op.add_column(sa.Column('need_summary', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+            batch_op.add_column(sa.Column('need_summary', sa.Boolean(), server_default=sa.text('false'), nullable=True))
     else:
         # MySQL: Use compatible syntax
         op.create_table(
@@ -83,7 +83,7 @@ def upgrade():
             batch_op.add_column(sa.Column('summary_index_setting', models.types.AdjustedJSON(), nullable=True))
 
         with op.batch_alter_table('documents', schema=None) as batch_op:
-            batch_op.add_column(sa.Column('need_summary', sa.Boolean(), server_default=sa.text('false'), nullable=False))
+            batch_op.add_column(sa.Column('need_summary', sa.Boolean(), server_default=sa.text('false'), nullable=True))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2026_01_29_1200-1c05e80d2380_add_customized_snippets_table.py

@@ -0,0 +1,83 @@
+"""add_customized_snippets_table
+
+Revision ID: 1c05e80d2380
+Revises: 788d3099ae3a
+Create Date: 2026-01-29 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
+# revision identifiers, used by Alembic.
+revision = "1c05e80d2380"
+down_revision = "788d3099ae3a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    conn = op.get_bind()
+
+    if _is_pg(conn):
+        op.create_table(
+            "customized_snippets",
+            sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuidv7()"), nullable=False),
+            sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.Text(), nullable=True),
+            sa.Column("type", sa.String(length=50), server_default=sa.text("'node'"), nullable=False),
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=True),
+            sa.Column("is_published", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+            sa.Column("version", sa.Integer(), server_default=sa.text("1"), nullable=False),
+            sa.Column("use_count", sa.Integer(), server_default=sa.text("0"), nullable=False),
+            sa.Column("icon_info", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+            sa.Column("graph", sa.Text(), nullable=True),
+            sa.Column("input_fields", sa.Text(), nullable=True),
+            sa.Column("created_by", models.types.StringUUID(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("updated_by", models.types.StringUUID(), nullable=True),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+            sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+        )
+    else:
+        op.create_table(
+            "customized_snippets",
+            sa.Column("id", models.types.StringUUID(), nullable=False),
+            sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", models.types.LongText(), nullable=True),
+            sa.Column("type", sa.String(length=50), server_default=sa.text("'node'"), nullable=False),
+            sa.Column("workflow_id", models.types.StringUUID(), nullable=True),
+            sa.Column("is_published", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+            sa.Column("version", sa.Integer(), server_default=sa.text("1"), nullable=False),
+            sa.Column("use_count", sa.Integer(), server_default=sa.text("0"), nullable=False),
+            sa.Column("icon_info", models.types.AdjustedJSON(astext_type=sa.Text()), nullable=True),
+            sa.Column("graph", models.types.LongText(), nullable=True),
+            sa.Column("input_fields", models.types.LongText(), nullable=True),
+            sa.Column("created_by", models.types.StringUUID(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("updated_by", models.types.StringUUID(), nullable=True),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+            sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+        )
+
+    with op.batch_alter_table("customized_snippets", schema=None) as batch_op:
+        batch_op.create_index("customized_snippet_tenant_idx", ["tenant_id"], unique=False)
+
+
+def downgrade():
+    with op.batch_alter_table("customized_snippets", schema=None) as batch_op:
+        batch_op.drop_index("customized_snippet_tenant_idx")
+
+    op.drop_table("customized_snippets")

api/models/__init__.py

@@ -79,6 +79,7 @@ from .provider import (
     TenantDefaultModel,
     TenantPreferredModelProvider,
 )
+from .snippet import CustomizedSnippet, SnippetType
 from .source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from .task import CeleryTask, CeleryTaskSet
 from .tools import (
@@ -138,6 +139,7 @@ __all__ = [
     "Conversation",
     "ConversationVariable",
     "CreatorUserRole",
+    "CustomizedSnippet",
     "DataSourceApiKeyAuthBinding",
     "DataSourceOauthBinding",
     "Dataset",
@@ -179,6 +181,7 @@ __all__ = [
     "RecommendedApp",
     "SavedMessage",
     "Site",
+    "SnippetType",
     "Tag",
     "TagBinding",
     "Tenant",

api/models/dataset.py

@@ -420,7 +420,7 @@ class Document(Base):
     doc_metadata = mapped_column(AdjustedJSON, nullable=True)
     doc_form = mapped_column(String(255), nullable=False, server_default=sa.text("'text_model'"))
     doc_language = mapped_column(String(255), nullable=True)
-    need_summary: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
+    need_summary: Mapped[bool | None] = mapped_column(sa.Boolean, nullable=True, server_default=sa.text("false"))
 
     DATA_SOURCES = ["upload_file", "notion_import", "website_crawl"]
 

api/models/snippet.py

@@ -0,0 +1,96 @@
+import json
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+
+import sqlalchemy as sa
+from sqlalchemy import DateTime, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from libs.uuid_utils import uuidv7
+
+from .account import Account
+from .base import Base
+from .engine import db
+from .types import AdjustedJSON, LongText, StringUUID
+
+
+class SnippetType(StrEnum):
+    """Snippet Type Enum"""
+
+    NODE = "node"
+    GROUP = "group"
+
+
+class CustomizedSnippet(Base):
+    """
+    Customized Snippet Model
+
+    Stores reusable workflow components (nodes or node groups) that can be
+    shared across applications within a workspace.
+    """
+
+    __tablename__ = "customized_snippets"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="customized_snippet_pkey"),
+        sa.Index("customized_snippet_tenant_idx", "tenant_id"),
+        sa.UniqueConstraint("tenant_id", "name", name="customized_snippet_tenant_name_key"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuidv7()))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    description: Mapped[str | None] = mapped_column(LongText, nullable=True)
+    type: Mapped[str] = mapped_column(String(50), nullable=False, server_default=sa.text("'node'"))
+
+    # Workflow reference for published version
+    workflow_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+
+    # State flags
+    is_published: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
+    version: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("1"))
+    use_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
+
+    # Visual customization
+    icon_info: Mapped[dict | None] = mapped_column(AdjustedJSON, nullable=True)
+
+    # Snippet configuration (stored as JSON text)
+    graph: Mapped[str | None] = mapped_column(LongText, nullable=True)
+    input_fields: Mapped[str | None] = mapped_column(LongText, nullable=True)
+
+    # Audit fields
+    created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+    @property
+    def graph_dict(self) -> dict[str, Any]:
+        """Parse graph JSON to dict."""
+        return json.loads(self.graph) if self.graph else {}
+
+    @property
+    def input_fields_list(self) -> list[dict[str, Any]]:
+        """Parse input_fields JSON to list."""
+        return json.loads(self.input_fields) if self.input_fields else []
+
+    @property
+    def created_by_account(self) -> Account | None:
+        """Get the account that created this snippet."""
+        if self.created_by:
+            return db.session.get(Account, self.created_by)
+        return None
+
+    @property
+    def updated_by_account(self) -> Account | None:
+        """Get the account that last updated this snippet."""
+        if self.updated_by:
+            return db.session.get(Account, self.updated_by)
+        return None
+
+    @property
+    def version_str(self) -> str:
+        """Get version as string for API response."""
+        return str(self.version)

api/models/workflow.py

@@ -65,6 +65,7 @@ class WorkflowType(StrEnum):
     WORKFLOW = "workflow"
     CHAT = "chat"
     RAG_PIPELINE = "rag-pipeline"
+    SNIPPET = "snippet"
 
     @classmethod
     def value_of(cls, value: str) -> "WorkflowType":

api/services/dataset_service.py

@@ -16,7 +16,6 @@ from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound
 
 from configs import dify_config
-from core.db.session_factory import session_factory
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.file import helpers as file_helpers
 from core.helper.name_generator import generate_incremental_name
@@ -1389,46 +1388,6 @@ class DocumentService:
         ).all()
         return documents
 
-    @staticmethod
-    def update_documents_need_summary(dataset_id: str, document_ids: Sequence[str], need_summary: bool = True) -> int:
-        """
-        Update need_summary field for multiple documents.
-
-        This method handles the case where documents were created when summary_index_setting was disabled,
-        and need to be updated when summary_index_setting is later enabled.
-
-        Args:
-            dataset_id: Dataset ID
-            document_ids: List of document IDs to update
-            need_summary: Value to set for need_summary field (default: True)
-
-        Returns:
-            Number of documents updated
-        """
-        if not document_ids:
-            return 0
-
-        document_id_list: list[str] = [str(document_id) for document_id in document_ids]
-
-        with session_factory.create_session() as session:
-            updated_count = (
-                session.query(Document)
-                .filter(
-                    Document.id.in_(document_id_list),
-                    Document.dataset_id == dataset_id,
-                    Document.doc_form != "qa_model",  # Skip qa_model documents
-                )
-                .update({Document.need_summary: need_summary}, synchronize_session=False)
-            )
-            session.commit()
-            logger.info(
-                "Updated need_summary to %s for %d documents in dataset %s",
-                need_summary,
-                updated_count,
-                dataset_id,
-            )
-            return updated_count
-
     @staticmethod
     def get_document_download_url(document: Document) -> str:
         """

api/services/rag_pipeline/rag_pipeline_transform_service.py

@@ -174,10 +174,6 @@ class RagPipelineTransformService:
         else:
             dataset.retrieval_model = knowledge_configuration.retrieval_model.model_dump()
 
-        # Copy summary_index_setting from dataset to knowledge_index node configuration
-        if dataset.summary_index_setting:
-            knowledge_configuration.summary_index_setting = dataset.summary_index_setting
-
         knowledge_configuration_dict.update(knowledge_configuration.model_dump())
         node["data"] = knowledge_configuration_dict
         return node

api/services/snippet_service.py

@@ -0,0 +1,542 @@
+import json
+import logging
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.variables.variables import VariableBase
+from core.workflow.enums import NodeType
+from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
+from extensions.ext_database import db
+from libs.infinite_scroll_pagination import InfiniteScrollPagination
+from models import Account
+from models.enums import WorkflowRunTriggeredFrom
+from models.snippet import CustomizedSnippet, SnippetType
+from models.workflow import (
+    Workflow,
+    WorkflowNodeExecutionModel,
+    WorkflowRun,
+    WorkflowType,
+)
+from repositories.factory import DifyAPIRepositoryFactory
+from services.errors.app import WorkflowHashNotEqualError
+
+logger = logging.getLogger(__name__)
+
+
+class SnippetService:
+    """Service for managing customized snippets."""
+
+    def __init__(self, session_maker: sessionmaker | None = None):
+        """Initialize SnippetService with repository dependencies."""
+        if session_maker is None:
+            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+        self._node_execution_service_repo = DifyAPIRepositoryFactory.create_api_workflow_node_execution_repository(
+            session_maker
+        )
+        self._workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+
+    # --- CRUD Operations ---
+
+    @staticmethod
+    def get_snippets(
+        *,
+        tenant_id: str,
+        page: int = 1,
+        limit: int = 20,
+        keyword: str | None = None,
+    ) -> tuple[Sequence[CustomizedSnippet], int, bool]:
+        """
+        Get paginated list of snippets with optional search.
+
+        :param tenant_id: Tenant ID
+        :param page: Page number (1-indexed)
+        :param limit: Number of items per page
+        :param keyword: Optional search keyword for name/description
+        :return: Tuple of (snippets list, total count, has_more flag)
+        """
+        stmt = (
+            select(CustomizedSnippet)
+            .where(CustomizedSnippet.tenant_id == tenant_id)
+            .order_by(CustomizedSnippet.created_at.desc())
+        )
+
+        if keyword:
+            stmt = stmt.where(
+                CustomizedSnippet.name.ilike(f"%{keyword}%") | CustomizedSnippet.description.ilike(f"%{keyword}%")
+            )
+
+        # Get total count
+        count_stmt = select(func.count()).select_from(stmt.subquery())
+        total = db.session.scalar(count_stmt) or 0
+
+        # Apply pagination
+        stmt = stmt.limit(limit + 1).offset((page - 1) * limit)
+        snippets = list(db.session.scalars(stmt).all())
+
+        has_more = len(snippets) > limit
+        if has_more:
+            snippets = snippets[:-1]
+
+        return snippets, total, has_more
+
+    @staticmethod
+    def get_snippet_by_id(
+        *,
+        snippet_id: str,
+        tenant_id: str,
+    ) -> CustomizedSnippet | None:
+        """
+        Get snippet by ID with tenant isolation.
+
+        :param snippet_id: Snippet ID
+        :param tenant_id: Tenant ID
+        :return: CustomizedSnippet or None
+        """
+        return (
+            db.session.query(CustomizedSnippet)
+            .where(
+                CustomizedSnippet.id == snippet_id,
+                CustomizedSnippet.tenant_id == tenant_id,
+            )
+            .first()
+        )
+
+    @staticmethod
+    def create_snippet(
+        *,
+        tenant_id: str,
+        name: str,
+        description: str | None,
+        snippet_type: SnippetType,
+        icon_info: dict | None,
+        graph: dict | None,
+        input_fields: list[dict] | None,
+        account: Account,
+    ) -> CustomizedSnippet:
+        """
+        Create a new snippet.
+
+        :param tenant_id: Tenant ID
+        :param name: Snippet name (must be unique per tenant)
+        :param description: Snippet description
+        :param snippet_type: Type of snippet (node or group)
+        :param icon_info: Icon information
+        :param graph: Workflow graph structure
+        :param input_fields: Input field definitions
+        :param account: Creator account
+        :return: Created CustomizedSnippet
+        :raises ValueError: If name already exists
+        """
+        # Check if name already exists for this tenant
+        existing = (
+            db.session.query(CustomizedSnippet)
+            .where(
+                CustomizedSnippet.tenant_id == tenant_id,
+                CustomizedSnippet.name == name,
+            )
+            .first()
+        )
+        if existing:
+            raise ValueError(f"Snippet with name '{name}' already exists")
+
+        snippet = CustomizedSnippet(
+            tenant_id=tenant_id,
+            name=name,
+            description=description or "",
+            type=snippet_type.value,
+            icon_info=icon_info,
+            graph=json.dumps(graph) if graph else None,
+            input_fields=json.dumps(input_fields) if input_fields else None,
+            created_by=account.id,
+        )
+
+        db.session.add(snippet)
+        db.session.commit()
+
+        return snippet
+
+    @staticmethod
+    def update_snippet(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        account_id: str,
+        data: dict,
+    ) -> CustomizedSnippet:
+        """
+        Update snippet attributes.
+
+        :param session: Database session
+        :param snippet: Snippet to update
+        :param account_id: ID of account making the update
+        :param data: Dictionary of fields to update
+        :return: Updated CustomizedSnippet
+        """
+        if "name" in data:
+            # Check if new name already exists for this tenant
+            existing = (
+                session.query(CustomizedSnippet)
+                .where(
+                    CustomizedSnippet.tenant_id == snippet.tenant_id,
+                    CustomizedSnippet.name == data["name"],
+                    CustomizedSnippet.id != snippet.id,
+                )
+                .first()
+            )
+            if existing:
+                raise ValueError(f"Snippet with name '{data['name']}' already exists")
+            snippet.name = data["name"]
+
+        if "description" in data:
+            snippet.description = data["description"]
+
+        if "icon_info" in data:
+            snippet.icon_info = data["icon_info"]
+
+        snippet.updated_by = account_id
+        snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+
+        session.add(snippet)
+        return snippet
+
+    @staticmethod
+    def delete_snippet(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+    ) -> bool:
+        """
+        Delete a snippet.
+
+        :param session: Database session
+        :param snippet: Snippet to delete
+        :return: True if deleted successfully
+        """
+        session.delete(snippet)
+        return True
+
+    # --- Workflow Operations ---
+
+    def get_draft_workflow(self, snippet: CustomizedSnippet) -> Workflow | None:
+        """
+        Get draft workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :return: Draft Workflow or None
+        """
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.tenant_id == snippet.tenant_id,
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.version == "draft",
+            )
+            .first()
+        )
+        return workflow
+
+    def get_published_workflow(self, snippet: CustomizedSnippet) -> Workflow | None:
+        """
+        Get published workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :return: Published Workflow or None
+        """
+        if not snippet.workflow_id:
+            return None
+
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.tenant_id == snippet.tenant_id,
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.id == snippet.workflow_id,
+            )
+            .first()
+        )
+        return workflow
+
+    def sync_draft_workflow(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        graph: dict,
+        unique_hash: str | None,
+        account: Account,
+        environment_variables: Sequence[VariableBase],
+        conversation_variables: Sequence[VariableBase],
+        input_variables: list[dict] | None = None,
+    ) -> Workflow:
+        """
+        Sync draft workflow for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :param graph: Workflow graph configuration
+        :param unique_hash: Hash for conflict detection
+        :param account: Account making the change
+        :param environment_variables: Environment variables
+        :param conversation_variables: Conversation variables
+        :param input_variables: Input variables for snippet
+        :return: Synced Workflow
+        :raises WorkflowHashNotEqualError: If hash mismatch
+        """
+        workflow = self.get_draft_workflow(snippet=snippet)
+
+        if workflow and workflow.unique_hash != unique_hash:
+            raise WorkflowHashNotEqualError()
+
+        # Create draft workflow if not found
+        if not workflow:
+            workflow = Workflow(
+                tenant_id=snippet.tenant_id,
+                app_id=snippet.id,
+                features="{}",
+                type=WorkflowType.SNIPPET.value,
+                version="draft",
+                graph=json.dumps(graph),
+                created_by=account.id,
+                environment_variables=environment_variables,
+                conversation_variables=conversation_variables,
+            )
+            db.session.add(workflow)
+            db.session.flush()
+        else:
+            # Update existing draft workflow
+            workflow.graph = json.dumps(graph)
+            workflow.updated_by = account.id
+            workflow.updated_at = datetime.now(UTC).replace(tzinfo=None)
+            workflow.environment_variables = environment_variables
+            workflow.conversation_variables = conversation_variables
+
+        # Update snippet's input_fields if provided
+        if input_variables is not None:
+            snippet.input_fields = json.dumps(input_variables)
+            snippet.updated_by = account.id
+            snippet.updated_at = datetime.now(UTC).replace(tzinfo=None)
+
+        db.session.commit()
+        return workflow
+
+    def publish_workflow(
+        self,
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        account: Account,
+    ) -> Workflow:
+        """
+        Publish the draft workflow as a new version.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        :param account: Account making the change
+        :return: Published Workflow
+        :raises ValueError: If no draft workflow exists
+        """
+        draft_workflow_stmt = select(Workflow).where(
+            Workflow.tenant_id == snippet.tenant_id,
+            Workflow.app_id == snippet.id,
+            Workflow.type == WorkflowType.SNIPPET.value,
+            Workflow.version == "draft",
+        )
+        draft_workflow = session.scalar(draft_workflow_stmt)
+        if not draft_workflow:
+            raise ValueError("No valid workflow found.")
+
+        # Create new published workflow
+        workflow = Workflow.new(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            type=draft_workflow.type,
+            version=str(datetime.now(UTC).replace(tzinfo=None)),
+            graph=draft_workflow.graph,
+            features=draft_workflow.features,
+            created_by=account.id,
+            environment_variables=draft_workflow.environment_variables,
+            conversation_variables=draft_workflow.conversation_variables,
+            marked_name="",
+            marked_comment="",
+        )
+        session.add(workflow)
+
+        # Update snippet version
+        snippet.version += 1
+        snippet.is_published = True
+        snippet.workflow_id = workflow.id
+        snippet.updated_by = account.id
+        session.add(snippet)
+
+        return workflow
+
+    def get_all_published_workflows(
+        self,
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+        page: int,
+        limit: int,
+    ) -> tuple[Sequence[Workflow], bool]:
+        """
+        Get all published workflow versions for snippet.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        :param page: Page number
+        :param limit: Items per page
+        :return: Tuple of (workflows list, has_more flag)
+        """
+        if not snippet.workflow_id:
+            return [], False
+
+        stmt = (
+            select(Workflow)
+            .where(
+                Workflow.app_id == snippet.id,
+                Workflow.type == WorkflowType.SNIPPET.value,
+                Workflow.version != "draft",
+            )
+            .order_by(Workflow.version.desc())
+            .limit(limit + 1)
+            .offset((page - 1) * limit)
+        )
+
+        workflows = list(session.scalars(stmt).all())
+        has_more = len(workflows) > limit
+        if has_more:
+            workflows = workflows[:-1]
+
+        return workflows, has_more
+
+    # --- Default Block Configs ---
+
+    def get_default_block_configs(self) -> list[dict]:
+        """
+        Get default block configurations for all node types.
+
+        :return: List of default configurations
+        """
+        default_block_configs: list[dict[str, Any]] = []
+        for node_class_mapping in NODE_TYPE_CLASSES_MAPPING.values():
+            node_class = node_class_mapping[LATEST_VERSION]
+            default_config = node_class.get_default_config()
+            if default_config:
+                default_block_configs.append(dict(default_config))
+
+        return default_block_configs
+
+    def get_default_block_config(self, node_type: str, filters: dict | None = None) -> Mapping[str, object] | None:
+        """
+        Get default config for specific node type.
+
+        :param node_type: Node type string
+        :param filters: Optional filters
+        :return: Default configuration or None
+        """
+        node_type_enum = NodeType(node_type)
+
+        if node_type_enum not in NODE_TYPE_CLASSES_MAPPING:
+            return None
+
+        node_class = NODE_TYPE_CLASSES_MAPPING[node_type_enum][LATEST_VERSION]
+        default_config = node_class.get_default_config(filters=filters)
+        if not default_config:
+            return None
+
+        return default_config
+
+    # --- Workflow Run Operations ---
+
+    def get_snippet_workflow_runs(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        args: dict,
+    ) -> InfiniteScrollPagination:
+        """
+        Get paginated workflow runs for snippet.
+
+        :param snippet: CustomizedSnippet instance
+        :param args: Request arguments (last_id, limit)
+        :return: InfiniteScrollPagination result
+        """
+        limit = int(args.get("limit", 20))
+        last_id = args.get("last_id")
+
+        triggered_from_values = [
+            WorkflowRunTriggeredFrom.DEBUGGING,
+        ]
+
+        return self._workflow_run_repo.get_paginated_workflow_runs(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            triggered_from=triggered_from_values,
+            limit=limit,
+            last_id=last_id,
+        )
+
+    def get_snippet_workflow_run(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        run_id: str,
+    ) -> WorkflowRun | None:
+        """
+        Get workflow run details.
+
+        :param snippet: CustomizedSnippet instance
+        :param run_id: Workflow run ID
+        :return: WorkflowRun or None
+        """
+        return self._workflow_run_repo.get_workflow_run_by_id(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            run_id=run_id,
+        )
+
+    def get_snippet_workflow_run_node_executions(
+        self,
+        *,
+        snippet: CustomizedSnippet,
+        run_id: str,
+    ) -> Sequence[WorkflowNodeExecutionModel]:
+        """
+        Get workflow run node execution list.
+
+        :param snippet: CustomizedSnippet instance
+        :param run_id: Workflow run ID
+        :return: List of WorkflowNodeExecutionModel
+        """
+        workflow_run = self.get_snippet_workflow_run(snippet=snippet, run_id=run_id)
+        if not workflow_run:
+            return []
+
+        node_executions = self._node_execution_service_repo.get_executions_by_workflow_run(
+            tenant_id=snippet.tenant_id,
+            app_id=snippet.id,
+            workflow_run_id=workflow_run.id,
+        )
+
+        return node_executions
+
+    # --- Use Count ---
+
+    @staticmethod
+    def increment_use_count(
+        *,
+        session: Session,
+        snippet: CustomizedSnippet,
+    ) -> None:
+        """
+        Increment the use_count when snippet is used.
+
+        :param session: Database session
+        :param snippet: CustomizedSnippet instance
+        """
+        snippet.use_count += 1
+        session.add(snippet)

api/services/summary_index_service.py

@@ -49,18 +49,11 @@ class SummaryIndexService:
         # Use lazy import to avoid circular import
         from core.rag.index_processor.processor.paragraph_index_processor import ParagraphIndexProcessor
 
-        # Get document language to ensure summary is generated in the correct language
-        # This is especially important for image-only chunks where text is empty or minimal
-        document_language = None
-        if segment.document and segment.document.doc_language:
-            document_language = segment.document.doc_language
-
         summary_content, usage = ParagraphIndexProcessor.generate_summary(
             tenant_id=dataset.tenant_id,
             text=segment.content,
             summary_index_setting=summary_index_setting,
             segment_id=segment.id,
-            document_language=document_language,
         )
 
         if not summary_content:
@@ -565,9 +558,6 @@ class SummaryIndexService:
                     )
                     session.add(summary_record)
 
-            # Commit the batch created records
-            session.commit()
-
     @staticmethod
     def update_summary_record_error(
         segment: DocumentSegment,
@@ -772,6 +762,7 @@ class SummaryIndexService:
                 dataset=dataset,
                 status="not_started",
             )
+            session.commit()  # Commit initial records
 
             summary_records = []
 

web/app/components/workflow/nodes/http/components/curl-panel.tsx

@@ -41,7 +41,7 @@ const parseCurl = (curlCommand: string): { node: HttpNodeType | null, error: str
       case '--request':
         if (i + 1 >= args.length)
           return { node: null, error: 'Missing HTTP method after -X or --request.' }
-        node.method = (args[++i].replace(/^['"]|['"]$/g, '') as Method) || Method.get
+        node.method = (args[++i].replace(/^['"]|['"]$/g, '').toLowerCase() as Method) || Method.get
         hasData = true
         break
       case '-H':