Pracovní postupy s AG-UI

Pracovní postupy orchestrují více agentů v definovaném prováděcím grafu. V .NET zpřístupňujete pracovní postup přes AG-UI přesně způsobem, jakým zveřejňujete libovolného agenta: převeďte ho na objekt s příponou AIAgentAsAIAgent() a namapujte ho na MapAGUIServer. Není k dispozici žádné rozhraní API serveru specifické pro pracovní postup.

using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.Agents.AI.Workflows;
using OpenAI.Chat;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddAGUIServer();

string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");

ChatClient chatClient = new AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName);

AIAgent researcher = chatClient.AsAIAgent(
    name: "researcher", instructions: "Research the user's topic and write a short, factual brief.");
AIAgent reporter = chatClient.AsAIAgent(
    name: "reporter", instructions: "Summarize the researcher's brief into a single clear paragraph.");

// A workflow-as-agent is just an AIAgent. Map it like any other agent.
AIAgent workflowAgent = AgentWorkflowBuilder.BuildSequential(researcher, reporter).AsAIAgent();

WebApplication app = builder.Build();
app.MapAGUIServer("/", workflowAgent);

await app.RunAsync();

Warning

DefaultAzureCredential je vhodný pro vývoj, ale vyžaduje pečlivé zvážení v produkčním prostředí. V produkčním prostředí zvažte použití konkrétních přihlašovacích údajů (např ManagedIdentityCredential. ) k zabránění problémům s latencí, neúmyslnému testování přihlašovacích údajů a potenciálním bezpečnostním rizikům z náhradních mechanismů.

Na straně klienta se nezmění. Spojte se AGUIChatClient jako v začínáme. Výstupní datové proudy každého agenta jako normální AG-UI události textu a volání nástrojů a AuthorName při každé aktualizaci zjistí, který agent v pracovním postupu ho vytvořil.

Note

Integrace .NET streamuje výstup agenta pracovního postupu (textová volání a volání nástrojů) přes uživatelské rozhraní skupiny dostupnosti, ale ne události specifické pro pracovní postup AG-UI zobrazené v Python verzi tohoto článku: sledování kroků (STEP_STARTED / STEP_FINISHED), snímky aktivit a přerušení na úrovni pracovního postupu. Tyto události se stále vyvíjejí pro .NET (sledované rozhraním agent-framework#2494).

Další kroky

V tomto kurzu se dozvíte, jak vystavit pracovní postupy agenta Framework prostřednictvím koncového bodu AG-UI. Pracovní postupy orchestrují v definovaném grafu spouštění několik agentů a nástrojů a AG-UI integrace streamuje bohaté události pracovního postupu – sledování kroků, snímky aktivit, přerušení a vlastní události – webovým klientům v reálném čase.

Předpoklady

Než začnete, ujistěte se, že máte:

Python 3.10 nebo novější
agent-framework-ag-ui a agent-framework-foundry nainstalované
Znalost kurzu Začínáme
Základní znalost konceptů pracovních postupů agenta Framework

Kdy použít pracovní postupy s AG-UI

Pokud potřebujete, použijte místo jednoho agenta pracovní postup:

Orchestrace s více agenty: Směrování úloh mezi specializovanými agenty (například třídění → refundace → objednávce)
Strukturované kroky provádění: Sledování průběhu prostřednictvím definovaných fází s událostmi STEP_STARTED / STEP_FINISHED
Přerušení / obnovení toků: Pozastavte provádění, abyste shromáždili lidské vstupy nebo schválení, a pak obnovili
Streamování vlastních událostí: Generování událostí specifických pro doménu (request_info, status, workflow_output) do klienta

Zabalení pracovního postupu pomocí AgentFrameworkWorkflow

AgentFrameworkWorkflow je jednoduchá obálka, která přizpůsobuje nativní Workflow protokol AG-UI. Můžete zadat předem vytvořenou instanci pracovního postupu nebo továrnu, která vytvoří nový pracovní postup pro každé vlákno.

Přímá instance

Použijte přímou instanci, když jeden objekt pracovního postupu může bezpečně obsluhovat všechny požadavky (například bezstavové kanály):

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

Továrna s oborem vláken

Používá se workflow_factory , když každé vlákno konverzace potřebuje svůj vlastní stav pracovního postupu. Továrna obdrží thread_id a vrátí novou 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

Musíte předat buďworkflowneboworkflow_factory, nikoliv obojí. Obálka vyvolá ValueError, pokud jsou poskytnuty obě.

Registrace koncového bodu

Zaregistrujte pracovní postup add_agent_framework_fastapi_endpoint stejným způsobem jako jednotlivého agenta:

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

Můžete také předat holý Workflow přímo – koncový bod ho automaticky zabalí do AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI události generované pracovními postupy

Spuštění pracovního postupu generuje bohatší sadu AG-UI událostí v porovnání se spuštěními s jedním agentem.

Událost	Při emitování	Description
`RUN_STARTED`	Běh začíná	Označí začátek provádění pracovního postupu.
`STEP_STARTED`	Začátek vykonavatele nebo superkroku	`step_name` identifikuje agenta nebo krok (například `"triage_agent"`)
`TEXT_MESSAGE_*`	Agent vytvoří text.	Standardní streamované textové události
`TOOL_CALL_*`	Agent vyvolá nástroj	Standardní události volání nástrojů
`STEP_FINISHED`	Exekutor nebo superkrok je dokončen	Zavře krok sledování průběhu uživatelského rozhraní.
`CUSTOM` (`status`)	Změny stavu pracovního postupu	Obsahuje hodnotu události `{"state": "<value>"}`.
`CUSTOM` (`request_info`)	Požadavky pracovního postupu na lidské vstupy	Obsahuje datovou část požadavku pro klienta k vykreslení výzvy.
`CUSTOM` (`workflow_output`)	Pracovní postup vytvoří výstup.	Generováno pro události `"output"` (terminálu) i `"intermediate"` pracovního postupu. Výstupy terminálu obsahují konečnou odpověď; mezivýstupy se při běhu workflow za `text_reasoning` zobrazují jako obsah `as_agent()`.
`RUN_FINISHED`	Běh dokončen	Zahrnuje `outcome.type == "interrupt"` a `outcome.interrupts` kdy pracovní postup čeká na vstup.

Klienti můžou pomocí STEP_STARTED / STEP_FINISHED událostí vykreslovat indikátory průběhu zobrazující, který agent je aktuálně aktivní.

Přerušení a pokračování

Pracovní postupy můžou pozastavit provádění, aby shromáždily lidské vstupy nebo schválení nástrojů. Integrace AG-UI to zpracovává prostřednictvím protokolu přerušení/obnovení.

Jak přerušení funguje

Během provádění pracovní postup vyvolá nevyřízenou žádost (například HandoffAgentUserRequest žádost o další podrobnosti nebo nástroj s approval_mode="always_require").
Most AG-UI vygeneruje CUSTOM událost obsahující name="request_info" data požadavku.

Běh se ukončí událostí RUN_FINISHED, jejíž pole outcome.interrupts obsahuje čekající požadavky:

{
  "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"
          }
        }
      }
    ]
  }
}

