feat: 合并dify1.1.3版本

# Conflicts:
#	README.md
#	api/.env.example
#	api/controllers/console/__init__.py
#	api/controllers/console/apikey.py
#	api/controllers/console/explore/completion.py
#	api/controllers/console/explore/workflow.py
#	api/controllers/service_api/app/workflow.py
#	api/controllers/service_api/wraps.py
#	api/controllers/web/workflow.py
#	api/core/model_runtime/model_providers/bedrock/get_bedrock_client.py
#	api/core/model_runtime/model_providers/bedrock/llm/llm.py
#	api/core/model_runtime/model_providers/openai_api_compatible/openai_api_compatible.yaml
#	api/core/model_runtime/model_providers/openai_api_compatible/text_embedding/text_embedding.py
#	api/models/model.py
#	api/poetry.lock
#	api/pyproject.toml
#	web/.env.example
#	web/Dockerfile
#	web/app/(commonLayout)/app/(appDetailLayout)/[appId]/layout.tsx
#	web/app/components/app/overview/appCard.tsx
#	web/app/components/base/chat/chat-with-history/chat-wrapper.tsx
#	web/app/components/base/chat/embedded-chatbot/index.tsx
#	web/app/components/base/mermaid/index.tsx
#	web/app/components/develop/index.tsx
#	web/app/components/develop/secret-key/secret-key-modal.tsx
#	web/app/components/explore/app-list/index.tsx
#	web/app/components/explore/item-operation/index.tsx
#	web/app/components/explore/sidebar/app-nav-item/index.tsx
#	web/app/components/explore/sidebar/index.tsx
#	web/app/components/header/account-setting/index.tsx
#	web/app/components/header/index.tsx
#	web/app/components/share/text-generation/index.tsx
#	web/app/components/tools/provider/detail.tsx
#	web/app/layout.tsx
#	web/package.json
#	web/service/base.ts
#	web/yarn.lock

api/core/workflow/callbacks/workflow_logging_callback.py

