工具#
工具是可由代理执行以执行操作的代码。工具可以是简单的函数，例如计算器，也可以是对第三方服务的 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 类是一个内置的代码执行器，它在 Docker 容器的命令行环境中子进程中运行 Python 代码片段。PythonCodeExecutionTool 类包装了代码执行器，并提供了一个简单的接口来执行 Python 代码片段。
其他内置工具示例
LocalSearchTool和GlobalSearchTool，用于使用 GraphRAG。mcp_server_tools，用于将 Model Context Protocol (MCP) 服务器用作工具。HttpTool，用于向 REST API 发送 HTTP 请求。LangChainToolAdapter，用于使用 LangChain 工具。
自定义函数工具#
工具也可以是一个执行特定操作的简单 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 模式来生成工具调用。
这是一个如何将 FunctionTool 类与 OpenAIChatCompletionClient 一起使用的示例。其他模型客户端类可以以类似的方式使用。有关更多详细信息，请参阅模型客户端。
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.
配备工具的代理#
将模型客户端和工具结合起来，您可以创建一个配备工具的代理，该代理可以使用工具执行操作，并反思这些操作的结果。
注意
核心 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:
# 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 和其他客户端，请参阅模型客户端。让我们用一个关于股票价格的问题来测试代理。
# 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.