autogen_agentchat.teams#

Dieses Modul implementiert verschiedene vordefinierte Multi-Agenten-Teams. Jedes Team erbt von der Klasse BaseGroupChat.

class BaseGroupChat(name: str, description: str, participants: List[ChatAgent | Team], group_chat_manager_name: str, group_chat_manager_class: type[SequentialRoutedAgent], termination_condition: TerminationCondition | None = None, max_turns: int | None = None, runtime: AgentRuntime | None = None, custom_message_types: List[type[BaseAgentEvent | BaseChatMessage]] | None = None, emit_team_events: bool = False)[Quelle]#

Bases: Team, ABC, ComponentBase[BaseModel]

Die Basisklasse für Gruppenchat-Teams.

In einem Gruppenchat-Team teilen die Teilnehmer den Kontext, indem sie ihre Nachrichten an alle anderen Teilnehmer veröffentlichen.

Wenn ein ChatAgent ein Teilnehmer ist, wird die BaseChatMessage aus der chat_message der Agentenantwort an andere Teilnehmer im Gruppenchat veröffentlicht.

Wenn ein Team ein Teilnehmer ist, wird die BaseChatMessage aus den messages des Teamergebnisses an andere Teilnehmer im Gruppenchat veröffentlicht.

Um ein Gruppenchat-Team zu implementieren, erstellen Sie zuerst eine Unterklasse von BaseGroupChatManager und dann eine Unterklasse von BaseGroupChat, die den Gruppenchat-Manager verwendet.

Diese Basisklasse bietet die Zuordnung zwischen den Agenten der AgentChat-API und der Agentenlaufzeit der Core-API und behandelt High-Level-Funktionen wie das Ausführen, Pausieren, Fortsetzen und Zurücksetzen des Teams.

component_type: ClassVar[ComponentType] = 'team'#

Der logische Typ der Komponente.

property name: str#

Der Name des Gruppenchat-Teams.

property description: str#

Eine Beschreibung des Gruppenchat-Teams.

async run(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None, output_task_messages: bool = True) → TaskResult[Quelle]#

Führt das Team aus und gibt das Ergebnis zurück. Die Basisimplementierung verwendet run_stream(), um das Team auszuführen und gibt dann das Endergebnis zurück. Sobald das Team gestoppt ist, wird die Abbruchbedingung zurückgesetzt.

Parameter:

  • task (str | BaseChatMessage | Sequence[BaseChatMessage] | None) – Die Aufgabe, mit der das Team ausgeführt werden soll. Kann ein String, eine einzelne BaseChatMessage oder eine Liste von BaseChatMessage sein.

  • cancellation_token (CancellationToken | None) – Das Abbruch-Token, um die Aufgabe sofort zu beenden. Das Setzen des Abbruch-Tokens kann das Team in einen inkonsistenten Zustand versetzen und die Abbruchbedingung möglicherweise nicht zurücksetzen. Um das Team ordnungsgemäß zu beenden, verwenden Sie stattdessen ExternalTermination.

Gibt zurück:

result – Das Ergebnis der Aufgabe als TaskResult. Das Ergebnis enthält die vom Team erzeugten Nachrichten und den Grund für den Stopp.

Beispiel mit dem Team RoundRobinGroupChat

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = MaxMessageTermination(3)
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)

    result = await team.run(task="Count from 1 to 10, respond one at a time.")
    print(result)

    # Run the team again without a task to continue the previous task.
    result = await team.run()
    print(result)


asyncio.run(main())

Beispiel für die Verwendung von bool, um die Aufgabe abzubrechen

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = MaxMessageTermination(3)
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)

    cancellation_token = CancellationToken()

    # Create a task to run the team in the background.
    run_task = asyncio.create_task(
        team.run(
            task="Count from 1 to 10, respond one at a time.",
            cancellation_token=cancellation_token,
        )
    )

    # Wait for 1 second and then cancel the task.
    await asyncio.sleep(1)
    cancellation_token.cancel()

    # This will raise a cancellation error.
    await run_task


asyncio.run(main())
async run_stream(*, task: str | BaseChatMessage | Sequence[BaseChatMessage] | None = None, cancellation_token: CancellationToken | None = None, output_task_messages: bool = True) → AsyncGenerator[BaseAgentEvent | BaseChatMessage | TaskResult, None][Quelle]#

Führt das Team aus und erzeugt einen Strom von Nachrichten und das Endergebnis vom Typ TaskResult als letztes Element im Strom. Sobald das Team gestoppt ist, wird die Abbruchbedingung zurückgesetzt.

Hinweis

Wenn ein Agent ModelClientStreamingChunkEvent erzeugt, wird die Nachricht im Strom ausgegeben, aber nicht in den messages enthalten sein.

Parameter:

  • task (str | BaseChatMessage | Sequence[BaseChatMessage] | None) – Die Aufgabe, mit der das Team ausgeführt werden soll. Kann ein String, eine einzelne BaseChatMessage oder eine Liste von BaseChatMessage sein.

  • cancellation_token (CancellationToken | None) – Das Abbruch-Token, um die Aufgabe sofort zu beenden. Das Setzen des Abbruch-Tokens kann das Team in einen inkonsistenten Zustand versetzen und die Abbruchbedingung möglicherweise nicht zurücksetzen. Um das Team ordnungsgemäß zu beenden, verwenden Sie stattdessen ExternalTermination.

  • output_task_messages (bool) – Ob Aufgabenmeldungen in den Ausgabestrom aufgenommen werden sollen. Standardmäßig True für Abwärtskompatibilität.

Gibt zurück:

stream – Ein AsyncGenerator, der BaseAgentEvent, BaseChatMessage und das Endergebnis TaskResult als letztes Element im Strom liefert.

Beispiel mit dem Team RoundRobinGroupChat

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = MaxMessageTermination(3)
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)

    stream = team.run_stream(task="Count from 1 to 10, respond one at a time.")
    async for message in stream:
        print(message)

    # Run the team again without a task to continue the previous task.
    stream = team.run_stream()
    async for message in stream:
        print(message)


asyncio.run(main())

Beispiel für die Verwendung von bool, um die Aufgabe abzubrechen

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = MaxMessageTermination(3)
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)

    cancellation_token = CancellationToken()

    # Create a task to run the team in the background.
    run_task = asyncio.create_task(
        Console(
            team.run_stream(
                task="Count from 1 to 10, respond one at a time.",
                cancellation_token=cancellation_token,
            )
        )
    )

    # Wait for 1 second and then cancel the task.
    await asyncio.sleep(1)
    cancellation_token.cancel()

    # This will raise a cancellation error.
    await run_task


asyncio.run(main())
async reset() → None[Quelle]#

Setzt das Team und seine Teilnehmer auf ihren Anfangszustand zurück.

Das Team muss gestoppt sein, bevor es zurückgesetzt werden kann.

Löst aus:

RuntimeError – Wenn das Team nicht initialisiert wurde oder gerade ausgeführt wird.

Beispiel mit dem Team RoundRobinGroupChat

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = MaxMessageTermination(3)
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)
    stream = team.run_stream(task="Count from 1 to 10, respond one at a time.")
    async for message in stream:
        print(message)

    # Reset the team.
    await team.reset()
    stream = team.run_stream(task="Count from 1 to 10, respond one at a time.")
    async for message in stream:
        print(message)


asyncio.run(main())
async pause() → None[Quelle]#

Pausiert seine Teilnehmer, während das Team läuft, indem seine on_pause() Methode über direkte RPC-Aufrufe aufgerufen wird.

Achtung

