Przepływy pracy z AG-UI

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 przepływów 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.

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.

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.
• 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
• Przepływy pracy platformy agentów
• 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{}))