Alt İş Akışları

Alt-İş Akışı Oluşturma

C# dilinde, alt iş akışlarını iki şekilde oluşturursunuz:

• Doğrudan bağlama — bir iş akışını doğrudan üst iş akışına yürütücü olarak eklemek için kullanın BindAsExecutor() . Bu, alt iş akışının yerel giriş/çıkış türlerini korur.
• Aracı sarmalama — bir iş akışını aracıya dönüştürmek ve ardından aracıyı üst iş akışına eklemek için kullanın AsAIAgent() . Bu, ana iş akışı ajan tabanlı çalıştırıcılar kullandığında kullanışlıdır.

BindAsExecutor ile Doğrudan Bağlama

BindAsExecutor() uzantı yöntemi, bir iş akışını doğrudan üst iş akışına ExecutorBinding eklenebilen bir iş akışına dönüştürür:

using Microsoft.Agents.AI.Workflows;

// Create executors for the inner workflow
UppercaseExecutor uppercase = new();
ReverseExecutor reverse = new();
AppendSuffixExecutor append = new(" [PROCESSED]");

// Build the inner workflow
var innerWorkflow = new WorkflowBuilder(uppercase)
    .AddEdge(uppercase, reverse)
    .AddEdge(reverse, append)
    .WithOutputFrom(append)
    .Build();

// Bind the inner workflow as an executor
ExecutorBinding subWorkflowExecutor = innerWorkflow.BindAsExecutor("TextProcessingSubWorkflow");

// Build the parent workflow using the sub-workflow executor
PrefixExecutor prefix = new("INPUT: ");
PostProcessExecutor postProcess = new();

var parentWorkflow = new WorkflowBuilder(prefix)
    .AddEdge(prefix, subWorkflowExecutor)
    .AddEdge(subWorkflowExecutor, postProcess)
    .WithOutputFrom(postProcess)
    .Build();

BindAsExecutor ile, alt iş akışının giriş ve çıkış türleri korunur. Üst iş akışı, iletileri alt iş akışının beklediği ve ürettiği gerçek türlere göre yönlendirir.

AsAIAgent ile Aracı Sarmalama

Üst iş akışı aracı tabanlı işleyiciler kullandığında, AsAIAgent() kullanarak iç iş akışını bir aracıya dönüştürün. Ajanı WorkflowBuilder otomatik olarak bir yürütücüye sarar.

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

// Create agents for the inner workflow
AIAgent specialist1 = chatClient.AsAIAgent("You are specialist 1. Analyze the data.");
AIAgent specialist2 = chatClient.AsAIAgent("You are specialist 2. Validate the analysis.");

// Build the inner workflow
var innerWorkflow = new WorkflowBuilder(specialist1)
    .AddEdge(specialist1, specialist2)
    .Build();

// Convert the inner workflow to an agent
AIAgent innerWorkflowAgent = innerWorkflow.AsAIAgent(
    id: "analysis-pipeline",
    name: "Analysis Pipeline",
    description: "A sub-workflow that analyzes and validates data"
);

// Create agents for the parent workflow
AIAgent coordinator = chatClient.AsAIAgent("You are a coordinator. Delegate tasks to the team.");
AIAgent reviewer = chatClient.AsAIAgent("You are a reviewer. Review the final output.");

// Build the parent workflow with the sub-workflow
var parentWorkflow = new WorkflowBuilder(coordinator)
    .AddEdge(coordinator, innerWorkflowAgent)
    .AddEdge(innerWorkflowAgent, reviewer)
    .Build();

Üst iş akışının bakış açısından iç iş akışı tek bir adım olarak çalışır. Koordinatör, dahili olarak specialist1 → specialist2 çalıştıran analiz işlem hattına ileti gönderir ve ardından sonucu gözden geçirene iletir.

Giriş ve Çıkış Türleri

Bir iş akışı alt iş akışı olarak kullanıldığında, iç yürütücülerinin tip protokollerini korur.

ile BindAsExecutor, alt iş akışı yürütücüsü iç iş akışının başlangıç yürütücüsüyle aynı giriş türlerini kabul eder ve iç iş akışının ürettiği çıkış türlerini gönderir. Üst iş akışının kenarları, çıkış türleri alt iş akışının beklenen giriş türleriyle eşleşen yürütücüleri bağlamalı ve alt iş akışının çıkış türleri aşağı akış yürütücülerinin beklenen girişleriyle eşleşmelidir.

