AutoGen Core : コンポーネントガイド : ツール

ツールはアクションを遂行するためにエージェントにより実行できるコードです。ツールは、電卓のような単純な関数や、株価検索や天気予報のようなサードパーティのサービスへの API 呼び出しです。

AutoGen Core : コンポーネントガイド : ツール

作成 : クラスキャット・セールスインフォメーション
作成日時 : 04/16/2025

* 本記事は github microsoft/autogen の以下のページを独自に翻訳した上でまとめ直し、補足説明を加えています :

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

クラスキャット 人工知能 研究開発支援サービス ⭐️ リニューアルしました 😉

クラスキャット は人工知能に関する各種サービスを提供しています。お気軽にご相談ください :

  • 人工知能導入個別相談会(無償)実施中! [詳細]

  • 人工知能研究開発支援 [詳細]
    1. 自社特有情報を含むチャットボット構築支援
    2. 画像認識 (医療系含む) / 画像生成

  • PoC(概念実証)を失敗させないための支援 [詳細]

お問合せ : 下記までお願いします。

  • クラスキャット セールス・インフォメーション
  • sales-info@classcat.com
  • ClassCatJP

 

 

AutoGen Core : コンポーネントガイド : ツール

ツールはアクションを遂行するためにエージェントにより実行できるコードです。ツールは、電卓のような単純な関数や、株価検索や天気予報のようなサードパーティのサービスへの API 呼び出しです。AI エージェントのコンテキストでは、ツールはモデルにより生成された関数呼び出しに応じて、エージェントにより実行されるように設計されています。

AutoGen は autogen_core.tools モジュールに、カスタムツールの作成と実行のために一連の組み込みツールとユーティリティを提供します。

 

組み込みツール

組み込みツールの一つは PythonCodeExecutionTool で、これはエージェントの Python コード・スニペットの実行を可能にします。

ここにツールを作成して使用する方法を示します。

from autogen_core import CancellationToken
from autogen_ext.code_executors.docker import DockerCommandLineCodeExecutor
from autogen_ext.tools.code_execution import PythonCodeExecutionTool

# Create the tool.
code_executor = DockerCommandLineCodeExecutor()
await code_executor.start()
code_execution_tool = PythonCodeExecutionTool(code_executor)
cancellation_token = CancellationToken()

# Use the tool directly without an agent.
code = "print('Hello, world!')"
result = await code_execution_tool.run_json({"code": code}, cancellation_token)
print(code_execution_tool.return_value_as_string(result))
Hello, world!

DockerCommandLineCodeExecutor クラスは組み込みのコード executor で、docker コンテナのコマンドライン環境内のサブプロセスで Python コードスニペットを実行します。PythonCodeExecutionTool はコード executor をラップして Python コードスニペットを実行する単純なインターフェイスを提供します。

他の組み込みツールの例は :

 

カスタム関数ツール

ツールはまた特定のアクションを遂行する単純な Python 関数でもあり得ます。カスタム関数ツールを作成するには、Python 関数を作成して、それをラップする FunctionTool クラスを使用する必要があるだけです。

FunctionTool は説明と型アノテーションを使用して、LLM に指定の関数をいつどのように使用するかを伝えます。説明は関数の目的と意図されるユースケースについてコンテキストを提供する一方で、型アノテーションは想定パラメータと戻り値の型について LLM に通知します。

例えば、会社の株価を取得する単純なツールはこのようなものです :

import random

from autogen_core import CancellationToken
from autogen_core.tools import FunctionTool
from typing_extensions import Annotated


async def get_stock_price(ticker: str, date: Annotated[str, "Date in YYYY/MM/DD"]) -> float:
    # Returns a random stock price for demonstration purposes.
    return random.uniform(10, 200)


# Create a function tool.
stock_price_tool = FunctionTool(get_stock_price, description="Get the stock price.")

# Run the tool.
cancellation_token = CancellationToken()
result = await stock_price_tool.run_json({"ticker": "AAPL", "date": "2021/01/01"}, cancellation_token)

# Print the result.
print(stock_price_tool.return_value_as_string(result))
143.83831971965762

 

モデルクライアントによるツール呼び出し

