AG-UI を含むワークフロー

MAF .NETは、ワークフローをAIAgentに変換し、他のエージェントと同様にマッピングすることで、AG-UI を通じてワークフローを公開できます。

AIAgent workflowAgent = AgentWorkflowBuilder
    .BuildSequential(researcher, reporter)
    .AsAIAgent();

app.MapAGUIServer("/", workflowAgent);

エンドポイントは、構成エージェントの標準テキストとツール呼び出しの出力をストリーミングします。 AuthorName は、各更新プログラムを生成したエージェントを識別します。

MAF .NETは現在、ワークフロー固有のライフサイクル動作を AG-UI にマップしていません。クライアントは、Python統合と同等のワークフローステップイベント、アクティビティスナップショット、ワークフロー割り込み、ワークフロー再開操作を受け取りません。ワークフローを AIAgent としてラップしても、これらのマッピングは追加されません。

現在の.NET追跡状態については、microsoft/agent-framework#2494 を参照してください。 AG UI に依存しないワークフローの構築と実行については、 MAF ワークフローの概念を参照してください。

次のステップ

運用環境とセキュリティに関する考慮事項を確認する

このチュートリアルでは、AG-UI エンドポイントを介して Agent Framework ワークフローを公開する方法について説明します。ワークフローは、定義された実行グラフ内の複数のエージェントとツールを調整し、AG-UI 統合により、ステップ追跡、アクティビティスナップショット、割り込み、カスタムイベントなどの豊富なワークフローイベントが Web クライアントにリアルタイムでストリーミングされます。

前提条件

始める前に、以下のことを確認してください:

Python 3.10 以降
agent-framework-ag-uiとagent-framework-foundryがインストール済み
作業の開始に関するチュートリアルに関する知識
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 イベントのセットが生成されます。

Event	放出時	説明
`RUN_STARTED`	実行の開始	ワークフロー実行の開始をマークします
`STEP_STARTED`	エグゼキューターまたはスーパーステップが開始する	`step_name` はエージェントまたはステップを識別します (例: `"triage_agent"`)
`TEXT_MESSAGE_*`	エージェントがテキストを生成する	標準のストリーミングテキストイベント
`TOOL_CALL_*`	エージェントがツールを呼び出す	標準ツール呼び出しイベント
`REASONING_*`	ワークフローは、`intermediate_output_from` で設定されたエグゼキューターからテキストを出力する	中間テキストを推論ブロックとしてストリームします。非推奨の `"data"` イベントエイリアスは、同じパスに従います。
`STEP_FINISHED`	スーパーステップまたはエグゼキューターが完了する	UI 進行状況の追跡の手順を閉じます。
`CUSTOM` (`status`)	ワークフローの状態の変更	イベント値に `{"state": "<value>"}` が含まれています
`CUSTOM` (`request_info`)	ワークフローが人間の入力を要求する	クライアントがプロンプトを表示するための要求ペイロードが含まれています
`CUSTOM` (`workflow_output`)	ワークフロー出力をメッセージコンテンツに変換できない	カスタムクライアントレンダリング用のシリアル化された出力が含まれます。
`RUN_FINISHED`	実行の完了	ワークフローが入力を待機している場合の `outcome.type == "interrupt"` と `outcome.interrupts` が含まれます

クライアントは、 STEP_STARTED / STEP_FINISHED イベントを使用して、現在アクティブなエージェントを示す進行状況インジケーターを表示できます。統合により、ターミナルイベントまたは人間入力要求の前にオープンな推論とテキストブロックが閉じられ、クライアントは完全なイベントシーケンスを受け取ります。

Python ワークフローが失敗した場合、RUN_ERRORは汎用パブリックメッセージ Workflow execution failed.とエラーコードを使用します。同様に、 executor_failed イベントは、汎用メッセージとエラーの種類を公開します。内部例外の詳細とトレースバックは、サーバーログに残ります。

中断して再開

ワークフローでは、実行を一時停止して、人間の入力またはツールの承認を収集できます。 AG-UI 統合では、割り込み/再開プロトコルを使用してこれを処理します。