@@ -11,6 +11,10 @@ from core.workflow.graph_engine.entities.event import (
     IterationRunNextEvent,
     IterationRunStartedEvent,
     IterationRunSucceededEvent,
+    LoopRunFailedEvent,
+    LoopRunNextEvent,
+    LoopRunStartedEvent,
+    LoopRunSucceededEvent,
     NodeRunFailedEvent,
     NodeRunStartedEvent,
     NodeRunStreamChunkEvent,
@@ -62,6 +66,12 @@ class WorkflowLoggingCallback(WorkflowCallback):
             self.on_workflow_iteration_next(event=event)
         elif isinstance(event, IterationRunSucceededEvent | IterationRunFailedEvent):
             self.on_workflow_iteration_completed(event=event)
+        elif isinstance(event, LoopRunStartedEvent):
+            self.on_workflow_loop_started(event=event)
+        elif isinstance(event, LoopRunNextEvent):
+            self.on_workflow_loop_next(event=event)
+        elif isinstance(event, LoopRunSucceededEvent | LoopRunFailedEvent):
+            self.on_workflow_loop_completed(event=event)
         else:
             self.print_text(f"\n[{event.__class__.__name__}]", color="blue")
 
@@ -160,6 +170,8 @@ class WorkflowLoggingCallback(WorkflowCallback):
         self.print_text(f"Branch ID: {event.parallel_start_node_id}", color="blue")
         if event.in_iteration_id:
             self.print_text(f"Iteration ID: {event.in_iteration_id}", color="blue")
+        if event.in_loop_id:
+            self.print_text(f"Loop ID: {event.in_loop_id}", color="blue")
 
     def on_workflow_parallel_completed(
         self, event: ParallelBranchRunSucceededEvent | ParallelBranchRunFailedEvent
@@ -182,6 +194,8 @@ class WorkflowLoggingCallback(WorkflowCallback):
         self.print_text(f"Branch ID: {event.parallel_start_node_id}", color=color)
         if event.in_iteration_id:
             self.print_text(f"Iteration ID: {event.in_iteration_id}", color=color)
+        if event.in_loop_id:
+            self.print_text(f"Loop ID: {event.in_loop_id}", color=color)
 
         if isinstance(event, ParallelBranchRunFailedEvent):
             self.print_text(f"Error: {event.error}", color=color)
@@ -213,6 +227,31 @@ class WorkflowLoggingCallback(WorkflowCallback):
         )
         self.print_text(f"Node ID: {event.iteration_id}", color="blue")
 
+    def on_workflow_loop_started(self, event: LoopRunStartedEvent) -> None:
+        """
+        Publish loop started
+        """
+        self.print_text("\n[LoopRunStartedEvent]", color="blue")
+        self.print_text(f"Loop Node ID: {event.loop_id}", color="blue")
+
+    def on_workflow_loop_next(self, event: LoopRunNextEvent) -> None:
+        """
+        Publish loop next
+        """
+        self.print_text("\n[LoopRunNextEvent]", color="blue")
+        self.print_text(f"Loop Node ID: {event.loop_id}", color="blue")
+        self.print_text(f"Loop Index: {event.index}", color="blue")
+
+    def on_workflow_loop_completed(self, event: LoopRunSucceededEvent | LoopRunFailedEvent) -> None:
+        """
+        Publish loop completed
+        """
+        self.print_text(
+            "\n[LoopRunSucceededEvent]" if isinstance(event, LoopRunSucceededEvent) else "\n[LoopRunFailedEvent]",
+            color="blue",
+        )
+        self.print_text(f"Node ID: {event.loop_id}", color="blue")
+
     def print_text(self, text: str, color: Optional[str] = None, end: str = "\n") -> None:
         """Print text with highlighting and no end characters."""
         text_to_print = self._get_colored_text(text, color) if color else text

api/core/workflow/entities/node_entities.py

@@ -17,14 +17,18 @@ class NodeRunMetadataKey(StrEnum):
     TOTAL_PRICE = "total_price"
     CURRENCY = "currency"
     TOOL_INFO = "tool_info"
+    AGENT_LOG = "agent_log"
     ITERATION_ID = "iteration_id"
     ITERATION_INDEX = "iteration_index"
+    LOOP_ID = "loop_id"
+    LOOP_INDEX = "loop_index"
     PARALLEL_ID = "parallel_id"
     PARALLEL_START_NODE_ID = "parallel_start_node_id"
     PARENT_PARALLEL_ID = "parent_parallel_id"
     PARENT_PARALLEL_START_NODE_ID = "parent_parallel_start_node_id"
     PARALLEL_MODE_RUN_ID = "parallel_mode_run_id"
     ITERATION_DURATION_MAP = "iteration_duration_map"  # single iteration duration if iteration node runs
+    LOOP_DURATION_MAP = "loop_duration_map"  # single loop duration if loop node runs
     ERROR_STRATEGY = "error_strategy"  # node in continue on error mode return the field
 
 
@@ -48,3 +52,8 @@ class NodeRunResult(BaseModel):
 
     # single step node run retry
     retry_index: int = 0
+
+
+class AgentNodeStrategyInit(BaseModel):
+    name: str
+    icon: str | None = None

api/core/workflow/entities/variable_pool.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from core.file import File, FileAttribute, file_manager
 from core.variables import Segment, SegmentGroup, Variable
-from core.variables.segments import FileSegment
+from core.variables.segments import FileSegment, NoneSegment
 from factories import variable_factory
 
 from ..constants import CONVERSATION_VARIABLE_NODE_ID, ENVIRONMENT_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
@@ -15,7 +15,6 @@ from ..enums import SystemVariableKey
 
 VariableValue = Union[str, int, float, dict, list, File]
 
-
 VARIABLE_PATTERN = re.compile(r"\{\{#([a-zA-Z0-9_]{1,50}(?:\.[a-zA-Z_][a-zA-Z0-9_]{0,29}){1,10})#\}\}")
 
 
@@ -131,11 +130,13 @@ class VariablePool(BaseModel):
             if attr not in {item.value for item in FileAttribute}:
                 return None
             value = self.get(selector)
-            if not isinstance(value, FileSegment):
+            if not isinstance(value, FileSegment | NoneSegment):
                 return None
-            attr = FileAttribute(attr)
-            attr_value = file_manager.get_attr(file=value.value, attr=attr)
-            return variable_factory.build_segment(attr_value)
+            if isinstance(value, FileSegment):
+                attr = FileAttribute(attr)
+                attr_value = file_manager.get_attr(file=value.value, attr=attr)
+                return variable_factory.build_segment(attr_value)
+            return value
 
         return value
 

api/core/workflow/entities/workflow_entities.py

@@ -3,7 +3,7 @@ from typing import Optional
 from pydantic import BaseModel
 
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.workflow.nodes.base import BaseIterationState, BaseNode
+from core.workflow.nodes.base import BaseIterationState, BaseLoopState, BaseNode
 from models.enums import UserFrom
 from models.workflow import Workflow, WorkflowType
 
@@ -41,11 +41,13 @@ class WorkflowRunState:
     class NodeRun(BaseModel):
         node_id: str
         iteration_node_id: str
+        loop_node_id: str
 
     workflow_node_runs: list[NodeRun]
     workflow_node_steps: int
 
     current_iteration_state: Optional[BaseIterationState]
+    current_loop_state: Optional[BaseLoopState]
 
     def __init__(
         self,
@@ -74,3 +76,4 @@ class WorkflowRunState:
         self.workflow_node_steps = 1
         self.workflow_node_runs = []
         self.current_iteration_state = None
+        self.current_loop_state = None

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

@@ -4,6 +4,7 @@ from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
+from core.workflow.entities.node_entities import AgentNodeStrategyInit
 from core.workflow.graph_engine.entities.runtime_route_state import RouteNodeState
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.base import BaseNodeData
@@ -62,12 +63,16 @@ class BaseNodeEvent(GraphEngineEvent):
     """parent parallel start node id if node is in parallel"""
     in_iteration_id: Optional[str] = None
     """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
 
 
 class NodeRunStartedEvent(BaseNodeEvent):
     predecessor_node_id: Optional[str] = None
-    parallel_mode_run_id: Optional[str] = None
     """predecessor node id"""
+    parallel_mode_run_id: Optional[str] = None
+    """iteration node parallel mode run id"""
+    agent_strategy: Optional[AgentNodeStrategyInit] = None
 
 
 class NodeRunStreamChunkEvent(BaseNodeEvent):
@@ -97,6 +102,10 @@ class NodeInIterationFailedEvent(BaseNodeEvent):
     error: str = Field(..., description="error")
 
 
+class NodeInLoopFailedEvent(BaseNodeEvent):
+    error: str = Field(..., description="error")
+
+
 class NodeRunRetryEvent(NodeRunStartedEvent):
     error: str = Field(..., description="error")
     retry_index: int = Field(..., description="which retry attempt is about to be performed")
@@ -119,6 +128,8 @@ class BaseParallelBranchEvent(GraphEngineEvent):
     """parent parallel start node id if node is in parallel"""
     in_iteration_id: Optional[str] = None
     """iteration id if node is in iteration"""
+    in_loop_id: Optional[str] = None
+    """loop id if node is in loop"""
 
 
 class ParallelBranchRunStartedEvent(BaseParallelBranchEvent):
@@ -164,8 +175,8 @@ class IterationRunStartedEvent(BaseIterationEvent):
 
 class IterationRunNextEvent(BaseIterationEvent):
     index: int = Field(..., description="index")
-    pre_iteration_output: Optional[Any] = Field(None, description="pre iteration output")
-    duration: Optional[float] = Field(None, description="duration")
+    pre_iteration_output: Optional[Any] = None
+    duration: Optional[float] = None
 
 
 class IterationRunSucceededEvent(BaseIterationEvent):
@@ -186,4 +197,78 @@ class IterationRunFailedEvent(BaseIterationEvent):
     error: str = Field(..., description="failed reason")
 
 
-InNodeEvent = BaseNodeEvent | BaseParallelBranchEvent | BaseIterationEvent
+###########################################
+# Loop Events
+###########################################
+
+
+class BaseLoopEvent(GraphEngineEvent):
+    loop_id: str = Field(..., description="loop node execution id")
+    loop_node_id: str = Field(..., description="loop node id")
+    loop_node_type: NodeType = Field(..., description="node type, loop or loop")
+    loop_node_data: BaseNodeData = Field(..., description="node data")
+    parallel_id: Optional[str] = None
+    """parallel id if node is in parallel"""
+    parallel_start_node_id: Optional[str] = None
+    """parallel start node id if node is in parallel"""
+    parent_parallel_id: Optional[str] = None
+    """parent parallel id if node is in parallel"""
+    parent_parallel_start_node_id: Optional[str] = None
+    """parent parallel start node id if node is in parallel"""
+    parallel_mode_run_id: Optional[str] = None
+    """loop run in parallel mode run id"""
+
+
+class LoopRunStartedEvent(BaseLoopEvent):
+    start_at: datetime = Field(..., description="start at")
+    inputs: Optional[Mapping[str, Any]] = None
+    metadata: Optional[Mapping[str, Any]] = None
+    predecessor_node_id: Optional[str] = None
+
+
+class LoopRunNextEvent(BaseLoopEvent):
+    index: int = Field(..., description="index")
+    pre_loop_output: Optional[Any] = None
+    duration: Optional[float] = None
+
+
+class LoopRunSucceededEvent(BaseLoopEvent):
+    start_at: datetime = Field(..., description="start at")
+    inputs: Optional[Mapping[str, Any]] = None
+    outputs: Optional[Mapping[str, Any]] = None
+    metadata: Optional[Mapping[str, Any]] = None
+    steps: int = 0
+    loop_duration_map: Optional[dict[str, float]] = None
+
+
+class LoopRunFailedEvent(BaseLoopEvent):
+    start_at: datetime = Field(..., description="start at")
+    inputs: Optional[Mapping[str, Any]] = None
+    outputs: Optional[Mapping[str, Any]] = None
+    metadata: Optional[Mapping[str, Any]] = None
+    steps: int = 0
+    error: str = Field(..., description="failed reason")
+
+
+###########################################
+# Agent Events
+###########################################
+
+
+class BaseAgentEvent(GraphEngineEvent):
+    pass
+
+
+class AgentLogEvent(BaseAgentEvent):
+    id: str = Field(..., description="id")
+    label: str = Field(..., description="label")
+    node_execution_id: str = Field(..., description="node execution id")
+    parent_id: str | None = Field(..., description="parent id")
+    error: str | None = Field(..., description="error")
+    status: str = Field(..., description="status")
+    data: Mapping[str, Any] = Field(..., description="data")
+    metadata: Optional[Mapping[str, Any]] = Field(default=None, description="metadata")
+    node_id: str = Field(..., description="agent node id")
+
+
+InNodeEvent = BaseNodeEvent | BaseParallelBranchEvent | BaseIterationEvent | BaseAgentEvent | BaseLoopEvent

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

@@ -590,8 +590,6 @@ class Graph(BaseModel):
                             start_node_id=node_id,
                             routes_node_ids=routes_node_ids,
                         )
-                        # Exclude conditional branch nodes
-                        and all(edge.run_condition is None for edge in reverse_edge_mapping.get(node_id, []))
                     ):
                         if node_id not in merge_branch_node_ids:
                             merge_branch_node_ids[node_id] = []

api/core/workflow/graph_engine/graph_engine.py

@@ -1,3 +1,4 @@
+import contextvars
 import logging
 import queue
 import time
@@ -13,11 +14,13 @@ from flask import Flask, current_app
 from configs import dify_config
 from core.app.apps.base_app_queue_manager import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.workflow.entities.node_entities import NodeRunMetadataKey, NodeRunResult
+from core.workflow.entities.node_entities import AgentNodeStrategyInit, NodeRunMetadataKey, NodeRunResult
 from core.workflow.entities.variable_pool import VariablePool, VariableValue
 from core.workflow.graph_engine.condition_handlers.condition_manager import ConditionManager
 from core.workflow.graph_engine.entities.event import (
+    BaseAgentEvent,
     BaseIterationEvent,
+    BaseLoopEvent,
     GraphEngineEvent,
     GraphRunFailedEvent,
     GraphRunPartialSucceededEvent,
@@ -39,6 +42,8 @@ from core.workflow.graph_engine.entities.graph_init_params import GraphInitParam
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.graph_engine.entities.runtime_route_state import RouteNodeState
 from core.workflow.nodes import NodeType
+from core.workflow.nodes.agent.agent_node import AgentNode
+from core.workflow.nodes.agent.entities import AgentNodeData
 from core.workflow.nodes.answer.answer_stream_processor import AnswerStreamProcessor
 from core.workflow.nodes.answer.base_stream_processor import StreamProcessor
 from core.workflow.nodes.base import BaseNode
@@ -477,6 +482,7 @@ class GraphEngine:
                 **{
                     "flask_app": current_app._get_current_object(),  # type: ignore[attr-defined]
                     "q": q,
+                    "context": contextvars.copy_context(),
                     "parallel_id": parallel_id,
                     "parallel_start_node_id": edge.target_node_id,
                     "parent_parallel_id": in_parallel_id,
@@ -497,7 +503,7 @@ class GraphEngine:
                     break
 
                 yield event
-                if event.parallel_id == parallel_id:
+                if not isinstance(event, BaseAgentEvent) and event.parallel_id == parallel_id:
                     if isinstance(event, ParallelBranchRunSucceededEvent):
                         succeeded_count += 1
                         if succeeded_count == len(futures):
@@ -520,6 +526,7 @@ class GraphEngine:
     def _run_parallel_node(
         self,
         flask_app: Flask,
+        context: contextvars.Context,
         q: queue.Queue,
         parallel_id: str,
         parallel_start_node_id: str,
@@ -530,6 +537,9 @@ class GraphEngine:
         """
         Run parallel nodes
         """
+        for var, val in context.items():
+            var.set(val)
+
         with flask_app.app_context():
             try:
                 q.put(
@@ -600,6 +610,14 @@ class GraphEngine:
         Run node
         """
         # trigger node run start event
+        agent_strategy = (
+            AgentNodeStrategyInit(
+                name=cast(AgentNodeData, node_instance.node_data).agent_strategy_name,
+                icon=cast(AgentNode, node_instance).agent_strategy_icon,
+            )
+            if node_instance.node_type == NodeType.AGENT
+            else None
+        )
         yield NodeRunStartedEvent(
             id=node_instance.id,
             node_id=node_instance.node_id,
@@ -611,6 +629,7 @@ class GraphEngine:
             parallel_start_node_id=parallel_start_node_id,
             parent_parallel_id=parent_parallel_id,
             parent_parallel_start_node_id=parent_parallel_start_node_id,
+            agent_strategy=agent_strategy,
         )
 
         db.session.close()
@@ -631,6 +650,12 @@ class GraphEngine:
                             item.parallel_start_node_id = parallel_start_node_id
                             item.parent_parallel_id = parent_parallel_id
                             item.parent_parallel_start_node_id = parent_parallel_start_node_id
+                        elif isinstance(item, BaseLoopEvent):
+                            # add parallel info to loop event
+                            item.parallel_id = parallel_id
+                            item.parallel_start_node_id = parallel_start_node_id
+                            item.parent_parallel_id = parent_parallel_id
+                            item.parent_parallel_start_node_id = parent_parallel_start_node_id
 
                         yield item
                     else:
@@ -648,7 +673,7 @@ class GraphEngine:
                                     retries += 1
                                     route_node_state.node_run_result = run_result
                                     yield NodeRunRetryEvent(
-                                        id=node_instance.id,
+                                        id=str(uuid.uuid4()),
                                         node_id=node_instance.node_id,
                                         node_type=node_instance.node_type,
                                         node_data=node_instance.node_data,
@@ -663,7 +688,7 @@ class GraphEngine:
                                         start_at=retry_start_at,
                                     )
                                     time.sleep(retry_interval)
-                                    continue
+                                    break
                             route_node_state.set_finished(run_result=run_result)
 
                             if run_result.status == WorkflowNodeExecutionStatus.FAILED:
@@ -713,8 +738,10 @@ class GraphEngine:
                                     )
                                 should_continue_retry = False
                             elif run_result.status == WorkflowNodeExecutionStatus.SUCCEEDED:
-                                if node_instance.should_continue_on_error and self.graph.edge_mapping.get(
-                                    node_instance.node_id
+                                if (
+                                    node_instance.should_continue_on_error
+                                    and self.graph.edge_mapping.get(node_instance.node_id)
+                                    and node_instance.node_data.error_strategy is ErrorStrategy.FAIL_BRANCH
                                 ):
                                     run_result.edge_source_handle = FailBranchSourceHandle.SUCCESS
                                 if run_result.metadata and run_result.metadata.get(NodeRunMetadataKey.TOTAL_TOKENS):
@@ -848,11 +875,12 @@ class GraphEngine:
     def create_copy(self):
         """
         create a graph engine copy
-        :return: with a new variable pool instance of graph engine
+        :return: graph engine with a new variable pool and initialized total tokens
         """
         new_instance = copy(self)
         new_instance.graph_runtime_state = copy(self.graph_runtime_state)
         new_instance.graph_runtime_state.variable_pool = deepcopy(self.graph_runtime_state.variable_pool)
+        new_instance.graph_runtime_state.total_tokens = 0
         return new_instance
 
     def _handle_continue_on_error(

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

@@ -0,0 +1,3 @@
+from .agent_node import AgentNode
+
+__all__ = ["AgentNode"]

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

@@ -0,0 +1,299 @@
+import json
+from collections.abc import Generator, Mapping, Sequence
+from typing import Any, cast
+
+from core.agent.entities import AgentToolEntity
+from core.agent.plugin_entities import AgentStrategyParameter
+from core.model_manager import ModelManager
+from core.model_runtime.entities.model_entities import ModelType
+from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.manager.plugin import PluginInstallationManager
+from core.tools.entities.tool_entities import ToolParameter, ToolProviderType
+from core.tools.tool_manager import ToolManager
+from core.workflow.entities.node_entities import NodeRunResult
+from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.enums import SystemVariableKey
+from core.workflow.nodes.agent.entities import AgentNodeData, ParamsAutoGenerated
+from core.workflow.nodes.base.entities import BaseNodeData
+from core.workflow.nodes.enums import NodeType
+from core.workflow.nodes.event.event import RunCompletedEvent
+from core.workflow.nodes.tool.tool_node import ToolNode
+from core.workflow.utils.variable_template_parser import VariableTemplateParser
+from factories.agent_factory import get_plugin_agent_strategy
+from models.workflow import WorkflowNodeExecutionStatus
+
+
+class AgentNode(ToolNode):
+    """
+    Agent Node
+    """
+
+    _node_data_cls = AgentNodeData  # type: ignore
+    _node_type = NodeType.AGENT
+
+    def _run(self) -> Generator:
+        """
+        Run the agent node
+        """
+        node_data = cast(AgentNodeData, self.node_data)
+
+        try:
+            strategy = get_plugin_agent_strategy(
+                tenant_id=self.tenant_id,
+                agent_strategy_provider_name=node_data.agent_strategy_provider_name,
+                agent_strategy_name=node_data.agent_strategy_name,
+            )
+        except Exception as e:
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs={},
+                    error=f"Failed to get agent strategy: {str(e)}",
+                )
+            )
+            return
+
+        agent_parameters = strategy.get_parameters()
+
+        # get parameters
+        parameters = self._generate_agent_parameters(
+            agent_parameters=agent_parameters,
+            variable_pool=self.graph_runtime_state.variable_pool,
+            node_data=node_data,
+        )
+        parameters_for_log = self._generate_agent_parameters(
+            agent_parameters=agent_parameters,
+            variable_pool=self.graph_runtime_state.variable_pool,
+            node_data=node_data,
+            for_log=True,
+        )
+
+        # get conversation id
+        conversation_id = self.graph_runtime_state.variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
+
+        try:
+            message_stream = strategy.invoke(
+                params=parameters,
+                user_id=self.user_id,
+                app_id=self.app_id,
+                conversation_id=conversation_id.text if conversation_id else None,
+            )
+        except Exception as e:
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs=parameters_for_log,
+                    error=f"Failed to invoke agent: {str(e)}",
+                )
+            )
+            return
+
+        try:
+            # convert tool messages
+
+            yield from self._transform_message(
+                message_stream,
+                {
+                    "icon": self.agent_strategy_icon,
+                    "agent_strategy": cast(AgentNodeData, self.node_data).agent_strategy_name,
+                },
+                parameters_for_log,
+            )
+        except PluginDaemonClientSideError as e:
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs=parameters_for_log,
+                    error=f"Failed to transform agent message: {str(e)}",
+                )
+            )
+
+    def _generate_agent_parameters(
+        self,
+        *,
+        agent_parameters: Sequence[AgentStrategyParameter],
+        variable_pool: VariablePool,
+        node_data: AgentNodeData,
+        for_log: bool = False,
+    ) -> dict[str, Any]:
+        """
+        Generate parameters based on the given tool parameters, variable pool, and node data.
+
+        Args:
+            agent_parameters (Sequence[AgentParameter]): The list of agent parameters.
+            variable_pool (VariablePool): The variable pool containing the variables.
+            node_data (AgentNodeData): The data associated with the agent node.
+
+        Returns:
+            Mapping[str, Any]: A dictionary containing the generated parameters.
+
+        """
+        agent_parameters_dictionary = {parameter.name: parameter for parameter in agent_parameters}
+
+        result: dict[str, Any] = {}
+        for parameter_name in node_data.agent_parameters:
+            parameter = agent_parameters_dictionary.get(parameter_name)
+            if not parameter:
+                result[parameter_name] = None
+                continue
+            agent_input = node_data.agent_parameters[parameter_name]
+            if agent_input.type == "variable":
+                variable = variable_pool.get(agent_input.value)  # type: ignore
+                if variable is None:
+                    raise ValueError(f"Variable {agent_input.value} does not exist")
+                parameter_value = variable.value
+            elif agent_input.type in {"mixed", "constant"}:
+                # variable_pool.convert_template expects a string template,
+                # but if passing a dict, convert to JSON string first before rendering
+                try:
+                    parameter_value = json.dumps(agent_input.value, ensure_ascii=False)
+                except TypeError:
+                    parameter_value = str(agent_input.value)
+                segment_group = variable_pool.convert_template(parameter_value)
+                parameter_value = segment_group.log if for_log else segment_group.text
+                # variable_pool.convert_template returns a string,
+                # so we need to convert it back to a dictionary
+                try:
+                    parameter_value = json.loads(parameter_value)
+                except json.JSONDecodeError:
+                    parameter_value = parameter_value
+            else:
+                raise ValueError(f"Unknown agent input type '{agent_input.type}'")
+            value = parameter_value
+            if parameter.type == "array[tools]":
+                value = cast(list[dict[str, Any]], value)
+                value = [tool for tool in value if tool.get("enabled", False)]
+
+                for tool in value:
+                    if "schemas" in tool:
+                        tool.pop("schemas")
+                    parameters = tool.get("parameters", {})
+                    if all(isinstance(v, dict) for _, v in parameters.items()):
+                        params = {}
+                        for key, param in parameters.items():
+                            if param.get("auto", ParamsAutoGenerated.OPEN.value) == ParamsAutoGenerated.CLOSE.value:
+                                value_param = param.get("value", {})
+                                params[key] = value_param.get("value", "") if value_param is not None else None
+                            else:
+                                params[key] = None
+                        parameters = params
+                    tool["settings"] = {k: v.get("value", None) for k, v in tool.get("settings", {}).items()}
+                    tool["parameters"] = parameters
+
+            if not for_log:
+                if parameter.type == "array[tools]":
+                    value = cast(list[dict[str, Any]], value)
+                    tool_value = []
+                    for tool in value:
+                        provider_type = ToolProviderType(tool.get("type", ToolProviderType.BUILT_IN.value))
+                        setting_params = tool.get("settings", {})
+                        parameters = tool.get("parameters", {})
+                        manual_input_params = [key for key, value in parameters.items() if value is not None]
+
+                        parameters = {**parameters, **setting_params}
+                        entity = AgentToolEntity(
+                            provider_id=tool.get("provider_name", ""),
+                            provider_type=provider_type,
+                            tool_name=tool.get("tool_name", ""),
+                            tool_parameters=parameters,
+                            plugin_unique_identifier=tool.get("plugin_unique_identifier", None),
+                        )
+
+                        extra = tool.get("extra", {})
+
+                        tool_runtime = ToolManager.get_agent_tool_runtime(
+                            self.tenant_id, self.app_id, entity, self.invoke_from
+                        )
+                        if tool_runtime.entity.description:
+                            tool_runtime.entity.description.llm = (
+                                extra.get("descrption", "") or tool_runtime.entity.description.llm
+                            )
+                        for tool_runtime_params in tool_runtime.entity.parameters:
+                            tool_runtime_params.form = (
+                                ToolParameter.ToolParameterForm.FORM
+                                if tool_runtime_params.name in manual_input_params
+                                else tool_runtime_params.form
+                            )
+                        manual_input_value = {}
+                        if tool_runtime.entity.parameters:
+                            manual_input_value = {
+                                key: value for key, value in parameters.items() if key in manual_input_params
+                            }
+                        runtime_parameters = {
+                            **tool_runtime.runtime.runtime_parameters,
+                            **manual_input_value,
+                        }
+                        tool_value.append(
+                            {
+                                **tool_runtime.entity.model_dump(mode="json"),
+                                "runtime_parameters": runtime_parameters,
+                                "provider_type": provider_type.value,
+                            }
+                        )
+                    value = tool_value
+                if parameter.type == "model-selector":
+                    value = cast(dict[str, Any], value)
+                    model_instance = ModelManager().get_model_instance(
+                        tenant_id=self.tenant_id,
+                        provider=value.get("provider", ""),
+                        model_type=ModelType(value.get("model_type", "")),
+                        model=value.get("model", ""),
+                    )
+                    models = model_instance.model_type_instance.plugin_model_provider.declaration.models
+                    finded_model = next((model for model in models if model.model == value.get("model", "")), None)
+
+                    value["entity"] = finded_model.model_dump(mode="json") if finded_model else None
+
+            result[parameter_name] = value
+
+        return result
+
+    @classmethod
+    def _extract_variable_selector_to_variable_mapping(
+        cls,
+        *,
+        graph_config: Mapping[str, Any],
+        node_id: str,
+        node_data: BaseNodeData,
+    ) -> Mapping[str, Sequence[str]]:
+        """
+        Extract variable selector to variable mapping
+        :param graph_config: graph config
+        :param node_id: node id
+        :param node_data: node data
+        :return:
+        """
+        node_data = cast(AgentNodeData, node_data)
+        result: dict[str, Any] = {}
+        for parameter_name in node_data.agent_parameters:
+            input = node_data.agent_parameters[parameter_name]
+            if input.type in ["mixed", "constant"]:
+                selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
+                for selector in selectors:
+                    result[selector.variable] = selector.value_selector
+            elif input.type == "variable":
+                result[parameter_name] = input.value
+
+        result = {node_id + "." + key: value for key, value in result.items()}
+
+        return result
+
+    @property
+    def agent_strategy_icon(self) -> str | None:
+        """
+        Get agent strategy icon
+        :return:
+        """
+        manager = PluginInstallationManager()
+        plugins = manager.list_plugins(self.tenant_id)
+        try:
+            current_plugin = next(
+                plugin
+                for plugin in plugins
+                if f"{plugin.plugin_id}/{plugin.name}"
+                == cast(AgentNodeData, self.node_data).agent_strategy_provider_name
+            )
+            icon = current_plugin.declaration.icon
+        except StopIteration:
+            icon = None
+        return icon

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

@@ -0,0 +1,24 @@
+from enum import Enum
+from typing import Any, Literal, Union
+
+from pydantic import BaseModel
+
+from core.tools.entities.tool_entities import ToolSelector
+from core.workflow.nodes.base.entities import BaseNodeData
+
+
+class AgentNodeData(BaseNodeData):
+    agent_strategy_provider_name: str  # redundancy
+    agent_strategy_name: str
+    agent_strategy_label: str  # redundancy
+
+    class AgentInput(BaseModel):
+        value: Union[list[str], list[ToolSelector], Any]
+        type: Literal["mixed", "variable", "constant"]
+
+    agent_parameters: dict[str, AgentInput]
+
+
+class ParamsAutoGenerated(Enum):
+    CLOSE = 0
+    OPEN = 1

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

