Agent 策略是一个定义了标准输入内容与输出格式的可扩展模板。通过开发具体 Agent 策略接口的功能代码，你可以实现众多不同的 Agent 策略如 CoT（思维链）/ ToT（思维树）/ GoT（思维图）/ BoT（思维骨架），实现一些诸如 Sementic Kernel 的复杂策略。

​
在 Manifest 内添加字段

在插件中添加 Agent 策略需要在 manifest.yaml 文件内新增 plugins.agent_strategies 字段，并且也需要定义 Agent 供应商，示例代码如下

version: 0.0.2
type: plugin
author: "langgenius"
name: "agent"
plugins:
  agent_strategies:
    - "provider/agent.yaml"

此处已省去 manifest 文件内部分无关的字段。如需了解 Manifest 的详细格式，请参考 通过清单文件定义插件信息 文档。

​
定义 Agent 供应商

随后，你需要新建 agent.yaml 文件并填写基础的 Agent 供应商信息。

identity:
  author: langgenius
  name: agent
  label:
    en_US: Agent
    zh_Hans: Agent
    pt_BR: Agent
  description:
    en_US: Agent
    zh_Hans: Agent
    pt_BR: Agent
  icon: icon.svg
strategies:
  - strategies/function_calling.yaml

其主要包含一些描述性质的基础内容，并且指明当前供应商包含哪些策略。在上述示例代码中仅指定了一个最基础的 function_calling.yaml 策略文件。

​
定义并实现 Agent 策略

​
定义

接下来需要定义能够实现 Agent 策略的代码。新建一个 function_calling.yaml 文件：

identity:
  name: function_calling
  author: Dify
  label:
    en_US: FunctionCalling
    zh_Hans: FunctionCalling
    pt_BR: FunctionCalling
description:
  en_US: Function Calling is a basic strategy for agent, model will use the tools provided to perform the task.
  zh_Hans: Function Calling 是一个基本的 Agent 策略，模型将使用提供的工具来执行任务。
  pt_BR: Function Calling is a basic strategy for agent, model will use the tools provided to perform the task.
parameters:
  - name: model
    type: model-selector
    scope: tool-call&llm
    required: true
    label:
      en_US: Model
      zh_Hans: 模型
      pt_BR: Model
  - name: tools
    type: array[tools]
    required: true
    label:
      en_US: Tools list
      zh_Hans: 工具列表
      pt_BR: Tools list
  - name: query
    type: string
    required: true
    label:
      en_US: Query
      zh_Hans: 用户提问
      pt_BR: Query
  - name: max_iterations
    type: number
    required: false
    default: 5
    label:
      en_US: Max Iterations
      zh_Hans: 最大迭代次数
      pt_BR: Max Iterations
    max: 50
    min: 1
extra:
  python:
    source: strategies/function_calling.py

代码格式类似 Tool 标准格式，定义了 model tools query max_iterations 等一共四个参数，以便于实现最基础的 Agent 策略。该代码的含义是可以允许用户选择模型和需要使用的工具，配置最大迭代次数并最终传入一个 query 后开始执行 Agent。

​
编写功能实现代码

获取参数

根据上文定义的四个参数，其中 model 类型参数为model-selector，tool 类型参数为特殊的 array[tools]。在参数中获取到的形式可以通过 SDK 中内置的 AgentModelConfig 和 list[ToolEntity]进行转换。

from dify_plugin.interfaces.agent import AgentModelConfig, AgentStrategy, ToolEntity

class FunctionCallingParams(BaseModel):
    query: str
    model: AgentModelConfig
    tools: list[ToolEntity] | None
    maximum_iterations: int = 3
    
 class FunctionCallingAgentStrategy(AgentStrategy):
    def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage]:
        """
        Run FunctionCall agent application
        """
        fc_params = FunctionCallingParams(**parameters)

调用模型

调用指定模型是 Agent 插件中必不可少的能力。通过 SDK 中的 session.model.invoke() 函数调用模型。可以从 model 中获取所需的传入参数。

invoke model 的方法签名示例代码：

def invoke(
        self,
        model_config: LLMModelConfig,
        prompt_messages: list[PromptMessage],
        tools: list[PromptMessageTool] | None = None,
        stop: list[str] | None = None,
        stream: bool = True,
    ) -> Generator[LLMResultChunk, None, None] | LLMResult:

需要传入模型信息 model_config，prompt 信息 prompt_messages 和工具信息 tools。

其中prompt_messages参数可以参考以下示例代码调用；而tool_messages则需要进行一定的转换。

请参考 invoke model 使用方法的示例代码：

from collections.abc import Generator
from typing import Any

from pydantic import BaseModel

from dify_plugin.entities.agent import AgentInvokeMessage
from dify_plugin.entities.model.llm import LLMModelConfig
from dify_plugin.entities.model.message import (
    PromptMessageTool,
    SystemPromptMessage,
    UserPromptMessage,
)
from dify_plugin.entities.tool import ToolParameter
from dify_plugin.interfaces.agent import AgentModelConfig, AgentStrategy, ToolEntity

class FunctionCallingParams(BaseModel):
    query: str
    instruction: str | None
    model: AgentModelConfig
    tools: list[ToolEntity] | None
    maximum_iterations: int = 3

class FunctionCallingAgentStrategy(AgentStrategy):
    def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage]:
        """
        Run FunctionCall agent application
        """
        # init params
        fc_params = FunctionCallingParams(**parameters)
        query = fc_params.query
        model = fc_params.model
        stop = fc_params.model.completion_params.get("stop", []) if fc_params.model.completion_params else []
        prompt_messages = [
            SystemPromptMessage(content="your system prompt message"),
            UserPromptMessage(content=query),
        ]
        tools = fc_params.tools
        prompt_messages_tools = self._init_prompt_tools(tools)

        # invoke llm
        chunks = self.session.model.llm.invoke(
            model_config=LLMModelConfig(**model.model_dump(mode="json")),
            prompt_messages=prompt_messages,
            stream=True,
            stop=stop,
            tools=prompt_messages_tools,
        )

    def _init_prompt_tools(self, tools: list[ToolEntity] | None) -> list[PromptMessageTool]:
        """
        Init tools
        """

        prompt_messages_tools = []
        for tool in tools or []:
            try:
                prompt_tool = self._convert_tool_to_prompt_message_tool(tool)
            except Exception:
                # api tool may be deleted
                continue

            # save prompt tool
            prompt_messages_tools.append(prompt_tool)

        return prompt_messages_tools

    def _convert_tool_to_prompt_message_tool(self, tool: ToolEntity) -> PromptMessageTool:
        """
        convert tool to prompt message tool
        """
        message_tool = PromptMessageTool(
            name=tool.identity.name,
            description=tool.description.llm if tool.description else "",
            parameters={
                "type": "object",
                "properties": {},
                "required": [],
            },
        )

        parameters = tool.parameters
        for parameter in parameters:
            if parameter.form != ToolParameter.ToolParameterForm.LLM:
                continue

            parameter_type = parameter.type
            if parameter.type in {
                ToolParameter.ToolParameterType.FILE,
                ToolParameter.ToolParameterType.FILES,
            }:
                continue
            enum = []
            if parameter.type == ToolParameter.ToolParameterType.SELECT:
                enum = [option.value for option in parameter.options] if parameter.options else []

            message_tool.parameters["properties"][parameter.name] = {
                "type": parameter_type,
                "description": parameter.llm_description or "",
            }

            if len(enum) > 0:
                message_tool.parameters["properties"][parameter.name]["enum"] = enum

            if parameter.required:
                message_tool.parameters["required"].append(parameter.name)

        return message_tool

调用工具

调用工具同样是 Agent 插件必不可少的能力。可以通过self.session.tool.invoke()进行调用。invoke tool 的方法签名示例代码：

def invoke(
        self,
        provider_type: ToolProviderType,
        provider: str,
        tool_name: str,
        parameters: dict[str, Any],
    ) -> Generator[ToolInvokeMessage, None, None]

必须的参数有 provider_type, provider, tool_name, parameters。其中 tool_name 和parameters在 Function Calling 中往往都由 LLM 生成。使用 invoke tool 的示例代码：

from dify_plugin.entities.tool import ToolProviderType