Dies ist eine experimentelle Funktion, die in v0.4.9 eingeführt wurde und sich in Zukunft ändern oder entfernt werden kann.

Das Team muss initialisiert werden, bevor es pausiert werden kann.

Im Gegensatz zum Beenden führt das Pausieren des Teams nicht dazu, dass die Methode run() oder run_stream() zurückkehrt. Sie ruft die Methode on_pause() für jeden Teilnehmer auf, und wenn der Teilnehmer die Methode nicht implementiert, ist dies eine No-Op.

Hinweis

Es liegt in der Verantwortung der Agentenklasse, das Pausieren zu handhaben und sicherzustellen, dass der Agent später fortgesetzt werden kann. Stellen Sie sicher, dass Sie die Methode on_pause() in Ihrer Agentenklasse für benutzerdefinierte Pausierungsverhalten implementieren. Standardmäßig tut der Agent nichts, wenn er aufgerufen wird.

Löst aus:

RuntimeError – Wenn das Team nicht initialisiert wurde. Ausnahmen von den Teilnehmern beim Aufruf ihrer Implementierungen der on_pause Methode werden an diese Methode weitergegeben und ausgelöst.

async resume() → None[Quelle]#

Setzt seine Teilnehmer fort, während das Team läuft und pausiert ist, indem seine on_resume() Methode über direkte RPC-Aufrufe aufgerufen wird.

Achtung

Dies ist eine experimentelle Funktion, die in v0.4.9 eingeführt wurde und sich in Zukunft ändern oder entfernt werden kann.

Das Team muss initialisiert werden, bevor es fortgesetzt werden kann.

Im Gegensatz zum Beenden und Neustarten mit einer neuen Aufgabe führt das Fortsetzen des Teams nicht dazu, dass die Methode run() oder run_stream() zurückkehrt. Sie ruft die Methode on_resume() für jeden Teilnehmer auf, und wenn der Teilnehmer die Methode nicht implementiert, ist dies eine No-Op.

Hinweis

Es liegt in der Verantwortung der Agentenklasse, das Fortsetzen zu handhaben und sicherzustellen, dass der Agent dort fortfährt, wo er pausiert wurde. Stellen Sie sicher, dass Sie die Methode on_resume() in Ihrer Agentenklasse für benutzerdefiniertes Fortsetzungsverhalten implementieren.

Löst aus:

RuntimeError – Wenn das Team nicht initialisiert wurde. Ausnahmen von den Teilnehmern beim Aufruf ihrer Implementierungen der on_resume Methode werden an diese Methode weitergegeben und ausgelöst.

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

Speichert den Zustand des Gruppenchat-Teams.

Der Zustand wird gespeichert, indem die Methode agent_save_state() für jeden Teilnehmer und den Gruppenchat-Manager mit deren interner Agenten-ID aufgerufen wird. Der Zustand wird als verschachteltes Wörterbuch zurückgegeben: ein Wörterbuch mit dem Schlüssel agent_states, das ein Wörterbuch mit Agentennamen als Schlüssel und dem Zustand als Werte ist.

{
    "agent_states": {
        "agent1": ...,
        "agent2": ...,
        "RoundRobinGroupChatManager": ...
    }
}

Hinweis

Ab Version v0.4.9 wird der Zustand mit dem Agentennamen als Schlüssel anstelle der Agenten-ID gespeichert, und das Feld team_id wird aus dem Zustand entfernt. Dies ermöglicht die Portabilität des Zustands über verschiedene Teams und Laufzeiten hinweg. Mit dem alten Format gespeicherte Zustände sind möglicherweise in Zukunft nicht mit dem neuen Format kompatibel.

Vorsicht

Wenn save_state() auf ein Team aufgerufen wird, während es ausgeführt wird, ist der Zustand möglicherweise nicht konsistent und kann zu einem unerwarteten Zustand führen. Es wird empfohlen, diese Methode aufzurufen, wenn das Team nicht ausgeführt wird oder nachdem es gestoppt wurde.

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

Lädt einen externen Zustand und überschreibt den aktuellen Zustand des Gruppenchat-Teams.

Der Zustand wird geladen, indem die Methode agent_load_state() für jeden Teilnehmer und den Gruppenchat-Manager mit deren interner Agenten-ID aufgerufen wird. Siehe save_state() für das erwartete Format des Zustands.

class RoundRobinGroupChat(teilnehmer: List[ChatAgent | Team], *, name: str | None = None, beschreibung: str | None = None, abbruchbedingung: TerminationCondition | None = None, max_runden: int | None = None, runtime: AgentRuntime | None = None, custom_nachrichtentypen: List[type[BaseAgentEvent | BaseChatMessage]] | None = None, emittiere_team_ereignisse: bool = False)[quelle]#

Basisklassen: BaseGroupChat, Component[RoundRobinGroupChatConfig]

Ein Team, das einen Gruppenchat ausführt, bei dem die Teilnehmer abwechselnd auf eine runde Weise eine Nachricht an alle senden.

Wenn ein ChatAgent ein Teilnehmer ist, wird die BaseChatMessage aus der chat_message der Agentenantwort an andere Teilnehmer im Gruppenchat veröffentlicht.

Wenn ein Team ein Teilnehmer ist, wird die BaseChatMessage aus den messages des Teamergebnisses an andere Teilnehmer im Gruppenchat veröffentlicht.

Wenn sich nur ein Teilnehmer im Team befindet, ist dieser Teilnehmer der einzige Sprecher.

Parameter:

  • teilnehmer (List[ChatAgent | Team]) – Die Teilnehmer am Gruppenchat.

  • name (str | None, optional) – Der Name des Gruppenchats, verwendet DEFAULT_NAME, falls nicht angegeben. Der Name wird von einem übergeordneten Team verwendet, um diesen Gruppenchat zu identifizieren, daher muss er innerhalb des übergeordneten Teams eindeutig sein.

  • beschreibung (str | None, optional) – Die Beschreibung des Gruppenchats, verwendet DEFAULT_DESCRIPTION, falls nicht angegeben.

  • abbruchbedingung (TerminationCondition, optional) – Die Abbruchbedingung für den Gruppenchat. Standardmäßig None. Ohne eine Abbruchbedingung läuft der Gruppenchat unbegrenzt.

  • max_runden (int, optional) – Die maximale Anzahl von Runden im Gruppenchat, bevor dieser beendet wird. Standardmäßig None, was bedeutet, dass keine Begrenzung besteht.

  • custom_nachrichtentypen (List[type[BaseAgentEvent | BaseChatMessage]], optional) – Eine Liste von benutzerdefinierten Nachrichtentypen, die im Gruppenchat verwendet werden. Wenn Sie benutzerdefinierte Nachrichtentypen verwenden oder Ihre Agenten benutzerdefinierte Nachrichtentypen erzeugen, müssen Sie diese hier angeben. Stellen Sie sicher, dass Ihre benutzerdefinierten Nachrichtentypen Unterklassen von BaseAgentEvent oder BaseChatMessage sind.

  • emittiere_team_ereignisse (bool, optional) – Ob Teamereignisse über BaseGroupChat.run_stream() emittiert werden sollen. Standardmäßig False.

Löst aus:

ValueError – Wenn keine Teilnehmer angegeben werden oder wenn die Namen der Teilnehmer nicht eindeutig sind.

Examples

Ein Team mit einem Teilnehmer mit Werkzeugen

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    async def get_weather(location: str) -> str:
        return f"The weather in {location} is sunny."

    assistant = AssistantAgent(
        "Assistant",
        model_client=model_client,
        tools=[get_weather],
    )
    termination = TextMentionTermination("TERMINATE")
    team = RoundRobinGroupChat([assistant], termination_condition=termination)
    await Console(team.run_stream(task="What's the weather in New York?"))


