Pracovní postupy s AG-UI

Poznámka:

Podpora pracovních postupů pro integraci .NET AG-UI bude brzy k dispozici.

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 je nainstalováno
Znalost kurzu Začínáme
Základní znalost pracovních postupů rozhraní Agent 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.",
)

Důležité

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.

Zvláštní 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.	Obsahuje konečná nebo zprostředkující výstupní data.
`RUN_FINISHED`	Běh dokončen	Může zahrnovat `interrupts` , pokud 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 dokončí událostí RUN_FINISHED, jejíž pole interrupts obsahuje seznam nevyřízených žádostí:

{
  "type": "RUN_FINISHED",
  "threadId": "abc123",
  "runId": "run_xyz",
  "interrupts": [
    {
      "id": "request-id-1",
      "value": { "request_type": "HandoffAgentUserRequest", "data": "..." }
    }
  ]
}

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 resume datovou částí, která obsahuje odpovědi uživatele, indexované podle ID přerušení.

{
  "threadId": "abc123",
  "messages": [],
  "resume": {
    "interrupts": [
      {
        "id": "request-id-1",
        "value": "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.

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

Sestavení pracovního postupu

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

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          interrupts=[{id: "...", value: {function_approval_request}}]

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

Přehled AG-UI
Začínáme
Pracovní postupy architektury agentů
Jedno úložiště GitHub pro Agent Framework

Váš názor

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

Last updated on 2026-05-11