class FunctionCallingAgentStrategy(AgentStrategy):
    def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage]:
        """
        Run FunctionCall agent application
        """
        fc_params = FunctionCallingParams(**parameters)
        
        # tool_call_name and tool_call_args parameter is obtained from the output of LLM
        tool_instances = {tool.identity.name: tool for tool in fc_params.tools} if fc_params.tools else {}
        tool_instance = tool_instances[tool_call_name]
        tool_invoke_responses = self.session.tool.invoke(
            provider_type=ToolProviderType.BUILT_IN,
            provider=tool_instance.identity.provider,
            tool_name=tool_instance.identity.name,
            # add the default value
            parameters={**tool_instance.runtime_parameters, **tool_call_args},
        )

self.session.tool.invoke()函数的输出是一个 Generator，代表着同样需要进行流式解析。

解析方法请参考以下函数：

import json
from collections.abc import Generator
from typing import cast

from dify_plugin.entities.agent import AgentInvokeMessage
from dify_plugin.entities.tool import ToolInvokeMessage

def parse_invoke_response(tool_invoke_responses: Generator[AgentInvokeMessage]) -> str:
    result = ""
    for response in tool_invoke_responses:
        if response.type == ToolInvokeMessage.MessageType.TEXT:
            result += cast(ToolInvokeMessage.TextMessage, response.message).text
        elif response.type == ToolInvokeMessage.MessageType.LINK:
            result += (
                f"result link: {cast(ToolInvokeMessage.TextMessage, response.message).text}."
                + " please tell user to check it."
            )
        elif response.type in {
            ToolInvokeMessage.MessageType.IMAGE_LINK,
            ToolInvokeMessage.MessageType.IMAGE,
        }:
            result += (
                "image has been created and sent to user already, "
                + "you do not need to create it, just tell the user to check it now."
            )
        elif response.type == ToolInvokeMessage.MessageType.JSON:
            text = json.dumps(cast(ToolInvokeMessage.JsonMessage, response.message).json_object, ensure_ascii=False)
            result += f"tool response: {text}."
        else:
            result += f"tool response: {response.message!r}."
    return result

​
Log

如果你希望看到 Agent 思考的过程，除了通过查看正常返回的消息以外，还可以使用专门的接口实现以树状结构展示整个 Agent 的思考过程。

创建日志

  • 该接口创建并返回一个 AgentLogMessage，该 Message 表示日志中树的一个节点。
  • 如果有传入 parent 则表示该节点具备父节点。
  • 状态默认为”Success”（成功）。但如果你想要更好地展示任务执行过程，可以先设置状态为”start”来显示”正在执行”的日志，等任务完成后再将该日志的状态更新为”Success”。这样用户就能清楚地看到任务从开始到完成的整个过程。
  • label 将用于最终给用户展示日志标题。
def create_log_message(
        self,
        label: str,
        data: Mapping[str, Any],
        status: AgentInvokeMessage.LogMessage.LogStatus = AgentInvokeMessage.LogMessage.LogStatus.SUCCESS,
        parent: AgentInvokeMessage | None = None,
    ) -> AgentInvokeMessage

完成日志

如果在前一个步骤选择了 start 状态作为初始状态，可以使用完成日志的接口来更改状态。

def finish_log_message(
        self,
        log: AgentInvokeMessage,
        status: AgentInvokeMessage.LogMessage.LogStatus = AgentInvokeMessage.LogMessage.LogStatus.SUCCESS,
        error: Optional[str] = None,
    ) -> AgentInvokeMessage

实例

这个示例展示了一个简单的两步执行过程：首先输出一条”正在思考”的状态日志，然后完成实际的任务处理。

class FunctionCallingAgentStrategy(AgentStrategy):
    def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage]:
        thinking_log = self.create_log_message(
            data={
                "Query": parameters.get("query"),
            },
            label="Thinking",
            status=AgentInvokeMessage.LogMessage.LogStatus.START,
        )

        yield thinking_log

        llm_response = self.session.model.llm.invoke(
            model_config=LLMModelConfig(
                provider="openai",
                model="gpt-4o-mini",
                mode="chat",
                completion_params={},
            ),
            prompt_messages=[
                SystemPromptMessage(content="you are a helpful assistant"),
                UserPromptMessage(content=parameters.get("query")),
            ],
            stream=False,
            tools=[],
        )

        thinking_log = self.finish_log_message(
            log=thinking_log,
        )

        yield thinking_log

        yield self.create_text_message(text=llm_response.message.content)

​
相关资源

编辑此页面 | 提交问题