AG-UI 사용하는 워크플로

이 자습서에서는 AG-UI 엔드포인트를 통해 에이전트 프레임워크 워크플로를 노출하는 방법을 보여 줍니다. 워크플로는 정의된 실행 그래프에서 여러 에이전트 및 도구를 오케스트레이션하고, AG-UI 통합은 단계 추적, 활동 스냅샷, 인터럽트 및 사용자 지정 이벤트와 같은 풍부한 워크플로 이벤트를 실시간으로 웹 클라이언트로 스트리밍합니다.

사전 요구 사항

시작하기 전에 다음 사항을 확인하세요.

• Python 3.10 이상
• agent-framework-ag-ui 설치
• 시작 자습서에 대한 숙지
• 에이전트 프레임워크 워크플로에 대한 기본 이해

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

엔드포인트 등록

단일 에이전트를 등록하는 것과 동일한 방식으로 워크플로 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",
)

bare Workflow 를 직접 전달할 수도 있습니다. 엔드포인트가 이를 AgentFrameworkWorkflow로 자동으로 래핑합니다.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

워크플로에서 발생된 AG-UI 이벤트

워크플로 실행은 단일 에이전트 실행에 비해 다양한 AG-UI 이벤트 집합을 내보냅니다.

클라이언트는 이벤트를 사용하여 STEP_STARTED / STEP_FINISHED 현재 활성 상태인 에이전트를 보여 주는 진행률 표시기를 렌더링할 수 있습니다.

인터럽트 및 다시 시작

워크플로는 실행을 일시 중지하여 사용자 입력 또는 도구 승인을 수집할 수 있습니다. AG-UI 통합은 인터럽트/다시 시작 프로토콜을 통해 이를 처리합니다.

인터럽트 작동 방식

1. 실행 중 워크플로는 (예: HandoffAgentUserRequest을 통한 추가 정보 요청이나 approval_mode="always_require"을 포함한 도구 사용 등의) 보류 중인 요청을 발생시킵니다.

2. AG-UI 브리지는 CUSTOM 이벤트를 name="request_info"으로 요청 데이터를 포함하여 내보냅니다.

3. 실행은 보류 중인 요청 개체 목록을 포함하는 필드의 이벤트 RUN_FINISHED 로 interrupts 완료됩니다.

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

4. 클라이언트는 사용자가 응답할 수 있도록 UI를 렌더링합니다(텍스트 입력, 승인 단추 등).

이력서 작동 방식

클라이언트는 인터럽트 ID로 키가 지정된 사용자의 응답을 포함하는 페이로드를 사용하여 새 요청을 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}}]

그런 다음 클라이언트는 승인 대화 상자를 표시하고 사용자의 결정에 따라 다시 시작할 수 있습니다.

다음 단계

추가 리소스

• AG-UI 개요
• 시작하기
• 에이전트 프레임워크 워크플로
• Agent Framework GitHub 리포지토리