AsAIAgent kullanılarak, alt iş akışı bir ajan olarak sarmalanır ve Aracı Yürütücüsü giriş/çıkış sözleşmelerine (string, ChatMessage, IEnumerable<ChatMessage>) uyar.

Çıkış Davranışı

Varsayılan olarak, bir alt iş akışı çıkış oluşturduğunda (aracılığıyla YieldOutputAsync), bu çıkışlar üst iş akışındaki bağlı yürütücülere ileti olarak iletilir. Bu, aşağı akış yürütücülerinin alt iş akışı sonuçlarını işlemesini sağlar.

sınıfı ExecutorOptions bu davranışı denetler:

Etkinleştirildiğinde AutoYieldOutputHandlerResultObject , alt iş akışı çıkışları üst öğesinin iç yönlendirmesini atlar ve doğrudan üst iş akışını çağırana teslim edilir.

var options = new ExecutorOptions
{
    AutoYieldOutputHandlerResultObject = true,
};

ExecutorBinding subWorkflowExecutor = innerWorkflow.BindAsExecutor("SubWorkflow", options);

İstekler ve Yanıtlar

Alt iş akışları , istek ve yanıt mekanizmasını tam olarak destekler. Alt iş akışı içindeki bir yürütücü bir istek gönderdiğinde (örneğin, insan girişi istemek için), WorkflowHostExecutor üst iş akışına RequestInfoEvent iletir; alt iş akışı yürütücüsü kimliği bağlantı noktası kimliğine (örneğin, SubWorkflow.GuessNumber) eklenir.

Bu niteleme, üst iş akışı bir yanıt aldığında yanıtı doğru alt iş akışı örneğine geri yönlendirebilmesini sağlar. Üst iş akışı, alt iş akışı isteklerini diğer isteklerle aynı yanıt mekanizmasını kullanarak işler:

await using StreamingRun handle = await InProcessExecution.RunStreamingAsync(parentWorkflow, input);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    switch (evt)
    {
        case RequestInfoEvent requestInfoEvt:
            // The request may originate from the sub-workflow
            // Handle it and send the response back
            var response = requestInfoEvt.Request.CreateResponse(myResponseData);
            await handle.SendResponseAsync(response);
            break;

        case WorkflowOutputEvent outputEvt:
            Console.WriteLine($"Output: {outputEvt.Data}");
            break;
    }
}

Nasıl Çalışır?

Üst iş akışı bir iletiyi alt iş akışı yürütücüsine yönlendirdiğinde:

1. Giriş teslimi — ileti iç iş akışının başlangıç yürütücüsine iletilir. ile BindAsExecutor, ileti türünün başlangıç yürütücüsunun beklenen türleriyle eşleşmesi gerekir. AsAIAgent ile, iletiler ChatMessage formatına normalleştirilir.
2. İç yürütme — iç iş akışı kendi üst adım döngüsünü çalıştırır.
3. Çıkış koleksiyonu — iç iş akışının çıkış olayları toplanır. ile BindAsExecutorçıkışlar özgün türlerini korur. ile AsAIAgent, çıkışlar aracı yanıt iletilerine dönüştürülür.
4. İstek iletme — İç iş akışında bekleyen istekler varsa, bunlar işlenmek üzere üst iş akışına iletilir (bkz. İstekler ve Yanıtlar).
5. Aşağı akış gönderimi — sonuçta elde edilen iletiler üst iş akışında sonraki yürütücüye gönderilir.

İç iş akışı kendi yürütme bağlamını koruduğundan, durumu üst iş akışından bağımsızdır.

Çok Düzeyli İç İçe Yerleştirme

Alt iş akışları rastgele derinliğe iç içe yerleştirilebilir. Her düzey kendi yürütme bağlamını korur:

// Level 1: Data preparation pipeline
var dataPipeline = new WorkflowBuilder(fetcher)
    .AddEdge(fetcher, cleaner)
    .Build();

AIAgent dataPipelineAgent = dataPipeline.AsAIAgent(
    id: "data-pipeline",
    name: "Data Pipeline"
);

// Level 2: Analysis pipeline (contains the data pipeline)
var analysisPipeline = new WorkflowBuilder(dataPipelineAgent)
    .AddEdge(dataPipelineAgent, analyzer)
    .Build();

AIAgent analysisPipelineAgent = analysisPipeline.AsAIAgent(
    id: "analysis-pipeline",
    name: "Analysis Pipeline"
);

// Level 3: Top-level orchestration
var topWorkflow = new WorkflowBuilder(coordinator)
    .AddEdge(coordinator, analysisPipelineAgent)
    .AddEdge(analysisPipelineAgent, reporter)
    .Build();

