注
.NET AG-UI統合のワークフロー サポートは近日公開予定です。
このチュートリアルでは、AG-UI エンドポイントを介して Agent Framework ワークフローを公開する方法について説明します。 ワークフローは、定義された実行グラフ内の複数のエージェントとツールを調整し、AG-UI 統合により、ステップ追跡、アクティビティ スナップショット、割り込み、カスタム イベントなどの豊富なワークフロー イベントが Web クライアントにリアルタイムでストリーミングされます。
前提条件
始める前に、以下のことを確認してください:
- Python 3.10 以降
-
agent-framework-ag-uiインストールされました - 作業の開始に関するチュートリアルに関する知識
- Agent Framework ワークフローの基本的な理解
AG-UI でワークフローを使用する場合
必要な場合は、1 つのエージェントではなくワークフローを使用します。
- マルチエージェント オーケストレーション: 特殊なエージェント間でタスクをルーティングする (トリアージ→払い戻し→注文など)
-
構造化された実行手順:
STEP_STARTED/STEP_FINISHEDイベントを使用して定義済みのステージの進行状況を追跡する - インタラプト/リジュームフロー: 実行を一時停止し、人間の入力や承認を収集した後、再開する
-
カスタム イベント ストリーミング: ドメイン固有のイベント (
request_info、status、workflow_output) をクライアントに出力する
AgentFrameworkWorkflow を使用したワークフローのラッピング
AgentFrameworkWorkflow は、ネイティブ Workflow を AG-UI プロトコルに適合させる軽量ラッパーです。 事前構築済みのワークフロー インスタンスまたはスレッドごとに新しいワークフローを作成するファクトリを提供できます。
直接インスタンス
1 つのワークフロー オブジェクトがすべての要求 (ステートレス パイプラインなど) を安全に処理できる場合は、ダイレクト インスタンスを使用します。
from agent_framework import Workflow
from agent_framework.ag_ui import AgentFrameworkWorkflow
workflow = build_my_workflow() # returns a Workflow
ag_ui_workflow = AgentFrameworkWorkflow(
workflow=workflow,
name="my-workflow",
description="Single-instance workflow.",
)
スレッド スコープ ファクトリ
各会話スレッドに独自のワークフロー状態が必要な場合は、 workflow_factory を使用します。 ファクトリは thread_id を受け取り、新しい Workflowを返します。
from agent_framework.ag_ui import AgentFrameworkWorkflow
ag_ui_workflow = AgentFrameworkWorkflow(
workflow_factory=lambda thread_id: build_my_workflow(),
name="my-workflow",
description="Thread-scoped workflow.",
)
Important
両方ではなくworkflowworkflow_factory渡す必要があります。 両方が指定されている場合、ラッパーは ValueError を発生させます。
エンドポイントの登録
1 つのエージェントを登録するのと同じ方法 add_agent_framework_fastapi_endpoint ワークフローを登録します。
from fastapi import FastAPI
from agent_framework.ag_ui import (
AgentFrameworkWorkflow,
add_agent_framework_fastapi_endpoint,
)
app = FastAPI(title="Workflow AG-UI Server")
ag_ui_workflow = AgentFrameworkWorkflow(
workflow_factory=lambda thread_id: build_my_workflow(),
name="handoff-demo",
description="Multi-agent handoff workflow.",
)
add_agent_framework_fastapi_endpoint(
app=app,
agent=ag_ui_workflow,
path="/workflow",
)
そのままの Workflow を直接渡すこともできます。エンドポイントは AgentFrameworkWorkflow で自動的にラップします。
add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")
ワークフローによって発生するAG-UIイベント
ワークフロー実行では、単一エージェントの実行と比較して、より豊富な AG-UI イベントのセットが生成されます。
| イベント | 放出時 | 説明 |
|---|---|---|
RUN_STARTED |
実行の開始 | ワークフロー実行の開始をマークします |
STEP_STARTED |
エグゼキューターまたはスーパーステップが開始する |
step_name はエージェントまたはステップを識別します (例: "triage_agent") |
TEXT_MESSAGE_* |
エージェントがテキストを生成する | 標準のストリーミング テキスト イベント |
TOOL_CALL_* |
エージェントがツールを呼び出す | 標準ツール呼び出しイベント |
STEP_FINISHED |
スーパーステップまたはエグゼキューターが完了する | UI 進行状況の追跡の手順を閉じます。 |
CUSTOM (status) |
ワークフローの状態の変更 | イベント値に {"state": "<value>"} が含まれています |
CUSTOM (request_info) |
ワークフローが人間の入力を要求する | クライアントがプロンプトを表示するための要求ペイロードが含まれています |
CUSTOM (workflow_output) |
ワークフローで出力が生成される | 最終的な出力データまたは中間出力データを格納します。 |
RUN_FINISHED |
実行の完了 | ワークフローが入力を待機している場合は、 interrupts を含めることができます |
クライアントは、 STEP_STARTED / STEP_FINISHED イベントを使用して、現在アクティブなエージェントを示す進行状況インジケーターを表示できます。
割り込みと再開
ワークフローでは、実行を一時停止して、人間の入力またはツールの承認を収集できます。 AG-UI 統合では、割り込み/再開プロトコルを使用してこれを処理します。
割り込みの動作原理
実行中に、ワークフローは保留中の要求 (詳細を要求する
HandoffAgentUserRequestや、approval_mode="always_require"を含むツールなど) を発生させます。AG-UI ブリッジは、要求データを含む
CUSTOMを含むname="request_info"イベントを出力します。RUN_FINISHEDフィールドに保留中の要求オブジェクトの一覧が含まれるinterruptsイベントで実行が完了します。{ "type": "RUN_FINISHED", "threadId": "abc123", "runId": "run_xyz", "interrupts": [ { "id": "request-id-1", "value": { "request_type": "HandoffAgentUserRequest", "data": "..." } } ] }クライアントは、ユーザーが応答するための UI (テキスト入力、承認ボタンなど) をレンダリングします。
履歴書のしくみ
クライアントは、割り込み ID でキー付けされたユーザーの応答を含む resume ペイロードで新しい要求を送信します。
{
"threadId": "abc123",
"messages": [],
"resume": {
"interrupts": [
{
"id": "request-id-1",
"value": "User's response text or approval decision"
}
]
}
}
サーバーは、再開ペイロードをワークフロー応答に変換し、一時停止した場所から実行を続行します。
完全な例: マルチエージェントハンドオフワークフロー
この例では、3 人のエージェントが互いに作業を引き渡し、承認を必要とするツールを使用し、必要に応じて人間の入力を要求するカスタマー サポート ワークフローを示します。
エージェントとツールを定義する
"""AG-UI workflow server with multi-agent handoff."""
import os
from agent_framework import Agent, Message, Workflow, tool
from agent_framework.ag_ui import (
AgentFrameworkWorkflow,
add_agent_framework_fastapi_endpoint,
)
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffBuilder
from azure.identity import AzureCliCredential
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
@tool(approval_mode="always_require")
def submit_refund(refund_description: str, amount: str, order_id: str) -> str:
"""Capture a refund request for manual review before processing."""
return f"Refund recorded for order {order_id} (amount: {amount}): {refund_description}"
@tool(approval_mode="always_require")
def submit_replacement(order_id: str, shipping_preference: str, replacement_note: str) -> str:
"""Capture a replacement request for manual review before processing."""
return f"Replacement recorded for order {order_id} (shipping: {shipping_preference}): {replacement_note}"
@tool(approval_mode="never_require")
def lookup_order_details(order_id: str) -> dict[str, str]:
"""Return order details for a given order ID."""
return {
"order_id": order_id,
"item_name": "Wireless Headphones",
"amount": "$129.99",
"status": "delivered",
}
ワークフローを構築する
def create_handoff_workflow() -> Workflow:
"""Build a handoff workflow with triage, refund, and order agents."""
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
triage = Agent(id="triage_agent", name="triage_agent", instructions="...", client=client)
refund = Agent(id="refund_agent", name="refund_agent", instructions="...", client=client,
tools=[lookup_order_details, submit_refund])
order = Agent(id="order_agent", name="order_agent", instructions="...", client=client,
tools=[lookup_order_details, submit_replacement])
def termination_condition(conversation: list[Message]) -> bool:
for msg in reversed(conversation):
if msg.role == "assistant" and (msg.text or "").strip().lower().endswith("case complete."):
return True
return False
builder = HandoffBuilder(
name="support_workflow",
participants=[triage, refund, order],
termination_condition=termination_condition,
)
builder.add_handoff(triage, [refund], description="Route refund requests.")
builder.add_handoff(triage, [order], description="Route replacement requests.")
builder.add_handoff(refund, [order], description="Route to order after refund.")
builder.add_handoff(order, [triage], description="Route back after completion.")
return builder.with_start_agent(triage).build()
FastAPI アプリを作成する
app = FastAPI(title="Workflow AG-UI Demo")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
ag_ui_workflow = AgentFrameworkWorkflow(
workflow_factory=lambda _thread_id: create_handoff_workflow(),
name="support_workflow",
description="Customer support handoff workflow.",
)
add_agent_framework_fastapi_endpoint(
app=app,
agent=ag_ui_workflow,
path="/support",
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8888)
イベント シーケンス
一般的なマルチターン操作では、次のようなイベントが生成されます。
RUN_STARTED threadId=abc123
STEP_STARTED stepName=triage_agent
TEXT_MESSAGE_START role=assistant
TEXT_MESSAGE_CONTENT delta="I'll look into your refund..."
TEXT_MESSAGE_END
STEP_FINISHED stepName=triage_agent
STEP_STARTED stepName=refund_agent
TOOL_CALL_START toolCallName=lookup_order_details
TOOL_CALL_ARGS delta='{"order_id":"12345"}'
TOOL_CALL_END
TOOL_CALL_START toolCallName=submit_refund
TOOL_CALL_ARGS delta='{"order_id":"12345","amount":"$129.99",...}'
TOOL_CALL_END
RUN_FINISHED interrupts=[{id: "...", value: {function_approval_request}}]
その後、クライアントは承認ダイアログを表示し、ユーザーの決定と共に再開できます。