refactor(api): move TokenBufferMemory to model_runtime

.github/workflows/pyrefly-diff.yml

@@ -29,26 +29,20 @@ jobs:
       - name: Install dependencies
         run: uv sync --project api --dev
 
-      - name: Prepare diagnostics extractor
-        run: |
-          git show ${{ github.event.pull_request.head.sha }}:api/libs/pyrefly_diagnostics.py > /tmp/pyrefly_diagnostics.py
-
       - name: Run pyrefly on PR branch
         run: |
-          uv run --directory api --dev pyrefly check 2>&1 \
-            | uv run --directory api python /tmp/pyrefly_diagnostics.py > /tmp/pyrefly_pr.txt || true
+          uv run --directory api pyrefly check > /tmp/pyrefly_pr.txt 2>&1 || true
 
       - name: Checkout base branch
         run: git checkout ${{ github.base_ref }}
 
       - name: Run pyrefly on base branch
         run: |
-          uv run --directory api --dev pyrefly check 2>&1 \
-            | uv run --directory api python /tmp/pyrefly_diagnostics.py > /tmp/pyrefly_base.txt || true
+          uv run --directory api pyrefly check > /tmp/pyrefly_base.txt 2>&1 || true
 
       - name: Compute diff
         run: |
-          diff -u /tmp/pyrefly_base.txt /tmp/pyrefly_pr.txt > pyrefly_diff.txt || true
+          diff /tmp/pyrefly_base.txt /tmp/pyrefly_pr.txt > pyrefly_diff.txt || true
 
       - name: Save PR number
         run: |

Makefile

@@ -68,9 +68,10 @@ lint:
 	@echo "✅ Linting complete"
 
 type-check:
-	@echo "📝 Running type checks (basedpyright + mypy)..."
+	@echo "📝 Running type checks (basedpyright + mypy + ty)..."
 	@./dev/basedpyright-check $(PATH_TO_CHECK)
 	@uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
+	@cd api && uv run ty check
 	@echo "✅ Type checks complete"
 
 test:
@@ -131,7 +132,7 @@ help:
 	@echo "  make format         - Format code with ruff"
 	@echo "  make check          - Check code with ruff"
 	@echo "  make lint           - Format, fix, and lint code (ruff, imports, dotenv)"
-	@echo "  make type-check     - Run type checks (basedpyright, mypy)"
+	@echo "  make type-check     - Run type checks (basedpyright, mypy, ty)"
 	@echo "  make test           - Run backend unit tests (or TARGET_TESTS=./api/tests/<target_tests>)"
 	@echo ""
 	@echo "Docker Build Targets:"

api/.importlinter

@@ -29,8 +29,6 @@ ignore_imports =
 
     core.workflow.nodes.iteration.iteration_node -> core.app.workflow.node_factory
     core.workflow.nodes.loop.loop_node -> core.app.workflow.node_factory
-    core.workflow.nodes.iteration.iteration_node -> core.app.workflow.layers.llm_quota
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
 
     core.workflow.nodes.iteration.iteration_node -> core.workflow.graph_engine
     core.workflow.nodes.iteration.iteration_node -> core.workflow.graph
@@ -54,6 +52,7 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_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
     # TODO(QuantumGhost): use DI to avoid depending on global DB.
@@ -107,12 +106,17 @@ ignore_imports =
     core.workflow.nodes.agent.agent_node -> core.provider_manager
     core.workflow.nodes.agent.agent_node -> core.tools.tool_manager
     core.workflow.nodes.document_extractor.node -> core.helper.ssrf_proxy
+    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.iteration.iteration_node -> core.app.workflow.layers.llm_quota
     core.workflow.nodes.knowledge_index.knowledge_index_node -> core.rag.index_processor.index_processor_factory
+    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.model_manager
     core.workflow.nodes.llm.protocols -> core.model_manager
     core.workflow.nodes.llm.llm_utils -> core.model_runtime.model_providers.__base.large_language_model
+    core.workflow.nodes.llm.llm_utils -> models.model
+    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.tool.tool_node -> core.callback_handler.workflow_tool_callback_handler
     core.workflow.nodes.tool.tool_node -> core.tools.tool_engine
@@ -125,19 +129,33 @@ 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.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
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.prompt.simple_prompt_transform
     core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.model_providers.__base.large_language_model
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.app.entities.app_invoke_entities
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.advanced_prompt_transform
     core.workflow.nodes.question_classifier.question_classifier_node -> core.prompt.simple_prompt_transform
+    core.workflow.nodes.start.entities -> core.app.app_config.entities
+    core.workflow.nodes.start.start_node -> core.app.app_config.entities
     core.workflow.workflow_entry -> core.app.apps.exc
     core.workflow.workflow_entry -> core.app.entities.app_invoke_entities
-    core.workflow.workflow_entry -> core.app.workflow.layers.llm_quota
     core.workflow.workflow_entry -> core.app.workflow.node_factory
+    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
+    core.workflow.nodes.llm.llm_utils -> core.variables.segments
+    core.workflow.nodes.loop.entities -> core.variables.types
     core.workflow.nodes.tool.tool_node -> core.tools.utils.message_transformer
     core.workflow.nodes.tool.tool_node -> models
     core.workflow.nodes.agent.agent_node -> models.model
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.code_node_provider
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.javascript.javascript_code_provider
+    core.workflow.nodes.code.code_node -> core.helper.code_executor.python3.python3_code_provider
+    core.workflow.nodes.code.entities -> core.helper.code_executor.code_executor
+    core.workflow.nodes.http_request.executor -> core.helper.ssrf_proxy
+    core.workflow.nodes.http_request.node -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.file_saver -> core.helper.ssrf_proxy
     core.workflow.nodes.llm.node -> core.helper.code_executor
     core.workflow.nodes.template_transform.template_renderer -> core.helper.code_executor.code_executor
@@ -146,6 +164,7 @@ ignore_imports =
     core.workflow.nodes.llm.node -> core.model_manager
     core.workflow.nodes.agent.entities -> core.prompt.entities.advanced_prompt_entities
     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
     core.workflow.nodes.llm.node -> core.prompt.utils.prompt_message_util
     core.workflow.nodes.parameter_extractor.entities -> core.prompt.entities.advanced_prompt_entities
@@ -164,9 +183,58 @@ ignore_imports =
     core.workflow.nodes.llm.file_saver -> core.tools.signature
     core.workflow.nodes.llm.file_saver -> core.tools.tool_file_manager
     core.workflow.nodes.tool.tool_node -> core.tools.errors
+    core.workflow.conversation_variable_updater -> core.variables
+    core.workflow.graph_engine.entities.commands -> core.variables.variables
+    core.workflow.nodes.agent.agent_node -> core.variables.segments
+    core.workflow.nodes.answer.answer_node -> core.variables
+    core.workflow.nodes.code.code_node -> core.variables.segments
+    core.workflow.nodes.code.code_node -> core.variables.types
+    core.workflow.nodes.code.entities -> core.variables.types
+    core.workflow.nodes.document_extractor.node -> core.variables
+    core.workflow.nodes.document_extractor.node -> core.variables.segments
+    core.workflow.nodes.http_request.executor -> core.variables.segments
+    core.workflow.nodes.http_request.node -> core.variables.segments
+    core.workflow.nodes.human_input.entities -> core.variables.consts
+    core.workflow.nodes.iteration.iteration_node -> core.variables
+    core.workflow.nodes.iteration.iteration_node -> core.variables.segments
+    core.workflow.nodes.iteration.iteration_node -> core.variables.variables
+    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables
+    core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node -> core.variables.segments
+    core.workflow.nodes.list_operator.node -> core.variables
+    core.workflow.nodes.list_operator.node -> core.variables.segments
+    core.workflow.nodes.llm.node -> core.variables
+    core.workflow.nodes.loop.loop_node -> core.variables
+    core.workflow.nodes.parameter_extractor.entities -> core.variables.types
+    core.workflow.nodes.parameter_extractor.exc -> core.variables.types
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.variables.types
+    core.workflow.nodes.tool.tool_node -> core.variables.segments
+    core.workflow.nodes.tool.tool_node -> core.variables.variables
+    core.workflow.nodes.trigger_webhook.node -> core.variables.types
+    core.workflow.nodes.trigger_webhook.node -> core.variables.variables
+    core.workflow.nodes.variable_aggregator.entities -> core.variables.types
+    core.workflow.nodes.variable_aggregator.variable_aggregator_node -> core.variables.segments
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.consts
+    core.workflow.nodes.variable_assigner.common.helpers -> core.variables.types
+    core.workflow.nodes.variable_assigner.v1.node -> core.variables
+    core.workflow.nodes.variable_assigner.v2.helpers -> core.variables
+    core.workflow.nodes.variable_assigner.v2.node -> core.variables
+    core.workflow.nodes.variable_assigner.v2.node -> core.variables.consts
+    core.workflow.runtime.graph_runtime_state_protocol -> core.variables.segments
+    core.workflow.runtime.read_only_wrappers -> core.variables.segments
+    core.workflow.runtime.variable_pool -> core.variables
+    core.workflow.runtime.variable_pool -> core.variables.consts
+    core.workflow.runtime.variable_pool -> core.variables.segments
+    core.workflow.runtime.variable_pool -> core.variables.variables
+    core.workflow.utils.condition.processor -> core.variables
+    core.workflow.utils.condition.processor -> core.variables.segments
+    core.workflow.variable_loader -> core.variables
+    core.workflow.variable_loader -> core.variables.consts
+    core.workflow.workflow_type_encoder -> core.variables
     core.workflow.nodes.agent.agent_node -> extensions.ext_database
     core.workflow.nodes.knowledge_index.knowledge_index_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.nodes.human_input.human_input_node -> extensions.ext_database
@@ -174,11 +242,16 @@ ignore_imports =
     core.workflow.workflow_entry -> extensions.otel.runtime
     core.workflow.nodes.agent.agent_node -> models
     core.workflow.nodes.base.node -> models.enums
-    core.workflow.nodes.loop.loop_node -> core.app.workflow.layers.llm_quota
+    core.workflow.nodes.llm.llm_utils -> models.provider_ids
     core.workflow.nodes.llm.node -> models.model
     core.workflow.workflow_entry -> models.enums
     core.workflow.nodes.agent.agent_node -> services
     core.workflow.nodes.tool.tool_node -> services
+    core.workflow.nodes.agent.agent_node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.llm.llm_utils -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.llm.node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.parameter_extractor.parameter_extractor_node -> core.model_runtime.token_buffer_memory
+    core.workflow.nodes.question_classifier.question_classifier_node -> core.model_runtime.token_buffer_memory
 
 [importlinter:contract:model-runtime-no-internal-imports]
 name = Model Runtime Internal Imports
@@ -231,6 +304,13 @@ ignore_imports =
     core.model_runtime.model_providers.model_provider_factory -> configs
     core.model_runtime.model_providers.model_provider_factory -> extensions.ext_redis
     core.model_runtime.model_providers.model_provider_factory -> models.provider_ids
+    core.model_runtime.token_buffer_memory -> core.app.app_config.features.file_upload.manager
+    core.model_runtime.token_buffer_memory -> core.model_manager
+    core.model_runtime.token_buffer_memory -> core.prompt.utils.extract_thread_messages
+    core.model_runtime.token_buffer_memory -> core.workflow.file.file_manager
+    core.model_runtime.token_buffer_memory -> extensions.ext_database
+    core.model_runtime.token_buffer_memory -> models.model
+    core.model_runtime.token_buffer_memory -> models.workflow
 
 [importlinter:contract:rsc]
 name = RSC

api/commands.py

