autogen_core.tools#

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

Bases: Protocol

property name: str#

property description: str#

property schema: ToolSchema#

args_type() → Type[BaseModel][Quelle]#

return_type() → Type[Any][Quelle]#

state_type() → Type[BaseModel] | None[Quelle]#

return_value_as_string(value: Any) → str[Quelle]#

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → Any[Quelle]#

async save_state_json() → Mapping[str, Any][Quelle]#

async load_state_json(state: Mapping[str, Any]) → None[Quelle]#

class StreamTool(*args, **kwargs)[Quelle]#

Bases: Tool, Protocol

run_json_stream(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → AsyncGenerator[Any, None][Quelle]#

class ToolSchema[Quelle]#

Bases: TypedDict

parameters: NotRequired[ParametersSchema]#

name: str#

description: NotRequired[str]#

strict: NotRequired[bool]#

class ParametersSchema[Quelle]#

Bases: 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)[Quelle]#

Bases: ABC, Tool, Generic[ArgsT, ReturnT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: Der logische Typ der Komponente.

property schema: ToolSchema#

property name: str#

property description: str#

args_type() → Type[BaseModel][Quelle]#

return_type() → Type[Any][Quelle]#

state_type() → Type[BaseModel] | None[Quelle]#

return_value_as_string(value: Any) → str[Quelle]#

abstract async run(args: ArgsT, cancellation_token: CancellationToken) → ReturnT[Quelle]#

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken, call_id: str | None = None) → Any[Quelle]#

Führt das Tool mit den bereitgestellten Argumenten in einem Dictionary aus.

Parameter:

args (Mapping[str, Any]) – Die Argumente, die an das Tool übergeben werden sollen.
cancellation_token (CancellationToken) – Ein Token zum Abbrechen der Operation, falls erforderlich.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die für die Rückverfolgung verwendet wird.

Gibt zurück:

Any – Der Rückgabewert der run-Methode des Tools.

async save_state_json() → Mapping[str, Any][Quelle]#

async load_state_json(state: Mapping[str, Any]) → None[Quelle]#

class BaseToolWithState(args_type: Type[ArgsT], return_type: Type[ReturnT], state_type: Type[StateT], name: str, description: str)[Quelle]#

Bases: BaseTool[ArgsT, ReturnT], ABC, Generic[ArgsT, ReturnT, StateT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: Der logische Typ der Komponente.

abstract save_state() → StateT[Quelle]#

abstract load_state(state: StateT) → None[Quelle]#

async save_state_json() → Mapping[str, Any][source]#

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

class BaseStreamTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)[source]#

Bases: BaseTool[ArgsT, ReturnT], StreamTool, ABC, Generic[ArgsT, StreamT, ReturnT], ComponentBase[BaseModel]

component_type: ClassVar[ComponentType] = 'tool'#: Der logische Typ der Komponente.

abstract run_stream(args: ArgsT, cancellation_token: CancellationToken) → AsyncGenerator[StreamT | ReturnT, None][source]#: Führt das Tool mit den bereitgestellten Argumenten aus und gibt einen Datenstrom zurück, der mit dem endgültigen Rückgabewert abschließt.

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

Führt das Tool mit den Argumenten im Wörterbuch aus und gibt einen Datenstrom vom run_stream()-Methode des Tools zurück, der mit dem endgültigen Rückgabewert abgeschlossen wird.

Parameter:

args (Mapping[str, Any]) – Die Argumente, die an das Tool übergeben werden sollen.
cancellation_token (CancellationToken) – Ein Token zum Abbrechen der Operation, falls erforderlich.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die für die Rückverfolgung verwendet wird.

Gibt zurück:

AsyncGenerator[StreamT | ReturnT, None] – Ein Generator, der Ergebnisse aus der run_stream()-Methode des Tools liefert.

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

Bases: BaseTool[BaseModel, BaseModel], Component[FunctionToolConfig]

Erstellen Sie benutzerdefinierte Tools durch Kapselung von Standard-Python-Funktionen.

FunctionTool bietet eine Schnittstelle zur Ausführung von Python-Funktionen, entweder asynchron oder synchron. Jede Funktion muss Typannotationen für alle Parameter und ihren Rückgabetyp enthalten. Diese Annotationen ermöglichen es FunctionTool, ein Schema zu generieren, das für die Eingabevalidierung, Serialisierung und zur Information des LLM über erwartete Parameter erforderlich ist. Wenn das LLM einen Funktionsaufruf vorbereitet, nutzt es dieses Schema, um Argumente zu generieren, die mit den Spezifikationen der Funktion übereinstimmen.

Hinweis

Es liegt in der Verantwortung des Benutzers, sicherzustellen, dass der Ausgabetyp des Tools mit dem erwarteten Typ übereinstimmt.

Parameter:

func (Callable[..., ReturnT | Awaitable[ReturnT]]) – Die Funktion, die gekapselt und als Tool verfügbar gemacht werden soll.
description (str) – Eine Beschreibung, die das Modell über den Zweck der Funktion informiert und angibt, was sie tut und in welchem Kontext sie aufgerufen werden sollte.
name (str, optional) – Ein optionaler benutzerdefinierter Name für das Tool. Standardmäßig wird der ursprüngliche Name der Funktion verwendet, wenn keiner angegeben ist.
strict (bool, optional) – Wenn auf True gesetzt, enthält das Tool-Schema nur Argumente, die explizit in der Funktionssignatur definiert sind, und es sind keine Standardwerte zulässig. Standardmäßig False. Dies muss auf True gesetzt werden, wenn es mit Modellen im strukturierten Ausgabemodus verwendet wird.

Beispiel

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'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

component_config_schema#: Alias von FunctionToolConfig

async run(args: BaseModel, cancellation_token: CancellationToken) → Any[source]#

_to_config() → FunctionToolConfig[source]#

Gib die Konfiguration aus, die erforderlich wäre, um eine neue Instanz einer Komponente zu erstellen, die der Konfiguration dieser Instanz entspricht.

Gibt zurück:: T – Die Konfiguration der Komponente.

classmethod _from_config(config: FunctionToolConfig) → Self[source]#

Erstelle eine neue Instanz der Komponente aus einem Konfigurationsobjekt.

Parameter:: config (T) – Das Konfigurationsobjekt.
Gibt zurück:: Self – Die neue Instanz der Komponente.

class Workbench[source]#

Bases: ABC, ComponentBase[BaseModel]

Eine Werkbank ist eine Komponente, die eine Reihe von Tools bereitstellt, die Ressourcen und Zustände gemeinsam nutzen können.

Eine Werkbank ist für die Verwaltung des Lebenszyklus der Tools und die Bereitstellung einer einzigen Schnittstelle für deren Aufruf verantwortlich. Die von der Werkbank bereitgestellten Tools können dynamisch sein und ihre Verfügbarkeit kann sich nach jeder Tool-Ausführung ändern.

Eine Werkbank kann durch Aufrufen der Methode start() gestartet und durch Aufrufen der Methode stop() gestoppt werden. Sie kann auch als asynchroner Kontextmanager verwendet werden, der die Werkbank beim Betreten und Verlassen des Kontexts automatisch startet und stoppt.

component_type: ClassVar[ComponentType] = 'workbench'#: Der logische Typ der Komponente.

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

Listet die aktuell verfügbaren Tools in der Werkbank als ToolSchema-Objekte auf.

Die Liste der Tools kann dynamisch sein und ihr Inhalt kann sich nach der Tool-Ausführung ändern.

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

Ruft ein Tool in der Werkbank auf.

Parameter:

name (str) – Der Name des aufzurufenden Tools.
arguments (Mapping[str, Any] | None) – Die an das Tool zu übergebenden Argumente. Wenn None, wird das Tool ohne Argumente aufgerufen.
cancellation_token (CancellationToken | None) – Ein optionaler Abbruch-Token zur Unterbrechung der Tool-Ausführung.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die für die Rückverfolgung verwendet wird.

Gibt zurück:

ToolResult – Das Ergebnis der Tool-Ausführung.

abstract async start() → None[source]#

Startet die Werkbank und initialisiert alle Ressourcen.

Diese Methode sollte aufgerufen werden, bevor die Werkbank verwendet wird.

abstract async stop() → None[source]#

Stoppt die Werkbank und gibt alle Ressourcen frei.

Diese Methode sollte aufgerufen werden, wenn die Werkbank nicht mehr benötigt wird.

abstract async reset() → None[source]#: Setzt die Werkbank auf ihren initialisierten, gestarteten Zustand zurück.

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

Speichert den Zustand der Werkbank.

Diese Methode sollte aufgerufen werden, um den Zustand der Werkbank zu persistieren.

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

Lädt den Zustand der Werkbank.

Parameter:: state (Mapping[str, Any]) – Der in die Werkbank zu ladende Zustand.

pydantic model ToolResult[source]#

Bases: BaseModel

Ein Ergebnis der Ausführung eines Tools durch eine Werkbank.

JSON-Schema anzeigen

{
   "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"
   ]
}

Felder:

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]#: Der Name des ausgeführten Tools.

field result: List[Annotated[TextResultContent | ImageResultContent, FieldInfo(annotation=NoneType, required=True, discriminator='type')]] [Required]#: Das Ergebnis der Tool-Ausführung.

field is_error: bool = False#: Ob die Ausführung des Tools zu einem Fehler geführt hat.

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

Das Ergebnis in einen Text-String umwandeln.

Parameter:: replace_image (str | None) – Der String, mit dem der Bildinhalt ersetzt werden soll. Wenn None, wird der Bildinhalt als Base64-String in den Text aufgenommen.
Gibt zurück:: str – Die Textdarstellung des Ergebnisses.

pydantic model TextResultContent[Quelle]#

Bases: BaseModel

Text-Ergebnisinhalt einer Tool-Ausführung.

JSON-Schema anzeigen

{
   "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"
   ]
}

Felder:

content (str)
type (Literal['TextResultContent'])

field type: Literal['TextResultContent'] = 'TextResultContent'#

