建立簡單的循序工作流程

本教學課程示範如何使用代理程式架構工作流程建立簡單的循序工作流程。

順序工作流程是建立複雜人工智慧代理系統的基礎。本教學課程示範如何建立簡單的兩步驟工作流程，其中每個步驟都會處理資料並將其傳遞至下一個步驟。

概觀

在本教學課程中，您將建立具有兩個執行程式的工作流程：

大寫執行器 - 將輸入文字轉換為大寫
反轉文字執行器 - 反轉文字並輸出最終結果

工作流程示範核心概念，例如：

建立帶有單一處理程序的自訂執行器
從函式建立自訂執行器
用於 WorkflowBuilder 將執行程式與邊緣連接起來
透過循序步驟處理資料
透過事件監控工作流程執行

涵蓋概念

先決條件

.NET 8.0 SDK 或更新版本
此基本範例不需要外部 AI 服務
新的主控台應用程式

逐步實施

下列各節說明如何逐步建置循序工作流程。

步驟 1：安裝 NuGet 套件

首先，安裝 .NET 專案所需的套件：

dotnet add package Microsoft.Agents.AI.Workflows --prerelease

第 2 步：定義大寫執行器

定義將文字轉換為大寫的執行器：

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Agents.AI.Workflows;

/// <summary>
/// First executor: converts input text to uppercase.
/// </summary>
Func<string, string> uppercaseFunc = s => s.ToUpperInvariant();
var uppercase = uppercaseFunc.BindExecutor("UppercaseExecutor");

關鍵點：

建立一個函式，將字串取回大寫版本
用 BindExecutor() 來從函式建立執行者

第 3 步：定義反向文字執行器

定義反轉文字的執行程式：

/// <summary>
/// Second executor: reverses the input text and completes the workflow.
/// </summary>
internal sealed class ReverseTextExecutor() : Executor<string, string>("ReverseTextExecutor")
{
    public override ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Reverse the input text
        return ValueTask.FromResult(new string(input.Reverse().ToArray()));
    }
}

ReverseTextExecutor reverse = new();

關鍵點：

建立一個繼承Executor<TInput, TOutput>的類別
實作 HandleAsync() 以處理輸入並回傳輸出

步驟 4：建立並連接工作流程

使用 WorkflowBuilder 連接執行單元

// Build the workflow by connecting executors sequentially
WorkflowBuilder builder = new(uppercase);
builder.AddEdge(uppercase, reverse).WithOutputFrom(reverse);
var workflow = builder.Build();

關鍵點：

WorkflowBuilder 建構函式會採用起始執行程式
AddEdge() 建立從大寫到反向的定向連接
WithOutputFrom() 指定哪些執行程式會產生工作流程輸出
Build() 建立不可變的工作流程

步驟 5：執行工作流程

執行工作流程並觀察結果：

// Execute the workflow with input data
await using Run run = await InProcessExecution.RunAsync(workflow, "Hello, World!");
foreach (WorkflowEvent evt in run.NewEvents)
{
    switch (evt)
    {
        case ExecutorCompletedEvent executorComplete:
            Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
            break;
    }
}

第六步：理解工作流程的輸出

當您執行工作流程時，您會看到如下的輸出：

UppercaseExecutor: HELLO, WORLD!
ReverseTextExecutor: !DLROW ,OLLEH

輸入 “Hello， World！” 會先轉換為大寫（“HELLO， WORLD！”），然後反轉（“！DLROW ，OLLEH“）。

關鍵概念解釋

執行程式介面

從函數中取得執行緒：

使用 BindExecutor() 從函式創建執行者

執行器實作 Executor<TInput, TOutput>：

TInput：此執行器接受的資料類型
TOutput：此執行器產生的資料類型
HandleAsync：處理輸入並傳回輸出的方法

.NET 工作流程產生器模式

WorkflowBuilder 提供流暢的 API 來建構工作流程。

