autogen_core.tools#

class Tool(*args, **kwargs)[源]#

基类:Protocol

property name: str#
property description: str#
property schema: ToolSchema#
args_type() Type[BaseModel][源]#
return_type() Type[Any][源]#
state_type() Type[BaseModel] | None[源]#
return_value_as_string(value: Any) str[源]#
async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) Any[源]#
async save_state_json() Mapping[str, Any][源]#
async load_state_json(state: Mapping[str, Any]) None[源]#
class StreamTool(*args, **kwargs)[源]#

基类:ToolProtocol

run_json_stream(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) AsyncGenerator[Any, None][源]#
class ToolSchema[源]#

基类:TypedDict

parameters: NotRequired[ParametersSchema]#
name: str#
description: NotRequired[str]#
strict: NotRequired[bool]#
class ParametersSchema[源]#

基类:TypedDict

type: str#
properties: Dict[str, Any]#
required: NotRequired[Sequence[str]]#
additionalProperties: NotRequired[bool]#
class BaseTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)[源]#

基类:ABCToolGeneric[ArgsTReturnT],ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#

组件的逻辑类型。

property schema: ToolSchema#
property name: str#
property description: str#
args_type() Type[BaseModel][源]#
return_type() Type[Any][源]#
state_type() Type[BaseModel] | None[源]#
return_value_as_string(value: Any) str[源]#
abstract async run(args: ArgsT, cancellation_token: CancellationToken) ReturnT[源]#
async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) Any[源]#

使用提供的字典参数运行工具。

参数:
  • args (Mapping[str, Any]) – 传递给工具的参数。

  • cancellation_token (CancellationToken) – 在需要时取消操作的令牌。

  • call_id (str | None) – 用于跟踪的工具调用的可选标识符。

返回:

Any – 工具运行方法的返回值。

async save_state_json() Mapping[str, Any][源]#
async load_state_json(state: Mapping[str, Any]) None[源]#
class BaseToolWithState(args_type: Type[ArgsT], return_type: Type[ReturnT], state_type: Type[StateT], name: str, description: str)[源]#

基类:BaseTool[ArgsTReturnT],ABCGeneric[ArgsTReturnTStateT],ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#

组件的逻辑类型。

abstract save_state() StateT[源]#
abstract load_state(state: StateT) None[源]#
async save_state_json() Mapping[str, Any][源]#
async load_state_json(state: Mapping[str, Any]) None[源]#
class BaseStreamTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)[源]#

基类:BaseTool[ArgsTReturnT],StreamToolABCGeneric[ArgsTStreamTReturnT],ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#

组件的逻辑类型。

abstract run_stream(args: ArgsT, cancellation_token: CancellationToken) AsyncGenerator[StreamT | ReturnT, None][源]#

使用提供的参数运行工具,并返回数据流,最后以最终返回值结束。

async run_json_stream(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) AsyncGenerator[StreamT | ReturnT, None][源]#

使用提供的字典参数运行工具,并从工具的 run_stream() 方法返回数据流,最后以最终返回值结束。

参数:
  • args (Mapping[str, Any]) – 传递给工具的参数。

  • cancellation_token (CancellationToken) – 在需要时取消操作的令牌。

  • call_id (str | None) – 用于跟踪的工具调用的可选标识符。

返回:

AsyncGenerator[StreamT | ReturnT, None] – 一个生成器,用于生成工具的 run_stream() 方法的结果。

class FunctionTool(func: Callable[[...], Any], description: str, name: str | None = None, global_imports: Sequence[str | ImportFromModule | Alias] = [], strict: bool = False)[源]#

基类:BaseTool[BaseModelBaseModel],Component[FunctionToolConfig]

通过包装标准 Python 函数创建自定义工具。

FunctionTool 提供了一个用于异步或同步执行 Python 函数的接口。每个函数必须包含所有参数及其返回类型的类型注释。这些注释使 FunctionTool 能够生成输入验证、序列化以及向 LLM 告知预期参数所需的模式。当 LLM 准备函数调用时,它会利用此模式生成与函数规范对齐的参数。

