Microsoft Agent Framework – Basit Düzeyde Workflow Tasarlama

Merhaba,

Bu içeriğimizde, Microsoft Agent Framework Nedir? Konseptleri Nelerdir? başlıklı makalemizde ele aldığımız Microsoft Agent Framework’ün ana konseptlerinden biri olan Workflow yapısına odaklanacak, pratiksel açıdan hangi kavramları barındırdığını ve davranışsal olarak hangi yaklaşımları sağladığını birlikte değerlendirerek, deneyimleyeceğiz. O halde buyurun başlayalım…

Workflow Nedir? Kısaca Hatırlayalım…

Microsoft Agent Framework’de workflow, birden fazla AI Agent’ı ve fonksiyonu bağlayan graph tabanlı (directed graph) akıştır. Özellikle karmaşık ve çok adımlı görevleri yönetmek için kullanılırken, executor’lar aracılığıyla agent’ları ve fonksiyonları birbirlerine Node olarak bağlamamızı ve Edge denen yapılarla da koşullu veri akışları oluşturarak akışta bir sonraki Node’u belirleyebilmemizi ve yapılandırabilmemizi sağlamaktadır. Ayrıca akışta checkpointing ile insan müdahalesine (human-in-the-loop) açık bir yapılanma imkanının yanında, bir yandan da paralel işlemeyi doğal olarak destekleyerek süreci işlevsel açıdan daha efektif yürütebilme imkanı sağlamaktadır.

Kısaca workflow, birden fazla agent’ı ve fonksiyonu, akış oklarıyla birbirine bağlayıp ‘şunu yap, sonra şuna göre devam et’ şeklinde talimat verip, planlama yapmamızı sağlayan akıllı bir iş akışıdır.

Workflow’da Executor Kavramı

Microsoft Agent Framework’de Workflow mantığını teknik açıdan anlayabilmek için öncelikle temel bilinmesi gereken kritik olgu Executor kavramıdır.

Executor, workflow graph’ında bir Node olarak davranan; agent’ları, fonksiyonları veya custom logic’i kapsayan ve mesajları (inputları) işleyip output üretebilen soyut bir sınıftır. Yani Agent Framework’te workflow mantığında işi yürüten (gerçekten yapan) executor’dır.

Basit mantıkla şöyle düşünülebilir; workflow’un içinde bir sürü Node diyeceğimiz kutucuk olduğunu varsayalım. Her kutucuğun içinde ‘kim bu işi yapacak?’ sorusunun yazdığını düşünelim. İşte o sorunun cevabı executor’dır. Executor’ın tek görevi şudur: “Tamam, sıra bende. Bana verilen görevi (mesela bir soruyu cevaplama, bir API’ye gitme, hesaplama yapma vs.) hemen yapmalıyım ve sonucu alıp bir sonraki Edge’e iletmeliyim…” Dolayısıyla bir sonraki Edge’de farklı bir kutucuk (Node), o kutucukta da farklı bir executor sırasını beklemektedir.

Executor, workflow’da işi fiilen yapan, talimatı uygulayan elemandır.

Teknik açıdan Executor’ı özetlememiz gerekirse eğer;

• IMessageHandler arayüzü sayesinde mesajları (input) alır, işler ve bağlı diğer executor’lara iletir.
• Parallel (fan-out/fan-in), sequential veya conditional akışları destekler.
• Her executor’un benzersiz bir ID’si mevcuttur ve belirli mesaj tiplerini filtreleyebilmektedir.
• Agent executor’ları (AgentExecutor) agent çağrılarını yönetir. Custom executor’lar ise özel logic için kullanılabilir.

Executor mantığının teknik açıdan en büyük avantajı; type-safe, stateful ve observable bir ortam sağlamasıdır. Ayrıca uzun süren workflow’larda durum kaydetme (checkpointing) yaparak işleri kolaylaştırmaktadır.

ReflectingExecutor<T> Class’ı Nedir?
ReflectingExecutor<T> class’ı, reflection kullanarak sınıf metotlarını otomatik taramakta ve bunları handler olarak kaydetmekte ve böylece, kodu yazarken metotları manuel kaydetmeye gerek bulunmamaktadır.

Örnek kullanımı aşağıdaki gibidir;

Executor tasarlayabilmek için öncelikle Microsoft.Agents.AI.Workflows kütüphanesinin yüklenmesi gerekmektedir!

using Microsoft.Agents.AI.Workflows.Reflection;
using Microsoft.Agents.AI.Workflows;

