Introduce Plugins (#13836)

Signed-off-by: yihong0618 <zouzou0208@gmail.com> Signed-off-by: -LAN- <laipz8200@outlook.com> Signed-off-by: xhe <xw897002528@gmail.com> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: takatost <takatost@gmail.com> Co-authored-by: kurokobo <kuro664@gmail.com> Co-authored-by: Novice Lee <novicelee@NoviPro.local> Co-authored-by: zxhlyh <jasonapring2015@outlook.com> Co-authored-by: AkaraChen <akarachen@outlook.com> Co-authored-by: Yi <yxiaoisme@gmail.com> Co-authored-by: Joel <iamjoel007@gmail.com> Co-authored-by: JzoNg <jzongcode@gmail.com> Co-authored-by: twwu <twwu@dify.ai> Co-authored-by: Hiroshi Fujita <fujita-h@users.noreply.github.com> Co-authored-by: AkaraChen <85140972+AkaraChen@users.noreply.github.com> Co-authored-by: NFish <douxc512@gmail.com> Co-authored-by: Wu Tianwei <30284043+WTW0313@users.noreply.github.com> Co-authored-by: 非法操作 <hjlarry@163.com> Co-authored-by: Novice <857526207@qq.com> Co-authored-by: Hiroki Nagai <82458324+nagaihiroki-git@users.noreply.github.com> Co-authored-by: Gen Sato <52241300+halogen22@users.noreply.github.com> Co-authored-by: eux <euxuuu@gmail.com> Co-authored-by: huangzhuo1949 <167434202+huangzhuo1949@users.noreply.github.com> Co-authored-by: huangzhuo <huangzhuo1@xiaomi.com> Co-authored-by: lotsik <lotsik@mail.ru> Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com> Co-authored-by: nite-knite <nkCoding@gmail.com> Co-authored-by: Jyong <76649700+JohnJyong@users.noreply.github.com> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: gakkiyomi <gakkiyomi@aliyun.com> Co-authored-by: CN-P5 <heibai2006@gmail.com> Co-authored-by: CN-P5 <heibai2006@qq.com> Co-authored-by: Chuehnone <1897025+chuehnone@users.noreply.github.com> Co-authored-by: yihong <zouzou0208@gmail.com> Co-authored-by: Kevin9703 <51311316+Kevin9703@users.noreply.github.com> Co-authored-by: -LAN- <laipz8200@outlook.com> Co-authored-by: Boris Feld <lothiraldan@gmail.com> Co-authored-by: mbo <himabo@gmail.com> Co-authored-by: mabo <mabo@aeyes.ai> Co-authored-by: Warren Chen <warren.chen830@gmail.com> Co-authored-by: JzoNgKVO <27049666+JzoNgKVO@users.noreply.github.com> Co-authored-by: jiandanfeng <chenjh3@wangsu.com> Co-authored-by: zhu-an <70234959+xhdd123321@users.noreply.github.com> Co-authored-by: zhaoqingyu.1075 <zhaoqingyu.1075@bytedance.com> Co-authored-by: 海狸大師 <86974027+yenslife@users.noreply.github.com> Co-authored-by: Xu Song <xusong.vip@gmail.com> Co-authored-by: rayshaw001 <396301947@163.com> Co-authored-by: Ding Jiatong <dingjiatong@gmail.com> Co-authored-by: Bowen Liang <liangbowen@gf.com.cn> Co-authored-by: JasonVV <jasonwangiii@outlook.com> Co-authored-by: le0zh <newlight@qq.com> Co-authored-by: zhuxinliang <zhuxinliang@didiglobal.com> Co-authored-by: k-zaku <zaku99@outlook.jp> Co-authored-by: luckylhb90 <luckylhb90@gmail.com> Co-authored-by: hobo.l <hobo.l@binance.com> Co-authored-by: jiangbo721 <365065261@qq.com> Co-authored-by: 刘江波 <jiangbo721@163.com> Co-authored-by: Shun Miyazawa <34241526+miya@users.noreply.github.com> Co-authored-by: EricPan <30651140+Egfly@users.noreply.github.com> Co-authored-by: crazywoola <427733928@qq.com> Co-authored-by: sino <sino2322@gmail.com> Co-authored-by: Jhvcc <37662342+Jhvcc@users.noreply.github.com> Co-authored-by: lowell <lowell.hu@zkteco.in> Co-authored-by: Boris Polonsky <BorisPolonsky@users.noreply.github.com> Co-authored-by: Ademílson Tonato <ademilsonft@outlook.com> Co-authored-by: Ademílson Tonato <ademilson.tonato@refurbed.com> Co-authored-by: IWAI, Masaharu <iwaim.sub@gmail.com> Co-authored-by: Yueh-Po Peng (Yabi) <94939112+y10ab1@users.noreply.github.com> Co-authored-by: Jason <ggbbddjm@gmail.com> Co-authored-by: Xin Zhang <sjhpzx@gmail.com> Co-authored-by: yjc980121 <3898524+yjc980121@users.noreply.github.com> Co-authored-by: heyszt <36215648+hieheihei@users.noreply.github.com> Co-authored-by: Abdullah AlOsaimi <osaimiacc@gmail.com> Co-authored-by: Abdullah AlOsaimi <189027247+osaimi@users.noreply.github.com> Co-authored-by: Yingchun Lai <laiyingchun@apache.org> Co-authored-by: Hash Brown <hi@xzd.me> Co-authored-by: zuodongxu <192560071+zuodongxu@users.noreply.github.com> Co-authored-by: Masashi Tomooka <tmokmss@users.noreply.github.com> Co-authored-by: aplio <ryo.091219@gmail.com> Co-authored-by: Obada Khalili <54270856+obadakhalili@users.noreply.github.com> Co-authored-by: Nam Vu <zuzoovn@gmail.com> Co-authored-by: Kei YAMAZAKI <1715090+kei-yamazaki@users.noreply.github.com> Co-authored-by: TechnoHouse <13776377+deephbz@users.noreply.github.com> Co-authored-by: Riddhimaan-Senapati <114703025+Riddhimaan-Senapati@users.noreply.github.com> Co-authored-by: MaFee921 <31881301+2284730142@users.noreply.github.com> Co-authored-by: te-chan <t-nakanome@sakura-is.co.jp> Co-authored-by: HQidea <HQidea@users.noreply.github.com> Co-authored-by: Joshbly <36315710+Joshbly@users.noreply.github.com> Co-authored-by: xhe <xw897002528@gmail.com> Co-authored-by: weiwenyan-dev <154779315+weiwenyan-dev@users.noreply.github.com> Co-authored-by: ex_wenyan.wei <ex_wenyan.wei@tcl.com> Co-authored-by: engchina <12236799+engchina@users.noreply.github.com> Co-authored-by: engchina <atjapan2015@gmail.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: 呆萌闷油瓶 <253605712@qq.com> Co-authored-by: Kemal <kemalmeler@outlook.com> Co-authored-by: Lazy_Frog <4590648+lazyFrogLOL@users.noreply.github.com> Co-authored-by: Yi Xiao <54782454+YIXIAO0@users.noreply.github.com> Co-authored-by: Steven sun <98230804+Tuyohai@users.noreply.github.com> Co-authored-by: steven <sunzwj@digitalchina.com> Co-authored-by: Kalo Chin <91766386+fdb02983rhy@users.noreply.github.com> Co-authored-by: Katy Tao <34019945+KatyTao@users.noreply.github.com> Co-authored-by: depy <42985524+h4ckdepy@users.noreply.github.com> Co-authored-by: 胡春东 <gycm520@gmail.com> Co-authored-by: Junjie.M <118170653@qq.com> Co-authored-by: MuYu <mr.muzea@gmail.com> Co-authored-by: Naoki Takashima <39912547+takatea@users.noreply.github.com> Co-authored-by: Summer-Gu <37869445+gubinjie@users.noreply.github.com> Co-authored-by: Fei He <droxer.he@gmail.com> Co-authored-by: ybalbert001 <120714773+ybalbert001@users.noreply.github.com> Co-authored-by: Yuanbo Li <ybalbert@amazon.com> Co-authored-by: douxc <7553076+douxc@users.noreply.github.com> Co-authored-by: liuzhenghua <1090179900@qq.com> Co-authored-by: Wu Jiayang <62842862+Wu-Jiayang@users.noreply.github.com> Co-authored-by: Your Name <you@example.com> Co-authored-by: kimjion <45935338+kimjion@users.noreply.github.com> Co-authored-by: AugNSo <song.tiankai@icloud.com> Co-authored-by: llinvokerl <38915183+llinvokerl@users.noreply.github.com> Co-authored-by: liusurong.lsr <liusurong.lsr@alibaba-inc.com> Co-authored-by: Vasu Negi <vasu-negi@users.noreply.github.com> Co-authored-by: Hundredwz <1808096180@qq.com> Co-authored-by: Xiyuan Chen <52963600+GareArc@users.noreply.github.com>
2025-12-10 03:16:51 +08:00 · 2025-02-17 17:05:13 +08:00
parent 222df44d21
commit 403e2d58b9
3272 changed files with 66339 additions and 281594 deletions
--- a/api/core/workflow/nodes/agent/init.py
+++ b/api/core/workflow/nodes/agent/init.py
@@ -0,0 +1,3 @@
+from .agent_node import AgentNode
+
+__all__ = ["AgentNode"]
--- a/api/core/workflow/nodes/agent/agent_node.py
+++ b/api/core/workflow/nodes/agent/agent_node.py
@@ -0,0 +1,254 @@
+from ast import literal_eval
+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 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
+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"}:
+                segment_group = variable_pool.convert_template(str(agent_input.value))
+                parameter_value = segment_group.log if for_log else segment_group.text
+            else:
+                raise ValueError(f"Unknown agent input type '{agent_input.type}'")
+            value = parameter_value.strip()
+            if (parameter_value.startswith("{") and parameter_value.endswith("}")) or (
+                parameter_value.startswith("[") and parameter_value.endswith("]")
+            ):
+                value = literal_eval(parameter_value)  # transform string to python object
+            if parameter.type == "array[tools]":
+                value = cast(list[dict[str, Any]], value)
+                value = [tool for tool in value if tool.get("enabled", False)]
+
+            if not for_log:
+                if parameter.type == "array[tools]":
+                    value = cast(list[dict[str, Any]], value)
+                    tool_value = []
+                    for tool in value:
+                        entity = AgentToolEntity(
+                            provider_id=tool.get("provider_name", ""),
+                            provider_type=ToolProviderType.BUILT_IN,
+                            tool_name=tool.get("tool_name", ""),
+                            tool_parameters=tool.get("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
+                            )
+
+                        tool_value.append(
+                            {
+                                **tool_runtime.entity.model_dump(mode="json"),
+                                "runtime_parameters": tool_runtime.runtime.runtime_parameters,
+                            }
+                        )
+                    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
--- a/api/core/workflow/nodes/agent/entities.py
+++ b/api/core/workflow/nodes/agent/entities.py
@@ -0,0 +1,18 @@
+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]
--- a/api/core/workflow/nodes/code/code_node.py
+++ b/api/core/workflow/nodes/code/code_node.py
@@ -195,7 +195,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(
@@ -223,7 +223,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(
@@ -244,7 +244,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(
@@ -265,7 +265,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(
--- a/api/core/workflow/nodes/enums.py
+++ b/api/core/workflow/nodes/enums.py
@@ -22,6 +22,7 @@ class NodeType(StrEnum):
    VARIABLE_ASSIGNER = "assigner"
    DOCUMENT_EXTRACTOR = "document-extractor"
    LIST_OPERATOR = "list-operator"
+    AGENT = "agent"


 class ErrorStrategy(StrEnum):
--- a/api/core/workflow/nodes/iteration/iteration_node.py
+++ b/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()
--- a/api/core/workflow/nodes/llm/node.py
+++ b/api/core/workflow/nodes/llm/node.py
@@ -236,9 +236,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,
        )
--- a/api/core/workflow/nodes/node_mapping.py
+++ b/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
@@ -101,4 +102,8 @@ NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[BaseNode]]] = {
        LATEST_VERSION: ListOperatorNode,
        "1": ListOperatorNode,
    },
+    NodeType.AGENT: {
+        LATEST_VERSION: AgentNode,
+        "1": AgentNode,
+    },
 }
--- a/api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py
+++ b/api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py
@@ -244,8 +244,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
--- a/api/core/workflow/nodes/tool/entities.py
+++ b/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
--- a/api/core/workflow/nodes/tool/tool_node.py
+++ b/api/core/workflow/nodes/tool/tool_node.py
@@ -1,24 +1,31 @@
-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.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 +43,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 +65,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 +90,45 @@ 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 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)}",
+                )
            )

-        # 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 +136,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 +172,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 +227,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 +235,139 @@ 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
+                    }
+                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,
+                )

-    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 +387,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":