Hata İşleme

Bir alt iş akışı başarısız olduğunda, hata üst iş akışına olarak SubworkflowErrorEventyayılır. Üst iş akışı, olay akışı aracılığıyla bu hataları gözlemleyebilir:

await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
    if (evt is SubworkflowErrorEvent subError)
    {
        Console.WriteLine($"Sub-workflow '{subError.ExecutorId}' failed: {subError.Data}");
    }
}

Alt iş akışı işlenmeyen bir özel durumla karşılaşırsa, üst iş akışının yürütülmesi devam eder ancak alt iş akışı yürütücüsü daha fazla ileti işlemeyi durdurur.

Kontrol noktası oluşturma

Üst iş akışında bir denetim noktası alındığında, alt iş akışı aracısının oturum durumu üst yürütücüsü denetim noktası verilerinin bir parçası olarak serileştirilir. Geri yükleme sırasında, oturum durumu seri hale getirilmiş durumdan çıkarılır ve böylece ana iş akışı, alt iş akışının durumunu bozulmadan devam ettirebilir.

CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();

// Run the parent workflow with checkpointing
StreamingRun run = await InProcessExecution
    .RunStreamingAsync(parentWorkflow, input, checkpointManager);

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    // Process events, including those from sub-workflows
}

// Resume from a checkpoint
CheckpointInfo checkpoint = run.Checkpoints[^1];
StreamingRun resumedRun = await InProcessExecution
    .ResumeStreamingAsync(parentWorkflow, checkpoint, checkpointManager);

Alt-İş Akışı Oluşturma

Python'da bir alt iş akışı oluşturmak için bir Workflow'yi bir WorkflowExecutor içine sarıp üst iş akışına eklersiniz.

from agent_framework import WorkflowBuilder, WorkflowExecutor

# Create agents for the inner workflow
specialist1 = client.as_agent(name="Specialist1", instructions="Analyze the data.")
specialist2 = client.as_agent(name="Specialist2", instructions="Validate the analysis.")

# Build the inner workflow
inner_workflow = (
    WorkflowBuilder(start_executor=specialist1)
    .add_edge(specialist1, specialist2)
    .build()
)

# Wrap as an executor
inner_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
)

# Create agents for the parent workflow
coordinator = client.as_agent(name="Coordinator", instructions="Delegate tasks to the team.")
reviewer = client.as_agent(name="Reviewer", instructions="Review the final output.")

# Build the parent workflow with the sub-workflow
parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

Üst iş akışının bakış açısından iç iş akışı tek bir adım olarak çalışır. Koordinatör, dahili olarak specialist1 → specialist2 çalıştıran analiz işlem hattına ileti gönderir ve ardından sonucu gözden geçirene iletir.

WorkflowExecutor Parametreleri

Alt iş akışlarını kapsülleme

Workflow örneklerini, bunları bir üst iş akışına eklemeden önce açıkça bir WorkflowExecutor içinde sarın. Aracılar doğrudan WorkflowBuilder öğesine geçirilebilir, ancak ham Workflow örnekleri bu sarmalayıcıyı gerektirir.

from agent_framework import WorkflowExecutor

inner_workflow_executor = WorkflowExecutor(inner_workflow, id="analysis_pipeline")

parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

Açık sarmalama şunları yapmanızı sağlar:

• Birden çok kenarda referans amacıyla belirli bir işlemci kimliği atayın.
• Aynı WorkflowExecutor örneği grafikte yeniden kullanma.

# Explicit wrapping — create the WorkflowExecutor yourself
inner_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
)

parent_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

Giriş ve Çıkış Türleri

, WorkflowExecutor sarmalanan iş akışından tür imzasını devralır:

• Giriş türleri sarmalanan iş akışının başlangıç yürütücüsü giriş türleriyle (artı olarak SubWorkflowResponseMessage iletilen isteklere verilen yanıtları işlemek için) eşleştirilir.
• Çıkış türleri sarmalanan iş akışının çıkış türleriyle eşleşer. Alt iş akışındaki herhangi bir yürütücü request-response özelliğine sahipse, SubWorkflowRequestMessage çıkış türü olarak da eklenir.

Bu, üst iş akışının kenarlarının, çıkış türleri alt iş akışının beklenen giriş türleriyle eşleşen yürütücülere bağlanması gerektiği anlamına gelir. Benzer şekilde, aşağı akış yürütücülerinin alt iş akışının ürettiği türleri kabul etmesi gerekir:

# The sub-workflow's start executor accepts TextProcessingRequest
# So the parent executor must send TextProcessingRequest
class Orchestrator(Executor):
    @handler
    async def start(self, texts: list[str], ctx: WorkflowContext[TextProcessingRequest]) -> None:
        for text in texts:
            await ctx.send_message(TextProcessingRequest(text=text))

# The sub-workflow yields TextProcessingResult
# So the downstream executor must handle TextProcessingResult
class ResultCollector(Executor):
    @handler
    async def collect(self, result: TextProcessingResult, ctx: WorkflowContext) -> None:
        print(f"Received: {result}")

Çıkış Davranışı

Varsayılan olarak (allow_direct_output=False), bir alt iş akışı üzerinden yield_outputçıkış oluşturduğunda, bu çıkışlar kullanılarak send_messageüst iş akışındaki bağlı yürütücülere ileti olarak iletilir. Bu, aşağı akış yürütücülerinin alt iş akışı sonuçlarını üst grafın bir parçası olarak işlemesini sağlar.

olduğunda allow_direct_output=True, alt iş akışı çıkışları doğrudan üst iş akışının olay akışına teslim edilir. Alt iş akışının çıkışları, üst iş akışının iç yürütücü yönlendirmesini atlayarak üst iş akışının çıkışları haline gelir:

# Outputs go directly to parent's event stream
sub_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
    allow_direct_output=True,
)

# The caller receives sub-workflow outputs directly
async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "output":
        # This output came from the sub-workflow
        print(event.data)

Alt iş akışlarından ara emisyonlar

"intermediate" bir alt iş akışında üretilen olaylar, otomatik olarak üst iş akışının olay akışına iletilir. Bunlar, kapsüllemeyi koruyacak şekilde, başlangıçta bunları yayan iç yürütücüye değil, WorkflowExecutor'ın kendi id öğesine atfedilir. En önemlisi, bu olaylar, üst öğenin kendi "intermediate" veya listelerinde WorkflowExecutor öğesini nasıl belirttiğine bakılmaksızın output_from.

async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "intermediate":
        # Attributed to the WorkflowExecutor id, e.g. "analysis-pipeline"
        print(f"[{event.executor_id}] intermediate: {event.data}")
    elif event.type == "output":
        print(f"Terminal output: {event.data}")

İstekler ve Yanıtlar

Alt iş akışları , istek ve yanıt mekanizmasını tam olarak destekler. Bir alt iş akışı içindeki bir yürütücü ctx.request_info() çağırdığında, WorkflowExecutor isteği yakalar ve ayarına göre propagate_request işleyerek yönetir.

Ana İş Akışında İstekleri Yakalama (Varsayılan)

ile propagate_request=False (varsayılan), alt iş akışından gelen istekler bir SubWorkflowRequestMessage içinde sarmalanıp üst iş akışındaki bağlı yürütücülere gönderilir. Bu, üst yürütücülerin isteği yerel olarak işlemesine olanak tanır:

from agent_framework import (
    SubWorkflowRequestMessage,
    SubWorkflowResponseMessage,
)


class ParentHandler(Executor):
    @handler
    async def handle_request(
        self,
        request: SubWorkflowRequestMessage,
        ctx: WorkflowContext[SubWorkflowResponseMessage],
    ) -> None:
        # Inspect the original request from the sub-workflow
        original_data = request.source_event.data

        # Create and send a response back to the sub-workflow
        response = request.create_response(my_response_data)
        await ctx.send_message(response, target_id=request.executor_id)

yöntemi, create_response() yanıt veri türünün özgün istekten beklenen türle eşleşdiğini doğrular. Türler eşleşmiyorsa, bir TypeError tetiklenir.

İstekleri Dış Arayanlara Yayma

ile propagate_request=True, alt iş akışından gelen istekler standart request_info mekanizma kullanılarak üst iş akışının olay akışına yayılır. Ana iş akışının çağıranı bu istekleri diğer insan-döngülü istekler gibi aynı şekilde ele alır.

sub_workflow_executor = WorkflowExecutor(
    workflow=inner_workflow,
    id="analysis-pipeline",
    propagate_request=True,
)

# Run the parent workflow and handle propagated requests
result = await parent_workflow.run(input_data)
request_info_events = result.get_request_info_events()
if request_info_events:
    responses = {}
    for event in request_info_events:
        # Handle each request (e.g., ask a human)
        responses[event.request_id] = get_human_response(event.data)
    result = await parent_workflow.run(responses=responses)

