Przepływy pracy z AG-UI

Usługa MAF .NET może uwidaczniać przepływ pracy za pośrednictwem AG-UI, konwertując przepływ pracy na obiekt AIAgent i mapując go tak jak każdy inny agent:

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

app.MapAGUIServer("/", workflowAgent);

Punkt końcowy przesyła strumieniowo standardowy tekst agentów składowych i dane wyjściowe wywołania narzędzia. AuthorName identyfikuje agenta, który wyprodukował każdą aktualizację.

MAF .NET obecnie nie odwzorowuje w AG-UI zachowań cyklu życia specyficznych dla przepływu pracy. Klienci nie otrzymują zdarzeń kroków przepływu pracy, migawek działań, przerwań przepływu pracy ani operacji wznawiania przepływu pracy odpowiadających integracji Python. Opakowanie przepływu pracy jako AIAgent nie dodaje tych mapowań.

Aby uzyskać bieżący stan śledzenia .NET, zobacz microsoft/agent-framework#2494. Informacje o tworzeniu i uruchamianiu przepływów pracy niezależnie od AG-UI można znaleźć w sekcji Podstawowe pojęcia dotyczące przepływów pracy w MAF.

Następne kroki

W tym samouczku pokazano, jak uwidaczniać przepływy pracy platformy Agent Framework za pośrednictwem punktu końcowego AG-UI. Przepływy pracy organizują wielu agentów i narzędzia w zdefiniowanym grafie wykonywania, a integracja AG-UI strumieniuje bogate zdarzenia przepływu pracy — śledzenie kroków, migawki działań, przerwania i zdarzenia niestandardowe — do klientów internetowych w czasie rzeczywistym.

Wymagania wstępne

Zanim zaczniesz, upewnij się, że masz:

• Python 3.10 lub nowszy
• agent-framework-ag-ui i agent-framework-foundry zainstalowane
• Znajomość samouczka Wprowadzenie
• Podstawowa wiedza na temat pojęć związanych z przepływem pracy platformy Agent Framework

Kiedy należy używać przepływów pracy z AG-UI

Jeśli potrzebujesz, użyj przepływu pracy zamiast jednego agenta:

• Orkiestracja wielu agentów: kierowanie zadań między wyspecjalizowanymi agentami (na przykład sortowanie → zwrot → zamówienie)
• Strukturalne kroki wykonawcze: Śledź postęp poprzez zdefiniowane etapy ze zdarzeniami STEP_STARTED / STEP_FINISHED
• Przerywanie/wznawianie przepływów: wstrzymywanie wykonywania w celu zbierania danych wejściowych lub zatwierdzeń przez człowieka, a następnie wznawiania
• Przesyłanie strumieniowe zdarzeń niestandardowych: emitowanie zdarzeń specyficznych dla domeny (request_info, status, workflow_output) do klienta

Zawijanie przepływu pracy z użyciem AgentFrameworkWorkflow

AgentFrameworkWorkflow to lekkie opakowanie, które dostosowuje natywny Workflow do protokołu AG-UI. Możesz podać wstępnie utworzone wystąpienie przepływu pracy lub fabrykę, która tworzy nowy przepływ pracy dla każdego wątku.

Bezpośrednia instancja

Użyj instancji bezpośredniej, gdy pojedynczy obiekt w przepływie pracy może bezpiecznie obsługiwać wszystkie żądania (na przykład potoki bezstanowe):

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

Fabryka o zakresie wątków

Użyj workflow_factory , gdy każdy wątek konwersacji potrzebuje własnego stanu przepływu pracy. Fabryka otrzymuje element thread_id i zwraca nowy Workflowelement :

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

Rejestrowanie punktu końcowego

Zarejestruj przepływ pracy add_agent_framework_fastapi_endpoint w taki sam sposób, jak w przypadku zarejestrowania pojedynczego 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",
)

Możesz również przekazać sam Workflow bezpośrednio — punkt końcowy automatycznie opakowuje go w AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

Zdarzenia AG-UI emitowane przez workflowy

Przebiegi przepływu pracy emitują bogatszy zestaw zdarzeń AG-UI w porównaniu z przebiegami pojedynczego agenta:

Klienci mogą używać zdarzeń STEP_STARTED / STEP_FINISHED do renderowania wskaźników postępu pokazujących, który agent jest obecnie aktywny. Integracja zamyka otwarte bloki rozumowania i tekstu przed zdarzeniem końcowym lub żądaniem danych wejściowych od użytkownika, dzięki czemu klienci otrzymują pełną sekwencję zdarzeń.

Przerywanie i wznawianie

Przepływy pracy mogą wstrzymywać wykonywanie, aby zebrać wkład ludzki lub zatwierdzenia dotyczące narzędzi. Integracja AG-UI obsługuje to za pośrednictwem protokołu przerwania/wznawiania.

Jak działają przerwania

1. Podczas wykonywania przepływu pracy zgłaszane jest oczekujące żądanie (na przykład HandoffAgentUserRequest z prośbą o więcej szczegółów lub narzędzie z approval_mode="always_require").

2. Mostek AG-UI emituje zdarzenie CUSTOM zawierające name="request_info" dane żądania.

