Alur kerja dengan AG-UI

Nota

Dukungan alur kerja untuk integrasi .NET AG-UI akan segera hadir.

Tutorial ini menunjukkan kepada Anda cara mengekspos alur kerja Agent Framework melalui titik akhir AG-UI. Alur kerja mengatur beberapa agen dan alat dalam grafik eksekusi yang ditentukan, dan integrasi AG-UI mengalirkan peristiwa alur kerja yang kaya — pelacakan langkah, rekam jepret aktivitas, gangguan, dan peristiwa kustom — ke klien web secara real time.

Prasyarat

Sebelum memulai, pastikan Anda memiliki:

Python 3.10 atau yang lebih baru
agent-framework-ag-ui Diinstal
Pemahaman tentang tutorial Pengantar Memulai
Pemahaman dasar tentang alur kerja Kerangka Kerja Agen

Kapan Menggunakan Alur Kerja dengan AG-UI

Gunakan alur kerja alih-alih satu agen saat Anda membutuhkan:

Orkestrasi multi-agen: Merutekan tugas antara agen khusus (misalnya, penyortiran → mengembalian dana → pesanan)
Langkah-langkah eksekusi terstruktur: Melacak kemajuan melalui tahap yang ditentukan dengan STEP_STARTED / STEP_FINISHED peristiwa
Hentikan sementara / lanjutkan alur: Menghentikan sementara eksekusi untuk mengumpulkan input atau persetujuan dari manusia, lalu melanjutkannya kembali
Streaming peristiwa kustom: Mengirimkan peristiwa khusus domain (request_info, status, workflow_output) kepada klien

Membungkus Alur Kerja dengan AgentFrameworkWorkflow

AgentFrameworkWorkflow adalah pembungkus ringan yang menyesuaikan Workflow asli dengan protokol AG-UI. Anda dapat menyediakan instans alur kerja bawaan atau pabrik yang membuat alur kerja baru per alur.

Instans langsung

Gunakan instans langsung saat satu objek alur kerja dapat melayani semua permintaan dengan aman (misalnya, alur stateless):

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

Pabrik berskala utas

Gunakan workflow_factory saat setiap utas percakapan memerlukan status alur kerjanya sendiri. Pabrik menerima thread_id dan mengembalikan yang baru 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.",
)

Penting

Anda harus lulus salah satuworkflowatauworkflow_factory, bukan keduanya. Pembungkus akan memunculkan ValueError jika keduanya disediakan.

Mendaftarkan Titik Akhir

Daftarkan alur kerja dengan add_agent_framework_fastapi_endpoint cara yang sama seperti Anda akan mendaftarkan satu agen:

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

Anda juga dapat meneruskan "bare" Workflow secara langsung — endpoint akan membungkusnya secara otomatis di AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

Peristiwa AG-UI Dikeluarkan oleh Proses Kerja

Alur kerja berjalan memancarkan serangkaian event AG-UI yang lebih kaya dibandingkan dengan menjalankan agen tunggal:

Event	Ketika dipancarkan	Deskripsi
`RUN_STARTED`	Proses dimulai	Menandai awal eksekusi alur kerja
`STEP_STARTED`	Eksekutor atau superstep dimulai	`step_name` mengidentifikasi agen atau langkah (misalnya, `"triage_agent"`)
`TEXT_MESSAGE_*`	Agen memproduksi teks	Peristiwa teks streaming standar
`TOOL_CALL_*`	Agen memanggil alat	Peristiwa panggilan alat standar
`STEP_FINISHED`	Proses pelaksana atau superstep selesai	Menutup langkah untuk pelacakan kemajuan UI
`CUSTOM`(`status`)	Perubahan status alur kerja	Berisi `{"state": "<value>"}` dalam nilai peristiwa
`CUSTOM`(`request_info`)	Alur kerja meminta input manusia	Berisi payload permintaan bagi klien untuk menghasilkan prompt.
`CUSTOM`(`workflow_output`)	Alur kerja menghasilkan output	Berisi data output akhir atau menengah
`RUN_FINISHED`	Proses selesai	Dapat mencakup `interrupts` jika alur kerja sedang menunggu input

Klien-klien dapat menggunakan event STEP_STARTED / STEP_FINISHED untuk merender indikator kemajuan yang menunjukkan agen yang saat ini aktif.

Interupsi dan Lanjutkan

Alur kerja dapat menjeda eksekusi untuk mengumpulkan input dari manusia atau persetujuan dari alat. Integrasi AG-UI menangani ini melalui protokol interupsi/lanjutkan.

Cara kerja interupsi

Selama eksekusi, alur kerja mengajukan permintaan yang tertunda (misalnya, HandoffAgentUserRequest untuk meminta detail lebih lanjut, atau alat dengan approval_mode="always_require").
Jembatan AG-UI mengeluarkan CUSTOM event dengan name="request_info" yang berisi data permintaan.

Eksekusi selesai dengan peristiwa yang RUN_FINISHED bidangnya interrupts berisi daftar objek permintaan yang tertunda:

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

Klien merender antarmuka pengguna untuk merespons (input teks, tombol persetujuan, dll.).

Cara kerja resume

Klien mengirimkan permintaan baru dengan resume payload yang berisi respons pengguna yang di-key oleh ID interupsi:

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

Server mengonversi payload resume menjadi respons alur kerja dan melanjutkan eksekusi dari tempatnya dijeda.

Contoh Lengkap: Alur Kerja Handoff Multi-Agen

Contoh ini menunjukkan alur kerja dukungan pelanggan dengan tiga agen yang menyerahkan pekerjaan satu sama lain, menggunakan alat yang memerlukan persetujuan, dan meminta input manusia saat diperlukan.

Menentukan agen dan alat

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

Menyusun alur kerja

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

Membuat aplikasi 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)

Urutan peristiwa

Interaksi multi-giliran yang khas menghasilkan peristiwa seperti:

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

Klien kemudian dapat menampilkan dialog persetujuan dan melanjutkan dengan keputusan pengguna.

Menerima Alat Peraga yang Diteruskan

AG-UI klien (seperti CopilotKit) dapat menyertakan forwarded_props bidang (atau forwardedProps) dalam payload input. Integrasi AG-UI secara otomatis meneruskan properti ini ke metode run alur kerja melalui argumen kata kunci 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.
        ...

Detail utama:

Baik forwarded_props dan forwardedProps diterima dalam payload input; secara internal mereka dinormalisasi menjadi forwarded_props.
Jika workflow.run() tidak menerima function_invocation_kwargs (atau **kwargs), alat peraga diam-diam dihilangkan — alur kerja yang ada tidak terpengaruh.
Alat peraga yang diteruskan juga disimpan dalam metadata sesi tetapi difilter dari metadata terikat LLM, sehingga tidak bocor ke permintaan klien obrolan.

Langkah berikutnya

Human-in-the-Loop

Sumber Daya Tambahan

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-05-11