public sealed class UppercaseExecutor() : ReflectingExecutor<UppercaseExecutor>(nameof(UppercaseExecutor)), IMessageHandler<string, string>
{
    public async ValueTask<string> HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        await Console.Out.WriteLineAsync(message);
        return message.ToUpper();
    }
}

Kodu incelerseniz eğer UppercaseExecutor adında bir executor oluşturulmuştur. Dikkat ederseniz bu sınıf ReflectingExecutor<T> base class’ından inherit edilmekte ve IMessageHandler<string, string> arayüzünü implement etmektedir. Bu executor mahiyeti doğrultusunda üretilecek sonuç otomatik olarak workflow’daki diğer executor’lara gönderilecektir. HandleAsync metodu ise reflection ile otomatik handle edilecektir.

Workflow’da bilinmesi gereken temel davranış bu executor’ların edge’lerle (bağlantılarla) birbirlerine bağlanmasıdır. Örneğin; .AddEdge(from_executor, to_executor)

Evet…

Artık workflow’lar da executor kavramının önemini kavradığımıza göre artık bu workflow üzerinde executor’ların davranışsal yaklaşımlarına geçiş yapabiliriz. Yani çok ajanlı orkestrasyon (multi-agent orchestration) süreçlerinde orchestration patterns olarak kullanacağımız workflow’ların execution stratejilerini belirleyen yaklaşımlara…

Bu yaklaşımlar özünde Sequential Workflow ve Concurrent Workflow olmak üzere iki tanedirler. Şimdi gelin tek tek bu yaklaşımları mercek altına alalım ve Agent Framework açısından nasıl yapılandırılabildiklerini pratiksel olarak değerlendirmeye başlayalım…

Sequential Workflow

Sequential Workflow, karmaşık AI Agent sistemleri oluşturmanın temelidir. Esasında executor’ların sırayla çalıştığı bir koordinasyon desenidir. Her executor, önceki executor’ın çıktısını alır, işler ve bir sonraki executor’a aktarır. Haliyle bu davranış, görevlerin belirli bir sırayla tamamlanması gerektiği durumlarda oldukça kullanışlıdır.

Sequential Workflow, executor’ların lineer bir zincirde çalıştığı bir graph yapısıdır. Her adım, yalnızca önceki adım tamamlandığında başlar.

Sequential workflow’un kullanım alanları, görevlerin işlenmesi için sıralı bağımlılıkların olduğu senaryolardır diyebiliriz. Misal olarak; veri ön işleme → analiz → raporlama gibi… Ya da buna daha net bir örnek vermemiz gerekirse eğer; bir müşteri hizmetleri akışını da düşünebilirsiniz. Önce kullanıcı talebini sınıflandırmak, sonra uygun agent’a yönlendirmek ve ardından yanıt oluşturmak bu mantıkta seyredeceğinden işte bu tarz durumlar için sequential workflow tercih edilmelidir.

Sequential Workflow’da insan onayı (human-in-the-loop) için belirli adımlarda duraklamalar eklenebilmektedir.

Sequential workflow için aşağıdaki örnek çalışmayı inceleyebilirsiniz;

public sealed class UppercaseExecutor() : ReflectingExecutor<UppercaseExecutor>(nameof(UppercaseExecutor)), IMessageHandler<string, string>
{
    public async ValueTask<string> HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        await Console.Out.WriteLineAsync(nameof(UppercaseExecutor));
        return message.ToUpper();
    }
}

public sealed class ReverseTextExecutor() : ReflectingExecutor<ReverseTextExecutor>(nameof(ReverseTextExecutor)), IMessageHandler<string, string>
{
    public async ValueTask<string> HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        await Console.Out.WriteLineAsync(nameof(ReverseTextExecutor));
        return new string(message.Reverse().ToArray());
    }
}

Görüldüğü üzere metinsel değerler üzerinden birinin tüm metni büyük karaktere, diğerinin ise tersine çeviren iki executor oluşturmuş bulunuyoruz. Şimdi bizler sequential workflow yaklaşımı ile bu executor’ları birbirlerine lineer bir şekilde bağlayarak, çalıştıralım.

UppercaseExecutor uppercaseExecutor = new();
ReverseTextExecutor reverseTextExecutor = new();

WorkflowBuilder workflowBuilder = new(uppercaseExecutor);
workflowBuilder.AddEdge(uppercaseExecutor, reverseTextExecutor)
    .WithOutputFrom(reverseTextExecutor);