建構函式：取得起始執行函式
AddEdge（）：在執行程式之間建立定向連線
WithOutputFrom（）：指定哪些執行程式產生工作流程輸出
Build（）：建立最終不可變的工作流程

.NET 事件類型

在執行期間，您可以觀察到以下事件類型：

ExecutorCompletedEvent - 當執行程式完成處理時

完整的 .NET 範例

如需完整的、隨時可執行的實作，請參閱代理程式架構存放庫中的 01_ExecutorsAndEdges範例。

此範例包括：

使用所有 using 語句和類別結構的完整實作
說明工作流程概念的其他註解
完成專案設定和配置

概觀

在本教學課程中，您將建立具有兩個執行程式的工作流程：

大寫執行器 - 將輸入文字轉換為大寫
反轉文字執行器 - 反轉文字並輸出最終結果

工作流程示範核心概念，例如：

定義工作單元（執行節點）的兩種方式：
1. 具備以 @handler 標記的非同步方法、以 Executor 為基底類別的自訂類別
2. 一個獨立的非同步函式，裝飾如下： @executor
將執行程式與 WorkflowBuilder
在步驟之間傳遞資料 ctx.send_message()
產生最終輸出以 ctx.yield_output()
串流事件以實現即時可觀察性

涵蓋概念

先決條件

Python 3.10 或更新版本
已安裝的代理程式架構核心 Python 套件： pip install agent-framework-core --pre
此基本範例不需要外部 AI 服務

逐步實施

下列各節說明如何逐步建置循序工作流程。

步驟 1：匯入所需模組

首先，從代理框架匯入必要的模組：

import asyncio
from typing_extensions import Never
from agent_framework import WorkflowBuilder, WorkflowContext, WorkflowOutputEvent, executor

第 2 步：建立第一個執行器

建立一個執行器，透過實作一個帶有處理方法的執行器，將文字轉換為大寫字母：

