Рабочие процессы с AG-UI

Замечание

Поддержка рабочих процессов для интеграции .NET AG-UI скоро ожидается.

В этом руководстве показано, как предоставлять рабочие процессы Agent Framework через конечную точку AG-UI. Рабочие процессы оркеструют несколько агентов и инструментов в определенном графе выполнения процессов, а потоки данных интеграции AG-UI передают разнообразные события рабочего процесса — отслеживание шагов, моментальные снимки действий, прерывания и пользовательские события — веб-клиентам в режиме реального времени.

Необходимые условия

Прежде чем начать, убедитесь, что у вас есть следующее:

Python 3.10 или более поздней версии
agent-framework-ag-ui установлен
Знакомство с руководством по началу работы
Основные сведения о рабочих процессах Agent Framework

Когда использовать воркфлоу с AG-UI

При необходимости используйте рабочий процесс вместо одного агента:

Оркестрация нескольких агентов: маршрутизация задач между специализированными агентами (например, триаж → возврат средств → заказ)
Структурированные шаги выполнения: Отслеживание прогресса через определенные этапы и STEP_STARTED / STEP_FINISHED события
Прерывание / возобновление потоков: приостановка выполнения для сбора данных или согласований от человека, а затем возобновление
Настраиваемая потоковая передача событий: выдача событий для конкретного домена (request_info, status, workflow_output) клиенту

Упаковка рабочего процесса с помощью AgentFrameworkWorkflow

AgentFrameworkWorkflow — это легкая оболочка, которая адаптирует встроенный Workflow к протоколу AG-UI. Можно предоставить предварительно созданный экземпляр рабочего процесса или фабрику, которая создает новый рабочий процесс для каждого потока.

Прямая инстанция

Используйте прямой экземпляр, когда один объект рабочего процесса может безопасно обслуживать все запросы (например, конвейеры без отслеживания состояния):

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

Фабрика с областью видимости потока

Используйте workflow_factory, когда каждому потоку беседы требуется собственное состояние рабочего процесса. Фабрика получает thread_id и возвращает свежий Workflow:

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

Это важно

Необходимо передать либоworkflowилиworkflow_factory, не оба. Оболочка вызывает ValueError, если предоставлены оба элемента.

Регистрация конечной точки

Зарегистрируйте рабочий процесс add_agent_framework_fastapi_endpoint так же, как регистрируете одного агента.

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

Вы также можете передать необработанный Workflow непосредственно, конечный узел автоматически упакует его в AgentFrameworkWorkflow.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

События AG-UI, испускаемые рабочими процессами

Выполнение рабочего процесса генерирует более богатый набор событий AG-UI по сравнению с одиночными запусками агентов.

Event	При излучении	Описание
`RUN_STARTED`	Запуск начинается	Помечает начало выполнения рабочего процесса
`STEP_STARTED`	Начинается выполнение или супершаг.	`step_name` определяет агент или шаг (например, `"triage_agent"`)
`TEXT_MESSAGE_*`	Агент создает текст	Стандартные события потоковой передачи текста
`TOOL_CALL_*`	Агент вызывает средство	События вызова стандартного инструмента
`STEP_FINISHED`	Исполнитель или супершаг завершен	Закрывает шаг отслеживания прогресса в пользовательском интерфейсе
`CUSTOM` (`status`)	Изменения состояния рабочего процесса	Содержится `{"state": "<value>"}` в значении события
`CUSTOM` (`request_info`)	Рабочий процесс запрашивает входные данные человека	Содержит полезные данные запроса для клиента для отображения подсказки.
`CUSTOM` (`workflow_output`)	Рабочий процесс создает выходные данные	Содержит окончательные или промежуточные выходные данные
`RUN_FINISHED`	Выполнение завершено	Может включаться `interrupts`, если рабочий процесс ждет входных данных

Клиенты могут использовать STEP_STARTED / STEP_FINISHED события для отображения индикаторов хода выполнения, показывающих, какой агент сейчас активен.

Прерывание и возобновление

Рабочие процессы могут приостановить выполнение для сбора человеческого ввода или одобрений инструментов. Интеграция AG-UI обрабатывает это с помощью протокола прерывания/возобновления.

Как работают прерывания

Во время выполнения рабочий процесс вызывает ожидающий запрос (например, HandoffAgentUserRequest запрос на дополнительные сведения или инструмент с approval_mode="always_require").
Мост AG-UI выдает CUSTOM событие с name="request_info" данными запроса.

Выполнение завершается событием RUN_FINISHED , поле которого interrupts содержит список ожидающих объектов запроса:

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

Клиент отрисовывает пользовательский интерфейс для ответа пользователя (текстовый ввод, кнопка утверждения и т. д.).

Как работает резюме

Клиент отправляет новый запрос с полезными данными resume, содержащими ответы пользователя, ключом которого является идентификатор прерывания:

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

Сервер преобразует полезную нагрузку возобновления в ответы рабочего процесса и продолжает выполнение с того момента, где он был приостановлен.

Полный пример: процесс передачи между несколькими агентами

В этом примере показан рабочий процесс поддержки клиентов с тремя агентами, которые передают работу друг другу, используют средства, требующие утверждения, и запрашивают человеческое участие при необходимости.

Определение агентов и средств

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

Создание рабочего процесса

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

Создание приложения 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)

Последовательность событий

Обычное многоэтапное взаимодействие создает такие события, как:

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

Затем клиент может отобразить диалоговое окно утверждения и возобновить работу с решением пользователя.

Дальнейшие действия

Человек в цикле

Дополнительные ресурсы

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-04-09

Поделиться через

Рабочие процессы с AG-UI

Необходимые условия

Когда использовать воркфлоу с AG-UI

Упаковка рабочего процесса с помощью AgentFrameworkWorkflow

Прямая инстанция

Фабрика с областью видимости потока

Регистрация конечной точки

События AG-UI, испускаемые рабочими процессами

Прерывание и возобновление

Как работают прерывания

Как работает резюме

Полный пример: процесс передачи между несколькими агентами

Определение агентов и средств

Создание рабочего процесса

Создание приложения FastAPI

Последовательность событий

Дальнейшие действия

Дополнительные ресурсы

Обратная связь

Дополнительные ресурсы