注意

用户有责任验证工具的输出类型与预期类型匹配。

参数:
  • func (Callable[..., ReturnT | Awaitable[ReturnT]]) – 要包装并作为工具公开的函数。

  • description (str) – 用于告知模型函数目的的描述,指定其功能以及应调用它的上下文。

  • name (str, optional) – 工具的可选自定义名称。如果未提供,则默认为函数的原始名称。

  • strict (bool, optional) – 如果设置为 True,则工具模式将仅包含函数签名中明确定义的参数,并且不允许默认值。默认为 False。在结构化输出模式下与模型一起使用时,需要将其设置为 True。

示例

import random
from autogen_core import CancellationToken
from autogen_core.tools import FunctionTool
from typing_extensions import Annotated
import asyncio


async def get_stock_price(ticker: str, date: Annotated[str, "Date in YYYY/MM/DD"]) -> float:
    # Simulates a stock price retrieval by returning a random float within a specified range.
    return random.uniform(10, 200)


async def example():
    # Initialize a FunctionTool instance for retrieving stock prices.
    stock_price_tool = FunctionTool(get_stock_price, description="Fetch the stock price for a given ticker.")

    # Execute the tool with cancellation support.
    cancellation_token = CancellationToken()
    result = await stock_price_tool.run_json({"ticker": "AAPL", "date": "2021/01/01"}, cancellation_token)

    # Output the result as a formatted string.
    print(stock_price_tool.return_value_as_string(result))


asyncio.run(example())
component_provider_override: ClassVar[str | None] = 'autogen_core.tools.FunctionTool'#

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

component_config_schema#

别名:FunctionToolConfig

async run(args: BaseModel, cancellation_token: CancellationToken) Any[源]#
_to_config() FunctionToolConfig[源]#

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: FunctionToolConfig) Self[源]#

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class Workbench[源]#

基类: ABC, ComponentBase[BaseModel]

工作台是一个提供一组工具的组件,这些工具可能共享资源和状态。

工作台负责管理工具的生命周期并提供一个调用它们的单一接口。工作台提供的工具可以是动态的,并且它们的可用性可能会在每次工具执行后发生变化。

工作台可以通过调用 start() 方法启动,并通过调用 stop() 方法停止。它也可以用作异步上下文管理器,在进入和退出上下文时会自动启动和停止工作台。

component_type: ClassVar[ComponentType] = 'workbench'#

组件的逻辑类型。

abstract async list_tools() List[ToolSchema][源]#

将工作台中当前可用的工具列为 ToolSchema 对象。

工具列表可以是动态的,其内容可能在工具执行后发生变化。

abstract async call_tool(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) ToolResult[源]#

在工作台中调用工具。

参数:
  • name (str) – 要调用的工具的名称。

  • arguments (Mapping[str, Any] | None) – 传递给工具的参数。如果为 None,则工具将不带参数调用。

  • cancellation_token (CancellationToken | None) – 可选的取消令牌,用于取消工具执行。

  • call_id (str | None) – 用于跟踪的工具调用的可选标识符。

返回:

ToolResult – 工具执行的结果。

abstract async start() None[源]#

启动工作台并初始化任何资源。

在使用工作台之前应该调用此方法。

abstract async stop() None[源]#

停止工作台并释放任何资源。

当不再需要工作台时,应该调用此方法。

abstract async reset() None[源]#

将工作台重置为其已初始化、已启动的状态。

abstract async save_state() Mapping[str, Any][source]#

保存工作台的状态。

应调用此方法以持久化工作台的状态。

abstract async load_state(state: Mapping[str, Any]) None[source]#

加载工作台的状态。

参数:

state (Mapping[str, Any]) – 要加载到工作台中的状态。

pydantic model ToolResult[source]#

