Agent und Agent Runtime

In diesem und den folgenden Abschnitten konzentrieren wir uns auf die Kernkonzepte von AutoGen: Agents, Agent Runtime, Nachrichten und Kommunikation – die grundlegenden Bausteine für Anwendungen mit mehreren Agenten.

Hinweis

Die Core API ist so konzipiert, dass sie unvoreingenommen und flexibel ist. Daher kann es manchmal herausfordernd sein. Machen Sie weiter, wenn Sie ein interaktives, skalierbares und verteiltes Multi-Agenten-System aufbauen und die volle Kontrolle über alle Arbeitsabläufe wünschen. Wenn Sie einfach nur schnell etwas zum Laufen bringen möchten, können Sie sich die AgentChat API ansehen.

Ein Agent in AutoGen ist eine Entität, die durch die Basis-Schnittstelle Agent definiert wird. Er hat eine eindeutige Kennung vom Typ AgentId, ein Metadaten-Dictionary vom Typ AgentMetadata.

In den meisten Fällen können Sie Ihre Agents von der übergeordneten Klasse RoutedAgent ableiten, die es Ihnen ermöglicht, Nachrichten an den entsprechenden Nachrichtenhandler weiterzuleiten, der mit dem message_handler()-Decorator und der richtigen Typ-Hint für die Variable message angegeben ist. Eine Agent Runtime ist die Ausführungsumgebung für Agents in AutoGen.

Ähnlich wie die Laufzeitumgebung einer Programmiersprache bietet eine Agent Runtime die notwendige Infrastruktur, um die Kommunikation zwischen Agents zu erleichtern, Agenten-Lebenszyklen zu verwalten, Sicherheitsgrenzen durchzusetzen und Monitoring und Debugging zu unterstützen.

Für die lokale Entwicklung können Entwickler SingleThreadedAgentRuntime verwenden, die in eine Python-Anwendung eingebettet werden kann.

Hinweis

Agents werden nicht direkt von der Anwendung durch Instanziierung verwaltet. Stattdessen werden sie von der Runtime bei Bedarf erstellt und von der Runtime verwaltet.

Wenn Sie bereits mit AgentChat vertraut sind, ist es wichtig zu beachten, dass die Agents von AgentChat, wie z. B. AssistantAgent, von der Anwendung erstellt werden und somit nicht direkt von der Runtime verwaltet werden. Um einen AgentChat-Agenten in Core zu verwenden, müssen Sie einen Wrapper-Core-Agenten erstellen, der Nachrichten an den AgentChat-Agenten delegiert und die Runtime den Wrapper-Agenten verwalten lässt.

Implementierung eines Agenten

Um einen Agenten zu implementieren, muss der Entwickler die Klasse RoutedAgent ableiten und für jeden Nachrichtentyp, den der Agent verarbeiten soll, eine Nachrichten-Handler-Methode mit dem message_handler()-Decorator implementieren. Zum Beispiel verarbeitet der folgende Agent einen einfachen Nachrichtentyp MyMessageType und gibt die empfangene Nachricht aus.

from dataclasses import dataclass

from autogen_core import AgentId, MessageContext, RoutedAgent, message_handler


@dataclass
class MyMessageType:
    content: str


class MyAgent(RoutedAgent):
    def __init__(self) -> None:
        super().__init__("MyAgent")

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")

Dieser Agent verarbeitet nur MyMessageType und Nachrichten werden an die Methode handle_my_message_type geliefert. Entwickler können mehrere Nachrichten-Handler für verschiedene Nachrichtentypen haben, indem sie den message_handler()-Decorator verwenden und den Typ-Hint für die Variable message in der Handler-Funktion festlegen. Sie können auch die Python-Typenvereinigung (Union) für die Variable message in einer Nachrichten-Handler-Funktion nutzen, wenn dies besser zur Logik des Agenten passt. Lesen Sie den nächsten Abschnitt über Nachrichten und Kommunikation.

Verwendung eines AgentChat-Agenten

Wenn Sie einen AgentChat-Agenten haben und ihn in der Core API verwenden möchten, können Sie einen Wrapper-RoutedAgent erstellen, der Nachrichten an den AgentChat-Agenten delegiert. Das folgende Beispiel zeigt, wie ein Wrapper-Agent für den AssistantAgent in AgentChat erstellt wird.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.models.openai import OpenAIChatCompletionClient


class MyAssistant(RoutedAgent):
    def __init__(self, name: str) -> None:
        super().__init__(name)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        self._delegate = AssistantAgent(name, model_client=model_client)

    @message_handler
    async def handle_my_message_type(self, message: MyMessageType, ctx: MessageContext) -> None:
        print(f"{self.id.type} received message: {message.content}")
        response = await self._delegate.on_messages(
            [TextMessage(content=message.content, source="user")], ctx.cancellation_token
        )
        print(f"{self.id.type} responded: {response.chat_message}")