asyncio.run(main())

Ein Team mit mehreren Teilnehmern

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent("Assistant1", model_client=model_client)
    agent2 = AssistantAgent("Assistant2", model_client=model_client)
    termination = TextMentionTermination("TERMINATE")
    team = RoundRobinGroupChat([agent1, agent2], termination_condition=termination)
    await Console(team.run_stream(task="Tell me some jokes."))


asyncio.run(main())

Ein Team aus User Proxy und einem verschachtelten Team aus Schreiber- und Prüfer-Agenten

import asyncio

from autogen_agentchat.agents import UserProxyAgent, AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

    writer = AssistantAgent(
        "writer", model_client=model_client, system_message="You are a writer.", model_client_stream=True
    )

    reviewer = AssistantAgent(
        "reviewer",
        model_client=model_client,
        system_message="Provide feedback to the input and suggest improvements.",
        model_client_stream=True,
    )

    # NOTE: you can skip input by pressing Enter.
    user_proxy = UserProxyAgent("user_proxy")

    # Maximum 1 round of review and revision.
    inner_termination = MaxMessageTermination(max_messages=4)

    # The outter-loop termination condition that will terminate the team when the user types "exit".
    outter_termination = TextMentionTermination("exit", sources=["user_proxy"])

    team = RoundRobinGroupChat(
        [
            # For each turn, the writer writes a summary and the reviewer reviews it.
            RoundRobinGroupChat([writer, reviewer], termination_condition=inner_termination),
            # The user proxy gets user input once the writer and reviewer have finished their actions.
            user_proxy,
        ],
        termination_condition=outter_termination,
    )
    # Start the team and wait for it to terminate.
    await Console(team.run_stream(task="Write a short essay about the impact of AI on society."))


asyncio.run(main())
component_config_schema#

Alias von RoundRobinGroupChatConfig

class_var component_provider_override: ClassVar[str | None] = 'autogen_agentchat.teams.RoundRobinGroupChat'#

Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

DEFAULT_NAME = 'RoundRobinGroupChat'#
DEFAULT_DESCRIPTION = 'A team of agents.'#
_to_config() → RoundRobinGroupChatConfig[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: RoundRobinGroupChatConfig) → 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 SelectorGroupChat(teilnehmer: List[ChatAgent | Team], modell_client: ChatCompletionClient, *, name: str | None = None, beschreibung: str | None = None, abbruchbedingung: TerminationCondition | None = None, max_runden: int | None = None, runtime: AgentRuntime | None = None, selector_prompt: str = 'Sie befinden sich in einem Rollenspiel. Die folgenden Rollen sind verfügbar:\n{rollen}.\nLesen Sie die folgende Konversation. Wählen Sie dann die nächste Rolle aus {teilnehmer} zum Spielen aus. Geben Sie nur die Rolle zurück.\n\n{verlauf}\n\nLesen Sie die obige Konversation. Wählen Sie dann die nächste Rolle aus {teilnehmer} zum Spielen aus. Geben Sie nur die Rolle zurück.\n', wiederholte_sprecher_erlaubt: bool = False, max_selektor_versuche: int = 3, selektor_funktion: Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], str | None] | Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[str | None]] | None = None, kandidaten_funktion: Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], List[str]] | Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[List[str]]] | None = None, custom_nachrichtentypen: List[type[BaseAgentEvent | BaseChatMessage]] | None = None, emittiere_team_ereignisse: bool = False, modell_client_streaming: bool = False, modell_kontext: ChatCompletionContext | None = None)[quelle]#

Basisklassen: BaseGroupChat, Component[SelectorGroupChatConfig]

Ein Gruppenchat-Team, bei dem die Teilnehmer abwechselnd eine Nachricht an alle senden und bei dem ein ChatCompletion-Modell den nächsten Sprecher nach jeder Nachricht auswählt.

Wenn ein ChatAgent ein Teilnehmer ist, wird die BaseChatMessage aus der chat_message der Agentenantwort an andere Teilnehmer im Gruppenchat veröffentlicht.

Wenn ein Team ein Teilnehmer ist, wird die BaseChatMessage aus den messages des Teamergebnisses an andere Teilnehmer im Gruppenchat veröffentlicht.

Parameter:

  • teilnehmer (List[ChatAgent | Team]) – Die Teilnehmer am Gruppenchat, müssen eindeutige Namen haben und mindestens zwei Teilnehmer.

  • modell_client (ChatCompletionClient) – Der ChatCompletion-Modell-Client, der zur Auswahl des nächsten Sprechers verwendet wird.

  • name (str | None, optional) – Der Name des Gruppenchats, verwendet DEFAULT_NAME, falls nicht angegeben. Der Name wird von einem übergeordneten Team verwendet, um diesen Gruppenchat zu identifizieren, daher muss er innerhalb des übergeordneten Teams eindeutig sein.

  • beschreibung (str | None, optional) – Die Beschreibung des Gruppenchats, verwendet DEFAULT_DESCRIPTION, falls nicht angegeben.

  • abbruchbedingung (TerminationCondition, optional) – Die Abbruchbedingung für den Gruppenchat. Standardmäßig None. Ohne eine Abbruchbedingung läuft der Gruppenchat unbegrenzt.

  • max_runden (int, optional) – Die maximale Anzahl von Runden im Gruppenchat, bevor dieser beendet wird. Standardmäßig None, was bedeutet, dass keine Begrenzung besteht.

  • selector_prompt (str, optional) – Die Prompt-Vorlage, die zur Auswahl des nächsten Sprechers verwendet wird. Verfügbare Felder: „{rollen}“, „{teilnehmer}“ und „{verlauf}“. {teilnehmer} sind die Namen der Kandidaten für die Auswahl. Das Format ist [„<name1>“, „<name2>“, …]. {rollen} ist eine mit Zeilenumbrüchen getrennte Liste von Namen und Beschreibungen der Kandidatenagenten. Das Format für jede Zeile ist: „<name> : <beschreibung>“. {verlauf} ist der Konversationsverlauf, formatiert als eine durch doppelte Zeilenumbrüche getrennte Liste von Namen und Nachrichteninhalt. Das Format für jede Nachricht ist: „<name> : <nachrichteninhalt>“.

  • wiederholte_sprecher_erlaubt (bool, optional) – Ob der vorherige Sprecher in die Liste der Kandidaten aufgenommen werden soll, die für die nächste Runde ausgewählt werden können. Standardmäßig False. Das Modell kann den vorherigen Sprecher trotzdem auswählen – eine Warnung wird protokolliert, wenn dies geschieht.

  • max_selektor_versuche (int, optional) – Die maximale Anzahl von Versuchen, einen Sprecher mithilfe des Modells auszuwählen. Standardmäßig 3. Wenn das Modell nach der maximalen Anzahl von Versuchen keinen Sprecher auswählen kann, wird der vorherige Sprecher verwendet, falls verfügbar, andernfalls der erste Teilnehmer.

  • selektor_funktion (Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], str | None], Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[str | None]], optional) – Eine benutzerdefinierte Selektorfunktion, die den Konversationsverlauf übernimmt und den Namen des nächsten Sprechers zurückgibt. Wenn diese Funktion bereitgestellt wird, wird sie verwendet, um das Modell zur Auswahl des nächsten Sprechers zu überschreiben. Wenn die Funktion None zurückgibt, wird das Modell verwendet, um den nächsten Sprecher auszuwählen. HINWEIS: selector_func ist nicht serialisierbar und wird während des Serialisierungs- und Deserialisierungsprozesses ignoriert.

  • candidate_func (Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], List[str]], Callable[[Sequence[BaseAgentEvent | BaseChatMessage]], Awaitable[List[str]]], optional) – Eine benutzerdefinierte Funktion, die den Gesprächsverlauf entgegennimmt und eine gefilterte Liste von Kandidaten für die Auswahl des nächsten Sprechers mithilfe eines Modells zurückgibt. Wenn die Funktion eine leere Liste oder None zurückgibt, löst SelectorGroupChat einen ValueError aus. Diese Funktion wird nur verwendet, wenn selector_func nicht gesetzt ist. allow_repeated_speaker wird ignoriert, wenn es gesetzt ist.

  • custom_nachrichtentypen (List[type[BaseAgentEvent | BaseChatMessage]], optional) – Eine Liste von benutzerdefinierten Nachrichtentypen, die im Gruppenchat verwendet werden. Wenn Sie benutzerdefinierte Nachrichtentypen verwenden oder Ihre Agenten benutzerdefinierte Nachrichtentypen erzeugen, müssen Sie diese hier angeben. Stellen Sie sicher, dass Ihre benutzerdefinierten Nachrichtentypen Unterklassen von BaseAgentEvent oder BaseChatMessage sind.

  • emittiere_team_ereignisse (bool, optional) – Ob Teamereignisse über BaseGroupChat.run_stream() emittiert werden sollen. Standardmäßig False.

  • model_client_streaming (bool, optional) – Ob Streaming für den Modellclient verwendet werden soll. (Dies ist nützlich für Reasoning-Modelle wie QwQ). Standardwert ist False.

  • model_context (ChatCompletionContext | None, optional) – Der Modellkontext zum Speichern und Abrufen von LLMMessage. Er kann mit anfänglichen Nachrichten vorab geladen werden. Nachrichten, die im Modellkontext gespeichert sind, werden für die Sprecherwahl verwendet. Die anfänglichen Nachrichten werden gelöscht, wenn das Team zurückgesetzt wird.