field content: str [Erforderlich]#: Der Textinhalt des Ergebnisses.

pydantic model ImageResultContent[Quelle]#

Bases: BaseModel

Bild-Ergebnisinhalt einer Tool-Ausführung.

JSON-Schema anzeigen

{
   "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"
   ]
}

Felder:

content (autogen_core._image.Image)
type (Literal['ImageResultContent'])

field type: Literal['ImageResultContent'] = 'ImageResultContent'#

field content: Image [Erforderlich]#: Der Bildinhalt des Ergebnisses.

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

Bases: Workbench, Component[StaticWorkbenchConfig]

Eine Werkbank, die einen statischen Satz von Werkzeugen bereitstellt, der sich nach jeder Tool-Ausführung nicht ändert.

Parameter:

tools (List[BaseTool[Any, Any]]) – Eine Liste von Werkzeugen, die in die Werkbank aufgenommen werden sollen. Die Werkzeuge sollten Unterklassen von BaseTool sein.
tool_overrides (Optional[Dict[str, ToolOverride]]) – Optionale Zuordnung von ursprünglichen Tool-Namen zu Override-Konfigurationen für Name und/oder Beschreibung. Dies ermöglicht die Anpassung, wie Werkzeuge für Konsumenten erscheinen, während die zugrunde liegende Werkzeugfunktionalität erhalten bleibt.

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticWorkbench'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

component_config_schema#: Alias von StaticWorkbenchConfig

async list_tools() → List[ToolSchema][Quelle]#

Listet die aktuell verfügbaren Tools in der Werkbank als ToolSchema-Objekte auf.

Die Liste der Tools kann dynamisch sein und ihr Inhalt kann sich nach der Tool-Ausführung ändern.

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

Ruft ein Tool in der Werkbank auf.

Parameter:

name (str) – Der Name des aufzurufenden Tools.
arguments (Mapping[str, Any] | None) – Die an das Tool zu übergebenden Argumente. Wenn None, wird das Tool ohne Argumente aufgerufen.
cancellation_token (CancellationToken | None) – Ein optionaler Abbruch-Token zur Unterbrechung der Tool-Ausführung.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die für die Rückverfolgung verwendet wird.

Gibt zurück:

ToolResult – Das Ergebnis der Tool-Ausführung.

async start() → None[Quelle]#

Startet die Werkbank und initialisiert alle Ressourcen.

Diese Methode sollte aufgerufen werden, bevor die Werkbank verwendet wird.

async stop() → None[Quelle]#

Stoppt die Werkbank und gibt alle Ressourcen frei.

Diese Methode sollte aufgerufen werden, wenn die Werkbank nicht mehr benötigt wird.

async reset() → None[Quelle]#: Setzt die Werkbank auf ihren initialisierten, gestarteten Zustand zurück.

async save_state() → Mapping[str, Any][Quelle]#

Speichert den Zustand der Werkbank.

Diese Methode sollte aufgerufen werden, um den Zustand der Werkbank zu persistieren.

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

Lädt den Zustand der Werkbank.

Parameter:: state (Mapping[str, Any]) – Der in die Werkbank zu ladende Zustand.

_to_config() → StaticWorkbenchConfig[Quelle]#

Gib die Konfiguration aus, die erforderlich wäre, um eine neue Instanz einer Komponente zu erstellen, die der Konfiguration dieser Instanz entspricht.

Gibt zurück:: T – Die Konfiguration der Komponente.

classmethod _from_config(config: StaticWorkbenchConfig) → Self[Quelle]#

Erstelle eine neue Instanz der Komponente aus einem Konfigurationsobjekt.

Parameter:: config (T) – Das Konfigurationsobjekt.
Gibt zurück:: Self – Die neue Instanz der Komponente.

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

Bases: StaticWorkbench, StreamWorkbench

Eine Werkbank, die einen statischen Satz von Werkzeugen bereitstellt, der sich nach jeder Tool-Ausführung nicht ändert und Streaming-Ergebnisse unterstützt.

component_provider_override: ClassVar[str | None] = 'autogen_core.tools.StaticStreamWorkbench'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

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][Quelle]#

Ruft ein Tool in der Werkbank auf und gibt einen Stream von Ergebnissen zurück.

Parameter:

name (str) – Der Name des aufzurufenden Tools.
arguments (Mapping[str, Any] | None) – Die Argumente, die an das Tool übergeben werden sollen. Wenn None, wird das Tool ohne Argumente aufgerufen.
cancellation_token (CancellationToken | None) – Ein optionaler Abbruch-Token zur Unterbrechung der Tool-Ausführung.
call_id (str | None) – Eine optionale Kennung für den Tool-Aufruf, die für die Rückverfolgung verwendet wird.

pydantic model ToolOverride[Quelle]#

Bases: BaseModel

Override-Konfiguration für Name und/oder Beschreibung eines Tools.

JSON-Schema anzeigen

{
   "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"
      }
   }
}

Felder:

description (str | None)
name (str | None)

field name: str | None = None#

field description: str | None = None#