AutoGen では、すべてのツールは BaseTool のサブクラスです、これはツール用に JSON スキーマを自動的に生成します。例えば、stock_price_tool の JSON スキーマを得るには、schema プロパティを使用できます。

stock_price_tool.schema
{'name': 'get_stock_price',
 'description': 'Get the stock price.',
 'parameters': {'type': 'object',
  'properties': {'ticker': {'description': 'ticker',
    'title': 'Ticker',
    'type': 'string'},
   'date': {'description': 'Date in YYYY/MM/DD',
    'title': 'Date',
    'type': 'string'}},
  'required': ['ticker', 'date'],
  'additionalProperties': False},
 'strict': False}

モデルクライアントはツールの JSON スキーマを使用してツール呼び出しを生成します。

OpenAIChatCompletionClient で FunctionTool クラスを使用する方法の例をここで示します。他のモデルクライアント・クラスも同様に使用できます。See Model Clients for more details.

import json

from autogen_core.models import AssistantMessage, FunctionExecutionResult, FunctionExecutionResultMessage, UserMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create the OpenAI chat completion client. Using OPENAI_API_KEY from environment variable.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

# Create a user message.
user_message = UserMessage(content="What is the stock price of AAPL on 2021/01/01?", source="user")

# Run the chat completion with the stock_price_tool defined above.
cancellation_token = CancellationToken()
create_result = await model_client.create(
    messages=[user_message], tools=[stock_price_tool], cancellation_token=cancellation_token
)
create_result.content
[FunctionCall(id='call_tpJ5J1Xoxi84Sw4v0scH0qBM', arguments='{"ticker":"AAPL","date":"2021/01/01"}', name='get_stock_price')]

create メソッドの呼び出しは実際には内部で何が起きているのでしょう?モデルクライアントはツールのリストを受け取り、各ツールのパラメータ用に JSON スキーマを生成します。そして、ツールの JSON スキーマと他のメッセージを使用してモデル API へのリクエストを生成して結果を取得します。

OpenAI の GPT-4o や Llama-3.2 のような多くのモデルは、ツールの JSON スキーマに準拠した構造化 JSON 文字列の形式でツール呼び出しを生成するようにトレーニングされています。そして AutoGen のモデルクライアントはモデルのレスポンスを解析して JSON 文字列からツール呼び出しを抽出します。

結果は FunctionCall オブジェクトのリストで、これは対応するツールを実行するために使用できます。

json.loads を使用して arguments フィールドの JSON 文字列を解析して Python 辞書に変換します。run_json() メソッドは辞書を受け取り提供された引数でツールを実行します。

assert isinstance(create_result.content, list)
arguments = json.loads(create_result.content[0].arguments)  # type: ignore
tool_result = await stock_price_tool.run_json(arguments, cancellation_token)
tool_result_str = stock_price_tool.return_value_as_string(tool_result)
tool_result_str
'32.381250753393104'

これでももう一つのモデルクライアント呼び出しを行い、ツール実行の結果のリフレクションをモデルに生成させることができます。

ツール呼び出しの結果は FunctionExecutionResult オブジェクト内にラップされます、これはツール実行の結果と呼び出されたツールの ID を含みます。モデルクライアントはこの情報を使用してツール実行の結果のリフレクションを生成できます。

# Create a function execution result
exec_result = FunctionExecutionResult(
    call_id=create_result.content[0].id,  # type: ignore
    content=tool_result_str,
    is_error=False,
    name=stock_price_tool.name,
)

# Make another chat completion with the history and function execution result message.
messages = [
    user_message,
    AssistantMessage(content=create_result.content, source="assistant"),  # assistant message with tool call
    FunctionExecutionResultMessage(content=[exec_result]),  # function execution result message
]
create_result = await model_client.create(messages=messages, cancellation_token=cancellation_token)  # type: ignore
print(create_result.content)
await model_client.close()
The stock price of AAPL (Apple Inc.) on January 1, 2021, was approximately $32.38.

 

ツール装備エージェント

モデルクライアントとツールを一つにまとめて、アクションを実行するためにツールを使用してそのアクションの結果を反映できるツール装備エージェントを作成できます。

Note : The Core API is designed to be minimal and you need to build your own agent logic around model clients and tools. For “pre-built” agents that can use tools, please refer to the AgentChat API.