Löst aus:

ValueError – Wenn die Anzahl der Teilnehmer weniger als zwei beträgt oder wenn der Selektor-Prompt ungültig ist.

Examples

Ein Team mit mehreren Teilnehmern

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    async def lookup_hotel(location: str) -> str:
        return f"Here are some hotels in {location}: hotel1, hotel2, hotel3."

    async def lookup_flight(origin: str, destination: str) -> str:
        return f"Here are some flights from {origin} to {destination}: flight1, flight2, flight3."

    async def book_trip() -> str:
        return "Your trip is booked!"

    travel_advisor = AssistantAgent(
        "Travel_Advisor",
        model_client,
        tools=[book_trip],
        description="Helps with travel planning.",
    )
    hotel_agent = AssistantAgent(
        "Hotel_Agent",
        model_client,
        tools=[lookup_hotel],
        description="Helps with hotel booking.",
    )
    flight_agent = AssistantAgent(
        "Flight_Agent",
        model_client,
        tools=[lookup_flight],
        description="Helps with flight booking.",
    )
    termination = TextMentionTermination("TERMINATE")
    team = SelectorGroupChat(
        [travel_advisor, hotel_agent, flight_agent],
        model_client=model_client,
        termination_condition=termination,
    )
    await Console(team.run_stream(task="Book a 3-day trip to new york."))


asyncio.run(main())

Ein Team mit einer benutzerdefinierten Selektorfunktion

import asyncio
from typing import Sequence
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_agentchat.messages import BaseAgentEvent, BaseChatMessage


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    def check_calculation(x: int, y: int, answer: int) -> str:
        if x + y == answer:
            return "Correct!"
        else:
            return "Incorrect!"

    agent1 = AssistantAgent(
        "Agent1",
        model_client,
        description="For calculation",
        system_message="Calculate the sum of two numbers",
    )
    agent2 = AssistantAgent(
        "Agent2",
        model_client,
        tools=[check_calculation],
        description="For checking calculation",
        system_message="Check the answer and respond with 'Correct!' or 'Incorrect!'",
    )

    def selector_func(messages: Sequence[BaseAgentEvent | BaseChatMessage]) -> str | None:
        if len(messages) == 1 or messages[-1].to_text() == "Incorrect!":
            return "Agent1"
        if messages[-1].source == "Agent1":
            return "Agent2"
        return None

    termination = TextMentionTermination("Correct!")
    team = SelectorGroupChat(
        [agent1, agent2],
        model_client=model_client,
        selector_func=selector_func,
        termination_condition=termination,
    )

    await Console(team.run_stream(task="What is 1 + 1?"))


asyncio.run(main())

Ein Team mit benutzerdefiniertem Modellkontext

import asyncio

from autogen_core.model_context import BufferedChatCompletionContext
from autogen_ext.models.openai import OpenAIChatCompletionClient

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")
    model_context = BufferedChatCompletionContext(buffer_size=5)

    async def lookup_hotel(location: str) -> str:
        return f"Here are some hotels in {location}: hotel1, hotel2, hotel3."

    async def lookup_flight(origin: str, destination: str) -> str:
        return f"Here are some flights from {origin} to {destination}: flight1, flight2, flight3."

    async def book_trip() -> str:
        return "Your trip is booked!"

    travel_advisor = AssistantAgent(
        "Travel_Advisor",
        model_client,
        tools=[book_trip],
        description="Helps with travel planning.",
    )
    hotel_agent = AssistantAgent(
        "Hotel_Agent",
        model_client,
        tools=[lookup_hotel],
        description="Helps with hotel booking.",
    )
    flight_agent = AssistantAgent(
        "Flight_Agent",
        model_client,
        tools=[lookup_flight],
        description="Helps with flight booking.",
    )
    termination = TextMentionTermination("TERMINATE")
    team = SelectorGroupChat(
        [travel_advisor, hotel_agent, flight_agent],
        model_client=model_client,
        termination_condition=termination,
        model_context=model_context,
    )
    await Console(team.run_stream(task="Book a 3-day trip to new york."))


asyncio.run(main())
component_config_schema#

alias von SelectorGroupChatConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.teams.SelectorGroupChat'#

Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

DEFAULT_NAME = 'SelectorGroupChat'#
DEFAULT_DESCRIPTION = 'A team of agents.'#
_to_config() → SelectorGroupChatConfig[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: SelectorGroupChatConfig) → 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 Swarm(participants: List[ChatAgent], *, name: str | None = None, description: str | None = None, termination_condition: TerminationCondition | None = None, max_turns: int | None = None, runtime: AgentRuntime | None = None, custom_message_types: List[type[BaseAgentEvent | BaseChatMessage]] | None = None, emit_team_events: bool = False)[source]#

Bases: BaseGroupChat, Component[SwarmConfig]

Ein Gruppenchat-Team, das den nächsten Sprecher basierend auf der Handoff-Nachricht auswählt.

Der erste Teilnehmer in der Teilnehmerliste ist der anfängliche Sprecher. Der nächste Sprecher wird basierend auf der HandoffMessage ausgewählt, die vom aktuellen Sprecher gesendet wird. Wenn keine Handoff-Nachricht gesendet wird, bleibt der aktuelle Sprecher der Sprecher.

Hinweis

Im Gegensatz zu RoundRobinGroupChat und SelectorGroupChat unterstützt dieses Gruppenchat-Team keine inneren Teams als Teilnehmer.