Nasıl Çalışır?

Üst iş akışı bir iletiyi WorkflowExecutor adresine yönlendirdiğinde:

1. Giriş teslimi — ileti iç iş akışının başlangıç yürütücüsine iletilir. İleti türü, başlangıç yürütücüsunun beklenen giriş türleriyle eşleşmelidir.
2. İç yürütme — iç iş akışı, tamamlanmaya kadar veya dış girişe ihtiyaç duyana kadar kendi üst adım döngüsünü çalıştırır.
3. Çıkış koleksiyonu — İç iş akışının çıkış olayları ayara göre allow_direct_output toplanır ve iletilir.
4. İstek iletme — İç iş akışında bekleyen istekler varsa, bunlar ayara göre propagate_request iletilir (bkz. İstekler ve Yanıtlar).
5. Yanıt birikmesi — WorkflowExecutor yalnızca belirli bir yürütme için beklenen tüm yanıtlar alındığında yanıtları toplar ve alt iş akışını sürdürür.
6. Aşağı akış gönderimi — çıkışlar üst iş akışında bir sonraki yürütücüye gönderilir.

Alt iş akışı kendi iç durumunu üst öğeden bağımsız olarak korur. İletiler yalnızca WorkflowExecutor'yu ebeveyn grafın geri kalanına bağlayan kenarlar üzerinden yönlendirilir; iç içe geçme düzeylerinde herhangi bir ileti yayını yapılmaz.

Çok Düzeyli İç İçe Yerleştirme

Alt iş akışları rastgele derinliğe iç içe yerleştirilebilir. Her düzey kendi yürütme bağlamını korur:

# Level 1: Data preparation pipeline
data_pipeline = (
    WorkflowBuilder(start_executor=fetcher)
    .add_edge(fetcher, cleaner)
    .build()
)

data_pipeline_executor = WorkflowExecutor(data_pipeline, id="data_pipeline")

# Level 2: Analysis pipeline (contains the data pipeline)
analysis_pipeline = (
    WorkflowBuilder(start_executor=data_pipeline_executor)
    .add_edge(data_pipeline_executor, analyzer)
    .build()
)

analysis_pipeline_executor = WorkflowExecutor(analysis_pipeline, id="analysis_pipeline")

# Level 3: Top-level orchestration
top_workflow = (
    WorkflowBuilder(start_executor=coordinator)
    .add_edge(coordinator, analysis_pipeline_executor)
    .add_edge(analysis_pipeline_executor, reporter)
    .build()
)

Hata İşleme

Bir alt iş akışı başarısız olduğunda, hata üst iş akışına yayılır. , WorkflowExecutor alt iş akışından başarısız olayı yakalar ve üst bağlamda bir hata olayına dönüştürür:

async for event in parent_workflow.run(input_data, stream=True):
    if event.type == "error":
        print(f"Sub-workflow failed: {event.details.message}")
    elif event.type == "output":
        print(event.data)

Alt iş akışı işlenmeyen bir özel durumla karşılaşırsa, üst iş akışı alt iş akışının kimliği de dahil olmak üzere özel durum ayrıntılarını içeren bir hata olayı alır.

Kontrol noktası oluşturma

Alt iş akışları denetim noktası oluşturmayı destekler. Ebeveyn iş akışında bir denetim noktası alındığında, WorkflowExecutor, iç iş akışının yürütme ilerleme durumu ve önbelleğe alınmış iletiler de dahil olmak üzere iç durumunu serileştirir. Geri yükleme sırasında bu durum deserileştirilir ve üst iş akışının alt iş akışıyla ilk durumuna devam etmesini sağlar.

from agent_framework import FileCheckpointStorage, WorkflowBuilder

checkpoint_storage = FileCheckpointStorage(storage_path="./checkpoints")

# Build the parent workflow with checkpointing
parent_workflow = (
    WorkflowBuilder(
        start_executor=coordinator,
        checkpoint_storage=checkpoint_storage,
    )
    .add_edge(coordinator, inner_workflow_executor)
    .add_edge(inner_workflow_executor, reviewer)
    .build()
)

# Run with automatic checkpointing
async for event in parent_workflow.run("Analyze the dataset", stream=True):
    if event.type == "output":
        print(event.data)

# Resume from a checkpoint
checkpoints = await checkpoint_storage.list_checkpoints(workflow_name=parent_workflow.name)
async for event in parent_workflow.run(
    checkpoint_id=checkpoints[-1].checkpoint_id,
    checkpoint_storage=checkpoint_storage,
    stream=True,
):
    if event.type == "output":
        print(event.data)

