AG-UI 工作流程

MAF .NET 可以透過 AG-UI 公開工作流程，方法是將工作流程轉換為 AIAgent，並像對應其他代理程式一樣對其進行映射：

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

app.MapAGUIServer("/", workflowAgent);

該端點會串流傳送各組成代理的標準文字輸出與工具呼叫輸出。 AuthorName 識別產生每次更新的代理程式。

MAF .NET 目前並不會將工作流程特定的生命週期行為映射到 AG-UI。 客戶端不會收到與 Python 整合相當的工作流程步驟事件、活動快照、工作流程中斷或工作流程恢復操作。 將工作流程包裝成 a AIAgent 並不會新增那些映射。

關於目前 .NET 的追蹤狀態，請參見 microsoft/agent-framework#2494。 關於獨立於 AG-UI 的工作流程建構與執行，請參見 MAF 工作流程概念。

下一步

這個教學會教你如何透過 AG-UI 端點暴露 Agent Framework 的工作流程。 工作流程在定義的執行圖中協調多個代理與工具，AG-UI 整合則即時將豐富的工作流程事件——步驟追蹤、活動快照、中斷及自訂事件——串流至網頁客戶端。

Prerequisites

在開始之前，請確保您擁有：

• Python 3.10 或更新版本
• agent-framework-ag-ui 和 agent-framework-foundry 已安裝
• 熟悉使用者入門教學課程
• 對代理框架工作流程概念的基本理解

在何種情況下使用 AG-UI 工作流程

當你需要時，使用工作流程而非單一代理：

• 多代理協調：將任務路由至專業代理間（例如，分類 → 退還 → 訂單）
• 結構化執行步驟：透過事件追蹤定義階段 STEP_STARTED / STEP_FINISHED 的進度
• 中斷/恢復流程：暫停執行以收集人工輸入或批准，然後繼續執行
• 自訂事件串流：向用戶端發出網域專屬事件 （request_info， status， workflow_output）

使用 AgentFrameworkWorkflow 包裝工作流程

AgentFrameworkWorkflow 是一個輕量化包裝器，能適應原生 Workflow 協定以符合 AG-UI 協定。 你可以提供預先建置的工作流程實例，或是使用工廠類別，讓每個執行緒建立其自己的工作流程。

直接實例

當單一工作流程物件能安全服務所有請求時，使用直接實例（例如無狀態管線）：

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.",
)

註冊端點

使用 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 事件集合：

用戶端可利用 STEP_STARTED / STEP_FINISHED 事件呈現進度指示器，顯示目前哪位代理在活動中。 該整合會在終端事件或人工輸入請求前關閉開放推理與文字區塊，讓用戶端收到完整的事件序列。

當 Python 工作流程失敗時，RUN_ERROR使用通用的公開訊息Workflow execution failed.加上錯誤代碼。 executor_failed事件同樣會暴露通用訊息與錯誤類型。 內部例外細節與追蹤資料仍保留在伺服器日誌中。

中斷與恢復

工作流程可以暫停執行以收集人工輸入或工具審核。 AG-UI 整合透過中斷/恢復協定處理此問題。

中斷機制運作方式

1. 執行過程中，工作流程會引發一個擱置要求 (例如，HandoffAgentUserRequest 要求更多詳細資料，或具有 approval_mode="always_require" 的工具)。

2. AG-UI 橋接器會觸發包含 CUSTOM 請求資料的 name="request_info" 事件。

3. 執行會以一個 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"
             }
           }
         }
       ]
     }
   }
   

4. 用戶端呈現用戶介面，讓使用者進行回應（如文字輸入、批准按鈕等）。

履歷的運作方式

用戶端會發送一個包含標準 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 請求酬載，因此用戶端會透過轉發屬性提供檢查點 ID，而不是透過 Python checkpoint_id 引數。 僅限檢查點的繼續不包含新的使用者訊息：

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

介面卡會恢復儲存的工作流程狀態並繼續執行。 如果檢查點包含待處理中斷，請在同一請求中同時包含檢查點 ID 與標準 resume 有效載荷。 配接器會在傳遞中斷回應之前還原檢查點。 使用該 metadata.agent_framework.checkpoint_id 值為 forwardedProps.checkpointId。

配接器會將每個新的檢查點繫結至請求的快照範圍以及用戶端提供的 threadId。 當任一值不符時，它會拒絕履歷請求。 在引入所有權元資料之前撰寫的檢查點仍可重複使用以促進相容性。

此檢查不會取代端點授權或受保護檢查點儲存。 如需詳細資訊，請參閱 安全性考慮。

InMemoryCheckpointStorage 無法存活，程序重啟。 關於耐用儲存選項及檢查點選擇，請參見檢查點。

工作流程檢查點與 AG-UI 執行緒快照

工作流程檢查點與 AG-UI 執行緒快照會保存不同的資料：

你可以同時設定兩種機制。 工作流程檢查點不會取代 AG-UI 執行緒快照，AG-UI 執行緒快照也不包含執行者狀態以恢復工作流程執行。

完整範例：多代理切換工作流程

此範例展示了一個由三位客服人員組成的客戶支援工作流程，他們彼此交接工作，使用需要核准的工具，並在需要時請求人工輸入。

定義代理人與工具

"""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 整合會透過 function_invocation_kwargs 關鍵字引數自動將這些屬性傳遞給工作流程的 run 方法：

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），則道具會靜默中移除——現有工作流程不受影響。
• 轉送的屬性也會存放在工作階段中繼資料中，但會被 LLM 綁定的中繼資料過濾，避免外洩到聊天用戶端要求中。

下一步

其他資源

• AG-UI 概覽
• 入門
• 工作流程
• Agent Framework GitHub 存放庫

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{}))