Klient vykreslí uživatelské rozhraní pro uživatele, aby odpověděl (textový vstup, tlačítko schválení atd.).

Jak funguje životopis

Klient odešle nový požadavek s kanonickým resume polem. Každá položka identifikuje přerušení a poskytuje odpověď uživatele:

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

Server převede datovou část životopisu na odpovědi pracovního postupu a pokračuje v provádění tam, kde se pozastavil. Chcete-li místo toho zrušit přerušené spuštění, nastavte status na "cancelled" a vynechte payload.

Kompletní příklad: Pracovní postup předávání mezi více agenty

Tento příklad ukazuje pracovní postup podpory zákazníků se třemi agenty, kteří vzájemně oddávají práci, používají nástroje vyžadující schválení a v případě potřeby vyžadují lidské vstupy.

Definování agentů a nástrojů

"""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",
    }

Sestavení pracovního postupu

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()

Vytvoření aplikace 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)

Posloupnost událostí

Typická vícekroková interakce produkuje události, jako například:

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"}]}

Klient pak může zobrazit dialogové okno schválení a pokračovat v rozhodování uživatele.

Příjem přeposlaných props

AG-UI klienti (například CopilotKit) můžou do vstupní části zahrnout pole forwarded_props (nebo forwardedProps). Integrace AG-UI automaticky předává tyto vlastnosti metodě pracovního postupu run prostřednictvím klíčového argumentu 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.
        ...

Klíčové podrobnosti:

Obě forwarded_props a forwardedProps jsou přijímány ve vstupní datové části; interně jsou normalizovány na forwarded_props.
Pokud workflow.run() neakceptuje function_invocation_kwargs (nebo **kwargs), jsou props tiše ignorovány – stávající pracovní postupy nejsou ovlivněné.
Předávané vlastnosti jsou také uloženy v metadatech relace, ale jsou filtrované z metadat vázaných na LLM, takže se nedostanou do požadavků chatu klienta.

Další kroky

Human-in-the-Loop

Další zdroje

Go může zpřístupnit pracovní postupy pro AG-UI zabalením workflow.Workflow jako agenta pomocí workflow/agentworkflow a následným hostováním tohoto agenta pomocí provider/aguiprovider.

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

Podívejte se na ukázku pracovního postupu agenta a ukázku serveru AG-UI pro kompletní spustitelné příklady.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-08-13

Pracovní postupy s AG-UI

Další kroky

Předpoklady

Kdy použít pracovní postupy s AG-UI

Zabalení pracovního postupu pomocí AgentFrameworkWorkflow

Přímá instance

Továrna s oborem vláken

Registrace koncového bodu

AG-UI události generované pracovními postupy

Přerušení a pokračování

Jak přerušení funguje

Jak funguje životopis

Kompletní příklad: Pracovní postup předávání mezi více agenty

Definování agentů a nástrojů

Sestavení pracovního postupu

Vytvoření aplikace FastAPI

Posloupnost událostí

Příjem přeposlaných props

Další kroky

Další zdroje

Váš názor

Další materiály