3. Przebieg kończy się zdarzeniem RUN_FINISHED , którego outcome.interrupts pole zawiera oczekujące żądania:

   {
     "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 renderuje interfejs użytkownika na potrzeby odpowiadania (wprowadzanie tekstu, przycisk zatwierdzenia itp.).

Jak działa wznawianie

Klient wysyła nowe żądanie z kanoniczną tablicą resume. Każdy wpis identyfikuje przerwanie i dostarcza odpowiedź użytkownika:

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

Serwer konwertuje ładunek życiorysu na odpowiedzi przepływu pracy i kontynuuje wykonywanie z miejsca wstrzymania. Aby zamiast tego anulować przerwane uruchomienie, ustaw status na "cancelled" i pomiń payload.

Utrwalanie i wznawianie punktów kontrolnych przepływu pracy

Skonfiguruj checkpoint_storage w AgentFrameworkWorkflow, aby zapisywać bazowy stan przepływu pracy na końcu każdego superkroku. Zamiast tego możesz przekazać ten sam argument do add_agent_framework_fastapi_endpoint podczas rejestrowania przepływu pracy. Pamięć masowa musi być dostępna dla nakładki AG-UI lub punktu końcowego, aby wznowić punkt kontrolny za pośrednictwem AG-UI.

W poniższym przykładzie użyto pamięci operacyjnej na potrzeby krótkotrwałego przepływu pracy:

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

AgentFrameworkWorkflow.run() odbiera treść żądania AG-UI, więc klient przekazuje identyfikator punktu kontrolnego za pośrednictwem przekazywanych właściwości zamiast argumentu Python checkpoint_id. Wznowienie wyłącznie z punktu kontrolnego nie obejmuje nowej wiadomości od użytkownika:

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

Adapter przywraca zapisany stan przepływu pracy i wznawia wykonywanie. Jeśli punkt kontrolny zawiera oczekujące przerwanie, dołącz do tego samego żądania zarówno identyfikator punktu kontrolnego, jak i kanoniczny ładunek resume. Adapter przywraca stan z punktu kontrolnego, zanim przekaże odpowiedź na przerwanie.

Adapter wiąże każdy nowy punkt kontrolny z zakresem „Snapshot Scope” żądania oraz elementem threadId dostarczonym przez klienta. Odrzuca żądanie wznowienia, gdy którakolwiek wartość nie jest zgodna. Punkty kontrolne zapisywane przed wprowadzeniem metadanych własności pozostają wznawiane pod kątem zgodności.

Ta kontrola nie zastępuje autoryzacji punktu końcowego ani chronionego magazynu punktów kontrolnych. Aby uzyskać więcej informacji, zobacz Zagadnienia dotyczące zabezpieczeń.

InMemoryCheckpointStorage nie przetrwa ponownego uruchomienia procesu. Aby uzyskać opcje trwałego magazynu i wybór punktu kontrolnego, zobacz Punkty kontrolne.

Punkty kontrolne przepływu pracy i migawki stanu wątków AG-UI

Punkty kontrolne przepływu pracy i migawki wątków AG-UI utrwalają różne dane:

Oba mechanizmy można skonfigurować. Punkt kontrolny workflow nie zastępuje migawki wątku AG-UI, a migawka wątku AG-UI nie zawiera stanu egzekutora wymaganego do wznowienia wykonywania workflow.

Przykład Kompletny: Przepływ pracy dotyczący przekazywania zadań między agentami

W tym przykładzie przedstawiono przepływ pracy z udziałem trzech agentów, którzy przekazują sobie nawzajem pracę, używają narzędzi wymagających zatwierdzenia i proszą o wkład ludzki w razie potrzeby.

Definiowanie agentów i narzędzi

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

Tworzenie przepływu pracy

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

Tworzenie aplikacji 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)

Sekwencja zdarzeń

Typowa interakcja obejmująca wiele tur generuje zdarzenia, takie jak:

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

Następnie aplikacja może wyświetlić okno dialogowe zatwierdzenia i kontynuować zgodnie z decyzją użytkownika.

Odbieranie przesłanych dalej rekwizytów

Klienci AG-UI (tacy jak CopilotKit) mogą zawierać pole forwarded_props (lub forwardedProps) w ładunku wejściowym. Integracja AG-UI automatycznie przekazuje te rekwizyty do metody przepływu pracy run za pośrednictwem argumentu kluczowego 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.
        ...

Kluczowe szczegóły:

• Zarówno forwarded_props , jak i forwardedProps są akceptowane w ładunku wejściowym; wewnętrznie są znormalizowane do forwarded_props.
• Wśród przekazywanych właściwości checkpoint_id i checkpointId są zarezerwowane do wznowienia z punktu kontrolnego przepływu pracy.
• Jeśli workflow.run() nie akceptuje function_invocation_kwargs (lub **kwargs), rekwizyty są dyskretnie porzucane — istniejące przepływy pracy nie mają wpływu.
• Przekazywane rekwizyty są również przechowywane w metadanych sesji, ale są filtrowane z metadanych powiązanych z usługą LLM, dzięki czemu nie wyciekają do żądań klientów czatu.

Następne kroki

Dodatkowe zasoby

• Omówienie AG-UI
• Pierwsze kroki
• Workflows
• Repozytorium GitHub Platformy agentów

Go może udostępniać przepływy pracy w AG-UI, opakowując workflow.Workflow jako agenta przy użyciu workflow/agentworkflow, a następnie hostując tego agenta za 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{}))