autogen_core.code_executor#

class CodeBlock(code: str, language: str)[Quelle]#

Basiert auf: object

Ein Codeblock, der aus einer Agentennachricht extrahiert wurde.

code: str#

language: str#

class CodeExecutor[Quelle]#

Bases: ABC, ComponentBase[BaseModel]

Führt Codeblöcke aus und gibt das Ergebnis zurück.

Dies ist eine abstrakte Basisklasse für Code-Executors. Sie definiert die Schnittstelle für die Ausführung von Codeblöcken und die Rückgabe des Ergebnisses. Eine konkrete Implementierung dieser Klasse sollte bereitgestellt werden, um Codeblöcke in einer bestimmten Umgebung auszuführen. Zum Beispiel führt DockerCommandLineCodeExecutor Codeblöcke in einer Kommandozeilenumgebung in einem Docker-Container aus.

Es wird empfohlen, Unterklassen als Kontextmanager zu verwenden, um sicherzustellen, dass Ressourcen ordnungsgemäß bereinigt werden. Implementieren Sie dazu die Methoden start() und stop(), die beim Betreten und Verlassen des Kontextmanagers aufgerufen werden.

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

abstract async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → CodeResult[Quelle]#

Führt Codeblöcke aus und gibt das Ergebnis zurück.

Diese Methode muss vom Code-Executor implementiert werden.

Parameter:

code_blocks (List[CodeBlock]) – Die auszuführenden Codeblöcke.

Gibt zurück:

CodeResult – Das Ergebnis der Codeausführung.

Löst aus:

ValueError – Fehler bei Benutzereingaben
TimeoutError – Codeausführung Timeouts
CancelledError – CancellationToken während der Ausführung ausgelöst

abstract async start() → None[Quelle]#: Startet den Code-Executor.

abstract async stop() → None[Quelle]#: Stoppt den Code-Executor und gibt alle Ressourcen frei.

abstract async restart() → None[Quelle]#

Startet den Code-Executor neu.

Diese Methode muss vom Code-Executor implementiert werden.

Diese Methode wird aufgerufen, wenn der Agent zurückgesetzt wird.

class CodeResult(exit_code: int, output: str)[Quelle]#

Basiert auf: object

Ergebnis einer Codeausführung.

exit_code: int#

output: str#

class Alias(name: 'str', alias: 'str')[Quelle]#

Basiert auf: object

name: str#

alias: str#

class ImportFromModule(module: 'str', imports: 'Union[Tuple[Union[str, Alias], ...], List[Union[str, Alias]]]')[Quelle]#

Basiert auf: object

module: str#

imports: Tuple[str | Alias, ...]#

class FunctionWithRequirements(func: 'Callable[P, T]', python_packages: 'Sequence[str]' = <factory>, global_imports: 'Sequence[Import]' = <factory>)[Quelle]#

Bases: Generic[T, P]

func: Callable[[P], T]#

python_packages: Sequence[str]#

global_imports: Sequence[str | ImportFromModule | Alias]#

classmethod from_callable(func: Callable[[P], T], python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirements[T, P][Quelle]#

static from_str(func: str, python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirementsStr[Quelle]#

class FunctionWithRequirementsStr(func: 'str', python_packages: 'Sequence[str]' = [], global_imports: 'Sequence[Import]' = [])[Quelle]#

Basiert auf: object

func: str#

python_packages: Sequence[str]#

global_imports: Sequence[str | ImportFromModule | Alias]#

compiled_func: Callable[[...], Any]#

with_requirements(python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → Callable[[Callable[[P], T]], FunctionWithRequirements[T, P]][Quelle]#

Dekoriert eine Funktion mit Paket- und Importanforderungen für Codeausführungsumgebungen.

Dieser Dekorator macht eine Funktion durch Einpacken in ein FunctionWithRequirements-Objekt, das ihre Abhängigkeiten verfolgt, für die Referenzierung in dynamisch ausgeführten Codeblöcken verfügbar. Wenn die dekorierte Funktion an einen Code-Executor übergeben wird, kann sie im ausgeführten Code anhand ihres Namens importiert werden, wobei alle Abhängigkeiten automatisch gehandhabt werden.

Parameter:

python_packages (Sequence[str], optional) – Von der Funktion benötigte Python-Pakete. Kann Versionsspezifikationen enthalten (z. B. [„pandas>=1.0.0“]). Standard ist [].
global_imports (Sequence[Import], optional) – Von der Funktion benötigte Importanweisungen. Können Strings („numpy“), ImportFromModule-Objekte oder Alias-Objekte sein. Standard ist [].

Gibt zurück:

Callable[[Callable[P, T]], FunctionWithRequirements[T, P]] – Ein Dekorator, der die Zielfunktion umschließt und ihre Funktionalität beibehält, während seine Abhängigkeiten registriert werden.

Beispiel

import tempfile
import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import with_requirements, CodeBlock
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
import pandas

@with_requirements(python_packages=["pandas"], global_imports=["pandas"])
def load_data() -> pandas.DataFrame:
    """Load some sample data.

    Returns:
        pandas.DataFrame: A DataFrame with sample data
    """
    data = {
        "name": ["John", "Anna", "Peter", "Linda"],
        "location": ["New York", "Paris", "Berlin", "London"],
        "age": [24, 13, 53, 33],
    }
    return pandas.DataFrame(data)

async def run_example():
    # The decorated function can be used in executed code
    with tempfile.TemporaryDirectory() as temp_dir:
        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir, functions=[load_data])
        code = f"""from {executor.functions_module} import load_data

        # Use the imported function
        data = load_data()
        print(data['name'][0])"""

        result = await executor.execute_code_blocks(
            code_blocks=[CodeBlock(language="python", code=code)],
            cancellation_token=CancellationToken(),
        )
        print(result.output)  # Output: John

# Run the async example
asyncio.run(run_example())