Alt-İş Akışı Oluşturma

Go'da, bir alt iş akışı oluşturmak için bir *workflow.Workflow oluşturur ve bunu inproc.BindSubworkflowAsExecutor ile ana iş akışına bağlarsınız.

package main

import (
    "context"
    "fmt"
    "slices"
    "strings"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func buildParentWorkflow() (*workflow.Workflow, error) {
    uppercase := workflow.NewExecutor("UppercaseExecutor", strings.ToUpper).Bind()
    reverse := workflow.NewExecutor("ReverseExecutor", reverseString).Bind()
    appendSuffix := workflow.NewExecutor("AppendSuffixExecutor", func(input string) string {
        return input + " [PROCESSED]"
    }).Bind()

    textProcessing, err := workflow.NewBuilder(uppercase).
        AddEdge(uppercase, reverse).
        AddEdge(reverse, appendSuffix).
        WithOutputFrom(appendSuffix).
        Build()
    if err != nil {
        return nil, err
    }

    textProcessingExecutor := inproc.BindSubworkflowAsExecutor(
        textProcessing,
        "TextProcessingSubWorkflow",
    )

    prefix := workflow.NewExecutor("PrefixExecutor", func(input string) string {
        return "INPUT: " + input
    }).Bind()
    postProcess := workflow.NewExecutor("PostProcessExecutor", func(input string) string {
        return "[FINAL] " + input + " [END]"
    }).Bind()

    return workflow.NewBuilder(prefix).
        AddEdge(prefix, textProcessingExecutor).
        AddEdge(textProcessingExecutor, postProcess).
        WithOutputFrom(postProcess).
        Build()
}

func reverseString(input string) string {
    runes := []rune(input)
    slices.Reverse(runes)
    return string(runes)
}

func runWorkflow(ctx context.Context, parentWorkflow *workflow.Workflow) error {
    run, err := inproc.Default.RunStreaming(ctx, parentWorkflow, "hello")
    if err != nil {
        return err
    }
    defer run.Close(ctx)

    for event, err := range run.WatchStream(ctx) {
        if err != nil {
            return err
        }
        if output, ok := event.(workflow.OutputEvent); ok {
            fmt.Println(output.Output)
        }
    }
    return nil
}

Bağlı alt iş akışı, üst iş akışının perspektifinden tek bir yürütme birimi olarak çalışır. İletiler bağlantı üzerinden gelir, alt iş akışı kendi iç grafiğini çalıştırır ve alt iş akışının çıktıları ana grafiğe geri yönlendirilir.

Giriş ve Çıkış Türleri

Bağlama, sarmalanan iş akışının protokolünü devralır. Üst iş akışı, çalışma zamanı türleri alt iş akışının kabul ettiği giriş türleriyle eşleşen iletiler gönderebilir ve bağlama, alt iş akışının üretilen çıkış türlerini hem ileti türleri hem de çıkış türleri olarak ortaya çıkarır.

Bu, üst iş akışının kenarlarının, çıktı türleri alt iş akışının kabul ettiği girdilerle eşleşen yürütücüleri bağlaması ve aşağı akıştaki yürütücülerin alt iş akışı tarafından üretilen türleri işlemesi gerektiği anlamına gelir:

type TextProcessingRequest struct {
    Text string
}

type TextProcessingResult struct {
    Text string
}

orchestrator := workflow.NewExecutor("Orchestrator", func(ctx *workflow.Context, texts []string) error {
    for _, text := range texts {
        if err := ctx.SendMessage("", TextProcessingRequest{Text: text}); err != nil {
            return err
        }
    }
    return nil
}).Bind()

collector := workflow.NewExecutor("Collector", func(result TextProcessingResult) {
    fmt.Println(result.Text)
}).Bind()

Çıkış Davranışı

Alt iş akışı bir çıktı ürettiğinde, alt iş akışı bağlaması bu çıktıyı, üst iş akışındaki bağlı yürütücülere bağlamadan gelen bir ileti olarak gönderir. Üst iş akışı, alt iş akışı bağını da WithOutputFrom ile işaretlerse, aynı değer, ExecutorID değeri alt iş akışı bağı kimliği olan bir üst workflow.OutputEvent olarak üretilir.

subWorkflowExecutor := inproc.BindSubworkflowAsExecutor(textProcessing, "TextProcessingSubWorkflow")
postProcess := workflow.NewExecutor("PostProcessExecutor", func(input string) string {
    return "[FINAL] " + input
}).Bind()

parentWorkflow, err := workflow.NewBuilder(subWorkflowExecutor).
    AddEdge(subWorkflowExecutor, postProcess).
    WithOutputFrom(subWorkflowExecutor).
    WithOutputFrom(postProcess).
    Build()

Alt iş akışı içinde oluşturulan özel iş akışı olayları, üst iş akışının olay akışına iletilir. Alt iş akışının kendi başlangıç ve superstep yaşam döngüsü olayları içeride tutulur; böylece ana akış, dışarıya anlam ifade eden olaylara odaklı kalır.

İstekler ve Yanıtlar

Alt iş akışları , istek ve yanıt mekanizmasını destekler. Alt iş akışı içindeki bir yürütücü bir dış istek gönderdiğinde, alt iş akışı bağlaması bağlama kimliğini önceden bağlayarak istek bağlantı noktası kimliğini niteler. Örneğin, ApprovalPort adlı bir alt istek bağlantı noktası, üst iş akışında ApprovalSubWorkflow.ApprovalPort olur.

Alt isteği ana iş akışı üzerinden görünür kılmak için, tam nitelikli kimliğe sahip bir üst RequestPort ekleyin ve isteklerle yanıtları alt iş akışı bağlaması ile bu bağlantı noktası arasında yönlendirin:

import "reflect"

approvalPort := workflow.RequestPort{
    ID:       "ApprovalPort",
    Request:  reflect.TypeFor[string](),
    Response: reflect.TypeFor[bool](),
}

approvalWorkflow, err := workflow.NewBuilder(approvalPort.Bind()).
    Build()
if err != nil {
    return err
}

approvalSubWorkflow := inproc.BindSubworkflowAsExecutor(
    approvalWorkflow,
    "ApprovalSubWorkflow",
)

qualifiedApprovalPort := workflow.RequestPort{
    ID:       "ApprovalSubWorkflow.ApprovalPort",
    Request:  approvalPort.Request,
    Response: approvalPort.Response,
}
qualifiedApproval := qualifiedApprovalPort.Bind()

parentWorkflow, err := workflow.NewBuilder(approvalSubWorkflow).
    AddDirectEdge(approvalSubWorkflow, qualifiedApproval, false, externalRequestOnly).
    AddDirectEdge(qualifiedApproval, approvalSubWorkflow, false, externalResponseOnly).
    Build()

Çağıran taraf, üst iş akışı akışından gelen isteği işler ve yanıtı aynı çalıştırma tutamacı üzerinden geri gönderir. Alt iş akışı bağlaması, alt iş akışına yanıt vermeden önce uygun ön eki kaldırır.

run, err := inproc.Default.RunStreaming(ctx, parentWorkflow, "Approve deployment?")
if err != nil {
    return err
}
defer run.Close(ctx)

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch event := event.(type) {
    case workflow.RequestInfoEvent:
        response, err := event.Request.CreateResponse(true)
        if err != nil {
            return err
        }
        if err := run.SendResponse(ctx, response); err != nil {
            return err
        }
    case workflow.OutputEvent:
        fmt.Println(event.Output)
    }
}