Parameter:

  • participants (List[ChatAgent]) – Die am Gruppenchat teilnehmenden Agenten. Der erste Agent in der Liste ist der anfängliche Sprecher.

  • name (str | None, optional) – Der Name des Gruppenchats, verwendet DEFAULT_NAME, falls nicht angegeben. Der Name wird von einem übergeordneten Team verwendet, um diesen Gruppenchat zu identifizieren, daher muss er innerhalb des übergeordneten Teams eindeutig sein.

  • description (str | None, optional) – Die Beschreibung des Gruppenchats, verwendet DEFAULT_DESCRIPTION, falls nicht angegeben.

  • abbruchbedingung (TerminationCondition, optional) – Die Abbruchbedingung für den Gruppenchat. Standardmäßig None. Ohne eine Abbruchbedingung läuft der Gruppenchat unbegrenzt.

  • max_runden (int, optional) – Die maximale Anzahl von Runden im Gruppenchat, bevor dieser beendet wird. Standardmäßig None, was bedeutet, dass keine Begrenzung besteht.

  • custom_nachrichtentypen (List[type[BaseAgentEvent | BaseChatMessage]], optional) – Eine Liste von benutzerdefinierten Nachrichtentypen, die im Gruppenchat verwendet werden. Wenn Sie benutzerdefinierte Nachrichtentypen verwenden oder Ihre Agenten benutzerdefinierte Nachrichtentypen erzeugen, müssen Sie diese hier angeben. Stellen Sie sicher, dass Ihre benutzerdefinierten Nachrichtentypen Unterklassen von BaseAgentEvent oder BaseChatMessage sind.

  • emittiere_team_ereignisse (bool, optional) – Ob Teamereignisse über BaseGroupChat.run_stream() emittiert werden sollen. Standardmäßig False.

Grundlegendes Beispiel

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import MaxMessageTermination


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent1 = AssistantAgent(
        "Alice",
        model_client=model_client,
        handoffs=["Bob"],
        system_message="You are Alice and you only answer questions about yourself.",
    )
    agent2 = AssistantAgent(
        "Bob", model_client=model_client, system_message="You are Bob and your birthday is on 1st January."
    )

    termination = MaxMessageTermination(3)
    team = Swarm([agent1, agent2], termination_condition=termination)

    stream = team.run_stream(task="What is bob's birthday?")
    async for message in stream:
        print(message)


asyncio.run(main())

Verwendung von HandoffTermination für Human-in-the-Loop Handoff

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import Swarm
from autogen_agentchat.conditions import HandoffTermination, MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_agentchat.messages import HandoffMessage


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    agent = AssistantAgent(
        "Alice",
        model_client=model_client,
        handoffs=["user"],
        system_message="You are Alice and you only answer questions about yourself, ask the user for help if needed.",
    )
    termination = HandoffTermination(target="user") | MaxMessageTermination(3)
    team = Swarm([agent], termination_condition=termination)

    # Start the conversation.
    await Console(team.run_stream(task="What is bob's birthday?"))

    # Resume with user feedback.
    await Console(
        team.run_stream(
            task=HandoffMessage(source="user", target="Alice", content="Bob's birthday is on 1st January.")
        )
    )


asyncio.run(main())
component_config_schema#

alias von SwarmConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.teams.Swarm'#

Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

DEFAULT_NAME = 'Swarm'#
DEFAULT_DESCRIPTION = 'A team of agents.'#
_to_config() → SwarmConfig[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: SwarmConfig) → Swarm[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 MagenticOneGroupChat(participants: List[ChatAgent], model_client: ChatCompletionClient, *, name: str | None = None, description: str | None = None, termination_condition: TerminationCondition | None = None, max_turns: int | None = 20, runtime: AgentRuntime | None = None, max_stalls: int = 3, final_answer_prompt: str = ORCHESTRATOR_FINAL_ANSWER_PROMPT, custom_message_types: List[type[BaseAgentEvent | BaseChatMessage]] | None = None, emit_team_events: bool = False)[source]#

Bases: BaseGroupChat, Component[MagenticOneGroupChatConfig]

Ein Team, das einen Gruppenchat mit Teilnehmern ausführt, die vom MagenticOneOrchestrator verwaltet werden.

Der Orchestrator steuert den Gesprächsfluss und stellt sicher, dass die Aufgabe effizient abgeschlossen wird, indem er die Interaktionen der Teilnehmer verwaltet.

Der Orchestrator basiert auf der Magentic-One-Architektur, einem allgemeinen Multi-Agenten-System zur Lösung komplexer Aufgaben (siehe Referenzen unten).

Im Gegensatz zu RoundRobinGroupChat und SelectorGroupChat unterstützt MagenticOneGroupChat nicht die Verwendung eines Teams als Teilnehmer.

Parameter:

  • participants (List[ChatAgent]) – Die Teilnehmer am Gruppenchat.

  • model_client (ChatCompletionClient) – Der Modellclient, der zum Generieren von Antworten verwendet wird.

  • termination_condition (TerminationCondition, optional) – Die Abbruchbedingung für den Gruppenchat. Standardwert ist None. Ohne Abbruchbedingung läuft der Gruppenchat basierend auf der Orchestrator-Logik oder bis die maximale Anzahl von Zügen erreicht ist.

  • max_turns (int, optional) – Die maximale Anzahl von Zügen im Gruppenchat, bevor dieser gestoppt wird. Standardwert ist 20.

  • max_stalls (int, optional) – Die maximale Anzahl von erlaubten Pausen, bevor eine Neuplanung erfolgt. Standardwert ist 3.

  • final_answer_prompt (str, optional) – Der LLM-Prompt, der zur Generierung der endgültigen Antwort oder Reaktion aus dem Transkript des Teams verwendet wird. Ein Standardwert (sinnvoll für Modelle der GPT-4o-Klasse) ist vorhanden.

  • custom_nachrichtentypen (List[type[BaseAgentEvent | BaseChatMessage]], optional) – Eine Liste von benutzerdefinierten Nachrichtentypen, die im Gruppenchat verwendet werden. Wenn Sie benutzerdefinierte Nachrichtentypen verwenden oder Ihre Agenten benutzerdefinierte Nachrichtentypen erzeugen, müssen Sie diese hier angeben. Stellen Sie sicher, dass Ihre benutzerdefinierten Nachrichtentypen Unterklassen von BaseAgentEvent oder BaseChatMessage sind.

  • emittiere_team_ereignisse (bool, optional) – Ob Teamereignisse über BaseGroupChat.run_stream() emittiert werden sollen. Standardmäßig False.

Löst aus:

ValueError – In der Orchestrierungslogik, wenn das Fortschrittsprotokoll erforderliche Schlüssel nicht enthält oder wenn der nächste Sprecher ungültig ist.

Examples

MagenticOneGroupChat mit einem Assistenten-Agenten

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import MagenticOneGroupChat
from autogen_agentchat.ui import Console


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    assistant = AssistantAgent(
        "Assistant",
        model_client=model_client,
    )
    team = MagenticOneGroupChat([assistant], model_client=model_client)
    await Console(team.run_stream(task="Provide a different proof to Fermat last theorem"))


asyncio.run(main())

Referenzen

Wenn Sie MagenticOneGroupChat in Ihrer Arbeit verwenden, zitieren Sie bitte die folgende Arbeit

@article{fourney2024magentic,
    title={Magentic-one: A generalist multi-agent system for solving complex tasks},
    author={Fourney, Adam and Bansal, Gagan and Mozannar, Hussein and Tan, Cheng and Salinas, Eduardo and Niedtner, Friederike and Proebsting, Grace and Bassman, Griffin and Gerrits, Jack and Alber, Jacob and others},
    journal={arXiv preprint arXiv:2411.04468},
    year={2024}
}
component_config_schema#

alias von MagenticOneGroupChatConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.teams.MagenticOneGroupChat'#

Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

DEFAULT_NAME = 'MagenticOneGroupChat'#
DEFAULT_DESCRIPTION = 'A team of agents.'#
_to_config() → MagenticOneGroupChatConfig[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: MagenticOneGroupChatConfig) → 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 DiGraphBuilder[source]#

Basiert auf: object

Ein Fluent Builder zum Konstruieren von DiGraph Ausführungsgraphen, die in GraphFlow verwendet werden.

Warnung

Dies ist eine experimentelle Funktion, und die API kann sich in zukünftigen Versionen ändern.

Dieses Dienstprogramm bietet eine bequeme Möglichkeit, programmgesteuert einen Graphen von Agenteninteraktionen zu erstellen, einschließlich komplexer Ausführungsflüsse wie

  • Sequentielle Ketten

  • Parallele Fan-outs

  • Bedingte Verzweigungen

  • Zyklische Schleifen mit sicheren Ausgängen

Jeder Knoten im Graphen repräsentiert einen Agenten. Kanten definieren Ausführungspfade zwischen Agenten und können optional basierend auf Nachrichteninhalt mithilfe von aufrufbaren Funktionen konditioniert werden.

Der Builder ist kompatibel mit dem Graph-Runner und unterstützt sowohl Standard- als auch gefilterte Agenten.

- add_node(agent, activation)

Fügt einen Agentenknoten zum Graphen hinzu.

- add_edge(source, target, condition)

Verbindet zwei Knoten optional mit einer Bedingung.

- add_conditional_edges(source, condition_to_target)

Fügt mehrere bedingte Kanten von einer Quelle hinzu.

- set_entry_point(agent)

Definiert den Standard-Startknoten (optional).

- build()

Generiert einen validierten DiGraph.

- get_participants()

Gibt die Liste der hinzugefügten Agenten zurück.

Beispiel — Sequenzieller Fluss A → B → C
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b).add_edge(agent_b, agent_c)
>>> team = Graph(
...     participants=builder.get_participants(),
...     graph=builder.build(),
...     termination_condition=MaxMessageTermination(5),
... )
Beispiel — Paralleler Fan-out A → (B, C)
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b).add_edge(agent_a, agent_c)
Beispiel — Bedingte Verzweigung A → B oder A → C
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> # Add conditional edges using keyword check
>>> builder.add_edge(agent_a, agent_b, condition="keyword1")
>>> builder.add_edge(agent_a, agent_c, condition="keyword2")
Beispiel — Verwendung benutzerdefinierter String-Bedingungen
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> # Add condition strings to check in messages
>>> builder.add_edge(agent_a, agent_b, condition="big")
>>> builder.add_edge(agent_a, agent_c, condition="small")
Beispiel — Schleife: A → B → A oder B → C
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b)
>> # Add a loop back to agent A
>>> builder.add_edge(agent_b, agent_a, condition=lambda msg: "loop" in msg.to_model_text())
>>> # Add exit condition to break the loop
>>> builder.add_edge(agent_b, agent_c, condition=lambda msg: "loop" not in msg.to_model_text())
Beispiel — Schleife mit mehreren Pfaden zum selben Knoten: A → B → C → B
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
>>> builder.add_edge(agent_a, agent_b)
>>> builder.add_edge(agent_b, agent_c)
>>> builder.add_edge(agent_c, agent_b, activation_group="loop_back")
Beispiel — Schleife mit mehreren Pfaden zum selben Knoten mit beliebiger Aktivierungsbedingung: A → B → (C1, C2) → B → E(exit)
>>> builder = GraphBuilder()
>>> builder.add_node(agent_a).add_node(agent_b).add_node(agent_c1).add_node(agent_c2).add_node(agent_e)
>>> builder.add_edge(agent_a, agent_b)
>>> builder.add_edge(agent_b, agent_c1)
>>> builder.add_edge(agent_b, agent_c2)
>>> builder.add_edge(agent_b, agent_e, condition="exit")
>>> builder.add_edge(agent_c1, agent_b, activation_group="loop_back_group", activation_condition="any")
>>> builder.add_edge(agent_c2, agent_b, activation_group="loop_back_group", activation_condition="any")
add_node(agent: ChatAgent, activation: Literal['all', 'any'] = 'all') → DiGraphBuilder[source]#