var workFlow = workflowBuilder.Build();

Burada WorkflowBuilder sınıfı ile adı üzerinde bir workflow builder yaratılmakta ve ilk adım olarak uppercaseExecutor constructor’dan verilmektedir. Böylece iş akışının ilk basamağı olarak gelen metni büyük harfe çevirme görevi tanımlanmaktadır. Devamında ise .AddEdge metodu aracılığıyla uppercaseExecutor işlevi bittikten sonra akış reverseTextExecutor‘a yönlendirilmektedir. .WithOutputFrom metodu ile de workflow’un son çıktısının (final output) reverseTextExecutor‘ın çıktısı olacağı bildirilmektedir.

Artık oluşturulan bu workflow’u aşağıdaki gibi kullanmaya başlayabiliriz;

await using Run run = await InProcessExecution.RunAsync(workFlow, "laylaylom galibA sanA göRe SevMeLer");
foreach (WorkflowEvent workflowEvent in run.NewEvents)
{
    var result = workflowEvent switch
    {
        ExecutorCompletedEvent executorCompletedEvent => $"{executorCompletedEvent.ExecutorId} : {executorCompletedEvent.Data}",
        WorkflowOutputEvent workflowOutputEvent => $"'{workflowOutputEvent.SourceId}' outputs : {workflowOutputEvent.Data}",
        _ => string.Empty
    };

    if (!string.IsNullOrEmpty(result))
        await Console.Out.WriteLineAsync(result);
}

Burada InProcessExecution.RunAsync metodu ile workflow aynı process içinde çalıştırılmaktadır. O niye la? Neden böyle bişeye ihtiyaç var ki? diye sorduğunuzu duyar gibiyim… Bir workflow’u aynı process içerisinde çalıştırmamızın sebebini bu örnekten yola çıkarak anlamaya çalışmak ilk etapta pek doğru olmayacaktır. Çünkü, bu çalışmada 2 executor örneklendiriliyor olsa da gerçek senaryolarda en az 5, 10 veya 15 adımlık senaryolar söz konusu olabilmektedir.

Bunu şöyle izah edelim;

Öncelikle workflow engine’in In-process execution ve Out-of-process / hosted execution olmak üzere iki farklı çalışma modeli sunduğunu bilmenizi isterim.

• In-process execution
  • Workflow, aynı process içinde senkron/async olarak çalışma sergiler.
  • Adımlar memory içinde çalışır.
  • Çok hızlıdır.
  • Debug etmek kolaydır.
  • WorkflowEvent, ExecutorCompletedEvent, WorkflowOutputEvent vs. gibi event’ler anında loop’a düşer.
• Out-of-process / hosted execution
  • Workflow başka bir worker/service içerisinde yürütülür.
  • Step execution mesajlaşma sistemi veya task scheduler ile çalışır.
  • Event’ler queue üzerinden geçerek daha geç gelir.
  • Performans daha düşüktür ama yatayda ölçeklendirme imkanı tanır.

Eğer ki, InProcessExecution kullanılmasaydı workflow ilgili noktada ve aynı anda değil, scheduler üzerinden background runner tarafından çalışacaktı. Böylece çağrı yapıldıktan sonra workflow hemen çalışmayabilecek, farklı bir thread pool/task yapısı içinde işlenebilecek veya step event’ler muhtemelen senkron olarak elde edilemiyor olacaktı. Bu ve bunlara benzer sebeplerden kaynaklı debugger ile süreci adım adım takip etme imkanı azalacak ve Out-of-process yaklaşımının getirisi olarak ciddi performans azalışı söz konusu olacaktı.

InProcessExecution kullandığımızda ise workflow, yazdığımız uygulamanın içinde çalışır hale getirilmiş olmakla birlikte, event’lere anında erişilebilmekte ve memory içi olduğu için geçici ve hız odaklı bir süreç söz konusu olabilmektedir.Sequential Workflow’un Önemli Noktaları
Sequential Workflow; basittir, öngörülebilirdir ve hata ayıklaması oldukça kolaydır. Özellikle yukarıda bahsedildiği gibi birbirine bağımlı görevler için oldukça idealdir. Ancak her adım önceki adımın tamamlanmasını beklediği için uzun süren görevlerde yavaş olabilir.

Concurrent Workflow

