工具#

工具是可以由代理执行以执行操作的代码。工具可以是简单的函数,如计算器,或者是对第三方服务的 API 调用,如股票价格查询或天气预报。在 AI 代理的上下文中,工具被设计为由代理响应模型生成的函数调用来执行。

AutoGen 提供了 autogen_core.components.tools 模块,其中包含一套内置工具和用于创建和运行自定义工具的实用程序。

内置工具#

其中一个内置工具是 PythonCodeExecutionTool,它允许代理执行 Python 代码片段。

以下是如何创建和使用该工具。

from autogen_core.base import CancellationToken
from autogen_core.components.tools import PythonCodeExecutionTool
from autogen_ext.code_executors import DockerCommandLineCodeExecutor

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

docker_executorCommandLineCodeExecutor 类是一个内置的代码执行器,它在本地命令行环境的子进程中运行 Python 代码片段。PythonCodeExecutionTool 类包装了代码执行器,并提供了一个简单的接口来执行 Python 代码片段。

其他内置工具将在未来添加。

自定义函数工具#

工具也可以是执行特定操作的简单 Python 函数。要创建自定义函数工具,您只需创建一个 Python 函数,并使用 FunctionTool 类来包装它。

FunctionTool 类使用描述和类型注解来告知 LLM 何时以及如何使用给定的函数。描述提供了关于函数目的和预期用例的上下文,而类型注解则告知 LLM 预期的参数和返回类型。

例如,一个获取公司股票价格的简单工具可能如下所示:

import random

from autogen_core.base import CancellationToken
from autogen_core.components.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))
80.44429939059668

Tool-Equipped Agent#

要在代理中使用工具,您可以使用 ToolAgent,通过在组合模式中使用它。 这里是一个使用 ToolAgent 作为内部代理来执行工具的工具使用代理示例。

from dataclasses import dataclass
from typing import List

from autogen_core.application import SingleThreadedAgentRuntime
from autogen_core.base import AgentId, AgentInstantiationContext, MessageContext
from autogen_core.components import RoutedAgent, message_handler
from autogen_core.components.models import (
    ChatCompletionClient,
    LLMMessage,
    OpenAIChatCompletionClient,
    SystemMessage,
    UserMessage,
)
from autogen_core.components.tool_agent import ToolAgent, tool_agent_caller_loop
from autogen_core.components.tools import FunctionTool, Tool, ToolSchema


@dataclass
class Message:
    content: str


class ToolUseAgent(RoutedAgent):
    def __init__(self, model_client: ChatCompletionClient, tool_schema: List[ToolSchema], tool_agent_type: str) -> None:
        super().__init__("An agent with tools")
        self._system_messages: List[LLMMessage] = [SystemMessage("You are a helpful AI assistant.")]
        self._model_client = model_client
        self._tool_schema = tool_schema
        self._tool_agent_id = AgentId(tool_agent_type, self.id.key)

    @message_handler
    async def handle_user_message(self, message: Message, ctx: MessageContext) -> Message:
        # Create a session of messages.
        session: List[LLMMessage] = [UserMessage(content=message.content, source="user")]
        # Run the caller loop to handle tool calls.
        messages = await tool_agent_caller_loop(
            self,
            tool_agent_id=self._tool_agent_id,
            model_client=self._model_client,
            input_messages=session,
            tool_schema=self._tool_schema,
            cancellation_token=ctx.cancellation_token,
        )
        # Return the final response.
        assert isinstance(messages[-1].content, str)
        return Message(content=messages[-1].content)

ToolUseAgent 类使用一个便利函数 tool_agent_caller_loop(),来处理模型和工具代理之间的交互。 核心思想可以用一个简单的控制流图来描述:

ToolUseAgent 控制流图

ToolUseAgenthandle_user_message 处理程序处理来自用户的消息,并确定模型是否生成了工具调用。如果模型生成了工具调用,那么处理程序会向 ToolAgent 代理发送函数调用消息来执行工具,然后使用工具调用的结果再次查询模型。这个过程会一直持续,直到模型停止生成工具调用,此时最终响应会返回给用户。

通过将工具执行逻辑放在单独的代理中,我们将模型-工具交互作为消息暴露给代理运行时,因此工具执行可以在外部观察并在必要时进行拦截。

要运行代理,我们需要创建一个运行时并注册代理。

# Create a runtime.
runtime = SingleThreadedAgentRuntime()
# Create the tools.
tools: List[Tool] = [FunctionTool(get_stock_price, description="Get the stock price.")]
# Register the agents.
await ToolAgent.register(runtime, "tool_executor_agent", lambda: ToolAgent("tool executor agent", tools))
await ToolUseAgent.register(
    runtime,
    "tool_use_agent",
    lambda: ToolUseAgent(
        OpenAIChatCompletionClient(model="gpt-4o-mini"), [tool.schema for tool in tools], "tool_executor_agent"
    ),
)
AgentType(type='tool_use_agent')

这个示例使用了 autogen_core.components.models.OpenAIChatCompletionClient,对于 Azure OpenAI 和其他客户端,请参见模型客户端。让我们用一个关于股票价格的问题来测试这个代理。

# 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()
The stock price of NVDA (NVIDIA Corporation) on June 1, 2024, was approximately $179.46.