fix(api): serialize pipeline file-upload created_at (#32098 )

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
fix(api): register knowledge pipeline service API routes (#32097 )
2026-02-09 17:50:29 +08:00 · 2026-02-09 17:43:36 +08:00 · 2026-02-09 17:12:16 +08:00 · 2026-02-09 16:42:53 +08:00 · 2026-02-09 15:32:40 +08:00 · 2026-02-09 15:10:29 +08:00
387 changed files with 34621 additions and 8366 deletions
@@ -1 +0,0 @@
-../../.agents/skills/component-refactoring
@@ -1 +0,0 @@
-../../.agents/skills/frontend-code-review
@@ -1 +0,0 @@
-../../.agents/skills/frontend-testing
@@ -1 +0,0 @@
-../../.agents/skills/orpc-contract-first
@@ -9,6 +9,9 @@
 # CODEOWNERS file
 /.github/CODEOWNERS @laipz8200 @crazywoola

+# Agents
+/.agents/skills/ @hyoban
+
 # Docs
 /docs/ @crazywoola

@@ -21,6 +24,10 @@
 /api/services/tools/mcp_tools_manage_service.py @Nov1c444
 /api/controllers/mcp/ @Nov1c444
 /api/controllers/console/app/mcp_server.py @Nov1c444
+
+# Backend - Tests
+/api/tests/ @laipz8200 @QuantumGhost
+
 /api/tests/**/*mcp* @Nov1c444

 # Backend - Workflow - Engine (Core graph execution engine)
@@ -231,6 +238,9 @@
 # Frontend - Base Components
 /web/app/components/base/ @iamjoel @zxhlyh

+# Frontend - Base Components Tests
+/web/app/components/base/**/*.spec.tsx @hyoban @CodingOnStar
+
 # Frontend - Utils and Hooks
 /web/utils/classnames.ts @iamjoel @zxhlyh
 /web/utils/time.ts @iamjoel @zxhlyh
@@ -79,29 +79,6 @@ jobs:
          find . -name "*.py" -type f -exec sed -i.bak -E 's/"([^"]+)" \| None/Optional["\1"]/g; s/'"'"'([^'"'"']+)'"'"' \| None/Optional['"'"'\1'"'"']/g' {} \;
          find . -name "*.py.bak" -type f -delete

-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v6
-        with:
-          node-version: 24
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install web dependencies
-        run: |
-          cd web
-          pnpm install --frozen-lockfile
-
-      - name: ESLint autofix
-        run: |
-          cd web
-          pnpm lint:fix || true
-
      # mdformat breaks YAML front matter in markdown files. Add --exclude for directories containing YAML front matter.
      - name: mdformat
        run: |
@@ -4,8 +4,7 @@ on:
  workflow_run:
    workflows: ["Build and Push API & Web"]
    branches:
-      - "feat/hitl-frontend"
-      - "feat/hitl-backend"
+      - "build/feat/hitl"
    types:
      - completed

@@ -14,10 +13,7 @@ jobs:
    runs-on: ubuntu-latest
    if: |
      github.event.workflow_run.conclusion == 'success' &&
-      (
-        github.event.workflow_run.head_branch == 'feat/hitl-frontend' ||
-        github.event.workflow_run.head_branch == 'feat/hitl-backend'
-      )
+      github.event.workflow_run.head_branch == 'build/feat/hitl'
    steps:
      - name: Deploy to server
        uses: appleboy/ssh-action@v1
@@ -39,7 +39,7 @@ jobs:
        run: pnpm install --frozen-lockfile

      - name: Run tests
-        run: pnpm test:coverage
+        run: pnpm test:ci

      - name: Coverage Summary
        if: always()
@@ -37,7 +37,7 @@
                "-c",
                "1",
                "-Q",
-                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention",
+                "dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention,workflow_based_app_execution",
                "--loglevel",
                "INFO"
            ],
@@ -52,14 +52,12 @@ ignore_imports =
    core.workflow.nodes.agent.agent_node -> extensions.ext_database
    core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
    core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
    core.workflow.nodes.llm.file_saver -> extensions.ext_database
    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
    core.workflow.nodes.llm.node -> extensions.ext_database
    core.workflow.nodes.tool.tool_node -> extensions.ext_database
    core.workflow.graph_engine.command_channels.redis_channel -> extensions.ext_redis
    core.workflow.graph_engine.manager -> extensions.ext_redis
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
    # TODO(QuantumGhost): use DI to avoid depending on global DB.
    core.workflow.nodes.human_input.human_input_node -> extensions.ext_database

@@ -126,11 +124,6 @@ ignore_imports =
    core.workflow.nodes.http_request.node -> core.tools.tool_file_manager
    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.datasource.retrieval_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.dataset_retrieval
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> models.dataset
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> services.feature_service
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_runtime.model_providers.__base.large_language_model
    core.workflow.nodes.llm.llm_utils -> configs
    core.workflow.nodes.llm.llm_utils -> core.app.entities.app_invoke_entities
    core.workflow.nodes.llm.llm_utils -> core.file.models
@@ -140,7 +133,6 @@ ignore_imports =
    core.workflow.nodes.llm.llm_utils -> models.provider
    core.workflow.nodes.llm.llm_utils -> services.credit_pool_service
    core.workflow.nodes.llm.node -> core.tools.signature
-    core.workflow.nodes.template_transform.template_transform_node -> configs
    core.workflow.nodes.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
    core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
    core.workflow.nodes.tool.tool_node -> core.tools.tool_manager
@@ -152,7 +144,6 @@ ignore_imports =
    core.workflow.nodes.human_input.human_input_node -> core.app.entities.app_invoke_entities
    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.app.entities.app_invoke_entities
    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.app_config.entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.app.entities.app_invoke_entities
    core.workflow.nodes.llm.node -> core.app.entities.app_invoke_entities
    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.app.entities.app_invoke_entities
    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.advanced_prompt_transform
@@ -168,9 +159,6 @@ ignore_imports =
    core.workflow.workflow_entry -> core.app.workflow.node_factory
    core.workflow.nodes.datasource.datasource_node -> core.datasource.datasource_manager
    core.workflow.nodes.datasource.datasource_node -> core.datasource.utils.message_transformer
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.agent_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.entities.model_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.model_manager
    core.workflow.nodes.llm.llm_utils -> core.entities.provider_entities
    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_manager
    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_manager
@@ -219,7 +207,6 @@ ignore_imports =
    core.workflow.nodes.llm.node -> core.llm_generator.output_parser.structured_output
    core.workflow.nodes.llm.node -> core.model_manager
    core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.prompt.simple_prompt_transform
    core.workflow.nodes.llm.entities -> core.prompt.entities.advanced_prompt_entities
    core.workflow.nodes.llm.llm_utils -> core.prompt.entities.advanced_prompt_entities
    core.workflow.nodes.llm.node -> core.prompt.entities.advanced_prompt_entities
@@ -235,7 +222,6 @@ ignore_imports =
    core.workflow.nodes.knowledge_index.knowledge_index_node -> services.summary_index_service
    core.workflow.nodes.knowledge_index.knowledge_index_node -> tasks.generate_summary_index_task
    core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.processor.paragraph_index_processor
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.rag.retrieval.retrieval_methods
    core.workflow.nodes.llm.node -> models.dataset
    core.workflow.nodes.agent.agent_node -> core.tools.utils.message_transformer
    core.workflow.nodes.llm.file_saver -> core.tools.signature
@@ -294,8 +280,6 @@ ignore_imports =
    core.workflow.nodes.agent.agent_node -> extensions.ext_database
    core.workflow.nodes.datasource.datasource_node -> extensions.ext_database
    core.workflow.nodes.knowledge_index.knowledge_index_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_database
-    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> extensions.ext_redis
    core.workflow.nodes.llm.file_saver -> extensions.ext_database
    core.workflow.nodes.llm.llm_utils -> extensions.ext_database
    core.workflow.nodes.llm.node -> extensions.ext_database
@@ -53,6 +53,7 @@ select = [
    "S301", # suspicious-pickle-usage, disallow use of `pickle` and its wrappers.
    "S302", # suspicious-marshal-usage, disallow use of `marshal` module
    "S311", # suspicious-non-cryptographic-random-usage,
+    "TID",   # flake8-tidy-imports

 ]

@@ -88,6 +89,7 @@ ignore = [
    "SIM113",  # enumerate-for-loop
    "SIM117",  # multiple-with-statements
    "SIM210",  # if-expr-with-true-false
+    "TID252",  # allow relative imports from parent modules
 ]

 [lint.per-file-ignores]
@@ -109,10 +111,20 @@ ignore = [
    "S110", # allow ignoring exceptions in tests code (currently)

 ]
+"controllers/console/explore/trial.py" = ["TID251"]
+"controllers/console/human_input_form.py" = ["TID251"]
+"controllers/web/human_input_form.py" = ["TID251"]

 [lint.pyflakes]
 allowed-unused-imports = [
-    "_pytest.monkeypatch",
    "tests.integration_tests",
    "tests.unit_tests",
 ]
+
+[lint.flake8-tidy-imports]
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse"]
+msg = "Use Pydantic payload/query models instead of reqparse."
+
+[lint.flake8-tidy-imports.banned-api."flask_restx.reqparse.RequestParser"]
+msg = "Use Pydantic payload/query models instead of reqparse."
@@ -122,7 +122,7 @@ These commands assume you start from the repository root.

   ```bash
   cd api
-   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
+   uv run celery -A app.celery worker -P threads -c 2 --loglevel INFO -Q api_token,dataset,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_executor,retention
   ```

 1. Optional: start Celery Beat (scheduled tasks, in a new terminal).
@@ -739,8 +739,10 @@ def upgrade_db():

            click.echo(click.style("Database migration successful!", fg="green"))

-        except Exception:
+        except Exception as e:
            logger.exception("Failed to execute database migration")
+            click.echo(click.style(f"Database migration failed: {e}", fg="red"))
+            raise SystemExit(1)
        finally:
            lock.release()
    else:
@@ -1450,54 +1452,58 @@ def clear_orphaned_file_records(force: bool):
        all_ids_in_tables = []
        for ids_table in ids_tables:
            query = ""
-            if ids_table["type"] == "uuid":
-                click.echo(
-                    click.style(
-                        f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
+            match ids_table["type"]:
+                case "uuid":
+                    click.echo(
+                        click.style(
+                            f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}",
+                            fg="white",
+                        )
                    )
-                )
-                query = (
-                    f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
-            elif ids_table["type"] == "text":
-                click.echo(
-                    click.style(
-                        f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
-                        fg="white",
+                    c = ids_table["column"]
+                    query = f"SELECT {c} FROM {ids_table['table']} WHERE {c} IS NOT NULL"
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
+                case "text":
+                    t = ids_table["table"]
+                    click.echo(
+                        click.style(
+                            f"- Listing file-id-like strings in column {ids_table['column']} in table {t}",
+                            fg="white",
+                        )
                    )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
-            elif ids_table["type"] == "json":
-                click.echo(
-                    click.style(
-                        (
-                            f"- Listing file-id-like JSON string in column {ids_table['column']} "
-                            f"in table {ids_table['table']}"
-                        ),
-                        fg="white",
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
                    )
-                )
-                query = (
-                    f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
-                    f"FROM {ids_table['table']}"
-                )
-                with db.engine.begin() as conn:
-                    rs = conn.execute(sa.text(query))
-                for i in rs:
-                    for j in i[0]:
-                        all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case "json":
+                    click.echo(
+                        click.style(
+                            (
+                                f"- Listing file-id-like JSON string in column {ids_table['column']} "
+                                f"in table {ids_table['table']}"
+                            ),
+                            fg="white",
+                        )
+                    )
+                    query = (
+                        f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
+                        f"FROM {ids_table['table']}"
+                    )
+                    with db.engine.begin() as conn:
+                        rs = conn.execute(sa.text(query))
+                    for i in rs:
+                        for j in i[0]:
+                            all_ids_in_tables.append({"table": ids_table["table"], "id": j})
+                case _:
+                    pass
        click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))

    except Exception as e:
@@ -1737,59 +1743,18 @@ def file_usage(
                if src_filter != src:
                    continue

-        if ids_table["type"] == "uuid":
-            # Direct UUID match
-            query = (
-                f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    ref_file_id = str(row[1])
-                    if ref_file_id not in file_key_map:
-                        continue
-                    storage_key = file_key_map[ref_file_id]
-
-                    # Apply filters
-                    if file_id and ref_file_id != file_id:
-                        continue
-                    if key and not storage_key.endswith(key):
-                        continue
-
-                    # Only collect items within the requested page range
-                    if offset <= total_count < offset + limit:
-                        paginated_usages.append(
-                            {
-                                "src": f"{ids_table['table']}.{ids_table['column']}",
-                                "record_id": record_id,
-                                "file_id": ref_file_id,
-                                "key": storage_key,
-                            }
-                        )
-                    total_count += 1
-
-        elif ids_table["type"] in ("text", "json"):
-            # Extract UUIDs from text/json content
-            column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
-            query = (
-                f"SELECT {ids_table['pk_column']}, {column_cast} "
-                f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
-            )
-            with db.engine.begin() as conn:
-                rs = conn.execute(sa.text(query))
-                for row in rs:
-                    record_id = str(row[0])
-                    content = str(row[1])
-
-                    # Find all UUIDs in the content
-                    import re
-
-                    uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
-                    matches = uuid_pattern.findall(content)
-
-                    for ref_file_id in matches:
+        match ids_table["type"]:
+            case "uuid":
+                # Direct UUID match
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {ids_table['column']} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        ref_file_id = str(row[1])
                        if ref_file_id not in file_key_map:
                            continue
                        storage_key = file_key_map[ref_file_id]
@@ -1812,6 +1777,50 @@ def file_usage(
                            )
                        total_count += 1

+            case "text" | "json":
+                # Extract UUIDs from text/json content
+                column_cast = f"{ids_table['column']}::text" if ids_table["type"] == "json" else ids_table["column"]
+                query = (
+                    f"SELECT {ids_table['pk_column']}, {column_cast} "
+                    f"FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
+                )
+                with db.engine.begin() as conn:
+                    rs = conn.execute(sa.text(query))
+                    for row in rs:
+                        record_id = str(row[0])
+                        content = str(row[1])
+
+                        # Find all UUIDs in the content
+                        import re
+
+                        uuid_pattern = re.compile(guid_regexp, re.IGNORECASE)
+                        matches = uuid_pattern.findall(content)
+
+                        for ref_file_id in matches:
+                            if ref_file_id not in file_key_map:
+                                continue
+                            storage_key = file_key_map[ref_file_id]
+
+                            # Apply filters
+                            if file_id and ref_file_id != file_id:
+                                continue
+                            if key and not storage_key.endswith(key):
+                                continue
+
+                            # Only collect items within the requested page range
+                            if offset <= total_count < offset + limit:
+                                paginated_usages.append(
+                                    {
+                                        "src": f"{ids_table['table']}.{ids_table['column']}",
+                                        "record_id": record_id,
+                                        "file_id": ref_file_id,
+                                        "key": storage_key,
+                                    }
+                                )
+                            total_count += 1
+            case _:
+                pass
+
    # Output results
    if output_json:
        result = {
@@ -1180,6 +1180,16 @@ class CeleryScheduleTasksConfig(BaseSettings):
        default=0,
    )

+    # API token last_used_at batch update
+    ENABLE_API_TOKEN_LAST_USED_UPDATE_TASK: bool = Field(
+        description="Enable periodic batch update of API token last_used_at timestamps",
+        default=True,
+    )
+    API_TOKEN_LAST_USED_UPDATE_INTERVAL: int = Field(
+        description="Interval in minutes for batch updating API token last_used_at (default 30)",
+        default=30,
+    )
+
    # Trigger provider refresh (simple version)
    ENABLE_TRIGGER_PROVIDER_REFRESH_TASK: bool = Field(
        description="Enable trigger provider refresh poller",
@@ -5,8 +5,6 @@ from enum import StrEnum
 from flask_restx import Namespace
 from pydantic import BaseModel, TypeAdapter

-from controllers.console import console_ns
-
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"


@@ -24,6 +22,9 @@ def register_schema_models(namespace: Namespace, *models: type[BaseModel]) -> No


 def get_or_create_model(model_name: str, field_def):
+    # Import lazily to avoid circular imports between console controllers and schema helpers.
+    from controllers.console import console_ns
+
    existing = console_ns.models.get(model_name)
    if existing is None:
        existing = console_ns.model(model_name, field_def)
@@ -10,6 +10,7 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models.dataset import Dataset
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache

 from . import console_ns
 from .wraps import account_initialization_required, edit_permission_required, setup_required
@@ -131,6 +132,11 @@ class BaseApiKeyResource(Resource):
        if key is None:
            flask_restx.abort(HTTPStatus.NOT_FOUND, message="API key not found")

+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
        db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
        db.session.commit()

@@ -1,10 +1,11 @@
 from typing import Any, Literal

 from flask import abort, make_response, request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter, field_validator

 from controllers.common.errors import NoFileUploadedError, TooManyFilesError
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import (
    account_initialization_required,
@@ -16,9 +17,11 @@ from controllers.console.wraps import (
 )
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
-    annotation_fields,
-    annotation_hit_history_fields,
-    build_annotation_model,
+    Annotation,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+    AnnotationList,
 )
 from libs.helper import uuid_value
 from libs.login import login_required
@@ -89,6 +92,14 @@ reg(CreateAnnotationPayload)
 reg(UpdateAnnotationPayload)
 reg(AnnotationReplyStatusQuery)
 reg(AnnotationFilePayload)
+register_schema_models(
+    console_ns,
+    Annotation,
+    AnnotationList,
+    AnnotationExportList,
+    AnnotationHitHistory,
+    AnnotationHitHistoryList,
+)


@console_ns.route("/apps/<uuid:app_id>/annotation-reply/<string:action>")
@@ -107,10 +118,11 @@ class AnnotationReplyActionApi(Resource):
    def post(self, app_id, action: Literal["enable", "disable"]):
        app_id = str(app_id)
        args = AnnotationReplyPayload.model_validate(console_ns.payload)
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args.model_dump(), app_id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_id)
        return result, 200


@@ -201,33 +213,33 @@ class AnnotationApi(Resource):

        app_id = str(app_id)
        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_id, page, limit, keyword)
-        response = {
-            "data": marshal(annotation_list, annotation_fields),
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response, 200
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response = AnnotationList(
+            data=annotation_models,
+            has_more=len(annotation_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json"), 200

    @console_ns.doc("create_annotation")
    @console_ns.doc(description="Create a new annotation for an app")
    @console_ns.doc(params={"app_id": "Application ID"})
    @console_ns.expect(console_ns.models[CreateAnnotationPayload.__name__])
-    @console_ns.response(201, "Annotation created successfully", build_annotation_model(console_ns))
+    @console_ns.response(201, "Annotation created successfully", console_ns.models[Annotation.__name__])
    @console_ns.response(403, "Insufficient permissions")
    @setup_required
    @login_required
    @account_initialization_required
    @cloud_edition_billing_resource_check("annotation")
-    @marshal_with(annotation_fields)
    @edit_permission_required
    def post(self, app_id):
        app_id = str(app_id)
        args = CreateAnnotationPayload.model_validate(console_ns.payload)
        data = args.model_dump(exclude_none=True)
        annotation = AppAnnotationService.up_insert_app_annotation_from_message(data, app_id)
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")

    @setup_required
    @login_required
@@ -264,7 +276,7 @@ class AnnotationExportApi(Resource):
    @console_ns.response(
        200,
        "Annotations exported successfully",
-        console_ns.model("AnnotationList", {"data": fields.List(fields.Nested(build_annotation_model(console_ns)))}),
+        console_ns.models[AnnotationExportList.__name__],
    )
    @console_ns.response(403, "Insufficient permissions")
    @setup_required
@@ -274,7 +286,8 @@ class AnnotationExportApi(Resource):
    def get(self, app_id):
        app_id = str(app_id)
        annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
-        response_data = {"data": marshal(annotation_list, annotation_fields)}
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response_data = AnnotationExportList(data=annotation_models).model_dump(mode="json")

        # Create response with secure headers for CSV export
        response = make_response(response_data, 200)
@@ -289,7 +302,7 @@ class AnnotationUpdateDeleteApi(Resource):
    @console_ns.doc("update_delete_annotation")
    @console_ns.doc(description="Update or delete an annotation")
    @console_ns.doc(params={"app_id": "Application ID", "annotation_id": "Annotation ID"})
-    @console_ns.response(200, "Annotation updated successfully", build_annotation_model(console_ns))
+    @console_ns.response(200, "Annotation updated successfully", console_ns.models[Annotation.__name__])
    @console_ns.response(204, "Annotation deleted successfully")
    @console_ns.response(403, "Insufficient permissions")
    @console_ns.expect(console_ns.models[UpdateAnnotationPayload.__name__])
@@ -298,7 +311,6 @@ class AnnotationUpdateDeleteApi(Resource):
    @account_initialization_required
    @cloud_edition_billing_resource_check("annotation")
    @edit_permission_required
-    @marshal_with(annotation_fields)
    def post(self, app_id, annotation_id):
        app_id = str(app_id)
        annotation_id = str(annotation_id)
@@ -306,7 +318,7 @@ class AnnotationUpdateDeleteApi(Resource):
        annotation = AppAnnotationService.update_app_annotation_directly(
            args.model_dump(exclude_none=True), app_id, annotation_id
        )
-        return annotation
+        return Annotation.model_validate(annotation, from_attributes=True).model_dump(mode="json")

    @setup_required
    @login_required
@@ -414,14 +426,7 @@ class AnnotationHitHistoryListApi(Resource):
    @console_ns.response(
        200,
        "Hit histories retrieved successfully",
-        console_ns.model(
-            "AnnotationHitHistoryList",
-            {
-                "data": fields.List(
-                    fields.Nested(console_ns.model("AnnotationHitHistoryItem", annotation_hit_history_fields))
-                )
-            },
-        ),
+        console_ns.models[AnnotationHitHistoryList.__name__],
    )
    @console_ns.response(403, "Insufficient permissions")
    @setup_required
@@ -436,11 +441,14 @@ class AnnotationHitHistoryListApi(Resource):
        annotation_hit_history_list, total = AppAnnotationService.get_annotation_hit_histories(
            app_id, annotation_id, page, limit
        )
-        response = {
-            "data": marshal(annotation_hit_history_list, annotation_hit_history_fields),
-            "has_more": len(annotation_hit_history_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
-        return response
+        history_models = TypeAdapter(list[AnnotationHitHistory]).validate_python(
+            annotation_hit_history_list, from_attributes=True
+        )
+        response = AnnotationHitHistoryList(
+            data=history_models,
+            has_more=len(annotation_hit_history_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json")
@@ -1,3 +1,4 @@
+import logging
 import uuid
 from datetime import datetime
 from typing import Any, Literal, TypeAlias
@@ -54,6 +55,8 @@ ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "co

 register_enum_models(console_ns, IconType)

+_logger = logging.getLogger(__name__)
+

 class AppListQuery(BaseModel):
    page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
@@ -499,6 +502,7 @@ class AppListApi(Resource):
                    select(Workflow).where(
                        Workflow.version == Workflow.VERSION_DRAFT,
                        Workflow.app_id.in_(workflow_capable_app_ids),
+                        Workflow.tenant_id == current_tenant_id,
                    )
                )
                .scalars()
@@ -510,12 +514,14 @@ class AppListApi(Resource):
                NodeType.TRIGGER_PLUGIN,
            }
            for workflow in draft_workflows:
+                node_id = None
                try:
-                    for _, node_data in workflow.walk_nodes():
+                    for node_id, node_data in workflow.walk_nodes():
                        if node_data.get("type") in trigger_node_types:
                            draft_trigger_app_ids.add(str(workflow.app_id))
                            break
                except Exception:
+                    _logger.exception("error while walking nodes, workflow_id=%s, node_id=%s", workflow.id, node_id)
                    continue

        for app in app_pagination.items:
@@ -6,6 +6,7 @@ from pydantic import BaseModel, Field
 from werkzeug.exceptions import InternalServerError

 import services
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
    AppUnavailableError,
@@ -33,7 +34,6 @@ from services.errors.audio import (
 )

 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"


 class TextToSpeechPayload(BaseModel):
@@ -47,13 +47,11 @@ class TextToSpeechVoiceQuery(BaseModel):
    language: str = Field(..., description="Language code")


-console_ns.schema_model(
-    TextToSpeechPayload.__name__, TextToSpeechPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    TextToSpeechVoiceQuery.__name__,
-    TextToSpeechVoiceQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+class AudioTranscriptResponse(BaseModel):
+    text: str = Field(description="Transcribed text from audio")
+
+
+register_schema_models(console_ns, AudioTranscriptResponse, TextToSpeechPayload, TextToSpeechVoiceQuery)


@console_ns.route("/apps/<uuid:app_id>/audio-to-text")
@@ -64,7 +62,7 @@ class ChatMessageAudioApi(Resource):
    @console_ns.response(
        200,
        "Audio transcription successful",
-        console_ns.model("AudioTranscriptResponse", {"text": fields.String(description="Transcribed text from audio")}),
+        console_ns.models[AudioTranscriptResponse.__name__],
    )
    @console_ns.response(400, "Bad request - No audio uploaded or unsupported type")
    @console_ns.response(413, "Audio file too large")
@@ -509,16 +509,19 @@ class ChatConversationApi(Resource):
                case "created_at" | "-created_at" | _:
                    query = query.where(Conversation.created_at <= end_datetime_utc)

-        if args.annotation_status == "annotated":
-            query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
-                MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
-            )
-        elif args.annotation_status == "not_annotated":
-            query = (
-                query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
-                .group_by(Conversation.id)
-                .having(func.count(MessageAnnotation.id) == 0)
-            )
+        match args.annotation_status:
+            case "annotated":
+                query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
+                    MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
+                )
+            case "not_annotated":
+                query = (
+                    query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
+                    .group_by(Conversation.id)
+                    .having(func.count(MessageAnnotation.id) == 0)
+                )
+            case "all":
+                pass

        if app_model.mode == AppMode.ADVANCED_CHAT:
            query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound

+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.app.error import (
    CompletionRequestError,
@@ -35,7 +36,6 @@ from services.errors.message import MessageNotExistsError, SuggestedQuestionsAft
 from services.message_service import MessageService, attach_message_extra_contents

 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"


 class ChatMessagesQuery(BaseModel):
@@ -90,13 +90,22 @@ class FeedbackExportQuery(BaseModel):
        raise ValueError("has_comment must be a boolean value")


-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class AnnotationCountResponse(BaseModel):
+    count: int = Field(description="Number of annotations")


-reg(ChatMessagesQuery)
-reg(MessageFeedbackPayload)
-reg(FeedbackExportQuery)
+class SuggestedQuestionsResponse(BaseModel):
+    data: list[str] = Field(description="Suggested question")
+
+
+register_schema_models(
+    console_ns,
+    ChatMessagesQuery,
+    MessageFeedbackPayload,
+    FeedbackExportQuery,
+    AnnotationCountResponse,
+    SuggestedQuestionsResponse,
+)

 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -232,7 +241,7 @@ class ChatMessageListApi(Resource):
    @marshal_with(message_infinite_scroll_pagination_model)
    @edit_permission_required
    def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = ChatMessagesQuery.model_validate(request.args.to_dict())

        conversation = (
            db.session.query(Conversation)
@@ -358,7 +367,7 @@ class MessageAnnotationCountApi(Resource):
    @console_ns.response(
        200,
        "Annotation count retrieved successfully",
-        console_ns.model("AnnotationCountResponse", {"count": fields.Integer(description="Number of annotations")}),
+        console_ns.models[AnnotationCountResponse.__name__],
    )
    @get_app_model
    @setup_required
@@ -378,9 +387,7 @@ class MessageSuggestedQuestionApi(Resource):
    @console_ns.response(
        200,
        "Suggested questions retrieved successfully",
-        console_ns.model(
-            "SuggestedQuestionsResponse", {"data": fields.List(fields.String(description="Suggested question"))}
-        ),
+        console_ns.models[SuggestedQuestionsResponse.__name__],
    )
    @console_ns.response(404, "Message or conversation not found")
    @setup_required
@@ -430,7 +437,7 @@ class MessageFeedbackExportApi(Resource):
    @login_required
    @account_initialization_required
    def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = FeedbackExportQuery.model_validate(request.args.to_dict())

        # Import the service function
        from services.feedback_service import FeedbackService
@@ -463,8 +463,9 @@ class WorkflowRunNodeExecutionListApi(Resource):
 class ConsoleWorkflowPauseDetailsApi(Resource):
    """Console API for getting workflow pause details."""

-    @account_initialization_required
+    @setup_required
    @login_required
+    @account_initialization_required
    def get(self, workflow_run_id: str):
        """
        Get workflow pause details.
@@ -477,10 +478,14 @@ class ConsoleWorkflowPauseDetailsApi(Resource):
        # Query WorkflowRun to determine if workflow is suspended
        session_maker = sessionmaker(bind=db.engine)
        workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=session_maker)
+
        workflow_run = db.session.get(WorkflowRun, workflow_run_id)
        if not workflow_run:
            raise NotFoundError("Workflow run not found")

+        if workflow_run.tenant_id != current_user.current_tenant_id:
+            raise NotFoundError("Workflow run not found")
+
        # Check if workflow is suspended
        is_paused = workflow_run.status == WorkflowExecutionStatus.PAUSED
        if not is_paused:
@@ -2,9 +2,11 @@ import logging

 import httpx
 from flask import current_app, redirect, request
-from flask_restx import Resource, fields
+from flask_restx import Resource
+from pydantic import BaseModel, Field

 from configs import dify_config
+from controllers.common.schema import register_schema_models
 from libs.login import login_required
 from libs.oauth_data_source import NotionOAuth

@@ -14,6 +16,26 @@ from ..wraps import account_initialization_required, is_admin_or_owner_required,
 logger = logging.getLogger(__name__)


+class OAuthDataSourceResponse(BaseModel):
+    data: str = Field(description="Authorization URL or 'internal' for internal setup")
+
+
+class OAuthDataSourceBindingResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+class OAuthDataSourceSyncResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    OAuthDataSourceResponse,
+    OAuthDataSourceBindingResponse,
+    OAuthDataSourceSyncResponse,
+)
+
+
 def get_oauth_providers():
    with current_app.app_context():
        notion_oauth = NotionOAuth(
@@ -34,10 +56,7 @@ class OAuthDataSource(Resource):
    @console_ns.response(
        200,
        "Authorization URL or internal setup success",
-        console_ns.model(
-            "OAuthDataSourceResponse",
-            {"data": fields.Raw(description="Authorization URL or 'internal' for internal setup")},
-        ),
+        console_ns.models[OAuthDataSourceResponse.__name__],
    )
    @console_ns.response(400, "Invalid provider")
    @console_ns.response(403, "Admin privileges required")
@@ -101,7 +120,7 @@ class OAuthDataSourceBinding(Resource):
    @console_ns.response(
        200,
        "Data source binding success",
-        console_ns.model("OAuthDataSourceBindingResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceBindingResponse.__name__],
    )
    @console_ns.response(400, "Invalid provider or code")
    def get(self, provider: str):
@@ -133,7 +152,7 @@ class OAuthDataSourceSync(Resource):
    @console_ns.response(
        200,
        "Data source sync success",
-        console_ns.model("OAuthDataSourceSyncResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[OAuthDataSourceSyncResponse.__name__],
    )
    @console_ns.response(400, "Invalid provider or sync failed")
    @setup_required
@@ -2,10 +2,11 @@ import base64
 import secrets

 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field, field_validator
 from sqlalchemy.orm import Session

+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
    EmailCodeError,
@@ -48,8 +49,31 @@ class ForgotPasswordResetPayload(BaseModel):
        return valid_password(value)


-for model in (ForgotPasswordSendPayload, ForgotPasswordCheckPayload, ForgotPasswordResetPayload):
-    console_ns.schema_model(model.__name__, model.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+class ForgotPasswordEmailResponse(BaseModel):
+    result: str = Field(description="Operation result")
+    data: str | None = Field(default=None, description="Reset token")
+    code: str | None = Field(default=None, description="Error code if account not found")
+
+
+class ForgotPasswordCheckResponse(BaseModel):
+    is_valid: bool = Field(description="Whether code is valid")
+    email: EmailStr = Field(description="Email address")
+    token: str = Field(description="New reset token")
+
+
+class ForgotPasswordResetResponse(BaseModel):
+    result: str = Field(description="Operation result")
+
+
+register_schema_models(
+    console_ns,
+    ForgotPasswordSendPayload,
+    ForgotPasswordCheckPayload,
+    ForgotPasswordResetPayload,
+    ForgotPasswordEmailResponse,
+    ForgotPasswordCheckResponse,
+    ForgotPasswordResetResponse,
+)


@console_ns.route("/forgot-password")
@@ -60,14 +84,7 @@ class ForgotPasswordSendEmailApi(Resource):
    @console_ns.response(
        200,
        "Email sent successfully",
-        console_ns.model(
-            "ForgotPasswordEmailResponse",
-            {
-                "result": fields.String(description="Operation result"),
-                "data": fields.String(description="Reset token"),
-                "code": fields.String(description="Error code if account not found"),
-            },
-        ),
+        console_ns.models[ForgotPasswordEmailResponse.__name__],
    )
    @console_ns.response(400, "Invalid email or rate limit exceeded")
    @setup_required
@@ -106,14 +123,7 @@ class ForgotPasswordCheckApi(Resource):
    @console_ns.response(
        200,
        "Code verified successfully",
-        console_ns.model(
-            "ForgotPasswordCheckResponse",
-            {
-                "is_valid": fields.Boolean(description="Whether code is valid"),
-                "email": fields.String(description="Email address"),
-                "token": fields.String(description="New reset token"),
-            },
-        ),
+        console_ns.models[ForgotPasswordCheckResponse.__name__],
    )
    @console_ns.response(400, "Invalid code or token")
    @setup_required
@@ -163,7 +173,7 @@ class ForgotPasswordResetApi(Resource):
    @console_ns.response(
        200,
        "Password reset successfully",
-        console_ns.model("ForgotPasswordResetResponse", {"result": fields.String(description="Operation result")}),
+        console_ns.models[ForgotPasswordResetResponse.__name__],
    )
    @console_ns.response(400, "Invalid token or password mismatch")
    @setup_required
@@ -155,43 +155,43 @@ class OAuthServerUserTokenApi(Resource):
            grant_type = OAuthGrantType(payload.grant_type)
        except ValueError:
            raise BadRequest("invalid grant_type")
+        match grant_type:
+            case OAuthGrantType.AUTHORIZATION_CODE:
+                if not payload.code:
+                    raise BadRequest("code is required")

-        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
-            if not payload.code:
-                raise BadRequest("code is required")
+                if payload.client_secret != oauth_provider_app.client_secret:
+                    raise BadRequest("client_secret is invalid")

-            if payload.client_secret != oauth_provider_app.client_secret:
-                raise BadRequest("client_secret is invalid")
+                if payload.redirect_uri not in oauth_provider_app.redirect_uris:
+                    raise BadRequest("redirect_uri is invalid")

-            if payload.redirect_uri not in oauth_provider_app.redirect_uris:
-                raise BadRequest("redirect_uri is invalid")
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, code=payload.code, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )
+            case OAuthGrantType.REFRESH_TOKEN:
+                if not payload.refresh_token:
+                    raise BadRequest("refresh_token is required")

-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, code=payload.code, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
-        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
-            if not payload.refresh_token:
-                raise BadRequest("refresh_token is required")
-
-            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
-                grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
-            )
-            return jsonable_encoder(
-                {
-                    "access_token": access_token,
-                    "token_type": "Bearer",
-                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
-                    "refresh_token": refresh_token,
-                }
-            )
+                access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                    grant_type, refresh_token=payload.refresh_token, client_id=oauth_provider_app.client_id
+                )
+                return jsonable_encoder(
+                    {
+                        "access_token": access_token,
+                        "token_type": "Bearer",
+                        "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                        "refresh_token": refresh_token,
+                    }
+                )


@console_ns.route("/oauth/provider/account")
@@ -1,6 +1,6 @@
 import json
 from collections.abc import Generator
-from typing import Any, cast
+from typing import Any, Literal, cast

 from flask import request
 from flask_restx import Resource, fields, marshal_with
@@ -157,9 +157,8 @@ class DataSourceApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    def patch(self, binding_id, action):
+    def patch(self, binding_id, action: Literal["enable", "disable"]):
        binding_id = str(binding_id)
-        action = str(action)
        with Session(db.engine) as session:
            data_source_binding = session.execute(
                select(DataSourceOauthBinding).filter_by(id=binding_id)
@@ -167,23 +166,24 @@ class DataSourceApi(Resource):
        if data_source_binding is None:
            raise NotFound("Data source binding not found.")
        # enable binding
-        if action == "enable":
-            if data_source_binding.disabled:
-                data_source_binding.disabled = False
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is not disabled.")
-        # disable binding
-        if action == "disable":
-            if not data_source_binding.disabled:
-                data_source_binding.disabled = True
-                data_source_binding.updated_at = naive_utc_now()
-                db.session.add(data_source_binding)
-                db.session.commit()
-            else:
-                raise ValueError("Data source is disabled.")
+        match action:
+            case "enable":
+                if data_source_binding.disabled:
+                    data_source_binding.disabled = False
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is not disabled.")
+            # disable binding
+            case "disable":
+                if not data_source_binding.disabled:
+                    data_source_binding.disabled = True
+                    data_source_binding.updated_at = naive_utc_now()
+                    db.session.add(data_source_binding)
+                    db.session.commit()
+                else:
+                    raise ValueError("Data source is disabled.")
        return {"result": "success"}, 200


@@ -55,6 +55,7 @@ from libs.login import current_account_with_tenant, login_required
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermissionEnum
 from models.provider_ids import ModelProviderID
+from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService

 # Register models for flask_restx to avoid dict type issues in Swagger
@@ -820,6 +821,11 @@ class DatasetApiDeleteApi(Resource):
        if key is None:
            console_ns.abort(404, message="API key not found")

+        # Invalidate cache before deleting from database
+        # Type assertion: key is guaranteed to be non-None here because abort() raises
+        assert key is not None  # nosec - for type checker only
+        ApiTokenCache.delete(key.token, key.type)
+
        db.session.query(ApiToken).where(ApiToken.id == api_key_id).delete()
        db.session.commit()

@@ -576,63 +576,62 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
            if document.indexing_status in {"completed", "error"}:
                raise DocumentAlreadyFinishedError()
            data_source_info = document.data_source_info_dict
+            match document.data_source_type:
+                case "upload_file":
+                    if not data_source_info:
+                        continue
+                    file_id = data_source_info["upload_file_id"]
+                    file_detail = (
+                        db.session.query(UploadFile)
+                        .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
+                        .first()
+                    )

-            if document.data_source_type == "upload_file":
-                if not data_source_info:
-                    continue
-                file_id = data_source_info["upload_file_id"]
-                file_detail = (
-                    db.session.query(UploadFile)
-                    .where(UploadFile.tenant_id == current_tenant_id, UploadFile.id == file_id)
-                    .first()
-                )
+                    if file_detail is None:
+                        raise NotFound("File not found.")

-                if file_detail is None:
-                    raise NotFound("File not found.")
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
+                    )
+                    extract_settings.append(extract_setting)
+                case "notion_import":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.NOTION,
+                        notion_info=NotionInfo.model_validate(
+                            {
+                                "credential_id": data_source_info.get("credential_id"),
+                                "notion_workspace_id": data_source_info["notion_workspace_id"],
+                                "notion_obj_id": data_source_info["notion_page_id"],
+                                "notion_page_type": data_source_info["type"],
+                                "tenant_id": current_tenant_id,
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)
+                case "website_crawl":
+                    if not data_source_info:
+                        continue
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.WEBSITE,
+                        website_info=WebsiteInfo.model_validate(
+                            {
+                                "provider": data_source_info["provider"],
+                                "job_id": data_source_info["job_id"],
+                                "url": data_source_info["url"],
+                                "tenant_id": current_tenant_id,
+                                "mode": data_source_info["mode"],
+                                "only_main_content": data_source_info["only_main_content"],
+                            }
+                        ),
+                        document_model=document.doc_form,
+                    )
+                    extract_settings.append(extract_setting)

-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
-                )
-                extract_settings.append(extract_setting)
-
-            elif document.data_source_type == "notion_import":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION,
-                    notion_info=NotionInfo.model_validate(
-                        {
-                            "credential_id": data_source_info.get("credential_id"),
-                            "notion_workspace_id": data_source_info["notion_workspace_id"],
-                            "notion_obj_id": data_source_info["notion_page_id"],
-                            "notion_page_type": data_source_info["type"],
-                            "tenant_id": current_tenant_id,
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-            elif document.data_source_type == "website_crawl":
-                if not data_source_info:
-                    continue
-                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE,
-                    website_info=WebsiteInfo.model_validate(
-                        {
-                            "provider": data_source_info["provider"],
-                            "job_id": data_source_info["job_id"],
-                            "url": data_source_info["url"],
-                            "tenant_id": current_tenant_id,
-                            "mode": data_source_info["mode"],
-                            "only_main_content": data_source_info["only_main_content"],
-                        }
-                    ),
-                    document_model=document.doc_form,
-                )
-                extract_settings.append(extract_setting)
-
-            else:
-                raise ValueError("Data source type not support")
+                case _:
+                    raise ValueError("Data source type not support")
            indexing_runner = IndexingRunner()
            try:
                response = indexing_runner.indexing_estimate(
@@ -954,23 +953,24 @@ class DocumentProcessingApi(DocumentResource):
        if not current_user.is_dataset_editor:
            raise Forbidden()

-        if action == "pause":
-            if document.indexing_status != "indexing":
-                raise InvalidActionError("Document not in indexing state.")
+        match action:
+            case "pause":
+                if document.indexing_status != "indexing":
+                    raise InvalidActionError("Document not in indexing state.")

-            document.paused_by = current_user.id
-            document.paused_at = naive_utc_now()
-            document.is_paused = True
-            db.session.commit()
+                document.paused_by = current_user.id
+                document.paused_at = naive_utc_now()
+                document.is_paused = True
+                db.session.commit()

-        elif action == "resume":
-            if document.indexing_status not in {"paused", "error"}:
-                raise InvalidActionError("Document not in paused or error state.")
+            case "resume":
+                if document.indexing_status not in {"paused", "error"}:
+                    raise InvalidActionError("Document not in paused or error state.")

-            document.paused_by = None
-            document.paused_at = None
-            document.is_paused = False
-            db.session.commit()
+                document.paused_by = None
+                document.paused_at = None
+                document.is_paused = False
+                db.session.commit()

        return {"result": "success"}, 200

@@ -1339,6 +1339,18 @@ 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
@@ -126,10 +126,11 @@ class DatasetMetadataBuiltInFieldActionApi(Resource):
            raise NotFound("Dataset not found.")
        DatasetService.check_dataset_permission(dataset, current_user)

-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
        return {"result": "success"}, 200


@@ -1,10 +1,9 @@
 import json
 import logging
 from typing import Any, Literal, cast
-from uuid import UUID

 from flask import abort, request
-from flask_restx import Resource, marshal_with, reqparse  # type: ignore
+from flask_restx import Resource, marshal_with  # type: ignore
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -38,7 +37,7 @@ from core.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from factories import variable_factory
 from libs import helper
-from libs.helper import TimestampField
+from libs.helper import TimestampField, UUIDStrOrEmpty
 from libs.login import current_account_with_tenant, current_user, login_required
 from models import Account
 from models.dataset import Pipeline
@@ -110,7 +109,7 @@ class NodeIdQuery(BaseModel):


 class WorkflowRunQuery(BaseModel):
-    last_id: UUID | None = None
+    last_id: UUIDStrOrEmpty | None = None
    limit: int = Field(default=20, ge=1, le=100)


@@ -121,6 +120,10 @@ class DatasourceVariablesPayload(BaseModel):
    start_node_title: str


+class RagPipelineRecommendedPluginQuery(BaseModel):
+    type: str = "all"
+
+
 register_schema_models(
    console_ns,
    DraftWorkflowSyncPayload,
@@ -135,6 +138,7 @@ register_schema_models(
    NodeIdQuery,
    WorkflowRunQuery,
    DatasourceVariablesPayload,
+    RagPipelineRecommendedPluginQuery,
 )


@@ -975,11 +979,8 @@ class RagPipelineRecommendedPluginApi(Resource):
    @login_required
    @account_initialization_required
    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("type", type=str, location="args", required=False, default="all")
-        args = parser.parse_args()
-        type = args["type"]
+        query = RagPipelineRecommendedPluginQuery.model_validate(request.args.to_dict())

        rag_pipeline_service = RagPipelineService()
-        recommended_plugins = rag_pipeline_service.get_recommended_plugins(type)
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins(query.type)
        return recommended_plugins
@@ -1,8 +1,9 @@
 import logging
-from typing import Any, cast
+from typing import Any, Literal, cast

 from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with, reqparse
+from flask_restx import Resource, fields, marshal, marshal_with
+from pydantic import BaseModel
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound

 import services
@@ -51,7 +52,7 @@ from fields.app_fields import (
    tag_fields,
 )
 from fields.dataset_fields import dataset_fields
-from fields.member_fields import build_simple_account_model
+from fields.member_fields import simple_account_fields
 from fields.workflow_fields import (
    conversation_variable_fields,
    pipeline_variable_fields,
@@ -103,7 +104,7 @@ app_detail_fields_with_site_copy["tags"] = fields.List(fields.Nested(tag_model))
 app_detail_fields_with_site_copy["site"] = fields.Nested(site_model)
 app_detail_with_site_model = get_or_create_model("TrialAppDetailWithSite", app_detail_fields_with_site_copy)

-simple_account_model = build_simple_account_model(console_ns)
+simple_account_model = get_or_create_model("SimpleAccount", simple_account_fields)
 conversation_variable_model = get_or_create_model("TrialConversationVariable", conversation_variable_fields)
 pipeline_variable_model = get_or_create_model("TrialPipelineVariable", pipeline_variable_fields)

@@ -117,7 +118,56 @@ workflow_fields_copy["rag_pipeline_variables"] = fields.List(fields.Nested(pipel
 workflow_model = get_or_create_model("TrialWorkflow", workflow_fields_copy)


+# Pydantic models for request validation
+DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
+
+
+class WorkflowRunRequest(BaseModel):
+    inputs: dict
+    files: list | None = None
+
+
+class ChatRequest(BaseModel):
+    inputs: dict
+    query: str
+    files: list | None = None
+    conversation_id: str | None = None
+    parent_message_id: str | None = None
+    retriever_from: str = "explore_app"
+
+
+class TextToSpeechRequest(BaseModel):
+    message_id: str | None = None
+    voice: str | None = None
+    text: str | None = None
+    streaming: bool | None = None
+
+
+class CompletionRequest(BaseModel):
+    inputs: dict
+    query: str = ""
+    files: list | None = None
+    response_mode: Literal["blocking", "streaming"] | None = None
+    retriever_from: str = "explore_app"
+
+
+# Register schemas for Swagger documentation
+console_ns.schema_model(
+    WorkflowRunRequest.__name__, WorkflowRunRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    ChatRequest.__name__, ChatRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    TextToSpeechRequest.__name__, TextToSpeechRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    CompletionRequest.__name__, CompletionRequest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class TrialAppWorkflowRunApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[WorkflowRunRequest.__name__])
    def post(self, trial_app):
        """
        Run workflow
@@ -129,10 +179,8 @@ class TrialAppWorkflowRunApi(TrialAppResource):
        if app_mode != AppMode.WORKFLOW:
            raise NotWorkflowAppError()

-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
+        request_data = WorkflowRunRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
        assert current_user is not None
        try:
            app_id = app_model.id
@@ -183,6 +231,7 @@ class TrialAppWorkflowTaskStopApi(TrialAppResource):


 class TrialChatApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[ChatRequest.__name__])
    @trial_feature_enable
    def post(self, trial_app):
        app_model = trial_app
@@ -190,14 +239,14 @@ class TrialChatApi(TrialAppResource):
        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
            raise NotChatAppError()

-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = ChatRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()
+
+        # Validate UUID values if provided
+        if args.get("conversation_id"):
+            args["conversation_id"] = uuid_value(args["conversation_id"])
+        if args.get("parent_message_id"):
+            args["parent_message_id"] = uuid_value(args["parent_message_id"])

        args["auto_generate_name"] = False

@@ -320,20 +369,16 @@ class TrialChatAudioApi(TrialAppResource):


 class TrialChatTextApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[TextToSpeechRequest.__name__])
    @trial_feature_enable
    def post(self, trial_app):
        app_model = trial_app
        try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
+            request_data = TextToSpeechRequest.model_validate(console_ns.payload)

-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
+            message_id = request_data.message_id
+            text = request_data.text
+            voice = request_data.voice
            if not isinstance(current_user, Account):
                raise ValueError("current_user must be an Account instance")

@@ -371,19 +416,15 @@ class TrialChatTextApi(TrialAppResource):


 class TrialCompletionApi(TrialAppResource):
+    @console_ns.expect(console_ns.models[CompletionRequest.__name__])
    @trial_feature_enable
    def post(self, trial_app):
        app_model = trial_app
        if app_model.mode != "completion":
            raise NotCompletionAppError()

-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
+        request_data = CompletionRequest.model_validate(console_ns.payload)
+        args = request_data.model_dump()

        streaming = args["response_mode"] == "streaming"
        args["auto_generate_name"] = False
@@ -1,87 +1,74 @@
 import os
+from typing import Literal

 from flask import session
-from flask_restx import Resource, fields
 from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.orm import Session

 from configs import dify_config
+from controllers.fastopenapi import console_router
 from extensions.ext_database import db
 from models.model import DifySetup
 from services.account_service import TenantService

-from . import console_ns
 from .error import AlreadySetupError, InitValidateFailedError
 from .wraps import only_edition_self_hosted

-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-

 class InitValidatePayload(BaseModel):
-    password: str = Field(..., max_length=30)
+    password: str = Field(..., max_length=30, description="Initialization password")


-console_ns.schema_model(
-    InitValidatePayload.__name__,
-    InitValidatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+class InitStatusResponse(BaseModel):
+    status: Literal["finished", "not_started"] = Field(..., description="Initialization status")
+
+
+class InitValidateResponse(BaseModel):
+    result: str = Field(description="Operation result", examples=["success"])
+
+
+@console_router.get(
+    "/init",
+    response_model=InitStatusResponse,
+    tags=["console"],
 )
+def get_init_status() -> InitStatusResponse:
+    """Get initialization validation status."""
+    init_status = get_init_validate_status()
+    if init_status:
+        return InitStatusResponse(status="finished")
+    return InitStatusResponse(status="not_started")


-@console_ns.route("/init")
-class InitValidateAPI(Resource):
-    @console_ns.doc("get_init_status")
-    @console_ns.doc(description="Get initialization validation status")
-    @console_ns.response(
-        200,
-        "Success",
-        model=console_ns.model(
-            "InitStatusResponse",
-            {"status": fields.String(description="Initialization status", enum=["finished", "not_started"])},
-        ),
-    )
-    def get(self):
-        """Get initialization validation status"""
-        init_status = get_init_validate_status()
-        if init_status:
-            return {"status": "finished"}
-        return {"status": "not_started"}
+@console_router.post(
+    "/init",
+    response_model=InitValidateResponse,
+    tags=["console"],
+    status_code=201,
+)
+@only_edition_self_hosted
+def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse:
+    """Validate initialization password."""
+    tenant_count = TenantService.get_tenant_count()
+    if tenant_count > 0:
+        raise AlreadySetupError()

-    @console_ns.doc("validate_init_password")
-    @console_ns.doc(description="Validate initialization password for self-hosted edition")
-    @console_ns.expect(console_ns.models[InitValidatePayload.__name__])
-    @console_ns.response(
-        201,
-        "Success",
-        model=console_ns.model("InitValidateResponse", {"result": fields.String(description="Operation result")}),
-    )
-    @console_ns.response(400, "Already setup or validation failed")
-    @only_edition_self_hosted
-    def post(self):
-        """Validate initialization password"""
-        # is tenant created
-        tenant_count = TenantService.get_tenant_count()
-        if tenant_count > 0:
-            raise AlreadySetupError()
+    if payload.password != os.environ.get("INIT_PASSWORD"):
+        session["is_init_validated"] = False
+        raise InitValidateFailedError()

-        payload = InitValidatePayload.model_validate(console_ns.payload)
-        input_password = payload.password
-
-        if input_password != os.environ.get("INIT_PASSWORD"):
-            session["is_init_validated"] = False
-            raise InitValidateFailedError()
-
-        session["is_init_validated"] = True
-        return {"result": "success"}, 201
+    session["is_init_validated"] = True
+    return InitValidateResponse(result="success")


-def get_init_validate_status():
+def get_init_validate_status() -> bool:
    if dify_config.EDITION == "SELF_HOSTED":
        if os.environ.get("INIT_PASSWORD"):
            if session.get("is_init_validated"):
                return True

            with Session(db.engine) as db_session:
-                return db_session.execute(select(DifySetup)).scalar_one_or_none()
+                return db_session.execute(select(DifySetup)).scalar_one_or_none() is not None

    return True
@@ -1,7 +1,6 @@
 import urllib.parse

 import httpx
-from flask_restx import Resource
 from pydantic import BaseModel, Field

 import services
@@ -11,7 +10,7 @@ from controllers.common.errors import (
    RemoteFileUploadError,
    UnsupportedFileTypeError,
 )
-from controllers.common.schema import register_schema_models
+from controllers.fastopenapi import console_router
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
@@ -19,84 +18,74 @@ from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
 from libs.login import current_account_with_tenant
 from services.file_service import FileService

-from . import console_ns
-
-register_schema_models(console_ns, RemoteFileInfo, FileWithSignedUrl)
-
-
-@console_ns.route("/remote-files/<path:url>")
-class RemoteFileInfoApi(Resource):
-    @console_ns.response(200, "Remote file info", console_ns.models[RemoteFileInfo.__name__])
-    def get(self, url):
-        decoded_url = urllib.parse.unquote(url)
-        resp = ssrf_proxy.head(decoded_url)
-        if resp.status_code != httpx.codes.OK:
-            # failed back to get method
-            resp = ssrf_proxy.get(decoded_url, timeout=3)
-        resp.raise_for_status()
-        info = RemoteFileInfo(
-            file_type=resp.headers.get("Content-Type", "application/octet-stream"),
-            file_length=int(resp.headers.get("Content-Length", 0)),
-        )
-        return info.model_dump(mode="json")
-

 class RemoteFileUploadPayload(BaseModel):
    url: str = Field(..., description="URL to fetch")


-console_ns.schema_model(
-    RemoteFileUploadPayload.__name__,
-    RemoteFileUploadPayload.model_json_schema(ref_template="#/definitions/{model}"),
+@console_router.get(
+    "/remote-files/<path:url>",
+    response_model=RemoteFileInfo,
+    tags=["console"],
 )
+def get_remote_file_info(url: str) -> RemoteFileInfo:
+    decoded_url = urllib.parse.unquote(url)
+    resp = ssrf_proxy.head(decoded_url)
+    if resp.status_code != httpx.codes.OK:
+        resp = ssrf_proxy.get(decoded_url, timeout=3)
+    resp.raise_for_status()
+    return RemoteFileInfo(
+        file_type=resp.headers.get("Content-Type", "application/octet-stream"),
+        file_length=int(resp.headers.get("Content-Length", 0)),
+    )


-@console_ns.route("/remote-files/upload")
-class RemoteFileUploadApi(Resource):
-    @console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
-    @console_ns.response(201, "Remote file uploaded", console_ns.models[FileWithSignedUrl.__name__])
-    def post(self):
-        args = RemoteFileUploadPayload.model_validate(console_ns.payload)
-        url = args.url
+@console_router.post(
+    "/remote-files/upload",
+    response_model=FileWithSignedUrl,
+    tags=["console"],
+    status_code=201,
+)
+def upload_remote_file(payload: RemoteFileUploadPayload) -> FileWithSignedUrl:
+    url = payload.url

-        try:
-            resp = ssrf_proxy.head(url=url)
-            if resp.status_code != httpx.codes.OK:
-                resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
-            if resp.status_code != httpx.codes.OK:
-                raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
-        except httpx.RequestError as e:
-            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")
+    try:
+        resp = ssrf_proxy.head(url=url)
+        if resp.status_code != httpx.codes.OK:
+            resp = ssrf_proxy.get(url=url, timeout=3, follow_redirects=True)
+        if resp.status_code != httpx.codes.OK:
+            raise RemoteFileUploadError(f"Failed to fetch file from {url}: {resp.text}")
+    except httpx.RequestError as e:
+        raise RemoteFileUploadError(f"Failed to fetch file from {url}: {str(e)}")

-        file_info = helpers.guess_file_info_from_response(resp)
+    file_info = helpers.guess_file_info_from_response(resp)

-        if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
-            raise FileTooLargeError
+    if not FileService.is_file_size_within_limit(extension=file_info.extension, file_size=file_info.size):
+        raise FileTooLargeError

-        content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
+    content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content

-        try:
-            user, _ = current_account_with_tenant()
-            upload_file = FileService(db.engine).upload_file(
-                filename=file_info.filename,
-                content=content,
-                mimetype=file_info.mimetype,
-                user=user,
-                source_url=url,
-            )
-        except services.errors.file.FileTooLargeError as file_too_large_error:
-            raise FileTooLargeError(file_too_large_error.description)
-        except services.errors.file.UnsupportedFileTypeError:
-            raise UnsupportedFileTypeError()
-
-        payload = FileWithSignedUrl(
-            id=upload_file.id,
-            name=upload_file.name,
-            size=upload_file.size,
-            extension=upload_file.extension,
-            url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
-            mime_type=upload_file.mime_type,
-            created_by=upload_file.created_by,
-            created_at=int(upload_file.created_at.timestamp()),
+    try:
+        user, _ = current_account_with_tenant()
+        upload_file = FileService(db.engine).upload_file(
+            filename=file_info.filename,
+            content=content,
+            mimetype=file_info.mimetype,
+            user=user,
+            source_url=url,
        )
-        return payload.model_dump(mode="json"), 201
+    except services.errors.file.FileTooLargeError as file_too_large_error:
+        raise FileTooLargeError(file_too_large_error.description)
+    except services.errors.file.UnsupportedFileTypeError:
+        raise UnsupportedFileTypeError()
+
+    return FileWithSignedUrl(
+        id=upload_file.id,
+        name=upload_file.name,
+        size=upload_file.size,
+        extension=upload_file.extension,
+        url=file_helpers.get_signed_file_url(upload_file_id=upload_file.id),
+        mime_type=upload_file.mime_type,
+        created_by=upload_file.created_by,
+        created_at=int(upload_file.created_at.timestamp()),
+    )
@@ -1,17 +1,27 @@
 from typing import Literal

 from flask import request
-from flask_restx import Resource, marshal_with
+from flask_restx import Namespace, Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden

 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 fields.tag_fields import dataset_tag_fields
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService

+dataset_tag_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "type": fields.String,
+    "binding_count": fields.String,
+}
+
+
+def build_dataset_tag_fields(api_or_ns: Namespace):
+    return api_or_ns.model("DataSetTag", dataset_tag_fields)
+

 class TagBasePayload(BaseModel):
    name: str = Field(description="Tag name", min_length=1, max_length=50)
@@ -110,7 +120,7 @@ class TagUpdateDeleteApi(Resource):

        TagService.delete_tag(tag_id)

-        return 204
+        return "", 204


@console_ns.route("/tag-bindings/create")
@@ -12,6 +12,7 @@ from sqlalchemy.orm import Session

 from configs import dify_config
 from constants.languages import supported_language
+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
    EmailAlreadyInUseError,
@@ -37,7 +38,7 @@ from controllers.console.wraps import (
    setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_fields
+from fields.member_fields import Account as AccountResponse
 from libs.datetime_utils import naive_utc_now
 from libs.helper import EmailStr, TimestampField, extract_remote_ip, timezone
 from libs.login import current_account_with_tenant, login_required
@@ -170,6 +171,12 @@ reg(ChangeEmailSendPayload)
 reg(ChangeEmailValidityPayload)
 reg(ChangeEmailResetPayload)
 reg(CheckEmailUniquePayload)
+register_schema_models(console_ns, AccountResponse)
+
+
+def _serialize_account(account) -> dict:
+    return AccountResponse.model_validate(account, from_attributes=True).model_dump(mode="json")
+

 integrate_fields = {
    "provider": fields.String,
@@ -236,11 +243,11 @@ class AccountProfileApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    @enterprise_license_required
    def get(self):
        current_user, _ = current_account_with_tenant()
-        return current_user
+        return _serialize_account(current_user)


@console_ns.route("/account/name")
@@ -249,14 +256,14 @@ class AccountNameApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
        args = AccountNamePayload.model_validate(payload)
        updated_account = AccountService.update_account(current_user, name=args.name)

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/avatar")
@@ -265,7 +272,7 @@ class AccountAvatarApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
@@ -273,7 +280,7 @@ class AccountAvatarApi(Resource):

        updated_account = AccountService.update_account(current_user, avatar=args.avatar)

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/interface-language")
@@ -282,7 +289,7 @@ class AccountInterfaceLanguageApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
@@ -290,7 +297,7 @@ class AccountInterfaceLanguageApi(Resource):

        updated_account = AccountService.update_account(current_user, interface_language=args.interface_language)

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/interface-theme")
@@ -299,7 +306,7 @@ class AccountInterfaceThemeApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
@@ -307,7 +314,7 @@ class AccountInterfaceThemeApi(Resource):

        updated_account = AccountService.update_account(current_user, interface_theme=args.interface_theme)

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/timezone")
@@ -316,7 +323,7 @@ class AccountTimezoneApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
@@ -324,7 +331,7 @@ class AccountTimezoneApi(Resource):

        updated_account = AccountService.update_account(current_user, timezone=args.timezone)

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/password")
@@ -333,7 +340,7 @@ class AccountPasswordApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        current_user, _ = current_account_with_tenant()
        payload = console_ns.payload or {}
@@ -344,7 +351,7 @@ class AccountPasswordApi(Resource):
        except ServiceCurrentPasswordIncorrectError:
            raise CurrentPasswordIncorrectError()

-        return {"result": "success"}
+        return _serialize_account(current_user)


@console_ns.route("/account/integrates")
@@ -620,7 +627,7 @@ class ChangeEmailResetApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_fields)
+    @console_ns.response(200, "Success", console_ns.models[AccountResponse.__name__])
    def post(self):
        payload = console_ns.payload or {}
        args = ChangeEmailResetPayload.model_validate(payload)
@@ -649,7 +656,7 @@ class ChangeEmailResetApi(Resource):
            email=normalized_new_email,
        )

-        return updated_account
+        return _serialize_account(updated_account)


@console_ns.route("/account/change-email/check-email-unique")
@@ -1,9 +1,10 @@
 from typing import Any

 from flask import request
-from flask_restx import Resource, fields
+from flask_restx import Resource
 from pydantic import BaseModel, Field

+from controllers.common.schema import register_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
@@ -38,15 +39,53 @@ class EndpointListForPluginQuery(EndpointListQuery):
    plugin_id: str


+class EndpointCreateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class PluginEndpointListResponse(BaseModel):
+    endpoints: list[dict[str, Any]] = Field(description="Endpoint information")
+
+
+class EndpointDeleteResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointUpdateResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointEnableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
+class EndpointDisableResponse(BaseModel):
+    success: bool = Field(description="Operation success")
+
+
 def reg(cls: type[BaseModel]):
    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))


-reg(EndpointCreatePayload)
-reg(EndpointIdPayload)
-reg(EndpointUpdatePayload)
-reg(EndpointListQuery)
-reg(EndpointListForPluginQuery)
+register_schema_models(
+    console_ns,
+    EndpointCreatePayload,
+    EndpointIdPayload,
+    EndpointUpdatePayload,
+    EndpointListQuery,
+    EndpointListForPluginQuery,
+    EndpointCreateResponse,
+    EndpointListResponse,
+    PluginEndpointListResponse,
+    EndpointDeleteResponse,
+    EndpointUpdateResponse,
+    EndpointEnableResponse,
+    EndpointDisableResponse,
+)


@console_ns.route("/workspaces/current/endpoints/create")
@@ -57,7 +96,7 @@ class EndpointCreateApi(Resource):
    @console_ns.response(
        200,
        "Endpoint created successfully",
-        console_ns.model("EndpointCreateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointCreateResponse.__name__],
    )
    @console_ns.response(403, "Admin privileges required")
    @setup_required
@@ -91,9 +130,7 @@ class EndpointListApi(Resource):
    @console_ns.response(
        200,
        "Success",
-        console_ns.model(
-            "EndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[EndpointListResponse.__name__],
    )
    @setup_required
    @login_required
@@ -126,9 +163,7 @@ class EndpointListForSinglePluginApi(Resource):
    @console_ns.response(
        200,
        "Success",
-        console_ns.model(
-            "PluginEndpointListResponse", {"endpoints": fields.List(fields.Raw(description="Endpoint information"))}
-        ),
+        console_ns.models[PluginEndpointListResponse.__name__],
    )
    @setup_required
    @login_required
@@ -163,7 +198,7 @@ class EndpointDeleteApi(Resource):
    @console_ns.response(
        200,
        "Endpoint deleted successfully",
-        console_ns.model("EndpointDeleteResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDeleteResponse.__name__],
    )
    @console_ns.response(403, "Admin privileges required")
    @setup_required
@@ -190,7 +225,7 @@ class EndpointUpdateApi(Resource):
    @console_ns.response(
        200,
        "Endpoint updated successfully",
-        console_ns.model("EndpointUpdateResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointUpdateResponse.__name__],
    )
    @console_ns.response(403, "Admin privileges required")
    @setup_required
@@ -221,7 +256,7 @@ class EndpointEnableApi(Resource):
    @console_ns.response(
        200,
        "Endpoint enabled successfully",
-        console_ns.model("EndpointEnableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointEnableResponse.__name__],
    )
    @console_ns.response(403, "Admin privileges required")
    @setup_required
@@ -248,7 +283,7 @@ class EndpointDisableApi(Resource):
    @console_ns.response(
        200,
        "Endpoint disabled successfully",
-        console_ns.model("EndpointDisableResponse", {"success": fields.Boolean(description="Operation success")}),
+        console_ns.models[EndpointDisableResponse.__name__],
    )
    @console_ns.response(403, "Admin privileges required")
    @setup_required
@@ -1,12 +1,12 @@
 from urllib import parse

 from flask import abort, request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource
+from pydantic import BaseModel, Field, TypeAdapter

 import services
 from configs import dify_config
-from controllers.common.schema import get_or_create_model, register_enum_models
+from controllers.common.schema import register_enum_models, register_schema_models
 from controllers.console import console_ns
 from controllers.console.auth.error import (
    CannotTransferOwnerToSelfError,
@@ -25,7 +25,7 @@ from controllers.console.wraps import (
    setup_required,
 )
 from extensions.ext_database import db
-from fields.member_fields import account_with_role_fields, account_with_role_list_fields
+from fields.member_fields import AccountWithRole, AccountWithRoleList
 from libs.helper import extract_remote_ip
 from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountRole
@@ -69,12 +69,7 @@ reg(OwnerTransferEmailPayload)
 reg(OwnerTransferCheckPayload)
 reg(OwnerTransferPayload)
 register_enum_models(console_ns, TenantAccountRole)
-
-account_with_role_model = get_or_create_model("AccountWithRole", account_with_role_fields)
-
-account_with_role_list_fields_copy = account_with_role_list_fields.copy()
-account_with_role_list_fields_copy["accounts"] = fields.List(fields.Nested(account_with_role_model))
-account_with_role_list_model = get_or_create_model("AccountWithRoleList", account_with_role_list_fields_copy)
+register_schema_models(console_ns, AccountWithRole, AccountWithRoleList)


@console_ns.route("/workspaces/current/members")
@@ -84,13 +79,15 @@ class MemberListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
    def get(self):
        current_user, _ = current_account_with_tenant()
        if not current_user.current_tenant:
            raise ValueError("No current tenant")
        members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200


@console_ns.route("/workspaces/current/members/invite-email")
@@ -235,13 +232,15 @@ class DatasetOperatorMemberListApi(Resource):
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(account_with_role_list_model)
+    @console_ns.response(200, "Success", console_ns.models[AccountWithRoleList.__name__])
    def get(self):
        current_user, _ = current_account_with_tenant()
        if not current_user.current_tenant:
            raise ValueError("No current tenant")
        members = TenantService.get_dataset_operator_members(current_user.current_tenant)
-        return {"result": "success", "accounts": members}, 200
+        member_models = TypeAdapter(list[AccountWithRole]).validate_python(members, from_attributes=True)
+        response = AccountWithRoleList(accounts=member_models)
+        return response.model_dump(mode="json"), 200


@console_ns.route("/workspaces/current/members/send-owner-transfer-confirm-email")
@@ -34,6 +34,8 @@ from .dataset import (
    metadata,
    segment,
 )
+from .dataset.rag_pipeline import rag_pipeline_workflow
+from .end_user import end_user
 from .workspace import models

 __all__ = [
@@ -44,6 +46,7 @@ __all__ = [
    "conversation",
    "dataset",
    "document",
+    "end_user",
    "file",
    "file_preview",
    "hit_testing",
@@ -51,6 +54,7 @@ __all__ = [
    "message",
    "metadata",
    "models",
+    "rag_pipeline_workflow",
    "segment",
    "site",
    "workflow",
@@ -1,16 +1,16 @@
 from typing import Literal

 from flask import request
-from flask_restx import Namespace, Resource, fields
+from flask_restx import Resource
 from flask_restx.api import HTTPStatus
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, TypeAdapter

 from controllers.common.schema import register_schema_models
 from controllers.console.wraps import edit_permission_required
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import validate_app_token
 from extensions.ext_redis import redis_client
-from fields.annotation_fields import annotation_fields, build_annotation_model
+from fields.annotation_fields import Annotation, AnnotationList
 from models.model import App
 from services.annotation_service import AppAnnotationService

@@ -26,7 +26,9 @@ class AnnotationReplyActionPayload(BaseModel):
    embedding_model_name: str = Field(description="Embedding model name")


-register_schema_models(service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload)
+register_schema_models(
+    service_api_ns, AnnotationCreatePayload, AnnotationReplyActionPayload, Annotation, AnnotationList
+)


@service_api_ns.route("/apps/annotation-reply/<string:action>")
@@ -45,10 +47,11 @@ class AnnotationReplyActionApi(Resource):
    def post(self, app_model: App, action: Literal["enable", "disable"]):
        """Enable or disable annotation reply feature."""
        args = AnnotationReplyActionPayload.model_validate(service_api_ns.payload or {}).model_dump()
-        if action == "enable":
-            result = AppAnnotationService.enable_app_annotation(args, app_model.id)
-        elif action == "disable":
-            result = AppAnnotationService.disable_app_annotation(app_model.id)
+        match action:
+            case "enable":
+                result = AppAnnotationService.enable_app_annotation(args, app_model.id)
+            case "disable":
+                result = AppAnnotationService.disable_app_annotation(app_model.id)
        return result, 200


@@ -82,23 +85,6 @@ class AnnotationReplyActionStatusApi(Resource):
        return {"job_id": job_id, "job_status": job_status, "error_msg": error_msg}, 200


-# Define annotation list response model
-annotation_list_fields = {
-    "data": fields.List(fields.Nested(annotation_fields)),
-    "has_more": fields.Boolean,
-    "limit": fields.Integer,
-    "total": fields.Integer,
-    "page": fields.Integer,
-}
-
-
-def build_annotation_list_model(api_or_ns: Namespace):
-    """Build the annotation list model for the API or Namespace."""
-    copied_annotation_list_fields = annotation_list_fields.copy()
-    copied_annotation_list_fields["data"] = fields.List(fields.Nested(build_annotation_model(api_or_ns)))
-    return api_or_ns.model("AnnotationList", copied_annotation_list_fields)
-
-
@service_api_ns.route("/apps/annotations")
 class AnnotationListApi(Resource):
    @service_api_ns.doc("list_annotations")
@@ -109,8 +95,12 @@ class AnnotationListApi(Resource):
            401: "Unauthorized - invalid API token",
        }
    )
+    @service_api_ns.response(
+        200,
+        "Annotations retrieved successfully",
+        service_api_ns.models[AnnotationList.__name__],
+    )
    @validate_app_token
-    @service_api_ns.marshal_with(build_annotation_list_model(service_api_ns))
    def get(self, app_model: App):
        """List annotations for the application."""
        page = request.args.get("page", default=1, type=int)
@@ -118,13 +108,15 @@ class AnnotationListApi(Resource):
        keyword = request.args.get("keyword", default="", type=str)

        annotation_list, total = AppAnnotationService.get_annotation_list_by_app_id(app_model.id, page, limit, keyword)
-        return {
-            "data": annotation_list,
-            "has_more": len(annotation_list) == limit,
-            "limit": limit,
-            "total": total,
-            "page": page,
-        }
+        annotation_models = TypeAdapter(list[Annotation]).validate_python(annotation_list, from_attributes=True)
+        response = AnnotationList(
+            data=annotation_models,
+            has_more=len(annotation_list) == limit,
+            limit=limit,
+            total=total,
+            page=page,
+        )
+        return response.model_dump(mode="json")

    @service_api_ns.expect(service_api_ns.models[AnnotationCreatePayload.__name__])
    @service_api_ns.doc("create_annotation")
@@ -135,13 +127,18 @@ class AnnotationListApi(Resource):
            401: "Unauthorized - invalid API token",
        }
    )
+    @service_api_ns.response(
+        HTTPStatus.CREATED,
+        "Annotation created successfully",
+        service_api_ns.models[Annotation.__name__],
+    )
    @validate_app_token
-    @service_api_ns.marshal_with(build_annotation_model(service_api_ns), code=HTTPStatus.CREATED)
    def post(self, app_model: App):
        """Create a new annotation."""
        args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
        annotation = AppAnnotationService.insert_app_annotation_directly(args, app_model.id)
-        return annotation, 201
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json"), HTTPStatus.CREATED


@service_api_ns.route("/apps/annotations/<uuid:annotation_id>")
@@ -158,14 +155,19 @@ class AnnotationUpdateDeleteApi(Resource):
            404: "Annotation not found",
        }
    )
+    @service_api_ns.response(
+        200,
+        "Annotation updated successfully",
+        service_api_ns.models[Annotation.__name__],
+    )
    @validate_app_token
    @edit_permission_required
-    @service_api_ns.marshal_with(build_annotation_model(service_api_ns))
    def put(self, app_model: App, annotation_id: str):
        """Update an existing annotation."""
        args = AnnotationCreatePayload.model_validate(service_api_ns.payload or {}).model_dump()
        annotation = AppAnnotationService.update_app_annotation_directly(args, app_model.id, annotation_id)
-        return annotation
+        response = Annotation.model_validate(annotation, from_attributes=True)
+        return response.model_dump(mode="json")

    @service_api_ns.doc("delete_annotation")
    @service_api_ns.doc(description="Delete an annotation")
@@ -30,6 +30,7 @@ from core.errors.error import (
 from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 from services.app_task_service import AppTaskService
@@ -52,7 +53,7 @@ class ChatRequestPayload(BaseModel):
    query: str
    files: list[dict[str, Any]] | None = None
    response_mode: Literal["blocking", "streaming"] | None = None
-    conversation_id: str | None = Field(default=None, description="Conversation UUID")
+    conversation_id: UUIDStrOrEmpty | None = Field(default=None, description="Conversation UUID")
    retriever_from: str = Field(default="dev")
    auto_generate_name: bool = Field(default=True, description="Auto generate conversation name")
    workflow_id: str | None = Field(default=None, description="Workflow ID for advanced chat")
@@ -1,5 +1,4 @@
 from typing import Any, Literal
-from uuid import UUID

 from flask import request
 from flask_restx import Resource
@@ -23,12 +22,13 @@ from fields.conversation_variable_fields import (
    build_conversation_variable_infinite_scroll_pagination_model,
    build_conversation_variable_model,
 )
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.conversation_service import ConversationService


 class ConversationListQuery(BaseModel):
-    last_id: UUID | None = Field(default=None, description="Last conversation ID for pagination")
+    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last conversation ID for pagination")
    limit: int = Field(default=20, ge=1, le=100, description="Number of conversations to return")
    sort_by: Literal["created_at", "-created_at", "updated_at", "-updated_at"] = Field(
        default="-updated_at", description="Sort order for conversations"
@@ -48,7 +48,7 @@ class ConversationRenamePayload(BaseModel):


 class ConversationVariablesQuery(BaseModel):
-    last_id: UUID | None = Field(default=None, description="Last variable ID for pagination")
+    last_id: UUIDStrOrEmpty | None = Field(default=None, description="Last variable ID for pagination")
    limit: int = Field(default=20, ge=1, le=100, description="Number of variables to return")
    variable_name: str | None = Field(
        default=None, description="Filter variables by name", min_length=1, max_length=255
@@ -1,6 +1,5 @@
 import logging
 from typing import Literal
-from uuid import UUID

 from flask import request
 from flask_restx import Resource
@@ -15,6 +14,7 @@ from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate
 from core.app.entities.app_invoke_entities import InvokeFrom
 from fields.conversation_fields import ResultResponse
 from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem
+from libs.helper import UUIDStrOrEmpty
 from models.model import App, AppMode, EndUser
 from services.errors.message import (
    FirstMessageNotExistsError,
@@ -27,8 +27,8 @@ logger = logging.getLogger(__name__)


 class MessageListQuery(BaseModel):
-    conversation_id: UUID
-    first_id: UUID | None = None
+    conversation_id: UUIDStrOrEmpty
+    first_id: UUIDStrOrEmpty | None = None
    limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return")


@@ -17,7 +17,7 @@ from controllers.service_api.wraps import (
 from core.model_runtime.entities.model_entities import ModelType
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
-from fields.tag_fields import build_dataset_tag_fields
+from fields.tag_fields import DataSetTag
 from libs.login import current_user
 from models.account import Account
 from models.dataset import DatasetPermissionEnum
@@ -114,6 +114,7 @@ register_schema_models(
    TagBindingPayload,
    TagUnbindingPayload,
    DatasetListQuery,
+    DataSetTag,
 )


@@ -395,7 +396,7 @@ class DatasetApi(DatasetApiResource):
        try:
            if DatasetService.delete_dataset(dataset_id_str, current_user):
                DatasetPermissionService.clear_partial_member_list(dataset_id_str)
-                return 204
+                return "", 204
            else:
                raise NotFound("Dataset not found.")
        except services.errors.dataset.DatasetInUseError:
@@ -480,15 +481,14 @@ class DatasetTagsApi(DatasetApiResource):
            401: "Unauthorized - invalid API token",
        }
    )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
    def get(self, _):
        """Get all knowledge type tags."""
        assert isinstance(current_user, Account)
        cid = current_user.current_tenant_id
        assert cid is not None
        tags = TagService.get_tags("knowledge", cid)
-
-        return tags, 200
+        tag_models = TypeAdapter(list[DataSetTag]).validate_python(tags, from_attributes=True)
+        return [tag.model_dump(mode="json") for tag in tag_models], 200

    @service_api_ns.expect(service_api_ns.models[TagCreatePayload.__name__])
    @service_api_ns.doc("create_dataset_tag")
@@ -500,7 +500,6 @@ class DatasetTagsApi(DatasetApiResource):
            403: "Forbidden - insufficient permissions",
        }
    )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
    def post(self, _):
        """Add a knowledge type tag."""
        assert isinstance(current_user, Account)
@@ -510,7 +509,9 @@ class DatasetTagsApi(DatasetApiResource):
        payload = TagCreatePayload.model_validate(service_api_ns.payload or {})
        tag = TagService.save_tags({"name": payload.name, "type": "knowledge"})

-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+        ).model_dump(mode="json")
        return response, 200

    @service_api_ns.expect(service_api_ns.models[TagUpdatePayload.__name__])
@@ -523,7 +524,6 @@ class DatasetTagsApi(DatasetApiResource):
            403: "Forbidden - insufficient permissions",
        }
    )
-    @service_api_ns.marshal_with(build_dataset_tag_fields(service_api_ns))
    def patch(self, _):
        assert isinstance(current_user, Account)
        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -536,8 +536,9 @@ class DatasetTagsApi(DatasetApiResource):

        binding_count = TagService.get_tag_binding_count(tag_id)

-        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
-
+        response = DataSetTag.model_validate(
+            {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+        ).model_dump(mode="json")
        return response, 200

    @service_api_ns.expect(service_api_ns.models[TagDeletePayload.__name__])
@@ -556,7 +557,7 @@ class DatasetTagsApi(DatasetApiResource):
        payload = TagDeletePayload.model_validate(service_api_ns.payload or {})
        TagService.delete_tag(payload.tag_id)

-        return 204
+        return "", 204


@service_api_ns.route("/datasets/tags/binding")
@@ -580,7 +581,7 @@ class DatasetTagBindingApi(DatasetApiResource):
        payload = TagBindingPayload.model_validate(service_api_ns.payload or {})
        TagService.save_tag_binding({"tag_ids": payload.tag_ids, "target_id": payload.target_id, "type": "knowledge"})

-        return 204
+        return "", 204


@service_api_ns.route("/datasets/tags/unbinding")
@@ -604,7 +605,7 @@ class DatasetTagUnbindingApi(DatasetApiResource):
        payload = TagUnbindingPayload.model_validate(service_api_ns.payload or {})
        TagService.delete_tag_binding({"tag_id": payload.tag_id, "target_id": payload.target_id, "type": "knowledge"})

-        return 204
+        return "", 204


@service_api_ns.route("/datasets/<uuid:dataset_id>/tags")
@@ -746,4 +746,4 @@ class DocumentApi(DatasetApiResource):
        except services.errors.document.DocumentIndexingError:
            raise DocumentIndexingError("Cannot delete document during indexing.")

-        return 204
+        return "", 204
@@ -1,7 +1,10 @@
-from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase
+from controllers.common.schema import register_schema_model
+from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase, HitTestingPayload
 from controllers.service_api import service_api_ns
 from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check

+register_schema_model(service_api_ns, HitTestingPayload)
+

@service_api_ns.route("/datasets/<uuid:dataset_id>/hit-testing", "/datasets/<uuid:dataset_id>/retrieve")
 class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
@@ -15,6 +18,7 @@ class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
            404: "Dataset not found",
        }
    )
+    @service_api_ns.expect(service_api_ns.models[HitTestingPayload.__name__])
    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
    def post(self, tenant_id, dataset_id):
        """Perform hit testing on a dataset.
@@ -128,7 +128,7 @@ class DatasetMetadataServiceApi(DatasetApiResource):
        DatasetService.check_dataset_permission(dataset, current_user)

        MetadataService.delete_metadata(dataset_id_str, metadata_id_str)
-        return 204
+        return "", 204


@service_api_ns.route("/datasets/<uuid:dataset_id>/metadata/built-in")
@@ -168,10 +168,11 @@ class DatasetMetadataBuiltInFieldActionServiceApi(DatasetApiResource):
            raise NotFound("Dataset not found.")
        DatasetService.check_dataset_permission(dataset, current_user)

-        if action == "enable":
-            MetadataService.enable_built_in_field(dataset)
-        elif action == "disable":
-            MetadataService.disable_built_in_field(dataset)
+        match action:
+            case "enable":
+                MetadataService.enable_built_in_field(dataset)
+            case "disable":
+                MetadataService.disable_built_in_field(dataset)
        return {"result": "success"}, 200


@@ -1,5 +1,3 @@
-import string
-import uuid
 from collections.abc import Generator
 from typing import Any

@@ -12,6 +10,7 @@ from controllers.common.errors import FilenameNotExistsError, NoFileUploadedErro
 from controllers.common.schema import register_schema_model
 from controllers.service_api import service_api_ns
 from controllers.service_api.dataset.error import PipelineRunError
+from controllers.service_api.dataset.rag_pipeline.serializers import serialize_upload_file
 from controllers.service_api.wraps import DatasetApiResource
 from core.app.apps.pipeline.pipeline_generator import PipelineGenerator
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -41,7 +40,7 @@ register_schema_model(service_api_ns, DatasourceNodeRunPayload)
 register_schema_model(service_api_ns, PipelineRunApiEntity)


-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource-plugins")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource-plugins")
 class DatasourcePluginsApi(DatasetApiResource):
    """Resource for datasource plugins."""

@@ -76,7 +75,7 @@ class DatasourcePluginsApi(DatasetApiResource):
        return datasource_plugins, 200


-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/datasource/nodes/{string:node_id}/run")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/datasource/nodes/<string:node_id>/run")
 class DatasourceNodeRunApi(DatasetApiResource):
    """Resource for datasource node run."""

@@ -131,7 +130,7 @@ class DatasourceNodeRunApi(DatasetApiResource):
        )


-@service_api_ns.route(f"/datasets/{uuid:dataset_id}/pipeline/run")
+@service_api_ns.route("/datasets/<uuid:dataset_id>/pipeline/run")
 class PipelineRunApi(DatasetApiResource):
    """Resource for datasource node run."""

@@ -232,12 +231,4 @@ class KnowledgebasePipelineFileUploadApi(DatasetApiResource):
        except services.errors.file.UnsupportedFileTypeError:
            raise UnsupportedFileTypeError()

-        return {
-            "id": upload_file.id,
-            "name": upload_file.name,
-            "size": upload_file.size,
-            "extension": upload_file.extension,
-            "mime_type": upload_file.mime_type,
-            "created_by": upload_file.created_by,
-            "created_at": upload_file.created_at,
-        }, 201
+        return serialize_upload_file(upload_file), 201
@@ -0,0 +1,22 @@
+"""
+Serialization helpers for Service API knowledge pipeline endpoints.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from models.model import UploadFile
+
+
+def serialize_upload_file(upload_file: UploadFile) -> dict[str, Any]:
+    return {
+        "id": upload_file.id,
+        "name": upload_file.name,
+        "size": upload_file.size,
+        "extension": upload_file.extension,
+        "mime_type": upload_file.mime_type,
+        "created_by": upload_file.created_by,
+        "created_at": upload_file.created_at.isoformat() if upload_file.created_at else None,
+    }
@@ -233,7 +233,7 @@ class DatasetSegmentApi(DatasetApiResource):
        if not segment:
            raise NotFound("Segment not found.")
        SegmentService.delete_segment(segment, document, dataset)
-        return 204
+        return "", 204

    @service_api_ns.expect(service_api_ns.models[SegmentUpdatePayload.__name__])
    @service_api_ns.doc("update_segment")
@@ -499,7 +499,7 @@ class DatasetChildChunkApi(DatasetApiResource):
        except ChildChunkDeleteIndexServiceError as e:
            raise ChildChunkDeleteIndexError(str(e))

-        return 204
+        return "", 204

    @service_api_ns.expect(service_api_ns.models[ChildChunkUpdatePayload.__name__])
    @service_api_ns.doc("update_child_chunk")
@@ -0,0 +1,3 @@
+from . import end_user
+
+__all__ = ["end_user"]
@@ -0,0 +1,41 @@
+from uuid import UUID
+
+from flask_restx import Resource
+
+from controllers.service_api import service_api_ns
+from controllers.service_api.end_user.error import EndUserNotFoundError
+from controllers.service_api.wraps import validate_app_token
+from fields.end_user_fields import EndUserDetail
+from models.model import App
+from services.end_user_service import EndUserService
+
+
+@service_api_ns.route("/end-users/<uuid:end_user_id>")
+class EndUserApi(Resource):
+    """Resource for retrieving end user details by ID."""
+
+    @service_api_ns.doc("get_end_user")
+    @service_api_ns.doc(description="Get an end user by ID")
+    @service_api_ns.doc(
+        params={"end_user_id": "End user ID"},
+        responses={
+            200: "End user retrieved successfully",
+            401: "Unauthorized - invalid API token",
+            404: "End user not found",
+        },
+    )
+    @validate_app_token
+    def get(self, app_model: App, end_user_id: UUID):
+        """Get end user detail.
+
+        This endpoint is scoped to the current app token's tenant/app to prevent
+        cross-tenant/app access when an end-user ID is known.
+        """
+
+        end_user = EndUserService.get_end_user_by_id(
+            tenant_id=app_model.tenant_id, app_id=app_model.id, end_user_id=str(end_user_id)
+        )
+        if end_user is None:
+            raise EndUserNotFoundError()
+
+        return EndUserDetail.model_validate(end_user).model_dump(mode="json")
@@ -0,0 +1,7 @@
+from libs.exception import BaseHTTPException
+
+
+class EndUserNotFoundError(BaseHTTPException):
+    error_code = "end_user_not_found"
+    description = "End user not found."
+    code = 404
@@ -1,27 +1,24 @@
 import logging
 import time
 from collections.abc import Callable
-from datetime import timedelta
 from enum import StrEnum, auto
 from functools import wraps
-from typing import Concatenate, ParamSpec, TypeVar
+from typing import Concatenate, ParamSpec, TypeVar, cast

 from flask import current_app, request
 from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
-from sqlalchemy import select, update
-from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized

 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from libs.datetime_utils import naive_utc_now
 from libs.login import current_user
 from models import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App
+from services.api_token_service import ApiTokenCache, fetch_token_with_single_flight, record_token_usage
 from services.end_user_service import EndUserService
 from services.feature_service import FeatureService

@@ -73,14 +70,14 @@ def validate_app_token(view: Callable[P, R] | None = None, *, fetch_user_arg: Fe

            # If caller needs end-user context, attach EndUser to current_user
            if fetch_user_arg:
-                if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
-                    user_id = request.args.get("user")
-                elif fetch_user_arg.fetch_from == WhereisUserArg.JSON:
-                    user_id = request.get_json().get("user")
-                elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
-                    user_id = request.form.get("user")
-                else:
-                    user_id = None
+                user_id = None
+                match fetch_user_arg.fetch_from:
+                    case WhereisUserArg.QUERY:
+                        user_id = request.args.get("user")
+                    case WhereisUserArg.JSON:
+                        user_id = request.get_json().get("user")
+                    case WhereisUserArg.FORM:
+                        user_id = request.form.get("user")

                if not user_id and fetch_user_arg.required:
                    raise ValueError("Arg user must be provided.")
@@ -220,6 +217,8 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
    def decorator(view: Callable[Concatenate[T, P], R]):
        @wraps(view)
        def decorated(*args: P.args, **kwargs: P.kwargs):
+            api_token = validate_and_get_api_token("dataset")
+
            # get url path dataset_id from positional args or kwargs
            # Flask passes URL path parameters as positional arguments
            dataset_id = None
@@ -256,12 +255,18 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):
            # Validate dataset if dataset_id is provided
            if dataset_id:
                dataset_id = str(dataset_id)
-                dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
+                dataset = (
+                    db.session.query(Dataset)
+                    .where(
+                        Dataset.id == dataset_id,
+                        Dataset.tenant_id == api_token.tenant_id,
+                    )
+                    .first()
+                )
                if not dataset:
                    raise NotFound("Dataset not found.")
                if not dataset.enable_api:
                    raise Forbidden("Dataset api access is not enabled.")
-            api_token = validate_and_get_api_token("dataset")
            tenant_account_join = (
                db.session.query(Tenant, TenantAccountJoin)
                .where(Tenant.id == api_token.tenant_id)
@@ -296,7 +301,14 @@ def validate_dataset_token(view: Callable[Concatenate[T, P], R] | None = None):

 def validate_and_get_api_token(scope: str | None = None):
    """
-    Validate and get API token.
+    Validate and get API token with Redis caching.
+
+    This function uses a two-tier approach:
+    1. First checks Redis cache for the token
+    2. If not cached, queries database and caches the result
+
+    The last_used_at field is updated asynchronously via Celery task
+    to avoid blocking the request.
    """
    auth_header = request.headers.get("Authorization")
    if auth_header is None or " " not in auth_header:
@@ -308,29 +320,18 @@ def validate_and_get_api_token(scope: str | None = None):
    if auth_scheme != "bearer":
        raise Unauthorized("Authorization scheme must be 'Bearer'")

-    current_time = naive_utc_now()
-    cutoff_time = current_time - timedelta(minutes=1)
-    with Session(db.engine, expire_on_commit=False) as session:
-        update_stmt = (
-            update(ApiToken)
-            .where(
-                ApiToken.token == auth_token,
-                (ApiToken.last_used_at.is_(None) | (ApiToken.last_used_at < cutoff_time)),
-                ApiToken.type == scope,
-            )
-            .values(last_used_at=current_time)
-        )
-        stmt = select(ApiToken).where(ApiToken.token == auth_token, ApiToken.type == scope)
-        result = session.execute(update_stmt)
-        api_token = session.scalar(stmt)
+    # Try to get token from cache first
+    # Returns a CachedApiToken (plain Python object), not a SQLAlchemy model
+    cached_token = ApiTokenCache.get(auth_token, scope)
+    if cached_token is not None:
+        logger.debug("Token validation served from cache for scope: %s", scope)
+        # Record usage in Redis for later batch update (no Celery task per request)
+        record_token_usage(auth_token, scope)
+        return cast(ApiToken, cached_token)

-        if hasattr(result, "rowcount") and result.rowcount > 0:
-            session.commit()
-
-        if not api_token:
-            raise Unauthorized("Access token is invalid")
-
-    return api_token
+    # Cache miss - use Redis lock for single-flight mode
+    # This ensures only one request queries DB for the same token concurrently
+    return fetch_token_with_single_flight(auth_token, scope)


 class DatasetApiResource(Resource):
@@ -65,15 +65,12 @@ def _jsonify_form_definition(form: Form, site_payload: dict | None = None) -> Re
    return Response(json.dumps(payload, ensure_ascii=False), mimetype="application/json")


-# TODO(QuantumGhost): disable authorization for web app
-# form api temporarily
-
-
@web_ns.route("/form/human_input/<string:form_token>")
-# class HumanInputFormApi(WebApiResource):
 class HumanInputFormApi(Resource):
    """API for getting and submitting human input forms via the web app."""

+    # NOTE(QuantumGhost): this endpoint is unauthenticated on purpose for now.
+
    # def get(self, _app_model: App, _end_user: EndUser, form_token: str):
    def get(self, form_token: str):
        """
@@ -14,16 +14,17 @@ class AgentConfigManager:
            agent_dict = config.get("agent_mode", {})
            agent_strategy = agent_dict.get("strategy", "cot")

-            if agent_strategy == "function_call":
-                strategy = AgentEntity.Strategy.FUNCTION_CALLING
-            elif agent_strategy in {"cot", "react"}:
-                strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
-            else:
-                # old configs, try to detect default strategy
-                if config["model"]["provider"] == "openai":
+            match agent_strategy:
+                case "function_call":
                    strategy = AgentEntity.Strategy.FUNCTION_CALLING
-                else:
+                case "cot" | "react":
                    strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT
+                case _:
+                    # old configs, try to detect default strategy
+                    if config["model"]["provider"] == "openai":
+                        strategy = AgentEntity.Strategy.FUNCTION_CALLING
+                    else:
+                        strategy = AgentEntity.Strategy.CHAIN_OF_THOUGHT

            agent_tools = []
            for tool in agent_dict.get("tools", []):
@@ -268,7 +268,7 @@ class WorkflowResponseConverter:
            data=WorkflowFinishStreamResponse.Data(
                id=run_id,
                workflow_id=workflow_id,
-                status=status.value,
+                status=status,
                outputs=encoded_outputs,
                error=error,
                elapsed_time=elapsed_time,
@@ -346,7 +346,7 @@ class WorkflowResponseConverter:
                    paused_nodes=list(event.paused_nodes),
                    outputs=encoded_outputs,
                    reasons=pause_reasons,
-                    status=WorkflowExecutionStatus.PAUSED.value,
+                    status=WorkflowExecutionStatus.PAUSED,
                    created_at=int(started_at.timestamp()),
                    elapsed_time=elapsed_time,
                    total_tokens=graph_runtime_state.total_tokens,
@@ -422,7 +422,7 @@ class WorkflowResponseConverter:
            data=WorkflowFinishStreamResponse.Data(
                id=run_id,
                workflow_id=workflow_run.workflow_id,
-                status=workflow_run.status.value,
+                status=workflow_run.status,
                outputs=encoded_outputs,
                error=workflow_run.error,
                elapsed_time=elapsed_time,
@@ -512,13 +512,13 @@ class WorkflowResponseConverter:
        metadata = self._merge_metadata(event.execution_metadata, snapshot)

        if isinstance(event, QueueNodeSucceededEvent):
-            status = WorkflowNodeExecutionStatus.SUCCEEDED.value
+            status = WorkflowNodeExecutionStatus.SUCCEEDED
            error_message = event.error
        elif isinstance(event, QueueNodeFailedEvent):
-            status = WorkflowNodeExecutionStatus.FAILED.value
+            status = WorkflowNodeExecutionStatus.FAILED
            error_message = event.error
        else:
-            status = WorkflowNodeExecutionStatus.EXCEPTION.value
+            status = WorkflowNodeExecutionStatus.EXCEPTION
            error_message = event.error

        return NodeFinishStreamResponse(
@@ -585,7 +585,7 @@ class WorkflowResponseConverter:
                process_data_truncated=process_data_truncated,
                outputs=outputs,
                outputs_truncated=outputs_truncated,
-                status=WorkflowNodeExecutionStatus.RETRY.value,
+                status=WorkflowNodeExecutionStatus.RETRY,
                error=event.error,
                elapsed_time=elapsed_time,
                execution_metadata=metadata,
@@ -120,7 +120,7 @@ class PipelineGenerator(BaseAppGenerator):
                raise ValueError("Pipeline dataset is required")
        inputs: Mapping[str, Any] = args["inputs"]
        start_node_id: str = args["start_node_id"]
-        datasource_type: str = args["datasource_type"]
+        datasource_type = DatasourceProviderType(args["datasource_type"])
        datasource_info_list: list[Mapping[str, Any]] = self._format_datasource_info_list(
            datasource_type, args["datasource_info_list"], pipeline, workflow, start_node_id, user
        )
@@ -660,7 +660,7 @@ class PipelineGenerator(BaseAppGenerator):
        tenant_id: str,
        dataset_id: str,
        built_in_field_enabled: bool,
-        datasource_type: str,
+        datasource_type: DatasourceProviderType,
        datasource_info: Mapping[str, Any],
        created_from: str,
        position: int,
@@ -668,17 +668,17 @@ class PipelineGenerator(BaseAppGenerator):
        batch: str,
        document_form: str,
    ):
-        if datasource_type == "local_file":
-            name = datasource_info.get("name", "untitled")
-        elif datasource_type == "online_document":
-            name = datasource_info.get("page", {}).get("page_name", "untitled")
-        elif datasource_type == "website_crawl":
-            name = datasource_info.get("title", "untitled")
-        elif datasource_type == "online_drive":
-            name = datasource_info.get("name", "untitled")
-        else:
-            raise ValueError(f"Unsupported datasource type: {datasource_type}")
-
+        match datasource_type:
+            case DatasourceProviderType.LOCAL_FILE:
+                name = datasource_info.get("name", "untitled")
+            case DatasourceProviderType.ONLINE_DOCUMENT:
+                name = datasource_info.get("page", {}).get("page_name", "untitled")
+            case DatasourceProviderType.WEBSITE_CRAWL:
+                name = datasource_info.get("title", "untitled")
+            case DatasourceProviderType.ONLINE_DRIVE:
+                name = datasource_info.get("name", "untitled")
+            case _:
+                raise ValueError(f"Unsupported datasource type: {datasource_type}")
        document = Document(
            tenant_id=tenant_id,
            dataset_id=dataset_id,
@@ -706,7 +706,7 @@ class PipelineGenerator(BaseAppGenerator):

    def _format_datasource_info_list(
        self,
-        datasource_type: str,
+        datasource_type: DatasourceProviderType,
        datasource_info_list: list[Mapping[str, Any]],
        pipeline: Pipeline,
        workflow: Workflow,
@@ -716,7 +716,7 @@ class PipelineGenerator(BaseAppGenerator):
        """
        Format datasource info list.
        """
-        if datasource_type == "online_drive":
+        if datasource_type == DatasourceProviderType.ONLINE_DRIVE:
            all_files: list[Mapping[str, Any]] = []
            datasource_node_data = None
            datasource_nodes = workflow.graph_dict.get("nodes", [])
@@ -8,7 +8,7 @@ from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.workflow.entities import AgentNodeStrategyInit
 from core.workflow.entities.workflow_start_reason import WorkflowStartReason
-from core.workflow.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
+from core.workflow.enums import WorkflowExecutionStatus, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from core.workflow.nodes.human_input.entities import FormInput, UserAction


@@ -231,7 +231,7 @@ class WorkflowFinishStreamResponse(StreamResponse):

        id: str
        workflow_id: str
-        status: str
+        status: WorkflowExecutionStatus
        outputs: Mapping[str, Any] | None = None
        error: str | None = None
        elapsed_time: float
@@ -262,7 +262,7 @@ class WorkflowPauseStreamResponse(StreamResponse):
        paused_nodes: Sequence[str] = Field(default_factory=list)
        outputs: Mapping[str, Any] = Field(default_factory=dict)
        reasons: Sequence[Mapping[str, Any]] = Field(default_factory=list)
-        status: str
+        status: WorkflowExecutionStatus
        created_at: int
        elapsed_time: float
        total_tokens: int
@@ -398,7 +398,7 @@ class NodeFinishStreamResponse(StreamResponse):
        process_data_truncated: bool = False
        outputs: Mapping[str, Any] | None = None
        outputs_truncated: bool = True
-        status: str
+        status: WorkflowNodeExecutionStatus
        error: str | None = None
        elapsed_time: float
        execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -462,7 +462,7 @@ class NodeRetryStreamResponse(StreamResponse):
        process_data_truncated: bool = False
        outputs: Mapping[str, Any] | None = None
        outputs_truncated: bool = False
-        status: str
+        status: WorkflowNodeExecutionStatus
        error: str | None = None
        elapsed_time: float
        execution_metadata: Mapping[WorkflowNodeExecutionMetadataKey, Any] | None = None
@@ -806,7 +806,7 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):

        id: str
        workflow_id: str
-        status: str
+        status: WorkflowExecutionStatus
        outputs: Mapping[str, Any] | None = None
        error: str | None = None
        elapsed_time: float
@@ -4,17 +4,20 @@ from typing import TYPE_CHECKING, final
 from typing_extensions import override

 from configs import dify_config
-from core.file import file_manager
-from core.helper import ssrf_proxy
+from core.file.file_manager import file_manager
 from core.helper.code_executor.code_executor import CodeExecutor
 from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.tools.tool_file_manager import ToolFileManager
+from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import NodeType
-from core.workflow.graph import NodeFactory
+from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.code_node import CodeNode
 from core.workflow.nodes.code.limits import CodeNodeLimits
 from core.workflow.nodes.http_request.node import HttpRequestNode
+from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from core.workflow.nodes.template_transform.template_renderer import (
@@ -22,7 +25,6 @@ from core.workflow.nodes.template_transform.template_renderer import (
    Jinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
-from libs.typing import is_str, is_str_dict

 if TYPE_CHECKING:
    from core.workflow.entities import GraphInitParams
@@ -47,9 +49,10 @@ class DifyNodeFactory(NodeFactory):
        code_providers: Sequence[type[CodeNodeProvider]] | None = None,
        code_limits: CodeNodeLimits | None = None,
        template_renderer: Jinja2TemplateRenderer | None = None,
-        http_request_http_client: HttpClientProtocol = ssrf_proxy,
+        template_transform_max_output_length: int | None = None,
+        http_request_http_client: HttpClientProtocol | None = None,
        http_request_tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        http_request_file_manager: FileManagerProtocol = file_manager,
+        http_request_file_manager: FileManagerProtocol | None = None,
    ) -> None:
        self.graph_init_params = graph_init_params
        self.graph_runtime_state = graph_runtime_state
@@ -68,12 +71,16 @@ class DifyNodeFactory(NodeFactory):
            max_object_array_length=dify_config.CODE_MAX_OBJECT_ARRAY_LENGTH,
        )
        self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()
-        self._http_request_http_client = http_request_http_client
+        self._template_transform_max_output_length = (
+            template_transform_max_output_length or dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
+        )
+        self._http_request_http_client = http_request_http_client or ssrf_proxy
        self._http_request_tool_file_manager_factory = http_request_tool_file_manager_factory
-        self._http_request_file_manager = http_request_file_manager
+        self._http_request_file_manager = http_request_file_manager or file_manager
+        self._rag_retrieval = DatasetRetrieval()

    @override
-    def create_node(self, node_config: dict[str, object]) -> Node:
+    def create_node(self, node_config: NodeConfigDict) -> Node:
        """
        Create a Node instance from node configuration data using the traditional mapping.

@@ -82,23 +89,14 @@ class DifyNodeFactory(NodeFactory):
        :raises ValueError: if node type is unknown or configuration is invalid
        """
        # Get node_id from config
-        node_id = node_config.get("id")
-        if not is_str(node_id):
-            raise ValueError("Node config missing id")
+        node_id = node_config["id"]

        # Get node type from config
-        node_data = node_config.get("data", {})
-        if not is_str_dict(node_data):
-            raise ValueError(f"Node {node_id} missing data information")
-
-        node_type_str = node_data.get("type")
-        if not is_str(node_type_str):
-            raise ValueError(f"Node {node_id} missing or invalid type information")
-
+        node_data = node_config["data"]
        try:
-            node_type = NodeType(node_type_str)
+            node_type = NodeType(node_data["type"])
        except ValueError:
-            raise ValueError(f"Unknown node type: {node_type_str}")
+            raise ValueError(f"Unknown node type: {node_data['type']}")

        # Get node class
        node_mapping = NODE_TYPE_CLASSES_MAPPING.get(node_type)
@@ -131,6 +129,7 @@ class DifyNodeFactory(NodeFactory):
                graph_init_params=self.graph_init_params,
                graph_runtime_state=self.graph_runtime_state,
                template_renderer=self._template_renderer,
+                max_output_length=self._template_transform_max_output_length,
            )

        if node_type == NodeType.HTTP_REQUEST:
@@ -144,6 +143,15 @@ class DifyNodeFactory(NodeFactory):
                file_manager=self._http_request_file_manager,
            )

+        if node_type == NodeType.KNOWLEDGE_RETRIEVAL:
+            return KnowledgeRetrievalNode(
+                id=node_id,
+                config=node_config,
+                graph_init_params=self.graph_init_params,
+                graph_runtime_state=self.graph_runtime_state,
+                rag_retrieval=self._rag_retrieval,
+            )
+
        return node_class(
            id=node_id,
            config=node_config,
@@ -168,3 +168,18 @@ def _to_url(f: File, /):
        return sign_tool_file(tool_file_id=f.related_id, extension=f.extension)
    else:
        raise ValueError(f"Unsupported transfer method: {f.transfer_method}")
+
+
+class FileManager:
+    """
+    Adapter exposing file manager helpers behind FileManagerProtocol.
+
+    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
+    where a protocol-typed file manager is expected.
+    """
+
+    def download(self, f: File, /) -> bytes:
+        return download(f)
+
+
+file_manager = FileManager()
@@ -47,15 +47,16 @@ class CodeNodeProvider(BaseModel, ABC):

    @classmethod
    def get_default_config(cls) -> DefaultConfig:
-        return {
-            "type": "code",
-            "config": {
-                "variables": [
-                    {"variable": "arg1", "value_selector": []},
-                    {"variable": "arg2", "value_selector": []},
-                ],
-                "code_language": cls.get_language(),
-                "code": cls.get_default_code(),
-                "outputs": {"result": {"type": "string", "children": None}},
-            },
+        variables: list[VariableConfig] = [
+            {"variable": "arg1", "value_selector": []},
+            {"variable": "arg2", "value_selector": []},
+        ]
+        outputs: dict[str, OutputConfig] = {"result": {"type": "string", "children": None}}
+
+        config: CodeConfig = {
+            "variables": variables,
+            "code_language": cls.get_language(),
+            "code": cls.get_default_code(),
+            "outputs": outputs,
        }
+        return {"type": "code", "config": config}
@@ -6,7 +6,8 @@ from yarl import URL

 from configs import dify_config
 from core.helper.download import download_with_size_limit
-from core.plugin.entities.marketplace import MarketplacePluginDeclaration
+from core.plugin.entities.marketplace import MarketplacePluginDeclaration, MarketplacePluginSnapshot
+from extensions.ext_redis import redis_client

 marketplace_api_url = URL(str(dify_config.MARKETPLACE_API_URL))
 logger = logging.getLogger(__name__)
@@ -43,28 +44,37 @@ def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
    return data.get("data", {}).get("plugins", [])


-def batch_fetch_plugin_manifests_ignore_deserialization_error(
-    plugin_ids: list[str],
-) -> Sequence[MarketplacePluginDeclaration]:
-    if len(plugin_ids) == 0:
-        return []
-
-    url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = httpx.post(url, json={"plugin_ids": plugin_ids}, headers={"X-Dify-Version": dify_config.project.version})
-    response.raise_for_status()
-    result: list[MarketplacePluginDeclaration] = []
-    for plugin in response.json()["data"]["plugins"]:
-        try:
-            result.append(MarketplacePluginDeclaration.model_validate(plugin))
-        except Exception:
-            logger.exception(
-                "Failed to deserialize marketplace plugin manifest for %s", plugin.get("plugin_id", "unknown")
-            )
-
-    return result
-
-
 def record_install_plugin_event(plugin_unique_identifier: str):
    url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
    response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier})
    response.raise_for_status()
+
+
+def fetch_global_plugin_manifest(cache_key_prefix: str, cache_ttl: int) -> None:
+    """
+    Fetch all plugin manifests from marketplace and cache them in Redis.
+    This should be called once per check cycle to populate the instance-level cache.
+
+    Args:
+        cache_key_prefix: Redis key prefix for caching plugin manifests
+        cache_ttl: Cache TTL in seconds
+
+    Raises:
+        httpx.HTTPError: If the HTTP request fails
+        Exception: If any other error occurs during fetching or caching
+    """
+    url = str(marketplace_api_url / "api/v1/dist/plugins/manifest.json")
+    response = httpx.get(url, headers={"X-Dify-Version": dify_config.project.version}, timeout=30)
+    response.raise_for_status()
+
+    raw_json = response.json()
+    plugins_data = raw_json.get("plugins", [])
+
+    # Parse and cache all plugin snapshots
+    for plugin_data in plugins_data:
+        plugin_snapshot = MarketplacePluginSnapshot.model_validate(plugin_data)
+        redis_client.setex(
+            name=f"{cache_key_prefix}{plugin_snapshot.plugin_id}",
+            time=cache_ttl,
+            value=plugin_snapshot.model_dump_json(),
+        )
@@ -230,3 +230,41 @@ def delete(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any)

 def head(url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
    return make_request("HEAD", url, max_retries=max_retries, **kwargs)
+
+
+class SSRFProxy:
+    """
+    Adapter exposing SSRF-protected HTTP helpers behind HttpClientProtocol.
+
+    This is intentionally a thin wrapper over the existing module-level functions so callers can inject it
+    where a protocol-typed HTTP client is expected.
+    """
+
+    @property
+    def max_retries_exceeded_error(self) -> type[Exception]:
+        return max_retries_exceeded_error
+
+    @property
+    def request_error(self) -> type[Exception]:
+        return request_error
+
+    def get(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return get(url=url, max_retries=max_retries, **kwargs)
+
+    def head(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return head(url=url, max_retries=max_retries, **kwargs)
+
+    def post(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return post(url=url, max_retries=max_retries, **kwargs)
+
+    def put(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return put(url=url, max_retries=max_retries, **kwargs)
+
+    def delete(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return delete(url=url, max_retries=max_retries, **kwargs)
+
+    def patch(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> httpx.Response:
+        return patch(url=url, max_retries=max_retries, **kwargs)
+
+
+ssrf_proxy = SSRFProxy()
@@ -369,77 +369,78 @@ 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)
+            preview_texts = index_processor.generate_summary_preview(
+                tenant_id, preview_texts, summary_index_setting, doc_language
+            )

        return IndexingEstimate(total_segments=total_segments, preview=preview_texts)

    def _extract(
        self, index_processor: BaseIndexProcessor, dataset_document: DatasetDocument, process_rule: dict
    ) -> list[Document]:
-        # load file
-        if dataset_document.data_source_type not in {"upload_file", "notion_import", "website_crawl"}:
-            return []
-
        data_source_info = dataset_document.data_source_info_dict
        text_docs = []
-        if dataset_document.data_source_type == "upload_file":
-            if not data_source_info or "upload_file_id" not in data_source_info:
-                raise ValueError("no upload file found")
-            stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
-            file_detail = db.session.scalars(stmt).one_or_none()
+        match dataset_document.data_source_type:
+            case "upload_file":
+                if not data_source_info or "upload_file_id" not in data_source_info:
+                    raise ValueError("no upload file found")
+                stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
+                file_detail = db.session.scalars(stmt).one_or_none()

-            if file_detail:
+                if file_detail:
+                    extract_setting = ExtractSetting(
+                        datasource_type=DatasourceType.FILE,
+                        upload_file=file_detail,
+                        document_model=dataset_document.doc_form,
+                    )
+                    text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case "notion_import":
+                if (
+                    not data_source_info
+                    or "notion_workspace_id" not in data_source_info
+                    or "notion_page_id" not in data_source_info
+                ):
+                    raise ValueError("no notion import info found")
                extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE,
-                    upload_file=file_detail,
+                    datasource_type=DatasourceType.NOTION,
+                    notion_info=NotionInfo.model_validate(
+                        {
+                            "credential_id": data_source_info.get("credential_id"),
+                            "notion_workspace_id": data_source_info["notion_workspace_id"],
+                            "notion_obj_id": data_source_info["notion_page_id"],
+                            "notion_page_type": data_source_info["type"],
+                            "document": dataset_document,
+                            "tenant_id": dataset_document.tenant_id,
+                        }
+                    ),
                    document_model=dataset_document.doc_form,
                )
                text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-        elif dataset_document.data_source_type == "notion_import":
-            if (
-                not data_source_info
-                or "notion_workspace_id" not in data_source_info
-                or "notion_page_id" not in data_source_info
-            ):
-                raise ValueError("no notion import info found")
-            extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.NOTION,
-                notion_info=NotionInfo.model_validate(
-                    {
-                        "credential_id": data_source_info.get("credential_id"),
-                        "notion_workspace_id": data_source_info["notion_workspace_id"],
-                        "notion_obj_id": data_source_info["notion_page_id"],
-                        "notion_page_type": data_source_info["type"],
-                        "document": dataset_document,
-                        "tenant_id": dataset_document.tenant_id,
-                    }
-                ),
-                document_model=dataset_document.doc_form,
-            )
-            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
-        elif dataset_document.data_source_type == "website_crawl":
-            if (
-                not data_source_info
-                or "provider" not in data_source_info
-                or "url" not in data_source_info
-                or "job_id" not in data_source_info
-            ):
-                raise ValueError("no website import info found")
-            extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.WEBSITE,
-                website_info=WebsiteInfo.model_validate(
-                    {
-                        "provider": data_source_info["provider"],
-                        "job_id": data_source_info["job_id"],
-                        "tenant_id": dataset_document.tenant_id,
-                        "url": data_source_info["url"],
-                        "mode": data_source_info["mode"],
-                        "only_main_content": data_source_info["only_main_content"],
-                    }
-                ),
-                document_model=dataset_document.doc_form,
-            )
-            text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case "website_crawl":
+                if (
+                    not data_source_info
+                    or "provider" not in data_source_info
+                    or "url" not in data_source_info
+                    or "job_id" not in data_source_info
+                ):
+                    raise ValueError("no website import info found")
+                extract_setting = ExtractSetting(
+                    datasource_type=DatasourceType.WEBSITE,
+                    website_info=WebsiteInfo.model_validate(
+                        {
+                            "provider": data_source_info["provider"],
+                            "job_id": data_source_info["job_id"],
+                            "tenant_id": dataset_document.tenant_id,
+                            "url": data_source_info["url"],
+                            "mode": data_source_info["mode"],
+                            "only_main_content": data_source_info["only_main_content"],
+                        }
+                    ),
+                    document_model=dataset_document.doc_form,
+                )
+                text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
+            case _:
+                return []
        # update document status to splitting
        self._update_document_index_status(
            document_id=dataset_document.id,
@@ -441,11 +441,13 @@ DEFAULT_GENERATOR_SUMMARY_PROMPT = (

 Requirements:
 1. Write a concise summary in plain text
-2. Use the same language as the input content
+2. You must write in {language}. No language other than {language} should be used.
 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:

@@ -88,7 +88,7 @@ PARAMETER_RULE_TEMPLATE: dict[DefaultParameterName, dict] = {
    DefaultParameterName.MAX_TOKENS: {
        "label": {
            "en_US": "Max Tokens",
-            "zh_Hans": "最大标记",
+            "zh_Hans": "最大 Token 数",
        },
        "type": "int",
        "help": {
@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, Field, computed_field, model_validator

 from core.model_runtime.entities.provider_entities import ProviderEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
@@ -48,3 +48,15 @@ class MarketplacePluginDeclaration(BaseModel):
        if "tool" in data and not data["tool"]:
            del data["tool"]
        return data
+
+
+class MarketplacePluginSnapshot(BaseModel):
+    org: str
+    name: str
+    latest_version: str
+    latest_package_identifier: str
+    latest_package_url: str
+
+    @computed_field
+    def plugin_id(self) -> str:
+        return f"{self.org}/{self.name}"
@@ -48,12 +48,22 @@ class BaseIndexProcessor(ABC):

    @abstractmethod
    def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
    ) -> 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

@@ -275,7 +275,11 @@ 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
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
    ) -> list[PreviewDetail]:
        """
        For each segment, concurrently call generate_summary to generate a summary
@@ -298,11 +302,15 @@ 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)
+                    summary, _ = self.generate_summary(
+                        tenant_id, preview.content, summary_index_setting, document_language=doc_language
+                    )
                    preview.summary = summary
            else:
                # Fallback: try without app context (may fail)
-                summary, _ = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                summary, _ = self.generate_summary(
+                    tenant_id, preview.content, summary_index_setting, document_language=doc_language
+                )
                preview.summary = summary

        # Generate summaries concurrently using ThreadPoolExecutor
@@ -356,6 +364,7 @@ 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,
@@ -366,6 +375,8 @@ 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
@@ -381,8 +392,22 @@ 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(
@@ -358,7 +358,11 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
        }

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

@@ -241,7 +241,11 @@ class QAIndexProcessor(BaseIndexProcessor):
        }

    def generate_summary_preview(
-        self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict
+        self,
+        tenant_id: str,
+        preview_texts: list[PreviewDetail],
+        summary_index_setting: dict,
+        doc_language: str | None = None,
    ) -> list[PreviewDetail]:
        """
        QA model doesn't generate summaries, so this method returns preview_texts unchanged.
@@ -1,13 +1,15 @@
 import json
+import logging
 import math
 import re
 import threading
+import time
 from collections import Counter, defaultdict
 from collections.abc import Generator, Mapping
 from typing import Any, Union, cast

 from flask import Flask, current_app
-from sqlalchemy import and_, literal, or_, select
+from sqlalchemy import and_, func, literal, or_, select
 from sqlalchemy.orm import Session

 from core.app.app_config.entities import (
@@ -18,6 +20,7 @@ from core.app.app_config.entities import (
 )
 from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
+from core.db.session_factory import session_factory
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
 from core.file import File, FileTransferMethod, FileType
@@ -58,12 +61,30 @@ from core.rag.retrieval.template_prompts import (
 )
 from core.tools.signature import sign_upload_file
 from core.tools.utils.dataset_retriever.dataset_retriever_base_tool import DatasetRetrieverBaseTool
+from core.workflow.nodes.knowledge_retrieval import exc
+from core.workflow.repositories.rag_retrieval_protocol import (
+    KnowledgeRetrievalRequest,
+    Source,
+    SourceChildChunk,
+    SourceMetadata,
+)
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from libs.json_in_md_parser import parse_and_check_json_markdown
 from models import UploadFile
-from models.dataset import ChildChunk, Dataset, DatasetMetadata, DatasetQuery, DocumentSegment, SegmentAttachmentBinding
+from models.dataset import (
+    ChildChunk,
+    Dataset,
+    DatasetMetadata,
+    DatasetQuery,
+    DocumentSegment,
+    RateLimitLog,
+    SegmentAttachmentBinding,
+)
 from models.dataset import Document as DatasetDocument
+from models.dataset import Document as DocumentModel
 from services.external_knowledge_service import ExternalDatasetService
+from services.feature_service import FeatureService

 default_retrieval_model: dict[str, Any] = {
    "search_method": RetrievalMethod.SEMANTIC_SEARCH,
@@ -73,6 +94,8 @@ default_retrieval_model: dict[str, Any] = {
    "score_threshold_enabled": False,
 }

+logger = logging.getLogger(__name__)
+

 class DatasetRetrieval:
    def __init__(self, application_generate_entity=None):
@@ -91,6 +114,233 @@ class DatasetRetrieval:
        else:
            self._llm_usage = self._llm_usage.plus(usage)

+    def knowledge_retrieval(self, request: KnowledgeRetrievalRequest) -> list[Source]:
+        self._check_knowledge_rate_limit(request.tenant_id)
+        available_datasets = self._get_available_datasets(request.tenant_id, request.dataset_ids)
+        available_datasets_ids = [i.id for i in available_datasets]
+        if not available_datasets_ids:
+            return []
+
+        if not request.query:
+            return []
+
+        metadata_filter_document_ids, metadata_condition = None, None
+
+        if request.metadata_filtering_mode != "disabled":
+            # Convert workflow layer types to app_config layer types
+            if not request.metadata_model_config:
+                raise ValueError("metadata_model_config is required for this method")
+
+            app_metadata_model_config = ModelConfig.model_validate(request.metadata_model_config.model_dump())
+
+            app_metadata_filtering_conditions = None
+            if request.metadata_filtering_conditions is not None:
+                app_metadata_filtering_conditions = MetadataFilteringCondition.model_validate(
+                    request.metadata_filtering_conditions.model_dump()
+                )
+
+            query = request.query if request.query is not None else ""
+
+            metadata_filter_document_ids, metadata_condition = self.get_metadata_filter_condition(
+                dataset_ids=available_datasets_ids,
+                query=query,
+                tenant_id=request.tenant_id,
+                user_id=request.user_id,
+                metadata_filtering_mode=request.metadata_filtering_mode,
+                metadata_model_config=app_metadata_model_config,
+                metadata_filtering_conditions=app_metadata_filtering_conditions,
+                inputs={},
+            )
+
+        if request.retrieval_mode == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE:
+            planning_strategy = PlanningStrategy.REACT_ROUTER
+            # Ensure required fields are not None for single retrieval mode
+            if request.model_provider is None or request.model_name is None or request.query is None:
+                raise ValueError("model_provider, model_name, and query are required for single retrieval mode")
+
+            model_manager = ModelManager()
+            model_instance = model_manager.get_model_instance(
+                tenant_id=request.tenant_id,
+                model_type=ModelType.LLM,
+                provider=request.model_provider,
+                model=request.model_name,
+            )
+
+            provider_model_bundle = model_instance.provider_model_bundle
+            model_type_instance = model_instance.model_type_instance
+            model_type_instance = cast(LargeLanguageModel, model_type_instance)
+
+            model_credentials = model_instance.credentials
+
+            # check model
+            provider_model = provider_model_bundle.configuration.get_provider_model(
+                model=request.model_name, model_type=ModelType.LLM
+            )
+
+            if provider_model is None:
+                raise exc.ModelNotExistError(f"Model {request.model_name} not exist.")
+
+            if provider_model.status == ModelStatus.NO_CONFIGURE:
+                raise exc.ModelCredentialsNotInitializedError(
+                    f"Model {request.model_name} credentials is not initialized."
+                )
+            elif provider_model.status == ModelStatus.NO_PERMISSION:
+                raise exc.ModelNotSupportedError(f"Dify Hosted OpenAI {request.model_name} currently not support.")
+            elif provider_model.status == ModelStatus.QUOTA_EXCEEDED:
+                raise exc.ModelQuotaExceededError(f"Model provider {request.model_provider} quota exceeded.")
+
+            stop = []
+            completion_params = (request.completion_params or {}).copy()
+            if "stop" in completion_params:
+                stop = completion_params["stop"]
+                del completion_params["stop"]
+
+            model_schema = model_type_instance.get_model_schema(request.model_name, model_credentials)
+
+            if not model_schema:
+                raise exc.ModelNotExistError(f"Model {request.model_name} not exist.")
+
+            model_config = ModelConfigWithCredentialsEntity(
+                provider=request.model_provider,
+                model=request.model_name,
+                model_schema=model_schema,
+                mode=request.model_mode or "chat",
+                provider_model_bundle=provider_model_bundle,
+                credentials=model_credentials,
+                parameters=completion_params,
+                stop=stop,
+            )
+            all_documents = self.single_retrieve(
+                request.app_id,
+                request.tenant_id,
+                request.user_id,
+                request.user_from,
+                request.query,
+                available_datasets,
+                model_instance,
+                model_config,
+                planning_strategy,
+                None,  # message_id
+                metadata_filter_document_ids,
+                metadata_condition,
+            )
+        else:
+            all_documents = self.multiple_retrieve(
+                app_id=request.app_id,
+                tenant_id=request.tenant_id,
+                user_id=request.user_id,
+                user_from=request.user_from,
+                available_datasets=available_datasets,
+                query=request.query,
+                top_k=request.top_k,
+                score_threshold=request.score_threshold,
+                reranking_mode=request.reranking_mode,
+                reranking_model=request.reranking_model,
+                weights=request.weights,
+                reranking_enable=request.reranking_enable,
+                metadata_filter_document_ids=metadata_filter_document_ids,
+                metadata_condition=metadata_condition,
+                attachment_ids=request.attachment_ids,
+            )
+
+        dify_documents = [item for item in all_documents if item.provider == "dify"]
+        external_documents = [item for item in all_documents if item.provider == "external"]
+        retrieval_resource_list = []
+        # deal with external documents
+        for item in external_documents:
+            source = Source(
+                metadata=SourceMetadata(
+                    source="knowledge",
+                    dataset_id=item.metadata.get("dataset_id"),
+                    dataset_name=item.metadata.get("dataset_name"),
+                    document_id=item.metadata.get("document_id"),
+                    document_name=item.metadata.get("title"),
+                    data_source_type="external",
+                    retriever_from="workflow",
+                    score=item.metadata.get("score"),
+                    doc_metadata=item.metadata,
+                ),
+                title=item.metadata.get("title"),
+                content=item.page_content,
+            )
+            retrieval_resource_list.append(source)
+        # deal with dify documents
+        if dify_documents:
+            records = RetrievalService.format_retrieval_documents(dify_documents)
+            dataset_ids = [i.segment.dataset_id for i in records]
+            document_ids = [i.segment.document_id for i in records]
+
+            with session_factory.create_session() as session:
+                datasets = session.query(Dataset).where(Dataset.id.in_(dataset_ids)).all()
+                documents = session.query(DatasetDocument).where(DatasetDocument.id.in_(document_ids)).all()
+
+            dataset_map = {i.id: i for i in datasets}
+            document_map = {i.id: i for i in documents}
+
+            if records:
+                for record in records:
+                    segment = record.segment
+                    dataset = dataset_map.get(segment.dataset_id)
+                    document = document_map.get(segment.document_id)
+
+                    if dataset and document:
+                        source = Source(
+                            metadata=SourceMetadata(
+                                source="knowledge",
+                                dataset_id=dataset.id,
+                                dataset_name=dataset.name,
+                                document_id=document.id,
+                                document_name=document.name,
+                                data_source_type=document.data_source_type,
+                                segment_id=segment.id,
+                                retriever_from="workflow",
+                                score=record.score or 0.0,
+                                segment_hit_count=segment.hit_count,
+                                segment_word_count=segment.word_count,
+                                segment_position=segment.position,
+                                segment_index_node_hash=segment.index_node_hash,
+                                doc_metadata=document.doc_metadata,
+                                child_chunks=[
+                                    SourceChildChunk(
+                                        id=str(getattr(chunk, "id", "")),
+                                        content=str(getattr(chunk, "content", "")),
+                                        position=int(getattr(chunk, "position", 0)),
+                                        score=float(getattr(chunk, "score", 0.0)),
+                                    )
+                                    for chunk in (record.child_chunks or [])
+                                ],
+                                position=None,
+                            ),
+                            title=document.name,
+                            files=list(record.files) if record.files else None,
+                            content=segment.get_sign_content(),
+                        )
+                        if segment.answer:
+                            source.content = f"question:{segment.get_sign_content()} \nanswer:{segment.answer}"
+
+                        if record.summary:
+                            source.summary = record.summary
+
+                        retrieval_resource_list.append(source)
+
+        if retrieval_resource_list:
+
+            def _score(item: Source) -> float:
+                meta = item.metadata
+                score = meta.score
+                if isinstance(score, (int, float)):
+                    return float(score)
+                return 0.0
+
+            retrieval_resource_list = sorted(
+                retrieval_resource_list,
+                key=_score,  # type: ignore[arg-type, return-value]
+                reverse=True,
+            )
+            for position, item in enumerate(retrieval_resource_list, start=1):
+                item.metadata.position = position  # type: ignore[index]
+        return retrieval_resource_list
+
    def retrieve(
        self,
        app_id: str,
@@ -150,14 +400,7 @@ class DatasetRetrieval:
        if features:
            if ModelFeature.TOOL_CALL in features or ModelFeature.MULTI_TOOL_CALL in features:
                planning_strategy = PlanningStrategy.ROUTER
-        available_datasets = []
-
-        dataset_stmt = select(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id.in_(dataset_ids))
-        datasets: list[Dataset] = db.session.execute(dataset_stmt).scalars().all()  # type: ignore
-        for dataset in datasets:
-            if dataset.available_document_count == 0 and dataset.provider != "external":
-                continue
-            available_datasets.append(dataset)
+        available_datasets = self._get_available_datasets(tenant_id, dataset_ids)

        if inputs:
            inputs = {key: str(value) for key, value in inputs.items()}
@@ -1161,7 +1404,6 @@ class DatasetRetrieval:
            query=query or "",
        )

-        result_text = ""
        try:
            # handle invoke result
            invoke_result = cast(
@@ -1192,7 +1434,8 @@ class DatasetRetrieval:
                                "condition": item.get("comparison_operator"),
                            }
                        )
-        except Exception:
+        except Exception as e:
+            logger.warning(e, exc_info=True)
            return None
        return automatic_metadata_filters

@@ -1406,7 +1649,12 @@ class DatasetRetrieval:
        usage = None
        for result in invoke_result:
            text = result.delta.message.content
-            full_text += text
+            if isinstance(text, str):
+                full_text += text
+            elif isinstance(text, list):
+                for i in text:
+                    if i.data:
+                        full_text += i.data

            if not model:
                model = result.model
@@ -1524,3 +1772,53 @@ class DatasetRetrieval:
                cancel_event.set()
            if thread_exceptions is not None:
                thread_exceptions.append(e)
+
+    def _get_available_datasets(self, tenant_id: str, dataset_ids: list[str]) -> list[Dataset]:
+        with session_factory.create_session() as session:
+            subquery = (
+                session.query(DocumentModel.dataset_id, func.count(DocumentModel.id).label("available_document_count"))
+                .where(
+                    DocumentModel.indexing_status == "completed",
+                    DocumentModel.enabled == True,
+                    DocumentModel.archived == False,
+                    DocumentModel.dataset_id.in_(dataset_ids),
+                )
+                .group_by(DocumentModel.dataset_id)
+                .having(func.count(DocumentModel.id) > 0)
+                .subquery()
+            )
+
+            results = (
+                session.query(Dataset)
+                .outerjoin(subquery, Dataset.id == subquery.c.dataset_id)
+                .where(Dataset.tenant_id == tenant_id, Dataset.id.in_(dataset_ids))
+                .where((subquery.c.available_document_count > 0) | (Dataset.provider == "external"))
+                .all()
+            )
+
+        available_datasets = []
+        for dataset in results:
+            if not dataset:
+                continue
+            available_datasets.append(dataset)
+        return available_datasets
+
+    def _check_knowledge_rate_limit(self, tenant_id: str):
+        knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(tenant_id)
+        if knowledge_rate_limit.enabled:
+            current_time = int(time.time() * 1000)
+            key = f"rate_limit_{tenant_id}"
+            redis_client.zadd(key, {current_time: current_time})
+            redis_client.zremrangebyscore(key, 0, current_time - 60000)
+            request_count = redis_client.zcard(key)
+            if request_count > knowledge_rate_limit.limit:
+                with session_factory.create_session() as session:
+                    rate_limit_log = RateLimitLog(
+                        tenant_id=tenant_id,
+                        subscription_plan=knowledge_rate_limit.subscription_plan,
+                        operation="knowledge",
+                    )
+                    session.add(rate_limit_log)
+                raise exc.RateLimitExceededError(
+                    "you have reached the knowledge base request rate limit of your subscription."
+                )
@@ -35,6 +35,7 @@ class SchemaRegistry:
                    registry.load_all_versions()

                    cls._default_instance = registry
+            return cls._default_instance

        return cls._default_instance

@@ -3,8 +3,8 @@ from __future__ import annotations
 import base64
 import json
 import logging
-from collections.abc import Generator
-from typing import Any
+from collections.abc import Generator, Mapping
+from typing import Any, cast

 from core.mcp.auth_client import MCPClientWithAuthRetry
 from core.mcp.error import MCPConnectionError
@@ -17,6 +17,7 @@ from core.mcp.types import (
    TextContent,
    TextResourceContents,
 )
+from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
@@ -46,6 +47,7 @@ class MCPTool(Tool):
        self.headers = headers or {}
        self.timeout = timeout
        self.sse_read_timeout = sse_read_timeout
+        self._latest_usage = LLMUsage.empty_usage()

    def tool_provider_type(self) -> ToolProviderType:
        return ToolProviderType.MCP
@@ -59,6 +61,10 @@ class MCPTool(Tool):
        message_id: str | None = None,
    ) -> Generator[ToolInvokeMessage, None, None]:
        result = self.invoke_remote_mcp_tool(tool_parameters)
+
+        # Extract usage metadata from MCP protocol's _meta field
+        self._latest_usage = self._derive_usage_from_result(result)
+
        # handle dify tool output
        for content in result.content:
            if isinstance(content, TextContent):
@@ -120,6 +126,99 @@ class MCPTool(Tool):
        for item in json_list:
            yield self.create_json_message(item)

+    @property
+    def latest_usage(self) -> LLMUsage:
+        return self._latest_usage
+
+    @classmethod
+    def _derive_usage_from_result(cls, result: CallToolResult) -> LLMUsage:
+        """
+        Extract usage metadata from MCP tool result's _meta field.
+
+        The MCP protocol's _meta field (aliased as 'meta' in Python) can contain
+        usage information such as token counts, costs, and other metadata.
+
+        Args:
+            result: The CallToolResult from MCP tool invocation
+
+        Returns:
+            LLMUsage instance with values from meta or empty_usage if not found
+        """
+        # Extract usage from the meta field if present
+        if result.meta:
+            usage_dict = cls._extract_usage_dict(result.meta)
+            if usage_dict is not None:
+                return LLMUsage.from_metadata(cast(LLMUsageMetadata, cast(object, dict(usage_dict))))
+
+        return LLMUsage.empty_usage()
+
+    @classmethod
+    def _extract_usage_dict(cls, payload: Mapping[str, Any]) -> Mapping[str, Any] | None:
+        """
+        Recursively search for usage dictionary in the payload.
+
+        The MCP protocol's _meta field can contain usage data in various formats:
+        - Direct usage field: {"usage": {...}}
+        - Nested in metadata: {"metadata": {"usage": {...}}}
+        - Or nested within other fields
+
+        Args:
+            payload: The payload to search for usage data
+
+        Returns:
+            The usage dictionary if found, None otherwise
+        """
+        # Check for direct usage field
+        usage_candidate = payload.get("usage")
+        if isinstance(usage_candidate, Mapping):
+            return usage_candidate
+
+        # Check for metadata nested usage
+        metadata_candidate = payload.get("metadata")
+        if isinstance(metadata_candidate, Mapping):
+            usage_candidate = metadata_candidate.get("usage")
+            if isinstance(usage_candidate, Mapping):
+                return usage_candidate
+
+        # Check for common token counting fields directly in payload
+        # Some MCP servers may include token counts directly
+        if "total_tokens" in payload or "prompt_tokens" in payload or "completion_tokens" in payload:
+            usage_dict: dict[str, Any] = {}
+            for key in (
+                "prompt_tokens",
+                "completion_tokens",
+                "total_tokens",
+                "prompt_unit_price",
+                "completion_unit_price",
+                "total_price",
+                "currency",
+                "prompt_price_unit",
+                "completion_price_unit",
+                "prompt_price",
+                "completion_price",
+                "latency",
+                "time_to_first_token",
+                "time_to_generate",
+            ):
+                if key in payload:
+                    usage_dict[key] = payload[key]
+            if usage_dict:
+                return usage_dict
+
+        # Recursively search through nested structures
+        for value in payload.values():
+            if isinstance(value, Mapping):
+                found = cls._extract_usage_dict(value)
+                if found is not None:
+                    return found
+            elif isinstance(value, list) and not isinstance(value, (str, bytes, bytearray)):
+                for item in value:
+                    if isinstance(item, Mapping):
+                        found = cls._extract_usage_dict(item)
+                        if found is not None:
+                            return found
+        return None
+
    def fork_tool_runtime(self, runtime: ToolRuntime) -> MCPTool:
        return MCPTool(
            entity=self.entity,
@@ -189,16 +189,13 @@ class ToolManager:
                raise ToolProviderNotFoundError(f"builtin tool {tool_name} not found")

            if not provider_controller.need_credentials:
-                return cast(
-                    BuiltinTool,
-                    builtin_tool.fork_tool_runtime(
-                        runtime=ToolRuntime(
-                            tenant_id=tenant_id,
-                            credentials={},
-                            invoke_from=invoke_from,
-                            tool_invoke_from=tool_invoke_from,
-                        )
-                    ),
+                return builtin_tool.fork_tool_runtime(
+                    runtime=ToolRuntime(
+                        tenant_id=tenant_id,
+                        credentials={},
+                        invoke_from=invoke_from,
+                        tool_invoke_from=tool_invoke_from,
+                    )
                )
            builtin_provider = None
            if isinstance(provider_controller, PluginToolProviderController):
@@ -300,18 +297,15 @@ class ToolManager:
                decrypted_credentials = refreshed_credentials.credentials
                cache.delete()

-            return cast(
-                BuiltinTool,
-                builtin_tool.fork_tool_runtime(
-                    runtime=ToolRuntime(
-                        tenant_id=tenant_id,
-                        credentials=dict(decrypted_credentials),
-                        credential_type=CredentialType.of(builtin_provider.credential_type),
-                        runtime_parameters={},
-                        invoke_from=invoke_from,
-                        tool_invoke_from=tool_invoke_from,
-                    )
-                ),
+            return builtin_tool.fork_tool_runtime(
+                runtime=ToolRuntime(
+                    tenant_id=tenant_id,
+                    credentials=dict(decrypted_credentials),
+                    credential_type=CredentialType.of(builtin_provider.credential_type),
+                    runtime_parameters={},
+                    invoke_from=invoke_from,
+                    tool_invoke_from=tool_invoke_from,
+                )
            )

        elif provider_type == ToolProviderType.API:
@@ -9,11 +9,6 @@ from core.workflow.nodes.base.entities import OutputVariableEntity


 class WorkflowToolConfigurationUtils:
-    @classmethod
-    def check_parameter_configurations(cls, configurations: list[Mapping[str, Any]]):
-        for configuration in configurations:
-            WorkflowToolParameterConfiguration.model_validate(configuration)
-
    @classmethod
    def get_workflow_graph_variables(cls, graph: Mapping[str, Any]) -> Sequence[VariableEntity]:
        """
@@ -112,7 +112,7 @@ class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):

 class RAGPipelineVariable(BaseModel):
    belong_to_node_id: str = Field(description="belong to which node id, shared means public")
-    type: str = Field(description="variable type, text-input, paragraph, select, number,  file, file-list")
+    type: str = Field(description="variable type, text-input, paragraph, select, number, file, file-list")
    label: str = Field(description="label")
    description: str | None = Field(description="description", default="")
    variable: str = Field(description="variable key", default="")
@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import sys
+
+from pydantic import TypeAdapter, with_config
+
+if sys.version_info >= (3, 12):
+    from typing import TypedDict
+else:
+    from typing_extensions import TypedDict
+
+
+@with_config(extra="allow")
+class NodeConfigData(TypedDict):
+    type: str
+
+
+@with_config(extra="allow")
+class NodeConfigDict(TypedDict):
+    id: str
+    data: NodeConfigData
+
+
+NodeConfigDictAdapter = TypeAdapter(NodeConfigDict)
@@ -5,15 +5,20 @@ from collections import defaultdict
 from collections.abc import Mapping, Sequence
 from typing import Protocol, cast, final

+from pydantic import TypeAdapter
+
+from core.workflow.entities.graph_config import NodeConfigDict
 from core.workflow.enums import ErrorStrategy, NodeExecutionType, NodeState, NodeType
 from core.workflow.nodes.base.node import Node
-from libs.typing import is_str, is_str_dict
+from libs.typing import is_str

 from .edge import Edge
 from .validation import get_graph_validator

 logger = logging.getLogger(__name__)

+_ListNodeConfigDict = TypeAdapter(list[NodeConfigDict])
+

 class NodeFactory(Protocol):
    """
@@ -23,7 +28,7 @@ class NodeFactory(Protocol):
    allowing for different node creation strategies while maintaining type safety.
    """

-    def create_node(self, node_config: dict[str, object]) -> Node:
+    def create_node(self, node_config: NodeConfigDict) -> Node:
        """
        Create a Node instance from node configuration data.

@@ -63,28 +68,24 @@ class Graph:
        self.root_node = root_node

    @classmethod
-    def _parse_node_configs(cls, node_configs: list[dict[str, object]]) -> dict[str, dict[str, object]]:
+    def _parse_node_configs(cls, node_configs: list[NodeConfigDict]) -> dict[str, NodeConfigDict]:
        """
        Parse node configurations and build a mapping of node IDs to configs.

        :param node_configs: list of node configuration dictionaries
        :return: mapping of node ID to node config
        """
-        node_configs_map: dict[str, dict[str, object]] = {}
+        node_configs_map: dict[str, NodeConfigDict] = {}

        for node_config in node_configs:
-            node_id = node_config.get("id")
-            if not node_id or not isinstance(node_id, str):
-                continue
-
-            node_configs_map[node_id] = node_config
+            node_configs_map[node_config["id"]] = node_config

        return node_configs_map

    @classmethod
    def _find_root_node_id(
        cls,
-        node_configs_map: Mapping[str, Mapping[str, object]],
+        node_configs_map: Mapping[str, NodeConfigDict],
        edge_configs: Sequence[Mapping[str, object]],
        root_node_id: str | None = None,
    ) -> str:
@@ -113,10 +114,8 @@ class Graph:
        # Prefer START node if available
        start_node_id = None
        for nid in root_candidates:
-            node_data = node_configs_map[nid].get("data")
-            if not is_str_dict(node_data):
-                continue
-            node_type = node_data.get("type")
+            node_data = node_configs_map[nid]["data"]
+            node_type = node_data["type"]
            if not isinstance(node_type, str):
                continue
            if NodeType(node_type).is_start_node:
@@ -176,7 +175,7 @@ class Graph:
    @classmethod
    def _create_node_instances(
        cls,
-        node_configs_map: dict[str, dict[str, object]],
+        node_configs_map: dict[str, NodeConfigDict],
        node_factory: NodeFactory,
    ) -> dict[str, Node]:
        """
@@ -303,7 +302,7 @@ class Graph:
        node_configs = graph_config.get("nodes", [])

        edge_configs = cast(list[dict[str, object]], edge_configs)
-        node_configs = cast(list[dict[str, object]], node_configs)
+        node_configs = _ListNodeConfigDict.validate_python(node_configs)

        if not node_configs:
            raise ValueError("Graph must have at least one node")
@@ -10,6 +10,7 @@ from pydantic import BaseModel, Field

 from core.workflow.entities.pause_reason import PauseReason
 from core.workflow.enums import NodeState
+from core.workflow.runtime.graph_runtime_state import GraphExecutionProtocol

 from .node_execution import NodeExecution

@@ -236,3 +237,6 @@ class GraphExecution:
    def record_node_failure(self) -> None:
        """Increment the count of node failures encountered during execution."""
        self.exceptions_count += 1
+
+
+_: GraphExecutionProtocol = GraphExecution(workflow_id="")
@@ -47,7 +47,6 @@ from .graph_traversal import EdgeProcessor, SkipPropagator
 from .layers.base import GraphEngineLayer
 from .orchestration import Dispatcher, ExecutionCoordinator
 from .protocols.command_channel import CommandChannel
-from .ready_queue import ReadyQueue
 from .worker_management import WorkerPool

 if TYPE_CHECKING:
@@ -94,7 +93,7 @@ class GraphEngine:
        self._graph_execution.workflow_id = workflow_id

        # === Execution Queues ===
-        self._ready_queue = cast(ReadyQueue, self._graph_runtime_state.ready_queue)
+        self._ready_queue = self._graph_runtime_state.ready_queue

        # Queue for events generated during execution
        self._event_queue: queue.Queue[GraphNodeEventBase] = queue.Queue()
@@ -15,10 +15,10 @@ from uuid import uuid4
 from pydantic import BaseModel, Field

 from core.workflow.enums import NodeExecutionType, NodeState
-from core.workflow.graph import Graph
 from core.workflow.graph_events import NodeRunStreamChunkEvent, NodeRunSucceededEvent
 from core.workflow.nodes.base.template import TextSegment, VariableSegment
 from core.workflow.runtime import VariablePool
+from core.workflow.runtime.graph_runtime_state import GraphProtocol

 from .path import Path
 from .session import ResponseSession
@@ -75,7 +75,7 @@ class ResponseStreamCoordinator:
    Ensures ordered streaming of responses based on upstream node outputs and constants.
    """

-    def __init__(self, variable_pool: "VariablePool", graph: "Graph") -> None:
+    def __init__(self, variable_pool: "VariablePool", graph: GraphProtocol) -> None:
        """
        Initialize coordinator with variable pool.

@@ -10,10 +10,10 @@ from __future__ import annotations
 from dataclasses import dataclass

 from core.workflow.nodes.answer.answer_node import AnswerNode
-from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.nodes.end.end_node import EndNode
 from core.workflow.nodes.knowledge_index import KnowledgeIndexNode
+from core.workflow.runtime.graph_runtime_state import NodeProtocol


@dataclass
@@ -29,21 +29,26 @@ class ResponseSession:
    index: int = 0  # Current position in the template segments

    @classmethod
-    def from_node(cls, node: Node) -> ResponseSession:
+    def from_node(cls, node: NodeProtocol) -> ResponseSession:
        """
-        Create a ResponseSession from an AnswerNode or EndNode.
+        Create a ResponseSession from a response-capable node.
+
+        The parameter is typed as `NodeProtocol` because the graph is exposed behind a protocol at the runtime layer,
+        but at runtime this must be an `AnswerNode`, `EndNode`, or `KnowledgeIndexNode` that provides:
+        - `id: str`
+        - `get_streaming_template() -> Template`

        Args:
-            node: Must be either an AnswerNode or EndNode instance
+            node: Node from the materialized workflow graph.

        Returns:
            ResponseSession configured with the node's streaming template

        Raises:
-            TypeError: If node is not an AnswerNode or EndNode
+            TypeError: If node is not a supported response node type.
        """
        if not isinstance(node, AnswerNode | EndNode | KnowledgeIndexNode):
-            raise TypeError
+            raise TypeError("ResponseSession.from_node only supports AnswerNode, EndNode, or KnowledgeIndexNode")
        return cls(
            node_id=node.id,
            template=node.get_streaming_template(),
@@ -192,32 +192,33 @@ class AgentNode(Node[AgentNodeData]):
                result[parameter_name] = None
                continue
            agent_input = node_data.agent_parameters[parameter_name]
-            if agent_input.type == "variable":
-                variable = variable_pool.get(agent_input.value)  # type: ignore
-                if variable is None:
-                    raise AgentVariableNotFoundError(str(agent_input.value))
-                parameter_value = variable.value
-            elif agent_input.type in {"mixed", "constant"}:
-                # variable_pool.convert_template expects a string template,
-                # but if passing a dict, convert to JSON string first before rendering
-                try:
-                    if not isinstance(agent_input.value, str):
-                        parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
-                    else:
+            match agent_input.type:
+                case "variable":
+                    variable = variable_pool.get(agent_input.value)  # type: ignore
+                    if variable is None:
+                        raise AgentVariableNotFoundError(str(agent_input.value))
+                    parameter_value = variable.value
+                case "mixed" | "constant":
+                    # variable_pool.convert_template expects a string template,
+                    # but if passing a dict, convert to JSON string first before rendering
+                    try:
+                        if not isinstance(agent_input.value, str):
+                            parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
+                        else:
+                            parameter_value = str(agent_input.value)
+                    except TypeError:
                        parameter_value = str(agent_input.value)
-                except TypeError:
-                    parameter_value = str(agent_input.value)
-                segment_group = variable_pool.convert_template(parameter_value)
-                parameter_value = segment_group.log if for_log else segment_group.text
-                # variable_pool.convert_template returns a string,
-                # so we need to convert it back to a dictionary
-                try:
-                    if not isinstance(agent_input.value, str):
-                        parameter_value = json.loads(parameter_value)
-                except json.JSONDecodeError:
-                    parameter_value = parameter_value
-            else:
-                raise AgentInputTypeError(agent_input.type)
+                    segment_group = variable_pool.convert_template(parameter_value)
+                    parameter_value = segment_group.log if for_log else segment_group.text
+                    # variable_pool.convert_template returns a string,
+                    # so we need to convert it back to a dictionary
+                    try:
+                        if not isinstance(agent_input.value, str):
+                            parameter_value = json.loads(parameter_value)
+                    except json.JSONDecodeError:
+                        parameter_value = parameter_value
+                case _:
+                    raise AgentInputTypeError(agent_input.type)
            value = parameter_value
            if parameter.type == "array[tools]":
                value = cast(list[dict[str, Any]], value)
@@ -374,12 +375,13 @@ class AgentNode(Node[AgentNodeData]):
        result: dict[str, Any] = {}
        for parameter_name in typed_node_data.agent_parameters:
            input = typed_node_data.agent_parameters[parameter_name]
-            if input.type in ["mixed", "constant"]:
-                selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
-                for selector in selectors:
-                    result[selector.variable] = selector.value_selector
-            elif input.type == "variable":
-                result[parameter_name] = input.value
+            match input.type:
+                case "mixed" | "constant":
+                    selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
+                    for selector in selectors:
+                        result[selector.variable] = selector.value_selector
+                case "variable":
+                    result[parameter_name] = input.value

        result = {node_id + "." + key: value for key, value in result.items()}

@@ -115,7 +115,7 @@ class DefaultValue(BaseModel):
    @model_validator(mode="after")
    def validate_value_type(self) -> DefaultValue:
        # Type validation configuration
-        type_validators = {
+        type_validators: dict[DefaultValueType, dict[str, Any]] = {
            DefaultValueType.STRING: {
                "type": str,
                "converter": lambda x: x,
@@ -1,4 +1,4 @@
-from typing import Annotated, Literal, Self
+from typing import Annotated, Literal

 from pydantic import AfterValidator, BaseModel

@@ -34,7 +34,7 @@ class CodeNodeData(BaseNodeData):

    class Output(BaseModel):
        type: Annotated[SegmentType, AfterValidator(_validate_type)]
-        children: dict[str, Self] | None = None
+        children: dict[str, "CodeNodeData.Output"] | None = None

    class Dependency(BaseModel):
        name: str
@@ -69,11 +69,13 @@ class DatasourceNode(Node[DatasourceNodeData]):
        if datasource_type is None:
            raise DatasourceNodeError("Datasource type is not set")

+        datasource_type = DatasourceProviderType.value_of(datasource_type)
+
        datasource_runtime = DatasourceManager.get_datasource_runtime(
            provider_id=f"{node_data.plugin_id}/{node_data.provider_name}",
            datasource_name=node_data.datasource_name or "",
            tenant_id=self.tenant_id,
-            datasource_type=DatasourceProviderType.value_of(datasource_type),
+            datasource_type=datasource_type,
        )
        datasource_info["icon"] = datasource_runtime.get_icon_url(self.tenant_id)

@@ -268,15 +270,18 @@ class DatasourceNode(Node[DatasourceNodeData]):
        if typed_node_data.datasource_parameters:
            for parameter_name in typed_node_data.datasource_parameters:
                input = typed_node_data.datasource_parameters[parameter_name]
-                if input.type == "mixed":
-                    assert isinstance(input.value, str)
-                    selectors = VariableTemplateParser(input.value).extract_variable_selectors()
-                    for selector in selectors:
-                        result[selector.variable] = selector.value_selector
-                elif input.type == "variable":
-                    result[parameter_name] = input.value
-                elif input.type == "constant":
-                    pass
+                match input.type:
+                    case "mixed":
+                        assert isinstance(input.value, str)
+                        selectors = VariableTemplateParser(input.value).extract_variable_selectors()
+                        for selector in selectors:
+                            result[selector.variable] = selector.value_selector
+                    case "variable":
+                        result[parameter_name] = input.value
+                    case "constant":
+                        pass
+                    case None:
+                        pass

            result = {node_id + "." + key: value for key, value in result.items()}

@@ -306,99 +311,107 @@ class DatasourceNode(Node[DatasourceNodeData]):
        variables: dict[str, Any] = {}

        for message in message_stream:
-            if message.type in {
-                DatasourceMessage.MessageType.IMAGE_LINK,
-                DatasourceMessage.MessageType.BINARY_LINK,
-                DatasourceMessage.MessageType.IMAGE,
-            }:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
+            match message.type:
+                case (
+                    DatasourceMessage.MessageType.IMAGE_LINK
+                    | DatasourceMessage.MessageType.BINARY_LINK
+                    | DatasourceMessage.MessageType.IMAGE
+                ):
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)

-                url = message.message.text
-                transfer_method = FileTransferMethod.TOOL_FILE
+                    url = message.message.text
+                    transfer_method = FileTransferMethod.TOOL_FILE

-                datasource_file_id = str(url).split("/")[-1].split(".")[0]
+                    datasource_file_id = str(url).split("/")[-1].split(".")[0]

-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
-                    datasource_file = session.scalar(stmt)
-                    if datasource_file is None:
-                        raise ToolFileError(f"Tool file {datasource_file_id} does not exist")
+                    with Session(db.engine) as session:
+                        stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
+                        datasource_file = session.scalar(stmt)
+                        if datasource_file is None:
+                            raise ToolFileError(f"Tool file {datasource_file_id} does not exist")

-                mapping = {
-                    "tool_file_id": datasource_file_id,
-                    "type": file_factory.get_file_type_by_mime_type(datasource_file.mimetype),
-                    "transfer_method": transfer_method,
-                    "url": url,
-                }
-                file = file_factory.build_from_mapping(
-                    mapping=mapping,
-                    tenant_id=self.tenant_id,
-                )
-                files.append(file)
-            elif message.type == DatasourceMessage.MessageType.BLOB:
-                # get tool file id
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                assert message.meta
-
-                datasource_file_id = message.message.text.split("/")[-1].split(".")[0]
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
-                    datasource_file = session.scalar(stmt)
-                    if datasource_file is None:
-                        raise ToolFileError(f"datasource file {datasource_file_id} not exists")
-
-                mapping = {
-                    "tool_file_id": datasource_file_id,
-                    "transfer_method": FileTransferMethod.TOOL_FILE,
-                }
-
-                files.append(
-                    file_factory.build_from_mapping(
+                    mapping = {
+                        "tool_file_id": datasource_file_id,
+                        "type": file_factory.get_file_type_by_mime_type(datasource_file.mimetype),
+                        "transfer_method": transfer_method,
+                        "url": url,
+                    }
+                    file = file_factory.build_from_mapping(
                        mapping=mapping,
                        tenant_id=self.tenant_id,
                    )
-                )
-            elif message.type == DatasourceMessage.MessageType.TEXT:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                text += message.message.text
-                yield StreamChunkEvent(
-                    selector=[self._node_id, "text"],
-                    chunk=message.message.text,
-                    is_final=False,
-                )
-            elif message.type == DatasourceMessage.MessageType.JSON:
-                assert isinstance(message.message, DatasourceMessage.JsonMessage)
-                json.append(message.message.json_object)
-            elif message.type == DatasourceMessage.MessageType.LINK:
-                assert isinstance(message.message, DatasourceMessage.TextMessage)
-                stream_text = f"Link: {message.message.text}\n"
-                text += stream_text
-                yield StreamChunkEvent(
-                    selector=[self._node_id, "text"],
-                    chunk=stream_text,
-                    is_final=False,
-                )
-            elif message.type == DatasourceMessage.MessageType.VARIABLE:
-                assert isinstance(message.message, DatasourceMessage.VariableMessage)
-                variable_name = message.message.variable_name
-                variable_value = message.message.variable_value
-                if message.message.stream:
-                    if not isinstance(variable_value, str):
-                        raise ValueError("When 'stream' is True, 'variable_value' must be a string.")
-                    if variable_name not in variables:
-                        variables[variable_name] = ""
-                    variables[variable_name] += variable_value
+                    files.append(file)
+                case DatasourceMessage.MessageType.BLOB:
+                    # get tool file id
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    assert message.meta

+                    datasource_file_id = message.message.text.split("/")[-1].split(".")[0]
+                    with Session(db.engine) as session:
+                        stmt = select(ToolFile).where(ToolFile.id == datasource_file_id)
+                        datasource_file = session.scalar(stmt)
+                        if datasource_file is None:
+                            raise ToolFileError(f"datasource file {datasource_file_id} not exists")
+
+                    mapping = {
+                        "tool_file_id": datasource_file_id,
+                        "transfer_method": FileTransferMethod.TOOL_FILE,
+                    }
+
+                    files.append(
+                        file_factory.build_from_mapping(
+                            mapping=mapping,
+                            tenant_id=self.tenant_id,
+                        )
+                    )
+                case DatasourceMessage.MessageType.TEXT:
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    text += message.message.text
                    yield StreamChunkEvent(
-                        selector=[self._node_id, variable_name],
-                        chunk=variable_value,
+                        selector=[self._node_id, "text"],
+                        chunk=message.message.text,
                        is_final=False,
                    )
-                else:
-                    variables[variable_name] = variable_value
-            elif message.type == DatasourceMessage.MessageType.FILE:
-                assert message.meta is not None
-                files.append(message.meta["file"])
+                case DatasourceMessage.MessageType.JSON:
+                    assert isinstance(message.message, DatasourceMessage.JsonMessage)
+                    json.append(message.message.json_object)
+                case DatasourceMessage.MessageType.LINK:
+                    assert isinstance(message.message, DatasourceMessage.TextMessage)
+                    stream_text = f"Link: {message.message.text}\n"
+                    text += stream_text
+                    yield StreamChunkEvent(
+                        selector=[self._node_id, "text"],
+                        chunk=stream_text,
+                        is_final=False,
+                    )
+                case DatasourceMessage.MessageType.VARIABLE:
+                    assert isinstance(message.message, DatasourceMessage.VariableMessage)
+                    variable_name = message.message.variable_name
+                    variable_value = message.message.variable_value
+                    if message.message.stream:
+                        if not isinstance(variable_value, str):
+                            raise ValueError("When 'stream' is True, 'variable_value' must be a string.")
+                        if variable_name not in variables:
+                            variables[variable_name] = ""
+                        variables[variable_name] += variable_value
+
+                        yield StreamChunkEvent(
+                            selector=[self._node_id, variable_name],
+                            chunk=variable_value,
+                            is_final=False,
+                        )
+                    else:
+                        variables[variable_name] = variable_value
+                case DatasourceMessage.MessageType.FILE:
+                    assert message.meta is not None
+                    files.append(message.meta["file"])
+                case (
+                    DatasourceMessage.MessageType.BLOB_CHUNK
+                    | DatasourceMessage.MessageType.LOG
+                    | DatasourceMessage.MessageType.RETRIEVER_RESOURCES
+                ):
+                    pass
+
        # mark the end of the stream
        yield StreamChunkEvent(
            selector=[self._node_id, "text"],
@@ -2,7 +2,7 @@ import base64
 import json
 import secrets
 import string
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from copy import deepcopy
 from typing import Any, Literal
 from urllib.parse import urlencode, urlparse
@@ -11,9 +11,9 @@ import httpx
 from json_repair import repair_json

 from configs import dify_config
-from core.file import file_manager
 from core.file.enums import FileTransferMethod
-from core.helper import ssrf_proxy
+from core.file.file_manager import file_manager as default_file_manager
+from core.helper.ssrf_proxy import ssrf_proxy
 from core.variables.segments import ArrayFileSegment, FileSegment
 from core.workflow.runtime import VariablePool

@@ -79,8 +79,8 @@ class Executor:
        timeout: HttpRequestNodeTimeout,
        variable_pool: VariablePool,
        max_retries: int = dify_config.SSRF_DEFAULT_MAX_RETRIES,
-        http_client: HttpClientProtocol = ssrf_proxy,
-        file_manager: FileManagerProtocol = file_manager,
+        http_client: HttpClientProtocol | None = None,
+        file_manager: FileManagerProtocol | None = None,
    ):
        # If authorization API key is present, convert the API key using the variable pool
        if node_data.authorization.type == "api-key":
@@ -107,8 +107,8 @@ class Executor:
        self.data = None
        self.json = None
        self.max_retries = max_retries
-        self._http_client = http_client
-        self._file_manager = file_manager
+        self._http_client = http_client or ssrf_proxy
+        self._file_manager = file_manager or default_file_manager

        # init template
        self.variable_pool = variable_pool
@@ -336,7 +336,7 @@ class Executor:
        """
        do http request depending on api bundle
        """
-        _METHOD_MAP = {
+        _METHOD_MAP: dict[str, Callable[..., httpx.Response]] = {
            "get": self._http_client.get,
            "head": self._http_client.head,
            "post": self._http_client.post,
@@ -348,7 +348,7 @@ class Executor:
        if method_lc not in _METHOD_MAP:
            raise InvalidHttpMethodError(f"Invalid http method {self.method}")

-        request_args = {
+        request_args: dict[str, Any] = {
            "data": self.data,
            "files": self.files,
            "json": self.json,
@@ -361,14 +361,13 @@ class Executor:
        }
        # request_args = {k: v for k, v in request_args.items() if v is not None}
        try:
-            response: httpx.Response = _METHOD_MAP[method_lc](
+            response = _METHOD_MAP[method_lc](
                url=self.url,
                **request_args,
                max_retries=self.max_retries,
            )
        except (self._http_client.max_retries_exceeded_error, self._http_client.request_error) as e:
            raise HttpRequestNodeError(str(e)) from e
-        # FIXME: fix type ignore, this maybe httpx type issue
        return response

    def invoke(self) -> Response:
@@ -4,8 +4,9 @@ from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any

 from configs import dify_config
-from core.file import File, FileTransferMethod, file_manager
-from core.helper import ssrf_proxy
+from core.file import File, FileTransferMethod
+from core.file.file_manager import file_manager as default_file_manager
+from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.tool_file_manager import ToolFileManager
 from core.variables.segments import ArrayFileSegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -47,9 +48,9 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
        graph_init_params: "GraphInitParams",
        graph_runtime_state: "GraphRuntimeState",
        *,
-        http_client: HttpClientProtocol = ssrf_proxy,
+        http_client: HttpClientProtocol | None = None,
        tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
-        file_manager: FileManagerProtocol = file_manager,
+        file_manager: FileManagerProtocol | None = None,
    ) -> None:
        super().__init__(
            id=id,
@@ -57,9 +58,9 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
            graph_init_params=graph_init_params,
            graph_runtime_state=graph_runtime_state,
        )
-        self._http_client = http_client
+        self._http_client = http_client or ssrf_proxy
        self._tool_file_manager_factory = tool_file_manager_factory
-        self._file_manager = file_manager
+        self._file_manager = file_manager or default_file_manager

    @classmethod
    def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
@@ -397,7 +397,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
            return outputs

        # Check if all non-None outputs are lists
-        non_none_outputs = [output for output in outputs if output is not None]
+        non_none_outputs: list[object] = [output for output in outputs if output is not None]
        if not non_none_outputs:
            return outputs

@@ -78,12 +78,21 @@ 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,
@@ -315,6 +324,7 @@ 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.
@@ -326,6 +336,7 @@ 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)
@@ -365,6 +376,7 @@ 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
@@ -374,6 +386,7 @@ 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
@@ -20,3 +20,7 @@ class ModelQuotaExceededError(KnowledgeRetrievalNodeError):

 class InvalidModelTypeError(KnowledgeRetrievalNodeError):
    """Raised when the model is not a Large Language Model."""
+
+
+class RateLimitExceededError(KnowledgeRetrievalNodeError):
+    """Raised when the rate limit is exceeded."""
@@ -1,29 +1,10 @@
-import json
 import logging
-import re
-import time
-from collections import defaultdict
 from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, cast
-
-from sqlalchemy import and_, func, or_, select
-from sqlalchemy.orm import sessionmaker
+from typing import TYPE_CHECKING, Any, Literal

 from core.app.app_config.entities import DatasetRetrieveConfigEntity
-from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.entities.agent_entities import PlanningStrategy
-from core.entities.model_entities import ModelStatus
-from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage
-from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.model_runtime.entities.model_entities import ModelFeature, ModelType
-from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.prompt.simple_prompt_transform import ModelMode
-from core.rag.datasource.retrieval_service import RetrievalService
-from core.rag.entities.metadata_entities import Condition, MetadataCondition
-from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
-from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.variables import (
    ArrayFileSegment,
    FileSegment,
@@ -36,35 +17,16 @@ from core.workflow.enums import (
    WorkflowNodeExecutionMetadataKey,
    WorkflowNodeExecutionStatus,
 )
-from core.workflow.node_events import ModelInvokeCompletedEvent, NodeRunResult
+from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.node import Node
-from core.workflow.nodes.knowledge_retrieval.template_prompts import (
-    METADATA_FILTER_ASSISTANT_PROMPT_1,
-    METADATA_FILTER_ASSISTANT_PROMPT_2,
-    METADATA_FILTER_COMPLETION_PROMPT,
-    METADATA_FILTER_SYSTEM_PROMPT,
-    METADATA_FILTER_USER_PROMPT_1,
-    METADATA_FILTER_USER_PROMPT_2,
-    METADATA_FILTER_USER_PROMPT_3,
-)
-from core.workflow.nodes.llm.entities import LLMNodeChatModelMessage, LLMNodeCompletionModelPromptTemplate, ModelConfig
 from core.workflow.nodes.llm.file_saver import FileSaverImpl, LLMFileSaver
-from core.workflow.nodes.llm.node import LLMNode
-from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from libs.json_in_md_parser import parse_and_check_json_markdown
-from models.dataset import Dataset, DatasetMetadata, Document, RateLimitLog
-from services.feature_service import FeatureService
+from core.workflow.repositories.rag_retrieval_protocol import KnowledgeRetrievalRequest, RAGRetrievalProtocol, Source

 from .entities import KnowledgeRetrievalNodeData
 from .exc import (
-    InvalidModelTypeError,
    KnowledgeRetrievalNodeError,
-    ModelCredentialsNotInitializedError,
-    ModelNotExistError,
-    ModelNotSupportedError,
-    ModelQuotaExceededError,
+    RateLimitExceededError,
 )

 if TYPE_CHECKING:
@@ -73,14 +35,6 @@ if TYPE_CHECKING:

 logger = logging.getLogger(__name__)

-default_retrieval_model = {
-    "search_method": RetrievalMethod.SEMANTIC_SEARCH,
-    "reranking_enable": False,
-    "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},
-    "top_k": 4,
-    "score_threshold_enabled": False,
-}
-

 class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeData]):
    node_type = NodeType.KNOWLEDGE_RETRIEVAL
@@ -97,6 +51,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        config: Mapping[str, Any],
        graph_init_params: "GraphInitParams",
        graph_runtime_state: "GraphRuntimeState",
+        rag_retrieval: RAGRetrievalProtocol,
        *,
        llm_file_saver: LLMFileSaver | None = None,
    ):
@@ -108,6 +63,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        )
        # LLM file outputs, used for MultiModal outputs.
        self._file_outputs = []
+        self._rag_retrieval = rag_retrieval

        if llm_file_saver is None:
            llm_file_saver = FileSaverImpl(
@@ -121,6 +77,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        return "1"

    def _run(self) -> NodeRunResult:
+        usage = LLMUsage.empty_usage()
        if not self._node_data.query_variable_selector and not self._node_data.query_attachment_selector:
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
@@ -128,7 +85,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                process_data={},
                outputs={},
                metadata={},
-                llm_usage=LLMUsage.empty_usage(),
+                llm_usage=usage,
            )
        variables: dict[str, Any] = {}
        # extract variables
@@ -156,36 +113,9 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
            else:
                variables["attachments"] = [variable.value]

-        # TODO(-LAN-): Move this check outside.
-        # check rate limit
-        knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(self.tenant_id)
-        if knowledge_rate_limit.enabled:
-            current_time = int(time.time() * 1000)
-            key = f"rate_limit_{self.tenant_id}"
-            redis_client.zadd(key, {current_time: current_time})
-            redis_client.zremrangebyscore(key, 0, current_time - 60000)
-            request_count = redis_client.zcard(key)
-            if request_count > knowledge_rate_limit.limit:
-                with sessionmaker(db.engine).begin() as session:
-                    # add ratelimit record
-                    rate_limit_log = RateLimitLog(
-                        tenant_id=self.tenant_id,
-                        subscription_plan=knowledge_rate_limit.subscription_plan,
-                        operation="knowledge",
-                    )
-                    session.add(rate_limit_log)
-                return NodeRunResult(
-                    status=WorkflowNodeExecutionStatus.FAILED,
-                    inputs=variables,
-                    error="Sorry, you have reached the knowledge base request rate limit of your subscription.",
-                    error_type="RateLimitExceeded",
-                )
-
-        # retrieve knowledge
-        usage = LLMUsage.empty_usage()
        try:
            results, usage = self._fetch_dataset_retriever(node_data=self._node_data, variables=variables)
-            outputs = {"result": ArrayObjectSegment(value=results)}
+            outputs = {"result": ArrayObjectSegment(value=[item.model_dump() for item in results])}
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.SUCCEEDED,
                inputs=variables,
@@ -198,9 +128,17 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                },
                llm_usage=usage,
            )
-
+        except RateLimitExceededError as e:
+            logger.warning(e, exc_info=True)
+            return NodeRunResult(
+                status=WorkflowNodeExecutionStatus.FAILED,
+                inputs=variables,
+                error=str(e),
+                error_type=type(e).__name__,
+                llm_usage=usage,
+            )
        except KnowledgeRetrievalNodeError as e:
-            logger.warning("Error when running knowledge retrieval node")
+            logger.warning("Error when running knowledge retrieval node", exc_info=True)
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.FAILED,
                inputs=variables,
@@ -210,6 +148,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
            )
        # Temporary handle all exceptions from DatasetRetrieval class here.
        except Exception as e:
+            logger.warning(e, exc_info=True)
            return NodeRunResult(
                status=WorkflowNodeExecutionStatus.FAILED,
                inputs=variables,
@@ -217,395 +156,104 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                error_type=type(e).__name__,
                llm_usage=usage,
            )
-        finally:
-            db.session.close()

    def _fetch_dataset_retriever(
        self, node_data: KnowledgeRetrievalNodeData, variables: dict[str, Any]
-    ) -> tuple[list[dict[str, Any]], LLMUsage]:
-        usage = LLMUsage.empty_usage()
-        available_datasets = []
+    ) -> tuple[list[Source], LLMUsage]:
        dataset_ids = node_data.dataset_ids
        query = variables.get("query")
        attachments = variables.get("attachments")
-        metadata_filter_document_ids = None
-        metadata_condition = None
-        metadata_usage = LLMUsage.empty_usage()
-        # Subquery: Count the number of available documents for each dataset
-        subquery = (
-            db.session.query(Document.dataset_id, func.count(Document.id).label("available_document_count"))
-            .where(
-                Document.indexing_status == "completed",
-                Document.enabled == True,
-                Document.archived == False,
-                Document.dataset_id.in_(dataset_ids),
-            )
-            .group_by(Document.dataset_id)
-            .having(func.count(Document.id) > 0)
-            .subquery()
-        )
+        retrieval_resource_list = []

-        results = (
-            db.session.query(Dataset)
-            .outerjoin(subquery, Dataset.id == subquery.c.dataset_id)
-            .where(Dataset.tenant_id == self.tenant_id, Dataset.id.in_(dataset_ids))
-            .where((subquery.c.available_document_count > 0) | (Dataset.provider == "external"))
-            .all()
-        )
+        metadata_filtering_mode: Literal["disabled", "automatic", "manual"] = "disabled"
+        if node_data.metadata_filtering_mode is not None:
+            metadata_filtering_mode = node_data.metadata_filtering_mode

-        # avoid blocking at retrieval
-        db.session.close()
-
-        for dataset in results:
-            # pass if dataset is not available
-            if not dataset:
-                continue
-            available_datasets.append(dataset)
-        if query:
-            metadata_filter_document_ids, metadata_condition, metadata_usage = self._get_metadata_filter_condition(
-                [dataset.id for dataset in available_datasets], query, node_data
-            )
-            usage = self._merge_usage(usage, metadata_usage)
-        all_documents = []
-        dataset_retrieval = DatasetRetrieval()
        if str(node_data.retrieval_mode) == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE and query:
            # fetch model config
            if node_data.single_retrieval_config is None:
-                raise ValueError("single_retrieval_config is required")
-            model_instance, model_config = self.get_model_config(node_data.single_retrieval_config.model)
-            # check model is support tool calling
-            model_type_instance = model_config.provider_model_bundle.model_type_instance
-            model_type_instance = cast(LargeLanguageModel, model_type_instance)
-            # get model schema
-            model_schema = model_type_instance.get_model_schema(
-                model=model_config.model, credentials=model_config.credentials
-            )
-
-            if model_schema:
-                planning_strategy = PlanningStrategy.REACT_ROUTER
-                features = model_schema.features
-                if features:
-                    if ModelFeature.TOOL_CALL in features or ModelFeature.MULTI_TOOL_CALL in features:
-                        planning_strategy = PlanningStrategy.ROUTER
-                all_documents = dataset_retrieval.single_retrieve(
-                    available_datasets=available_datasets,
+                raise ValueError("single_retrieval_config is required for single retrieval mode")
+            model = node_data.single_retrieval_config.model
+            retrieval_resource_list = self._rag_retrieval.knowledge_retrieval(
+                request=KnowledgeRetrievalRequest(
                    tenant_id=self.tenant_id,
                    user_id=self.user_id,
                    app_id=self.app_id,
                    user_from=self.user_from.value,
+                    dataset_ids=dataset_ids,
+                    retrieval_mode=DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE.value,
+                    completion_params=model.completion_params,
+                    model_provider=model.provider,
+                    model_mode=model.mode,
+                    model_name=model.name,
+                    metadata_model_config=node_data.metadata_model_config,
+                    metadata_filtering_conditions=node_data.metadata_filtering_conditions,
+                    metadata_filtering_mode=metadata_filtering_mode,
                    query=query,
-                    model_config=model_config,
-                    model_instance=model_instance,
-                    planning_strategy=planning_strategy,
-                    metadata_filter_document_ids=metadata_filter_document_ids,
-                    metadata_condition=metadata_condition,
                )
+            )
        elif str(node_data.retrieval_mode) == DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE:
            if node_data.multiple_retrieval_config is None:
                raise ValueError("multiple_retrieval_config is required")
-            if node_data.multiple_retrieval_config.reranking_mode == "reranking_model":
-                if node_data.multiple_retrieval_config.reranking_model:
-                    reranking_model = {
-                        "reranking_provider_name": node_data.multiple_retrieval_config.reranking_model.provider,
-                        "reranking_model_name": node_data.multiple_retrieval_config.reranking_model.model,
-                    }
-                else:
-                    reranking_model = None
-                weights = None
-            elif node_data.multiple_retrieval_config.reranking_mode == "weighted_score":
-                if node_data.multiple_retrieval_config.weights is None:
-                    raise ValueError("weights is required")
-                reranking_model = None
-                vector_setting = node_data.multiple_retrieval_config.weights.vector_setting
-                weights = {
-                    "vector_setting": {
-                        "vector_weight": vector_setting.vector_weight,
-                        "embedding_provider_name": vector_setting.embedding_provider_name,
-                        "embedding_model_name": vector_setting.embedding_model_name,
-                    },
-                    "keyword_setting": {
-                        "keyword_weight": node_data.multiple_retrieval_config.weights.keyword_setting.keyword_weight
-                    },
-                }
-            else:
-                reranking_model = None
-                weights = None
-            all_documents = dataset_retrieval.multiple_retrieve(
-                app_id=self.app_id,
-                tenant_id=self.tenant_id,
-                user_id=self.user_id,
-                user_from=self.user_from.value,
-                available_datasets=available_datasets,
-                query=query,
-                top_k=node_data.multiple_retrieval_config.top_k,
-                score_threshold=node_data.multiple_retrieval_config.score_threshold
-                if node_data.multiple_retrieval_config.score_threshold is not None
-                else 0.0,
-                reranking_mode=node_data.multiple_retrieval_config.reranking_mode,
-                reranking_model=reranking_model,
-                weights=weights,
-                reranking_enable=node_data.multiple_retrieval_config.reranking_enable,
-                metadata_filter_document_ids=metadata_filter_document_ids,
-                metadata_condition=metadata_condition,
-                attachment_ids=[attachment.related_id for attachment in attachments] if attachments else None,
-            )
-        usage = self._merge_usage(usage, dataset_retrieval.llm_usage)
-
-        dify_documents = [item for item in all_documents if item.provider == "dify"]
-        external_documents = [item for item in all_documents if item.provider == "external"]
-        retrieval_resource_list = []
-        # deal with external documents
-        for item in external_documents:
-            source: dict[str, dict[str, str | Any | dict[Any, Any] | None] | Any | str | None] = {
-                "metadata": {
-                    "_source": "knowledge",
-                    "dataset_id": item.metadata.get("dataset_id"),
-                    "dataset_name": item.metadata.get("dataset_name"),
-                    "document_id": item.metadata.get("document_id") or item.metadata.get("title"),
-                    "document_name": item.metadata.get("title"),
-                    "data_source_type": "external",
-                    "retriever_from": "workflow",
-                    "score": item.metadata.get("score"),
-                    "doc_metadata": item.metadata,
-                },
-                "title": item.metadata.get("title"),
-                "content": item.page_content,
-            }
-            retrieval_resource_list.append(source)
-        # deal with dify documents
-        if dify_documents:
-            records = RetrievalService.format_retrieval_documents(dify_documents)
-            if records:
-                for record in records:
-                    segment = record.segment
-                    dataset = db.session.query(Dataset).filter_by(id=segment.dataset_id).first()  # type: ignore
-                    stmt = select(Document).where(
-                        Document.id == segment.document_id,
-                        Document.enabled == True,
-                        Document.archived == False,
-                    )
-                    document = db.session.scalar(stmt)
-                    if dataset and document:
-                        source = {
-                            "metadata": {
-                                "_source": "knowledge",
-                                "dataset_id": dataset.id,
-                                "dataset_name": dataset.name,
-                                "document_id": document.id,
-                                "document_name": document.name,
-                                "data_source_type": document.data_source_type,
-                                "segment_id": segment.id,
-                                "retriever_from": "workflow",
-                                "score": record.score or 0.0,
-                                "child_chunks": [
-                                    {
-                                        "id": str(getattr(chunk, "id", "")),
-                                        "content": str(getattr(chunk, "content", "")),
-                                        "position": int(getattr(chunk, "position", 0)),
-                                        "score": float(getattr(chunk, "score", 0.0)),
-                                    }
-                                    for chunk in (record.child_chunks or [])
-                                ],
-                                "segment_hit_count": segment.hit_count,
-                                "segment_word_count": segment.word_count,
-                                "segment_position": segment.position,
-                                "segment_index_node_hash": segment.index_node_hash,
-                                "doc_metadata": document.doc_metadata,
-                            },
-                            "title": document.name,
-                            "files": list(record.files) if record.files else None,
+            reranking_model = None
+            weights = None
+            match node_data.multiple_retrieval_config.reranking_mode:
+                case "reranking_model":
+                    if node_data.multiple_retrieval_config.reranking_model:
+                        reranking_model = {
+                            "reranking_provider_name": node_data.multiple_retrieval_config.reranking_model.provider,
+                            "reranking_model_name": node_data.multiple_retrieval_config.reranking_model.model,
                        }
-                        if segment.answer:
-                            source["content"] = f"question:{segment.get_sign_content()} \nanswer:{segment.answer}"
-                        else:
-                            source["content"] = segment.get_sign_content()
-                        # Add summary if available
-                        if record.summary:
-                            source["summary"] = record.summary
-                        retrieval_resource_list.append(source)
-        if retrieval_resource_list:
-            retrieval_resource_list = sorted(
-                retrieval_resource_list,
-                key=self._score,  # type: ignore[arg-type, return-value]
-                reverse=True,
+                    else:
+                        reranking_model = None
+                    weights = None
+                case "weighted_score":
+                    if node_data.multiple_retrieval_config.weights is None:
+                        raise ValueError("weights is required")
+                    reranking_model = None
+                    vector_setting = node_data.multiple_retrieval_config.weights.vector_setting
+                    weights = {
+                        "vector_setting": {
+                            "vector_weight": vector_setting.vector_weight,
+                            "embedding_provider_name": vector_setting.embedding_provider_name,
+                            "embedding_model_name": vector_setting.embedding_model_name,
+                        },
+                        "keyword_setting": {
+                            "keyword_weight": node_data.multiple_retrieval_config.weights.keyword_setting.keyword_weight
+                        },
+                    }
+                case _:
+                    # Handle any other reranking_mode values
+                    reranking_model = None
+                    weights = None
+
+            retrieval_resource_list = self._rag_retrieval.knowledge_retrieval(
+                request=KnowledgeRetrievalRequest(
+                    app_id=self.app_id,
+                    tenant_id=self.tenant_id,
+                    user_id=self.user_id,
+                    user_from=self.user_from.value,
+                    dataset_ids=dataset_ids,
+                    query=query,
+                    retrieval_mode=DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE.value,
+                    top_k=node_data.multiple_retrieval_config.top_k,
+                    score_threshold=node_data.multiple_retrieval_config.score_threshold
+                    if node_data.multiple_retrieval_config.score_threshold is not None
+                    else 0.0,
+                    reranking_mode=node_data.multiple_retrieval_config.reranking_mode,
+                    reranking_model=reranking_model,
+                    weights=weights,
+                    reranking_enable=node_data.multiple_retrieval_config.reranking_enable,
+                    metadata_model_config=node_data.metadata_model_config,
+                    metadata_filtering_conditions=node_data.metadata_filtering_conditions,
+                    metadata_filtering_mode=metadata_filtering_mode,
+                    attachment_ids=[attachment.related_id for attachment in attachments] if attachments else None,
+                )
            )
-            for position, item in enumerate(retrieval_resource_list, start=1):
-                item["metadata"]["position"] = position  # type: ignore[index]
+
+        usage = self._rag_retrieval.llm_usage
        return retrieval_resource_list, usage

-    def _score(self, item: dict[str, Any]) -> float:
-        meta = item.get("metadata")
-        if isinstance(meta, dict):
-            s = meta.get("score")
-            if isinstance(s, (int, float)):
-                return float(s)
-        return 0.0
-
-    def _get_metadata_filter_condition(
-        self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
-    ) -> tuple[dict[str, list[str]] | None, MetadataCondition | None, LLMUsage]:
-        usage = LLMUsage.empty_usage()
-        document_query = db.session.query(Document).where(
-            Document.dataset_id.in_(dataset_ids),
-            Document.indexing_status == "completed",
-            Document.enabled == True,
-            Document.archived == False,
-        )
-        filters: list[Any] = []
-        metadata_condition = None
-        if node_data.metadata_filtering_mode == "disabled":
-            return None, None, usage
-        elif node_data.metadata_filtering_mode == "automatic":
-            automatic_metadata_filters, automatic_usage = self._automatic_metadata_filter_func(
-                dataset_ids, query, node_data
-            )
-            usage = self._merge_usage(usage, automatic_usage)
-            if automatic_metadata_filters:
-                conditions = []
-                for sequence, filter in enumerate(automatic_metadata_filters):
-                    DatasetRetrieval.process_metadata_filter_func(
-                        sequence,
-                        filter.get("condition", ""),
-                        filter.get("metadata_name", ""),
-                        filter.get("value"),
-                        filters,
-                    )
-                    conditions.append(
-                        Condition(
-                            name=filter.get("metadata_name"),  # type: ignore
-                            comparison_operator=filter.get("condition"),  # type: ignore
-                            value=filter.get("value"),
-                        )
-                    )
-                metadata_condition = MetadataCondition(
-                    logical_operator=node_data.metadata_filtering_conditions.logical_operator
-                    if node_data.metadata_filtering_conditions
-                    else "or",
-                    conditions=conditions,
-                )
-        elif node_data.metadata_filtering_mode == "manual":
-            if node_data.metadata_filtering_conditions:
-                conditions = []
-                for sequence, condition in enumerate(node_data.metadata_filtering_conditions.conditions):  # type: ignore
-                    metadata_name = condition.name
-                    expected_value = condition.value
-                    if expected_value is not None and condition.comparison_operator not in ("empty", "not empty"):
-                        if isinstance(expected_value, str):
-                            expected_value = self.graph_runtime_state.variable_pool.convert_template(
-                                expected_value
-                            ).value[0]
-                            if expected_value.value_type in {"number", "integer", "float"}:
-                                expected_value = expected_value.value
-                            elif expected_value.value_type == "string":
-                                expected_value = re.sub(r"[\r\n\t]+", " ", expected_value.text).strip()
-                            else:
-                                raise ValueError("Invalid expected metadata value type")
-                    conditions.append(
-                        Condition(
-                            name=metadata_name,
-                            comparison_operator=condition.comparison_operator,
-                            value=expected_value,
-                        )
-                    )
-                    filters = DatasetRetrieval.process_metadata_filter_func(
-                        sequence,
-                        condition.comparison_operator,
-                        metadata_name,
-                        expected_value,
-                        filters,
-                    )
-                metadata_condition = MetadataCondition(
-                    logical_operator=node_data.metadata_filtering_conditions.logical_operator,
-                    conditions=conditions,
-                )
-        else:
-            raise ValueError("Invalid metadata filtering mode")
-        if filters:
-            if (
-                node_data.metadata_filtering_conditions
-                and node_data.metadata_filtering_conditions.logical_operator == "and"
-            ):
-                document_query = document_query.where(and_(*filters))
-            else:
-                document_query = document_query.where(or_(*filters))
-        documents = document_query.all()
-        # group by dataset_id
-        metadata_filter_document_ids = defaultdict(list) if documents else None  # type: ignore
-        for document in documents:
-            metadata_filter_document_ids[document.dataset_id].append(document.id)  # type: ignore
-        return metadata_filter_document_ids, metadata_condition, usage
-
-    def _automatic_metadata_filter_func(
-        self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
-    ) -> tuple[list[dict[str, Any]], LLMUsage]:
-        usage = LLMUsage.empty_usage()
-        # get all metadata field
-        stmt = select(DatasetMetadata).where(DatasetMetadata.dataset_id.in_(dataset_ids))
-        metadata_fields = db.session.scalars(stmt).all()
-        all_metadata_fields = [metadata_field.name for metadata_field in metadata_fields]
-        if node_data.metadata_model_config is None:
-            raise ValueError("metadata_model_config is required")
-        # get metadata model instance and fetch model config
-        model_instance, model_config = self.get_model_config(node_data.metadata_model_config)
-        # fetch prompt messages
-        prompt_template = self._get_prompt_template(
-            node_data=node_data,
-            metadata_fields=all_metadata_fields,
-            query=query or "",
-        )
-        prompt_messages, stop = LLMNode.fetch_prompt_messages(
-            prompt_template=prompt_template,
-            sys_query=query,
-            memory=None,
-            model_config=model_config,
-            sys_files=[],
-            vision_enabled=node_data.vision.enabled,
-            vision_detail=node_data.vision.configs.detail,
-            variable_pool=self.graph_runtime_state.variable_pool,
-            jinja2_variables=[],
-            tenant_id=self.tenant_id,
-        )
-
-        result_text = ""
-        try:
-            # handle invoke result
-            generator = LLMNode.invoke_llm(
-                node_data_model=node_data.metadata_model_config,
-                model_instance=model_instance,
-                prompt_messages=prompt_messages,
-                stop=stop,
-                user_id=self.user_id,
-                structured_output_enabled=self.node_data.structured_output_enabled,
-                structured_output=None,
-                file_saver=self._llm_file_saver,
-                file_outputs=self._file_outputs,
-                node_id=self._node_id,
-                node_type=self.node_type,
-            )
-
-            for event in generator:
-                if isinstance(event, ModelInvokeCompletedEvent):
-                    result_text = event.text
-                    usage = self._merge_usage(usage, event.usage)
-                    break
-
-            result_text_json = parse_and_check_json_markdown(result_text, [])
-            automatic_metadata_filters = []
-            if "metadata_map" in result_text_json:
-                metadata_map = result_text_json["metadata_map"]
-                for item in metadata_map:
-                    if item.get("metadata_field_name") in all_metadata_fields:
-                        automatic_metadata_filters.append(
-                            {
-                                "metadata_name": item.get("metadata_field_name"),
-                                "value": item.get("metadata_field_value"),
-                                "condition": item.get("comparison_operator"),
-                            }
-                        )
-        except Exception:
-            return [], usage
-        return automatic_metadata_filters, usage
-
    @classmethod
    def _extract_variable_selector_to_variable_mapping(
        cls,
@@ -624,107 +272,3 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
        if typed_node_data.query_attachment_selector:
            variable_mapping[node_id + ".queryAttachment"] = typed_node_data.query_attachment_selector
        return variable_mapping
-
-    def get_model_config(self, model: ModelConfig) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
-        model_name = model.name
-        provider_name = model.provider
-
-        model_manager = ModelManager()
-        model_instance = model_manager.get_model_instance(
-            tenant_id=self.tenant_id, model_type=ModelType.LLM, provider=provider_name, model=model_name
-        )
-
-        provider_model_bundle = model_instance.provider_model_bundle
-        model_type_instance = model_instance.model_type_instance
-        model_type_instance = cast(LargeLanguageModel, model_type_instance)
-
-        model_credentials = model_instance.credentials
-
-        # check model
-        provider_model = provider_model_bundle.configuration.get_provider_model(
-            model=model_name, model_type=ModelType.LLM
-        )
-
-        if provider_model is None:
-            raise ModelNotExistError(f"Model {model_name} not exist.")
-
-        if provider_model.status == ModelStatus.NO_CONFIGURE:
-            raise ModelCredentialsNotInitializedError(f"Model {model_name} credentials is not initialized.")
-        elif provider_model.status == ModelStatus.NO_PERMISSION:
-            raise ModelNotSupportedError(f"Dify Hosted OpenAI {model_name} currently not support.")
-        elif provider_model.status == ModelStatus.QUOTA_EXCEEDED:
-            raise ModelQuotaExceededError(f"Model provider {provider_name} quota exceeded.")
-
-        # model config
-        completion_params = model.completion_params
-        stop = []
-        if "stop" in completion_params:
-            stop = completion_params["stop"]
-            del completion_params["stop"]
-
-        # get model mode
-        model_mode = model.mode
-        if not model_mode:
-            raise ModelNotExistError("LLM mode is required.")
-
-        model_schema = model_type_instance.get_model_schema(model_name, model_credentials)
-
-        if not model_schema:
-            raise ModelNotExistError(f"Model {model_name} not exist.")
-
-        return model_instance, ModelConfigWithCredentialsEntity(
-            provider=provider_name,
-            model=model_name,
-            model_schema=model_schema,
-            mode=model_mode,
-            provider_model_bundle=provider_model_bundle,
-            credentials=model_credentials,
-            parameters=completion_params,
-            stop=stop,
-        )
-
-    def _get_prompt_template(self, node_data: KnowledgeRetrievalNodeData, metadata_fields: list, query: str):
-        model_mode = ModelMode(node_data.metadata_model_config.mode)  # type: ignore
-        input_text = query
-
-        prompt_messages: list[LLMNodeChatModelMessage] = []
-        if model_mode == ModelMode.CHAT:
-            system_prompt_messages = LLMNodeChatModelMessage(
-                role=PromptMessageRole.SYSTEM, text=METADATA_FILTER_SYSTEM_PROMPT
-            )
-            prompt_messages.append(system_prompt_messages)
-            user_prompt_message_1 = LLMNodeChatModelMessage(
-                role=PromptMessageRole.USER, text=METADATA_FILTER_USER_PROMPT_1
-            )
-            prompt_messages.append(user_prompt_message_1)
-            assistant_prompt_message_1 = LLMNodeChatModelMessage(
-                role=PromptMessageRole.ASSISTANT, text=METADATA_FILTER_ASSISTANT_PROMPT_1
-            )
-            prompt_messages.append(assistant_prompt_message_1)
-            user_prompt_message_2 = LLMNodeChatModelMessage(
-                role=PromptMessageRole.USER, text=METADATA_FILTER_USER_PROMPT_2
-            )
-            prompt_messages.append(user_prompt_message_2)
-            assistant_prompt_message_2 = LLMNodeChatModelMessage(
-                role=PromptMessageRole.ASSISTANT, text=METADATA_FILTER_ASSISTANT_PROMPT_2
-            )
-            prompt_messages.append(assistant_prompt_message_2)
-            user_prompt_message_3 = LLMNodeChatModelMessage(
-                role=PromptMessageRole.USER,
-                text=METADATA_FILTER_USER_PROMPT_3.format(
-                    input_text=input_text,
-                    metadata_fields=json.dumps(metadata_fields, ensure_ascii=False),
-                ),
-            )
-            prompt_messages.append(user_prompt_message_3)
-            return prompt_messages
-        elif model_mode == ModelMode.COMPLETION:
-            return LLMNodeCompletionModelPromptTemplate(
-                text=METADATA_FILTER_COMPLETION_PROMPT.format(
-                    input_text=input_text,
-                    metadata_fields=json.dumps(metadata_fields, ensure_ascii=False),
-                )
-            )
-
-        else:
-            raise InvalidModelTypeError(f"Model mode {model_mode} not support.")
@@ -196,13 +196,13 @@ def _get_file_extract_string_func(*, key: str) -> Callable[[File], str]:
        case "name":
            return lambda x: x.filename or ""
        case "type":
-            return lambda x: x.type
+            return lambda x: str(x.type)
        case "extension":
            return lambda x: x.extension or ""
        case "mime_type":
            return lambda x: x.mime_type or ""
        case "transfer_method":
-            return lambda x: x.transfer_method
+            return lambda x: str(x.transfer_method)
        case "url":
            return lambda x: x.remote_url or ""
        case "related_id":
@@ -276,7 +276,6 @@ def _get_boolean_filter_func(*, condition: FilterOperator, value: bool) -> Calla


 def _get_file_filter_func(*, key: str, condition: str, value: str | Sequence[str]) -> Callable[[File], bool]:
-    extract_func: Callable[[File], Any]
    if key in {"name", "extension", "mime_type", "url", "related_id"} and isinstance(value, str):
        extract_func = _get_file_extract_string_func(key=key)
        return lambda x: _get_string_filter_func(condition=condition, value=value)(extract_func(x))
@@ -284,8 +283,8 @@ def _get_file_filter_func(*, key: str, condition: str, value: str | Sequence[str
        extract_func = _get_file_extract_string_func(key=key)
        return lambda x: _get_sequence_filter_func(condition=condition, value=value)(extract_func(x))
    elif key == "size" and isinstance(value, str):
-        extract_func = _get_file_extract_number_func(key=key)
-        return lambda x: _get_number_filter_func(condition=condition, value=float(value))(extract_func(x))
+        extract_number = _get_file_extract_number_func(key=key)
+        return lambda x: _get_number_filter_func(condition=condition, value=float(value))(extract_number(x))
    else:
        raise InvalidKeyError(f"Invalid key: {key}")

@@ -852,18 +852,16 @@ class LLMNode(Node[LLMNodeData]):
            # Insert histories into the prompt
            prompt_content = prompt_messages[0].content
            # For issue #11247 - Check if prompt content is a string or a list
-            prompt_content_type = type(prompt_content)
-            if prompt_content_type == str:
+            if isinstance(prompt_content, str):
                prompt_content = str(prompt_content)
                if "#histories#" in prompt_content:
                    prompt_content = prompt_content.replace("#histories#", memory_text)
                else:
                    prompt_content = memory_text + "\n" + prompt_content
                prompt_messages[0].content = prompt_content
-            elif prompt_content_type == list:
-                prompt_content = prompt_content if isinstance(prompt_content, list) else []
+            elif isinstance(prompt_content, list):
                for content_item in prompt_content:
-                    if content_item.type == PromptMessageContentType.TEXT:
+                    if isinstance(content_item, TextPromptMessageContent):
                        if "#histories#" in content_item.data:
                            content_item.data = content_item.data.replace("#histories#", memory_text)
                        else:
@@ -873,13 +871,12 @@ class LLMNode(Node[LLMNodeData]):

            # Add current query to the prompt message
            if sys_query:
-                if prompt_content_type == str:
+                if isinstance(prompt_content, str):
                    prompt_content = str(prompt_messages[0].content).replace("#sys.query#", sys_query)
                    prompt_messages[0].content = prompt_content
-                elif prompt_content_type == list:
-                    prompt_content = prompt_content if isinstance(prompt_content, list) else []
+                elif isinstance(prompt_content, list):
                    for content_item in prompt_content:
-                        if content_item.type == PromptMessageContentType.TEXT:
+                        if isinstance(content_item, TextPromptMessageContent):
                            content_item.data = sys_query + "\n" + content_item.data
                else:
                    raise ValueError("Invalid prompt content type")
@@ -1033,14 +1030,14 @@ class LLMNode(Node[LLMNodeData]):
        if typed_node_data.prompt_config:
            enable_jinja = False

-            if isinstance(prompt_template, list):
+            if isinstance(prompt_template, LLMNodeCompletionModelPromptTemplate):
+                if prompt_template.edition_type == "jinja2":
+                    enable_jinja = True
+            else:
                for prompt in prompt_template:
                    if prompt.edition_type == "jinja2":
                        enable_jinja = True
                        break
-            else:
-                if prompt_template.edition_type == "jinja2":
-                    enable_jinja = True

            if enable_jinja:
                for variable_selector in typed_node_data.prompt_config.jinja2_variables or []:
@@ -1,4 +1,4 @@
-from typing import Protocol
+from typing import Any, Protocol

 import httpx

@@ -12,17 +12,17 @@ class HttpClientProtocol(Protocol):
    @property
    def request_error(self) -> type[Exception]: ...

-    def get(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def get(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...

-    def head(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def head(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...

-    def post(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def post(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...

-    def put(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def put(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...

-    def delete(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def delete(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...

-    def patch(self, url: str, max_retries: int = ..., **kwargs: object) -> httpx.Response: ...
+    def patch(self, url: str, max_retries: int = ..., **kwargs: Any) -> httpx.Response: ...


 class FileManagerProtocol(Protocol):
@@ -1,7 +1,6 @@
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any

-from configs import dify_config
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
@@ -16,12 +15,13 @@ if TYPE_CHECKING:
    from core.workflow.entities import GraphInitParams
    from core.workflow.runtime import GraphRuntimeState

-MAX_TEMPLATE_TRANSFORM_OUTPUT_LENGTH = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
+DEFAULT_TEMPLATE_TRANSFORM_MAX_OUTPUT_LENGTH = 400_000


 class TemplateTransformNode(Node[TemplateTransformNodeData]):
    node_type = NodeType.TEMPLATE_TRANSFORM
    _template_renderer: Jinja2TemplateRenderer
+    _max_output_length: int

    def __init__(
        self,
@@ -31,6 +31,7 @@ class TemplateTransformNode(Node[TemplateTransformNodeData]):
        graph_runtime_state: "GraphRuntimeState",
        *,
        template_renderer: Jinja2TemplateRenderer | None = None,
+        max_output_length: int | None = None,
    ) -> None:
        super().__init__(
            id=id,
@@ -40,6 +41,10 @@ class TemplateTransformNode(Node[TemplateTransformNodeData]):
        )
        self._template_renderer = template_renderer or CodeExecutorJinja2TemplateRenderer()

+        if max_output_length is not None and max_output_length <= 0:
+            raise ValueError("max_output_length must be a positive integer")
+        self._max_output_length = max_output_length or DEFAULT_TEMPLATE_TRANSFORM_MAX_OUTPUT_LENGTH
+
    @classmethod
    def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
        """
@@ -69,11 +74,11 @@ class TemplateTransformNode(Node[TemplateTransformNodeData]):
        except TemplateRenderError as e:
            return NodeRunResult(inputs=variables, status=WorkflowNodeExecutionStatus.FAILED, error=str(e))

-        if len(rendered) > MAX_TEMPLATE_TRANSFORM_OUTPUT_LENGTH:
+        if len(rendered) > self._max_output_length:
            return NodeRunResult(
                inputs=variables,
                status=WorkflowNodeExecutionStatus.FAILED,
-                error=f"Output length exceeds {MAX_TEMPLATE_TRANSFORM_OUTPUT_LENGTH} characters",
+                error=f"Output length exceeds {self._max_output_length} characters",
            )

        return NodeRunResult(
@@ -482,16 +482,17 @@ class ToolNode(Node[ToolNodeData]):
        result = {}
        for parameter_name in typed_node_data.tool_parameters:
            input = typed_node_data.tool_parameters[parameter_name]
-            if input.type == "mixed":
-                assert isinstance(input.value, str)
-                selectors = VariableTemplateParser(input.value).extract_variable_selectors()
-                for selector in selectors:
-                    result[selector.variable] = selector.value_selector
-            elif input.type == "variable":
-                selector_key = ".".join(input.value)
-                result[f"#{selector_key}#"] = input.value
-            elif input.type == "constant":
-                pass
+            match input.type:
+                case "mixed":
+                    assert isinstance(input.value, str)
+                    selectors = VariableTemplateParser(input.value).extract_variable_selectors()
+                    for selector in selectors:
+                        result[selector.variable] = selector.value_selector
+                case "variable":
+                    selector_key = ".".join(input.value)
+                    result[f"#{selector_key}#"] = input.value
+                case "constant":
+                    pass

        result = {node_id + "." + key: value for key, value in result.items()}

@@ -0,0 +1,108 @@
+from typing import Any, Literal, Protocol
+
+from pydantic import BaseModel, Field
+
+from core.model_runtime.entities import LLMUsage
+from core.workflow.nodes.knowledge_retrieval.entities import MetadataFilteringCondition
+from core.workflow.nodes.llm.entities import ModelConfig
+
+
+class SourceChildChunk(BaseModel):
+    id: str = Field(default="", description="Child chunk ID")
+    content: str = Field(default="", description="Child chunk content")
+    position: int = Field(default=0, description="Child chunk position")
+    score: float = Field(default=0.0, description="Child chunk relevance score")
+
+
+class SourceMetadata(BaseModel):
+    source: str = Field(
+        default="knowledge",
+        serialization_alias="_source",
+        description="Data source identifier",
+    )
+    dataset_id: str = Field(description="Dataset unique identifier")
+    dataset_name: str = Field(description="Dataset display name")
+    document_id: str = Field(description="Document unique identifier")
+    document_name: str = Field(description="Document display name")
+    data_source_type: str = Field(description="Type of data source")
+    segment_id: str | None = Field(default=None, description="Segment unique identifier")
+    retriever_from: str = Field(default="workflow", description="Retriever source context")
+    score: float = Field(default=0.0, description="Retrieval relevance score")
+    child_chunks: list[SourceChildChunk] = Field(default=[], description="List of child chunks")
+    segment_hit_count: int | None = Field(default=0, description="Number of times segment was retrieved")
+    segment_word_count: int | None = Field(default=0, description="Word count of the segment")
+    segment_position: int | None = Field(default=0, description="Position of segment in document")
+    segment_index_node_hash: str | None = Field(default=None, description="Hash of index node for the segment")
+    doc_metadata: dict[str, Any] | None = Field(default=None, description="Additional document metadata")
+    position: int | None = Field(default=0, description="Position of the document in the dataset")
+
+    class Config:
+        populate_by_name = True
+
+
+class Source(BaseModel):
+    metadata: SourceMetadata = Field(description="Source metadata information")
+    title: str = Field(description="Document title")
+    files: list[Any] | None = Field(default=None, description="Associated file references")
+    content: str | None = Field(description="Segment content text")
+    summary: str | None = Field(default=None, description="Content summary if available")
+
+
+class KnowledgeRetrievalRequest(BaseModel):
+    tenant_id: str = Field(description="Tenant unique identifier")
+    user_id: str = Field(description="User unique identifier")
+    app_id: str = Field(description="Application unique identifier")
+    user_from: str = Field(description="Source of the user request (e.g., 'workflow', 'api')")
+    dataset_ids: list[str] = Field(description="List of dataset IDs to retrieve from")
+    query: str | None = Field(default=None, description="Query text for knowledge retrieval")
+    retrieval_mode: str = Field(description="Retrieval strategy: 'single' or 'multiple'")
+    model_provider: str | None = Field(default=None, description="Model provider name (e.g., 'openai', 'anthropic')")
+    completion_params: dict[str, Any] | None = Field(
+        default=None, description="Model completion parameters (e.g., temperature, max_tokens)"
+    )
+    model_mode: str | None = Field(default=None, description="Model mode (e.g., 'chat', 'completion')")
+    model_name: str | None = Field(default=None, description="Model name (e.g., 'gpt-4', 'claude-3-opus')")
+    metadata_model_config: ModelConfig | None = Field(
+        default=None, description="Model config for metadata-based filtering"
+    )
+    metadata_filtering_conditions: MetadataFilteringCondition | None = Field(
+        default=None, description="Conditions for filtering by metadata"
+    )
+    metadata_filtering_mode: Literal["disabled", "automatic", "manual"] = Field(
+        default="disabled", description="Metadata filtering mode: 'disabled', 'automatic', or 'manual'"
+    )
+    top_k: int = Field(default=0, description="Number of top results to return")
+    score_threshold: float = Field(default=0.0, description="Minimum relevance score threshold")
+    reranking_mode: str = Field(default="reranking_model", description="Reranking strategy")
+    reranking_model: dict | None = Field(default=None, description="Reranking model configuration")
+    weights: dict[str, Any] | None = Field(default=None, description="Weights for weighted score reranking")
+    reranking_enable: bool = Field(default=True, description="Whether reranking is enabled")
+    attachment_ids: list[str] | None = Field(default=None, description="List of attachment file IDs for retrieval")
+
+
+class RAGRetrievalProtocol(Protocol):
+    """Protocol for RAG-based knowledge retrieval implementations.
+
+    Implementations of this protocol handle knowledge retrieval from datasets
+    including rate limiting, dataset filtering, and document retrieval.
+    """
+
+    @property
+    def llm_usage(self) -> LLMUsage:
+        """Return accumulated LLM usage for retrieval operations."""
+        ...
+
+    def knowledge_retrieval(self, request: KnowledgeRetrievalRequest) -> list[Source]:
+        """Retrieve knowledge from datasets based on the provided request.
+
+        Args:
+            request: Knowledge retrieval request with search parameters
+
+        Returns:
+            List of sources matching the search criteria
+
+        Raises:
+            RateLimitExceededError: If rate limit is exceeded
+            ModelNotExistError: If specified model doesn't exist
+        """
+        ...
--- a/Show More
+++ b/Show More