建立簡單的循序工作流程

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

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

概觀

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

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

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

建立帶有單一處理程序的自訂執行器
從函式建立自訂執行器
用於 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 語句和類別結構的完整實作
說明工作流程概念的其他註解
完成專案設定和配置

概觀

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

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

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

使用 @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)

關鍵點：

裝飾器將此函式註冊為工作流程中的一個節點。
WorkflowContext[str] 表示此執行者透過指定第一個型別參數將字串向下送
ctx.send_message() 將資料傳遞至下一個步驟

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

建立一個執行器，將文字反轉，並從裝飾為 @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)

關鍵點：

WorkflowContext[Never, str] 表示這是一個終端執行器，以指定為第一類型參數 Never 而不發送任何訊息，但以指定為第二類型參數 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

關鍵概念解釋

工作流程前後關聯類型

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

WorkflowContext[str] - 將字串傳送至下一個執行程式
WorkflowContext[Never, str] - 產生類型為字串的工作流程輸出的終端執行程式

事件類型

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

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

Python 工作流程產生器模式

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

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

完整範例

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

此範例包括：

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

後續步驟

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

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-11-21

共用方式為

建立簡單的循序工作流程

概觀

涵蓋概念

先決條件

逐步實施

步驟 1：安裝 NuGet 套件

第 2 步：定義大寫執行器

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

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

步驟 5：執行工作流程

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

關鍵概念解釋

執行程式介面

.NET 工作流程產生器模式

.NET 事件類型

完整的 .NET 範例

概觀

涵蓋概念

先決條件

逐步實施

步驟 1：匯入所需模組

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

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

步驟 4：建立工作流程

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

第 6 步：了解輸出

關鍵概念解釋

工作流程前後關聯類型

事件類型

Python 工作流程產生器模式

完整範例

後續步驟

意見反應

其他資源