Concurrent Workflow, Microsoft Agent Framework’te birden fazla executor’ın aynı anda paralel olarak çalışabildiği bir workflow desenidir. Bu, Sequential Workflow’un tersine, görevlerin sırayla değil, eşzamanlı (concurrent) olarak yürütülmesini sağlamaktadır. Framework bu özelliği fan-out / fan-in (açılım / birleşme) mekanizmasıyla desteklemektedir.

Fan-out / Fan-in Nedir?
Fan-out / Fan-in, concurrent workflow’ün temel mekanizmasıdır. Şöyle ki;
1 mesaj → N tane executor’a dağıtılır (fan-out), hepsi bitince sonuçlar toplanır (fan-in)

WorkflowContext Nedir?
WorkflowContext, executor’lar arasında veri akışını ve iletişimini yöneten context objesidir. Bu objenin ana özellikleri aşağıdaki gibidir;

• Veri Paylaşımı / State Sharing veya Shared State
  Bir AI agent’ın herhangi bir görevi tamamlandığı durumlarda executor’lar, sonuçlarını WorkflowContext üzerinden sonraki adımlara aktarmaktadır. Böylece, workflow’un durumu (state) korunmuş ve executor’ların birbirlerine bağımlı çalışması sağlanmış olacaktır.
• Durum Yönetimi / State Management
  Uzun süren veya insan müdahalesi gerektiren (human-in-the-loop) senaryolarda, thread tabanlı state yönetimi için kullanılmaktadır. Checkpointing ve routing gibi özelliklerle workflow’u kesintiye uğratmadan devam ettirebilir.

Evet… Workflow’ler, başlangıç executor’undan başlayarak edge’ler (koşullu yollar) ile ilerler ve WorkflowContext, bu edge’lerdeki veri akışını koordine eder.

Nasıl çalışmaktadır?

1. Başlangıç mesajı gelir.
2. Mesaj aynı anda n tane executor’a gönderilir (fan-out)
3. Her executor bağımsız olarak çalışır (farklı AI modelleri, farklı analizler vs.)
4. Tüm executor’lar bittiğinde sonuçlar toplanır, birleştirilir ve bir sonraki adıma geçilir (fan-in)

Aşağıdaki örnek çalışmayı inceleyebilirsiniz;

public sealed class ConcurrentStartExecutor() : Executor<string>(nameof(ConcurrentStartExecutor))
{
    public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        await context.SendMessageAsync(new ChatMessage(ChatRole.User, message), cancellationToken);
        await context.SendMessageAsync(new TurnToken(emitEvents: true));
    }
}

public sealed class ConcurrentAggregationExecutor() : Executor<string>(nameof(ConcurrentAggregationExecutor))
{
    private readonly List<ChatMessage> _messages = [];
    public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        _messages.Add(new ChatMessage(ChatRole.User, message));

        if (_messages.Count == 2)
        {
            string aggregatedResult = string.Join(Environment.NewLine, _messages.Select(m => $"{m.AuthorName}: {m.Text}"));
            await context.YieldOutputAsync(aggregatedResult, cancellationToken);
        }
    }
}

Burada görüldüğü üzere fan-out yapabilmek için ConcurrentStartExecutor, aynı şekilde fan-in için de ConcurrentAggregationExecutor executor sınıfları tanımlanmıştır. Bu sınıflar, dikkat ederseniz ReflectingExecutor ile değil de onun base class’ı olan Executor sınıfı ile türetilmişlerdir. Burada reflection’lık bir davranış tercih edilmediği için bu sınıfların kullanıldığını söyleyebilirim.

Ayrıca ConcurrentStartExecutor içerisinde kullandığımız TurnToken‘dan da bahsetmek istiyorum. TurnToken, workflow içerisinde sıra/tur yönetimi için kullanılan özel bir kontrol mesajıdır. Şöyle ki; TurnToken, workflow’da bir ‘tur’un (turn) bittiğini ve sonraki executor’ların çalışmaya başlayabileceğini belirten bir senkronizasyon token’ıdır. Yani, ilgili executor işini bitirdiği taktirde sıranın başka executor’da olduğunu ifade eden ya da işlev açısından bir sonraki adıma geçilebileceğinin sinyalini veren bir aparattır. Bu aparat ile bizler fan-out senaryolarında paralel executor’ların başlamasını tetikleyebilmekteyiz. İşte bundan kaynaklı birazdan workflow’u tasarlarken ConcurrentStartExecutor‘ın kullanıcı mesajını sisteme gönderdikten sonra TurnToken ile işlevinin bittiğini ve sürecin agent’lar/executor’lar tarafından paralel olarak çalışmaya başlayacağını ifade ediyor olacağız. Ee peki ’emitEvents: true’ parametresi de neyin nesi hoca la? dediğinizi duyar gibiyim… Bu parametre de workflow event’lerinin yayınlanıp yayınlanmayacağının kararını vermemizi sağlamakta; true dendiği taktirde tüm logların ve metriklerin yayınlanmasına, false dendiğinde ise sessizce geçiş yapılmasına ve böylece performans açısından küçükte olsa bir optimizasyona yaramaktadır.