Wie man Client-Modelle verwendet, finden Sie im Abschnitt Model Client.

Da die Core API unvoreingenommen ist, sind Sie nicht verpflichtet, die AgentChat API zu verwenden, um die Core API zu nutzen. Sie können Ihre eigenen Agents implementieren oder ein anderes Agenten-Framework verwenden.

Registrierung von Agententypen

Damit Agents für die Runtime verfügbar sind, können Entwickler die Klassenmethode register() der Klasse BaseAgent verwenden. Der Registrierungsprozess ordnet einem Agententyp, der eindeutig durch einen String identifiziert wird, und einer Factory-Funktion zu, die eine Instanz des Agententyps der gegebenen Klasse erstellt. Die Factory-Funktion wird verwendet, um die automatische Erstellung von Agent-Instanzen zu ermöglichen, wenn diese benötigt werden.

Agententyp (AgentType) ist nicht dasselbe wie die Agentenklasse. In diesem Beispiel ist der Agententyp AgentType("my_agent") oder AgentType("my_assistant") und die Agentenklasse ist die Python-Klasse MyAgent oder MyAssistantAgent. Von der Factory-Funktion wird erwartet, dass sie eine Instanz der Agentenklasse zurückgibt, auf der die Klassenmethode register() aufgerufen wird. Lesen Sie Agent Identity and Lifecycles, um mehr über Agententypen und Identitäten zu erfahren.

Hinweis

Verschiedene Agententypen können mit Factory-Funktionen registriert werden, die dieselbe Agentenklasse zurückgeben. Zum Beispiel können in den Factory-Funktionen Variationen der Konstruktorparameter verwendet werden, um verschiedene Instanzen derselben Agentenklasse zu erstellen.

Um unsere Agententypen bei der SingleThreadedAgentRuntime zu registrieren, kann der folgende Code verwendet werden.

from autogen_core import SingleThreadedAgentRuntime

runtime = SingleThreadedAgentRuntime()
await MyAgent.register(runtime, "my_agent", lambda: MyAgent())
await MyAssistant.register(runtime, "my_assistant", lambda: MyAssistant("my_assistant"))

AgentType(type='my_assistant')

Sobald ein Agententyp registriert ist, können wir eine direkte Nachricht an eine Agenteninstanz mit einer AgentId senden. Die Runtime erstellt die Instanz beim ersten Mal, wenn sie eine Nachricht an diese Instanz liefert.

runtime.start()  # Start processing messages in the background.
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_agent", "default"))
await runtime.send_message(MyMessageType("Hello, World!"), AgentId("my_assistant", "default"))
await runtime.stop()  # Stop processing messages in the background.

my_agent received message: Hello, World!
my_assistant received message: Hello, World!
my_assistant responded: Hello! How can I assist you today?

Hinweis

Da die Runtime den Lebenszyklus von Agents verwaltet, wird eine AgentId nur zur Kommunikation mit dem Agenten oder zum Abrufen seiner Metadaten (z. B. Beschreibung) verwendet.

Ausführung der Single-Threaded Agent Runtime

Der obige Code-Schnipsel verwendet start(), um einen Hintergrundtask zu starten, der Nachrichten verarbeitet und an die Nachrichten-Handler der Empfänger liefert. Dies ist ein Merkmal der lokalen eingebetteten Runtime SingleThreadedAgentRuntime.

Um den Hintergrundtask sofort zu beenden, verwenden Sie die Methode stop().

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop()  # This will return immediately but will not cancel
# any in-progress message handling.

Sie können den Hintergrundtask wieder aufnehmen, indem Sie erneut start() aufrufen.

Für Batch-Szenarien, wie z. B. das Ausführen von Benchmarks zur Bewertung von Agents, möchten Sie möglicherweise warten, bis der Hintergrundtask automatisch beendet wird, wenn keine unverarbeiteten Nachrichten vorhanden sind und kein Agent Nachrichten verarbeitet – der Batch kann als abgeschlossen betrachtet werden. Dies können Sie mit der Methode stop_when_idle() erreichen.

runtime.start()
# ... Send messages, publish messages, etc.
await runtime.stop_when_idle()  # This will block until the runtime is idle.

Um die Runtime zu schließen und Ressourcen freizugeben, verwenden Sie die Methode close().

await runtime.close()

Andere Runtime-Implementierungen haben ihre eigenen Wege, die Runtime auszuführen.