@@ -158,6 +158,7 @@ class AnswerStreamGeneratorRouter:
                     NodeType.IF_ELSE,
                     NodeType.QUESTION_CLASSIFIER,
                     NodeType.ITERATION,
+                    NodeType.LOOP,
                     NodeType.VARIABLE_ASSIGNER,
                 }
                 or source_node_data.get("error_strategy") == ErrorStrategy.FAIL_BRANCH

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

@@ -35,7 +35,7 @@ class AnswerStreamProcessor(StreamProcessor):
 
                 yield event
             elif isinstance(event, NodeRunStreamChunkEvent):
-                if event.in_iteration_id:
+                if event.in_iteration_id or event.in_loop_id:
                     yield event
                     continue
 
@@ -82,7 +82,7 @@ class AnswerStreamProcessor(StreamProcessor):
         :param event: node run succeeded event
         :return:
         """
-        for answer_node_id, position in self.route_position.items():
+        for answer_node_id in self.route_position:
             # all depends on answer node id not in rest node ids
             if event.route_node_state.node_id != answer_node_id and (
                 answer_node_id not in self.rest_node_ids
@@ -155,11 +155,13 @@ class AnswerStreamProcessor(StreamProcessor):
         for answer_node_id, route_position in self.route_position.items():
             if answer_node_id not in self.rest_node_ids:
                 continue
-
+            # exclude current node id
+            answer_dependencies = self.generate_routes.answer_dependencies
+            if event.node_id in answer_dependencies[answer_node_id]:
+                answer_dependencies[answer_node_id].remove(event.node_id)
+            answer_dependencies_ids = answer_dependencies.get(answer_node_id, [])
             # all depends on answer node id not in rest node ids
-            if all(
-                dep_id not in self.rest_node_ids for dep_id in self.generate_routes.answer_dependencies[answer_node_id]
-            ):
+            if all(dep_id not in self.rest_node_ids for dep_id in answer_dependencies_ids):
                 if route_position >= len(self.generate_routes.answer_generate_route[answer_node_id]):
                     continue
 

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

@@ -57,11 +57,19 @@ class StreamProcessor(ABC):
 
                     # The branch_identify parameter is added to ensure that
                     # only nodes in the correct logical branch are included.
+                    reachable_node_ids.append(edge.target_node_id)
                     ids = self._fetch_node_ids_in_reachable_branch(edge.target_node_id, run_result.edge_source_handle)
                     reachable_node_ids.extend(ids)
                 else:
+                    # if the condition edge in parallel, and the target node is not in parallel, we should not remove it
+                    # Issues: #13626
+                    if (
+                        finished_node_id in self.graph.node_parallel_mapping
+                        and edge.target_node_id not in self.graph.node_parallel_mapping
+                    ):
+                        continue
                     unreachable_first_node_ids.append(edge.target_node_id)
-
+            unreachable_first_node_ids = list(set(unreachable_first_node_ids) - set(reachable_node_ids))
             for node_id in unreachable_first_node_ids:
                 self._remove_node_ids_in_unreachable_branch(node_id, reachable_node_ids)
 

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

@@ -1,4 +1,11 @@
-from .entities import BaseIterationNodeData, BaseIterationState, BaseNodeData
+from .entities import BaseIterationNodeData, BaseIterationState, BaseLoopNodeData, BaseLoopState, BaseNodeData
 from .node import BaseNode
 
-__all__ = ["BaseIterationNodeData", "BaseIterationState", "BaseNode", "BaseNodeData"]
+__all__ = [
+    "BaseIterationNodeData",
+    "BaseIterationState",
+    "BaseLoopNodeData",
+    "BaseLoopState",
+    "BaseNode",
+    "BaseNodeData",
+]

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

@@ -147,3 +147,18 @@ class BaseIterationState(BaseModel):
         pass
 
     metadata: MetaData
+
+
+class BaseLoopNodeData(BaseNodeData):
+    start_node_id: Optional[str] = None
+
+
+class BaseLoopState(BaseModel):
+    loop_node_id: str
+    index: int
+    inputs: dict
+
+    class MetaData(BaseModel):
+        pass
+
+    metadata: MetaData

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

@@ -22,7 +22,7 @@ GenericNodeData = TypeVar("GenericNodeData", bound=BaseNodeData)
 
 
 class BaseNode(Generic[GenericNodeData]):
-    _node_data_cls: type[BaseNodeData]
+    _node_data_cls: type[GenericNodeData]
     _node_type: NodeType
 
     def __init__(
@@ -57,7 +57,7 @@ class BaseNode(Generic[GenericNodeData]):
         self.node_id = node_id
 
         node_data = self._node_data_cls.model_validate(config.get("data", {}))
-        self.node_data = cast(GenericNodeData, node_data)
+        self.node_data = node_data
 
     @abstractmethod
     def _run(self) -> NodeRunResult | Generator[Union[NodeEvent, "InNodeEvent"], None, None]:

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

@@ -200,7 +200,7 @@ class CodeNode(BaseNode[CodeNodeData]):
             if output_config.type == "object":
                 # check if output is object
                 if not isinstance(result.get(output_name), dict):
-                    if isinstance(result.get(output_name), type(None)):
+                    if result[output_name] is None:
                         transformed_result[output_name] = None
                     else:
                         raise OutputValidationError(
@@ -228,7 +228,7 @@ class CodeNode(BaseNode[CodeNodeData]):
             elif output_config.type == "array[number]":
                 # check if array of number available
                 if not isinstance(result[output_name], list):
-                    if isinstance(result[output_name], type(None)):
+                    if result[output_name] is None:
                         transformed_result[output_name] = None
                     else:
                         raise OutputValidationError(
@@ -249,7 +249,7 @@ class CodeNode(BaseNode[CodeNodeData]):
             elif output_config.type == "array[string]":
                 # check if array of string available
                 if not isinstance(result[output_name], list):
-                    if isinstance(result[output_name], type(None)):
+                    if result[output_name] is None:
                         transformed_result[output_name] = None
                     else:
                         raise OutputValidationError(
@@ -270,7 +270,7 @@ class CodeNode(BaseNode[CodeNodeData]):
             elif output_config.type == "array[object]":
                 # check if array of object available
                 if not isinstance(result[output_name], list):
-                    if isinstance(result[output_name], type(None)):
+                    if result[output_name] is None:
                         transformed_result[output_name] = None
                     else:
                         raise OutputValidationError(

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

@@ -2,7 +2,6 @@ import csv
 import io
 import json
 import logging
-import operator
 import os
 import tempfile
 from collections.abc import Mapping, Sequence
@@ -12,6 +11,9 @@ import docx
 import pandas as pd
 import pypdfium2  # type: ignore
 import yaml  # type: ignore
+from docx.document import Document
+from docx.oxml.table import CT_Tbl
+from docx.oxml.text.paragraph import CT_P
 from docx.table import Table
 from docx.text.paragraph import Paragraph
 
@@ -107,8 +109,10 @@ def _extract_text_by_mime_type(*, file_content: bytes, mime_type: str) -> str:
             return _extract_text_from_plain_text(file_content)
         case "application/pdf":
             return _extract_text_from_pdf(file_content)
-        case "application/vnd.openxmlformats-officedocument.wordprocessingml.document" | "application/msword":
+        case "application/msword":
             return _extract_text_from_doc(file_content)
+        case "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
+            return _extract_text_from_docx(file_content)
         case "text/csv":
             return _extract_text_from_csv(file_content)
         case "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" | "application/vnd.ms-excel":
@@ -142,8 +146,10 @@ def _extract_text_by_file_extension(*, file_content: bytes, file_extension: str)
             return _extract_text_from_yaml(file_content)
         case ".pdf":
             return _extract_text_from_pdf(file_content)
-        case ".doc" | ".docx":
+        case ".doc":
             return _extract_text_from_doc(file_content)
+        case ".docx":
+            return _extract_text_from_docx(file_content)
         case ".csv":
             return _extract_text_from_csv(file_content)
         case ".xls" | ".xlsx":
@@ -203,7 +209,40 @@ def _extract_text_from_pdf(file_content: bytes) -> str:
 
 def _extract_text_from_doc(file_content: bytes) -> str:
     """
-    Extract text from a DOC/DOCX file.
+    Extract text from a DOC file.
+    """
+    from unstructured.partition.api import partition_via_api
+
+    if not (dify_config.UNSTRUCTURED_API_URL and dify_config.UNSTRUCTURED_API_KEY):
+        raise TextExtractionError("UNSTRUCTURED_API_URL and UNSTRUCTURED_API_KEY must be set")
+
+    try:
+        with tempfile.NamedTemporaryFile(suffix=".doc", delete=False) as temp_file:
+            temp_file.write(file_content)
+            temp_file.flush()
+            with open(temp_file.name, "rb") as file:
+                elements = partition_via_api(
+                    file=file,
+                    metadata_filename=temp_file.name,
+                    api_url=dify_config.UNSTRUCTURED_API_URL,
+                    api_key=dify_config.UNSTRUCTURED_API_KEY,
+                )
+            os.unlink(temp_file.name)
+        return "\n".join([getattr(element, "text", "") for element in elements])
+    except Exception as e:
+        raise TextExtractionError(f"Failed to extract text from DOC: {str(e)}") from e
+
+
+def paser_docx_part(block, doc: Document, content_items, i):
+    if isinstance(block, CT_P):
+        content_items.append((i, "paragraph", Paragraph(block, doc)))
+    elif isinstance(block, CT_Tbl):
+        content_items.append((i, "table", Table(block, doc)))
+
+
+def _extract_text_from_docx(file_content: bytes) -> str:
+    """
+    Extract text from a DOCX file.
     For now support only paragraph and table add more if needed
     """
     try:
@@ -214,16 +253,13 @@ def _extract_text_from_doc(file_content: bytes) -> str:
         # Keep track of paragraph and table positions
         content_items: list[tuple[int, str, Table | Paragraph]] = []
 
-        # Process paragraphs and tables
-        for i, paragraph in enumerate(doc.paragraphs):
-            if paragraph.text.strip():
-                content_items.append((i, "paragraph", paragraph))
-
-        for i, table in enumerate(doc.tables):
-            content_items.append((i, "table", table))
-
-        # Sort content items based on their original position
-        content_items.sort(key=operator.itemgetter(0))
+        it = iter(doc.element.body)
+        part = next(it, None)
+        i = 0
+        while part is not None:
+            paser_docx_part(part, doc, content_items, i)
+            i = i + 1
+            part = next(it, None)
 
         # Process sorted content
         for _, item_type, item in content_items:
@@ -255,13 +291,13 @@ def _extract_text_from_doc(file_content: bytes) -> str:
 
                         text.append(markdown_table)
                 except Exception as e:
-                    logger.warning(f"Failed to extract table from DOC/DOCX: {e}")
+                    logger.warning(f"Failed to extract table from DOC: {e}")
                     continue
 
         return "\n".join(text)
 
     except Exception as e:
-        raise TextExtractionError(f"Failed to extract text from DOC/DOCX: {str(e)}") from e
+        raise TextExtractionError(f"Failed to extract text from DOCX: {str(e)}") from e
 
 
 def _download_file_content(file: File) -> bytes:
@@ -329,14 +365,29 @@ def _extract_text_from_excel(file_content: bytes) -> str:
 
 
 def _extract_text_from_ppt(file_content: bytes) -> str:
+    from unstructured.partition.api import partition_via_api
     from unstructured.partition.ppt import partition_ppt
 
     try:
-        with io.BytesIO(file_content) as file:
-            elements = partition_ppt(file=file)
+        if dify_config.UNSTRUCTURED_API_URL and dify_config.UNSTRUCTURED_API_KEY:
+            with tempfile.NamedTemporaryFile(suffix=".ppt", delete=False) as temp_file:
+                temp_file.write(file_content)
+                temp_file.flush()
+                with open(temp_file.name, "rb") as file:
+                    elements = partition_via_api(
+                        file=file,
+                        metadata_filename=temp_file.name,
+                        api_url=dify_config.UNSTRUCTURED_API_URL,
+                        api_key=dify_config.UNSTRUCTURED_API_KEY,
+                    )
+                os.unlink(temp_file.name)
+        else:
+            with io.BytesIO(file_content) as file:
+                elements = partition_ppt(file=file)
         return "\n".join([getattr(element, "text", "") for element in elements])
+
     except Exception as e:
-        raise TextExtractionError(f"Failed to extract text from PPT: {str(e)}") from e
+        raise TextExtractionError(f"Failed to extract text from PPTX: {str(e)}") from e
 
 
 def _extract_text_from_pptx(file_content: bytes) -> str:

api/core/workflow/nodes/end/end_node.py

@@ -1,6 +1,3 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
-
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.end.entities import EndNodeData
@@ -30,20 +27,3 @@ class EndNode(BaseNode[EndNodeData]):
             inputs=outputs,
             outputs=outputs,
         )
-
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls,
-        *,
-        graph_config: Mapping[str, Any],
-        node_id: str,
-        node_data: EndNodeData,
-    ) -> Mapping[str, Sequence[str]]:
-        """
-        Extract variable selector to variable mapping
-        :param graph_config: graph config
-        :param node_id: node id
-        :param node_data: node data
-        :return:
-        """
-        return {}

api/core/workflow/nodes/end/end_stream_processor.py

@@ -33,7 +33,7 @@ class EndStreamProcessor(StreamProcessor):
 
                 yield event
             elif isinstance(event, NodeRunStreamChunkEvent):
-                if event.in_iteration_id:
+                if event.in_iteration_id or event.in_loop_id:
                     if self.has_output and event.node_id not in self.output_node_ids:
                         event.chunk_content = "\n" + event.chunk_content
 

api/core/workflow/nodes/enums.py

@@ -16,12 +16,14 @@ class NodeType(StrEnum):
     VARIABLE_AGGREGATOR = "variable-aggregator"
     LEGACY_VARIABLE_AGGREGATOR = "variable-assigner"  # TODO: Merge this into VARIABLE_AGGREGATOR in the database.
     LOOP = "loop"
+    LOOP_START = "loop-start"
     ITERATION = "iteration"
     ITERATION_START = "iteration-start"  # Fake start node for iteration.
     PARAMETER_EXTRACTOR = "parameter-extractor"
     VARIABLE_ASSIGNER = "assigner"
     DOCUMENT_EXTRACTOR = "document-extractor"
     LIST_OPERATOR = "list-operator"
+    AGENT = "agent"
 
 
 class ErrorStrategy(StrEnum):

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

@@ -109,17 +109,19 @@ class Response:
         3. MIME type analysis
         """
         content_type = self.content_type.split(";")[0].strip().lower()
-        content_disposition = self.response.headers.get("content-disposition", "")
+        parsed_content_disposition = self.parsed_content_disposition
 
         # Check if it's explicitly marked as an attachment
-        if content_disposition:
-            msg = Message()
-            msg["content-disposition"] = content_disposition
-            disp_type = msg.get_content_disposition()  # Returns 'attachment', 'inline', or None
-            filename = msg.get_filename()  # Returns filename if present, None otherwise
+        if parsed_content_disposition:
+            disp_type = parsed_content_disposition.get_content_disposition()  # Returns 'attachment', 'inline', or None
+            filename = parsed_content_disposition.get_filename()  # Returns filename if present, None otherwise
             if disp_type == "attachment" or filename is not None:
                 return True
 
+        # For 'text/' types, only 'csv' should be downloaded as file
+        if content_type.startswith("text/") and "csv" not in content_type:
+            return False
+
         # For application types, try to detect if it's a text-based format
         if content_type.startswith("application/"):
             # Common text-based application types
@@ -178,3 +180,12 @@ class Response:
             return f"{(self.size / 1024):.2f} KB"
         else:
             return f"{(self.size / 1024 / 1024):.2f} MB"
+
+    @property
+    def parsed_content_disposition(self) -> Optional[Message]:
+        content_disposition = self.headers.get("content-disposition", "")
+        if content_disposition:
+            msg = Message()
+            msg["content-disposition"] = content_disposition
+            return msg
+        return None

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

@@ -10,6 +10,7 @@ import httpx
 from configs import dify_config
 from core.file import file_manager
 from core.helper import ssrf_proxy