Fügt einen Knoten zum Graphen hinzu und registriert dessen Agenten.

add_edge(source: str | ChatAgent, target: str | ChatAgent, condition: str | Callable[[BaseChatMessage], bool] | None = None, activation_group: str | None = None, activation_condition: Literal['all', 'any'] | None = None) → DiGraphBuilder[source]#

Fügt eine gerichtete Kante von Quelle zu Ziel hinzu, optional mit einer Bedingung.

Parameter:

  • source – Quellknoten (Agentenname oder Agentenobjekt)

  • target – Zielknoten (Agentenname oder Agentenobjekt)

  • condition – Optionale Bedingung für die Aktivierung der Kante. Wenn es sich um einen String handelt, wird die Kante aktiviert, wenn der Substring in der Nachricht gefunden wird. Wenn es sich um eine aufrufbare Funktion handelt, wird die Kante aktiviert, wenn die Funktion True für die Nachricht zurückgibt.

Gibt zurück:

Self für Methodenverkettung

Löst aus:

ValueError – Wenn Quell- oder Zielknoten im Builder nicht existieren

add_conditional_edges(source: str | ChatAgent, condition_to_target: Dict[str, str | ChatAgent]) → DiGraphBuilder[source]#

Fügt mehrere bedingte Kanten von einem Quellknoten basierend auf Schlüsselwortprüfungen hinzu.

Warnung

Dieses Methodenschnittstelle wird in Zukunft geändert, um aufrufbare Bedingungen zu unterstützen. Verwenden Sie add_edge, wenn Sie benutzerdefinierte Bedingungen angeben müssen.

Parameter:

  • source – Quellknoten (Agentenname oder Agentenobjekt)

  • condition_to_target –

    Zuordnung von Bedingungszeichenketten zu Zielknoten. Jeder Schlüssel ist ein Schlüsselwort, das im Nachrichteninhalt überprüft wird. Jeder Wert ist der Zielknoten, der aktiviert wird, wenn die Bedingung erfüllt ist.

    Für jeden Schlüssel (Schlüsselwort) wird eine Lambda-Funktion erstellt, die prüft, ob das Schlüsselwort im Nachrichtentext vorhanden ist.

Gibt zurück:

Self für Methodenverkettung

set_entry_point(name: str | ChatAgent) → DiGraphBuilder[source]#

Legt den Standard-Startknoten des Graphen fest.

build() → DiGraph[source]#

Erstellt und validiert den DiGraph.

get_participants() → list[ChatAgent][source]#

Gibt die Liste der Agenten im Builder zurück, in Einfügereihenfolge.

pydantic model DiGraph[source]#

Bases: BaseModel

Definiert eine gerichtete Graphstruktur mit Knoten und Kanten. GraphFlow verwendet dies, um die Ausführungsreihenfolge und Bedingungen zu bestimmen.

Warnung

Dies ist eine experimentelle Funktion, und die API kann sich in zukünftigen Versionen ändern.

JSON-Schema anzeigen
{
   "title": "DiGraph",
   "type": "object",
   "properties": {
      "nodes": {
         "default": null,
         "title": "Nodes"
      },
      "default_start_node": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Default Start Node"
      }
   }
}

Felder:

  • default_start_node (str | None)

  • nodes (Dict[str, autogen_agentchat.teams._group_chat._graph._digraph_group_chat.DiGraphNode])

