Agent戦略は、標準の入力内容と出力形式を定義する拡張可能なテンプレートです。具体的なAgent戦略インターフェースの機能コードを開発することで、CoT（思考の連鎖）/ ToT（思考の木）/ GoT（思考のグラフ）/ BoT（思考の骨格）など、さまざまなAgent戦略を実現し、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など合計4つのパラメータを定義し、最も基本的なAgent戦略の実装を容易にします。このコードの意味は、ユーザーがモデルと使用するツールを選択し、最大反復回数を設定し、最終的にqueryを渡してAgentの実行を開始できるようにすることです。
機能実装コードの作成
パラメータの取得
上記で定義された4つのパラメータに基づき、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_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="あなたのシステムプロンプトメッセージ"),
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ツールは削除された可能性があります
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,
# デフォルト値を追加
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"結果リンク: {cast(ToolInvokeMessage.TextMessage, response.message).text}。"
+ " ユーザーに確認するよう伝えてください。"
)
elif response.type in {
ToolInvokeMessage.MessageType.IMAGE_LINK,
ToolInvokeMessage.MessageType.IMAGE,
}:
result += (
"画像は既に作成されユーザーに送信済みです。"
+ "作成する必要はありません。ユーザーに今すぐ確認するよう伝えてください。"
)
elif response.type == ToolInvokeMessage.MessageType.JSON:
text = json.dumps(cast(ToolInvokeMessage.JsonMessage, response.message).json_object, ensure_ascii=False)
result += f"ツール応答: {text}。"
else:
result += f"ツール応答: {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
インスタンス
この例では、単純な2ステップの実行プロセスを示しています。まず「思考中」のステータスログを出力し、次に実際のタスク処理を完了します。
class FunctionCallingAgentStrategy(AgentStrategy):
def _invoke(self, parameters: dict[str, Any]) -> Generator[AgentInvokeMessage]:
thinking_log = self.create_log_message(
data={
"Query": parameters.get("query"),
},
label="思考中",
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="あなたは役立つアシスタントです"),
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)
関連リソース
このページを編集する | 問題を報告する
