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 Zainstalowano
• 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 interrupts pole zawiera listę oczekujących obiektów zapytań:

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

4. Klient renderuje interfejs użytkownika na potrzeby odpowiadania (wprowadzanie tekstu, przycisk zatwierdzenia itp.).

Jak działa wznawianie

Klient wysyła nowe żądanie z ładunkiem resume zawierającym odpowiedzi użytkownika kluczowane według identyfikatora przerwania.

{
  "threadId": "abc123",
  "messages": [],
  "resume": {
    "interrupts": [
      {
        "id": "request-id-1",
        "value": "User's response text or approval decision"
      }
    ]
  }
}

Serwer konwertuje ładunek życiorysu na odpowiedzi przepływu pracy i kontynuuje wykonywanie z miejsca wstrzymania.

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

Tworzenie przepływu pracy

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

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

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

Następne kroki

Dodatkowe zasoby

• Omówienie AG-UI
• Pierwsze kroki
• Przepływy pracy platformy agentów
• Agent Framework GitHub Repository