Human-in-the-Loop#

Im vorherigen Abschnitt Teams haben wir gesehen, wie man ein Team von Agenten erstellt, beobachtet und steuert. Dieser Abschnitt konzentriert sich darauf, wie man mit dem Team aus Ihrer Anwendung interagiert und dem Team menschliches Feedback gibt.

Es gibt zwei Hauptmöglichkeiten, aus Ihrer Anwendung mit dem Team zu interagieren

Während der Ausführung eines Teams – der Ausführung von run() oder run_stream() – geben Sie Feedback über einen UserProxyAgent.
Nachdem die Ausführung beendet ist, geben Sie Feedback, indem Sie die nächste Aufforderung zu run() oder run_stream() übergeben.

Wir werden beide Methoden in diesem Abschnitt behandeln.

Um direkt zu Codebeispielen für die Integration mit Web- und UI-Frameworks zu gelangen, siehe die folgenden Links

Feedback während einer Ausführung geben#

Der UserProxyAgent ist ein spezieller integrierter Agent, der als Proxy für einen Benutzer fungiert, um dem Team Feedback zu geben.

Um den UserProxyAgent zu verwenden, können Sie eine Instanz davon erstellen und sie vor der Ausführung des Teams in das Team aufnehmen. Das Team entscheidet, wann der UserProxyAgent aufgerufen wird, um Feedback vom Benutzer zu erhalten.

Beispielsweise in einem RoundRobinGroupChat-Team wird der UserProxyAgent in der Reihenfolge aufgerufen, in der er dem Team übergeben wird, während in einem SelectorGroupChat-Team die Selector-Prompt oder die Selector-Funktion bestimmt, wann der UserProxyAgent aufgerufen wird.

Das folgende Diagramm veranschaulicht, wie Sie den UserProxyAgent verwenden können, um während der Ausführung eines Teams Feedback vom Benutzer zu erhalten.

human-in-the-loop-user-proxy

Die fetten Pfeile zeigen den Kontrollfluss während der Ausführung eines Teams: Wenn das Team den UserProxyAgent aufruft, wird die Kontrolle an die Anwendung/den Benutzer übertragen und auf Feedback gewartet; sobald das Feedback gegeben wurde, wird die Kontrolle an das Team zurückgegeben und das Team setzt seine Ausführung fort.

Hinweis

Wenn der UserProxyAgent während einer Ausführung aufgerufen wird, blockiert er die Ausführung des Teams, bis der Benutzer Feedback gibt oder einen Fehler meldet. Dies verzögert den Fortschritt des Teams und versetzt das Team in einen instabilen Zustand, der nicht gespeichert oder fortgesetzt werden kann.

Aufgrund der blockierenden Natur dieses Ansatzes wird empfohlen, ihn nur für kurze Interaktionen zu verwenden, die sofortiges Feedback vom Benutzer erfordern, z. B. die Aufforderung zur Genehmigung oder Ablehnung per Mausklick oder eine Benachrichtigung, die sofortige Aufmerksamkeit erfordert, andernfalls schlägt die Aufgabe fehl.

Hier ist ein Beispiel für die Verwendung des UserProxyAgent in einem RoundRobinGroupChat für eine Gedichtgenerierungsaufgabe.

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

# Create the agents.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)
user_proxy = UserProxyAgent("user_proxy", input_func=input)  # Use input() to get user input from console.

# Create the termination condition which will end the conversation when the user says "APPROVE".
termination = TextMentionTermination("APPROVE")

# Create the team.
team = RoundRobinGroupChat([assistant, user_proxy], termination_condition=termination)

# Run the conversation and stream to the console.
stream = team.run_stream(task="Write a 4-line poem about the ocean.")
# Use asyncio.run(...) when running in a script.
await Console(stream)
await model_client.close()

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
In endless blue where whispers play,  
The ocean's waves dance night and day.  
A world of depths, both calm and wild,  
Nature's heart, forever beguiled.  
TERMINATE
---------- user_proxy ----------
APPROVE