field nodes: Dict[str, DiGraphNode] [Required]#
field default_start_node: str | None = None#
get_parents() → Dict[str, List[str]][source]#

Berechnet eine Zuordnung jedes Knotens zu seinen Elternknoten.

get_start_nodes() → Set[str][source]#

Gibt die Knoten zurück, die keine eingehenden Kanten haben (Einstiegspunkte).

get_leaf_nodes() → Set[str][source]#

Gibt die Knoten zurück, die keine ausgehenden Kanten haben (endgültige Ausgabeknoten).

has_cycles_with_exit() → bool[source]#

Prüft, ob der Graph Zyklen aufweist, und validiert, dass jeder Zyklus mindestens eine bedingte Kante hat.

Gibt zurück:

bool – True, wenn mindestens ein Zyklus vorhanden ist und alle Zyklen eine Ausfallbedingung haben. False, wenn keine Zyklen vorhanden sind.

Löst aus:

ValueError – Wenn ein Zyklus ohne bedingte Kante vorhanden ist.

get_has_cycles() → bool[source]#

Zeigt an, ob der Graph mindestens einen Zyklus (mit gültigen Ausfallbedingungen) hat.

graph_validate() → None[source]#

Validiert die Graphstruktur und die Ausführungsregeln.

get_remaining_map() → Dict[str, Dict[str, int]][source]#

Ruft die verbleibende Zuordnung ab, die verfolgt, wie viele Kanten mit jeder Aktivierungsgruppe auf jeden Zielknoten zeigen.

Gibt zurück:

Verzeichnis, das Zielknoten ihren Aktivierungsgruppen und verbleibenden Zählungen zuordnet

model_post_init(context: Any, /) → None#

Diese Funktion soll sich wie eine BaseModel-Methode verhalten, um private Attribute zu initialisieren.

Sie nimmt den Kontext als Argument entgegen, da pydantic-core ihn beim Aufruf übergibt.

Parameter:

  • self – Die BaseModel-Instanz.

  • context – Der Kontext.

pydantic model DiGraphNode[source]#

Bases: BaseModel

Stellt einen Knoten (Agenten) in einem DiGraph mit seinen ausgehenden Kanten und dem Aktivierungstyp dar.

Warnung

Dies ist eine experimentelle Funktion, und die API kann sich in zukünftigen Versionen ändern.

JSON-Schema anzeigen
{
   "title": "DiGraphNode",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "edges": {
         "default": null,
         "title": "Edges"
      },
      "activation": {
         "default": "all",
         "enum": [
            "all",
            "any"
         ],
         "title": "Activation",
         "type": "string"
      }
   },
   "required": [
      "name"
   ]
}

Felder:

  • activation (Literal['all', 'any'])

  • edges (List[autogen_agentchat.teams._group_chat._graph._digraph_group_chat.DiGraphEdge])

  • name (str)

field name: str [Required]#
field edges: List[DiGraphEdge] = []#
field activation: Literal['all', 'any'] = 'all'#
pydantic model DiGraphEdge[source]#

Bases: BaseModel

Stellt eine gerichtete Kante in einem DiGraph dar, mit einer optionalen Ausführungsbedingung.

Warnung

Dies ist eine experimentelle Funktion, und die API kann sich in zukünftigen Versionen ändern.

Warnung

Wenn die Bedingung aufrufbar ist, wird sie nicht im Modell serialisiert.

JSON-Schema anzeigen
{
   "title": "DiGraphEdge",
   "type": "object",
   "properties": {
      "target": {
         "title": "Target",
         "type": "string"
      },
      "condition": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Condition"
      },
      "condition_function": {
         "default": null,
         "title": "Condition Function"
      },
      "activation_group": {
         "default": "",
         "title": "Activation Group",
         "type": "string"
      },
      "activation_condition": {
         "default": "all",
         "enum": [
            "all",
            "any"
         ],
         "title": "Activation Condition",
         "type": "string"
      }
   },
   "required": [
      "target"
   ]
}

Felder:

  • activation_condition (Literal['all', 'any'])

  • activation_group (str)

  • condition (str | Callable[[autogen_agentchat.messages.BaseChatMessage], bool] | None)

  • condition_function (Callable[[autogen_agentchat.messages.BaseChatMessage], bool] | None)

  • target (str)

Validatoren:

  • _validate_condition » all fields

field target: str [Required]#
Validiert durch:

  • _validate_condition

field condition: str | Callable[[BaseChatMessage], bool] | None = None#

(Experimentell) Bedingung zum Ausführen dieser Kante. Wenn None, ist die Kante bedingungslos. Wenn eine Zeichenkette, ist die Kante bedingt durch das Vorhandensein dieser Zeichenkette in der letzten Agenten-Chat-Nachricht. Wenn aufrufbar, ist die Kante bedingt durch die Rückgabe von True, wenn die aufrufbare Funktion die letzte Nachricht erhält.

Validiert durch:

  • _validate_condition

field condition_function: Callable[[BaseChatMessage], bool] | None = None#
Validiert durch:

  • _validate_condition

field activation_group: str = ''#

Gruppenbezeichner für vorwärtsgerichtete Abhängigkeiten.

Wenn mehrere Kanten auf denselben Zielknoten zeigen, werden sie nach diesem Feld gruppiert. Dies ermöglicht die Unterscheidung zwischen verschiedenen Zyklen oder Abhängigkeitsmustern.

Beispiel: In einem Graphen mit einem Zyklus wie A → B → C → B können die beiden Kanten, die auf B zeigen (A → B und C → B), in unterschiedlichen Aktivierungsgruppen sein, um zu steuern, wie B aktiviert wird. Standardmäßig ist dies der Name des Zielknotens, wenn nicht anders angegeben.

Validiert durch:

  • _validate_condition

field activation_condition: Literal['all', 'any'] = 'all'#

Bestimmt, wie vorwärtsgerichtete Abhängigkeiten innerhalb derselben Aktivierungsgruppe ausgewertet werden.

  • „all“: Alle Kanten in dieser Aktivierungsgruppe müssen erfüllt sein, bevor der Zielknoten ausgeführt werden kann.

  • „any“: Jede einzelne Kante in dieser Aktivierungsgruppe, die erfüllt ist, erlaubt die Ausführung des Zielknotens.

Dies wird verwendet, um komplexe Abhängigkeitsmuster in zyklischen Graphen zu handhaben, bei denen mehrere Pfade zum selben Zielknoten führen können.

Validiert durch:

  • _validate_condition

check_condition(message: BaseChatMessage) → bool[source]#

Prüft, ob die Kantenbedingung für die gegebene Nachricht erfüllt ist.

Parameter:

message – Die Nachricht, gegen die die Bedingung geprüft wird.

Gibt zurück:

  • True, wenn die Bedingung erfüllt ist (None-Bedingung gibt immer True zurück)

  • Andernfalls False.

class GraphFlow(participants: List[ChatAgent], graph: DiGraph, *, name: str | None = None, description: str | None = None, termination_condition: TerminationCondition | None = None, max_turns: int | None = None, runtime: AgentRuntime | None = None, custom_message_types: List[type[BaseAgentEvent | BaseChatMessage]] | None = None)[source]#

Bases: BaseGroupChat, Component[GraphFlowConfig]

Ein Team, das einen Gruppenchat mit einem gerichteten Graph-Ausführungsmuster ausführt.

Warnung

Dies ist eine experimentelle Funktion, und die API kann sich in zukünftigen Versionen ändern.