基类: BaseModel

工作台执行工具的结果。

显示 JSON 模式
{
   "title": "ToolResult",
   "description": "A result of a tool execution by a workbench.",
   "type": "object",
   "properties": {
      "type": {
         "const": "ToolResult",
         "default": "ToolResult",
         "title": "Type",
         "type": "string"
      },
      "name": {
         "title": "Name",
         "type": "string"
      },
      "result": {
         "items": {
            "discriminator": {
               "mapping": {
                  "ImageResultContent": "#/$defs/ImageResultContent",
                  "TextResultContent": "#/$defs/TextResultContent"
               },
               "propertyName": "type"
            },
            "oneOf": [
               {
                  "$ref": "#/$defs/TextResultContent"
               },
               {
                  "$ref": "#/$defs/ImageResultContent"
               }
            ]
         },
         "title": "Result",
         "type": "array"
      },
      "is_error": {
         "default": false,
         "title": "Is Error",
         "type": "boolean"
      }
   },
   "$defs": {
      "ImageResultContent": {
         "description": "Image result content of a tool execution.",
         "properties": {
            "type": {
               "const": "ImageResultContent",
               "default": "ImageResultContent",
               "title": "Type",
               "type": "string"
            },
            "content": {
               "title": "Content"
            }
         },
         "required": [
            "content"
         ],
         "title": "ImageResultContent",
         "type": "object"
      },
      "TextResultContent": {
         "description": "Text result content of a tool execution.",
         "properties": {
            "type": {
               "const": "TextResultContent",
               "default": "TextResultContent",
               "title": "Type",
               "type": "string"
            },
            "content": {
               "title": "Content",
               "type": "string"
            }
         },
         "required": [
            "content"
         ],
         "title": "TextResultContent",
         "type": "object"
      }
   },
   "required": [
      "name",
      "result"
   ]
}

字段:
  • is_error (bool)

  • name (str)

  • result (List[autogen_core.tools._workbench.TextResultContent | autogen_core.tools._workbench.ImageResultContent])

  • type (Literal['ToolResult'])

field type: Literal['ToolResult'] = 'ToolResult'#
field name: str [Required]#

已执行工具的名称。

