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

Usługa MAF .NET nie mapuje obecnie zachowania cyklu życia specyficznego dla przepływu pracy na interfejs użytkownika grupy dostępności. 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. Zawijanie przepływu pracy jako elementu AIAgent nie powoduje dodania tych mapowań.

Aby uzyskać bieżący stan śledzenia .NET, zobacz microsoft/agent-framework#2494. Aby zapoznać się z tworzeniem i wykonywaniem przepływu pracy niezależnie od interfejsu użytkownika grupy dostępności, zobacz Pojęcia dotyczące przepływu pracy narzędzia MAF.

Następne kroki

Przegląd zagadnień dotyczących środowiska produkcyjnego i zabezpieczeń

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

Ważna

Musisz przekazać alboworkflowalboworkflow_factory, nie oba. Otoka zgłasza wartość ValueError , jeśli podano oba te elementy.

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:

Zdarzenie	Po emisji	Description
`RUN_STARTED`	Uruchamianie rozpoczyna się	Oznacza początek wykonywania przepływu pracy
`STEP_STARTED`	Rozpoczyna się funkcja wykonawcza lub superkrok	`step_name` identyfikuje agenta lub krok (na przykład `"triage_agent"`)
`TEXT_MESSAGE_*`	Agent tworzy tekst	Standardowe zdarzenia tekstowe przesyłania strumieniowego
`TOOL_CALL_*`	Agent wywołuje narzędzie	Zdarzenia wywołań standardowych narzędzi
`STEP_FINISHED`	Zakończenie operacji wykonawcy lub superkroku	Zamyka krok śledzenia postępu interfejsu użytkownika
`CUSTOM` (`status`)	Zmiany stanu przepływu pracy	Zawiera wartość zdarzenia `{"state": "<value>"}`
`CUSTOM` (`request_info`)	Przepływ pracy wymaga wkładu człowieka	Zawiera ładunek żądania dla klienta w celu renderowania monitu
`CUSTOM` (`workflow_output`)	Przepływ pracy generuje dane wyjściowe	Emitowane zarówno dla zdarzeń `"output"` (terminal), jak i `"intermediate"` przepływu pracy. Dane wyjściowe terminalu zawierają ostateczną odpowiedź; pośrednie dane wyjściowe są wyświetlane jako `text_reasoning` zawartość, gdy przepływ pracy działa za `as_agent()`.
`RUN_FINISHED`	Uruchamianie ukończone	Uwzględnia `outcome.type == "interrupt"` i `outcome.interrupts` gdy przepływ pracy oczekuje na dane wejściowe

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

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").
Mostek AG-UI emituje zdarzenie CUSTOM zawierające name="request_info" dane żądania.

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

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

Człowiek w pętli

Dodatkowe zasoby

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

Wskazówka

Zobacz przykład przepływu pracy jako agenta i przykład serwera AG-UI, aby zapoznać się z kompletnymi przykładami gotowymi do uruchomienia.

Opinia

Czy ta strona była pomocna?

Last updated on 2026-08-14