İstek ve yanıt kenarlarını dar tutmak için önkoşulları kullanın:

func externalRequestOnly(msg any) bool {
    _, ok := msg.(*workflow.ExternalRequest)
    return ok
}

func externalResponseOnly(msg any) bool {
    _, ok := msg.(*workflow.ExternalResponse)
    return ok
}

Nasıl Çalışır?

Üst iş akışı bir iletiyi alt iş akışı bağlamasına yönlendirdiğinde:

1. Giriş teslimi — bağlama, alt iş akışının kabul edilen giriş türleriyle eşleşen iletileri kabul eder ve bunları alt iş akışının başlangıç yürütücüsuna sıralar.
2. İç yürütme — alt iş akışı aynı işlem içi yürütme ortamında çalışır ve kendi üst adım döngüsünü korur.
3. Çıkış yönlendirme — alt öğe workflow.OutputEvent değerleri, bağlamadan alt akıştaki üst yürütücülere ileti olarak gönderilir ve bağlama WithOutputFrom içinde listelenmişse üst yürütücü tarafından da üretilir.
4. İstek iletme — alt workflow.RequestInfoEvent istekler, nitelikli bağlantı noktası kimlikleriyle yeniden gönderilir ve üst RequestPort bağlamalar aracılığıyla yönlendirilebilir.
5. Olay yönlendirme — alt iş akışlarına ait özel olaylar üst akışa eklenir. Hatalar, alt iş akışı kimliği kaydedilerek ana workflow.ErrorEvent değerleri olarak görüntülenir.
6. Aşağı akış gönderimi — sonuçta elde edilen iletiler üst iş akışı kenarlarında devam eder.