import asyncio
import json
from dataclasses import dataclass
from typing import List

from autogen_core import (
    AgentId,
    FunctionCall,
    MessageContext,
    RoutedAgent,
    SingleThreadedAgentRuntime,
    message_handler,
)
from autogen_core.models import (
    ChatCompletionClient,
    LLMMessage,
    SystemMessage,
    UserMessage,
)
from autogen_core.tools import FunctionTool, Tool
from autogen_ext.models.openai import OpenAIChatCompletionClient


@dataclass
class Message:
    content: str


class ToolUseAgent(RoutedAgent):
    def __init__(self, model_client: ChatCompletionClient, tool_schema: List[Tool]) -> None:
        super().__init__("An agent with tools")
        self._system_messages: List[LLMMessage] = [SystemMessage(content="You are a helpful AI assistant.")]
        self._model_client = model_client
        self._tools = tool_schema

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:
        # Create a session of messages.
        session: List[LLMMessage] = self._system_messages + [UserMessage(content=message.content, source="user")]

        # Run the chat completion with the tools.
        create_result = await self._model_client.create(
            messages=session,
            tools=self._tools,
            cancellation_token=ctx.cancellation_token,
        )

        # If there are no tool calls, return the result.
        if isinstance(create_result.content, str):
            return Message(content=create_result.content)
        assert isinstance(create_result.content, list) and all(
            isinstance(call, FunctionCall) for call in create_result.content
        )

        # Add the first model create result to the session.
        session.append(AssistantMessage(content=create_result.content, source="assistant"))

        # Execute the tool calls.
        results = await asyncio.gather(
            *[self._execute_tool_call(call, ctx.cancellation_token) for call in create_result.content]
        )

        # Add the function execution results to the session.
        session.append(FunctionExecutionResultMessage(content=results))

        # Run the chat completion again to reflect on the history and function execution results.
        create_result = await self._model_client.create(
            messages=session,
            cancellation_token=ctx.cancellation_token,
        )
        assert isinstance(create_result.content, str)

        # Return the result as a message.
        return Message(content=create_result.content)

    async def _execute_tool_call(
        self, call: FunctionCall, cancellation_token: CancellationToken
    ) -> FunctionExecutionResult:
        # Find the tool by name.
        tool = next((tool for tool in self._tools if tool.name == call.name), None)
        assert tool is not None

        # Run the tool and capture the result.
        try:
            arguments = json.loads(call.arguments)
            result = await tool.run_json(arguments, cancellation_token)
            return FunctionExecutionResult(
                call_id=call.id, content=tool.return_value_as_string(result), is_error=False, name=tool.name
            )
        except Exception as e:
            return FunctionExecutionResult(call_id=call.id, content=str(e), is_error=True, name=tool.name)

ユーザメッセージを処理する際、ToolUseAgent クラスは最初にモデルクライアントを使用してツールへの関数呼び出しのリストを生成してから、ツールを実行してツール実行の結果のリフレクションを生成します。そしてエージェントのレスポンスとしてリフレクションがユーザに返されます。

エージェントを実行するため、ランタイムを作成して、そのランタイムでエージェントを登録しましょう。

# Create the model client.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
# Create a runtime.
runtime = SingleThreadedAgentRuntime()
# Create the tools.
tools: List[Tool] = [FunctionTool(get_stock_price, description="Get the stock price.")]
# Register the agents.
await ToolUseAgent.register(
    runtime,
    "tool_use_agent",
    lambda: ToolUseAgent(
        model_client=model_client,
        tool_schema=tools,
    ),
)
AgentType(type='tool_use_agent')

この例は OpenAIChatCompletionClient を使用しています、Azure OpenAI と他のクライアントについては、Model Clients をご覧ください。Let’s test the agent with a question about stock price.

# Start processing messages.
runtime.start()
# Send a direct message to the tool agent.
tool_use_agent = AgentId("tool_use_agent", "default")
response = await runtime.send_message(Message("What is the stock price of NVDA on 2024/06/01?"), tool_use_agent)
print(response.content)
# Stop processing messages.
await runtime.stop()
await model_client.close()
The stock price of NVIDIA (NVDA) on June 1, 2024, was approximately $140.05.

 

以上