Dieser Gruppenchat führt Agenten basierend auf einer gerichteten Graphstruktur (DiGraph) aus, die komplexe Workflows wie sequentielle Ausführung, paralleles Fan-out, bedingte Verzweigung, Join-Muster und Schleifen mit expliziten Ausfallbedingungen ermöglicht.

Die Ausführungsreihenfolge wird durch die im DiGraph definierten Kanten bestimmt. Jeder Knoten im Graphen entspricht einem Agenten, und Kanten definieren den Nachrichtenfluss zwischen den Agenten. Knoten können so konfiguriert werden, dass sie aktiviert werden, wenn

  • Alle übergeordneten Knoten abgeschlossen sind (activation=”all”) → Standard

  • Ein beliebiger übergeordneter Knoten abgeschlossen ist (activation=”any”)

Bedingte Verzweigung wird mit Kantenbedingungen unterstützt, bei denen die nächsten Agenten basierend auf dem Inhalt des Chatverlaufs ausgewählt werden. Schleifen sind zulässig, solange es eine Bedingung gibt, die die Schleife schließlich beendet.

Hinweis

Verwenden Sie die Klasse DiGraphBuilder, um einfach einen DiGraph zu erstellen. Sie bietet eine flüssige API zum Hinzufügen von Knoten und Kanten, zum Festlegen von Einstiegspunkten und zum Validieren der Graphstruktur. Weitere Details finden Sie in der Dokumentation zu DiGraphBuilder. Die Klasse GraphFlow ist für die Verwendung mit DiGraphBuilder zur Erstellung komplexer Workflows konzipiert.

Warnung

Bei der Verwendung von aufrufbaren Bedingungen in Kanten werden diese beim Aufruf von dump_component() nicht serialisiert. Dies wird in zukünftigen Versionen behoben.

Parameter:

  • participants (List[ChatAgent]) – Die Teilnehmer am Gruppenchat.

  • termination_condition (TerminationCondition, optional) – Abbruchbedingung für den Chat.

  • max_turns (int, optional) – Maximale Anzahl von Zügen, bevor ein Abbruch erzwungen wird.

  • graph (DiGraph) – Gerichteter Ausführungsgraph, der den Knotenfluss und die Bedingungen definiert.

Löst aus:

ValueError – Wenn Teilnehmersnamen nicht eindeutig sind oder die Graphvalidierung fehlschlägt (z. B. Zyklen ohne Ausfahrt).

Examples

Sequentieller Fluss: A → B → C

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main():
    # Initialize agents with OpenAI model clients.
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
    agent_a = AssistantAgent("A", model_client=model_client, system_message="You are a helpful assistant.")
    agent_b = AssistantAgent("B", model_client=model_client, system_message="Translate input to Chinese.")
    agent_c = AssistantAgent("C", model_client=model_client, system_message="Translate input to English.")

    # Create a directed graph with sequential flow A -> B -> C.
    builder = DiGraphBuilder()
    builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
    builder.add_edge(agent_a, agent_b).add_edge(agent_b, agent_c)
    graph = builder.build()

    # Create a GraphFlow team with the directed graph.
    team = GraphFlow(
        participants=[agent_a, agent_b, agent_c],
        graph=graph,
        termination_condition=MaxMessageTermination(5),
    )

    # Run the team and print the events.
    async for event in team.run_stream(task="Write a short story about a cat."):
        print(event)


asyncio.run(main())

Paralleles Fan-out: A → (B, C)

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main():
    # Initialize agents with OpenAI model clients.
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
    agent_a = AssistantAgent("A", model_client=model_client, system_message="You are a helpful assistant.")
    agent_b = AssistantAgent("B", model_client=model_client, system_message="Translate input to Chinese.")
    agent_c = AssistantAgent("C", model_client=model_client, system_message="Translate input to Japanese.")

    # Create a directed graph with fan-out flow A -> (B, C).
    builder = DiGraphBuilder()
    builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
    builder.add_edge(agent_a, agent_b).add_edge(agent_a, agent_c)
    graph = builder.build()

    # Create a GraphFlow team with the directed graph.
    team = GraphFlow(
        participants=[agent_a, agent_b, agent_c],
        graph=graph,
        termination_condition=MaxMessageTermination(5),
    )

    # Run the team and print the events.
    async for event in team.run_stream(task="Write a short story about a cat."):
        print(event)


asyncio.run(main())

Bedingte Verzweigung: A → B (wenn „yes“) oder C (andernfalls)

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main():
    # Initialize agents with OpenAI model clients.
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")
    agent_a = AssistantAgent(
        "A",
        model_client=model_client,
        system_message="Detect if the input is in Chinese. If it is, say 'yes', else say 'no', and nothing else.",
    )
    agent_b = AssistantAgent("B", model_client=model_client, system_message="Translate input to English.")
    agent_c = AssistantAgent("C", model_client=model_client, system_message="Translate input to Chinese.")

    # Create a directed graph with conditional branching flow A -> B ("yes"), A -> C (otherwise).
    builder = DiGraphBuilder()
    builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
    # Create conditions as callables that check the message content.
    builder.add_edge(agent_a, agent_b, condition=lambda msg: "yes" in msg.to_model_text())
    builder.add_edge(agent_a, agent_c, condition=lambda msg: "yes" not in msg.to_model_text())
    graph = builder.build()

    # Create a GraphFlow team with the directed graph.
    team = GraphFlow(
        participants=[agent_a, agent_b, agent_c],
        graph=graph,
        termination_condition=MaxMessageTermination(5),
    )

    # Run the team and print the events.
    async for event in team.run_stream(task="AutoGen is a framework for building AI agents."):
        print(event)


asyncio.run(main())

Schleife mit Ausfallbedingung: A → B → C (wenn „APPROVE“) oder A (andernfalls)

import asyncio

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main():
    # Initialize agents with OpenAI model clients.
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")
    agent_a = AssistantAgent(
        "A",
        model_client=model_client,
        system_message="You are a helpful assistant.",
    )
    agent_b = AssistantAgent(
        "B",
        model_client=model_client,
        system_message="Provide feedback on the input, if your feedback has been addressed, "
        "say 'APPROVE', otherwise provide a reason for rejection.",
    )
    agent_c = AssistantAgent(
        "C", model_client=model_client, system_message="Translate the final product to Korean."
    )

    # Create a loop graph with conditional exit: A -> B -> C ("APPROVE"), B -> A (otherwise).
    builder = DiGraphBuilder()
    builder.add_node(agent_a).add_node(agent_b).add_node(agent_c)
    builder.add_edge(agent_a, agent_b)

    # Create conditional edges using strings
    builder.add_edge(agent_b, agent_c, condition=lambda msg: "APPROVE" in msg.to_model_text())
    builder.add_edge(agent_b, agent_a, condition=lambda msg: "APPROVE" not in msg.to_model_text())

    builder.set_entry_point(agent_a)
    graph = builder.build()

    # Create a GraphFlow team with the directed graph.
    team = GraphFlow(
        participants=[agent_a, agent_b, agent_c],
        graph=graph,
        termination_condition=MaxMessageTermination(20),  # Max 20 messages to avoid infinite loop.
    )

    # Run the team and print the events.
    async for event in team.run_stream(task="Write a short poem about AI Agents."):
        print(event)


asyncio.run(main())
component_config_schema#

Alias von GraphFlowConfig

component_provider_override: ClassVar[str | None] = 'autogen_agentchat.teams.GraphFlow'#

Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

DEFAULT_NAME = 'GraphFlow'#
DEFAULT_DESCRIPTION = 'A team of agents'#
Auf dieser Seite
Auf GitHub bearbeiten
Quellcode anzeigen