工具

工具是可由代理执行以完成操作的代码。一个工具可以是一个简单的函数（如计算器），也可以是对第三方服务的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

# 创建工具。
code_executor = DockerCommandLineCodeExecutor()
await code_executor.start()
code_execution_tool = PythonCodeExecutionTool(code_executor)
cancellation_token = CancellationToken()

# 不通过代理直接使用工具。
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类是一个内置代码执行器，它在Docker容器的命令行环境中以子进程方式运行Python代码片段。 PythonCodeExecutionTool类封装了代码执行器，并提供了执行Python代码片段的简单接口。

其他内置工具示例：

• LocalSearchTool和GlobalSearchTool用于使用GraphRAG。

• mcp_server_tools用于将模型上下文协议(MCP)服务器作为工具使用。

• HttpTool用于向REST API发起HTTP请求。

• LangChainToolAdapter用于使用LangChain工具。

自定义函数工具

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

FunctionTool 类利用描述和类型注解 来告知大语言模型何时以及如何使用给定函数。描述提供了 关于函数用途和预期使用场景的上下文，而类型注解则告知大语言模型 预期的参数和返回类型。

例如，一个获取公司股价的简单工具可能如下所示：

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:
    # 返回随机股价用于演示目的。
    return random.uniform(10, 200)


# 创建函数工具。
stock_price_tool = FunctionTool(get_stock_price, description="Get the stock price.")

# 运行工具。
cancellation_token = CancellationToken()
result = await stock_price_tool.run_json({"ticker": "AAPL", "date": "2021/01/01"}, cancellation_token)

# 打印结果。
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 模式来生成工具调用。

以下示例展示了如何将 FunctionTool 类 与 OpenAIChatCompletionClient 结合使用。 其他模型客户端类的使用方式类似。详见模型客户端 获取更多信息。

import json

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

# 创建 OpenAI 聊天补全客户端，使用环境变量中的 OPENAI_API_KEY
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

# 创建用户消息
user_message = UserMessage(content="What is the stock price of AAPL on 2021/01/01?", source="user")

# 使用上面定义的 stock_price_tool 运行聊天补全
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')]

调用 {py:class}~autogen_ext.models.openai.BaseOpenAIChatCompletionClient.create` 方法时，底层实际发生了什么？模型客户端会获取工具列表并为每个工具的参数生成 JSON 模式。然后，它会向模型 API 发起请求，携带工具的 JSON 模式和其他消息以获取结果。

许多模型（如 OpenAI 的 GPT-4o 和 Llama-3.2）经过训练后，能够生成符合工具 JSON 模式的结构化 JSON 字符串形式的工具调用。AutoGen 的模型客户端随后会解析模型的响应，并从 JSON 字符串中提取工具调用。

最终结果是生成一个 {py:class}~autogen_core.FunctionCall` 对象列表，这些对象可用于运行对应的工具。

我们使用 json.loads 将 {py:class}~autogen_core.FunctionCall.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'

现在你可以发起另一个模型客户端调用，让模型对工具执行结果进行反思。

工具调用的结果会被包装在 {py:class}~autogen_core.models.FunctionExecutionResult` 对象中，该对象包含工具执行结果和被调用工具的 ID。模型客户端可以利用这些信息生成对工具执行结果的反思。

# 创建函数执行结果
exec_result = FunctionExecutionResult(
    call_id=create_result.content[0].id,  # type: ignore
    content=tool_result_str,
    is_error=False,
    name=stock_price_tool.name,
)

# 使用历史记录和函数执行结果消息进行另一次聊天完成
messages = [
    user_message,
    AssistantMessage(content=create_result.content, source="assistant"),  # 带有工具调用的助手消息
    FunctionExecutionResultMessage(content=[exec_result]),  # 函数执行结果消息
]
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.

配备工具的智能体

将模型客户端与工具结合起来，您可以创建一个配备工具的智能体 它能够使用工具执行操作，并反思这些操作的结果。

备注

核心API设计为最小化，您需要围绕模型客户端和工具构建自己的智能体逻辑。 关于可以使用工具的"预构建"智能体，请参考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:
        # 创建消息会话
        session: List[LLMMessage] = self._system_messages + [UserMessage(content=message.content, source="user")]

        # 使用工具运行聊天补全
        create_result = await self._model_client.create(
            messages=session,
            tools=self._tools,
            cancellation_token=ctx.cancellation_token,
        )

        # 如果没有工具调用，返回结果
        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
        )

        # 将首个模型创建结果添加到会话中
        session.append(AssistantMessage(content=create_result.content, source="assistant"))

        # 执行工具调用
        results = await asyncio.gather(
            *[self._execute_tool_call(call, ctx.cancellation_token) for call in create_result.content]
        )

        # 将函数执行结果添加到会话中。
        session.append(FunctionExecutionResultMessage(content=results))

        # 再次运行聊天补全以反映历史记录和函数执行结果。
        create_result = await self._model_client.create(
            messages=session,
            cancellation_token=ctx.cancellation_token,
        )
        assert isinstance(create_result.content, str)

        # 将结果作为消息返回。
        return Message(content=create_result.content)

    async def _execute_tool_call(
        self, call: FunctionCall, cancellation_token: CancellationToken
    ) -> FunctionExecutionResult:
        # 按名称查找工具。
        tool = next((tool for tool in self._tools if tool.name == call.name), None)
        assert tool is not None

        # 运行工具并捕获结果。
        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 类首先使用模型客户端 生成一系列工具函数调用，然后运行这些工具 并对工具执行结果进行反思。 随后将反思结果作为代理的响应返回给用户。

要运行该代理，我们需要创建一个运行时环境并将代理注册到运行时中。

# 创建模型客户端。
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
# 创建运行时环境。
runtime = SingleThreadedAgentRuntime()
# 创建工具集。
tools: List[Tool] = [FunctionTool(get_stock_price, description="Get the stock price.")]
# 注册代理程序。
await ToolUseAgent.register(
    runtime,
    "tool_use_agent",
    lambda: ToolUseAgent(
        model_client=model_client,
        tool_schema=tools,
    ),
)

AgentType(type='tool_use_agent')

本示例使用 OpenAIChatCompletionClient， 关于Azure OpenAI和其他客户端，请参阅模型客户端。 让我们用一个关于股票价格的问题来测试这个智能体。

# 开始处理消息。
runtime.start()
# 向工具智能体发送直接消息。
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)
# 停止处理消息。
await runtime.stop()
await model_client.close()

The stock price of NVIDIA (NVDA) on June 1, 2024, was approximately $140.05.