+from core.variables.segments import ArrayFileSegment, FileSegment
 from core.workflow.entities.variable_pool import VariablePool
 
 from .entities import (
@@ -57,7 +58,7 @@ class Executor:
     params: list[tuple[str, str]] | None
     content: str | bytes | None
     data: Mapping[str, Any] | None
-    files: Mapping[str, tuple[str | None, bytes, str]] | None
+    files: list[tuple[str, tuple[str | None, bytes, str]]] | None
     json: Any
     headers: dict[str, str]
     auth: HttpRequestNodeAuthorization
@@ -207,17 +208,38 @@ class Executor:
                         self.variable_pool.convert_template(item.key).text: item.file
                         for item in filter(lambda item: item.type == "file", data)
                     }
-                    files: dict[str, Any] = {}
-                    files = {k: self.variable_pool.get_file(selector) for k, selector in file_selectors.items()}
-                    files = {k: v for k, v in files.items() if v is not None}
-                    files = {k: variable.value for k, variable in files.items() if variable is not None}
-                    files = {
-                        k: (v.filename, file_manager.download(v), v.mime_type or "application/octet-stream")
-                        for k, v in files.items()
-                        if v.related_id is not None
-                    }
+
+                    # get files from file_selectors, add support for array file variables
+                    files_list = []
+                    for key, selector in file_selectors.items():
+                        segment = self.variable_pool.get(selector)
+                        if isinstance(segment, FileSegment):
+                            files_list.append((key, [segment.value]))
+                        elif isinstance(segment, ArrayFileSegment):
+                            files_list.append((key, list(segment.value)))
+
+                    # get files from file_manager
+                    files: dict[str, list[tuple[str | None, bytes, str]]] = {}
+                    for key, files_in_segment in files_list:
+                        for file in files_in_segment:
+                            if file.related_id is not None:
+                                file_tuple = (
+                                    file.filename,
+                                    file_manager.download(file),
+                                    file.mime_type or "application/octet-stream",
+                                )
+                                if key not in files:
+                                    files[key] = []
+                                files[key].append(file_tuple)
+
+                    # convert files to list for httpx request
+                    if files:
+                        self.files = []
+                        for key, file_tuples in files.items():
+                            for file_tuple in file_tuples:
+                                self.files.append((key, file_tuple))
+
                     self.data = form_data
-                    self.files = files or None
 
     def _assembling_headers(self) -> dict[str, Any]:
         authorization = deepcopy(self.auth)
@@ -344,10 +366,16 @@ class Executor:
 
         body_string = ""
         if self.files:
-            for k, v in self.files.items():
+            for key, (filename, content, mime_type) in self.files:
                 body_string += f"--{boundary}\r\n"
-                body_string += f'Content-Disposition: form-data; name="{k}"\r\n\r\n'
-                body_string += f"{v[1]}\r\n"
+                body_string += f'Content-Disposition: form-data; name="{key}"\r\n\r\n'
+                # decode content
+                try:
+                    body_string += content.decode("utf-8")
+                except UnicodeDecodeError:
+                    # fix: decode binary content
+                    pass
+                body_string += "\r\n"
             body_string += f"--{boundary}--\r\n"
         elif self.node_data.body:
             if self.content:

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

@@ -169,32 +169,44 @@ class HttpRequestNode(BaseNode[HttpRequestNodeData]):
         """
         Extract files from response by checking both Content-Type header and URL
         """
-        files = []
+        files: list[File] = []
         is_file = response.is_file
         content_type = response.content_type
         content = response.content
+        parsed_content_disposition = response.parsed_content_disposition
+        content_disposition_type = None
 
-        if is_file:
-            # Guess file extension from URL or Content-Type header
-            filename = url.split("?")[0].split("/")[-1] or ""
-            mime_type = content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream"
+        if not is_file:
+            return files
 
-            tool_file = ToolFileManager.create_file_by_raw(
-                user_id=self.user_id,
-                tenant_id=self.tenant_id,
-                conversation_id=None,
-                file_binary=content,
-                mimetype=mime_type,
-            )
+        if parsed_content_disposition:
+            content_disposition_filename = parsed_content_disposition.get_filename()
+            if content_disposition_filename:
+                # If filename is available from content-disposition, use it to guess the content type
+                content_disposition_type = mimetypes.guess_type(content_disposition_filename)[0]
 
-            mapping = {
-                "tool_file_id": tool_file.id,
-                "transfer_method": FileTransferMethod.TOOL_FILE.value,
-            }
-            file = file_factory.build_from_mapping(
-                mapping=mapping,
-                tenant_id=self.tenant_id,
-            )
-            files.append(file)
+        # Guess file extension from URL or Content-Type header
+        filename = url.split("?")[0].split("/")[-1] or ""
+        mime_type = (
+            content_disposition_type or content_type or mimetypes.guess_type(filename)[0] or "application/octet-stream"
+        )
+
+        tool_file = ToolFileManager.create_file_by_raw(
+            user_id=self.user_id,
+            tenant_id=self.tenant_id,
+            conversation_id=None,
+            file_binary=content,
+            mimetype=mime_type,
+        )
+
+        mapping = {
+            "tool_file_id": tool_file.id,
+            "transfer_method": FileTransferMethod.TOOL_FILE.value,
+        }
+        file = file_factory.build_from_mapping(
+            mapping=mapping,
+            tenant_id=self.tenant_id,
+        )
+        files.append(file)
 
         return files

api/core/workflow/nodes/if_else/if_else_node.py

@@ -1,5 +1,4 @@
-from collections.abc import Mapping, Sequence
-from typing import Any, Literal
+from typing import Literal
 
 from typing_extensions import deprecated
 
@@ -88,23 +87,6 @@ class IfElseNode(BaseNode[IfElseNodeData]):
 
         return data
 
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls,
-        *,
-        graph_config: Mapping[str, Any],
-        node_id: str,
-        node_data: IfElseNodeData,
-    ) -> Mapping[str, Sequence[str]]:
-        """
-        Extract variable selector to variable mapping
-        :param graph_config: graph config
-        :param node_id: node id
-        :param node_data: node data
-        :return:
-        """
-        return {}
-
 
 @deprecated("This function is deprecated. You should use the new cases structure.")
 def _should_not_use_old_function(

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

@@ -1,3 +1,4 @@
+import contextvars
 import logging
 import uuid
 from collections.abc import Generator, Mapping, Sequence
@@ -174,6 +175,7 @@ class IterationNode(BaseNode[IterationNodeData]):
                         self._run_single_iter_parallel,
                         flask_app=current_app._get_current_object(),  # type: ignore
                         q=q,
+                        context=contextvars.copy_context(),
                         iterator_list_value=iterator_list_value,
                         inputs=inputs,
                         outputs=outputs,
@@ -568,6 +570,7 @@ class IterationNode(BaseNode[IterationNodeData]):
         self,
         *,
         flask_app: Flask,
+        context: contextvars.Context,
         q: Queue,
         iterator_list_value: Sequence[str],
         inputs: Mapping[str, list],
@@ -582,6 +585,8 @@ class IterationNode(BaseNode[IterationNodeData]):
         """
         run single iteration in parallel mode
         """
+        for var, val in context.items():
+            var.set(val)
         with flask_app.app_context():
             parallel_mode_run_id = uuid.uuid4().hex
             graph_engine_copy = graph_engine.create_copy()

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

@@ -1,14 +1,11 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
-
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
-from core.workflow.nodes.iteration.entities import IterationNodeData, IterationStartNodeData
+from core.workflow.nodes.iteration.entities import IterationStartNodeData
 from models.workflow import WorkflowNodeExecutionStatus
 
 
-class IterationStartNode(BaseNode):
+class IterationStartNode(BaseNode[IterationStartNodeData]):
     """
     Iteration Start Node.
     """
@@ -21,16 +18,3 @@ class IterationStartNode(BaseNode):
         Run the node.
         """
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED)
-
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls, graph_config: Mapping[str, Any], node_id: str, node_data: IterationNodeData
-    ) -> Mapping[str, Sequence[str]]:
-        """
-        Extract variable selector to variable mapping
-        :param graph_config: graph config
-        :param node_id: node id
-        :param node_data: node data
-        :return:
-        """
-        return {}

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

@@ -1,8 +1,10 @@
+from collections.abc import Sequence
 from typing import Any, Literal, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from core.workflow.nodes.base import BaseNodeData
+from core.workflow.nodes.llm.entities import VisionConfig
 
 
 class RerankingModelConfig(BaseModel):
@@ -73,6 +75,48 @@ class SingleRetrievalConfig(BaseModel):
     model: ModelConfig
 
 
+SupportedComparisonOperator = Literal[
+    # for string or array
+    "contains",
+    "not contains",
+    "start with",
+    "end with",
+    "is",
+    "is not",
+    "empty",
+    "not empty",
+    # for number
+    "=",
+    "≠",
+    ">",
+    "<",
+    "≥",
+    "≤",
+    # for time
+    "before",
+    "after",
+]
+
+
+class Condition(BaseModel):
+    """
+    Conditon detail
+    """
+
+    name: str
+    comparison_operator: SupportedComparisonOperator
+    value: str | Sequence[str] | None | int | float = None
+
+
+class MetadataFilteringCondition(BaseModel):
+    """
+    Metadata Filtering Condition.
+    """
+
+    logical_operator: Optional[Literal["and", "or"]] = "and"
+    conditions: Optional[list[Condition]] = Field(default=None, deprecated=True)
+
+
 class KnowledgeRetrievalNodeData(BaseNodeData):
     """
     Knowledge retrieval Node Data.
@@ -84,3 +128,7 @@ class KnowledgeRetrievalNodeData(BaseNodeData):
     retrieval_mode: Literal["single", "multiple"]
     multiple_retrieval_config: Optional[MultipleRetrievalConfig] = None
     single_retrieval_config: Optional[SingleRetrievalConfig] = None
+    metadata_filtering_mode: Optional[Literal["disabled", "automatic", "manual"]] = "disabled"
+    metadata_model_config: Optional[ModelConfig] = None
+    metadata_filtering_conditions: Optional[MetadataFilteringCondition] = None
+    vision: VisionConfig = Field(default_factory=VisionConfig)

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

@@ -16,3 +16,7 @@ class ModelNotSupportedError(KnowledgeRetrievalNodeError):
 
 class ModelQuotaExceededError(KnowledgeRetrievalNodeError):
     """Raised when the model provider quota is exceeded."""
+
+
+class InvalidModelTypeError(KnowledgeRetrievalNodeError):
+    """Raised when the model is not a Large Language Model."""

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

@@ -1,29 +1,51 @@
+import json
 import logging
+import time
+from collections import defaultdict
 from collections.abc import Mapping, Sequence
-from typing import Any, cast
+from typing import Any, Optional, cast
 
-from sqlalchemy import func
+from sqlalchemy import Integer, and_, func, or_, text
+from sqlalchemy import cast as sqlalchemy_cast
 
 from core.app.app_config.entities import DatasetRetrieveConfigEntity
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.entities.agent_entities import PlanningStrategy
 from core.entities.model_entities import ModelStatus
 from core.model_manager import ModelInstance, ModelManager
+from core.model_runtime.entities.message_entities import PromptMessageRole
 from core.model_runtime.entities.model_entities import ModelFeature, ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from core.prompt.simple_prompt_transform import ModelMode
 from core.rag.datasource.retrieval_service import RetrievalService
+from core.rag.entities.metadata_entities import Condition, MetadataCondition
 from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.variables import StringSegment
 from core.workflow.entities.node_entities import NodeRunResult
-from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
+from core.workflow.nodes.event.event import ModelInvokeCompletedEvent
+from core.workflow.nodes.knowledge_retrieval.template_prompts import (
+    METADATA_FILTER_ASSISTANT_PROMPT_1,
+    METADATA_FILTER_ASSISTANT_PROMPT_2,
+    METADATA_FILTER_COMPLETION_PROMPT,
+    METADATA_FILTER_SYSTEM_PROMPT,
+    METADATA_FILTER_USER_PROMPT_1,
+    METADATA_FILTER_USER_PROMPT_3,
+)
+from core.workflow.nodes.llm.entities import LLMNodeChatModelMessage, LLMNodeCompletionModelPromptTemplate
+from core.workflow.nodes.llm.node import LLMNode
+from core.workflow.nodes.question_classifier.template_prompts import QUESTION_CLASSIFIER_USER_PROMPT_2
 from extensions.ext_database import db
-from models.dataset import Dataset, Document
+from extensions.ext_redis import redis_client
+from libs.json_in_md_parser import parse_and_check_json_markdown
+from models.dataset import Dataset, DatasetMetadata, Document, RateLimitLog
 from models.workflow import WorkflowNodeExecutionStatus
+from services.feature_service import FeatureService
 
-from .entities import KnowledgeRetrievalNodeData
+from .entities import KnowledgeRetrievalNodeData, ModelConfig
 from .exc import (
+    InvalidModelTypeError,
     KnowledgeRetrievalNodeError,
     ModelCredentialsNotInitializedError,
     ModelNotExistError,
@@ -42,13 +64,14 @@ default_retrieval_model = {
 }
 
 
-class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
-    _node_data_cls = KnowledgeRetrievalNodeData
+class KnowledgeRetrievalNode(LLMNode):
+    _node_data_cls = KnowledgeRetrievalNodeData  # type: ignore
     _node_type = NodeType.KNOWLEDGE_RETRIEVAL
 
-    def _run(self) -> NodeRunResult:
+    def _run(self) -> NodeRunResult:  # type: ignore
+        node_data = cast(KnowledgeRetrievalNodeData, self.node_data)
         # extract variables
-        variable = self.graph_runtime_state.variable_pool.get(self.node_data.query_variable_selector)
+        variable = self.graph_runtime_state.variable_pool.get(node_data.query_variable_selector)
         if not isinstance(variable, StringSegment):
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED,
@@ -61,9 +84,34 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED, inputs=variables, error="Query is required."
             )
+        # check rate limit
+        if self.tenant_id:
+            knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(self.tenant_id)
+            if knowledge_rate_limit.enabled:
+                current_time = int(time.time() * 1000)
+                key = f"rate_limit_{self.tenant_id}"
+                redis_client.zadd(key, {current_time: current_time})
+                redis_client.zremrangebyscore(key, 0, current_time - 60000)
+                request_count = redis_client.zcard(key)
+                if request_count > knowledge_rate_limit.limit:
+                    # add ratelimit record
+                    rate_limit_log = RateLimitLog(
+                        tenant_id=self.tenant_id,
+                        subscription_plan=knowledge_rate_limit.subscription_plan,
+                        operation="knowledge",
+                    )
+                    db.session.add(rate_limit_log)
+                    db.session.commit()
+                    return NodeRunResult(
+                        status=WorkflowNodeExecutionStatus.FAILED,
+                        inputs=variables,
+                        error="Sorry, you have reached the knowledge base request rate limit of your subscription.",
+                        error_type="RateLimitExceeded",
+                    )
+
         # retrieve knowledge
         try:
-            results = self._fetch_dataset_retriever(node_data=self.node_data, query=query)
+            results = self._fetch_dataset_retriever(node_data=node_data, query=query)
             outputs = {"result": results}
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=variables, process_data=None, outputs=outputs
@@ -117,11 +165,14 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
             if not dataset:
                 continue
             available_datasets.append(dataset)
+        metadata_filter_document_ids, metadata_condition = self._get_metadata_filter_condition(
+            [dataset.id for dataset in available_datasets], query, node_data
+        )
         all_documents = []
         dataset_retrieval = DatasetRetrieval()
         if node_data.retrieval_mode == DatasetRetrieveConfigEntity.RetrieveStrategy.SINGLE.value:
             # fetch model config
-            model_instance, model_config = self._fetch_model_config(node_data)
+            model_instance, model_config = self._fetch_model_config(node_data.single_retrieval_config.model)  # type: ignore
             # check model is support tool calling
             model_type_instance = model_config.provider_model_bundle.model_type_instance
             model_type_instance = cast(LargeLanguageModel, model_type_instance)
@@ -146,6 +197,8 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
                     model_config=model_config,
                     model_instance=model_instance,
                     planning_strategy=planning_strategy,
+                    metadata_filter_document_ids=metadata_filter_document_ids,
+                    metadata_condition=metadata_condition,
                 )
         elif node_data.retrieval_mode == DatasetRetrieveConfigEntity.RetrieveStrategy.MULTIPLE.value:
             if node_data.multiple_retrieval_config is None:
@@ -192,6 +245,8 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
                 reranking_model=reranking_model,
                 weights=weights,
                 reranking_enable=node_data.multiple_retrieval_config.reranking_enable,
+                metadata_filter_document_ids=metadata_filter_document_ids,
+                metadata_condition=metadata_condition,
             )
         dify_documents = [item for item in all_documents if item.provider == "dify"]
         external_documents = [item for item in all_documents if item.provider == "external"]
@@ -240,6 +295,7 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
                                 "segment_word_count": segment.word_count,
                                 "segment_position": segment.position,
                                 "segment_index_node_hash": segment.index_node_hash,
+                                "doc_metadata": document.doc_metadata,
                             },
                             "title": document.name,
                         }
@@ -258,13 +314,187 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
                 item["metadata"]["position"] = position
         return retrieval_resource_list
 
+    def _get_metadata_filter_condition(
+        self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
+    ) -> tuple[Optional[dict[str, list[str]]], Optional[MetadataCondition]]:
+        document_query = db.session.query(Document).filter(
+            Document.dataset_id.in_(dataset_ids),
+            Document.indexing_status == "completed",
+            Document.enabled == True,
+            Document.archived == False,
+        )
+        filters = []  # type: ignore
+        metadata_condition = None
+        if node_data.metadata_filtering_mode == "disabled":
+            return None, None
+        elif node_data.metadata_filtering_mode == "automatic":
+            automatic_metadata_filters = self._automatic_metadata_filter_func(dataset_ids, query, node_data)
+            if automatic_metadata_filters:
+                conditions = []
+                for filter in automatic_metadata_filters:
+                    self._process_metadata_filter_func(
+                        filter.get("condition", ""),
+                        filter.get("metadata_name", ""),
+                        filter.get("value"),
+                        filters,  # type: ignore
+                    )
+                    conditions.append(
+                        Condition(
+                            name=filter.get("metadata_name"),  # type: ignore
+                            comparison_operator=filter.get("condition"),  # type: ignore
+                            value=filter.get("value"),
+                        )
+                    )
+                metadata_condition = MetadataCondition(
+                    logical_operator=node_data.metadata_filtering_conditions.logical_operator,  # type: ignore
+                    conditions=conditions,
+                )
+        elif node_data.metadata_filtering_mode == "manual":
+            if node_data.metadata_filtering_conditions:
+                metadata_condition = MetadataCondition(**node_data.metadata_filtering_conditions.model_dump())
+                if node_data.metadata_filtering_conditions:
+                    for condition in node_data.metadata_filtering_conditions.conditions:  # type: ignore
+                        metadata_name = condition.name
+                        expected_value = condition.value
+                        if expected_value is not None or condition.comparison_operator in ("empty", "not empty"):
+                            if isinstance(expected_value, str):
+                                expected_value = self.graph_runtime_state.variable_pool.convert_template(
+                                    expected_value
+                                ).text
+
+                            filters = self._process_metadata_filter_func(
+                                condition.comparison_operator, metadata_name, expected_value, filters
+                            )
+        else:
+            raise ValueError("Invalid metadata filtering mode")
+        if filters:
+            if node_data.metadata_filtering_conditions.logical_operator == "and":  # type: ignore
+                document_query = document_query.filter(and_(*filters))
+            else:
+                document_query = document_query.filter(or_(*filters))
+        documents = document_query.all()
+        # group by dataset_id
+        metadata_filter_document_ids = defaultdict(list) if documents else None  # type: ignore
+        for document in documents:
+            metadata_filter_document_ids[document.dataset_id].append(document.id)  # type: ignore
+        return metadata_filter_document_ids, metadata_condition
+
+    def _automatic_metadata_filter_func(
+        self, dataset_ids: list, query: str, node_data: KnowledgeRetrievalNodeData
+    ) -> list[dict[str, Any]]:
+        # get all metadata field
+        metadata_fields = db.session.query(DatasetMetadata).filter(DatasetMetadata.dataset_id.in_(dataset_ids)).all()
+        all_metadata_fields = [metadata_field.name for metadata_field in metadata_fields]
+        # get metadata model config
+        metadata_model_config = node_data.metadata_model_config
+        if metadata_model_config is None:
+            raise ValueError("metadata_model_config is required")
+        # get metadata model instance
+        # fetch model config
+        model_instance, model_config = self._fetch_model_config(node_data.metadata_model_config)  # type: ignore
+        # fetch prompt messages
+        prompt_template = self._get_prompt_template(
+            node_data=node_data,
+            metadata_fields=all_metadata_fields,
+            query=query or "",
+        )
+        prompt_messages, stop = self._fetch_prompt_messages(
+            prompt_template=prompt_template,
+            sys_query=query,
+            memory=None,
+            model_config=model_config,
+            sys_files=[],
+            vision_enabled=node_data.vision.enabled,
+            vision_detail=node_data.vision.configs.detail,
+            variable_pool=self.graph_runtime_state.variable_pool,
+            jinja2_variables=[],
+        )
+
+        result_text = ""
+        try:
+            # handle invoke result
+            generator = self._invoke_llm(
+                node_data_model=node_data.metadata_model_config,  # type: ignore
+                model_instance=model_instance,
+                prompt_messages=prompt_messages,
+                stop=stop,
+            )
+
+            for event in generator:
+                if isinstance(event, ModelInvokeCompletedEvent):
+                    result_text = event.text
+                    break
+
+            result_text_json = parse_and_check_json_markdown(result_text, [])
+            automatic_metadata_filters = []
+            if "metadata_map" in result_text_json:
+                metadata_map = result_text_json["metadata_map"]
+                for item in metadata_map:
+                    if item.get("metadata_field_name") in all_metadata_fields:
+                        automatic_metadata_filters.append(
+                            {
+                                "metadata_name": item.get("metadata_field_name"),
+                                "value": item.get("metadata_field_value"),
+                                "condition": item.get("comparison_operator"),
+                            }
+                        )
+        except Exception as e:
+            return []
+        return automatic_metadata_filters
+
+    def _process_metadata_filter_func(self, condition: str, metadata_name: str, value: Optional[str], filters: list):
+        match condition:
+            case "contains":
+                filters.append(
+                    (text("documents.doc_metadata ->> :key LIKE :value")).params(key=metadata_name, value=f"%{value}%")
+                )
+            case "not contains":
+                filters.append(
+                    (text("documents.doc_metadata ->> :key NOT LIKE :value")).params(
+                        key=metadata_name, value=f"%{value}%"
+                    )
+                )
+            case "start with":
+                filters.append(
+                    (text("documents.doc_metadata ->> :key LIKE :value")).params(key=metadata_name, value=f"{value}%")
+                )
+            case "end with":
+                filters.append(
+                    (text("documents.doc_metadata ->> :key LIKE :value")).params(key=metadata_name, value=f"%{value}")
+                )
+            case "=" | "is":
+                if isinstance(value, str):
+                    filters.append(Document.doc_metadata[metadata_name] == f'"{value}"')
+                else:
+                    filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) == value)
+            case "is not" | "≠":
+                if isinstance(value, str):
+                    filters.append(Document.doc_metadata[metadata_name] != f'"{value}"')
+                else:
+                    filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) != value)
+            case "empty":
+                filters.append(Document.doc_metadata[metadata_name].is_(None))
+            case "not empty":
+                filters.append(Document.doc_metadata[metadata_name].isnot(None))
+            case "before" | "<":
+                filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) < value)
+            case "after" | ">":
+                filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) > value)
+            case "≤" | ">=":
+                filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) <= value)
+            case "≥" | ">=":
+                filters.append(sqlalchemy_cast(Document.doc_metadata[metadata_name].astext, Integer) >= value)
+            case _:
+                pass
+        return filters
+
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,
         *,
         graph_config: Mapping[str, Any],
         node_id: str,
-        node_data: KnowledgeRetrievalNodeData,
+        node_data: KnowledgeRetrievalNodeData,  # type: ignore
     ) -> Mapping[str, Sequence[str]]:
         """
         Extract variable selector to variable mapping
@@ -277,18 +507,16 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
         variable_mapping[node_id + ".query"] = node_data.query_variable_selector
         return variable_mapping
 
-    def _fetch_model_config(
-        self, node_data: KnowledgeRetrievalNodeData
-    ) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:
+    def _fetch_model_config(self, model: ModelConfig) -> tuple[ModelInstance, ModelConfigWithCredentialsEntity]:  # type: ignore
         """
         Fetch model config
