Munkafolyamatok AG-UI-val

A MAF .NET az AG-UI-n keresztül elérhetővé tehet egy munkafolyamatot úgy, hogy azt AIAgent alakítja, és ugyanúgy képezi le, mint bármely más ügynököt:

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

app.MapAGUIServer("/", workflowAgent);

A végpont adatfolyamként továbbítja az összetevő ügynökök normál szövegét és eszközhívási kimenetét. AuthorName azonosítja az egyes frissítéseket előállító ügynököt.

A MAF .NET jelenleg nem képez le munkafolyamat-specifikus életciklus-viselkedést AG-UI-ra. Az ügyfelek nem kapnak munkafolyamat-lépéseseményeket, tevékenység-pillanatképeket, munkafolyamat-megszakításokat vagy munkafolyamat-folytatási műveleteket az Python integrációval megegyezően. Ha egy munkafolyamatot AIAgent elembe csomagol, az nem adja hozzá ezeket a hozzárendeléseket.

Az aktuális .NET nyomon követés állapotáról a microsoft/agent-framework#2494 című cikkben olvashat. A munkafolyamatok AG-UI-tól független felépítéséről és végrehajtásáról a MAF munkafolyamat-fogalmaiban olvashat.

Következő lépések

Éles és biztonsági szempontok áttekintése

Ez az oktatóanyag bemutatja, hogyan teheti elérhetővé az Ügynök-keretrendszer munkafolyamatait egy AG-UI végponton keresztül. A munkafolyamatok több ügynököt és eszközt vezényelnek egy meghatározott végrehajtási gráfban, és az AG-UI integrációs folyamat gazdag munkafolyamat-eseményeket – lépéskövetést, tevékenység-pillanatképeket, megszakításokat és egyéni eseményeket – továbbít a webes ügyfeleknek valós időben.

Prerequisites

Mielőtt hozzákezdene, győződjön meg arról, hogy:

• Python 3.10 vagy újabb verzió
• agent-framework-ag-ui és agent-framework-foundry telepítve
• Az első lépések oktatóanyagának ismerete
• Az Agent Framework munkafolyamat-fogalmainak alapszintű ismerete

Mikor érdemes munkafolyamatokat használni AG-UI

Ha szüksége van rá, használjon munkafolyamatot egyetlen ügynök helyett:

• Több ügynök vezénylése: Feladatok irányítása specializált ügynökök között (például triázs → visszatérítés → megrendelés)
• Strukturált végrehajtási lépések: Az előrehaladás nyomon követése meghatározott szakaszokon keresztül események használatával STEP_STARTED / STEP_FINISHED
• Folyamatok megszakítása/folytatása: A végrehajtás felfüggesztése emberi bemenetek vagy jóváhagyások gyűjtéséhez, majd folytatás
• Egyéni eseménystreamelés: Tartományspecifikus események (request_info, status, workflow_output) küldése az ügyfélnek

Munkafolyamat burkolása az AgentFrameworkWorkflow használatával

AgentFrameworkWorkflow egy egyszerű burkoló, amely a natív Workflow AG-UI protokollhoz igazodik. Megadhat egy előre elkészített munkafolyamat-példányt vagy egy olyan gyárat, amely szálonként új munkafolyamatot hoz létre.

Közvetlen példány

Használjon közvetlen példányt, ha egyetlen munkafolyamat-objektum biztonságosan kiszolgálhatja az összes kérést (például állapot nélküli folyamatokat):

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

Szál hatókörű gyár

Akkor használható workflow_factory , ha minden beszélgetési szálnak saját munkafolyamat-állapotra van szüksége. A gyár megkapja a thread_id-t, és visszaad egy új Workflow-t.

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

Vagyworkflow eztvagyworkflow_factory azt kell teljesíteni, de nem mindkettőt. A burkoló hibát jelez, ValueError ha mindkettő meg van adva.

A végpont regisztrálása

A munkafolyamatot add_agent_framework_fastapi_endpoint ugyanúgy regisztrálja, mint egy ügynököt:

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

Közvetlenül átadhat egy üres Workflow is — a végpont automatikusan becsomagolja a AgentFrameworkWorkflow-ba.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI munkafolyamatok által kibocsátott események

A munkafolyamat-futtatások AG-UI események gazdagabb készletét bocsátják ki az együgynök-futtatásokhoz képest:

Esemény	Amikor kibocsátják	Description
RUN_STARTED	A futtatás elkezdődött	A munkafolyamat-végrehajtás kezdetének megjelölése
STEP_STARTED	A végrehajtás vagy a szuperlépés elindul	step_name azonosítja az ügynököt vagy a lépést (például "triage_agent")
TEXT_MESSAGE_*	Az ügynök szöveget állít elő	Szabványos folyamatos adatátviteli szöveges események
TOOL_CALL_*	Az ügynök meghív egy eszközt	Standard eszközhívási események
STEP_FINISHED	A futtató vagy a superstep befejeződött	A felhasználói felület állapotkövetési lépésének bezárása
CUSTOM (status)	Munkafolyamat állapotának változásai	Tartalmazza {"state": "<value>"} az eseményértékben
CUSTOM (request_info)	A munkafolyamat emberi bemenetet kér	A kliens számára megjelenítendő prompt kérés hasznos adatait tartalmazza.
CUSTOM (workflow_output)	A munkafolyamat kimenetet hoz létre	Kiadva mind a "output" (terminál), mind a "intermediate" munkafolyamat-események esetén. A terminálkimenetek a végső választ hordozzák; A köztes kimenetek tartalomként text_reasoning jelennek meg, amikor a munkafolyamat mögötte as_agent()fut.
RUN_FINISHED	A futtatás befejeződik	Tartalmazza outcome.type == "interrupt" és outcome.interrupts amikor a munkafolyamat bemenetre vár

Az ügyfelek események használatával STEP_STARTED / STEP_FINISHED megjeleníthetik az aktuálisan aktív ügynök állapotát jelző állapotjelzőket.

Megszakítás és folytatás

A munkafolyamatok szüneteltethetik a végrehajtást, hogy emberi bemeneteket vagy eszközjóváhagyásokat gyűjtsenek. A AG-UI integráció ezt a megszakítási/folytatási protokollon keresztül kezeli.

A megszakítások működése

1. A végrehajtás során a munkafolyamat függőben lévő kérést hoz létre (például HandoffAgentUserRequest, amely további részleteket kér, vagy egy eszköz approval_mode="always_require"-vel).

2. A AG-UI híd a kérelemadatokat tartalmazó eseményt CUSTOMname="request_info" bocsát ki.

3. A futtatás egy RUN_FINISHED olyan eseménysel fejeződik be, amelynek outcome.interrupts mezője tartalmazza a függőben lévő kéréseket:

   {
     "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. Az ügyfél megjeleníti a felhasználói felületet, hogy a felhasználó válaszoljon (szöveges bemenet, jóváhagyási gomb stb.).

Az önéletrajz működése

Az ügyfél egy új kérést küld egy canonical resume array használatával. Minden bejegyzés azonosítja a megszakítást, és megadja a felhasználó válaszát:

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

A kiszolgáló munkafolyamat-válaszokká alakítja az önéletrajz hasznos adatait, és onnan folytatja a végrehajtást, ahol szüneteltette. Ha ehelyett meg szeretné szakítani a megszakított futtatást, állítsa a(z) status értékét erre: "cancelled", és hagyja el a(z) payload elemet.

Teljes példa: Multi-Agent átvételi munkafolyamata

Ez a példa egy ügyféltámogatási munkafolyamatot mutat be három ügynökkel, amelyek átadják egymásnak a munkát, jóváhagyást igénylő eszközöket használnak, és szükség esetén emberi bemenetet kérnek.

Az ügynökök és eszközök meghatározása

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

A munkafolyamat létrehozása

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

A FastAPI-alkalmazás létrehozása

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)

Eseménysorozat

Egy tipikus többfordulós interakció olyan eseményeket hoz létre, mint:

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

Az ügyfél ezután megjeleníthet egy jóváhagyási párbeszédpanelt, és folytathatja a felhasználó döntésével.

Továbbított kellékek fogadása

AG-UI ügyfelek (például a CopilotKit) tartalmazhatnak egy forwarded_props (vagy forwardedProps) mezőt a bemeneti adatcsomagban. Az AG-UI integráció automatikusan átadja ezeket a paramétereket a munkafolyamat run metódusának a function_invocation_kwargs kulcsszóargumentumon keresztül.

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

Főbb részletek:

• forwarded_props és forwardedProps egyaránt elfogadott a bemeneti adatban; belsőleg forwarded_props-re normalizálva lesznek.
• Ha workflow.run() nem fogadja el function_invocation_kwargs (vagy **kwargs), a tulajdonságok csendben elvetésre kerülnek, így a meglévő munkafolyamatok nem változnak.
• A továbbított tulajdonságokat a munkamenet metaadatai is tárolják, de kiszűrik őket az LLM-hez kötött metaadatok közül, így nem kerülnek be a csevegőügyfelek kéréseibe.

Következő lépések

Human-in-the-Loop

További források

• AG-UI áttekintése
• Első lépések
• Munkafolyamatok
• Agent Framework GitHub-adattár

A Go elérhetővé teheti a munkafolyamatokat az AG-UI számára úgy, hogy egy workflow.Workflow-t ügynökként becsomagol a workflow/agentworkflow segítségével, majd azt a provider/aguiprovider segítségével hosztolja.

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

A teljes futtatható példákért tekintse meg a munkafolyamatot ügynökmintaként és AG-UI kiszolgálómintát .