TaskResult(messages=[TextMessage(source='user', models_usage=None, metadata={}, content='Write a 4-line poem about the ocean.', type='TextMessage'), TextMessage(source='assistant', models_usage=RequestUsage(prompt_tokens=46, completion_tokens=43), metadata={}, content="In endless blue where whispers play,  \nThe ocean's waves dance night and day.  \nA world of depths, both calm and wild,  \nNature's heart, forever beguiled.  \nTERMINATE", type='TextMessage'), UserInputRequestedEvent(source='user_proxy', models_usage=None, metadata={}, request_id='2622a0aa-b776-4e54-9e8f-4ecbdf14b78d', content='', type='UserInputRequestedEvent'), TextMessage(source='user_proxy', models_usage=None, metadata={}, content='APPROVE', type='TextMessage')], stop_reason="Text 'APPROVE' mentioned")

Aus der Konsolenausgabe können Sie sehen, dass das Team über user_proxy Feedback vom Benutzer eingeholt hat, um das generierte Gedicht zu genehmigen.

Sie können Ihre eigene Eingabefunktion für den UserProxyAgent bereitstellen, um den Feedbackprozess anzupassen. Wenn das Team beispielsweise als Webdienst läuft, können Sie eine benutzerdefinierte Eingabefunktion verwenden, um auf Nachrichten von einer WebSocket-Verbindung zu warten. Der folgende Codeausschnitt zeigt ein Beispiel für eine benutzerdefinierte Eingabefunktion bei Verwendung des FastAPI-Webframeworks.

@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await websocket.accept()

    async def _user_input(prompt: str, cancellation_token: CancellationToken | None) -> str:
        data = await websocket.receive_json() # Wait for user message from websocket.
        message = TextMessage.model_validate(data) # Assume user message is a TextMessage.
        return message.content
    
    # Create user proxy with custom input function
    # Run the team with the user proxy
    # ...

Siehe die AgentChat FastAPI Sample für ein vollständiges Beispiel.

Für die Integration von ChainLit mit UserProxyAgent siehe die AgentChat ChainLit Sample.

Feedback für die nächste Ausführung geben#

Oft interagiert eine Anwendung oder ein Benutzer mit dem Agententeam in einer interaktiven Schleife: Das Team läuft bis zum Ende, die Anwendung oder der Benutzer gibt Feedback, und das Team läuft mit dem Feedback erneut.

Dieser Ansatz ist nützlich in einer persistenten Sitzung mit asynchroner Kommunikation zwischen dem Team und der Anwendung/dem Benutzer: Nachdem ein Team eine Ausführung beendet hat, speichert die Anwendung den Zustand des Teams, legt ihn in einem persistenten Speicher ab und setzt das Team fort, wenn das Feedback eintrifft.

Hinweis

Informationen zum Speichern und Laden des Zustands eines Teams finden Sie unter State Management. Dieser Abschnitt konzentriert sich auf die Feedbackmechanismen.

Das folgende Diagramm veranschaulicht den Kontrollfluss in diesem Ansatz.

human-in-the-loop-termination

Es gibt zwei Möglichkeiten, diesen Ansatz zu implementieren.

Legen Sie die maximale Anzahl von Runden fest, sodass das Team immer nach der angegebenen Anzahl von Runden stoppt.
Verwenden Sie Abbruchbedingungen wie TextMentionTermination und HandoffTermination, damit das Team entscheiden kann, wann es stoppt und die Kontrolle zurückgibt, basierend auf dem internen Zustand des Teams.

Sie können beide Methoden kombinieren, um das gewünschte Verhalten zu erzielen.

Verwendung von Max Turns#

