AG-UI ile iş akışları

Uyarı

.NET AG-UI tümleştirmesi için iş akışı desteği yakında sunulacaktır.

Bu öğreticide, bir AG-UI uç noktası aracılığıyla Agent Framework iş akışlarını nasıl kullanıma sunabileceğiniz gösterilmektedir. İş akışları tanımlı bir yürütme grafiğinde birden çok aracıyı ve aracıyı düzenler ve AG-UI tümleştirmesi, web istemcilerine gerçek zamanlı olarak zengin iş akışı olayları (adım izleme, etkinlik anlık görüntüleri, kesmeler ve özel olaylar) aktarır.

Önkoşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

• Python 3.10 veya üzeri
• agent-framework-ag-ui Yüklü
• Başlama eğitimine aşinalık
• Agent Framework iş akışları hakkında temel bilgiler

AG-UI ile İş Akışları Ne Zaman Kullanılır?

İhtiyacınız olduğunda tek bir aracı yerine iş akışı kullanın:

• Çoklu ajan koordinasyonu: Özel aracılar arasındaki görevleri yönlendirme (örneğin, triyaj → para iadesi → sipariş)
• Yapılandırılmış yürütme adımları: Olaylarla STEP_STARTED / STEP_FINISHED tanımlı aşamalarda ilerleme durumunu izleme
• Kesme/sürdürme akışları: İnsan girişi veya onaylarını toplamak için yürütmeyi duraklatıp sürdürme
• Özel olay akışı: Etki alanına özgü olayları (request_info, status, workflow_output) istemciye yayar

AgentFrameworkWorkflow ile İş Akışını Sarmalama

AgentFrameworkWorkflow, AG-UI protokolüne uyarlamak için yerel Workflow'yi kullanan hafif bir sarmalayıcıdır. Önceden oluşturulmuş bir iş akışı örneği veya iş parçacığı başına yeni iş akışı oluşturan bir fabrika sağlayabilirsiniz.

Doğrudan örnek

Tek bir iş akışı nesnesi tüm isteklere (örneğin durum bilgisi olmayan işlem hatları) güvenli bir şekilde hizmet verebiliyorsa doğrudan örnek kullanın:

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

İş parçacığı kapsamlı fabrika

Her konuşma dizisinin kendi iş akışı durumuna ihtiyacı olduğunda workflow_factory kullanın. Fabrika thread_id alır ve yeni bir Workflow döndürür.

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

Önemli

Yaworkflow ya daworkflow_factory Sarmalayıcı, her ikisi de sağlanmışsa bir ValueError oluşturur.

Uç Noktayı Kaydetme

İş akışını add_agent_framework_fastapi_endpoint tek bir ajan kaydettiğiniz gibi kaydedin.

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

Ayrıca yalnızca Workflow geçirmeniz de mümkündür — uç nokta bunu AgentFrameworkWorkflow ile otomatik olarak sarar.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI İş Akışları Tarafından Oluşturulan Olaylar

İş akışı çalıştırmaları, tek aracılı çalıştırmalara kıyasla daha zengin bir AG-UI olay kümesi yayar:

Etkinlik	Yayıldığında	Açıklama
RUN_STARTED	Çalışma başlıyor	İş akışı yürütmenin başlangıcını işaretler
STEP_STARTED	Yürütücü veya üst adım başlar	step_name aracıyı veya adımı tanımlar (örneğin, "triage_agent")
TEXT_MESSAGE_*	Yazılım aracı metin üretir	Standart akış metin olayları
TOOL_CALL_*	Temsilci bir aracı çalıştırır	Standart araç çağrısı olayları
STEP_FINISHED	Yürütücü veya üst adım tamamlar	Kullanıcı arabirimi ilerleme durumunu izleme adımını kapatır
CUSTOM (status)	İş akışı durumu değişiklikleri	Olay değerinde {"state": "<value>"} içerir
CUSTOM (request_info)	İş akışı insan girişi isteği	İstemcinin bir uyarı görüntülemesi için istek yükünü içerir
CUSTOM (workflow_output)	İş akışı çıktı oluşturuyor	Son veya ara çıktı verilerini içerir
RUN_FINISHED	Çalıştırma tamamlandı	İş akışı giriş bekliyorsa interrupts içerebilir.

İstemciler, hangi aracının şu anda etkin olduğunu gösteren ilerleme göstergelerini işlemek için olayları kullanabilir STEP_STARTED / STEP_FINISHED .

Kesme ve Sürdürme

İş akışları, insan girişi veya araç onaylarını toplamak için yürütmeyi duraklatabilir. AG-UI entegrasyonu, bu işlemi kesme/sürdürme protokolü aracılığıyla gerçekleştirir.

Kesinti işlemleri nasıl çalışır?

1. Yürütme sırasında iş akışı, bekleyen bir istek oluşturur (örneğin, daha fazla ayrıntı isteyen bir HandoffAgentUserRequest veya bir approval_mode="always_require" araç).

2. AG-UI köprüsü, istek verilerini içeren bir CUSTOM olayını name="request_info" yayar.

3. Çalıştırma, RUN_FINISHED alanında bekleyen istek nesnelerinin listesini içeren bir interrupts olayı ile tamamlanır.

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

4. İstemci, kullanıcının yanıt vermesi için kullanıcı arabirimini işler (metin girişi, onay düğmesi vb.).

Özgeçmiş nasıl çalışır?

İstemci, kullanıcının yanıtlarının kesme kimliğiyle resume anahtarlandığı yükü içeren yeni bir istek gönderir:

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

Sunucu, özgeçmiş yükünü iş akışı yanıtlarına dönüştürür ve duraklatıldığı yerden yürütmeye devam eder.

Tam Örnek: Çok Aracılı İletim İş Akışı

Bu örnekte, çalışmayı birbirine devreden, onay gerektiren araçları kullanan ve gerektiğinde insan girişi isteyen üç aracıya sahip bir müşteri desteği iş akışı gösterilmektedir.

Aracıları ve araçları tanımlayın

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

İş akışını oluşturma

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 uygulamasını oluşturma

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)

Olay dizisi

Tipik bir çok dönüşlü etkileşim aşağıdakiler gibi olaylar üretir:

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

İstemci daha sonra bir onay iletişim kutusu görüntüleyebilir ve kullanıcının kararıyla devam edebilir.

İletilen Prop'ları Karşılama

AG-UI istemcileri (CopilotKit gibi) giriş yüküne bir forwarded_props (veya forwardedProps) alanı içerebilir. AG-UI tümleştirmesi, anahtar sözcük bağımsız değişkeni aracılığıyla bu prop'ları iş akışının run yöntemine function_invocation_kwargs otomatik olarak geçirir:

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

Önemli ayrıntılar:

• Giriş yükünde forwarded_props ve forwardedProps kabul edilir; dahili olarak forwarded_props normalleştirilir.
• Eğer workflow.run() (veya **kwargs) kabul etmezse, function_invocation_kwargs özellikleri sessizce bırakılır; mevcut iş akışları etkilenmez.
• İletilen prop'lar da oturum meta verilerinde depolanır, ancak LLM'ye bağlı meta verilerden filtrelenir, bu nedenle sohbet istemci isteklerine sızmaz.

Sonraki Adımlar

Döngüdeki İnsan

Ek Kaynaklar

• AG-UI Genel Bakış
• Başlarken
• Agent Framework İş Akışları
• Agent Framework GitHub Deposu