Şimdi AI Agent’larımızı oluşturalım;

string key = "sk-or-v1-69337dd***debd44acf50172532d";
string model = "openai/gpt-oss-120b";
string endpoint = "https://openrouter.ai/api/v1";
AIAgent physicist = new OpenAIClient(credential: new ApiKeyCredential(key), options: new OpenAIClientOptions
{
    Endpoint = new Uri(endpoint)
})
    .GetChatClient(model)
    .CreateAIAgent
    (
        name: "Physicist",
        instructions: "Fizik konusunda uzmansın. Sorulara fizik perspektifinden cevap veriyorsun."
    );

AIAgent chemist = new OpenAIClient(credential: new ApiKeyCredential(key), options: new OpenAIClientOptions
{
    Endpoint = new Uri(endpoint)
})
    .GetChatClient(model)
    .CreateAIAgent
    (
        name: "Chemist",
        instructions: "Kimya konusunda uzmansın. Sorulara fizik perspektifinden cevap veriyorsun."
    );

Görüldüğü üzere biri fizik diğeri ise kimya konusunda uzman iki adet agent’ımız mevcuttur. Şimdi bu agent’ların süreçteki fan-out ve fan-in davranışlarını modelleyerek workflow’umuzu oluşturalım.

ConcurrentStartExecutor startExecutor = new();
ConcurrentAggregationExecutor aggregationExecutor = new();

WorkflowBuilder workflowBuilder = new(startExecutor);
var workflow = workflowBuilder
    .AddFanOutEdge
    (
        source: startExecutor,
        targets: [physicist, chemist]
    )
    .AddFanInEdge
    (
        [physicist, chemist],
        aggregationExecutor
    )
    .WithOutputFrom(aggregationExecutor)
    .Build();

Yukarıdaki kodu incelerseniz eğer; AddFanOutEdge metodu ile bir açılım yapılmaktadır. source olarak tanımlanmış olan startExecutor‘ın çıktısı aynı anda physicist ve chemist executor’larına dağıtılmakta ve ikisi de eşzamanlı çalıştırılmaktadır.

AddFanInEdge metodu ile de physicist ve chemist agent’larının işlevi bitene kadar beklenmekte ve sonra her ikisinin çıktıları aggregationExecutor isimli executor’a gönderilmekte ve böylece tüm sonuçlar alınarak, toplanmaktadır.

WithOutputFrom metodu ile ise workflow bittiğinde son cevabın aggregationExecutor‘dan geleceği bildirilmekte, böylece de kullanıcıya gösterilecek cevabın aggregationExecutor‘ın ürettiği özet olduğu ifade edilmektedir.

Artık oluşturulan bu workflow’u aşağıdaki gibi çalıştırabilir ve neticeyi elde edebiliriz:

await using StreamingRun run = await InProcessExecution.StreamAsync(workflow, "Sıcaklık nedir?");

await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    if (workflowEvent is AgentRunUpdateEvent agentRunUpdateEvent)
    {
        Console.Write(agentRunUpdateEvent.Data);
    }
}

Nihai olarak;

Netice itibariyle bu içerik sürecinde, günümüzün çağdaş teknoloji davranışlarında giderek daha fazla benimseneceğini düşündüğüm workflow yaklaşımını Microsoft Agent Framework çerçevesinde ele almış ve multi-agent orkestrasyon ihtiyaçlarını karşılarken agent’lar arası planlamanın nasıl daha kolay ve etkili kılınabileceğini hem teorik hem de pratik olarak incelemiş bulunuyoruz.

İlgilenenlerin faydalanması dileğiyle…
Sonraki yazılarımda görüşmek üzere…
İyi çalışmalar…

Not : Örnek çalışmaya aşağıdaki GitHub adresinden erişebilirsiniz.
https://github.com/gncyyldz/Microsoft_Agent_Framework_Workflows_Implementation