-        :param node_data: node data
+        :param model: model
         :return:
         """
-        if node_data.single_retrieval_config is None:
-            raise ValueError("single_retrieval_config is required")
-        model_name = node_data.single_retrieval_config.model.name
-        provider_name = node_data.single_retrieval_config.model.provider
+        if model is None:
+            raise ValueError("model is required")
+        model_name = model.name
+        provider_name = model.provider
 
         model_manager = ModelManager()
         model_instance = model_manager.get_model_instance(
@@ -317,14 +545,14 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
             raise ModelQuotaExceededError(f"Model provider {provider_name} quota exceeded.")
 
         # model config
-        completion_params = node_data.single_retrieval_config.model.completion_params
+        completion_params = model.completion_params
         stop = []
         if "stop" in completion_params:
             stop = completion_params["stop"]
             del completion_params["stop"]
 
         # get model mode
-        model_mode = node_data.single_retrieval_config.model.mode
+        model_mode = model.mode
         if not model_mode:
             raise ModelNotExistError("LLM mode is required.")
 
@@ -343,3 +571,50 @@ class KnowledgeRetrievalNode(BaseNode[KnowledgeRetrievalNodeData]):
             parameters=completion_params,
             stop=stop,
         )
+
+    def _get_prompt_template(self, node_data: KnowledgeRetrievalNodeData, metadata_fields: list, query: str):
+        model_mode = ModelMode.value_of(node_data.metadata_model_config.mode)  # type: ignore
+        input_text = query
+        memory_str = ""
+
+        prompt_messages: list[LLMNodeChatModelMessage] = []
+        if model_mode == ModelMode.CHAT:
+            system_prompt_messages = LLMNodeChatModelMessage(
+                role=PromptMessageRole.SYSTEM, text=METADATA_FILTER_SYSTEM_PROMPT
+            )
+            prompt_messages.append(system_prompt_messages)
+            user_prompt_message_1 = LLMNodeChatModelMessage(
+                role=PromptMessageRole.USER, text=METADATA_FILTER_USER_PROMPT_1
+            )
+            prompt_messages.append(user_prompt_message_1)
+            assistant_prompt_message_1 = LLMNodeChatModelMessage(
+                role=PromptMessageRole.ASSISTANT, text=METADATA_FILTER_ASSISTANT_PROMPT_1
+            )
+            prompt_messages.append(assistant_prompt_message_1)
+            user_prompt_message_2 = LLMNodeChatModelMessage(
+                role=PromptMessageRole.USER, text=QUESTION_CLASSIFIER_USER_PROMPT_2
+            )
+            prompt_messages.append(user_prompt_message_2)
+            assistant_prompt_message_2 = LLMNodeChatModelMessage(
+                role=PromptMessageRole.ASSISTANT, text=METADATA_FILTER_ASSISTANT_PROMPT_2
+            )
+            prompt_messages.append(assistant_prompt_message_2)
+            user_prompt_message_3 = LLMNodeChatModelMessage(
+                role=PromptMessageRole.USER,
+                text=METADATA_FILTER_USER_PROMPT_3.format(
+                    input_text=input_text,
+                    metadata_fields=json.dumps(metadata_fields, ensure_ascii=False),
+                ),
+            )
+            prompt_messages.append(user_prompt_message_3)
+            return prompt_messages
+        elif model_mode == ModelMode.COMPLETION:
+            return LLMNodeCompletionModelPromptTemplate(
+                text=METADATA_FILTER_COMPLETION_PROMPT.format(
+                    input_text=input_text,
+                    metadata_fields=json.dumps(metadata_fields, ensure_ascii=False),
+                )
+            )
+
+        else:
+            raise InvalidModelTypeError(f"Model mode {model_mode} not support.")

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

@@ -0,0 +1,66 @@
+METADATA_FILTER_SYSTEM_PROMPT = """
+    ### Job Description',
+    You are a text metadata extract engine that extract text's metadata based on user input and set the metadata value
+    ### Task
+    Your task is to ONLY extract the metadatas that exist in the input text from the provided metadata list and Use the following operators ["=", "!=", ">", "<", ">=", "<="] to express logical relationships, then return result in JSON format with the key "metadata_fields" and value "metadata_field_value" and comparison operator "comparison_operator".
+    ### Format
+    The input text is in the variable input_text. Metadata are specified as a list in the variable metadata_fields.
+    ### Constraint
+    DO NOT include anything other than the JSON array in your response.
+"""  # noqa: E501
+
+METADATA_FILTER_USER_PROMPT_1 = """
+    { "input_text": "I want to know which company’s email address test@example.com is?",
+    "metadata_fields": ["filename", "email", "phone", "address"]
+    }
+"""
+
+METADATA_FILTER_ASSISTANT_PROMPT_1 = """
+```json
+    {"metadata_map": [
+        {"metadata_field_name": "email", "metadata_field_value": "test@example.com", "comparison_operator": "="}
+    ]
+    }
+```
+"""
+
+METADATA_FILTER_USER_PROMPT_2 = """
+    {"input_text": "What are the movies with a score of more than 9 in 2024?",
+    "metadata_fields": ["name", "year", "rating", "country"]}
+"""
+
+METADATA_FILTER_ASSISTANT_PROMPT_2 = """
+```json
+    {"metadata_map": [
+        {"metadata_field_name": "year", "metadata_field_value": "2024", "comparison_operator": "="},
+        {"metadata_field_name": "rating", "metadata_field_value": "9", "comparison_operator": ">"},
+    ]}
+```
+"""
+
+METADATA_FILTER_USER_PROMPT_3 = """
+    '{{"input_text": "{input_text}",',
+    '"metadata_fields": {metadata_fields}}}'
+"""
+
+METADATA_FILTER_COMPLETION_PROMPT = """
+### Job Description
+You are a text metadata extract engine that extract text's metadata based on user input and set the metadata value
+### Task
+# Your task is to ONLY extract the metadatas that exist in the input text from the provided metadata list and Use the following operators ["=", "!=", ">", "<", ">=", "<="] to express logical relationships, then return result in JSON format with the key "metadata_fields" and value "metadata_field_value" and comparison operator "comparison_operator".
+### Format
+The input text is in the variable input_text. Metadata are specified as a list in the variable metadata_fields.
+### Constraint 
+DO NOT include anything other than the JSON array in your response.
+### Example
+Here is the chat example between human and assistant, inside <example></example> XML tags.
+<example>
+User:{{"input_text": ["I want to know which company’s email address test@example.com is?"], "metadata_fields": ["filename", "email", "phone", "address"]}}
+Assistant:{{"metadata_map": [{{"metadata_field_name": "email", "metadata_field_value": "test@example.com", "comparison_operator": "="}}]}}
+User:{{"input_text": "What are the movies with a score of more than 9 in 2024?", "metadata_fields": ["name", "year", "rating", "country"]}}
+Assistant:{{"metadata_map": [{{"metadata_field_name": "year", "metadata_field_value": "2024", "comparison_operator": "="}, {{"metadata_field_name": "rating", "metadata_field_value": "9", "comparison_operator": ">"}}]}}
+</example> 
+### User Input
+{{"input_text" : "{input_text}", "metadata_fields" : {metadata_fields}}}
+### Assistant Output
+"""  # noqa: E501

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

@@ -1,6 +1,7 @@
 import json
 import logging
 from collections.abc import Generator, Mapping, Sequence
+from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any, Optional, cast
 
 from configs import dify_config
@@ -29,6 +30,7 @@ from core.model_runtime.entities.message_entities import (
 from core.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey, ModelType
 from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin import ModelProviderID
 from core.prompt.entities.advanced_prompt_entities import CompletionModelPromptTemplate, MemoryConfig
 from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.variables import (
@@ -92,6 +94,9 @@ class LLMNode(BaseNode[LLMNodeData]):
     def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
         node_inputs: Optional[dict[str, Any]] = None
         process_data = None
+        result_text = ""
+        usage = LLMUsage.empty_usage()
+        finish_reason = None
 
         try:
             # init messages template
@@ -176,9 +181,6 @@ class LLMNode(BaseNode[LLMNodeData]):
                 stop=stop,
             )
 
-            result_text = ""
-            usage = LLMUsage.empty_usage()
-            finish_reason = None
             for event in generator:
                 if isinstance(event, RunStreamChunkEvent):
                     yield event
@@ -189,6 +191,22 @@ class LLMNode(BaseNode[LLMNodeData]):
                     # deduct quota
                     self.deduct_llm_quota(tenant_id=self.tenant_id, model_instance=model_instance, usage=usage)
                     break
+            outputs = {"text": result_text, "usage": jsonable_encoder(usage), "finish_reason": finish_reason}
+
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                    inputs=node_inputs,
+                    process_data=process_data,
+                    outputs=outputs,
+                    metadata={
+                        NodeRunMetadataKey.TOTAL_TOKENS: usage.total_tokens,
+                        NodeRunMetadataKey.TOTAL_PRICE: usage.total_price,
+                        NodeRunMetadataKey.CURRENCY: usage.currency,
+                    },
+                    llm_usage=usage,
+                )
+            )
         except LLMNodeError as e:
             yield RunCompletedEvent(
                 run_result=NodeRunResult(
@@ -209,23 +227,6 @@ class LLMNode(BaseNode[LLMNodeData]):
                 )
             )
 
-        outputs = {"text": result_text, "usage": jsonable_encoder(usage), "finish_reason": finish_reason}
-
-        yield RunCompletedEvent(
-            run_result=NodeRunResult(
-                status=WorkflowNodeExecutionStatus.SUCCEEDED,
-                inputs=node_inputs,
-                process_data=process_data,
-                outputs=outputs,
-                metadata={
-                    NodeRunMetadataKey.TOTAL_TOKENS: usage.total_tokens,
-                    NodeRunMetadataKey.TOTAL_PRICE: usage.total_price,
-                    NodeRunMetadataKey.CURRENCY: usage.currency,
-                },
-                llm_usage=usage,
-            )
-        )
-
     def _invoke_llm(
         self,
         node_data_model: ModelConfig,
@@ -236,9 +237,9 @@ class LLMNode(BaseNode[LLMNodeData]):
         db.session.close()
 
         invoke_result = model_instance.invoke_llm(
-            prompt_messages=prompt_messages,
+            prompt_messages=list(prompt_messages),
             model_parameters=node_data_model.completion_params,
-            stop=stop,
+            stop=list(stop or []),
             stream=True,
             user=self.user_id,
         )
@@ -247,6 +248,24 @@ class LLMNode(BaseNode[LLMNodeData]):
 
     def _handle_invoke_result(self, invoke_result: LLMResult | Generator) -> Generator[NodeEvent, None, None]:
         if isinstance(invoke_result, LLMResult):
+            content = invoke_result.message.content
+            if content is None:
+                message_text = ""
+            elif isinstance(content, str):
+                message_text = content
+            elif isinstance(content, list):
+                # Assuming the list contains PromptMessageContent objects with a "data" attribute
+                message_text = "".join(
+                    item.data if hasattr(item, "data") and isinstance(item.data, str) else str(item) for item in content
+                )
+            else:
+                message_text = str(content)
+
+            yield ModelInvokeCompletedEvent(
+                text=message_text,
+                usage=invoke_result.usage,
+                finish_reason=None,
+            )
             return
 
         model = None
@@ -439,6 +458,7 @@ class LLMNode(BaseNode[LLMNodeData]):
                 "index_node_hash": metadata.get("segment_index_node_hash"),
                 "content": context_dict.get("content"),
                 "page": metadata.get("page"),
+                "doc_metadata": metadata.get("doc_metadata"),
             }
 
             return source
@@ -740,11 +760,17 @@ class LLMNode(BaseNode[LLMNodeData]):
         if used_quota is not None and system_configuration.current_quota_type is not None:
             db.session.query(Provider).filter(
                 Provider.tenant_id == tenant_id,
-                Provider.provider_name == model_instance.provider,
+                # 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,
-            ).update({"quota_used": Provider.quota_used + used_quota})
+            ).update(
+                {
+                    "quota_used": Provider.quota_used + used_quota,
+                    "last_used": datetime.now(tz=UTC).replace(tzinfo=None),
+                }
+            )
             db.session.commit()
 
     @classmethod

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

@@ -0,0 +1,5 @@
+from .entities import LoopNodeData
+from .loop_node import LoopNode
+from .loop_start_node import LoopStartNode
+
+__all__ = ["LoopNode", "LoopNodeData", "LoopStartNode"]

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

@@ -1,13 +1,54 @@
-from core.workflow.nodes.base import BaseIterationNodeData, BaseIterationState
+from typing import Any, Literal, Optional
+
+from pydantic import Field
+
+from core.workflow.nodes.base import BaseLoopNodeData, BaseLoopState, BaseNodeData
+from core.workflow.utils.condition.entities import Condition
 
 
-class LoopNodeData(BaseIterationNodeData):
+class LoopNodeData(BaseLoopNodeData):
     """
     Loop Node Data.
     """
 
+    loop_count: int  # Maximum number of loops
+    break_conditions: list[Condition]  # Conditions to break the loop
+    logical_operator: Literal["and", "or"]
 
-class LoopState(BaseIterationState):
+
+class LoopStartNodeData(BaseNodeData):
+    """
+    Loop Start Node Data.
+    """
+
+    pass
+
+
+class LoopState(BaseLoopState):
     """
     Loop State.
     """
+
+    outputs: list[Any] = Field(default_factory=list)
+    current_output: Optional[Any] = None
+
+    class MetaData(BaseLoopState.MetaData):
+        """
+        Data.
+        """
+
+        loop_length: int
+
+    def get_last_output(self) -> Optional[Any]:
+        """
+        Get last output.
+        """
+        if self.outputs:
+            return self.outputs[-1]
+        return None
+
+    def get_current_output(self) -> Optional[Any]:
+        """
+        Get current output.
+        """
+        return self.current_output

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

@@ -1,9 +1,35 @@
-from typing import Any
+import logging
+from collections.abc import Generator, Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any, cast
 
+from configs import dify_config
+from core.variables import IntegerSegment
+from core.workflow.entities.node_entities import NodeRunMetadataKey, NodeRunResult
+from core.workflow.graph_engine.entities.event import (
+    BaseGraphEvent,
+    BaseNodeEvent,
+    BaseParallelBranchEvent,
+    GraphRunFailedEvent,
+    InNodeEvent,
+    LoopRunFailedEvent,
+    LoopRunNextEvent,
+    LoopRunStartedEvent,
+    LoopRunSucceededEvent,
+    NodeRunFailedEvent,
+    NodeRunStartedEvent,
+    NodeRunStreamChunkEvent,
+    NodeRunSucceededEvent,
+)
+from core.workflow.graph_engine.entities.graph import Graph
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
-from core.workflow.nodes.loop.entities import LoopNodeData, LoopState
-from core.workflow.utils.condition.entities import Condition
+from core.workflow.nodes.event import NodeEvent, RunCompletedEvent
+from core.workflow.nodes.loop.entities import LoopNodeData
+from core.workflow.utils.condition.processor import ConditionProcessor
+from models.workflow import WorkflowNodeExecutionStatus
+
+logger = logging.getLogger(__name__)
 
 
 class LoopNode(BaseNode[LoopNodeData]):
@@ -14,24 +40,323 @@ class LoopNode(BaseNode[LoopNodeData]):
     _node_data_cls = LoopNodeData
     _node_type = NodeType.LOOP
 
-    def _run(self) -> LoopState:  # type: ignore
-        return super()._run()  # type: ignore
+    def _run(self) -> Generator[NodeEvent | InNodeEvent, None, None]:
+        """Run the node."""
+        # Get inputs
+        loop_count = self.node_data.loop_count
+        break_conditions = self.node_data.break_conditions
+        logical_operator = self.node_data.logical_operator
+
+        inputs = {"loop_count": loop_count}
+
+        if not self.node_data.start_node_id:
+            raise ValueError(f"field start_node_id in loop {self.node_id} not found")
+
+        # Initialize graph
+        loop_graph = Graph.init(graph_config=self.graph_config, root_node_id=self.node_data.start_node_id)
+        if not loop_graph:
+            raise ValueError("loop graph not found")
+
+        # Initialize variable pool
+        variable_pool = self.graph_runtime_state.variable_pool
+        variable_pool.add([self.node_id, "index"], 0)
+
+        from core.workflow.graph_engine.graph_engine import GraphEngine
+
+        graph_engine = GraphEngine(
+            tenant_id=self.tenant_id,
+            app_id=self.app_id,
+            workflow_type=self.workflow_type,
+            workflow_id=self.workflow_id,
+            user_id=self.user_id,
+            user_from=self.user_from,
+            invoke_from=self.invoke_from,
+            call_depth=self.workflow_call_depth,
+            graph=loop_graph,
+            graph_config=self.graph_config,
+            variable_pool=variable_pool,
+            max_execution_steps=dify_config.WORKFLOW_MAX_EXECUTION_STEPS,
+            max_execution_time=dify_config.WORKFLOW_MAX_EXECUTION_TIME,
+            thread_pool_id=self.thread_pool_id,
+        )
+
+        start_at = datetime.now(UTC).replace(tzinfo=None)
+        condition_processor = ConditionProcessor()
+
+        # Start Loop event
+        yield LoopRunStartedEvent(
+            loop_id=self.id,
+            loop_node_id=self.node_id,
+            loop_node_type=self.node_type,
+            loop_node_data=self.node_data,
+            start_at=start_at,
+            inputs=inputs,
+            metadata={"loop_length": loop_count},
+            predecessor_node_id=self.previous_node_id,
+        )
+
+        yield LoopRunNextEvent(
+            loop_id=self.id,
+            loop_node_id=self.node_id,
+            loop_node_type=self.node_type,
+            loop_node_data=self.node_data,
+            index=0,
+            pre_loop_output=None,
+        )
+
+        try:
+            check_break_result = False
+            for i in range(loop_count):
+                # Run workflow
+                rst = graph_engine.run()
+                current_index_variable = variable_pool.get([self.node_id, "index"])
+                if not isinstance(current_index_variable, IntegerSegment):
+                    raise ValueError(f"loop {self.node_id} current index not found")
+                current_index = current_index_variable.value
+
+                check_break_result = False
+
+                for event in rst:
+                    if isinstance(event, (BaseNodeEvent | BaseParallelBranchEvent)) and not event.in_loop_id:
+                        event.in_loop_id = self.node_id
+
+                    if (
+                        isinstance(event, BaseNodeEvent)
+                        and event.node_type == NodeType.LOOP_START
+                        and not isinstance(event, NodeRunStreamChunkEvent)
+                    ):
+                        continue
+
+                    if isinstance(event, NodeRunSucceededEvent):
+                        yield self._handle_event_metadata(event=event, iter_run_index=current_index)
+
+                        # Check if all variables in break conditions exist
+                        exists_variable = False
+                        for condition in break_conditions:
+                            if not self.graph_runtime_state.variable_pool.get(condition.variable_selector):
+                                exists_variable = False
+                                break
+                            else:
+                                exists_variable = True
+                        if exists_variable:
+                            input_conditions, group_result, check_break_result = condition_processor.process_conditions(
+                                variable_pool=self.graph_runtime_state.variable_pool,
+                                conditions=break_conditions,
+                                operator=logical_operator,
+                            )
+                            if check_break_result:
+                                break
+
+                    elif isinstance(event, BaseGraphEvent):
+                        if isinstance(event, GraphRunFailedEvent):
+                            # Loop run failed
+                            yield LoopRunFailedEvent(
+                                loop_id=self.id,
+                                loop_node_id=self.node_id,
+                                loop_node_type=self.node_type,
+                                loop_node_data=self.node_data,
+                                start_at=start_at,
+                                inputs=inputs,
+                                steps=i,
+                                metadata={
+                                    NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens,
+                                    "completed_reason": "error",
+                                },
+                                error=event.error,
+                            )
+                            yield RunCompletedEvent(
+                                run_result=NodeRunResult(
+                                    status=WorkflowNodeExecutionStatus.FAILED,
+                                    error=event.error,
+                                    metadata={
+                                        NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens
+                                    },
+                                )
+                            )
+                            return
+                    elif isinstance(event, NodeRunFailedEvent):
+                        # Loop run failed
+                        yield event
+                        yield LoopRunFailedEvent(
+                            loop_id=self.id,
+                            loop_node_id=self.node_id,
+                            loop_node_type=self.node_type,
+                            loop_node_data=self.node_data,
+                            start_at=start_at,
+                            inputs=inputs,
+                            steps=i,
+                            metadata={
+                                NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens,
+                                "completed_reason": "error",
+                            },
+                            error=event.error,
+                        )
+                        yield RunCompletedEvent(
+                            run_result=NodeRunResult(
+                                status=WorkflowNodeExecutionStatus.FAILED,
+                                error=event.error,
+                                metadata={
+                                    NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens
+                                },
+                            )
+                        )
+                        return
+                    else:
+                        yield self._handle_event_metadata(event=cast(InNodeEvent, event), iter_run_index=current_index)
+
+                # Remove all nodes outputs from variable pool
+                for node_id in loop_graph.node_ids:
+                    variable_pool.remove([node_id])
+
+                if check_break_result:
+                    break
+
+                # Move to next loop
+                next_index = current_index + 1
+                variable_pool.add([self.node_id, "index"], next_index)
+
+                yield LoopRunNextEvent(
+                    loop_id=self.id,
+                    loop_node_id=self.node_id,
+                    loop_node_type=self.node_type,
+                    loop_node_data=self.node_data,
+                    index=next_index,
+                    pre_loop_output=None,
+                )
+
+            # Loop completed successfully
+            yield LoopRunSucceededEvent(
+                loop_id=self.id,
+                loop_node_id=self.node_id,
+                loop_node_type=self.node_type,
+                loop_node_data=self.node_data,
+                start_at=start_at,
+                inputs=inputs,
+                steps=loop_count,
+                metadata={
+                    NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens,
+                    "completed_reason": "loop_break" if check_break_result else "loop_completed",
+                },
+            )
+
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                    metadata={NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens},
+                )
+            )
+
+        except Exception as e:
+            # Loop failed
+            logger.exception("Loop run failed")
+            yield LoopRunFailedEvent(
+                loop_id=self.id,
+                loop_node_id=self.node_id,
+                loop_node_type=self.node_type,
+                loop_node_data=self.node_data,
+                start_at=start_at,
+                inputs=inputs,
+                steps=loop_count,
+                metadata={
+                    "total_tokens": graph_engine.graph_runtime_state.total_tokens,
+                    "completed_reason": "error",
+                },
+                error=str(e),
+            )
+
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    error=str(e),
+                    metadata={NodeRunMetadataKey.TOTAL_TOKENS: graph_engine.graph_runtime_state.total_tokens},
+                )
+            )
+
+        finally:
+            # Clean up
+            variable_pool.remove([self.node_id, "index"])
+
+    def _handle_event_metadata(
+        self,
+        *,
+        event: BaseNodeEvent | InNodeEvent,
+        iter_run_index: int,
+    ) -> NodeRunStartedEvent | BaseNodeEvent | InNodeEvent:
+        """
+        add iteration metadata to event.
+        """
+        if not isinstance(event, BaseNodeEvent):
+            return event
+        if event.route_node_state.node_run_result:
+            metadata = event.route_node_state.node_run_result.metadata
+            if not metadata:
+                metadata = {}
+            if NodeRunMetadataKey.LOOP_ID not in metadata:
+                metadata = {
+                    **metadata,
+                    NodeRunMetadataKey.LOOP_ID: self.node_id,
+                    NodeRunMetadataKey.LOOP_INDEX: iter_run_index,
+                }
+                event.route_node_state.node_run_result.metadata = metadata
+        return event
 
     @classmethod
-    def get_conditions(cls, node_config: dict[str, Any]) -> list[Condition]:
+    def _extract_variable_selector_to_variable_mapping(
+        cls,
+        *,
+        graph_config: Mapping[str, Any],
+        node_id: str,
+        node_data: LoopNodeData,
+    ) -> Mapping[str, Sequence[str]]:
         """
-        Get conditions.
+        Extract variable selector to variable mapping
+        :param graph_config: graph config
+        :param node_id: node id
+        :param node_data: node data
+        :return:
         """
-        node_id = node_config.get("id")
-        if not node_id:
-            return []
+        variable_mapping = {}
 
-        # TODO waiting for implementation
-        return [
-            Condition(  # type: ignore
-                variable_selector=[node_id, "index"],
-                comparison_operator="≤",
-                value_type="value_selector",
-                value_selector=[],
-            )
-        ]
+        # init graph
+        loop_graph = Graph.init(graph_config=graph_config, root_node_id=node_data.start_node_id)
+
+        if not loop_graph:
+            raise ValueError("loop graph not found")
+
+        for sub_node_id, sub_node_config in loop_graph.node_id_config_mapping.items():
+            if sub_node_config.get("data", {}).get("loop_id") != node_id:
+                continue
+
+            # variable selector to variable mapping
+            try:
+                # Get node class
+                from core.workflow.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
+
+                node_type = NodeType(sub_node_config.get("data", {}).get("type"))
+                if node_type not in NODE_TYPE_CLASSES_MAPPING:
+                    continue
+                node_version = sub_node_config.get("data", {}).get("version", "1")
+                node_cls = NODE_TYPE_CLASSES_MAPPING[node_type][node_version]
+
+                sub_node_variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
+                    graph_config=graph_config, config=sub_node_config
+                )
+                sub_node_variable_mapping = cast(dict[str, Sequence[str]], sub_node_variable_mapping)
+            except NotImplementedError:
+                sub_node_variable_mapping = {}
+
+            # remove loop variables
+            sub_node_variable_mapping = {
+                sub_node_id + "." + key: value
+                for key, value in sub_node_variable_mapping.items()
+                if value[0] != node_id
+            }
+
+            variable_mapping.update(sub_node_variable_mapping)
+
+        # remove variable out from loop
+        variable_mapping = {
+            key: value for key, value in variable_mapping.items() if value[0] not in loop_graph.node_ids
+        }
+
+        return variable_mapping

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

@@ -0,0 +1,20 @@
+from core.workflow.entities.node_entities import NodeRunResult
+from core.workflow.nodes.base import BaseNode
+from core.workflow.nodes.enums import NodeType
+from core.workflow.nodes.loop.entities import LoopStartNodeData
+from models.workflow import WorkflowNodeExecutionStatus
+
+
+class LoopStartNode(BaseNode[LoopStartNodeData]):
+    """
+    Loop Start Node.
+    """
+
+    _node_data_cls = LoopStartNodeData
+    _node_type = NodeType.LOOP_START
+
+    def _run(self) -> NodeRunResult:
+        """
+        Run the node.
+        """
+        return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED)

api/core/workflow/nodes/node_mapping.py

@@ -1,5 +1,6 @@
 from collections.abc import Mapping
 
+from core.workflow.nodes.agent.agent_node import AgentNode
 from core.workflow.nodes.answer import AnswerNode
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.code import CodeNode
@@ -12,6 +13,7 @@ from core.workflow.nodes.iteration import IterationNode, IterationStartNode
 from core.workflow.nodes.knowledge_retrieval import KnowledgeRetrievalNode
 from core.workflow.nodes.list_operator import ListOperatorNode
 from core.workflow.nodes.llm import LLMNode
+from core.workflow.nodes.loop import LoopNode, LoopStartNode
 from core.workflow.nodes.parameter_extractor import ParameterExtractorNode
 from core.workflow.nodes.question_classifier import QuestionClassifierNode
 from core.workflow.nodes.start import StartNode
@@ -84,6 +86,14 @@ NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[BaseNode]]] = {
         LATEST_VERSION: IterationStartNode,
         "1": IterationStartNode,
     },
+    NodeType.LOOP: {
+        LATEST_VERSION: LoopNode,
+        "1": LoopNode,
+    },
+    NodeType.LOOP_START: {
+        LATEST_VERSION: LoopStartNode,
+        "1": LoopStartNode,
+    },
     NodeType.PARAMETER_EXTRACTOR: {
         LATEST_VERSION: ParameterExtractorNode,
         "1": ParameterExtractorNode,
@@ -101,4 +111,8 @@ NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[BaseNode]]] = {
         LATEST_VERSION: ListOperatorNode,
         "1": ListOperatorNode,
     },
+    NodeType.AGENT: {
+        LATEST_VERSION: AgentNode,
+        "1": AgentNode,
+    },
 }

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

@@ -7,6 +7,7 @@ from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEnti
 from core.file import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
+from core.model_runtime.entities import ImagePromptMessageContent
 from core.model_runtime.entities.llm_entities import LLMResult, LLMUsage
 from core.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
@@ -129,6 +130,7 @@ class ParameterExtractorNode(LLMNode):
                 model_config=model_config,
                 memory=memory,
                 files=files,
+                vision_detail=node_data.vision.configs.detail,
             )
         else:
             # use prompt engineering
@@ -139,6 +141,7 @@ class ParameterExtractorNode(LLMNode):
                 model_config=model_config,
                 memory=memory,
                 files=files,
+                vision_detail=node_data.vision.configs.detail,
             )
 
             prompt_message_tools = []
@@ -244,8 +247,8 @@ class ParameterExtractorNode(LLMNode):
         if not isinstance(invoke_result, LLMResult):
             raise InvalidInvokeResultError(f"Invalid invoke result: {invoke_result}")
 
-        text = invoke_result.message.content
-        if not isinstance(text, str | None):
+        text = invoke_result.message.content or ""
+        if not isinstance(text, str):
             raise InvalidTextContentTypeError(f"Invalid text content type: {type(text)}. Expected str.")
 
         usage = invoke_result.usage
@@ -267,6 +270,7 @@ class ParameterExtractorNode(LLMNode):
         model_config: ModelConfigWithCredentialsEntity,
         memory: Optional[TokenBufferMemory],
         files: Sequence[File],
+        vision_detail: Optional[ImagePromptMessageContent.DETAIL] = None,
     ) -> tuple[list[PromptMessage], list[PromptMessageTool]]:
         """
         Generate function call prompt.
@@ -289,6 +293,7 @@ class ParameterExtractorNode(LLMNode):
             memory_config=node_data.memory,
             memory=None,
             model_config=model_config,
+            image_detail_config=vision_detail,
         )
 
         # find last user message
