工作流程建構器與執行

工作流程將執行者與邊綁定成有向圖並管理執行。它協調執行者調用、訊息路由及事件串流。

建立工作流程

工作流程是使用類別 WorkflowBuilder 建構的，該類別提供用於定義工作流程結構的流暢 API：

using Microsoft.Agents.AI.Workflows;

var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();

// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();

工作流程是透過 WorkflowBuilder 以下類別建構的：

from agent_framework import WorkflowBuilder

processor = DataProcessor()
validator = Validator()
formatter = Formatter()

# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()

workflow 套件提供一種基於圖的執行模型，其中執行器之間以邊相連。

執行者（Executor） -接收輸入並產生輸出的處理單元
Edge - 將一個執行者的輸出連接到另一個執行者的輸入
建構器 - 透過定義執行器與邊緣來建構工作流程
執行 - 執行給定輸入的工作流程

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

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

reverse := workflow.NewExecutor("ReverseExecutor", func(input string) string {
    runes := []rune(input)
    slices.Reverse(runes)
    return string(runes)
}).Bind()

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

工作流程執行

工作流程支援串流和非串流執行模式：

using Microsoft.Agents.AI.Workflows;

// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is ExecutorCompletedEvent executorComplete)
    {
        Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
    }

    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Workflow completed: {outputEvt.Data}");
    }
}

// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
    if (evt is WorkflowOutputEvent outputEvt)
    {
        Console.WriteLine($"Final result: {outputEvt.Data}");
    }
}

# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
    if event.type == "output":
        print(f"Workflow completed: {event.data}")

# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")

當你想在事件發生時立即取得相關事件時，請使用 RunStreaming：

stream, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}
defer stream.Close(context.Background())

for evt, err := range stream.WatchStream(context.Background()) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Workflow completed: %v\n", output.Output)
    }
}

當你想等待工作流程完成後再檢查收集到的事件時，請使用 Run ：

run, err := inproc.Default.Run(context.Background(), wf, "Hello, World!")
if err != nil {
    return err
}

for evt := range run.NewEvents() {
    if output, ok := evt.(workflow.OutputEvent); ok {
        fmt.Printf("Final result: %v\n", output.Output)
    }
}

你也可以檢查非串流運行所收集的執行者事件：

for evt := range run.NewEvents() {
    if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
        fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
    }
}

Tip

請參閱工作流程範例以取得完整的可執行範例。

工作流程驗證

架構在建置工作流程時會執行全面的驗證：

類型相容性：確保訊息類型在連接的執行器之間相容
圖形連線能力：驗證所有執行程式是否可從起始執行程式連線
執行器綁定：確認所有執行器都已正確綁定和實例化
Edge 驗證：檢查重複的 Edge 和無效的連線

執行模型：超步

該框架採用修改過的 Pregel 執行模型——一種基於超步驟處理的批量同步平行（BSP）方法。

超階的運作原理

工作流程執行被組織成離散的超步驟。每個超步：

收集前一超步中所有未處理的訊息
根據邊緣定義將訊息路由至目標執行者
在超步中同時運行所有目標執行器
等待所有執行者完成後才能推進（同步障礙）
** 將執行器發出的新訊息排程於下一次超步

Superstep N:
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│  Collect All    │───▶│  Route Messages │───▶│  Execute All    │
│  Pending        │    │  Based on Type  │    │  Target         │
│  Messages       │    │  & Conditions   │    │  Executors      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                       │
                                                       │ (barrier: wait for all)
┌─────────────────┐    ┌─────────────────┐             │
│  Start Next     │◀───│  Emit Events &  │◀────────────┘
│  Superstep      │    │  New Messages   │
└─────────────────┘    └─────────────────┘

同步障礙

最重要的特徵是超步之間的同步障礙。在單一超步驟內，所有觸發的執行器會同時執行，但工作流程不會進入下一個超步驟，直到每個執行器都完成。

這會影響扇出模式：如果您將多個路徑扇開，一個是由一系列執行單元組成的路徑，另一個是單一長時間運行的執行單元，則在長時間運行的執行單元完成之前，串聯路徑無法繼續推進。

為什麼是 Supersteps？

BSP模式提供了重要的保證：

確定性執行：在相同輸入下，工作流程總是以相同順序執行
可靠的檢查點保存：狀態可儲存在超步邊界，以達成容錯的目的。
更簡單的推理：超步之間沒有競賽條件；每個超步都能看到一致的訊息視圖

使用超階模型

如果你需要真正獨立且不會互相阻塞的平行路徑，可以將連續步驟合併到單一執行者中。與其串接 step1 → step2 → step3，不如將這個邏輯合併成一個執行者。這兩條平行路徑會在單一超步內執行。

下一步

工作流程中的代理

相關主題：

執行者 — 工作流程中的處理單元
邊 — 執行單元之間的連結
事件 — 工作流程可觀察性
狀態管理

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-08-13