Alt iş akışı, durumunu ve ileti yönlendirmesini üst iş akışından ayrı tutar. İletiler sınırı yalnızca alt iş akışı bağlayıcısına bağlı kenarlar üzerinden aşar.

Çok Düzeyli İç İçe Yerleştirme

Alt iş akışları rastgele derinliğe iç içe yerleştirilebilir. Her alt iş akışı, kendisini içeren iş akışına eklenmeden önce bağlanır:

fraudCheck, err := workflow.NewBuilder(analyzePatterns).
    AddEdge(analyzePatterns, calculateRiskScore).
    WithOutputFrom(calculateRiskScore).
    Build()
if err != nil {
    return err
}

fraudCheckExecutor := inproc.BindSubworkflowAsExecutor(fraudCheck, "FraudCheck")

payment, err := workflow.NewBuilder(validatePayment).
    AddEdge(validatePayment, fraudCheckExecutor).
    AddEdge(fraudCheckExecutor, chargePayment).
    WithOutputFrom(chargePayment).
    Build()
if err != nil {
    return err
}

paymentExecutor := inproc.BindSubworkflowAsExecutor(payment, "Payment")
shippingExecutor := inproc.BindSubworkflowAsExecutor(shipping, "Shipping")

orderWorkflow, err := workflow.NewBuilder(orderReceived).
    AddEdge(orderReceived, paymentExecutor).
    AddEdge(paymentExecutor, shippingExecutor).
    AddEdge(shippingExecutor, orderCompleted).
    WithOutputFrom(orderCompleted).
    Build()

Hata İşleme

Bir alt iş akışı hata oluşturduğunda, alt iş akışı bağlaması bunu workflow.ErrorEvent olarak üst iş akışına iletir ve SubWorkflowID değerini bağlama kimliğine ayarlar. Üst iş akışı, üst düzey iş akışı hataları için kullandığı aynı olay akışı aracılığıyla bu hataları gözlemleyebilir:

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    switch event := event.(type) {
    case workflow.ErrorEvent:
        if event.SubWorkflowID != "" {
            return fmt.Errorf("sub-workflow %q failed: %w", event.SubWorkflowID, event.Error)
        }
        return event.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", event.ExecutorID, event.Error)
    }
}

Alt olaylar iletilirken oluşan hatalar da, alt iş akışı kimliği eklenmiş olarak ana workflow.ErrorEvent değerlerine dönüştürülür.

Kontrol noktası oluşturma

Alt iş akışları denetim noktası oluşturmayı destekler. Üst iş akışı bir denetim noktası oluşturduğunda, alt iş akışı bağlaması, alt iş akışının denetim noktası yöneticisini ve bekleyen tüm nitelikli yanıt bağlantı noktası eşlemelerini üst iş akışının yürütücü durumu içinde depolar. Geri yükleme sırasında alt iş akışı, bekleyen istekler de dahil olmak üzere iç içe yürütme durumu bozulmadan devam edebilir.

checkpointManager := checkpoint.NewInMemoryManager()
environment := inproc.Default.WithCheckpointing(checkpointManager)

var checkpoints []workflow.CheckpointInfo
run, err := environment.RunStreaming(ctx, parentWorkflow, "hello")
if err != nil {
    return err
}
defer run.Close(ctx)

for event, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if completed, ok := event.(workflow.SuperStepCompletedEvent); ok {
        if completed.CompletionInfo != nil && completed.CompletionInfo.CheckpointInfo != nil {
            checkpoints = append(checkpoints, *completed.CompletionInfo.CheckpointInfo)
        }
    }
}

if len(checkpoints) == 0 {
    return fmt.Errorf("no checkpoints were created")
}

resumedRun, err := environment.ResumeStreaming(ctx, parentWorkflow, checkpoints[len(checkpoints)-1])
if err != nil {
    return err
}
defer resumedRun.Close(ctx)

Alt iş akışında bekleyen bir istek varken bir denetim noktası geri yüklenirse, geri yüklenen üst çalıştırma nitelikli istek bilgisi olayını yeniden yayımlar. Çağıran taraf, yeniden yayımlanmış bu istekten bir yanıt oluşturabilir ve bunu üst çalıştırma tanıtıcısı aracılığıyla geri gönderebilir.