Tools#

Tools sind Code, der von einem Agenten zur Ausführung von Aktionen verwendet werden kann. Ein Tool kann eine einfache Funktion wie ein Taschenrechner oder ein API-Aufruf an einen Drittanbieterdienst wie die Abfrage von Aktienkursen oder Wettervorhersagen sein. Im Kontext von KI-Agenten sind Tools so konzipiert, dass sie von Agenten als Reaktion auf modellgenerierte Funktionsaufrufe ausgeführt werden.

AutoGen stellt das Modul autogen_core.tools mit einer Reihe von integrierten Tools und Dienstprogrammen zur Erstellung und Ausführung benutzerdefinierter Tools bereit.

Integrierte Tools#

Eines der integrierten Tools ist das PythonCodeExecutionTool, das es Agenten ermöglicht, Python-Code-Schnipsel auszuführen.

So erstellen Sie das Tool und verwenden es.

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!

Die Klasse DockerCommandLineCodeExecutor ist ein integrierter Code-Executor, der Python-Code-Schnipsel in einem Subprozess in der Kommandozeilenumgebung eines Docker-Containers ausführt. Die Klasse PythonCodeExecutionTool umschließt den Code-Executor und bietet eine einfache Schnittstelle zur Ausführung von Python-Code-Schnipseln.

Beispiele für andere integrierte Tools

Benutzerdefinierte Funktionswerkzeuge#

Ein Tool kann auch eine einfache Python-Funktion sein, die eine bestimmte Aktion ausführt. Um ein benutzerdefiniertes Funktionswerkzeug zu erstellen, müssen Sie lediglich eine Python-Funktion erstellen und die Klasse FunctionTool verwenden, um sie zu umschließen.

Die Klasse FunctionTool verwendet Beschreibungen und Typannotationen, um die LLM zu informieren, wann und wie eine bestimmte Funktion verwendet werden soll. Die Beschreibung liefert Kontext über den Zweck und die beabsichtigten Anwendungsfälle der Funktion, während Typannotationen die LLM über die erwarteten Parameter und den Rückgabetyp informieren.

Zum Beispiel könnte ein einfaches Tool zur Ermittlung des Aktienkurses eines Unternehmens wie folgt aussehen

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

Tools mit Modellclients aufrufen#

In AutoGen ist jedes Tool eine Unterklasse von BaseTool, die automatisch das JSON-Schema für das Tool generiert. Um beispielsweise das JSON-Schema für das stock_price_tool zu erhalten, können wir die Eigenschaft schema verwenden.

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}

Modellclients verwenden das JSON-Schema der Tools, um Tool-Aufrufe zu generieren.

Hier ist ein Beispiel für die Verwendung der Klasse FunctionTool mit einem OpenAIChatCompletionClient. Andere Modellclientklassen können auf ähnliche Weise verwendet werden. Weitere Informationen finden Sie unter Modellclients.

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')]

Was passiert tatsächlich unter der Haube des Aufrufs der Methode create? Der Modellclient nimmt die Liste der Tools entgegen und generiert ein JSON-Schema für die Parameter jedes Tools. Anschließend generiert er eine Anfrage an die Modell-API mit dem JSON-Schema des Tools und den anderen Nachrichten, um ein Ergebnis zu erhalten.

Viele Modelle, wie z. B. GPT-4o von OpenAI und Llama-3.2, sind darauf trainiert, Tool-Aufrufe in Form von strukturierten JSON-Strings zu erzeugen, die dem JSON-Schema des Tools entsprechen. Die Modellclients von AutoGen analysieren dann die Antwort des Modells und extrahieren den Tool-Aufruf aus dem JSON-String.

Das Ergebnis ist eine Liste von FunctionCall-Objekten, die zum Ausführen der entsprechenden Tools verwendet werden können.

Wir verwenden json.loads, um den JSON-String im Feld arguments in ein Python-Dictionary zu parsen. Die Methode run_json() nimmt das Dictionary entgegen und führt das Tool mit den angegebenen Argumenten aus.

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'

Jetzt können Sie einen weiteren Modellclient-Aufruf tätigen, damit das Modell eine Reflexion über das Ergebnis der Tool-Ausführung generiert.

Das Ergebnis des Tool-Aufrufs ist in einem Objekt vom Typ FunctionExecutionResult verpackt, das das Ergebnis der Tool-Ausführung und die ID des aufgerufenen Tools enthält. Der Modellclient kann diese Informationen verwenden, um eine Reflexion über das Ergebnis der Tool-Ausführung zu generieren.

# 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.

Mit Tools ausgestatteter Agent#

Wenn Sie den Modellclient und die Tools zusammenfügen, können Sie einen mit Tools ausgestatteten Agenten erstellen, der Tools zur Ausführung von Aktionen verwenden und über die Ergebnisse dieser Aktionen reflektieren kann.

Hinweis

Die Kern-API ist minimalistisch gestaltet, und Sie müssen Ihre eigene Agentenlogik um Modellclients und Tools herum aufbauen. Für "vorgefertigte" Agenten, die Tools verwenden können, beachten Sie bitte die 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)

Beim Behandeln einer Benutzernachricht verwendet die Klasse ToolUseAgent zuerst den Modellclient, um eine Liste von Funktionsaufrufen an die Tools zu generieren, führt dann die Tools aus und generiert eine Reflexion über die Ergebnisse der Tool-Ausführung. Die Reflexion wird dann als Antwort des Agenten an den Benutzer zurückgegeben.

Um den Agenten auszuführen, erstellen wir eine Laufzeitumgebung und registrieren den Agenten bei der Laufzeitumgebung.

# 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')

Dieses Beispiel verwendet den OpenAIChatCompletionClient. Für Azure OpenAI und andere Clients siehe Modellclients. Testen wir den Agenten mit einer Frage zum Aktienkurs.

# 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.