class UpperCase(Executor):
    def __init__(self, id: str):
        super().__init__(id=id)

    @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.

        Note: The WorkflowContext is parameterized with the type this handler will
        emit. Here WorkflowContext[str] means downstream nodes should expect str.
        """
        result = text.upper()

        # Send the result to the next executor in the workflow.
        await ctx.send_message(result)

關鍵點：

透過子類化 Executor 讓你可以定義一個具有生命週期掛鉤的命名節點（如有需要）
裝飾者會 @handler 標記非同步方法來完成這項工作
處理器簽名遵循合約：
- 第一個參數是此節點的輸入的數據類型（此處為：text: str）
- 第二個參數為 WorkflowContext[T_Out]，其中 T_Out 是該節點將透過 ctx.send_message() （此處： str）所發出的資料類型
在處理程式中，你通常會計算結果並使用 ctx.send_message(result) 轉發給下游節點

第 3 步：建立第二個執行器

對於簡單的步驟，你可以跳過子分類，定義一個帶有相同簽名模式（類型輸入 + WorkflowContext）的非同步函式，並用 @executor來裝飾。這會建立一個功能完整的節點，可以嵌入到流程中。

@executor(id="reverse_text_executor")
async def reverse_text(text: str, ctx: WorkflowContext[Never, str]) -> None:
    """Reverse the input and yield the workflow output."""
    result = text[::-1]

    # Yield the final output for this workflow run
    await ctx.yield_output(result)

關鍵點：

裝飾器將 @executor 一個獨立的非同步函式轉換成工作流程節點
參數 WorkflowContext 有兩種：
- T_Out = Never：此節點不會向下游節點發送訊息
- T_W_Out = str：此節點產生工作流程輸出，其類型為str
終端節點會產生輸出，使用 ctx.yield_output() 來提供工作流程結果
工作流程在進入閒置狀態（不再需要工作）時完成

步驟 4：建立工作流程

使用 WorkflowBuilder 連接執行單元

upper_case = UpperCase(id="upper_case_executor")

workflow = (
    WorkflowBuilder()
    .add_edge(upper_case, reverse_text)
    .set_start_executor(upper_case)
    .build()
)

關鍵點：

add_edge() 在執行程式之間建立定向連線
set_start_executor() 定義進入點
build() 完成工作流程

步驟 5：使用串流執行工作流程

執行工作流程並即時觀察事件：

async def main():
    # Run the workflow and stream events
    async for event in workflow.run_stream("hello world"):
        print(f"Event: {event}")
        if isinstance(event, WorkflowOutputEvent):
            print(f"Workflow completed with result: {event.data}")

if __name__ == "__main__":
    asyncio.run(main())

第 6 步：了解輸出

當您執行工作流程時，您會看到類似下列事件：

Event: ExecutorInvokedEvent(executor_id=upper_case_executor)
Event: ExecutorCompletedEvent(executor_id=upper_case_executor)
Event: ExecutorInvokedEvent(executor_id=reverse_text_executor)
Event: ExecutorCompletedEvent(executor_id=reverse_text_executor)
Event: WorkflowOutputEvent(data='DLROW OLLEH', source_executor_id=reverse_text_executor)
Workflow completed with result: DLROW OLLEH

關鍵概念解釋

定義執行人的兩種方式

自訂類別（子類別 Executor）：最適合需要生命週期鉤子或複雜狀態時。使用 @handler 裝飾器定義一個非同步方法。
函式導向（@executor 裝飾器）：最適合簡單步驟。定義一個獨立的非同步函式，且具有相同的簽名模式。

兩種方法都使用相同的處理者簽名：

第一個參數：此節點的類型化輸入
第二個參數：a WorkflowContext[T_Out, T_W_Out]

工作流程前後關聯類型

WorkflowContext泛型類型定義了執行程式之間的資料流：

WorkflowContext[T_Out] - 用於向下游節點傳送T_Out類型訊息的節點，透過ctx.send_message()
WorkflowContext[T_Out, T_W_Out] - 用於同時產生以下類型的 T_W_Out 工作流程輸出的節點 ctx.yield_output()
WorkflowContext 若無型別參數，則等同於 WorkflowContext[Never, Never]，意即此節點既不向下游節點發送訊息，也不產生工作流程輸出

事件類型

在串流執行期間，您會觀察到下列事件類型：

ExecutorInvokedEvent - 當執行程式開始處理時
ExecutorCompletedEvent - 當執行程式完成處理時
WorkflowOutputEvent - 包含最終工作流程結果

Python 工作流程產生器模式

WorkflowBuilder 提供流暢的 API 來建構工作流程。

add_edge（）：在執行器之間建立定向連接
set_start_executor（）：定義工作流程進入點
build（）：完成並傳回不可變的工作流程物件

完整範例

完整且可執行的實作，請參閱 Agent Framework 資料庫中的範例。

此範例包括：

完整實施所有匯入和文件
說明工作流程概念的其他註解
顯示預期結果的範例輸出

後續步驟

瞭解如何建立簡單的並行工作流程

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-01-13

共用方式為

建立簡單的循序工作流程

概觀

涵蓋概念

先決條件

逐步實施

步驟 1：安裝 NuGet 套件

第 2 步：定義大寫執行器

第 3 步：定義反向文字執行器

步驟 4：建立並連接工作流程

步驟 5：執行工作流程

第六步：理解工作流程的輸出

關鍵概念解釋

執行程式介面

.NET 工作流程產生器模式

.NET 事件類型

完整的 .NET 範例

概觀

涵蓋概念

先決條件

逐步實施

步驟 1：匯入所需模組

第 2 步：建立第一個執行器

第 3 步：建立第二個執行器

步驟 4：建立工作流程

步驟 5：使用串流執行工作流程

第 6 步：了解輸出

關鍵概念解釋

定義執行人的兩種方式

工作流程前後關聯類型

事件類型

Python 工作流程產生器模式

完整範例

後續步驟

意見反應

其他資源