Diese Methode ermöglicht es Ihnen, das Team für Benutzereingaben zu pausieren, indem Sie eine maximale Anzahl von Runden festlegen. Sie können das Team beispielsweise so konfigurieren, dass es stoppt, nachdem der erste Agent geantwortet hat, indem Sie max_turns auf 1 setzen. Dies ist besonders nützlich in Szenarien, in denen eine kontinuierliche Benutzerinteraktion erforderlich ist, z. B. in einem Chatbot.

Um dies zu implementieren, setzen Sie den Parameter max_turns im Konstruktor von RoundRobinGroupChat().

team = RoundRobinGroupChat([...], max_turns=1)

Sobald das Team stoppt, wird der Rundenzähler zurückgesetzt. Wenn Sie das Team fortsetzen, beginnt es wieder bei 0. Der interne Zustand des Teams bleibt jedoch erhalten, z. B. wird der RoundRobinGroupChat vom nächsten Agenten in der Liste mit demselben Gesprächsverlauf fortgesetzt.

Hinweis

max_turn ist spezifisch für die Teamklasse und wird derzeit nur von RoundRobinGroupChat, SelectorGroupChat und Swarm unterstützt. In Verbindung mit Abbruchbedingungen stoppt das Team, wenn eine der Bedingungen erfüllt ist.

Hier ist ein Beispiel für die Verwendung von max_turns in einem RoundRobinGroupChat für eine Gedichtgenerierungsaufgabe mit maximal 1 Runde.

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

# Create the agents.
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")
assistant = AssistantAgent("assistant", model_client=model_client)

# Create the team setting a maximum number of turns to 1.
team = RoundRobinGroupChat([assistant], max_turns=1)

task = "Write a 4-line poem about the ocean."
while True:
    # Run the conversation and stream to the console.
    stream = team.run_stream(task=task)
    # Use asyncio.run(...) when running in a script.
    await Console(stream)
    # Get the user response.
    task = input("Enter your feedback (type 'exit' to leave): ")
    if task.lower().strip() == "exit":
        break
await model_client.close()

---------- user ----------
Write a 4-line poem about the ocean.
---------- assistant ----------
Endless waves in a dance with the shore,  
Whispers of secrets in tales from the roar,  
Beneath the vast sky, where horizons blend,  
The ocean’s embrace is a timeless friend.  
TERMINATE
[Prompt tokens: 46, Completion tokens: 48]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of turns 1 reached.
Total prompt tokens: 46
Total completion tokens: 48
Duration: 1.63 seconds
---------- user ----------
Can you make it about a person and its relationship with the ocean
---------- assistant ----------
She walks along the tide, where dreams intertwine,  
With every crashing wave, her heart feels aligned,  
In the ocean's embrace, her worries dissolve,  
A symphony of solace, where her spirit evolves.  
TERMINATE
[Prompt tokens: 117, Completion tokens: 49]
---------- Summary ----------
Number of messages: 2
Finish reason: Maximum number of turns 1 reached.
Total prompt tokens: 117
Total completion tokens: 49
Duration: 1.21 seconds

Sie sehen, dass das Team sofort nach der Antwort eines Agenten gestoppt hat.

Verwendung von Abbruchbedingungen#

Wir haben bereits mehrere Beispiele für Abbruchbedingungen in den vorherigen Abschnitten gesehen. In diesem Abschnitt konzentrieren wir uns auf HandoffTermination, das das Team stoppt, wenn ein Agent eine HandoffMessage sendet.

Erstellen wir ein Team mit einem einzelnen AssistantAgent-Agenten mit Handoff-Einstellung und führen das Team mit einer Aufgabe aus, die zusätzliche Eingaben vom Benutzer erfordert, da der Agent keine relevanten Werkzeuge zur weiteren Verarbeitung der Aufgabe hat.

Hinweis

Das mit AssistantAgent verwendete Modell muss Tool-Aufrufe unterstützen, um die Handoff-Funktion nutzen zu können.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import Handoff
from autogen_agentchat.conditions import HandoffTermination, TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient

# Create an OpenAI model client.
model_client = OpenAIChatCompletionClient(
    model="gpt-4o",
    # api_key="sk-...", # Optional if you have an OPENAI_API_KEY env variable set.
)

# Create a lazy assistant agent that always hands off to the user.
lazy_agent = AssistantAgent(
    "lazy_assistant",
    model_client=model_client,
    handoffs=[Handoff(target="user", message="Transfer to user.")],
    system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with 'TERMINATE'.",
)

# Define a termination condition that checks for handoff messages.
handoff_termination = HandoffTermination(target="user")
# Define a termination condition that checks for a specific text mention.
text_termination = TextMentionTermination("TERMINATE")

# Create a single-agent team with the lazy assistant and both termination conditions.
lazy_agent_team = RoundRobinGroupChat([lazy_agent], termination_condition=handoff_termination | text_termination)

# Run the team and stream to the console.
task = "What is the weather in New York?"
await Console(lazy_agent_team.run_stream(task=task), output_stats=True)

---------- user ----------
What is the weather in New York?
---------- lazy_assistant ----------
[FunctionCall(id='call_EAcMgrLGHdLw0e7iJGoMgxuu', arguments='{}', name='transfer_to_user')]
[Prompt tokens: 69, Completion tokens: 12]
---------- lazy_assistant ----------
[FunctionExecutionResult(content='Transfer to user.', call_id='call_EAcMgrLGHdLw0e7iJGoMgxuu')]
---------- lazy_assistant ----------
Transfer to user.
---------- Summary ----------
Number of messages: 4
Finish reason: Handoff to user from lazy_assistant detected.
Total prompt tokens: 69
Total completion tokens: 12
Duration: 0.69 seconds

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='What is the weather in New York?', type='TextMessage'), ToolCallRequestEvent(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=69, completion_tokens=12), content=[FunctionCall(id='call_EAcMgrLGHdLw0e7iJGoMgxuu', arguments='{}', name='transfer_to_user')], type='ToolCallRequestEvent'), ToolCallExecutionEvent(source='lazy_assistant', models_usage=None, content=[FunctionExecutionResult(content='Transfer to user.', call_id='call_EAcMgrLGHdLw0e7iJGoMgxuu')], type='ToolCallExecutionEvent'), HandoffMessage(source='lazy_assistant', models_usage=None, target='user', content='Transfer to user.', context=[], type='HandoffMessage')], stop_reason='Handoff to user from lazy_assistant detected.')

Sie sehen, dass das Team aufgrund der erkannten Handoff-Nachricht gestoppt wurde. Lassen Sie uns das Team fortsetzen, indem wir die Informationen bereitstellen, die der Agent benötigt.

await Console(lazy_agent_team.run_stream(task="The weather in New York is sunny."))

---------- user ----------
The weather in New York is sunny.
---------- lazy_assistant ----------
Great! Enjoy the sunny weather in New York! Is there anything else you'd like to know?
---------- lazy_assistant ----------
TERMINATE

TaskResult(messages=[TextMessage(source='user', models_usage=None, content='The weather in New York is sunny.', type='TextMessage'), TextMessage(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=110, completion_tokens=21), content="Great! Enjoy the sunny weather in New York! Is there anything else you'd like to know?", type='TextMessage'), TextMessage(source='lazy_assistant', models_usage=RequestUsage(prompt_tokens=137, completion_tokens=5), content='TERMINATE', type='TextMessage')], stop_reason="Text 'TERMINATE' mentioned")

Sie sehen, dass das Team nach der Bereitstellung der Informationen durch den Benutzer fortgesetzt wurde.

Hinweis

Wenn Sie ein Swarm-Team mit HandoffTermination, das auf den Benutzer abzielt, verwenden, müssen Sie zum Fortsetzen des Teams die task auf eine HandoffMessage mit dem target auf den nächsten auszuführenden Agenten setzen. Weitere Einzelheiten finden Sie unter Swarm.