割り込みの動作原理

実行中に、ワークフローは保留中の要求 (詳細を要求する HandoffAgentUserRequest や、 approval_mode="always_require"を含むツールなど) を発生させます。
AG-UI ブリッジは、要求データを含むCUSTOMを含むname="request_info" イベントを出力します。

RUN_FINISHED フィールドに保留中の要求が含まれるoutcome.interrupts イベントで実行が完了します。

{
  "type": "RUN_FINISHED",
  "threadId": "abc123",
  "runId": "run_xyz",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "request-id-1",
        "reason": "input_required",
        "message": "Provide the requested information.",
        "responseSchema": { "type": "string" },
        "metadata": {
          "agent_framework": {
            "request_type": "HandoffAgentUserRequest"
          }
        }
      }
    ]
  }
}

クライアントは、ユーザーが応答するための UI (テキスト入力、承認ボタンなど) をレンダリングします。

履歴書のしくみ

クライアントは、正規の resume 配列を使用して新しい要求を送信します。各エントリは割り込みを識別し、ユーザーの応答を提供します。

{
  "threadId": "abc123",
  "messages": [],
  "resume": [
    {
      "interruptId": "request-id-1",
      "status": "resolved",
      "payload": "User's response text or approval decision"
    }
  ]
}

サーバーは、再開ペイロードをワークフロー応答に変換し、一時停止した場所から実行を続行します。代わりに中断された実行をキャンセルするには、status を "cancelled" に設定し、payload を省略します。

ワークフローチェックポイントを永続化および再開する

AgentFrameworkWorkflowのcheckpoint_storageを構成して、各スーパーステップの最後に基になるワークフローの状態を保存します。代わりに、ワークフローを登録するときに同じ引数を add_agent_framework_fastapi_endpoint に渡すことができます。基になるワークフローがチェックポイントストレージを使用して構築された場合、アダプターはそのビルダーまたはランタイムストレージを直接使用できるため、ラッパーまたはエンドポイントで構成を複製する必要はありません。

次の例では、有効期間の短いワークフローにメモリ内ストレージを使用します。

from agent_framework import InMemoryCheckpointStorage
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)
from fastapi import FastAPI

app = FastAPI()
checkpoint_storage = InMemoryCheckpointStorage()
workflow = build_my_workflow()

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    checkpoint_storage=checkpoint_storage,
)
add_agent_framework_fastapi_endpoint(
    app,
    ag_ui_workflow,
    "/workflow",
)

実行が一時停止し、一時停止チェックポイントが使用可能な場合、 RUN_FINISHED イベント内の各割り込みには、 metadata.agent_framework.checkpoint_idのチェックポイント ID が含まれます。出力された値を使用して、別の最新チェックポイント参照なしでアプリケーションインスタンス間で完全な一時停止を再開します。

AgentFrameworkWorkflow.run()は AG-UI 要求ペイロードを受け取るので、クライアントは、Python checkpoint_id引数の代わりに転送されたプロパティを介してチェックポイント ID を提供します。チェックポイントのみの再開には、新しいユーザーメッセージは含まれません。

{
  "threadId": "abc123",
  "messages": [],
  "forwardedProps": {
    "checkpointId": "checkpoint-id-from-interrupt-metadata"
  }
}

アダプターは、保存されたワークフローの状態を復元し、実行を続行します。チェックポイントに保留中の割り込みが含まれている場合は、チェックポイント ID と正規 resume ペイロードの両方を同じ要求に含めます。アダプターは、割り込み応答を配信する前にチェックポイントを復元します。 forwardedProps.checkpointIdとしてmetadata.agent_framework.checkpoint_idの値を使用します。

アダプターは、各新しいチェックポイントを要求のスナップショットスコープとクライアント提供の threadIdにバインドします。どちらの値も一致しない場合、再開要求は拒否されます。所有権メタデータが導入される前に書き込まれたチェックポイントは、互換性のために再開できます。

このチェックでは、エンドポイントの承認または保護されたチェックポイントストレージは置き換えられません。詳細については、「セキュリティに関する考慮事項を参照してください。

