autogen_core.tool_agent._tool_agent 源代码

import json
from dataclasses import dataclass
from typing import List

from .. import FunctionCall, MessageContext, RoutedAgent, message_handler
from ..models import FunctionExecutionResult
from ..tools import Tool

__all__ = [
    "ToolAgent",
    "ToolException",
    "ToolNotFoundException",
    "InvalidToolArgumentsException",
    "ToolExecutionException",
]




@dataclass
class ToolException(BaseException):
    call_id: str
    content: str
    name: str





@dataclass
class ToolNotFoundException(ToolException):
    pass





@dataclass
class InvalidToolArgumentsException(ToolException):
    pass





@dataclass
class ToolExecutionException(ToolException):
    pass





class ToolAgent(RoutedAgent):
    """一个工具代理接收 `FunctionCall` 类型的直接消息，
    使用提供的参数执行请求的工具，并将结果作为
    `FunctionExecutionResult` 消息返回。

    Args:
        description (str): 代理的描述信息。
        tools (List[Tool]): 代理可以执行的工具列表。
    """

    def __init__(
        self,
        description: str,
        tools: List[Tool],
    ) -> None:
        super().__init__(description)
        self._tools = tools

    @property
    def tools(self) -> List[Tool]:
        return self._tools



    @message_handler
    async def handle_function_call(self, message: FunctionCall, ctx: MessageContext) -> FunctionExecutionResult:
        """处理 `FunctionCall` 消息，使用提供的参数执行请求的工具。

        Args:
            message (FunctionCall): 函数调用消息。
            cancellation_token (CancellationToken): 取消令牌。

        Returns:
            FunctionExecutionResult: 函数执行结果。

        Raises:
            ToolNotFoundException: 如果找不到工具。
            InvalidToolArgumentsException: 如果工具参数无效。
            ToolExecutionException: 如果工具执行失败。
        """
        tool = next((tool for tool in self._tools if tool.name == message.name), None)
        if tool is None:
            raise ToolNotFoundException(
                call_id=message.id, content=f"Error: Tool not found: {message.name}", name=message.name
            )
        else:
            try:
                arguments = json.loads(message.arguments)
                result = await tool.run_json(args=arguments, cancellation_token=ctx.cancellation_token)
                result_as_str = tool.return_value_as_string(result)
            except json.JSONDecodeError as e:
                raise InvalidToolArgumentsException(
                    call_id=message.id, content=f"Error: Invalid arguments: {message.arguments}", name=message.name
                ) from e
            except Exception as e:
                raise ToolExecutionException(call_id=message.id, content=f"Error: {e}", name=message.name) from e
        return FunctionExecutionResult(content=result_as_str, call_id=message.id, is_error=False, name=message.name)