@@ -2598,29 +2598,15 @@ def migrate_oss(
 @click.option(
     "--start-from",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=False,
-    default=None,
+    required=True,
     help="Lower bound (inclusive) for created_at.",
 )
 @click.option(
     "--end-before",
     type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
-    required=False,
-    default=None,
+    required=True,
     help="Upper bound (exclusive) for created_at.",
 )
-@click.option(
-    "--from-days-ago",
-    type=int,
-    default=None,
-    help="Relative lower bound in days ago (inclusive). Must be used with --before-days.",
-)
-@click.option(
-    "--before-days",
-    type=int,
-    default=None,
-    help="Relative upper bound in days ago (exclusive). Required for relative mode.",
-)
 @click.option("--batch-size", default=1000, show_default=True, help="Batch size for selecting messages.")
 @click.option(
     "--graceful-period",
@@ -2632,10 +2618,8 @@ def migrate_oss(
 def clean_expired_messages(
     batch_size: int,
     graceful_period: int,
-    start_from: datetime.datetime | None,
-    end_before: datetime.datetime | None,
-    from_days_ago: int | None,
-    before_days: int | None,
+    start_from: datetime.datetime,
+    end_before: datetime.datetime,
     dry_run: bool,
 ):
     """
@@ -2646,64 +2630,18 @@ def clean_expired_messages(
     start_at = time.perf_counter()
 
     try:
-        abs_mode = start_from is not None and end_before is not None
-        rel_mode = before_days is not None
-
-        if abs_mode and rel_mode:
-            raise click.UsageError(
-                "Options are mutually exclusive: use either (--start-from,--end-before) "
-                "or (--from-days-ago,--before-days)."
-            )
-
-        if from_days_ago is not None and before_days is None:
-            raise click.UsageError("--from-days-ago must be used together with --before-days.")
-
-        if (start_from is None) ^ (end_before is None):
-            raise click.UsageError("Both --start-from and --end-before are required when using absolute time range.")
-
-        if not abs_mode and not rel_mode:
-            raise click.UsageError(
-                "You must provide either (--start-from,--end-before) or (--before-days [--from-days-ago])."
-            )
-
-        if rel_mode:
-            if before_days < 0:
-                raise click.UsageError("--before-days must be >= 0.")
-            if from_days_ago is not None:
-                if from_days_ago < 0:
-                    raise click.UsageError("--from-days-ago must be >= 0.")
-                if from_days_ago <= before_days:
-                    raise click.UsageError("--from-days-ago must be greater than --before-days.")
-
         # Create policy based on billing configuration
         # NOTE: graceful_period will be ignored when billing is disabled.
         policy = create_message_clean_policy(graceful_period_days=graceful_period)
 
         # Create and run the cleanup service
-        if abs_mode:
-            service = MessagesCleanService.from_time_range(
-                policy=policy,
-                start_from=start_from,
-                end_before=end_before,
-                batch_size=batch_size,
-                dry_run=dry_run,
-            )
-        elif from_days_ago is None:
-            service = MessagesCleanService.from_days(
-                policy=policy,
-                days=before_days,
-                batch_size=batch_size,
-                dry_run=dry_run,
-            )
-        else:
-            now = datetime.datetime.now()
-            service = MessagesCleanService.from_time_range(
-                policy=policy,
-                start_from=now - datetime.timedelta(days=from_days_ago),
-                end_before=now - datetime.timedelta(days=before_days),
-                batch_size=batch_size,
-                dry_run=dry_run,
-            )
+        service = MessagesCleanService.from_time_range(
+            policy=policy,
+            start_from=start_from,
+            end_before=end_before,
+            batch_size=batch_size,
+            dry_run=dry_run,
+        )
         stats = service.run()
 
         end_at = time.perf_counter()

api/controllers/console/app/workflow_draft_variable.py

@@ -15,11 +15,11 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.segment_group import SegmentGroup
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from core.workflow.file import helpers as file_helpers
-from core.workflow.variables.segment_group import SegmentGroup
-from core.workflow.variables.segments import ArrayFileSegment, FileSegment, Segment
-from core.workflow.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type

api/controllers/console/datasets/rag_pipeline/rag_pipeline_draft_variable.py

@@ -21,8 +21,8 @@ from controllers.console.app.workflow_draft_variable import (
 from controllers.console.datasets.wraps import get_rag_pipeline
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
-from core.workflow.variables.types import SegmentType
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping, build_from_mappings
 from factories.variable_factory import build_segment_with_type

api/controllers/console/wraps.py

@@ -36,9 +36,9 @@ ERROR_MSG_INVALID_ENCRYPTED_DATA = "Invalid encrypted data"
 ERROR_MSG_INVALID_ENCRYPTED_CODE = "Invalid encrypted code"
 
 
-def account_initialization_required(view: Callable[P, R]) -> Callable[P, R]:
+def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
         current_user, _ = current_account_with_tenant()
         if current_user.status == AccountStatus.UNINITIALIZED:
@@ -214,9 +214,9 @@ def cloud_utm_record(view: Callable[P, R]):
     return decorated
 
 
-def setup_required(view: Callable[P, R]) -> Callable[P, R]:
+def setup_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check setup
         if (
             dify_config.EDITION == "SELF_HOSTED"

api/controllers/mcp/mcp.py

@@ -8,9 +8,9 @@ from sqlalchemy.orm import Session
 from controllers.common.schema import register_schema_model
 from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
+from core.app.app_config.entities import VariableEntity
 from core.mcp import types as mcp_types
 from core.mcp.server.streamable_http import handle_mcp_request
-from core.workflow.variables.input_entities import VariableEntity
 from extensions.ext_database import db
 from libs import helper
 from models.model import App, AppMCPServer, AppMode, EndUser

api/core/agent/base_agent_runner.py

@@ -17,7 +17,6 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
@@ -32,6 +31,7 @@ from core.model_runtime.entities import (
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from core.model_runtime.entities.model_entities import ModelFeature
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.prompt.utils.extract_thread_messages import extract_thread_messages
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (

api/core/app/app_config/easy_ui_based_app/variables/manager.py

@@ -1,8 +1,7 @@
 import re
 
-from core.app.app_config.entities import ExternalDataVariableEntity
+from core.app.app_config.entities import ExternalDataVariableEntity, VariableEntity, VariableEntityType
 from core.external_data_tool.factory import ExternalDataToolFactory
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 
 _ALLOWED_VARIABLE_ENTITY_TYPE = frozenset(
     [

api/core/app/app_config/entities.py

@@ -2,12 +2,12 @@ from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
 
-from pydantic import BaseModel, Field
+from jsonschema import Draft7Validator, SchemaError
+from pydantic import BaseModel, Field, field_validator
 
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.model_runtime.entities.message_entities import PromptMessageRole
-from core.workflow.file import FileUploadConfig
-from core.workflow.variables.input_entities import VariableEntity as WorkflowVariableEntity
+from core.workflow.file import FileTransferMethod, FileType, FileUploadConfig
 from models.model import AppMode
 
 
@@ -90,7 +90,61 @@ class PromptTemplateEntity(BaseModel):
     advanced_completion_prompt_template: AdvancedCompletionPromptTemplateEntity | None = None
 
 
-class RagPipelineVariableEntity(WorkflowVariableEntity):
+class VariableEntityType(StrEnum):
+    TEXT_INPUT = "text-input"
+    SELECT = "select"
+    PARAGRAPH = "paragraph"
+    NUMBER = "number"
+    EXTERNAL_DATA_TOOL = "external_data_tool"
+    FILE = "file"
+    FILE_LIST = "file-list"
+    CHECKBOX = "checkbox"
+    JSON_OBJECT = "json_object"
+
+
+class VariableEntity(BaseModel):
+    """
+    Variable Entity.
+    """
+
+    # `variable` records the name of the variable in user inputs.
+    variable: str
+    label: str
+    description: str = ""
+    type: VariableEntityType
+    required: bool = False
+    hide: bool = False
+    default: Any = None
+    max_length: int | None = None
+    options: Sequence[str] = Field(default_factory=list)
+    allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
+    allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
+    allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
+    json_schema: dict | None = Field(default=None)
+
+    @field_validator("description", mode="before")
+    @classmethod
+    def convert_none_description(cls, v: Any) -> str:
+        return v or ""
+
+    @field_validator("options", mode="before")
+    @classmethod
+    def convert_none_options(cls, v: Any) -> Sequence[str]:
+        return v or []
+
+    @field_validator("json_schema")
+    @classmethod
+    def validate_json_schema(cls, schema: dict | None) -> dict | None:
+        if schema is None:
+            return None
+        try:
+            Draft7Validator.check_schema(schema)
+        except SchemaError as e:
+            raise ValueError(f"Invalid JSON schema: {e.message}")
+        return schema
+
+
+class RagPipelineVariableEntity(VariableEntity):
     """
     Rag Pipeline Variable Entity.
     """
@@ -260,7 +314,7 @@ class AppConfig(BaseModel):
     app_id: str
     app_mode: AppMode
     additional_features: AppAdditionalFeatures | None = None
-    variables: list[WorkflowVariableEntity] = []
+    variables: list[VariableEntity] = []
     sensitive_word_avoidance: SensitiveWordAvoidanceEntity | None = None
 
 

api/core/app/app_config/workflow_ui_based_app/variables/manager.py

@@ -1,7 +1,6 @@
 import re
 
-from core.app.app_config.entities import RagPipelineVariableEntity
-from core.workflow.variables.input_entities import VariableEntity
+from core.app.app_config.entities import RagPipelineVariableEntity, VariableEntity
 from models.workflow import Workflow
 
 

api/core/app/apps/advanced_chat/app_runner.py

@@ -25,6 +25,7 @@ from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, Workfl
 from core.db.session_factory import session_factory
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
+from core.variables.variables import Variable
 from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
@@ -33,7 +34,6 @@ from core.workflow.repositories.workflow_node_execution_repository import Workfl
 from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
-from core.workflow.variables.variables import Variable
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client

api/core/app/apps/agent_chat/app_runner.py

@@ -12,11 +12,11 @@ from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.base_app_runner import AppRunner
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMMode
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.moderation.base import ModerationError
 from extensions.ext_database import db
 from models.model import App, Conversation, Message

api/core/app/apps/base_app_generator.py

@@ -3,6 +3,7 @@ from typing import TYPE_CHECKING, Any, Union, final
 
 from sqlalchemy.orm import Session
 
+from core.app.app_config.entities import VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.workflow.enums import NodeType
 from core.workflow.file import File, FileUploadConfig
@@ -11,14 +12,13 @@ from core.workflow.repositories.draft_variable_repository import (
     DraftVariableSaverFactory,
     NoopDraftVariableSaver,
 )
-from core.workflow.variables.input_entities import VariableEntityType
 from factories import file_factory
 from libs.orjson import orjson_dumps
 from models import Account, EndUser
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
-    from core.workflow.variables.input_entities import VariableEntity
+    from core.app.app_config.entities import VariableEntity
 
 
 class BaseAppGenerator:

api/core/app/apps/base_app_runner.py

@@ -22,7 +22,6 @@ from core.app.entities.queue_entities import (
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
 from core.app.features.hosting_moderation.hosting_moderation import HostingModerationFeature
 from core.external_data_tool.external_data_fetch import ExternalDataFetch
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMResultChunk, LLMResultChunkDelta, LLMUsage
 from core.model_runtime.entities.message_entities import (
@@ -33,6 +32,7 @@ from core.model_runtime.entities.message_entities import (
 )
 from core.model_runtime.entities.model_entities import ModelPropertyKey
 from core.model_runtime.errors.invoke import InvokeBadRequestError
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.moderation.input_moderation import InputModeration
 from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig

api/core/app/apps/chat/app_runner.py

@@ -11,9 +11,9 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.entities.queue_entities import QueueAnnotationReplyEvent
 from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCallbackHandler
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.moderation.base import ModerationError
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.workflow.file import File

api/core/app/apps/common/workflow_response_converter.py

@@ -49,6 +49,7 @@ from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
 from core.trigger.trigger_manager import TriggerManager
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities.pause_reason import HumanInputRequired
 from core.workflow.entities.workflow_start_reason import WorkflowStartReason
 from core.workflow.enums import (
@@ -61,7 +62,6 @@ from core.workflow.enums import (
 from core.workflow.file import FILE_MODEL_IDENTITY, File
 from core.workflow.runtime import GraphRuntimeState
 from core.workflow.system_variable import SystemVariable
-from core.workflow.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.workflow_entry import WorkflowEntry
 from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from extensions.ext_database import db

api/core/app/apps/pipeline/pipeline_runner.py

@@ -11,6 +11,7 @@ from core.app.entities.app_invoke_entities import (
 )
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.app.workflow.node_factory import DifyNodeFactory
+from core.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from core.workflow.entities.graph_init_params import GraphInitParams
 from core.workflow.enums import WorkflowType
 from core.workflow.graph import Graph
@@ -20,7 +21,6 @@ from core.workflow.repositories.workflow_node_execution_repository import Workfl
 from core.workflow.runtime import GraphRuntimeState, VariablePool
 from core.workflow.system_variable import SystemVariable
 from core.workflow.variable_loader import VariableLoader
-from core.workflow.variables.variables import RAGPipelineVariable, RAGPipelineVariableInput
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.dataset import Document, Pipeline

api/core/app/layers/conversation_variable_persist_layer.py

@@ -1,12 +1,12 @@
 import logging
 
+from core.variables import VariableBase
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.conversation_variable_updater import ConversationVariableUpdater
 from core.workflow.enums import NodeType
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_events import GraphEngineEvent, NodeRunSucceededEvent
 from core.workflow.nodes.variable_assigner.common import helpers as common_helpers
-from core.workflow.variables import VariableBase
 
 logger = logging.getLogger(__name__)
 

api/core/app/llm/__init__.py

@@ -1,5 +1 @@
 """LLM-related application services."""
-
-from .quota import deduct_llm_quota, ensure_llm_quota_available
-
-__all__ = ["deduct_llm_quota", "ensure_llm_quota_available"]

api/core/app/llm/model_access.py

@@ -83,21 +83,14 @@ def fetch_model_config(
         raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
     provider_model.raise_for_status()
 
-    completion_params = dict(node_data_model.completion_params)
-    stop = completion_params.pop("stop", [])
-    if not isinstance(stop, list):
-        stop = []
+    stop: list[str] = []
+    if "stop" in node_data_model.completion_params:
+        stop = node_data_model.completion_params.pop("stop")
 
     model_schema = model_instance.model_type_instance.get_model_schema(node_data_model.name, credentials)
     if not model_schema:
         raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
 
-    model_instance.provider = node_data_model.provider
-    model_instance.model_name = node_data_model.name
-    model_instance.credentials = credentials
-    model_instance.parameters = completion_params
-    model_instance.stop = tuple(stop)
-
     return model_instance, ModelConfigWithCredentialsEntity(
         provider=node_data_model.provider,
         model=node_data_model.name,
@@ -105,6 +98,6 @@ def fetch_model_config(
         mode=node_data_model.mode,
         provider_model_bundle=provider_model_bundle,
         credentials=credentials,
-        parameters=completion_params,
+        parameters=node_data_model.completion_params,
         stop=stop,
     )

api/core/app/llm/quota.py

@@ -1,93 +0,0 @@
-from sqlalchemy import update
-from sqlalchemy.orm import Session
-
-from configs import dify_config
-from core.entities.model_entities import ModelStatus
-from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
-from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
-from core.model_runtime.entities.llm_entities import LLMUsage
-from extensions.ext_database import db
-from libs.datetime_utils import naive_utc_now
-from models.provider import Provider, ProviderType
-from models.provider_ids import ModelProviderID
-
-
-def ensure_llm_quota_available(*, model_instance: ModelInstance) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
-
-    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
-        return
-
-    provider_model = provider_configuration.get_provider_model(
-        model_type=model_instance.model_type_instance.model_type,
-        model=model_instance.model_name,
-    )
-    if provider_model and provider_model.status == ModelStatus.QUOTA_EXCEEDED:
-        raise QuotaExceededError(f"Model provider {model_instance.provider} quota exceeded.")
-
-
-def deduct_llm_quota(*, tenant_id: str, model_instance: ModelInstance, usage: LLMUsage) -> None:
-    provider_model_bundle = model_instance.provider_model_bundle
-    provider_configuration = provider_model_bundle.configuration
-
-    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
-        return
-
-    system_configuration = provider_configuration.system_configuration
-
-    quota_unit = None
-    for quota_configuration in system_configuration.quota_configurations:
-        if quota_configuration.quota_type == system_configuration.current_quota_type:
-            quota_unit = quota_configuration.quota_unit
-
-            if quota_configuration.quota_limit == -1:
-                return
-
-            break
-
-    used_quota = None
-    if quota_unit:
-        if quota_unit == QuotaUnit.TOKENS:
-            used_quota = usage.total_tokens
-        elif quota_unit == QuotaUnit.CREDITS:
-            used_quota = dify_config.get_model_credits(model_instance.model_name)
-        else:
-            used_quota = 1
-
-    if used_quota is not None and system_configuration.current_quota_type is not None:
-        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
-            from services.credit_pool_service import CreditPoolService
-
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-            )
-        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
-            from services.credit_pool_service import CreditPoolService
-
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-                pool_type="paid",
-            )
-        else:
-            with Session(db.engine) as session:
-                stmt = (
-                    update(Provider)
-                    .where(
-                        Provider.tenant_id == tenant_id,
-                        # TODO: Use provider name with prefix after the data migration.
-                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                        Provider.provider_type == ProviderType.SYSTEM.value,
-                        Provider.quota_type == system_configuration.current_quota_type.value,
-                        Provider.quota_limit > Provider.quota_used,
-                    )
-                    .values(
-                        quota_used=Provider.quota_used + used_quota,
-                        last_used=naive_utc_now(),
-                    )
-                )
-                session.execute(stmt)
-                session.commit()

api/core/app/workflow/layers/__init__.py

@@ -1,11 +1,9 @@
 """Workflow-level GraphEngine layers that depend on outer infrastructure."""
 
-from .llm_quota import LLMQuotaLayer
 from .observability import ObservabilityLayer
 from .persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 
 __all__ = [
-    "LLMQuotaLayer",
     "ObservabilityLayer",
     "PersistenceWorkflowInfo",
     "WorkflowPersistenceLayer",

api/core/app/workflow/layers/llm_quota.py

@@ -1,128 +0,0 @@
-"""
-LLM quota deduction layer for GraphEngine.
-
-This layer centralizes model-quota deduction outside node implementations.
-"""
-
-import logging
-from typing import TYPE_CHECKING, cast, final
-
-from typing_extensions import override
-
-from core.app.llm import deduct_llm_quota, ensure_llm_quota_available
-from core.errors.error import QuotaExceededError
-from core.model_manager import ModelInstance
-from core.workflow.enums import NodeType
-from core.workflow.graph_engine.entities.commands import AbortCommand, CommandType
-from core.workflow.graph_engine.layers.base import GraphEngineLayer
-from core.workflow.graph_events import GraphEngineEvent, GraphNodeEventBase
-from core.workflow.graph_events.node import NodeRunSucceededEvent
-from core.workflow.nodes.base.node import Node
-
-if TYPE_CHECKING:
-    from core.workflow.nodes.llm.node import LLMNode
-    from core.workflow.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
-    from core.workflow.nodes.question_classifier.question_classifier_node import QuestionClassifierNode
-
-logger = logging.getLogger(__name__)
-
-
-@final
-class LLMQuotaLayer(GraphEngineLayer):
-    """Graph layer that applies LLM quota deduction after node execution."""
-
-    def __init__(self) -> None:
-        super().__init__()
-        self._abort_sent = False
-
-    @override
-    def on_graph_start(self) -> None:
-        self._abort_sent = False
-
-    @override
-    def on_event(self, event: GraphEngineEvent) -> None:
-        _ = event
-
-    @override
-    def on_graph_end(self, error: Exception | None) -> None:
-        _ = error
-
-    @override
-    def on_node_run_start(self, node: Node) -> None:
-        if self._abort_sent:
-            return
-
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
-            return
-
-        try:
-            ensure_llm_quota_available(model_instance=model_instance)
-        except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
-            logger.warning("LLM quota check failed, node_id=%s, error=%s", node.id, exc)
-
-    @override
-    def on_node_run_end(
-        self, node: Node, error: Exception | None, result_event: GraphNodeEventBase | None = None
-    ) -> None:
-        if error is not None or not isinstance(result_event, NodeRunSucceededEvent):
-            return
-
-        model_instance = self._extract_model_instance(node)
-        if model_instance is None:
-            return
-
-        try:
-            deduct_llm_quota(
-                tenant_id=node.tenant_id,
-                model_instance=model_instance,
-                usage=result_event.node_run_result.llm_usage,
-            )
-        except QuotaExceededError as exc:
-            self._set_stop_event(node)
-            self._send_abort_command(reason=str(exc))
-            logger.warning("LLM quota deduction exceeded, node_id=%s, error=%s", node.id, exc)
-        except Exception:
-            logger.exception("LLM quota deduction failed, node_id=%s", node.id)
-
-    @staticmethod
-    def _set_stop_event(node: Node) -> None:
-        stop_event = getattr(node.graph_runtime_state, "stop_event", None)
-        if stop_event is not None:
-            stop_event.set()
-
-    def _send_abort_command(self, *, reason: str) -> None:
-        if not self.command_channel or self._abort_sent:
-            return
-
-        try:
-            self.command_channel.send_command(
-                AbortCommand(
-                    command_type=CommandType.ABORT,
-                    reason=reason,
-                )
-            )
-            self._abort_sent = True
-        except Exception:
-            logger.exception("Failed to send quota abort command")
-
-    @staticmethod
-    def _extract_model_instance(node: Node) -> ModelInstance | None:
-        try:
-            match node.node_type:
-                case NodeType.LLM:
-                    return cast("LLMNode", node).model_instance
-                case NodeType.PARAMETER_EXTRACTOR:
-                    return cast("ParameterExtractorNode", node).model_instance
-                case NodeType.QUESTION_CLASSIFIER:
-                    return cast("QuestionClassifierNode", node).model_instance
-                case _:
-                    return None
-        except AttributeError:
-            logger.warning(
-                "LLMQuotaLayer skipped quota deduction because node does not expose a model instance, node_id=%s",
-                node.id,
-            )
-            return None

api/core/app/workflow/node_factory.py

@@ -1,28 +1,18 @@
 from collections.abc import Mapping
-from typing import TYPE_CHECKING, Any, cast, final
+from typing import TYPE_CHECKING, Any, final
 
-from sqlalchemy import select
-from sqlalchemy.orm import Session
 from typing_extensions import override
 
 from configs import dify_config
 from core.app.llm.model_access import build_dify_model_access
 from core.datasource.datasource_manager import DatasourceManager
-from core.helper.code_executor.code_executor import (
-    CodeExecutionError,
-    CodeExecutor,
-)
+from core.helper.code_executor.code_executor import CodeExecutionError, CodeExecutor
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.ssrf_proxy import ssrf_proxy
-from core.memory.token_buffer_memory import TokenBufferMemory
-from core.model_manager import ModelInstance
-from core.model_runtime.entities.model_entities import ModelType
-from core.model_runtime.memory import PromptMessageMemory
-from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
-from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 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, SystemVariableKey
+from core.workflow.enums import NodeType
 from core.workflow.file.file_manager import file_manager
 from core.workflow.graph.graph import NodeFactory
 from core.workflow.nodes.base.node import Node
@@ -33,8 +23,6 @@ from core.workflow.nodes.datasource import DatasourceNode
 from core.workflow.nodes.document_extractor import DocumentExtractorNode, UnstructuredApiConfig
 from core.workflow.nodes.http_request import HttpRequestNode, build_http_request_config
 from core.workflow.nodes.knowledge_retrieval.knowledge_retrieval_node import KnowledgeRetrievalNode
-from core.workflow.nodes.llm.entities import ModelConfig
-from core.workflow.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
 from core.workflow.nodes.llm.node import LLMNode
 from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
 from core.workflow.nodes.parameter_extractor.parameter_extractor_node import ParameterExtractorNode
@@ -43,34 +31,12 @@ from core.workflow.nodes.template_transform.template_renderer import (
     CodeExecutorJinja2TemplateRenderer,
 )
 from core.workflow.nodes.template_transform.template_transform_node import TemplateTransformNode
-from core.workflow.variables.segments import StringSegment
-from extensions.ext_database import db
-from models.model import Conversation
 
 if TYPE_CHECKING:
     from core.workflow.entities import GraphInitParams
     from core.workflow.runtime import GraphRuntimeState
 
 
-def fetch_memory(
-    *,
-    conversation_id: str | None,
-    app_id: str,
-    node_data_memory: MemoryConfig | None,
-    model_instance: ModelInstance,
-) -> TokenBufferMemory | None:
-    if not node_data_memory or not conversation_id:
-        return None
-
-    with Session(db.engine, expire_on_commit=False) as session:
-        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
-        conversation = session.scalar(stmt)
-        if not conversation:
-            return None
-
-    return TokenBufferMemory(conversation=conversation, model_instance=model_instance)
-
-
 class DefaultWorkflowCodeExecutor:
     def execute(
         self,
@@ -106,6 +72,7 @@ class DifyNodeFactory(NodeFactory):
         self.graph_init_params = graph_init_params
         self.graph_runtime_state = graph_runtime_state
         self._code_executor: WorkflowCodeExecutor = DefaultWorkflowCodeExecutor()
+        self._code_providers: tuple[type[CodeNodeProvider], ...] = CodeNode.default_code_providers()
         self._code_limits = CodeNodeLimits(
             max_string_length=dify_config.CODE_MAX_STRING_LENGTH,
             max_number=dify_config.CODE_MAX_NUMBER,
@@ -177,6 +144,7 @@ class DifyNodeFactory(NodeFactory):
                 graph_init_params=self.graph_init_params,
                 graph_runtime_state=self.graph_runtime_state,
                 code_executor=self._code_executor,
+                code_providers=self._code_providers,
                 code_limits=self._code_limits,
             )
 
@@ -203,8 +171,6 @@ class DifyNodeFactory(NodeFactory):
             )
 
         if node_type == NodeType.LLM:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
             return LLMNode(
                 id=node_id,
                 config=node_config,
@@ -212,8 +178,6 @@ class DifyNodeFactory(NodeFactory):
                 graph_runtime_state=self.graph_runtime_state,
                 credentials_provider=self._llm_credentials_provider,
                 model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-                memory=memory,
             )
 
         if node_type == NodeType.DATASOURCE:
@@ -244,8 +208,6 @@ class DifyNodeFactory(NodeFactory):
             )
 
         if node_type == NodeType.QUESTION_CLASSIFIER:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
             return QuestionClassifierNode(
                 id=node_id,
                 config=node_config,
@@ -253,13 +215,9 @@ class DifyNodeFactory(NodeFactory):
                 graph_runtime_state=self.graph_runtime_state,
                 credentials_provider=self._llm_credentials_provider,
                 model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-                memory=memory,
             )
 
         if node_type == NodeType.PARAMETER_EXTRACTOR:
-            model_instance = self._build_model_instance_for_llm_node(node_data)
-            memory = self._build_memory_for_llm_node(node_data=node_data, model_instance=model_instance)
             return ParameterExtractorNode(
                 id=node_id,
                 config=node_config,
@@ -267,8 +225,6 @@ class DifyNodeFactory(NodeFactory):
                 graph_runtime_state=self.graph_runtime_state,
                 credentials_provider=self._llm_credentials_provider,
                 model_factory=self._llm_model_factory,
-                model_instance=model_instance,
-                memory=memory,
             )
 
         return node_class(
@@ -277,61 +233,3 @@ class DifyNodeFactory(NodeFactory):
             graph_init_params=self.graph_init_params,
             graph_runtime_state=self.graph_runtime_state,
         )
-
-    def _build_model_instance_for_llm_node(self, node_data: Mapping[str, Any]) -> ModelInstance:
-        node_data_model = ModelConfig.model_validate(node_data["model"])
-        if not node_data_model.mode:
-            raise LLMModeRequiredError("LLM mode is required.")
-
-        credentials = self._llm_credentials_provider.fetch(node_data_model.provider, node_data_model.name)
-        model_instance = self._llm_model_factory.init_model_instance(node_data_model.provider, node_data_model.name)
-        provider_model_bundle = model_instance.provider_model_bundle
-
-        provider_model = provider_model_bundle.configuration.get_provider_model(
-            model=node_data_model.name,
-            model_type=ModelType.LLM,
-        )
-        if provider_model is None:
-            raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-        provider_model.raise_for_status()
-
-        completion_params = dict(node_data_model.completion_params)
-        stop = completion_params.pop("stop", [])
-        if not isinstance(stop, list):
-            stop = []
-
-        model_schema = model_instance.model_type_instance.get_model_schema(node_data_model.name, credentials)
-        if not model_schema:
-            raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
-
-        model_instance.provider = node_data_model.provider
-        model_instance.model_name = node_data_model.name
-        model_instance.credentials = credentials
-        model_instance.parameters = completion_params
-        model_instance.stop = tuple(stop)
-        model_instance.model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
-        return model_instance
-
-    def _build_memory_for_llm_node(
-        self,
-        *,
-        node_data: Mapping[str, Any],
-        model_instance: ModelInstance,
-    ) -> PromptMessageMemory | None:
-        raw_memory_config = node_data.get("memory")
-        if raw_memory_config is None:
-            return None
-
-        node_memory = MemoryConfig.model_validate(raw_memory_config)
-        conversation_id_variable = self.graph_runtime_state.variable_pool.get(
-            ["sys", SystemVariableKey.CONVERSATION_ID]
-        )
-        conversation_id = (
-            conversation_id_variable.value if isinstance(conversation_id_variable, StringSegment) else None
-        )
-        return fetch_memory(
-            conversation_id=conversation_id,
-            app_id=self.graph_init_params.app_id,
-            node_data_memory=node_memory,
-            model_instance=model_instance,
-        )

api/core/helper/code_executor/code_executor.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Mapping
+from enum import StrEnum
 from threading import Lock
 from typing import Any
 
@@ -13,7 +14,6 @@ from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTr
 from core.helper.code_executor.python3.python3_transformer import Python3TemplateTransformer
 from core.helper.code_executor.template_transformer import TemplateTransformer
 from core.helper.http_client_pooling import get_pooled_http_client
-from core.workflow.nodes.code.entities import CodeLanguage
 
 logger = logging.getLogger(__name__)
 code_execution_endpoint_url = URL(str(dify_config.CODE_EXECUTION_ENDPOINT))
@@ -40,6 +40,12 @@ class CodeExecutionResponse(BaseModel):
     data: Data
 
 
+class CodeLanguage(StrEnum):
+    PYTHON3 = "python3"
+    JINJA2 = "jinja2"
+    JAVASCRIPT = "javascript"
+
+
 def _build_code_executor_client() -> httpx.Client:
     return httpx.Client(
         verify=CODE_EXECUTION_SSL_VERIFY,

api/core/helper/code_executor/template_transformer.py

@@ -5,7 +5,7 @@ from base64 import b64encode
 from collections.abc import Mapping
 from typing import Any
 
-from core.workflow.variables.utils import dumps_with_segments
+from core.variables.utils import dumps_with_segments
 
 
 class TemplateTransformer(ABC):

api/core/mcp/server/streamable_http.py

@@ -4,10 +4,10 @@ from collections.abc import Mapping
 from typing import Any, cast
 
 from configs import dify_config
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
 from core.mcp import types as mcp_types
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from models.model import App, AppMCPServer, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 

api/core/model_manager.py

@@ -1,5 +1,5 @@
 import logging
-from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
+from collections.abc import Callable, Generator, Iterable, Sequence
 from typing import IO, Any, Literal, Optional, Union, cast, overload
 
 from configs import dify_config
@@ -38,9 +38,6 @@ class ModelInstance:
         self.model_name = model
         self.provider = provider_model_bundle.configuration.provider.provider
         self.credentials = self._fetch_credentials_from_bundle(provider_model_bundle, model)
-        # Runtime LLM invocation fields.
-        self.parameters: Mapping[str, Any] = {}
-        self.stop: Sequence[str] = ()
         self.model_type_instance = self.provider_model_bundle.model_type_instance
         self.load_balancing_manager = self._get_load_balancing_manager(
             configuration=provider_model_bundle.configuration,

api/core/model_runtime/memory/__init__.py

@@ -1,3 +0,0 @@
-from .prompt_message_memory import DEFAULT_MEMORY_MAX_TOKEN_LIMIT, PromptMessageMemory
-
-__all__ = ["DEFAULT_MEMORY_MAX_TOKEN_LIMIT", "PromptMessageMemory"]

api/core/model_runtime/memory/prompt_message_memory.py

@@ -1,18 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Sequence
-from typing import Protocol
-
-from core.model_runtime.entities import PromptMessage
-
-DEFAULT_MEMORY_MAX_TOKEN_LIMIT = 2000
-
-
-class PromptMessageMemory(Protocol):
-    """Port for loading memory as prompt messages."""
-
-    def get_history_prompt_messages(
-        self, max_token_limit: int = DEFAULT_MEMORY_MAX_TOKEN_LIMIT, message_limit: int | None = None
-    ) -> Sequence[PromptMessage]:
-        """Return historical prompt messages constrained by token/message limits."""
-        ...

api/core/model_runtime/model_providers/__base/large_language_model.py

@@ -83,21 +83,19 @@ def _merge_tool_call_delta(
         tool_call.function.arguments += delta.function.arguments
 
 
-def _build_llm_result_from_chunks(
+def _build_llm_result_from_first_chunk(
     model: str,
     prompt_messages: Sequence[PromptMessage],
     chunks: Iterator[LLMResultChunk],
 ) -> LLMResult:
     """
-    Build a single `LLMResult` by accumulating all returned chunks.
+    Build a single `LLMResult` from the first returned chunk.
 
-    Some models only support streaming output (e.g. Qwen3 open-source edition)
-    and the plugin side may still implement the response via a chunked stream,
-    so all chunks must be consumed and concatenated into a single ``LLMResult``.
+    This is used for `stream=False` because the plugin side may still implement the response via a chunked stream.
 
-    The ``usage`` is taken from the last chunk that carries it, which is the
-    typical convention for streaming responses (the final chunk contains the
-    aggregated token counts).
+    Note:
+        This function always drains the `chunks` iterator after reading the first chunk to ensure any underlying
+        streaming resources are released (e.g., HTTP connections owned by the plugin runtime).
     """
     content = ""
     content_list: list[PromptMessageContentUnionTypes] = []
@@ -106,27 +104,24 @@ def _build_llm_result_from_chunks(
     tools_calls: list[AssistantPromptMessage.ToolCall] = []
 
     try:
-        for chunk in chunks:
-            if isinstance(chunk.delta.message.content, str):
-                content += chunk.delta.message.content
-            elif isinstance(chunk.delta.message.content, list):
-                content_list.extend(chunk.delta.message.content)
+        first_chunk = next(chunks, None)
+        if first_chunk is not None:
+            if isinstance(first_chunk.delta.message.content, str):
+                content += first_chunk.delta.message.content
+            elif isinstance(first_chunk.delta.message.content, list):
+                content_list.extend(first_chunk.delta.message.content)
 
-            if chunk.delta.message.tool_calls:
-                _increase_tool_call(chunk.delta.message.tool_calls, tools_calls)
+            if first_chunk.delta.message.tool_calls:
+                _increase_tool_call(first_chunk.delta.message.tool_calls, tools_calls)
 
-            if chunk.delta.usage:
-                usage = chunk.delta.usage
-            if chunk.system_fingerprint:
-                system_fingerprint = chunk.system_fingerprint
-    except Exception:
-        logger.exception("Error while consuming non-stream plugin chunk iterator.")
-        raise
+            usage = first_chunk.delta.usage or LLMUsage.empty_usage()
+            system_fingerprint = first_chunk.system_fingerprint
     finally:
-        # Drain any remaining chunks to release underlying streaming resources (e.g. HTTP connections).
-        close = getattr(chunks, "close", None)
-        if callable(close):
-            close()
+        try:
+            for _ in chunks:
+                pass
+        except Exception:
+            logger.debug("Failed to drain non-stream plugin chunk iterator.", exc_info=True)
 
     return LLMResult(
         model=model,
@@ -179,7 +174,7 @@ def _normalize_non_stream_plugin_result(
 ) -> LLMResult:
     if isinstance(result, LLMResult):
         return result
-    return _build_llm_result_from_chunks(model=model, prompt_messages=prompt_messages, chunks=result)
+    return _build_llm_result_from_first_chunk(model=model, prompt_messages=prompt_messages, chunks=result)
 
 
 def _increase_tool_call(

api/core/ops/base_trace_instance.py

@@ -14,9 +14,10 @@ class BaseTraceInstance(ABC):
     Base trace instance for ops trace services
     """
 
+    @abstractmethod
     def __init__(self, trace_config: BaseTracingConfig):
         """
-        Initializer for the trace instance.
+        Abstract initializer for the trace instance.
         Distribute trace tasks by matching entities
         """
         self.trace_config = trace_config

api/core/ops/ops_trace_manager.py

@@ -41,8 +41,8 @@ logger = logging.getLogger(__name__)
 
 
 class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
-    def __getitem__(self, key: str) -> dict[str, Any]:
-        match key:
+    def __getitem__(self, provider: str) -> dict[str, Any]:
+        match provider:
             case TracingProviderEnum.LANGFUSE:
                 from core.ops.entities.config_entity import LangfuseConfig
                 from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
@@ -149,7 +149,7 @@ class OpsTraceProviderConfigMap(collections.UserDict[str, dict[str, Any]]):
                 }
 
             case _:
-                raise KeyError(f"Unsupported tracing provider: {key}")
+                raise KeyError(f"Unsupported tracing provider: {provider}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()

api/core/plugin/backwards_invocation/model.py

@@ -2,7 +2,6 @@ import tempfile
 from binascii import hexlify, unhexlify
 from collections.abc import Generator
 
-from core.app.llm import deduct_llm_quota
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import (
@@ -30,6 +29,7 @@ from core.plugin.entities.request import (
 )
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.utils.model_invocation_utils import ModelInvocationUtils
+from core.workflow.nodes.llm import llm_utils
 from models.account import Tenant
 
 
@@ -63,14 +63,16 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
             def handle() -> Generator[LLMResultChunk, None, None]:
                 for chunk in response:
                     if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
+                        llm_utils.deduct_llm_quota(
+                            tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage
+                        )
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
             if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
+                llm_utils.deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(response: LLMResult) -> Generator[LLMResultChunk, None, None]:
                 yield LLMResultChunk(
@@ -124,14 +126,16 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
             def handle() -> Generator[LLMResultChunkWithStructuredOutput, None, None]:
                 for chunk in response:
                     if chunk.delta.usage:
-                        deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage)
+                        llm_utils.deduct_llm_quota(
+                            tenant_id=tenant.id, model_instance=model_instance, usage=chunk.delta.usage
+                        )
                     chunk.prompt_messages = []
                     yield chunk
 
             return handle()
         else:
             if response.usage:
-                deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
+                llm_utils.deduct_llm_quota(tenant_id=tenant.id, model_instance=model_instance, usage=response.usage)
 
             def handle_non_streaming(
                 response: LLMResultWithStructuredOutput,

api/core/prompt/advanced_prompt_transform.py

@@ -3,8 +3,6 @@ from typing import cast
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.helper.code_executor.jinja2.jinja2_formatter import Jinja2Formatter
-from core.memory.token_buffer_memory import TokenBufferMemory
-from core.model_manager import ModelInstance
 from core.model_runtime.entities import (
     AssistantPromptMessage,
     PromptMessage,
@@ -14,6 +12,7 @@ from core.model_runtime.entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.prompt_transform import PromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
@@ -45,8 +44,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         prompt_messages = []
@@ -61,7 +59,6 @@ class AdvancedPromptTransform(PromptTransform):
                 memory_config=memory_config,
                 memory=memory,
                 model_config=model_config,
-                model_instance=model_instance,
                 image_detail_config=image_detail_config,
             )
         elif isinstance(prompt_template, list) and all(isinstance(item, ChatModelMessage) for item in prompt_template):
@@ -74,7 +71,6 @@ class AdvancedPromptTransform(PromptTransform):
                 memory_config=memory_config,
                 memory=memory,
                 model_config=model_config,
-                model_instance=model_instance,
                 image_detail_config=image_detail_config,
             )
 
@@ -89,8 +85,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         """
@@ -116,7 +111,6 @@ class AdvancedPromptTransform(PromptTransform):
                     parser=parser,
                     prompt_inputs=prompt_inputs,
                     model_config=model_config,
-                    model_instance=model_instance,
                 )
 
             if query:
@@ -152,8 +146,7 @@ class AdvancedPromptTransform(PromptTransform):
         context: str | None,
         memory_config: MemoryConfig | None,
         memory: TokenBufferMemory | None,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
         image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
         """
@@ -205,13 +198,8 @@ class AdvancedPromptTransform(PromptTransform):
 
         prompt_message_contents: list[PromptMessageContentUnionTypes] = []
         if memory and memory_config:
-            prompt_messages = self._append_chat_histories(
-                memory,
-                memory_config,
-                prompt_messages,
-                model_config=model_config,
-                model_instance=model_instance,
-            )
+            prompt_messages = self._append_chat_histories(memory, memory_config, prompt_messages, model_config)
+
             if files and query is not None:
                 for file in files:
                     prompt_message_contents.append(
@@ -288,8 +276,7 @@ class AdvancedPromptTransform(PromptTransform):
         role_prefix: MemoryConfig.RolePrefix,
         parser: PromptTemplateParser,
         prompt_inputs: Mapping[str, str],
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
     ) -> Mapping[str, str]:
         prompt_inputs = dict(prompt_inputs)
         if "#histories#" in parser.variable_keys:
@@ -299,11 +286,7 @@ class AdvancedPromptTransform(PromptTransform):
                 prompt_inputs = {k: inputs[k] for k in parser.variable_keys if k in inputs}
                 tmp_human_message = UserPromptMessage(content=parser.format(prompt_inputs))
 
-                rest_tokens = self._calculate_rest_token(
-                    [tmp_human_message],
-                    model_config=model_config,
-                    model_instance=model_instance,
-                )
+                rest_tokens = self._calculate_rest_token([tmp_human_message], model_config)
 
                 histories = self._get_history_messages_from_memory(
                     memory=memory,

api/core/prompt/agent_history_prompt_transform.py

@@ -3,13 +3,13 @@ from typing import cast
 from core.app.entities.app_invoke_entities import (
     ModelConfigWithCredentialsEntity,
 )
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.entities.message_entities import (
     PromptMessage,
     SystemPromptMessage,
     UserPromptMessage,
 )
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.prompt.prompt_transform import PromptTransform
 
 
@@ -41,7 +41,7 @@ class AgentHistoryPromptTransform(PromptTransform):
         if not self.memory:
             return prompt_messages
 
-        max_token_limit = self._calculate_rest_token(self.prompt_messages, model_config=self.model_config)
+        max_token_limit = self._calculate_rest_token(self.prompt_messages, self.model_config)
 
         model_type_instance = self.model_config.provider_model_bundle.model_type_instance
         model_type_instance = cast(LargeLanguageModel, model_type_instance)

api/core/prompt/prompt_transform.py

@@ -1,86 +1,48 @@
 from typing import Any
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.message_entities import PromptMessage
-from core.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey
+from core.model_runtime.entities.model_entities import ModelPropertyKey
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 
 
 class PromptTransform:
-    def _resolve_model_runtime(
-        self,
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
-    ) -> tuple[ModelInstance, AIModelEntity]:
-        if model_instance is None:
-            if model_config is None:
-                raise ValueError("Either model_config or model_instance must be provided.")
-            model_instance = ModelInstance(
-                provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
-            )
-            model_instance.credentials = model_config.credentials
-            model_instance.parameters = model_config.parameters
-            model_instance.stop = model_config.stop
-
-        model_schema = model_instance.model_type_instance.get_model_schema(
-            model=model_instance.model_name,
-            credentials=model_instance.credentials,
-        )
-        if model_schema is None:
-            if model_config is None:
-                raise ValueError("Model schema not found for the provided model instance.")
-            model_schema = model_config.model_schema
-
-        return model_instance, model_schema
-
     def _append_chat_histories(
         self,
         memory: TokenBufferMemory,
         memory_config: MemoryConfig,
         prompt_messages: list[PromptMessage],
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        model_config: ModelConfigWithCredentialsEntity,
     ) -> list[PromptMessage]:
-        rest_tokens = self._calculate_rest_token(
-            prompt_messages,
-            model_config=model_config,
-            model_instance=model_instance,
-        )
+        rest_tokens = self._calculate_rest_token(prompt_messages, model_config)
         histories = self._get_history_messages_list_from_memory(memory, memory_config, rest_tokens)
         prompt_messages.extend(histories)
 
         return prompt_messages
 
     def _calculate_rest_token(
-        self,
-        prompt_messages: list[PromptMessage],
-        *,
-        model_config: ModelConfigWithCredentialsEntity | None = None,
-        model_instance: ModelInstance | None = None,
+        self, prompt_messages: list[PromptMessage], model_config: ModelConfigWithCredentialsEntity
     ) -> int:
-        model_instance, model_schema = self._resolve_model_runtime(
-            model_config=model_config,
-            model_instance=model_instance,
-        )
-        model_parameters = model_instance.parameters
         rest_tokens = 2000
 
-        model_context_tokens = model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
         if model_context_tokens:
+            model_instance = ModelInstance(
+                provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
+            )
+
             curr_message_tokens = model_instance.get_llm_num_tokens(prompt_messages)
 
             max_tokens = 0
-            for parameter_rule in model_schema.parameter_rules:
+            for parameter_rule in model_config.model_schema.parameter_rules:
                 if parameter_rule.name == "max_tokens" or (
                     parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
                 ):
                     max_tokens = (
-                        model_parameters.get(parameter_rule.name)
-                        or model_parameters.get(parameter_rule.use_template or "")
+                        model_config.parameters.get(parameter_rule.name)
+                        or model_config.parameters.get(parameter_rule.use_template or "")
                     ) or 0
 
             rest_tokens = model_context_tokens - max_tokens - curr_message_tokens

api/core/prompt/simple_prompt_transform.py

@@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
 
 from core.app.app_config.entities import PromptTemplateEntity
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.entities.message_entities import (
     ImagePromptMessageContent,
     PromptMessage,
@@ -15,6 +14,7 @@ from core.model_runtime.entities.message_entities import (
     TextPromptMessageContent,
     UserPromptMessage,
 )
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.prompt.prompt_transform import PromptTransform
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
@@ -252,7 +252,7 @@ class SimplePromptTransform(PromptTransform):
         if memory:
             tmp_human_message = UserPromptMessage(content=prompt)
 
-            rest_tokens = self._calculate_rest_token([tmp_human_message], model_config=model_config)
+            rest_tokens = self._calculate_rest_token([tmp_human_message], model_config)
             histories = self._get_history_messages_from_memory(
                 memory=memory,
                 memory_config=MemoryConfig(

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py

@@ -192,8 +192,8 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             metrics=self.config.metrics,
             include_values=True,
-            vector=None,
-            content=None,
+            vector=None,  # ty: ignore [invalid-argument-type]
+            content=None,  # ty: ignore [invalid-argument-type]
             top_k=1,
             filter=f"ref_doc_id='{id}'",
         )
@@ -211,7 +211,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,
+            collection_data=None,  # ty: ignore [invalid-argument-type]
             collection_data_filter=f"ref_doc_id IN {ids_str}",
         )
         self._client.delete_collection_data(request)
@@ -225,7 +225,7 @@ class AnalyticdbVectorOpenAPI:
             namespace=self.config.namespace,
             namespace_password=self.config.namespace_password,
             collection=self._collection_name,
-            collection_data=None,
+            collection_data=None,  # ty: ignore [invalid-argument-type]
             collection_data_filter=f"metadata_ ->> '{key}' = '{value}'",
         )
         self._client.delete_collection_data(request)
@@ -249,7 +249,7 @@ class AnalyticdbVectorOpenAPI:
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
             vector=query_vector,
-            content=None,
+            content=None,  # ty: ignore [invalid-argument-type]
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,
         )
@@ -285,7 +285,7 @@ class AnalyticdbVectorOpenAPI:
             collection=self._collection_name,
             include_values=kwargs.pop("include_values", True),
             metrics=self.config.metrics,
-            vector=None,
+            vector=None,  # ty: ignore [invalid-argument-type]
             content=query,
             top_k=kwargs.get("top_k", 4),
             filter=where_clause,

api/core/rag/datasource/vdb/couchbase/couchbase_vector.py

@@ -306,7 +306,7 @@ class CouchbaseVector(BaseVector):
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         top_k = kwargs.get("top_k", 4)
         try:
-            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))
+            CBrequest = search.SearchRequest.create(search.QueryStringQuery("text:" + query))  # ty: ignore [too-many-positional-arguments]
             search_iter = self._scope.search(
                 self._collection_name + "_search", CBrequest, SearchOptions(limit=top_k, fields=["*"])
             )

api/core/rag/index_processor/index_processor_base.py

@@ -75,15 +75,15 @@ class BaseIndexProcessor(ABC):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         raise NotImplementedError
 
     @abstractmethod
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         raise NotImplementedError
 
     @abstractmethod
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         raise NotImplementedError
 
     @abstractmethod

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

@@ -8,7 +8,6 @@ from typing import Any, cast
 
 logger = logging.getLogger(__name__)
 
-from core.app.llm import deduct_llm_quota
 from core.entities.knowledge_entities import PreviewDetail
 from core.llm_generator.prompts import DEFAULT_GENERATOR_SUMMARY_PROMPT
 from core.model_manager import ModelInstance
@@ -36,6 +35,7 @@ from core.rag.models.document import AttachmentDocument, Document, MultimodalGen
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
 from core.workflow.file import File, FileTransferMethod, FileType, file_manager
+from core.workflow.nodes.llm import llm_utils
 from extensions.ext_database import db
 from factories.file_factory import build_from_mapping
 from libs import helper
@@ -115,7 +115,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
@@ -130,7 +130,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             else:
                 keyword.add_texts(documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -196,7 +196,7 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         documents: list[Any] = []
         all_multimodal_documents: list[Any] = []
         if isinstance(chunks, list):
@@ -469,12 +469,12 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         if not isinstance(result, LLMResult):
             raise ValueError("Expected LLMResult when stream=False")
 
-        summary_content = result.message.get_text_content()
+        summary_content = getattr(result.message, "content", "")
         usage = result.usage
 
         # Deduct quota for summary generation (same as workflow nodes)
         try:
-            deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
+            llm_utils.deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
         except Exception as e:
             # Log but don't fail summary generation if quota deduction fails
             logger.warning("Failed to deduct quota for summary generation: %s", str(e))

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

@@ -126,7 +126,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             for document in documents:
@@ -139,7 +139,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # node_ids is segment's node_ids
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
@@ -272,7 +272,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
                     child_nodes.append(child_document)
         return child_nodes
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         parent_childs = ParentChildStructureChunk.model_validate(chunks)
         documents = []
         for parent_child in parent_childs.parent_child_chunks:

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

@@ -139,14 +139,14 @@ class QAIndexProcessor(BaseIndexProcessor):
         multimodal_documents: list[AttachmentDocument] | None = None,
         with_keywords: bool = True,
         **kwargs,
-    ) -> None:
+    ):
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             vector.create(documents)
             if multimodal_documents and dataset.is_multimodal:
                 vector.create_multimodal(multimodal_documents)
 
-    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs) -> None:
+    def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
         # This method is called for actual deletion scenarios (e.g., when segment is deleted).
         # For disable operations, disable_summaries_for_segments is called directly in the task.