field result: List[Annotated[TextResultContent | ImageResultContent, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] [Required]#

工具执行的结果。

field is_error: bool = False#

工具执行是否导致错误。

to_text(replace_image: str | None = None) str[source]#

将结果转换为文本字符串。

参数:

replace_image (str | None) – 用于替换图像内容的字符串。如果为None,图像内容将作为base64字符串包含在文本中。

返回:

str – 结果的文本表示。

pydantic model TextResultContent[source]#

基类: BaseModel

工具执行的文本结果内容。

显示 JSON 模式
{
   "title": "TextResultContent",
   "description": "Text result content of a tool execution.",
   "type": "object",
   "properties": {
      "type": {
         "const": "TextResultContent",
         "default": "TextResultContent",
         "title": "Type",
         "type": "string"
      },
      "content": {
         "title": "Content",
         "type": "string"
      }
   },
   "required": [
      "content"
   ]
}

字段:
  • content (str)

  • type (Literal['TextResultContent'])

field type: Literal['TextResultContent'] = 'TextResultContent'#
field content: str [Required]#

结果的文本内容。

pydantic model ImageResultContent[source]#

基类: BaseModel

工具执行的图像结果内容。

显示 JSON 模式
{
   "title": "ImageResultContent",
   "description": "Image result content of a tool execution.",
   "type": "object",
   "properties": {
      "type": {
         "const": "ImageResultContent",
         "default": "ImageResultContent",
         "title": "Type",
         "type": "string"
      },
      "content": {
         "title": "Content"
      }
   },
   "required": [
      "content"
   ]
}

字段:
  • content (autogen_core._image.Image)

  • type (Literal['ImageResultContent'])

field type: Literal['ImageResultContent'] = 'ImageResultContent'#
field content: Image [Required]#

结果的图像内容。

class StaticWorkbench(tools: List[BaseTool[Any, Any]], tool_overrides: Dict[str, ToolOverride] | None = None)[source]#

基类:Workbench, Component[StaticWorkbenchConfig]

一个工作台,提供一组静态工具,这些工具在每次工具执行后都不会改变。

参数:
  • tools (List[BaseTool[Any, Any]]) – 要包含在工作台中的工具列表。工具应为 BaseTool 的子类。

  • tool_overrides (Optional[Dict[str, ToolOverride]]) – 原始工具名称到名称和/或描述覆盖配置的可选映射。这允许自定义工具如何向使用者显示,同时保持底层工具功能。

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticWorkbench'#

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

component_config_schema#

别名:StaticWorkbenchConfig

async list_tools() List[ToolSchema][source]#

将工作台中当前可用的工具列为 ToolSchema 对象。

工具列表可以是动态的,其内容可能在工具执行后发生变化。

async call_tool(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) ToolResult[source]#

在工作台中调用工具。

参数:
  • name (str) – 要调用的工具的名称。

  • arguments (Mapping[str, Any] | None) – 传递给工具的参数。如果为 None,则工具将不带参数调用。

  • cancellation_token (CancellationToken | None) – 可选的取消令牌,用于取消工具执行。

  • call_id (str | None) – 用于跟踪的工具调用的可选标识符。

返回:

ToolResult – 工具执行的结果。

async start() None[source]#

启动工作台并初始化任何资源。

在使用工作台之前应该调用此方法。

async stop() None[source]#

停止工作台并释放任何资源。

当不再需要工作台时,应该调用此方法。

async reset() None[source]#

将工作台重置为其已初始化、已启动的状态。

async save_state() Mapping[str, Any][source]#

保存工作台的状态。

应调用此方法以持久化工作台的状态。

async load_state(state: Mapping[str, Any]) None[source]#

加载工作台的状态。

参数:

state (Mapping[str, Any]) – 要加载到工作台中的状态。

_to_config() StaticWorkbenchConfig[source]#

转储创建与此实例配置匹配的组件新实例所需的配置。

返回:

T – 组件的配置。

classmethod _from_config(config: StaticWorkbenchConfig) Self[source]#

从配置对象创建组件的新实例。

参数:

config (T) – 配置对象。

返回:

Self – 组件的新实例。

class StaticStreamWorkbench(tools: List[BaseTool[Any, Any]], tool_overrides: Dict[str, ToolOverride] | None = None)[source]#

基类:StaticWorkbench, StreamWorkbench

一个工作台,提供一组静态工具,这些工具在每次工具执行后都不会改变,并支持流式结果。

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticStreamWorkbench'#

覆盖组件的提供者字符串。这应该用于防止内部模块名称成为模块名称的一部分。

async call_tool_stream(name: str, arguments: Mapping[str, Any] | None = None, cancellation_token: CancellationToken | None = None, call_id: str | None = None) AsyncGenerator[Any | ToolResult, None]#

在工作台中调用工具并返回结果流。

参数:
  • name (str) – 要调用的工具的名称。

  • arguments (Mapping[str, Any] | None) – 要传递给工具的参数。如果为None,则以无参数方式调用工具。

  • cancellation_token (CancellationToken | None) – 可选的取消令牌,用于取消工具执行。

  • call_id (str | None) – 用于跟踪的工具调用的可选标识符。

pydantic model ToolOverride[source]#

基类: BaseModel

工具名称和/或描述的覆盖配置。

显示 JSON 模式
{
   "title": "ToolOverride",
   "description": "Override configuration for a tool's name and/or description.",
   "type": "object",
   "properties": {
      "name": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Name"
      },
      "description": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Description"
      }
   }
}

字段:
  • description (str | None)

  • name (str | None)

field name: str | None = None#
field description: str | None = None#