Pracovní postupy s AG-UI

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

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.

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

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

2. Most AG-UI vygeneruje CUSTOM událost obsahující name="request_info" data požadavku.

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

4. 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

Další zdroje

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

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