@@ -206,7 +206,7 @@ class QAIndexProcessor(BaseIndexProcessor):
                 docs.append(doc)
         return docs
 
-    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any) -> None:
+    def index(self, dataset: Dataset, document: DatasetDocument, chunks: Any):
         qa_chunks = QAStructureChunk.model_validate(chunks)
         documents = []
         for qa_chunk in qa_chunks.qa_chunks:

api/core/rag/retrieval/dataset_retrieval.py

@@ -23,12 +23,12 @@ from core.callback_handler.index_tool_callback_handler import DatasetIndexToolCa
 from core.db.session_factory import session_factory
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 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.token_buffer_memory import TokenBufferMemory
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time

api/core/rag/retrieval/router/multi_dataset_react_route.py

@@ -2,7 +2,6 @@ from collections.abc import Generator, Sequence
 from typing import Union
 
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.app.llm import deduct_llm_quota
 from core.model_manager import ModelInstance
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
@@ -10,6 +9,7 @@ from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
 from core.rag.retrieval.output_parser.react_output import ReactAction
 from core.rag.retrieval.output_parser.structured_chat import StructuredChatOutputParser
+from core.workflow.nodes.llm import llm_utils
 
 PREFIX = """Respond to the human as helpfully and accurately as possible. You have access to the following tools:"""
 