InMemoryCheckpointStorage は、プロセスの再起動後も存続しません。永続的なストレージオプションとチェックポイントの選択については、「チェックポイント」を参照してください。

ワークフローチェックポイントと AG-UI スレッドスナップショット

ワークフローチェックポイントと AG-UI スレッドスナップショットは、次の異なるデータを保持します。

永続化メカニズム	ストア	Purpose
Agent Framework ワークフローチェックポイント	Executor とランタイムの状態 (保留中の要求を含む)	保存されたランタイム状態からワークフロー実行を再開する
AG-UI スレッドスナップショット	再生可能なプロトコル出力 (メッセージ、共有状態、最新の割り込みなど)	クライアントに表示されるスレッドをリハイドレートする

両方のメカニズムを構成できます。ワークフローチェックポイントは AG-UI スレッドスナップショットを置き換えません。また、AG-UI スレッドスナップショットには、ワークフローの実行を再開するために必要な Executor 状態が含まれません。

完全な例: マルチエージェントハンドオフワークフロー

この例では、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.foundry import FoundryChatClient
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 = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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          outcome={type: "interrupt", interrupts: [{id: "...", reason: "tool_call"}]}

その後、クライアントは承認ダイアログを表示し、ユーザーの決定と共に再開できます。

転送された小道具の受信

AG-UI クライアント (CopilotKit など) は、入力ペイロードに forwarded_props (または forwardedProps) フィールドを含めることができます。 AG-UI 統合では、run キーワード引数を使用して、これらのプロパティがワークフローのfunction_invocation_kwargs メソッドに自動的に渡されます。

class MyWorkflow(Workflow):
    async def run(
        self,
        *,
        message=None,
        responses=None,
        stream: bool = False,
        function_invocation_kwargs: dict | None = None,
    ):
        forwarded_props = (function_invocation_kwargs or {}).get("forwarded_props", {})
        # Use forwarded_props for custom routing, feature flags, etc.
        ...

重要な詳細:

forwarded_propsとforwardedPropsの両方が入力ペイロードで受け入れられます。内部的には、forwarded_propsに正規化されます。
転送されたプロパティ内では、 checkpoint_id と checkpointId はワークフローチェックポイントの再開用に予約されます。
workflow.run()がfunction_invocation_kwargs (または**kwargs) を受け入れない場合、props は自動的に削除されます。既存のワークフローは影響を受けません。
転送されたプロパティもセッションメタデータに格納されますが、LLM バインドメタデータからフィルター処理されるため、チャットクライアント要求にリークすることはありません。

次のステップ

ヒューマン・イン・ザ・ループ

その他のリソース

Go では、 workflow.Workflow をエージェントとして workflow/agentworkflow でラップし、そのエージェントを provider/aguiprovider でホストすることで、ワークフローを AG-UI に公開できます。

workflowAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
    IncludeOutputsInResponse: true,
    Config: agent.Config{
        Name: "WorkflowAgent",
    },
})
if err != nil {
    panic(err)
}

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(workflowAgent, aguiprovider.HandlerConfig{}))

Tip

完全な実行可能な例については、エージェントサンプルとしてのワークフローと AG-UI サーバーのサンプルを参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-09-18

AG-UI を含むワークフロー

次のステップ

前提条件

AG-UI でワークフローを使用する場合

AgentFrameworkWorkflow を使用したワークフローのラッピング

直接インスタンス

スレッド スコープ ファクトリ

エンドポイントの登録

ワークフローによって発生するAG-UIイベント

中断して再開

割り込みの動作原理

履歴書のしくみ

ワークフロー チェックポイントを永続化および再開する

ワークフロー チェックポイントと AG-UI スレッド スナップショット

完全な例: マルチエージェントハンドオフワークフロー

エージェントとツールを定義する

ワークフローを構築する

FastAPI アプリを作成する

イベント シーケンス

転送された小道具の受信

次のステップ

その他のリソース

フィードバック

その他のリソース

スレッドスコープファクトリ

ワークフローチェックポイントを永続化および再開する

ワークフローチェックポイントと AG-UI スレッドスナップショット

イベントシーケンス