@@ -347,6 +352,7 @@ class ParameterExtractorNode(LLMNode):
         model_config: ModelConfigWithCredentialsEntity,
         memory: Optional[TokenBufferMemory],
         files: Sequence[File],
+        vision_detail: Optional[ImagePromptMessageContent.DETAIL] = None,
     ) -> list[PromptMessage]:
         """
         Generate prompt engineering prompt.
@@ -361,6 +367,7 @@ class ParameterExtractorNode(LLMNode):
                 model_config=model_config,
                 memory=memory,
                 files=files,
+                vision_detail=vision_detail,
             )
         elif model_mode == ModelMode.CHAT:
             return self._generate_prompt_engineering_chat_prompt(
@@ -370,6 +377,7 @@ class ParameterExtractorNode(LLMNode):
                 model_config=model_config,
                 memory=memory,
                 files=files,
+                vision_detail=vision_detail,
             )
         else:
             raise InvalidModelModeError(f"Invalid model mode: {model_mode}")
@@ -382,6 +390,7 @@ class ParameterExtractorNode(LLMNode):
         model_config: ModelConfigWithCredentialsEntity,
         memory: Optional[TokenBufferMemory],
         files: Sequence[File],
+        vision_detail: Optional[ImagePromptMessageContent.DETAIL] = None,
     ) -> list[PromptMessage]:
         """
         Generate completion prompt.
@@ -402,6 +411,7 @@ class ParameterExtractorNode(LLMNode):
             memory_config=node_data.memory,
             memory=memory,
             model_config=model_config,
+            image_detail_config=vision_detail,
         )
 
         return prompt_messages
@@ -414,6 +424,7 @@ class ParameterExtractorNode(LLMNode):
         model_config: ModelConfigWithCredentialsEntity,
         memory: Optional[TokenBufferMemory],
         files: Sequence[File],
+        vision_detail: Optional[ImagePromptMessageContent.DETAIL] = None,
     ) -> list[PromptMessage]:
         """
         Generate chat prompt.
@@ -441,6 +452,7 @@ class ParameterExtractorNode(LLMNode):
             memory_config=node_data.memory,
             memory=None,
             model_config=model_config,
+            image_detail_config=vision_detail,
         )
 
         # find last user message

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

@@ -1,6 +1,3 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
-
 from core.workflow.constants import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.nodes.base import BaseNode
@@ -23,13 +20,3 @@ class StartNode(BaseNode[StartNodeData]):
             node_inputs[SYSTEM_VARIABLE_NODE_ID + "." + var] = system_inputs[var]
 
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, inputs=node_inputs, outputs=node_inputs)
-
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls,
-        *,
-        graph_config: Mapping[str, Any],
-        node_id: str,
-        node_data: StartNodeData,
-    ) -> Mapping[str, Sequence[str]]:
-        return {}

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

@@ -3,16 +3,18 @@ from typing import Any, Literal, Union
 from pydantic import BaseModel, field_validator
 from pydantic_core.core_schema import ValidationInfo
 
-from core.workflow.nodes.base import BaseNodeData
+from core.tools.entities.tool_entities import ToolProviderType
+from core.workflow.nodes.base.entities import BaseNodeData
 
 
 class ToolEntity(BaseModel):
     provider_id: str
-    provider_type: Literal["builtin", "api", "workflow"]
+    provider_type: ToolProviderType
     provider_name: str  # redundancy
     tool_name: str
     tool_label: str  # redundancy
     tool_configurations: dict[str, Any]
+    plugin_unique_identifier: str | None = None  # redundancy
 
     @field_validator("tool_configurations", mode="before")
     @classmethod

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

@@ -1,24 +1,32 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
-from uuid import UUID
+from collections.abc import Generator, Mapping, Sequence
+from typing import Any, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
-from core.file import File, FileTransferMethod, FileType
+from core.file import File, FileTransferMethod
+from core.plugin.manager.exc import PluginDaemonClientSideError
+from core.plugin.manager.plugin import PluginInstallationManager
 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
+from core.variables.variables import ArrayAnyVariable
 from core.workflow.entities.node_entities import NodeRunMetadataKey, NodeRunResult
 from core.workflow.entities.variable_pool import VariablePool
+from core.workflow.enums import SystemVariableKey
+from core.workflow.graph_engine.entities.event import AgentLogEvent
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
+from core.workflow.nodes.event import RunCompletedEvent, RunStreamChunkEvent
 from core.workflow.utils.variable_template_parser import VariableTemplateParser
 from extensions.ext_database import db
 from factories import file_factory
 from models import ToolFile
 from models.workflow import WorkflowNodeExecutionStatus
+from services.tools.builtin_tools_manage_service import BuiltinToolManageService
 
 from .entities import ToolNodeData
 from .exc import (
@@ -36,11 +44,18 @@ class ToolNode(BaseNode[ToolNodeData]):
     _node_data_cls = ToolNodeData
     _node_type = NodeType.TOOL
 
-    def _run(self) -> NodeRunResult:
+    def _run(self) -> Generator:
+        """
+        Run the tool node
+        """
+
+        node_data = cast(ToolNodeData, self.node_data)
+
         # fetch tool icon
         tool_info = {
-            "provider_type": self.node_data.provider_type,
-            "provider_id": self.node_data.provider_id,
+            "provider_type": node_data.provider_type.value,
+            "provider_id": node_data.provider_id,
+            "plugin_unique_identifier": node_data.plugin_unique_identifier,
         }
 
         # get tool runtime
@@ -51,18 +66,19 @@ class ToolNode(BaseNode[ToolNodeData]):
                 self.tenant_id, self.app_id, self.node_id, self.node_data, self.invoke_from
             )
         except ToolNodeError as e:
-            return NodeRunResult(
-                status=WorkflowNodeExecutionStatus.FAILED,
-                inputs={},
-                metadata={
-                    NodeRunMetadataKey.TOOL_INFO: tool_info,
-                },
-                error=f"Failed to get tool runtime: {str(e)}",
-                error_type=type(e).__name__,
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs={},
+                    metadata={NodeRunMetadataKey.TOOL_INFO: tool_info},
+                    error=f"Failed to get tool runtime: {str(e)}",
+                    error_type=type(e).__name__,
+                )
             )
+            return
 
         # get parameters
-        tool_parameters = tool_runtime.parameters or []
+        tool_parameters = tool_runtime.get_merged_runtime_parameters() or []
         parameters = self._generate_parameters(
             tool_parameters=tool_parameters,
             variable_pool=self.graph_runtime_state.variable_pool,
@@ -75,52 +91,46 @@ class ToolNode(BaseNode[ToolNodeData]):
             for_log=True,
         )
 
+        # get conversation id
+        conversation_id = self.graph_runtime_state.variable_pool.get(["sys", SystemVariableKey.CONVERSATION_ID])
+
         try:
-            messages = ToolEngine.workflow_invoke(
+            message_stream = ToolEngine.generic_invoke(
                 tool=tool_runtime,
                 tool_parameters=parameters,
                 user_id=self.user_id,
                 workflow_tool_callback=DifyWorkflowCallbackHandler(),
                 workflow_call_depth=self.workflow_call_depth,
                 thread_pool_id=self.thread_pool_id,
+                app_id=self.app_id,
+                conversation_id=conversation_id.text if conversation_id else None,
             )
         except ToolNodeError as e:
-            return NodeRunResult(
-                status=WorkflowNodeExecutionStatus.FAILED,
-                inputs=parameters_for_log,
-                metadata={
-                    NodeRunMetadataKey.TOOL_INFO: tool_info,
-                },
-                error=f"Failed to invoke tool: {str(e)}",
-                error_type=type(e).__name__,
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs=parameters_for_log,
+                    metadata={NodeRunMetadataKey.TOOL_INFO: tool_info},
+                    error=f"Failed to invoke tool: {str(e)}",
+                    error_type=type(e).__name__,
+                )
             )
-        except Exception as e:
-            return NodeRunResult(
-                status=WorkflowNodeExecutionStatus.FAILED,
-                inputs=parameters_for_log,
-                metadata={
-                    NodeRunMetadataKey.TOOL_INFO: tool_info,
-                },
-                error=f"Failed to invoke tool: {str(e)}",
-                error_type="UnknownError",
+            return
+
+        try:
+            # convert tool messages
+            yield from self._transform_message(message_stream, tool_info, parameters_for_log)
+        except (PluginDaemonClientSideError, ToolInvokeError) as e:
+            yield RunCompletedEvent(
+                run_result=NodeRunResult(
+                    status=WorkflowNodeExecutionStatus.FAILED,
+                    inputs=parameters_for_log,
+                    metadata={NodeRunMetadataKey.TOOL_INFO: tool_info},
+                    error=f"Failed to transform tool message: {str(e)}",
+                    error_type=type(e).__name__,
+                )
             )
 
-        # convert tool messages
-        plain_text, files, json = self._convert_tool_messages(messages)
-
-        return NodeRunResult(
-            status=WorkflowNodeExecutionStatus.SUCCEEDED,
-            outputs={
-                "text": plain_text,
-                "files": files,
-                "json": json,
-            },
-            metadata={
-                NodeRunMetadataKey.TOOL_INFO: tool_info,
-            },
-            inputs=parameters_for_log,
-        )
-
     def _generate_parameters(
         self,
         *,
@@ -128,7 +138,7 @@ class ToolNode(BaseNode[ToolNodeData]):
         variable_pool: VariablePool,
         node_data: ToolNodeData,
         for_log: bool = False,
-    ) -> Mapping[str, Any]:
+    ) -> dict[str, Any]:
         """
         Generate parameters based on the given tool parameters, variable pool, and node data.
 
@@ -164,37 +174,52 @@ class ToolNode(BaseNode[ToolNodeData]):
 
         return result
 
-    def _convert_tool_messages(
+    def _fetch_files(self, variable_pool: VariablePool) -> list[File]:
+        variable = variable_pool.get(["sys", SystemVariableKey.FILES.value])
+        assert isinstance(variable, ArrayAnyVariable | ArrayAnySegment)
+        return list(variable.value) if variable else []
+
+    def _transform_message(
         self,
-        messages: list[ToolInvokeMessage],
-    ):
+        messages: Generator[ToolInvokeMessage, None, None],
+        tool_info: Mapping[str, Any],
+        parameters_for_log: dict[str, Any],
+    ) -> Generator:
         """
         Convert ToolInvokeMessages into tuple[plain_text, files]
         """
         # transform message and handle file storage
