工作流程將執行器與邊緣結合成有方向的圖形，並管理執行作業。 它協調執行者調用、訊息路由及事件串流。
建立工作流程
工作流程是使用類別 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()
在建置時驗證時，處理器可以使用裸容器註解，例如 list、 dict、 tuple或 Sequence。 它可以接受使用對應參數化容器的輸出，例如 list[str]。
workflow 套件提供一種基於圖的執行模型，其中執行器之間以邊相連。
- 執行者（Executor） -接收輸入並產生輸出的處理單元
- Edge - 將一個執行者的輸出連接到另一個執行者的輸入
- 建構器 - 透過定義執行器與邊緣來建構工作流程
- 執行 - 執行給定輸入的工作流程
此函式建立一個不需模型或憑證的文字處理工作流程：
import (
"slices"
"strings"
"github.com/microsoft/agent-framework-go/workflow"
)
func buildWorkflow() (*workflow.Workflow, error) {
uppercase := workflow.NewExecutor(
"UppercaseExecutor", strings.ToUpper,
).Bind()
reverse := workflow.NewExecutor(
"ReverseExecutor", func(input string) string {
runes := []rune(input)
slices.Reverse(runes)
return string(runes)
},
).Bind()
return workflow.NewBuilder(uppercase).
AddEdge(uppercase, reverse).
WithOutputFrom(reverse).
Build()
}
WithOutputFrom(reverse) 將反向執行者的結果作為工作流程輸出公開。 無論輸出指定為何，邊仍會將大寫結果傳送給反向執行者。
將 buildWorkflow 回傳的工作流程傳給下方任一執行函式。 對於輸入 "Hello, World!"，兩者都印出 Workflow output: !DLROW ,OLLEH。
工作流程執行
工作流程支援串流和非串流執行模式：
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。 傳遞呼叫者的上下文，讓取消訊號傳遞到工作流程，迭代結束後檢查錯誤，完成後關閉執行：
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runStreaming(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
}()
for evt, streamErr := range stream.WatchStream(ctx) {
if streamErr != nil {
return streamErr
}
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
return ctx.Err()
}
WatchStream 會在外部請求待處理時等待。 針對需要人工輸入的工作流程，請處理要求事件，或使用 WatchUntilHalt 在暫停時傳回。 請參閱 工作流程事件。
用 Run 來前進到下一個停靠點，然後檢查收集到的事件。 一個運行可以在等待外部輸入時停止，而不僅僅是在工作完成後：
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runToHalt(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()
for evt := range run.NewEvents() {
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
for evt := range run.OutgoingEvents() {
if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
}
}
return ctx.Err()
}
每次呼叫NewEvents() 時，它都會推進其書籤。 第二個迴圈用於以 OutgoingEvents() 重播累積的執行者事件，而非嘗試再次用 NewEvents() 消耗這些事件。
這些函數在觀察結果後會結束其執行。 如果你的應用程式需要發送更多訊息或暫停後繼續，請保持 run 開啟。
請參閱 工作流程事件 以了解完整消費者、輸出分類及暫停語意。
Tip
請參閱 工作流程範例 以取得完整的可執行範例。
工作流程驗證
工作流程建置器會在執行前驗證圖形。 具體檢查依據 SDK 及可用的類型元資料而異：
- 型別相容性：在有元資料可用的情況下，檢查連接執行者間宣告的訊息協定。
- 圖連通性：檢查已註冊的執行者是否可從起始執行器存取。
- 執行器繫結：檢查是否存在必要的執行器註冊。
- 邊緣驗證：根據建置器的規則，檢查是否有無效的連線與重複的邊緣。
Go Build() 會拒絕未解決的綁定、無法存取的註冊執行程式、重複的無條件邊，以及不相容的宣告訊息類型。 條件路由與動態協定仍需在執行階段進行測試。
型別驗證可以建立暫時的執行程式執行個體，以檢查其通訊協定。 如果無法檢查工廠或協定，則跳過該相容性檢查。 成功的建置並不保證執行器建立或每次訊息傳遞都能在執行時成功。 處理由執行建立、事件反覆運算以及工作流程失敗事件所產生的錯誤。
BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) 僅停用孤兒執行者檢定。 其他驗證仍在執行。
執行模型：超步
該框架採用修改過的 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，不如將這個邏輯合併成一個執行者。 這兩條平行路徑會在單一超步內執行。
下一步
相關主題：