@@ -162,7 +162,7 @@ class ReactMultiDatasetRouter:
         text, usage = self._handle_invoke_result(invoke_result=invoke_result)
 
         # deduct quota
-        deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
+        llm_utils.deduct_llm_quota(tenant_id=tenant_id, model_instance=model_instance, usage=usage)
 
         return text, usage
 

api/core/tools/builtin_tool/providers/webscraper/webscraper.yaml

@@ -6,9 +6,9 @@ identity:
     zh_Hans: 网页抓取
     pt_BR: WebScraper
   description:
-    en_US: Web Scraper tool kit is used to scrape web
+    en_US: Web Scrapper tool kit is used to scrape web
     zh_Hans: 一个用于抓取网页的工具。
-    pt_BR: Web Scraper tool kit is used to scrape web
+    pt_BR: Web Scrapper tool kit is used to scrape web
   icon: icon.svg
   tags:
     - productivity

api/core/tools/utils/workflow_configuration_sync.py

@@ -1,11 +1,11 @@
 from collections.abc import Mapping, Sequence
 from typing import Any
 
+from core.app.app_config.entities import VariableEntity
 from core.tools.entities.tool_entities import WorkflowToolParameterConfiguration
 from core.tools.errors import WorkflowToolHumanInputNotSupportedError
 from core.workflow.enums import NodeType
 from core.workflow.nodes.base.entities import OutputVariableEntity
-from core.workflow.variables.input_entities import VariableEntity
 
 
 class WorkflowToolConfigurationUtils:

api/core/tools/workflow_as_tool/provider.py

@@ -5,6 +5,7 @@ from collections.abc import Mapping
 from pydantic import Field
 from sqlalchemy.orm import Session
 
+from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.db.session_factory import session_factory
 from core.plugin.entities.parameters import PluginParameterOption
@@ -22,7 +23,6 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.utils.workflow_configuration_sync import WorkflowToolConfigurationUtils
 from core.tools.workflow_as_tool.tool import WorkflowTool
-from core.workflow.variables.input_entities import VariableEntity, VariableEntityType
 from extensions.ext_database import db
 from models.account import Account
 from models.model import App, AppMode

api/core/variables/__init__.py

