AG-UI ile iş akışları

MAF .NET, iş akışını bir AIAgent öğesine dönüştürüp diğer aracılar gibi eşleyerek AG-UI üzerinden sunabilir:

AIAgent workflowAgent = AgentWorkflowBuilder
    .BuildSequential(researcher, reporter)
    .AsAIAgent();

app.MapAGUIServer("/", workflowAgent);

Uç nokta, bileşen aracılarının standart metin ve araç çağrısı çıktılarının akışını sağlar. AuthorName her güncelleştirmeyi üreten aracıyı tanımlar.

MAF .NET şu anda iş akışına özgü yaşam döngüsü davranışını AG-UI ile eşlemez. İstemciler Python tümleştirmesine eşdeğer iş akışı adımı olaylarını, etkinlik anlık görüntülerini, iş akışı kesintilerini veya iş akışı sürdürme işlemlerini almaz. Bir iş akışını AIAgent olarak sarmak, bu eşlemeleri eklemez.

Geçerli .NET izleme durumu için bkz. microsoft/agent-framework#2494. AG-UI'lerden bağımsız iş akışı oluşturma ve yürütme için bkz. MAF iş akışı kavramları.

Sonraki Adımlar

Üretim ve güvenlikle ilgili dikkat edilmesi gerekenleri gözden geçirin

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.

Prerequisites

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

Python 3.10 veya üzeri
agent-framework-ag-ui ve agent-framework-foundry yüklendi
Başlama eğitimine aşinalık
Agent Framework iş akışı kavramlarını temel olarak anlama

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:

Event	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ı
`REASONING_*`	İş akışı, `intermediate_output_from` içinde yapılandırılmış bir yürütücüden metin çıktısı verir.	Ara metni bir akıl yürütme bloğu olarak aktarır. Kullanım dışı bırakılan `"data"` olay diğer adı aynı yolu izler.
`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ışı çıkışı ileti içeriğine dönüştürülemiyor	Özel istemci işleme için serileştirilmiş çıktıyı içerir.
`RUN_FINISHED`	Çalıştırma tamamlandı	İş akışı girdi beklediğinde `outcome.type == "interrupt"` ve `outcome.interrupts` öğelerini içerir

İ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 . Tümleştirme, terminal olayından veya insan girişi isteğinden önce açık mantık ve metin bloklarını kapatır, böylece istemciler eksiksiz bir olay dizisi alır.

bir Python iş akışı başarısız olduğunda, RUN_ERROR genel genel iletiyi Workflow execution failed. ve bir hata kodunu kullanır. Bir executor_failed olay da aynı şekilde genel mesajı ve hata türünü sunar. İç özel durum ayrıntıları ve izlemeler sunucu günlüklerinde kalır.

Duraklat ve devam et

İş 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?

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ç).
AG-UI köprüsü, istek verilerini içeren bir CUSTOM olayını name="request_info" yayar.

Çalıştırma, RUN_FINISHED alanı bekleyen istekleri içeren bir outcome.interrupts olayla sona erer:

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

İ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 kurallı resume bir dizi ile yeni bir istek gönderir. Her girdi kesmeyi tanımlar ve kullanıcının yanıtını sağlar:

{
  "threadId": "abc123",
  "messages": [],
  "resume": [
    {
      "interruptId": "request-id-1",
      "status": "resolved",
      "payload": "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. Bunun yerine, kesintiye uğrayan çalıştırmayı iptal etmek için status değerini "cancelled" olarak ayarlayın ve payload belirtmeyin.

İş akışı denetim noktalarını kalıcı hale alma ve sürdürme

AgentFrameworkWorkflow üzerinde checkpoint_storage öğesini, her bir süper adımın sonunda temel iş akışı durumunu kaydedecek şekilde yapılandırın. Bunun yerine, bir iş akışını kaydederken aynı argümanı add_agent_framework_fastapi_endpoint'a iletebilirsiniz. Altta yatan iş akışı denetim noktası depolamasıyla oluşturulduysa, bağdaştırıcı ilgili oluşturucuyu veya çalışma zamanı depolamasını doğrudan kullanabilir; bu nedenle yapılandırmayı sarmalayıcıda ya da uç noktada yeniden tanımlamanız gerekmez.

Aşağıdaki örnek, kısa süreli bir iş akışı için bellek içi depolama kullanır:

from agent_framework import InMemoryCheckpointStorage
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)
from fastapi import FastAPI

app = FastAPI()
checkpoint_storage = InMemoryCheckpointStorage()
workflow = build_my_workflow()

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    checkpoint_storage=checkpoint_storage,
)
add_agent_framework_fastapi_endpoint(
    app,
    ag_ui_workflow,
    "/workflow",
)

Bir çalıştırma duraklatıldığında ve bir duraklatma denetim noktası kullanılabiliyorsa, RUN_FINISHED olayındaki her kesme, denetim noktası kimliğini metadata.agent_framework.checkpoint_id içinde içerir. Ayrı bir en son denetim noktasını arama işlemi yapmadan, uygulama örnekleri arasında tam olarak aynı duraklatılmış noktadan devam etmek için bu üretilen değeri kullanın.

AgentFrameworkWorkflow.run() AG-UI istek yükünü aldığı için istemci, denetim noktası kimliğini Python checkpoint_id argümanı yerine iletilen özellikler üzerinden sağlar. Yalnızca denetim noktası özgeçmişi yeni bir kullanıcı iletisi içermez:

{
  "threadId": "abc123",
  "messages": [],
  "forwardedProps": {
    "checkpointId": "checkpoint-id-from-interrupt-metadata"
  }
}

Bağdaştırıcı kaydedilen iş akışı durumunu geri yükler ve yürütmeye devam eder. Denetim noktası bekleyen bir kesme içeriyorsa, aynı istekte hem denetim noktası kimliğini hem de kanonik resume yükünü ekleyin. Bağdaştırıcı, kesme yanıtını vermeden önce denetim noktasını geri yükler. metadata.agent_framework.checkpoint_id değerini forwardedProps.checkpointId olarak kullanın.

Bağdaştırıcı her yeni denetim noktasını isteğin Anlık Görüntü Kapsamına ve istemci tarafından sağlanan threadIdöğesine bağlar. Değerlerden biri eşleşmediğinde özgeçmiş isteğini reddeder. Sahiplik meta verileri kullanılmadan önce yazılan denetim noktaları uyumluluk açısından devam ettirilebilir durumda kalır.

Bu denetim, uç nokta yetkilendirmesi veya korumalı denetim noktası depolama alanının yerini almaz. Daha fazla bilgi için bkz. güvenlikle ilgili dikkat edilmesi gerekenler.

InMemoryCheckpointStorage, işlem yeniden başlatmalarından sonra korunmaz. Dayanıklı depolama seçenekleri ve denetim noktası seçimi için bkz. Denetim noktaları.

İş akışı denetim noktaları ve AG-UI iş parçacığı anlık görüntüleri

İş akışı denetim noktaları ile AG-UI iş parçacığı anlık görüntüleri farklı veriler saklar:

Kalıcılık mekanizması	Mağazalar	Purpose
Agent Framework iş akışı denetim noktası	Bekleyen istekler de dahil olmak üzere yürütücü ve çalışma zamanı durumu	Kaydedilen çalışma zamanı durumundan iş akışı yürütmeyi sürdürme
AG-UI iş parçacığı anlık görüntüsü	İletiler, paylaşılan durum ve en son kesme gibi yeniden oynatılabilir protokol çıktıları	İstemciye görünen iş parçacığını yeniden etkinleştir

Her iki mekanizmayı da yapılandırabilirsiniz. İş akışı denetim noktası, AG-UI iş parçacığı anlık görüntüsünün yerini almaz ve AG-UI iş parçacığı anlık görüntüsü, iş akışı yürütmeyi sürdürmek için gereken yürütücü durumunu içermez.

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

İş akışını oluşturma

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

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          outcome={type: "interrupt", interrupts: [{id: "...", reason: "tool_call"}]}

İ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.
İletilen özellikler içinde, checkpoint_id ve checkpointId, iş akışı denetim noktasından sürdürme için ayrılmıştır.
Eğer workflow.run() (veya function_invocation_kwargs) kabul etmezse, **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

Go, workflow.Workflow öğesini workflow/agentworkflow ile bir aracı olarak sarıp ardından bu aracıyı provider/aguiprovider ile barındırarak iş akışlarını AG-UI’ye sunabilir.

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

Tip

Tam çalıştırılabilir örnekler için iş akışını aracı örneği olarak ve AG-UI sunucu örneğine bakın.

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Last updated on 2026-09-18

AG-UI ile iş akışları

Sonraki Adımlar

Prerequisites

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

AgentFrameworkWorkflow ile İş Akışını Sarmalama

Doğrudan örnek

İş parçacığı kapsamlı fabrika

Uç Noktayı Kaydetme

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

Duraklat ve devam et

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

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

İş akışı denetim noktalarını kalıcı hale alma ve sürdürme

İş akışı denetim noktaları ve AG-UI iş parçacığı anlık görüntüleri

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

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

İş akışını oluşturma

FastAPI uygulamasını oluşturma

Olay dizisi

İletilen Prop'ları Karşılama

Sonraki Adımlar

Ek Kaynaklar

Geri Bildirim

Ek kaynaklar