Yürütücüler

Temel Yürütücü Yapısı

Yürütücüler Executor temel sınıfından türetilir ve [MessageHandler] özniteliğini işleyici yöntemlerini bildirmek için kullanır. Kaynak oluşturmayı etkinleştirmek için sınıfı işaretlenmelidir partial .

using Microsoft.Agents.AI.Workflows;

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        return ValueTask.FromResult(result); // Return value is automatically sent to connected executors
    }
}

Ayrıca, bir değer döndürmeden iletileri el ile de gönderebilirsiniz:

internal sealed partial class UppercaseExecutor() : Executor("UppercaseExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        string result = message.ToUpperInvariant();
        await context.SendMessageAsync(result); // Manually send messages to connected executors
    }
}

Birden Çok Giriş Türü

Birden çok yöntem tanımlayarak birden çok [MessageHandler] giriş türünü işleyebilir:

internal sealed partial class SampleExecutor() : Executor("SampleExecutor")
{
    [MessageHandler]
    private ValueTask<string> HandleStringAsync(string message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message.ToUpperInvariant());
    }

    [MessageHandler]
    private ValueTask<int> HandleIntAsync(int message, IWorkflowContext context)
    {
        return ValueTask.FromResult(message * 2);
    }
}

Function-Based Yürütücüleri

Uzantı yöntemini kullanarak BindExecutor bir işlevden yürütücü oluşturun:

Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

IWorkflowContext Nesnesi

, IWorkflowContext yürütme sırasında iş akışıyla etkileşime yönelik yöntemler sağlar:

• SendMessageAsync — bağlı yürütücülere ileti gönderme
• YieldOutputAsync — çağırana döndürülen/akışa alınan iş akışı çıkışlarını üretin

internal sealed partial class OutputExecutor() : Executor("OutputExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.YieldOutputAsync("Hello, World!");
    }
}

İşleyici ne ileti gönderirse ne de çıkış verirse, yalnızca yan etkiler gerçekleştirebilir:

internal sealed partial class LogExecutor() : Executor("LogExecutor")
{
    [MessageHandler]
    private void Handle(string message, IWorkflowContext context)
    {
        Console.WriteLine("Doing some work...");
    }
}

Temel Yürütücü Yapısı

Yürütücüler Executor temel sınıfından devralır. Her bir yürütücü, @handler dekoratör ile süslenmiş yöntemleri kullanır. İşleyiciler, işledikleri ileti türlerini belirtmek için uygun tür açıklamalarına sahip olmalıdır.

from agent_framework import (
    Executor,
    WorkflowContext,
    handler,
)

class UpperCase(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        """Convert the input to uppercase and forward it to the next node."""
        await ctx.send_message(text.upper())

Function-Based Yürütücüleri

@executor dekoratörünü kullanarak bir fonksiyondan yürütücü oluşturun.

from agent_framework import (
    WorkflowContext,
    executor,
)

@executor(id="upper_case_executor")
async def upper_case(text: str, ctx: WorkflowContext[str]) -> None:
    """Convert the input to uppercase and forward it to the next node."""
    await ctx.send_message(text.upper())

Birden Çok Giriş Türü

Birden çok işleyici tanımlayarak birden çok giriş türünü işleyebilir:

class SampleExecutor(Executor):

    @handler
    async def to_upper_case(self, text: str, ctx: WorkflowContext[str]) -> None:
        await ctx.send_message(text.upper())

    @handler
    async def double_integer(self, number: int, ctx: WorkflowContext[int]) -> None:
        await ctx.send_message(number * 2)

Açık Tür Parametreleri

Tür ek açıklamalarına alternatif olarak, türleri dekoratör parametreleri aracılığıyla açıkça belirtebilirsiniz:

class ExplicitTypesExecutor(Executor):

    @handler(input=str, output=str)
    async def to_upper_case(self, text, ctx) -> None:
        await ctx.send_message(text.upper())

    @handler(input=str | int, output=str)
    async def handle_mixed(self, message, ctx) -> None:
        await ctx.send_message(str(message).upper())

    @handler(input=str, output=int, workflow_output=bool)
    async def process_with_workflow_output(self, message, ctx) -> None:
        await ctx.send_message(len(message))
        await ctx.yield_output(True)

WorkflowContext Nesnesi

, WorkflowContext yürütme sırasında iş akışıyla etkileşime yönelik yöntemler sağlar:

• send_message — bağlı yürütücülere ileti gönderme
• yield_output — çağırana döndürülen/akışa alınan iş akışı çıkışlarını üretin

class OutputExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[Never, str]) -> None:
        await ctx.yield_output("Hello, World!")