@@ -1,4 +1,3 @@
-from .input_entities import VariableEntity, VariableEntityType
 from .segment_group import SegmentGroup
 from .segments import (
     ArrayAnySegment,
@@ -65,6 +64,4 @@ __all__ = [
     "StringVariable",
     "Variable",
     "VariableBase",
-    "VariableEntity",
-    "VariableEntityType",
 ]

api/core/variables/variables.py

@@ -4,6 +4,8 @@ from uuid import uuid4
 
 from pydantic import BaseModel, Discriminator, Field, Tag
 
+from core.helper import encrypter
+
 from .segments import (
     ArrayAnySegment,
     ArrayBooleanSegment,
@@ -25,14 +27,6 @@ from .segments import (
 from .types import SegmentType
 
 
-def _obfuscated_token(token: str) -> str:
-    if not token:
-        return token
-    if len(token) <= 8:
-        return "*" * 20
-    return token[:6] + "*" * 12 + token[-2:]
-
-
 class VariableBase(Segment):
     """
     A variable is a segment that has a name.
@@ -92,7 +86,7 @@ class SecretVariable(StringVariable):
 
     @property
     def log(self) -> str:
-        return _obfuscated_token(self.value)
+        return encrypter.obfuscated_token(self.value)
 
 
 class NoneVariable(NoneSegment, VariableBase):

api/core/workflow/conversation_variable_updater.py

@@ -1,7 +1,7 @@
 import abc
 from typing import Protocol
 
-from core.workflow.variables import VariableBase
+from core.variables import VariableBase
 
 
 class ConversationVariableUpdater(Protocol):

api/core/workflow/graph_engine/entities/commands.py

@@ -11,7 +11,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
-from core.workflow.variables.variables import Variable
+from core.variables.variables import Variable
 
 
 class CommandType(StrEnum):

api/core/workflow/graph_engine/graph_engine.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import logging
 import queue
+import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, cast, final
 
@@ -76,10 +77,13 @@ class GraphEngine:
         config: GraphEngineConfig = _DEFAULT_CONFIG,
     ) -> None:
         """Initialize the graph engine with all subsystems and dependencies."""
+        # stop event
+        self._stop_event = threading.Event()
 
         # Bind runtime state to current workflow context
         self._graph = graph
         self._graph_runtime_state = graph_runtime_state
+        self._graph_runtime_state.stop_event = self._stop_event
         self._graph_runtime_state.configure(graph=cast("GraphProtocol", graph))
         self._command_channel = command_channel
         self._config = config
@@ -159,6 +163,7 @@ class GraphEngine:
             layers=self._layers,
             execution_context=execution_context,
             config=self._config,
+            stop_event=self._stop_event,
         )
 
         # === Orchestration ===
@@ -189,6 +194,7 @@ class GraphEngine:
             event_handler=self._event_handler_registry,
             execution_coordinator=self._execution_coordinator,
             event_emitter=self._event_manager,
+            stop_event=self._stop_event,
         )
 
         # === Validation ===
@@ -308,6 +314,7 @@ class GraphEngine:
 
     def _start_execution(self, *, resume: bool = False) -> None:
         """Start execution subsystems."""
+        self._stop_event.clear()
         paused_nodes: list[str] = []
         deferred_nodes: list[str] = []
         if resume:
@@ -341,6 +348,7 @@ class GraphEngine:
 
     def _stop_execution(self) -> None:
         """Stop execution subsystems."""
+        self._stop_event.set()
         self._dispatcher.stop()
         self._worker_pool.stop()
         # Don't mark complete here as the dispatcher already does it

api/core/workflow/graph_engine/orchestration/dispatcher.py

@@ -44,6 +44,7 @@ class Dispatcher:
         event_queue: queue.Queue[GraphNodeEventBase],
         event_handler: "EventHandler",
         execution_coordinator: ExecutionCoordinator,
+        stop_event: threading.Event,
         event_emitter: EventManager | None = None,
     ) -> None:
         """
@@ -61,7 +62,7 @@ class Dispatcher:
         self._event_emitter = event_emitter
 
         self._thread: threading.Thread | None = None
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._start_time: float | None = None
 
     def start(self) -> None:
@@ -69,14 +70,12 @@ class Dispatcher:
         if self._thread and self._thread.is_alive():
             return
 
-        self._stop_event.clear()
         self._start_time = time.time()
         self._thread = threading.Thread(target=self._dispatcher_loop, name="GraphDispatcher", daemon=True)
         self._thread.start()
 
     def stop(self) -> None:
         """Stop the dispatcher thread."""
-        self._stop_event.set()
         if self._thread and self._thread.is_alive():
             self._thread.join(timeout=2.0)
 

api/core/workflow/graph_engine/worker.py

@@ -42,6 +42,7 @@ class Worker(threading.Thread):
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: Sequence[GraphEngineLayer],
+        stop_event: threading.Event,
         worker_id: int = 0,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -62,13 +63,16 @@ class Worker(threading.Thread):
         self._graph = graph
         self._worker_id = worker_id
         self._execution_context = execution_context
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._layers = layers if layers is not None else []
         self._last_task_time = time.time()
 
     def stop(self) -> None:
-        """Signal the worker to stop processing."""
-        self._stop_event.set()
+        """Worker is controlled via shared stop_event from GraphEngine.
+
+        This method is a no-op retained for backward compatibility.
+        """
+        pass
 
     @property
     def is_idle(self) -> bool:

api/core/workflow/graph_engine/worker_management/worker_pool.py

@@ -37,6 +37,7 @@ class WorkerPool:
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: list[GraphEngineLayer],
+        stop_event: threading.Event,
         config: GraphEngineConfig,
         execution_context: IExecutionContext | None = None,
     ) -> None:
@@ -63,6 +64,7 @@ class WorkerPool:
         self._worker_counter = 0
         self._lock = threading.RLock()
         self._running = False
+        self._stop_event = stop_event
 
         # No longer tracking worker states with callbacks to avoid lock contention
 
@@ -133,6 +135,7 @@ class WorkerPool:
             layers=self._layers,
             worker_id=worker_id,
             execution_context=self._execution_context,
+            stop_event=self._stop_event,
         )
 
         worker.start()

api/core/workflow/nodes/agent/agent_node.py

@@ -11,10 +11,10 @@ from sqlalchemy.orm import Session
 
 from core.agent.entities import AgentToolEntity
 from core.agent.plugin_entities import AgentStrategyParameter
-from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities.llm_entities import LLMUsage, LLMUsageMetadata
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.provider_manager import ProviderManager
 from core.tools.entities.tool_entities import (
@@ -25,6 +25,7 @@ from core.tools.entities.tool_entities import (
 )
 from core.tools.tool_manager import ToolManager
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
+from core.variables.segments import ArrayFileSegment, StringSegment
 from core.workflow.enums import (
     NodeType,
     SystemVariableKey,
@@ -43,7 +44,6 @@ from core.workflow.nodes.agent.entities import AgentNodeData, AgentOldVersionMod
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.segments import ArrayFileSegment, StringSegment
 from extensions.ext_database import db
 from factories import file_factory
 from factories.agent_factory import get_plugin_agent_strategy

api/core/workflow/nodes/answer/answer_node.py

@@ -1,13 +1,13 @@
 from collections.abc import Mapping, Sequence
 from typing import Any
 
+from core.variables import ArrayFileSegment, FileSegment, Segment
 from core.workflow.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.answer.entities import AnswerNodeData
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
-from core.workflow.variables import ArrayFileSegment, FileSegment, Segment
 
 
 class AnswerNode(Node[AnswerNodeData]):

api/core/workflow/nodes/base/node.py

@@ -302,6 +302,10 @@ class Node(Generic[NodeDataT]):
         """
         raise NotImplementedError
 
+    def _should_stop(self) -> bool:
+        """Check if execution should be stopped."""
+        return self.graph_runtime_state.stop_event.is_set()
+
     def run(self) -> Generator[GraphNodeEventBase, None, None]:
         execution_id = self.ensure_execution_id()
         self._start_at = naive_utc_now()
@@ -370,6 +374,21 @@ class Node(Generic[NodeDataT]):
                     yield event
                 else:
                     yield event
+
+                if self._should_stop():
+                    error_message = "Execution cancelled"
+                    yield NodeRunFailedEvent(
+                        id=self.execution_id,
+                        node_id=self._node_id,
+                        node_type=self.node_type,
+                        start_at=self._start_at,
+                        node_run_result=NodeRunResult(
+                            status=WorkflowNodeExecutionStatus.FAILED,
+                            error=error_message,
+                        ),
+                        error=error_message,
+                    )
+                    return
         except Exception as e:
             logger.exception("Node %s failed to run", self._node_id)
             result = NodeRunResult(

api/core/workflow/nodes/code/code_node.py

@@ -1,15 +1,17 @@
 from collections.abc import Mapping, Sequence
 from decimal import Decimal
-from textwrap import dedent
-from typing import TYPE_CHECKING, Any, Protocol, cast
+from typing import TYPE_CHECKING, Any, ClassVar, Protocol, cast
 
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
+from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
+from core.variables.segments import ArrayFileSegment
+from core.variables.types import SegmentType
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.code.entities import CodeLanguage, CodeNodeData
 from core.workflow.nodes.code.limits import CodeNodeLimits
-from core.workflow.variables.segments import ArrayFileSegment
-from core.workflow.variables.types import SegmentType
 
 from .exc import (
     CodeNodeError,
@@ -34,44 +36,12 @@ class WorkflowCodeExecutor(Protocol):
     def is_execution_error(self, error: Exception) -> bool: ...
 
 
-def _build_default_config(*, language: CodeLanguage, code: str) -> Mapping[str, object]:
-    return {
-        "type": "code",
-        "config": {
-            "variables": [
-                {"variable": "arg1", "value_selector": []},
-                {"variable": "arg2", "value_selector": []},
-            ],
-            "code_language": language,
-            "code": code,
-            "outputs": {"result": {"type": "string", "children": None}},
-        },
-    }
-
-
-_DEFAULT_CODE_BY_LANGUAGE: Mapping[CodeLanguage, str] = {
-    CodeLanguage.PYTHON3: dedent(
-        """
-        def main(arg1: str, arg2: str):
-            return {
-                "result": arg1 + arg2,
-            }
-        """
-    ),
-    CodeLanguage.JAVASCRIPT: dedent(
-        """
-        function main({arg1, arg2}) {
-            return {
-                result: arg1 + arg2
-            }
-        }
-        """
-    ),
-}
-
-
 class CodeNode(Node[CodeNodeData]):
     node_type = NodeType.CODE
+    _DEFAULT_CODE_PROVIDERS: ClassVar[tuple[type[CodeNodeProvider], ...]] = (
+        Python3CodeProvider,
+        JavascriptCodeProvider,
+    )
     _limits: CodeNodeLimits
 
     def __init__(
@@ -82,6 +52,7 @@ class CodeNode(Node[CodeNodeData]):
         graph_runtime_state: "GraphRuntimeState",
         *,
         code_executor: WorkflowCodeExecutor,
+        code_providers: Sequence[type[CodeNodeProvider]] | None = None,
         code_limits: CodeNodeLimits,
     ) -> None:
         super().__init__(
@@ -91,6 +62,9 @@ class CodeNode(Node[CodeNodeData]):
             graph_runtime_state=graph_runtime_state,
         )
         self._code_executor: WorkflowCodeExecutor = code_executor
+        self._code_providers: tuple[type[CodeNodeProvider], ...] = (
+            tuple(code_providers) if code_providers else self._DEFAULT_CODE_PROVIDERS
+        )
         self._limits = code_limits
 
     @classmethod
@@ -104,10 +78,15 @@ class CodeNode(Node[CodeNodeData]):
         if filters:
             code_language = cast(CodeLanguage, filters.get("code_language", CodeLanguage.PYTHON3))
 
-        default_code = _DEFAULT_CODE_BY_LANGUAGE.get(code_language)
-        if default_code is None:
-            raise CodeNodeError(f"Unsupported code language: {code_language}")
-        return _build_default_config(language=code_language, code=default_code)
+        code_provider: type[CodeNodeProvider] = next(
+            provider for provider in cls._DEFAULT_CODE_PROVIDERS if provider.is_accept_language(code_language)
+        )
+
+        return code_provider.get_default_config()
+
+    @classmethod
+    def default_code_providers(cls) -> tuple[type[CodeNodeProvider], ...]:
+        return cls._DEFAULT_CODE_PROVIDERS
 
     @classmethod
     def version(cls) -> str:
@@ -129,6 +108,7 @@ class CodeNode(Node[CodeNodeData]):
                 variables[variable_name] = variable.to_object() if variable else None
         # Run code
         try:
+            _ = self._select_code_provider(code_language)
             result = self._code_executor.execute(
                 language=code_language,
                 code=code,
@@ -150,6 +130,12 @@ class CodeNode(Node[CodeNodeData]):
 
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=variables, outputs=result)
 
+    def _select_code_provider(self, code_language: CodeLanguage) -> type[CodeNodeProvider]:
+        for provider in self._code_providers:
+            if provider.is_accept_language(code_language):
+                return provider
+        raise CodeNodeError(f"Unsupported code language: {code_language}")
+
     def _check_string(self, value: str | None, variable: str) -> str | None:
         """
         Check string

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

@@ -1,18 +1,11 @@
-from enum import StrEnum
 from typing import Annotated, Literal
 
 from pydantic import AfterValidator, BaseModel
 
+from core.helper.code_executor.code_executor import CodeLanguage
+from core.variables.types import SegmentType
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.base.entities import VariableSelector
-from core.workflow.variables.types import SegmentType
-
-
-class CodeLanguage(StrEnum):
-    PYTHON3 = "python3"
-    JINJA2 = "jinja2"
-    JAVASCRIPT = "javascript"
-
 
 _ALLOWED_OUTPUT_FROM_CODE = frozenset(
     [

api/core/workflow/nodes/document_extractor/node.py

@@ -21,12 +21,12 @@ from docx.table import Table
 from docx.text.paragraph import Paragraph
 
 from core.helper import ssrf_proxy
+from core.variables import ArrayFileSegment
+from core.variables.segments import ArrayStringSegment, FileSegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.file import File, FileTransferMethod, file_manager
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
-from core.workflow.variables import ArrayFileSegment
-from core.workflow.variables.segments import ArrayStringSegment, FileSegment
 
 from .entities import DocumentExtractorNodeData, UnstructuredApiConfig
 from .exc import DocumentExtractorError, FileDownloadError, TextExtractionError, UnsupportedFileTypeError

api/core/workflow/nodes/http_request/executor.py

@@ -10,9 +10,11 @@ from urllib.parse import urlencode, urlparse
 import httpx
 from json_repair import repair_json
 
+from core.helper.ssrf_proxy import ssrf_proxy
+from core.variables.segments import ArrayFileSegment, FileSegment
 from core.workflow.file.enums import FileTransferMethod
+from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.segments import ArrayFileSegment, FileSegment
 
 from ..protocols import FileManagerProtocol, HttpClientProtocol
 from .entities import (
@@ -79,8 +81,8 @@ class Executor:
         http_request_config: HttpRequestNodeConfig,
         max_retries: int | None = None,
         ssl_verify: bool | None = None,
-        http_client: HttpClientProtocol,
-        file_manager: FileManagerProtocol,
+        http_client: HttpClientProtocol | None = None,
+        file_manager: FileManagerProtocol | None = None,
     ):
         self._http_request_config = http_request_config
         # If authorization API key is present, convert the API key using the variable pool
@@ -114,8 +116,8 @@ class Executor:
         self.max_retries = (
             max_retries if max_retries is not None else self._http_request_config.ssrf_default_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

api/core/workflow/nodes/http_request/node.py

@@ -3,15 +3,18 @@ import mimetypes
 from collections.abc import Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+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
 from core.workflow.file import File, FileTransferMethod
+from core.workflow.file.file_manager import file_manager as default_file_manager
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.http_request.executor import Executor
-from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol, ToolFileManagerProtocol
-from core.workflow.variables.segments import ArrayFileSegment
+from core.workflow.nodes.protocols import FileManagerProtocol, HttpClientProtocol
 from factories import file_factory
 
 from .config import build_http_request_config, resolve_http_request_config
@@ -42,9 +45,9 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         graph_runtime_state: "GraphRuntimeState",
         *,
         http_request_config: HttpRequestNodeConfig,
-        http_client: HttpClientProtocol,
-        tool_file_manager_factory: Callable[[], ToolFileManagerProtocol],
-        file_manager: FileManagerProtocol,
+        http_client: HttpClientProtocol | None = None,
+        tool_file_manager_factory: Callable[[], ToolFileManager] = ToolFileManager,
+        file_manager: FileManagerProtocol | None = None,
     ) -> None:
         super().__init__(
             id=id,
@@ -52,11 +55,10 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,
         )
-
         self._http_request_config = http_request_config
-        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]:

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

@@ -10,10 +10,10 @@ from typing import Annotated, Any, ClassVar, Literal, Self
 
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+from core.variables.consts import SELECTORS_LENGTH
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.consts import SELECTORS_LENGTH
 
 from .enums import ButtonStyle, DeliveryMethodType, EmailRecipientType, FormInputType, PlaceholderType, TimeoutUnit
 

api/core/workflow/nodes/iteration/iteration_node.py

@@ -7,6 +7,9 @@ from typing import TYPE_CHECKING, Any, NewType, cast
 from typing_extensions import TypeIs
 
 from core.model_runtime.entities.llm_entities import LLMUsage
+from core.variables import IntegerVariable, NoneSegment
+from core.variables.segments import ArrayAnySegment, ArraySegment
+from core.variables.variables import Variable
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.enums import (
     NodeExecutionType,
@@ -33,9 +36,6 @@ from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.iteration.entities import ErrorHandleMode, IterationNodeData
 from core.workflow.runtime import VariablePool
-from core.workflow.variables import IntegerVariable, NoneSegment
-from core.workflow.variables.segments import ArrayAnySegment, ArraySegment
-from core.workflow.variables.variables import Variable
 from libs.datetime_utils import naive_utc_now
 
 from .exc import (
@@ -588,7 +588,6 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
     def _create_graph_engine(self, index: int, item: object):
         # Import dependencies
-        from core.app.workflow.layers.llm_quota import LLMQuotaLayer
         from core.app.workflow.node_factory import DifyNodeFactory
         from core.workflow.entities import GraphInitParams
         from core.workflow.graph import Graph
@@ -643,6 +642,5 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
             command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
             config=GraphEngineConfig(),
         )
-        graph_engine.layer(LLMQuotaLayer())
 
         return graph_engine

api/core/workflow/nodes/knowledge_retrieval/knowledge_retrieval_node.py

@@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any, Literal
 from core.app.app_config.entities import DatasetRetrieveConfigEntity
 from core.model_runtime.entities.llm_entities import LLMUsage
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.variables import (
+    ArrayFileSegment,
+    FileSegment,
+    StringSegment,
+)
+from core.variables.segments import ArrayObjectSegment
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import (
     NodeType,
@@ -16,12 +22,6 @@ from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.llm.file_saver import FileSaverImpl, LLMFileSaver
 from core.workflow.repositories.rag_retrieval_protocol import KnowledgeRetrievalRequest, RAGRetrievalProtocol, Source
-from core.workflow.variables import (
-    ArrayFileSegment,
-    FileSegment,
-    StringSegment,
-)
-from core.workflow.variables.segments import ArrayObjectSegment
 
 from .entities import KnowledgeRetrievalNodeData
 from .exc import (

api/core/workflow/nodes/list_operator/node.py

@@ -1,12 +1,12 @@
 from collections.abc import Callable, Sequence
 from typing import Any, TypeAlias, TypeVar
 
+from core.variables import ArrayFileSegment, ArrayNumberSegment, ArrayStringSegment
+from core.variables.segments import ArrayAnySegment, ArrayBooleanSegment, ArraySegment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
-from core.workflow.variables import ArrayFileSegment, ArrayNumberSegment, ArrayStringSegment
-from core.workflow.variables.segments import ArrayAnySegment, ArrayBooleanSegment, ArraySegment
 
 from .entities import FilterOperator, ListOperatorNodeData, Order
 from .exc import InvalidConditionError, InvalidFilterValueError, InvalidKeyError, ListOperatorError

api/core/workflow/nodes/llm/llm_utils.py

@@ -1,31 +1,74 @@
 from collections.abc import Sequence
 from typing import cast
 
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+
+from configs import dify_config
+from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.model_manager import ModelInstance
-from core.model_runtime.entities import PromptMessageRole
-from core.model_runtime.entities.message_entities import (
-    ImagePromptMessageContent,
-    PromptMessage,
-    TextPromptMessageContent,
-)
-from core.model_runtime.entities.model_entities import AIModelEntity
-from core.model_runtime.memory import PromptMessageMemory
+from core.model_runtime.entities.llm_entities import LLMUsage
+from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
+from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment, StringSegment
+from core.workflow.enums import SystemVariableKey
 from core.workflow.file.models import File
+from core.workflow.nodes.llm.entities import ModelConfig
+from core.workflow.nodes.llm.exc import LLMModeRequiredError, ModelNotExistError
+from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.segments import ArrayAnySegment, ArrayFileSegment, FileSegment, NoneSegment
+from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
+from models.model import Conversation
+from models.provider import Provider, ProviderType
+from models.provider_ids import ModelProviderID
 
 from .exc import InvalidVariableTypeError
 
 
-def fetch_model_schema(*, model_instance: ModelInstance) -> AIModelEntity:
-    model_schema = cast(LargeLanguageModel, model_instance.model_type_instance).get_model_schema(
-        model_instance.model_name,
-        model_instance.credentials,
+def fetch_model_config(
+    *,
+    node_data_model: ModelConfig,
+    credentials_provider: CredentialsProvider,
+    model_factory: ModelFactory,
+) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
+    if not node_data_model.mode:
+        raise LLMModeRequiredError("LLM mode is required.")
+
+    credentials = credentials_provider.fetch(node_data_model.provider, node_data_model.name)
+    model_instance = model_factory.init_model_instance(node_data_model.provider, node_data_model.name)
+    provider_model_bundle = model_instance.provider_model_bundle
+
+    provider_model = provider_model_bundle.configuration.get_provider_model(
+        model=node_data_model.name,
+        model_type=ModelType.LLM,
     )
+    if provider_model is None:
+        raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
+    provider_model.raise_for_status()
+
+    stop: list[str] = []
+    if "stop" in node_data_model.completion_params:
+        stop = node_data_model.completion_params.pop("stop")
+
+    model_schema = model_instance.model_type_instance.get_model_schema(node_data_model.name, credentials)
     if not model_schema:
-        raise ValueError(f"Model schema not found for {model_instance.model_name}")
-    return model_schema
+        raise ModelNotExistError(f"Model {node_data_model.name} not exist.")
+
+    model_instance.model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
+    return model_instance, ModelConfigWithCredentialsEntity(
+        provider=node_data_model.provider,
+        model=node_data_model.name,
+        model_schema=model_schema,
+        mode=node_data_model.mode,
+        provider_model_bundle=provider_model_bundle,
+        credentials=credentials,
+        parameters=node_data_model.completion_params,
+        stop=stop,
+    )
 
 
 def fetch_files(variable_pool: VariablePool, selector: Sequence[str]) -> Sequence["File"]:
@@ -41,51 +84,88 @@ def fetch_files(variable_pool: VariablePool, selector: Sequence[str]) -> Sequenc
     raise InvalidVariableTypeError(f"Invalid variable type: {type(variable)}")
 
 
-def convert_history_messages_to_text(
-    *,
-    history_messages: Sequence[PromptMessage],
-    human_prefix: str,
-    ai_prefix: str,
-) -> str:
-    string_messages: list[str] = []
-    for message in history_messages:
-        if message.role == PromptMessageRole.USER:
-            role = human_prefix
-        elif message.role == PromptMessageRole.ASSISTANT:
-            role = ai_prefix
+def fetch_memory(
+    variable_pool: VariablePool, app_id: str, node_data_memory: MemoryConfig | None, model_instance: ModelInstance
+) -> TokenBufferMemory | None:
+    if not node_data_memory:
+        return None
+
+    # get conversation id
+    conversation_id_variable = variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
+    if not isinstance(conversation_id_variable, StringSegment):
+        return None
+    conversation_id = conversation_id_variable.value
+
+    with Session(db.engine, expire_on_commit=False) as session:
+        stmt = select(Conversation).where(Conversation.app_id == app_id, Conversation.id == conversation_id)
+        conversation = session.scalar(stmt)
+        if not conversation:
+            return None
+
+    memory = TokenBufferMemory(conversation=conversation, model_instance=model_instance)
+    return memory
+
+
+def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUsage):
+    provider_model_bundle = model_instance.provider_model_bundle
+    provider_configuration = provider_model_bundle.configuration
+
+    if provider_configuration.using_provider_type != ProviderType.SYSTEM:
+        return
+
+    system_configuration = provider_configuration.system_configuration
+
+    quota_unit = None
+    for quota_configuration in system_configuration.quota_configurations:
+        if quota_configuration.quota_type == system_configuration.current_quota_type:
+            quota_unit = quota_configuration.quota_unit
+
+            if quota_configuration.quota_limit == -1:
+                return
+
+            break
+
+    used_quota = None
+    if quota_unit:
+        if quota_unit == QuotaUnit.TOKENS:
+            used_quota = usage.total_tokens
+        elif quota_unit == QuotaUnit.CREDITS:
+            used_quota = dify_config.get_model_credits(model_instance.model_name)
         else:
-            continue
+            used_quota = 1
 
-        if isinstance(message.content, list):
-            content_parts = []
-            for content in message.content:
-                if isinstance(content, TextPromptMessageContent):
-                    content_parts.append(content.data)
-                elif isinstance(content, ImagePromptMessageContent):
-                    content_parts.append("[image]")
+    if used_quota is not None and system_configuration.current_quota_type is not None:
+        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+            from services.credit_pool_service import CreditPoolService
 
-            inner_msg = "\n".join(content_parts)
-            string_messages.append(f"{role}: {inner_msg}")
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+            )
+        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
+            from services.credit_pool_service import CreditPoolService
+
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+                pool_type="paid",
+            )
         else:
-            string_messages.append(f"{role}: {message.content}")
-
-    return "\n".join(string_messages)
-
-
-def fetch_memory_text(
-    *,
-    memory: PromptMessageMemory,
-    max_token_limit: int,
-    message_limit: int | None = None,
-    human_prefix: str = "Human",
-    ai_prefix: str = "Assistant",
-) -> str:
-    history_messages = memory.get_history_prompt_messages(
-        max_token_limit=max_token_limit,
-        message_limit=message_limit,
-    )
-    return convert_history_messages_to_text(
-        history_messages=history_messages,
-        human_prefix=human_prefix,
-        ai_prefix=ai_prefix,
-    )
+            with Session(db.engine) as session:
+                stmt = (
+                    update(Provider)
+                    .where(
+                        Provider.tenant_id == tenant_id,
+                        # TODO: Use provider name with prefix after the data migration.
+                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                        Provider.provider_type == ProviderType.SYSTEM.value,
+                        Provider.quota_type == system_configuration.current_quota_type.value,
+                        Provider.quota_limit > Provider.quota_used,
+                    )
+                    .values(
+                        quota_used=Provider.quota_used + used_quota,
+                        last_used=naive_utc_now(),
+                    )
+                )
+                session.execute(stmt)
+                session.commit()

api/core/workflow/nodes/llm/node.py

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Literal
 
 from sqlalchemy import select
 
+from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.helper.code_executor import CodeExecutor, CodeLanguage
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
@@ -36,13 +37,21 @@ from core.model_runtime.entities.message_entities import (
     SystemPromptMessage,
     UserPromptMessage,
 )
-from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from core.model_runtime.memory import PromptMessageMemory
+from core.model_runtime.entities.model_entities import AIModelEntity, ModelFeature, ModelPropertyKey
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.prompt.entities.advanced_prompt_entities import CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.tools.signature import sign_upload_file
+from core.variables import (
+    ArrayFileSegment,
+    ArraySegment,
+    FileSegment,
+    NoneSegment,
+    ObjectSegment,
+    StringSegment,
+)
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import (
@@ -65,14 +74,6 @@ from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from core.workflow.runtime import VariablePool
-from core.workflow.variables import (
-    ArrayFileSegment,
-    ArraySegment,
-    FileSegment,
-    NoneSegment,
-    ObjectSegment,
-    StringSegment,
-)
 from extensions.ext_database import db
 from models.dataset import SegmentAttachmentBinding
 from models.model import UploadFile
@@ -82,6 +83,7 @@ from .entities import (
     LLMNodeChatModelMessage,
     LLMNodeCompletionModelPromptTemplate,
     LLMNodeData,
+    ModelConfig,
 )
 from .exc import (
     InvalidContextStructureError,
@@ -114,8 +116,6 @@ class LLMNode(Node[LLMNodeData]):
     _llm_file_saver: LLMFileSaver
     _credentials_provider: CredentialsProvider
     _model_factory: ModelFactory
-    _model_instance: ModelInstance
-    _memory: PromptMessageMemory | None
 
     def __init__(
         self,
@@ -126,8 +126,6 @@ class LLMNode(Node[LLMNodeData]):
         *,
         credentials_provider: CredentialsProvider,
         model_factory: ModelFactory,
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None = None,
         llm_file_saver: LLMFileSaver | None = None,
     ):
         super().__init__(
@@ -141,8 +139,6 @@ class LLMNode(Node[LLMNodeData]):
 
         self._credentials_provider = credentials_provider
         self._model_factory = model_factory
-        self._model_instance = model_instance
-        self._memory = memory
 
         if llm_file_saver is None:
             llm_file_saver = FileSaverImpl(
@@ -206,12 +202,29 @@ class LLMNode(Node[LLMNodeData]):
                 node_inputs["#context_files#"] = [file.model_dump() for file in context_files]
 
             # fetch model config
-            model_instance = self._model_instance
-            model_name = model_instance.model_name
-            model_provider = model_instance.provider
-            model_stop = model_instance.stop
+            model_instance, model_config = self._fetch_model_config(
+                node_data_model=self.node_data.model,
+            )
+            model_name = getattr(model_instance, "model_name", None)
+            if not isinstance(model_name, str):
+                model_name = model_config.model
+            model_provider = getattr(model_instance, "provider", None)
+            if not isinstance(model_provider, str):
+                model_provider = model_config.provider
+            model_schema = model_instance.model_type_instance.get_model_schema(
+                model_name,
+                model_instance.credentials,
+            )
+            if not model_schema:
+                raise ValueError(f"Model schema not found for {model_name}")
 
-            memory = self._memory
+            # fetch memory
+            memory = llm_utils.fetch_memory(
+                variable_pool=variable_pool,
+                app_id=self.app_id,
+                node_data_memory=self.node_data.memory,
+                model_instance=model_instance,
+            )
 
             query: str | None = None
             if self.node_data.memory:
@@ -227,7 +240,9 @@ class LLMNode(Node[LLMNodeData]):
                 context=context,
                 memory=memory,
                 model_instance=model_instance,
-                stop=model_stop,
+                model_schema=model_schema,
+                model_parameters=self.node_data.model.completion_params,
+                stop=model_config.stop,
                 prompt_template=self.node_data.prompt_template,
                 memory_config=self.node_data.memory,
                 vision_enabled=self.node_data.vision.enabled,
@@ -239,6 +254,7 @@ class LLMNode(Node[LLMNodeData]):
 
             # handle invoke result
             generator = LLMNode.invoke_llm(
+                node_data_model=self.node_data.model,
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 stop=stop,
@@ -279,6 +295,8 @@ class LLMNode(Node[LLMNodeData]):
                         else None
                     )
 
+                    # deduct quota
+                    llm_utils.deduct_llm_quota(tenant_id=self.tenant_id, model_instance=model_instance, usage=usage)
                     break
                 elif isinstance(event, LLMStructuredOutput):
                     structured_output = event
@@ -353,6 +371,7 @@ class LLMNode(Node[LLMNodeData]):
     @staticmethod
     def invoke_llm(
         *,
+        node_data_model: ModelConfig,
         model_instance: ModelInstance,
         prompt_messages: Sequence[PromptMessage],
         stop: Sequence[str] | None = None,
@@ -365,10 +384,11 @@ class LLMNode(Node[LLMNodeData]):
         node_type: NodeType,
         reasoning_format: Literal["separated", "tagged"] = "tagged",
     ) -> Generator[NodeEventBase | LLMStructuredOutput, None, None]:
-        model_parameters = model_instance.parameters
-        invoke_model_parameters = dict(model_parameters)
-
-        model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
+        model_schema = model_instance.model_type_instance.get_model_schema(
+            node_data_model.name, model_instance.credentials
+        )
+        if not model_schema:
+            raise ValueError(f"Model schema not found for {node_data_model.name}")
 
         if structured_output_enabled:
             output_schema = LLMNode.fetch_structured_output_schema(
@@ -382,7 +402,7 @@ class LLMNode(Node[LLMNodeData]):
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 json_schema=output_schema,
-                model_parameters=invoke_model_parameters,
+                model_parameters=node_data_model.completion_params,
                 stop=list(stop or []),
                 stream=True,
                 user=user_id,
@@ -392,7 +412,7 @@ class LLMNode(Node[LLMNodeData]):
 
             invoke_result = model_instance.invoke_llm(
                 prompt_messages=list(prompt_messages),
-                model_parameters=invoke_model_parameters,
+                model_parameters=node_data_model.completion_params,
                 stop=list(stop or []),
                 stream=True,
                 user=user_id,
@@ -751,14 +771,33 @@ class LLMNode(Node[LLMNodeData]):
 
         return None
 
+    def _fetch_model_config(
+        self,
+        *,
+        node_data_model: ModelConfig,
+    ) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
+        model, model_config_with_cred = llm_utils.fetch_model_config(
+            node_data_model=node_data_model,
+            credentials_provider=self._credentials_provider,
+            model_factory=self._model_factory,
+        )
+        completion_params = model_config_with_cred.parameters
+
+        model_config_with_cred.parameters = completion_params
+        # NOTE(-LAN-): This line modify the `self.node_data.model`, which is used in `_invoke_llm()`.
+        node_data_model.completion_params = completion_params
+        return model, model_config_with_cred
+
     @staticmethod
     def fetch_prompt_messages(
         *,
         sys_query: str | None = None,
         sys_files: Sequence[File],
         context: str | None = None,
-        memory: PromptMessageMemory | None = None,
+        memory: TokenBufferMemory | None = None,
         model_instance: ModelInstance,
+        model_schema: AIModelEntity,
+        model_parameters: Mapping[str, Any],
         prompt_template: Sequence[LLMNodeChatModelMessage] | LLMNodeCompletionModelPromptTemplate,
         stop: Sequence[str] | None = None,
         memory_config: MemoryConfig | None = None,
@@ -769,7 +808,6 @@ class LLMNode(Node[LLMNodeData]):
         context_files: list[File] | None = None,
     ) -> tuple[Sequence[PromptMessage], Sequence[str] | None]:
         prompt_messages: list[PromptMessage] = []
-        model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
 
         if isinstance(prompt_template, list):
             # For chat model
@@ -788,6 +826,8 @@ class LLMNode(Node[LLMNodeData]):
                 memory=memory,
                 memory_config=memory_config,
                 model_instance=model_instance,
+                model_schema=model_schema,
+                model_parameters=model_parameters,
             )
             # Extend prompt_messages with memory messages
             prompt_messages.extend(memory_messages)
@@ -825,6 +865,8 @@ class LLMNode(Node[LLMNodeData]):
                 memory=memory,
                 memory_config=memory_config,
                 model_instance=model_instance,
+                model_schema=model_schema,
+                model_parameters=model_parameters,
             )
             # Insert histories into the prompt
             prompt_content = prompt_messages[0].content
@@ -1233,10 +1275,6 @@ class LLMNode(Node[LLMNodeData]):
     def retry(self) -> bool:
         return self.node_data.retry_config.retry_enabled
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
-
 
 def _combine_message_content_with_role(
     *, contents: str | list[PromptMessageContentUnionTypes] | None = None, role: PromptMessageRole
@@ -1278,23 +1316,23 @@ def _calculate_rest_token(
     *,
     prompt_messages: list[PromptMessage],
     model_instance: ModelInstance,
+    model_schema: AIModelEntity,
+    model_parameters: Mapping[str, Any],
 ) -> int:
     rest_tokens = 2000
-    runtime_model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
-    runtime_model_parameters = model_instance.parameters
 
-    model_context_tokens = runtime_model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+    model_context_tokens = model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
     if model_context_tokens:
         curr_message_tokens = model_instance.get_llm_num_tokens(prompt_messages)
 
         max_tokens = 0
-        for parameter_rule in runtime_model_schema.parameter_rules:
+        for parameter_rule in model_schema.parameter_rules:
             if parameter_rule.name == "max_tokens" or (
                 parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
             ):
                 max_tokens = (
-                    runtime_model_parameters.get(parameter_rule.name)
-                    or runtime_model_parameters.get(str(parameter_rule.use_template))
+                    model_parameters.get(parameter_rule.name)
+                    or model_parameters.get(str(parameter_rule.use_template))
                     or 0
                 )
 
@@ -1306,9 +1344,11 @@ def _calculate_rest_token(
 
 def _handle_memory_chat_mode(
     *,
-    memory: PromptMessageMemory | None,
+    memory: TokenBufferMemory | None,
     memory_config: MemoryConfig | None,
     model_instance: ModelInstance,
+    model_schema: AIModelEntity,
+    model_parameters: Mapping[str, Any],
 ) -> Sequence[PromptMessage]:
     memory_messages: Sequence[PromptMessage] = []
     # Get messages from memory for chat model
@@ -1316,6 +1356,8 @@ def _handle_memory_chat_mode(
         rest_tokens = _calculate_rest_token(
             prompt_messages=[],
             model_instance=model_instance,
+            model_schema=model_schema,
+            model_parameters=model_parameters,
         )
         memory_messages = memory.get_history_prompt_messages(
             max_token_limit=rest_tokens,
@@ -1326,9 +1368,11 @@ def _handle_memory_chat_mode(
 
 def _handle_memory_completion_mode(
     *,
-    memory: PromptMessageMemory | None,
+    memory: TokenBufferMemory | None,
     memory_config: MemoryConfig | None,
     model_instance: ModelInstance,
+    model_schema: AIModelEntity,
+    model_parameters: Mapping[str, Any],
 ) -> str:
     memory_text = ""
     # Get history text from memory for completion model
@@ -1336,11 +1380,12 @@ def _handle_memory_completion_mode(
         rest_tokens = _calculate_rest_token(
             prompt_messages=[],
             model_instance=model_instance,
+            model_schema=model_schema,
+            model_parameters=model_parameters,
         )
         if not memory_config.role_prefix:
             raise MemoryRolePrefixRequiredError("Memory role prefix is required for completion model.")
-        memory_text = llm_utils.fetch_memory_text(
-            memory=memory,
+        memory_text = memory.get_history_prompt_text(
             max_token_limit=rest_tokens,
             message_limit=memory_config.window.size if memory_config.window.enabled else None,
             human_prefix=memory_config.role_prefix.user,

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

@@ -3,9 +3,9 @@ from typing import Annotated, Any, Literal
 
 from pydantic import AfterValidator, BaseModel, Field, field_validator
 
+from core.variables.types import SegmentType
 from core.workflow.nodes.base import BaseLoopNodeData, BaseLoopState, BaseNodeData
 from core.workflow.utils.condition.entities import Condition
-from core.workflow.variables.types import SegmentType
 
 _VALID_VAR_TYPE = frozenset(
     [

api/core/workflow/nodes/loop/loop_node.py

@@ -6,6 +6,7 @@ from datetime import datetime
 from typing import TYPE_CHECKING, Any, Literal, cast
 
 from core.model_runtime.entities.llm_entities import LLMUsage
+from core.variables import Segment, SegmentType
 from core.workflow.enums import (
     NodeExecutionType,
     NodeType,
@@ -30,7 +31,6 @@ from core.workflow.nodes.base import LLMUsageTrackingMixin
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.loop.entities import LoopCompletedReason, LoopNodeData, LoopVariableData
 from core.workflow.utils.condition.processor import ConditionProcessor
-from core.workflow.variables import Segment, SegmentType
 from factories.variable_factory import TypeMismatchError, build_segment_with_type, segment_to_variable
 from libs.datetime_utils import naive_utc_now
 
@@ -413,7 +413,6 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
 
     def _create_graph_engine(self, start_at: datetime, root_node_id: str):
         # Import dependencies
-        from core.app.workflow.layers.llm_quota import LLMQuotaLayer
         from core.app.workflow.node_factory import DifyNodeFactory
         from core.workflow.entities import GraphInitParams
         from core.workflow.graph import Graph
@@ -455,6 +454,5 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             command_channel=InMemoryChannel(),  # Use InMemoryChannel for sub-graphs
             config=GraphEngineConfig(),
         )
-        graph_engine.layer(LLMQuotaLayer())
 
         return graph_engine

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

@@ -8,9 +8,9 @@ from pydantic import (
 )
 
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
+from core.variables.types import SegmentType
 from core.workflow.nodes.base import BaseNodeData
 from core.workflow.nodes.llm.entities import ModelConfig, VisionConfig
-from core.workflow.variables.types import SegmentType
 
 _OLD_BOOL_TYPE_NAME = "bool"
 _OLD_SELECT_TYPE_NAME = "select"

api/core/workflow/nodes/parameter_extractor/exc.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from core.workflow.variables.types import SegmentType
+from core.variables.types import SegmentType
 
 
 class ParameterExtractorNodeError(ValueError):

api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py

@@ -5,6 +5,7 @@ import uuid
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, cast
 
+from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import ImagePromptMessageContent
 from core.model_runtime.entities.llm_entities import LLMUsage
@@ -17,25 +18,21 @@ from core.model_runtime.entities.message_entities import (
     UserPromptMessage,
 )
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from core.model_runtime.memory import PromptMessageMemory
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.entities.advanced_prompt_entities import ChatModelMessage, CompletionModelPromptTemplate
 from core.prompt.simple_prompt_transform import ModelMode
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
-from core.workflow.enums import (
-    NodeType,
-    WorkflowNodeExecutionMetadataKey,
-    WorkflowNodeExecutionStatus,
-)
+from core.variables.types import ArrayValidation, SegmentType
+from core.workflow.enums import NodeType, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from core.workflow.file import File
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base import variable_template_parser
 from core.workflow.nodes.base.node import Node
-from core.workflow.nodes.llm import llm_utils
+from core.workflow.nodes.llm import ModelConfig, llm_utils
 from core.workflow.runtime import VariablePool
-from core.workflow.variables.types import ArrayValidation, SegmentType
 from factories.variable_factory import build_segment_with_type
 
 from .entities import ParameterExtractorNodeData
@@ -98,10 +95,10 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
 
     node_type = NodeType.PARAMETER_EXTRACTOR
 
-    _model_instance: ModelInstance
+    _model_instance: ModelInstance | None = None
+    _model_config: ModelConfigWithCredentialsEntity | None = None
     _credentials_provider: "CredentialsProvider"
     _model_factory: "ModelFactory"
-    _memory: PromptMessageMemory | None
 
     def __init__(
         self,
@@ -112,8 +109,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         *,
         credentials_provider: "CredentialsProvider",
         model_factory: "ModelFactory",
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None = None,
     ) -> None:
         super().__init__(
             id=id,
@@ -123,8 +118,6 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         )
         self._credentials_provider = credentials_provider
         self._model_factory = model_factory
-        self._model_instance = model_instance
-        self._memory = memory
 
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
@@ -162,15 +155,25 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             else []
         )
 
-        model_instance = self._model_instance
+        model_instance, model_config = self._fetch_model_config(node_data.model)
         if not isinstance(model_instance.model_type_instance, LargeLanguageModel):
             raise InvalidModelTypeError("Model is not a Large Language Model")
 
-        try:
-            model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
-        except ValueError as exc:
-            raise ModelSchemaNotFoundError("Model schema not found") from exc
-        memory = self._memory
+        llm_model = model_instance.model_type_instance
+        model_schema = llm_model.get_model_schema(
+            model=model_config.model,
+            credentials=model_config.credentials,
+        )
+        if not model_schema:
+            raise ModelSchemaNotFoundError("Model schema not found")
+
+        # fetch memory
+        memory = llm_utils.fetch_memory(
+            variable_pool=variable_pool,
+            app_id=self.app_id,
+            node_data_memory=node_data.memory,
+            model_instance=model_instance,
+        )
 
         if (
             set(model_schema.features or []) & {ModelFeature.TOOL_CALL, ModelFeature.MULTI_TOOL_CALL}
@@ -181,7 +184,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
                 node_data=node_data,
                 query=query,
                 variable_pool=self.graph_runtime_state.variable_pool,
-                model_instance=model_instance,
+                model_config=model_config,
                 memory=memory,
                 files=files,
                 vision_detail=node_data.vision.configs.detail,
@@ -192,7 +195,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
                 data=node_data,
                 query=query,
                 variable_pool=self.graph_runtime_state.variable_pool,
-                model_instance=model_instance,
+                model_config=model_config,
                 memory=memory,
                 files=files,
                 vision_detail=node_data.vision.configs.detail,
@@ -208,23 +211,24 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         }
 
         process_data = {
-            "model_mode": node_data.model.mode,
+            "model_mode": model_config.mode,
             "prompts": PromptMessageUtil.prompt_messages_to_prompt_for_saving(
-                model_mode=node_data.model.mode, prompt_messages=prompt_messages
+                model_mode=model_config.mode, prompt_messages=prompt_messages
             ),
             "usage": None,
             "function": {} if not prompt_message_tools else jsonable_encoder(prompt_message_tools[0]),
             "tool_call": None,
-            "model_provider": model_instance.provider,
-            "model_name": model_instance.model_name,
+            "model_provider": model_config.provider,
+            "model_name": model_config.model,
         }
 
         try:
             text, usage, tool_call = self._invoke(
+                node_data_model=node_data.model,
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 tools=prompt_message_tools,
-                stop=model_instance.stop,
+                stop=model_config.stop,
             )
             process_data["usage"] = jsonable_encoder(usage)
             process_data["tool_call"] = jsonable_encoder(tool_call)
@@ -286,16 +290,17 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
 
     def _invoke(
         self,
+        node_data_model: ModelConfig,
         model_instance: ModelInstance,
         prompt_messages: list[PromptMessage],
         tools: list[PromptMessageTool],
-        stop: Sequence[str],
+        stop: list[str],
     ) -> tuple[str, LLMUsage, AssistantPromptMessage.ToolCall | None]:
         invoke_result = model_instance.invoke_llm(
             prompt_messages=prompt_messages,
-            model_parameters=dict(model_instance.parameters),
+            model_parameters=node_data_model.completion_params,
             tools=tools,
-            stop=list(stop),
+            stop=stop,
             stream=False,
             user=self.user_id,
         )
@@ -309,6 +314,9 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         usage = invoke_result.usage
         tool_call = invoke_result.message.tool_calls[0] if invoke_result.message.tool_calls else None
 
+        # deduct quota
+        llm_utils.deduct_llm_quota(tenant_id=self.tenant_id, model_instance=model_instance, usage=usage)
+
         return text, usage, tool_call
 
     def _generate_function_call_prompt(
@@ -316,8 +324,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        model_config: ModelConfigWithCredentialsEntity,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> tuple[list[PromptMessage], list[PromptMessageTool]]:
@@ -329,13 +337,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         )
 
         prompt_transform = AdvancedPromptTransform(with_variable_tmpl=True)
-        rest_token = self._calculate_rest_token(
-            node_data=node_data,
-            query=query,
-            variable_pool=variable_pool,
-            model_instance=model_instance,
-            context="",
-        )
+        rest_token = self._calculate_rest_token(node_data, query, variable_pool, model_config, "")
         prompt_template = self._get_function_calling_prompt_template(
             node_data, query, variable_pool, memory, rest_token
         )
@@ -347,7 +349,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             context="",
             memory_config=node_data.memory,
             memory=None,
-            model_instance=model_instance,
+            model_config=model_config,
             image_detail_config=vision_detail,
         )
 
@@ -404,8 +406,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        model_config: ModelConfigWithCredentialsEntity,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -419,7 +421,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
                 node_data=data,
                 query=query,
                 variable_pool=variable_pool,
-                model_instance=model_instance,
+                model_config=model_config,
                 memory=memory,
                 files=files,
                 vision_detail=vision_detail,
@@ -429,7 +431,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
                 node_data=data,
                 query=query,
                 variable_pool=variable_pool,
-                model_instance=model_instance,
+                model_config=model_config,
                 memory=memory,
                 files=files,
                 vision_detail=vision_detail,
@@ -442,8 +444,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        model_config: ModelConfigWithCredentialsEntity,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -452,11 +454,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         """
         prompt_transform = AdvancedPromptTransform(with_variable_tmpl=True)
         rest_token = self._calculate_rest_token(
-            node_data=node_data,
-            query=query,
-            variable_pool=variable_pool,
-            model_instance=model_instance,
-            context="",
+            node_data=node_data, query=query, variable_pool=variable_pool, model_config=model_config, context=""
         )
         prompt_template = self._get_prompt_engineering_prompt_template(
             node_data=node_data, query=query, variable_pool=variable_pool, memory=memory, max_token_limit=rest_token
@@ -468,9 +466,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             files=files,
             context="",
             memory_config=node_data.memory,
-            # AdvancedPromptTransform is still typed against TokenBufferMemory.
-            memory=cast(Any, memory),
-            model_instance=model_instance,
+            memory=memory,
+            model_config=model_config,
             image_detail_config=vision_detail,
         )
 
@@ -481,8 +478,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None,
+        model_config: ModelConfigWithCredentialsEntity,
+        memory: TokenBufferMemory | None,
         files: Sequence[File],
         vision_detail: ImagePromptMessageContent.DETAIL | None = None,
     ) -> list[PromptMessage]:
@@ -491,11 +488,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         """
         prompt_transform = AdvancedPromptTransform(with_variable_tmpl=True)
         rest_token = self._calculate_rest_token(
-            node_data=node_data,
-            query=query,
-            variable_pool=variable_pool,
-            model_instance=model_instance,
-            context="",
+            node_data=node_data, query=query, variable_pool=variable_pool, model_config=model_config, context=""
         )
         prompt_template = self._get_prompt_engineering_prompt_template(
             node_data=node_data,
@@ -515,7 +508,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             context="",
             memory_config=node_data.memory,
             memory=None,
-            model_instance=model_instance,
+            model_config=model_config,
             image_detail_config=vision_detail,
         )
 
@@ -714,7 +707,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ) -> list[ChatModelMessage]:
         model_mode = ModelMode(node_data.model.mode)
@@ -723,8 +716,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         instruction = variable_pool.convert_template(node_data.instruction or "").text
 
         if memory and node_data.memory and node_data.memory.window:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory, max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
+            memory_str = memory.get_history_prompt_text(
+                max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
             )
         if model_mode == ModelMode.CHAT:
             system_prompt_messages = ChatModelMessage(
@@ -741,7 +734,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)
@@ -750,8 +743,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         instruction = variable_pool.convert_template(node_data.instruction or "").text
 
         if memory and node_data.memory and node_data.memory.window:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory, max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
+            memory_str = memory.get_history_prompt_text(
+                max_token_limit=max_token_limit, message_limit=node_data.memory.window.size
             )
         if model_mode == ModelMode.CHAT:
             system_prompt_messages = ChatModelMessage(
@@ -776,16 +769,21 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         node_data: ParameterExtractorNodeData,
         query: str,
         variable_pool: VariablePool,
-        model_instance: ModelInstance,
+        model_config: ModelConfigWithCredentialsEntity,
         context: str | None,
     ) -> int:
-        try:
-            model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
-        except ValueError as exc:
-            raise ModelSchemaNotFoundError("Model schema not found") from exc
         prompt_transform = AdvancedPromptTransform(with_variable_tmpl=True)
 
-        if set(model_schema.features or []) & {ModelFeature.TOOL_CALL, ModelFeature.MULTI_TOOL_CALL}:
+        model_instance, model_config = self._fetch_model_config(node_data.model)
+        if not isinstance(model_instance.model_type_instance, LargeLanguageModel):
+            raise InvalidModelTypeError("Model is not a Large Language Model")
+
+        llm_model = model_instance.model_type_instance
+        model_schema = llm_model.get_model_schema(model_config.model, model_config.credentials)
+        if not model_schema:
+            raise ModelSchemaNotFoundError("Model schema not found")
+
+        if set(model_schema.features or []) & {ModelFeature.MULTI_TOOL_CALL, ModelFeature.MULTI_TOOL_CALL}:
             prompt_template = self._get_function_calling_prompt_template(node_data, query, variable_pool, None, 2000)
         else:
             prompt_template = self._get_prompt_engineering_prompt_template(node_data, query, variable_pool, None, 2000)
@@ -798,28 +796,27 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
             context=context,
             memory_config=node_data.memory,
             memory=None,
-            model_instance=model_instance,
+            model_config=model_config,
         )
         rest_tokens = 2000
 
-        model_context_tokens = model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
         if model_context_tokens:
-            model_type_instance = cast(LargeLanguageModel, model_instance.model_type_instance)
+            model_type_instance = model_config.provider_model_bundle.model_type_instance
+            model_type_instance = cast(LargeLanguageModel, model_type_instance)
+
             curr_message_tokens = (
-                model_type_instance.get_num_tokens(
-                    model_instance.model_name, model_instance.credentials, prompt_messages
-                )
-                + 1000
+                model_type_instance.get_num_tokens(model_config.model, model_config.credentials, prompt_messages) + 1000
             )  # add 1000 to ensure tool call messages
 
             max_tokens = 0
-            for parameter_rule in model_schema.parameter_rules:
+            for parameter_rule in model_config.model_schema.parameter_rules:
                 if parameter_rule.name == "max_tokens" or (
                     parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
                 ):
                     max_tokens = (
-                        model_instance.parameters.get(parameter_rule.name)
-                        or model_instance.parameters.get(parameter_rule.use_template or "")
+                        model_config.parameters.get(parameter_rule.name)
+                        or model_config.parameters.get(parameter_rule.use_template or "")
                     ) or 0
 
             rest_tokens = model_context_tokens - max_tokens - curr_message_tokens
@@ -827,9 +824,20 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
 
         return rest_tokens
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
+    def _fetch_model_config(
+        self, node_data_model: ModelConfig
+    ) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
+        """
+        Fetch model config.
+        """
+        if not self._model_instance or not self._model_config:
+            self._model_instance, self._model_config = llm_utils.fetch_model_config(
+                node_data_model=node_data_model,
+                credentials_provider=self._credentials_provider,
+                model_factory=self._model_factory,
+            )
+
+        return self._model_instance, self._model_config
 
     @classmethod
     def _extract_variable_selector_to_variable_mapping(

api/core/workflow/nodes/protocols.py

@@ -27,16 +27,3 @@ class HttpClientProtocol(Protocol):
 
 class FileManagerProtocol(Protocol):
     def download(self, f: File, /) -> bytes: ...
-
-
-class ToolFileManagerProtocol(Protocol):
-    def create_file_by_raw(
-        self,
-        *,
-        user_id: str,
-        tenant_id: str,
-        conversation_id: str | None,
-        file_binary: bytes,
-        mimetype: str,
-        filename: str | None = None,
-    ) -> Any: ...

api/core/workflow/nodes/question_classifier/question_classifier_node.py

@@ -3,10 +3,12 @@ import re
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.model_manager import ModelInstance
 from core.model_runtime.entities import LLMUsage, ModelPropertyKey, PromptMessageRole
-from core.model_runtime.memory import PromptMessageMemory
+from core.model_runtime.token_buffer_memory import TokenBufferMemory
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.prompt.advanced_prompt_transform import AdvancedPromptTransform
 from core.prompt.simple_prompt_transform import ModelMode
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.workflow.entities import GraphInitParams
@@ -20,12 +22,7 @@ from core.workflow.node_events import ModelInvokeCompletedEvent, NodeRunResult
 from core.workflow.nodes.base.entities import VariableSelector
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
-from core.workflow.nodes.llm import (
-    LLMNode,
-    LLMNodeChatModelMessage,
-    LLMNodeCompletionModelPromptTemplate,
-    llm_utils,
-)
+from core.workflow.nodes.llm import LLMNode, LLMNodeChatModelMessage, LLMNodeCompletionModelPromptTemplate, llm_utils
 from core.workflow.nodes.llm.file_saver import FileSaverImpl, LLMFileSaver
 from core.workflow.nodes.llm.protocols import CredentialsProvider, ModelFactory
 from libs.json_in_md_parser import parse_and_check_json_markdown
@@ -55,8 +52,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
     _llm_file_saver: LLMFileSaver
     _credentials_provider: "CredentialsProvider"
     _model_factory: "ModelFactory"
-    _model_instance: ModelInstance
-    _memory: PromptMessageMemory | None
 
     def __init__(
         self,
@@ -67,8 +62,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         *,
         credentials_provider: "CredentialsProvider",
         model_factory: "ModelFactory",
-        model_instance: ModelInstance,
-        memory: PromptMessageMemory | None = None,
         llm_file_saver: LLMFileSaver | None = None,
     ):
         super().__init__(
@@ -82,8 +75,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
 
         self._credentials_provider = credentials_provider
         self._model_factory = model_factory
-        self._model_instance = model_instance
-        self._memory = memory
 
         if llm_file_saver is None:
             llm_file_saver = FileSaverImpl(
@@ -104,9 +95,25 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         variable = variable_pool.get(node_data.query_variable_selector) if node_data.query_variable_selector else None
         query = variable.value if variable else None
         variables = {"query": query}
-        # fetch model instance
-        model_instance = self._model_instance
-        memory = self._memory
+        # fetch model config
+        model_instance, model_config = llm_utils.fetch_model_config(
+            node_data_model=node_data.model,
+            credentials_provider=self._credentials_provider,
+            model_factory=self._model_factory,
+        )
+        model_schema = model_instance.model_type_instance.get_model_schema(
+            model_instance.model_name,
+            model_instance.credentials,
+        )
+        if not model_schema:
+            raise ValueError(f"Model schema not found for {model_instance.model_name}")
+        # fetch memory
+        memory = llm_utils.fetch_memory(
+            variable_pool=variable_pool,
+            app_id=self.app_id,
+            node_data_memory=node_data.memory,
+            model_instance=model_instance,
+        )
         # fetch instruction
         node_data.instruction = node_data.instruction or ""
         node_data.instruction = variable_pool.convert_template(node_data.instruction).text
@@ -124,7 +131,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         rest_token = self._calculate_rest_token(
             node_data=node_data,
             query=query or "",
-            model_instance=model_instance,
+            model_config=model_config,
             context="",
         )
         prompt_template = self._get_prompt_template(
@@ -142,7 +149,9 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
             sys_query="",
             memory=memory,
             model_instance=model_instance,
-            stop=model_instance.stop,
+            model_schema=model_schema,
+            model_parameters=node_data.model.completion_params,
+            stop=model_config.stop,
             sys_files=files,
             vision_enabled=node_data.vision.enabled,
             vision_detail=node_data.vision.configs.detail,
@@ -157,6 +166,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         try:
             # handle invoke result
             generator = LLMNode.invoke_llm(
+                node_data_model=node_data.model,
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 stop=stop,
@@ -195,14 +205,14 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
                     category_name = classes_map[category_id_result]
                     category_id = category_id_result
             process_data = {
-                "model_mode": node_data.model.mode,
+                "model_mode": model_config.mode,
                 "prompts": PromptMessageUtil.prompt_messages_to_prompt_for_saving(
-                    model_mode=node_data.model.mode, prompt_messages=prompt_messages
+                    model_mode=model_config.mode, prompt_messages=prompt_messages
                 ),
                 "usage": jsonable_encoder(usage),
                 "finish_reason": finish_reason,
-                "model_provider": model_instance.provider,
-                "model_name": model_instance.model_name,
+                "model_provider": model_config.provider,
+                "model_name": model_config.model,
             }
             outputs = {
                 "class_name": category_name,
@@ -237,10 +247,6 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
                 llm_usage=usage,
             )
 
-    @property
-    def model_instance(self) -> ModelInstance:
-        return self._model_instance
-
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,
@@ -279,40 +285,39 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self,
         node_data: QuestionClassifierNodeData,
         query: str,
-        model_instance: ModelInstance,
+        model_config: ModelConfigWithCredentialsEntity,
         context: str | None,
     ) -> int:
-        model_schema = llm_utils.fetch_model_schema(model_instance=model_instance)
-
+        prompt_transform = AdvancedPromptTransform(with_variable_tmpl=True)
         prompt_template = self._get_prompt_template(node_data, query, None, 2000)
-        prompt_messages, _ = LLMNode.fetch_prompt_messages(
+        prompt_messages = prompt_transform.get_prompt(
             prompt_template=prompt_template,
-            sys_query="",
-            sys_files=[],
+            inputs={},
+            query="",
+            files=[],
             context=context,
-            memory=None,
-            model_instance=model_instance,
-            stop=model_instance.stop,
             memory_config=node_data.memory,
-            vision_enabled=False,
-            vision_detail=node_data.vision.configs.detail,
-            variable_pool=self.graph_runtime_state.variable_pool,
-            jinja2_variables=[],
+            memory=None,
+            model_config=model_config,
         )
         rest_tokens = 2000
 
-        model_context_tokens = model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
+        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
         if model_context_tokens:
+            model_instance = ModelInstance(
+                provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
+            )
+
             curr_message_tokens = model_instance.get_llm_num_tokens(prompt_messages)
 
             max_tokens = 0
-            for parameter_rule in model_schema.parameter_rules:
+            for parameter_rule in model_config.model_schema.parameter_rules:
                 if parameter_rule.name == "max_tokens" or (
                     parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
                 ):
                     max_tokens = (
-                        model_instance.parameters.get(parameter_rule.name)
-                        or model_instance.parameters.get(parameter_rule.use_template or "")
+                        model_config.parameters.get(parameter_rule.name)
+                        or model_config.parameters.get(parameter_rule.use_template or "")
                     ) or 0
 
             rest_tokens = model_context_tokens - max_tokens - curr_message_tokens
@@ -324,7 +329,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         self,
         node_data: QuestionClassifierNodeData,
         query: str,
-        memory: PromptMessageMemory | None,
+        memory: TokenBufferMemory | None,
         max_token_limit: int = 2000,
     ):
         model_mode = ModelMode(node_data.model.mode)
@@ -337,8 +342,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         input_text = query
         memory_str = ""
         if memory:
-            memory_str = llm_utils.fetch_memory_text(
-                memory=memory,
+            memory_str = memory.get_history_prompt_text(
                 max_token_limit=max_token_limit,
                 message_limit=node_data.memory.window.size if node_data.memory and node_data.memory.window else None,
             )

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

@@ -2,8 +2,8 @@ from collections.abc import Sequence
 
 from pydantic import Field
 
+from core.app.app_config.entities import VariableEntity
 from core.workflow.nodes.base import BaseNodeData
-from core.workflow.variables.input_entities import VariableEntity
 
 
 class StartNodeData(BaseNodeData):

api/core/workflow/nodes/start/start_node.py

@@ -2,12 +2,12 @@ from typing import Any
 
 from jsonschema import Draft7Validator, ValidationError
 
+from core.app.app_config.entities import VariableEntityType
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.enums import NodeExecutionType, NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.start.entities import StartNodeData
-from core.workflow.variables.input_entities import VariableEntityType
 
 
 class StartNode(Node[StartNodeData]):

api/core/workflow/nodes/tool/tool_node.py

@@ -11,6 +11,8 @@ from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 from core.tools.errors import ToolInvokeError
 from core.tools.tool_engine import ToolEngine
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
+from core.variables.segments import ArrayAnySegment, ArrayFileSegment
+from core.variables.variables import ArrayAnyVariable
 from core.workflow.enums import (
     NodeType,
     SystemVariableKey,
@@ -21,8 +23,6 @@ from core.workflow.file import File, FileTransferMethod
 from core.workflow.node_events import NodeEventBase, NodeRunResult, StreamChunkEvent, StreamCompletedEvent
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
-from core.workflow.variables.segments import ArrayAnySegment, ArrayFileSegment
-from core.workflow.variables.variables import ArrayAnyVariable
 from extensions.ext_database import db
 from factories import file_factory
 from models import ToolFile

api/core/workflow/nodes/trigger_webhook/node.py

@@ -2,14 +2,14 @@ import logging
 from collections.abc import Mapping
 from typing import Any
 
+from core.variables.types import SegmentType
+from core.variables.variables import FileVariable
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionStatus
 from core.workflow.enums import NodeExecutionType, NodeType
 from core.workflow.file import FileTransferMethod
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
-from core.workflow.variables.types import SegmentType
-from core.workflow.variables.variables import FileVariable
 from factories import file_factory
 from factories.variable_factory import build_segment_with_type
 

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

@@ -1,7 +1,7 @@
 from pydantic import BaseModel
 
+from core.variables.types import SegmentType
 from core.workflow.nodes.base import BaseNodeData
-from core.workflow.variables.types import SegmentType
 
 
 class AdvancedSettings(BaseModel):

api/core/workflow/nodes/variable_aggregator/variable_aggregator_node.py

@@ -1,10 +1,10 @@
 from collections.abc import Mapping
 
+from core.variables.segments import Segment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.variable_aggregator.entities import VariableAggregatorNodeData
-from core.workflow.variables.segments import Segment
 
 
 class VariableAggregatorNode(Node[VariableAggregatorNodeData]):

api/core/workflow/nodes/variable_assigner/common/helpers.py

@@ -3,9 +3,9 @@ from typing import Any, TypeVar
 
 from pydantic import BaseModel
 
-from core.workflow.variables import Segment
-from core.workflow.variables.consts import SELECTORS_LENGTH
-from core.workflow.variables.types import SegmentType
+from core.variables import Segment
+from core.variables.consts import SELECTORS_LENGTH
+from core.variables.types import SegmentType
 
 # Use double underscore (`__`) prefix for internal variables
 # to minimize risk of collision with user-defined variable names.

api/core/workflow/nodes/variable_assigner/v1/node.py

@@ -1,6 +1,7 @@
 from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+from core.variables import SegmentType, VariableBase
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.entities import GraphInitParams
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -8,7 +9,6 @@ from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.variable_assigner.common import helpers as common_helpers
 from core.workflow.nodes.variable_assigner.common.exc import VariableOperatorNodeError
-from core.workflow.variables import SegmentType, VariableBase
 
 from .node_data import VariableAssignerData, WriteMode
 

api/core/workflow/nodes/variable_assigner/v2/helpers.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from core.workflow.variables import SegmentType
+from core.variables import SegmentType
 
 from .enums import Operation
 

api/core/workflow/nodes/variable_assigner/v2/node.py

@@ -2,14 +2,14 @@ import json
 from collections.abc import Mapping, MutableMapping, Sequence
 from typing import TYPE_CHECKING, Any
 
+from core.variables import SegmentType, VariableBase
+from core.variables.consts import SELECTORS_LENGTH
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.variable_assigner.common import helpers as common_helpers
 from core.workflow.nodes.variable_assigner.common.exc import VariableOperatorNodeError
-from core.workflow.variables import SegmentType, VariableBase
-from core.workflow.variables.consts import SELECTORS_LENGTH
 
 from . import helpers
 from .entities import VariableAssignerNodeData, VariableOperationItem

api/core/workflow/runtime/graph_runtime_state.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import importlib
 import json
+import threading
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
@@ -218,6 +219,8 @@ class GraphRuntimeState:
         self._pending_graph_node_states: dict[str, NodeState] | None = None
         self._pending_graph_edge_states: dict[str, NodeState] | None = None
 
+        self.stop_event: threading.Event = threading.Event()
+
         if graph is not None:
             self.attach_graph(graph)
 

api/core/workflow/runtime/graph_runtime_state_protocol.py

@@ -2,8 +2,8 @@ from collections.abc import Mapping, Sequence
 from typing import Any, Protocol
 
 from core.model_runtime.entities.llm_entities import LLMUsage
+from core.variables.segments import Segment
 from core.workflow.system_variable import SystemVariableReadOnlyView
-from core.workflow.variables.segments import Segment
 
 
 class ReadOnlyVariablePool(Protocol):