-        messages = ToolFileMessageTransformer.transform_tool_invoke_messages(
+        message_stream = ToolFileMessageTransformer.transform_tool_invoke_messages(
             messages=messages,
             user_id=self.user_id,
             tenant_id=self.tenant_id,
             conversation_id=None,
         )
-        # extract plain text and files
-        files = self._extract_tool_response_binary(messages)
-        plain_text = self._extract_tool_response_text(messages)
-        json = self._extract_tool_response_json(messages)
 
-        return plain_text, files, json
+        text = ""
+        files: list[File] = []
+        json: list[dict] = []
+
+        agent_logs: list[AgentLogEvent] = []
+        agent_execution_metadata: Mapping[NodeRunMetadataKey, Any] = {}
+
+        variables: dict[str, Any] = {}
+
+        for message in message_stream:
+            if message.type in {
+                ToolInvokeMessage.MessageType.IMAGE_LINK,
+                ToolInvokeMessage.MessageType.BINARY_LINK,
+                ToolInvokeMessage.MessageType.IMAGE,
+            }:
+                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
+
+                url = message.message.text
+                if message.meta:
+                    transfer_method = message.meta.get("transfer_method", FileTransferMethod.TOOL_FILE)
+                else:
+                    transfer_method = FileTransferMethod.TOOL_FILE
 
-    def _extract_tool_response_binary(self, tool_response: list[ToolInvokeMessage]) -> list[File]:
-        """
-        Extract tool response binary
-        """
-        result = []
-        for response in tool_response:
-            if response.type in {ToolInvokeMessage.MessageType.IMAGE_LINK, ToolInvokeMessage.MessageType.IMAGE}:
-                url = str(response.message) if response.message else None
                 tool_file_id = str(url).split("/")[-1].split(".")[0]
-                transfer_method = response.meta.get("transfer_method", FileTransferMethod.TOOL_FILE)
 
                 with Session(db.engine) as session:
                     stmt = select(ToolFile).where(ToolFile.id == tool_file_id)
@@ -204,7 +229,7 @@ class ToolNode(BaseNode[ToolNodeData]):
 
                 mapping = {
                     "tool_file_id": tool_file_id,
-                    "type": FileType.IMAGE,
+                    "type": file_factory.get_file_type_by_mime_type(tool_file.mimetype),
                     "transfer_method": transfer_method,
                     "url": url,
                 }
@@ -212,70 +237,142 @@ class ToolNode(BaseNode[ToolNodeData]):
                     mapping=mapping,
                     tenant_id=self.tenant_id,
                 )
-                result.append(file)
-            elif response.type == ToolInvokeMessage.MessageType.BLOB:
-                tool_file_id = str(response.message).split("/")[-1].split(".")[0]
+                files.append(file)
+            elif message.type == ToolInvokeMessage.MessageType.BLOB:
+                # get tool file id
+                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
+                assert message.meta
+
+                tool_file_id = message.message.text.split("/")[-1].split(".")[0]
                 with Session(db.engine) as session:
                     stmt = select(ToolFile).where(ToolFile.id == tool_file_id)
                     tool_file = session.scalar(stmt)
                     if tool_file is None:
-                        raise ValueError(f"tool file {tool_file_id} not exists")
+                        raise ToolFileError(f"tool file {tool_file_id} not exists")
+
                 mapping = {
                     "tool_file_id": tool_file_id,
                     "transfer_method": FileTransferMethod.TOOL_FILE,
                 }
-                file = file_factory.build_from_mapping(
-                    mapping=mapping,
-                    tenant_id=self.tenant_id,
+
+                files.append(
+                    file_factory.build_from_mapping(
+                        mapping=mapping,
+                        tenant_id=self.tenant_id,
+                    )
                 )
-                result.append(file)
-            elif response.type == ToolInvokeMessage.MessageType.LINK:
-                url = str(response.message)
-                transfer_method = FileTransferMethod.TOOL_FILE
-                tool_file_id = url.split("/")[-1].split(".")[0]
-                try:
-                    UUID(tool_file_id)
-                except ValueError:
-                    raise ToolFileError(f"cannot extract tool file id from url {url}")
-                with Session(db.engine) as session:
-                    stmt = select(ToolFile).where(ToolFile.id == tool_file_id)
-                    tool_file = session.scalar(stmt)
-                    if tool_file is None:
-                        raise ToolFileError(f"Tool file {tool_file_id} does not exist")
-                mapping = {
-                    "tool_file_id": tool_file_id,
-                    "transfer_method": transfer_method,
-                    "url": url,
-                }
-                file = file_factory.build_from_mapping(
-                    mapping=mapping,
-                    tenant_id=self.tenant_id,
+            elif message.type == ToolInvokeMessage.MessageType.TEXT:
+                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
+                text += message.message.text
+                yield RunStreamChunkEvent(
+                    chunk_content=message.message.text, from_variable_selector=[self.node_id, "text"]
                 )
-                result.append(file)
+            elif message.type == ToolInvokeMessage.MessageType.JSON:
+                assert isinstance(message.message, ToolInvokeMessage.JsonMessage)
+                if self.node_type == NodeType.AGENT:
+                    msg_metadata = message.message.json_object.pop("execution_metadata", {})
+                    agent_execution_metadata = {
+                        key: value
+                        for key, value in msg_metadata.items()
+                        if key in NodeRunMetadataKey.__members__.values()
+                    }
+                json.append(message.message.json_object)
+            elif message.type == ToolInvokeMessage.MessageType.LINK:
+                assert isinstance(message.message, ToolInvokeMessage.TextMessage)
+                stream_text = f"Link: {message.message.text}\n"
+                text += stream_text
+                yield RunStreamChunkEvent(chunk_content=stream_text, from_variable_selector=[self.node_id, "text"])
+            elif message.type == ToolInvokeMessage.MessageType.VARIABLE:
+                assert isinstance(message.message, ToolInvokeMessage.VariableMessage)
+                variable_name = message.message.variable_name
+                variable_value = message.message.variable_value
+                if message.message.stream:
+                    if not isinstance(variable_value, str):
+                        raise ValueError("When 'stream' is True, 'variable_value' must be a string.")
+                    if variable_name not in variables:
+                        variables[variable_name] = ""
+                    variables[variable_name] += variable_value
 
-            elif response.type == ToolInvokeMessage.MessageType.FILE:
-                assert response.meta is not None
-                result.append(response.meta["file"])
+                    yield RunStreamChunkEvent(
+                        chunk_content=variable_value, from_variable_selector=[self.node_id, variable_name]
+                    )
+                else:
+                    variables[variable_name] = variable_value
+            elif message.type == ToolInvokeMessage.MessageType.FILE:
+                assert message.meta is not None
+                files.append(message.meta["file"])
+            elif message.type == ToolInvokeMessage.MessageType.LOG:
+                assert isinstance(message.message, ToolInvokeMessage.LogMessage)
+                if message.message.metadata:
+                    icon = tool_info.get("icon", "")
+                    dict_metadata = dict(message.message.metadata)
+                    if dict_metadata.get("provider"):
+                        manager = PluginInstallationManager()
+                        plugins = manager.list_plugins(self.tenant_id)
+                        try:
+                            current_plugin = next(
+                                plugin
+                                for plugin in plugins
+                                if f"{plugin.plugin_id}/{plugin.name}" == dict_metadata["provider"]
+                            )
+                            icon = current_plugin.declaration.icon
+                        except StopIteration:
+                            pass
+                        try:
+                            builtin_tool = next(
+                                provider
+                                for provider in BuiltinToolManageService.list_builtin_tools(
+                                    self.user_id,
+                                    self.tenant_id,
+                                )
+                                if provider.name == dict_metadata["provider"]
+                            )
+                            icon = builtin_tool.icon
+                        except StopIteration:
+                            pass
 
-        return result
+                        dict_metadata["icon"] = icon
+                        message.message.metadata = dict_metadata
+                agent_log = AgentLogEvent(
+                    id=message.message.id,
+                    node_execution_id=self.id,
+                    parent_id=message.message.parent_id,
+                    error=message.message.error,
+                    status=message.message.status.value,
+                    data=message.message.data,
+                    label=message.message.label,
+                    metadata=message.message.metadata,
+                    node_id=self.node_id,
+                )
 
-    def _extract_tool_response_text(self, tool_response: list[ToolInvokeMessage]) -> str:
-        """
-        Extract tool response text
-        """
-        return "\n".join(
-            [
-                str(message.message)
-                if message.type == ToolInvokeMessage.MessageType.TEXT
-                else f"Link: {str(message.message)}"
-                for message in tool_response
-                if message.type in {ToolInvokeMessage.MessageType.TEXT, ToolInvokeMessage.MessageType.LINK}
-            ]
+                # check if the agent log is already in the list
+                for log in agent_logs:
+                    if log.id == agent_log.id:
+                        # update the log
+                        log.data = agent_log.data
+                        log.status = agent_log.status
+                        log.error = agent_log.error
+                        log.label = agent_log.label
+                        log.metadata = agent_log.metadata
+                        break
+                else:
+                    agent_logs.append(agent_log)
+
+                yield agent_log
+
+        yield RunCompletedEvent(
+            run_result=NodeRunResult(
+                status=WorkflowNodeExecutionStatus.SUCCEEDED,
+                outputs={"text": text, "files": files, "json": json, **variables},
+                metadata={
+                    **agent_execution_metadata,
+                    NodeRunMetadataKey.TOOL_INFO: tool_info,
+                    NodeRunMetadataKey.AGENT_LOG: agent_logs,
+                },
+                inputs=parameters_for_log,
+            )
         )
 
-    def _extract_tool_response_json(self, tool_response: list[ToolInvokeMessage]):
-        return [message.message for message in tool_response if message.type == ToolInvokeMessage.MessageType.JSON]
-
     @classmethod
     def _extract_variable_selector_to_variable_mapping(
         cls,
@@ -295,7 +392,8 @@ class ToolNode(BaseNode[ToolNodeData]):
         for parameter_name in node_data.tool_parameters:
             input = node_data.tool_parameters[parameter_name]
             if input.type == "mixed":
-                selectors = VariableTemplateParser(str(input.value)).extract_variable_selectors()
+                assert isinstance(input.value, str)
+                selectors = VariableTemplateParser(input.value).extract_variable_selectors()
                 for selector in selectors:
                     result[selector.variable] = selector.value_selector
             elif input.type == "variable":

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

@@ -1,6 +1,3 @@
-from collections.abc import Mapping, Sequence
-from typing import Any
-
 from core.workflow.entities.node_entities import NodeRunResult
 from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
@@ -36,16 +33,3 @@ class VariableAggregatorNode(BaseNode[VariableAssignerNodeData]):
                         break
 
         return NodeRunResult(status=WorkflowNodeExecutionStatus.SUCCEEDED, outputs=outputs, inputs=inputs)
-
-    @classmethod
-    def _extract_variable_selector_to_variable_mapping(
-        cls, *, graph_config: Mapping[str, Any], node_id: str, node_data: VariableAssignerNodeData
-    ) -> Mapping[str, Sequence[str]]:
-        """
-        Extract variable selector to variable mapping
-        :param graph_config: graph config
-        :param node_id: node id
-        :param node_data: node data
-        :return:
-        """
-        return {}

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

@@ -1,6 +1,6 @@
 from core.variables import SegmentType, Variable
 from core.workflow.entities.node_entities import NodeRunResult
-from core.workflow.nodes.base import BaseNode, BaseNodeData
+from core.workflow.nodes.base import BaseNode
 from core.workflow.nodes.enums import NodeType
 from core.workflow.nodes.variable_assigner.common import helpers as common_helpers
 from core.workflow.nodes.variable_assigner.common.exc import VariableOperatorNodeError
@@ -11,7 +11,7 @@ from .node_data import VariableAssignerData, WriteMode
 
 
 class VariableAssignerNode(BaseNode[VariableAssignerData]):
-    _node_data_cls: type[BaseNodeData] = VariableAssignerData
+    _node_data_cls = VariableAssignerData
     _node_type = NodeType.VARIABLE_ASSIGNER
 
     def _run(self) -> NodeRunResult:

api/core/workflow/utils/condition/processor.py

@@ -64,6 +64,10 @@ class ConditionProcessor:
                     expected=expected_value,
                 )
             group_results.append(result)
+            # Implemented short-circuit evaluation for logical conditions
+            if (operator == "and" and not result) or (operator == "or" and result):
+                final_result = result
+                return input_conditions, group_results, final_result
 
         final_result = all(group_results) if operator == "and" else any(group_results)
         return input_conditions, group_results, final_result

api/core/workflow/workflow_entry.py

@@ -2,7 +2,7 @@ import logging
 import time
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import Any, Optional
+from typing import Any, Optional, cast
 
 from configs import dify_config
 from core.app.apps.base_app_queue_manager import GenerateTaskStoppedError
@@ -194,6 +194,105 @@ class WorkflowEntry:
             raise WorkflowNodeRunFailedError(node_instance=node_instance, error=str(e))
         return node_instance, generator
 
+    @classmethod
+    def run_free_node(
+        cls, node_data: dict, node_id: str, tenant_id: str, user_id: str, user_inputs: dict[str, Any]
+    ) -> tuple[BaseNode, Generator[NodeEvent | InNodeEvent, None, None]]:
+        """
+        Run free node
+
+        NOTE: only parameter_extractor/question_classifier are supported
+
+        :param node_data: node data
+        :param user_id: user id
+        :param user_inputs: user inputs
+        :return:
+        """
+        # generate a fake graph
+        node_config = {"id": node_id, "width": 114, "height": 514, "type": "custom", "data": node_data}
+        start_node_config = {
+            "id": "start",
+            "width": 114,
+            "height": 514,
+            "type": "custom",
+            "data": {
+                "type": NodeType.START.value,
+                "title": "Start",
+                "desc": "Start",
+            },
+        }
+        graph_dict = {
+            "nodes": [start_node_config, node_config],
+            "edges": [
+                {
+                    "source": "start",
+                    "target": node_id,
+                    "sourceHandle": "source",
+                    "targetHandle": "target",
+                }
+            ],
+        }
+
+        node_type = NodeType(node_data.get("type", ""))
+        if node_type not in {NodeType.PARAMETER_EXTRACTOR, NodeType.QUESTION_CLASSIFIER}:
+            raise ValueError(f"Node type {node_type} not supported")
+
+        node_cls = NODE_TYPE_CLASSES_MAPPING[node_type]["1"]
+        if not node_cls:
+            raise ValueError(f"Node class not found for node type {node_type}")
+
+        graph = Graph.init(graph_config=graph_dict)
+
+        # init variable pool
+        variable_pool = VariablePool(
+            system_variables={},
+            user_inputs={},
+            environment_variables=[],
+        )
+
+        node_cls = cast(type[BaseNode], node_cls)
+        # init workflow run state
+        node_instance: BaseNode = node_cls(
+            id=str(uuid.uuid4()),
+            config=node_config,
+            graph_init_params=GraphInitParams(
+                tenant_id=tenant_id,
+                app_id="",
+                workflow_type=WorkflowType.WORKFLOW,
+                workflow_id="",
+                graph_config=graph_dict,
+                user_id=user_id,
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph=graph,
+            graph_runtime_state=GraphRuntimeState(variable_pool=variable_pool, start_at=time.perf_counter()),
+        )
+
+        try:
+            # variable selector to variable mapping
+            try:
+                variable_mapping = node_cls.extract_variable_selector_to_variable_mapping(
+                    graph_config=graph_dict, config=node_config
+                )
+            except NotImplementedError:
+                variable_mapping = {}
+
+            cls.mapping_user_inputs_to_variable_pool(
+                variable_mapping=variable_mapping,
+                user_inputs=user_inputs,
+                variable_pool=variable_pool,
+                tenant_id=tenant_id,
+            )
+
+            # run node
+            generator = node_instance.run()
+
+            return node_instance, generator
+        except Exception as e:
+            raise WorkflowNodeRunFailedError(node_instance=node_instance, error=str(e))
+
     @staticmethod
     def handle_special_values(value: Optional[Mapping[str, Any]]) -> Mapping[str, Any] | None:
         result = WorkflowEntry._handle_special_values(value)