İşleyici ileti göndermez veya çıkış vermezse, tür parametresi gerekmez:

class LogExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext) -> None:
        print("Doing some work...")

Terminal ve Ara Çıkış Yürütücülerini Belirleme

İş akışının nihai yanıtına hangi yürütücülerin katkıda bulunduğu ve hangilerinin gözlemsel ilerleme olayları yayımladığı, yayım başına bir bayrak değil, üzerinde yapılandırılan bir WorkflowBuilder kararıdır.

• output_from — ctx.yield_output(...) çağrıları "output" olayları üreten ve WorkflowRunResult.get_outputs() tarafından döndürülen yürütücüler.
• intermediate_output_from — ctx.yield_output(...) çağrıları "intermediate" olayları üreten ve WorkflowRunResult.get_intermediate_outputs() tarafından döndürülen yürütücüler.

from agent_framework import WorkflowBuilder

workflow = WorkflowBuilder(
    start_executor=analysis_executor,
    output_from=[summary_executor],
    intermediate_output_from=[analysis_executor],
).build()

Her iki liste de isteğe bağlıdır. Çıkış seçimi listelerinden biri belirtilmişse, listelerin hiçbirinde yer almayan bir yürütücü yine de ctx.send_message(...) aracılığıyla alt akıştaki yürütücülere mesaj gönderebilir, ancak yield_output çağrıları gizlenir. Her iki liste de belirtilmezse, uyumluluk için her yield_output yine de "output" oluşturmaya devam eder.

Temel Yürütücü Yapısı

Yürütücüler bir iş akışındaki işleme birimleridir. Giriş alır, iş gerçekleştirir ve çıkış üretir.

Birden Çok Giriş Türü

Yürütücüde yolları yapılandırarak birden çok işleyici kaydedin:

sample := (&workflow.Executor{
    ID: "SampleExecutor",
    ConfigureProtocol: func(pb *workflow.ProtocolBuilder) (*workflow.ProtocolBuilder, error) {
        pb.RouteBuilder.
            AddHandlerRaw(reflect.TypeFor[string](), reflect.TypeFor[string](), func(_ *workflow.Context, msg any) (any, error) {
                return strings.ToUpper(msg.(string)), nil
            }).
            AddHandlerRaw(reflect.TypeFor[int](), reflect.TypeFor[int](), func(_ *workflow.Context, msg any) (any, error) {
                return msg.(int) * 2, nil
            })
        return pb, nil
    },
}).Bind()

Function-Based Yürütücüleri

Yürütücü oluşturmanın en basit yolu iledir workflow.NewExecutor(...).Bind():

uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
    return strings.ToUpper(input)
}).Bind()

İşlev yürütücüleri giriş türünü otomatik olarak kaydeder ve döndürülen değerleri otomatik olarak gönderebilir ve otomatik olarak verebilir.

İş akışı. Bağlam Nesnesi

İşleyiciler, yürütme sırasında iş akışıyla etkileşim kurmak için *workflow.Context kabul edebilir:

output := workflow.NewExecutor("OutputExecutor", func(ctx *workflow.Context, message string) error {
    return ctx.YieldOutput("Hello, World!")
}).Bind()

Bağlam, SendMessage, AddEvent, PostRequest, ReadState ve QueueStateUpdate gibi API'leri de sunar.

Ajan Yürütücüleri

Aracılar, agentworkflow.New aracılığıyla iş akışı yürütücüleri olarak kullanılabilir:

agentExecutor := agentworkflow.New(myAgent, agentworkflow.Config{
    EmitUpdateEvents: true,
})

Yürütücü Yaşam Döngüsü

Yürütücüler, workflow.Executor üzerindeki alanlar aracılığıyla yaşam döngüsü kancalarını destekler:

stateful := workflow.NewExecutor("StatefulExecutor", handleMessage).Extend(&workflow.Executor{
    InitializeFunc: func(ctx *workflow.Context) error {
        return nil
    },
    ResetFunc: func() error {
        return nil
    },
    OnCheckpointFunc: func(ctx *workflow.Context) error {
        return ctx.QueueStateUpdate("StatefulExecutorState", "", currentState)
    },
    OnCheckpointRestoredFunc: func(ctx *workflow.Context) error {
        restored, err := ctx.ReadState("StatefulExecutorState", "")
        if err != nil {
            return err
        }
        currentState = restored
        return nil
    },
}).Bind()