前のセクションでは、エージェントのチームを作成し、観察し、そして制御する方法を見てきました。このセクションは、アプリケーションからチームと相互作用し、人間のフィードバックをチームに提供する方法に焦点を当てていきます。
AutoGen AgentChat : Tutorial : ヒューマン・イン・ザ・ループ (人間参加型)
作成 : クラスキャット・セールスインフォメーション
作成日時 : 04/30/2025
* 本記事は github microsoft/autogen の以下のページを独自に翻訳した上でまとめ直し、補足説明を加えています :
* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。
◆ お問合せ : 下記までお願いします。
- クラスキャット セールス・インフォメーション
- sales-info@classcat.com
- ClassCatJP
AutoGen AgentChat : Tutorial : ヒューマン・イン・ザ・ループ (人間参加型)
前のセクション チーム では、エージェントのチームを作成し、観察し、そして制御する方法を見てきました。このセクションは、アプリケーションからチームと相互作用し、人間のフィードバックをチームに提供する方法に焦点を当てていきます。
アプリケーションからチームと相互作用するには 2 つの主要な方法があります :
- チームの実行中 – run() or run_stream() の実行中、UserProxyAgent を通してフィードバックを提供します。
- 実行が終了すれば、フィードバックは run() or run_stream() の次の呼び出しへの入力を通して提供します。
We will cover both methods in this section.
web と UI フレームワークとの統合についてのコードサンプルに直接移行するには、以下のリンクをご覧ください :
実行中にフィードバックを提供する
UserProxyAgent は特別な組み込みエージェントで、ユーザがフィードバックをチームに提供するためのプロキシーとして機能します。
UserProxyAgent を使用するには、チームを実行する前にそのインスタンスを作成して、それをチームに含めることができます。チームはいつ UserProxyAgent を呼び出してユーザからのフィードバックを求めるか決定します。
例えば RoundRobinGroupChat チームでは、UserProxyAgent はチームに渡される順序で呼び出されますが、SelectorGroupChat チームでは、selector プロンプトや selector 関数がいつ UserProxyAgent が呼び出されるか決定します。
次の図は、チームの実行中に UserProxyAgent を使用してユーザからフィードバックを取得する方法を示します :
太字の矢印はチーム実行中の制御フローを示しています : チームが UserProxyAgent を呼び出すと、それは制御をアプリケーション/ユーザに移し、フィードバックを待ちます ; フィードバックが提供されれば、制御はチームに戻されてチームは実行を続行します。
Note : UserProxyAgent が実行中に呼び出されると、ユーザがフィードバックを提供するかエラーが発生するまで、それはチームの実行をブロックします。これはチームの進捗を一時的に止めて、チームを保存したり再開できない unstable な状態に置きます。
このアプローチのブロックする性質により、ボタンクリックによる承認/非承認を求めたり、即時の注目が必要な (さもないとタスクが失敗する) アラートのような、ユーザからの即時のフィードバックを必要とする短いインタラクションのためだけに使用することが推奨されます。
詩生成タスクについて RoundRobinGroupChat で UserProxyAgent を使用する方法の例があります :
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")
コンソール出力から、チームが生成された詩を承認するために user_proxy を通してユーザからのフィードバックを求めたことがわかります。
UserProxyAgent に独自の入力関数を提供してフィードバック・プロセスをカスタマイズすることができます。例えば、チームが web サービスとして動作している場合、カスタム入力関数を使用して web ソケット接続からのメッセージを待機できます。次のコードスニペットは、FastAPI web フレームワークを使用している場合のカスタム入力関数の例を示しています :
@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
# ...
See the AgentChat FastAPI sample for a complete example.
For ChainLit integration with UserProxyAgent, see the AgentChat ChainLit sample.
次の実行へのフィードバックの提供
多くの場合、アプリケーションやユーザは対話型 (インタラクティブ) ループでエージェントのチームと相互作用します : チームは終了まで実行され、アプリケーションやユーザはフィードバックを提供し、そしてチームはそのフィードバックを使用して再度実行されます。
このアプローチは、チームとアプリケーション/ユーザ間の非同期通信を伴う永続化されたセッションで役立ちます : チームが実行を終了すれば、アプリケーションはチームの状態を保存し、永続的なストレージに置いて、そしてフィードバックを取得したらチームを再開します。
Note : チームの状態を保存してロードする方法については、Managing State を参照してください。このフィードバックはフィードバック機構に焦点を当てます。
次の図はこのアプローチでの制御フローを図示しています :
このアプローチを実装するには 2 つの方法があります :
- 指定したターン数後にチームが必ず停止するように、最大ターン数を設定します。
- TextMentionTermination や HandoffTermination のような終了条件を使用して、チームの内部状態が与えられたとき、チームがいつ停止して制御を戻すか決定することを可能にします。
両方の手法を一緒に使用して望ましい動作を実装することができます。
最大ターン数の使用
この手法は最大ターン数を設定することでチームをユーザ入力のために一時停止することを可能にします。例えば、max_turns を 1 に設定して最初のエージェントが応答した後にチームが停止するように構成できます。これは、チャットボットのように連続的なユーザのエンゲージメント (関与) が必要なシナリオで特に役立ちます。
これを実装するには、RoundRobinGroupChat() コンストラクタで max_turns パラメータを設定します。
team = RoundRobinGroupChat([...], max_turns=1)
チームが停止したら、ターンカウントはリセットされます。チームを再開すると、それは再度 0 から始まります。ただし、チームの内部状態は保存されます、例えば、RoundRobinGroupChat は同じ会話履歴のリストの次のエージェントから再開されます。
Note : max_turn はチームのクラス固有のもので、現在は RoundRobinGroupChat, SelectorGroupChat と Swarm によりサポートされているだけです。終了条件とともに使用されると、チームはいずれかの条件が満たされた場合に停止します。
ここに、最大 1 のターンを持つ、詩生成タスクのための RoundRobinGroupChat で max_turns を使用する方法の例があります :
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
You can see that the team stopped immediately after one agent responded.
終了条件の使用
前のセクションで終了条件の幾つかの例を既に見てきました。このセクションでは、HandoffTermination に焦点を当てます、これはエージェントが HandoffMessage メッセージを送信する場合にチームを停止します。
ハンドオフ設定を持つ単一の AssistantAgent エージェントでチームを作成して、(エージェントがタスクを処理し続ける関連ツールを持たないために) ユーザからの追加の入力を必要とするタスクを持つチームを実行しましょう。
Note : AssistantAgent で使用されるモデルはハンドオフ機能を使用するためにツール呼び出しをサポートしている必要があります。
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.')
ハンドオフ・メッセージが検出されたためにチームが停止したことがわかります。エージェントが必要とする情報の提供によりチームを継続しましょう。
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")
You can see the team continued after the user provided the information.
以上