创建简单的顺序工作流

概述

在本教程中，你将创建包含两个执行器的工作流：

1. 大写执行程序 - 将输入文本转换为大写
2. 反向文本执行程序 - 反转文本并输出最终结果

工作流演示以下核心概念：

• 创建实现IMessageHandler<TInput, TOutput>的自定义执行程序
• 使用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>
internal sealed class UppercaseExecutor() : ReflectingExecutor<UppercaseExecutor>("UppercaseExecutor"),
    IMessageHandler<string, string>
{
    public ValueTask<string> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken = default)
    {
        // Convert input to uppercase and pass to next executor
        return ValueTask.FromResult(input.ToUpper());
    }
}

要点：

• 从 ReflectingExecutor<T> 继承基本执行器功能
• IMessageHandler<string, string> 实现 - 采用字符串输入，生成字符串输出
• 该方法 HandleAsync 处理输入并返回结果

步骤 3：定义反向文本执行程序

定义反转文本的执行程序：

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

要点：

• 与第一个执行者相同的模式。
• 使用 LINQ 的方法 Reverse() 反转字符串。
• 这是工作流中的最终执行者。

步骤 4：生成和连接工作流

使用 WorkflowBuilder 连接执行程序。

// Create the executors
UppercaseExecutor uppercase = new();
ReverseTextExecutor reverse = new();

// 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;
        case WorkflowOutputEvent workflowOutput:
            Console.WriteLine($"Workflow '{workflowOutput.SourceId}' outputs: {workflowOutput.Data}");
            break;
    }
}

步骤 6：了解工作流输出

运行工作流时，会看到如下所示的输出：

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

输入“Hello, World!” 首先转换为大写（“HELLO, WORLD!”），然后反转（“!DLROW, OLLEH”）。

关键概念解析

执行程序接口

执行程序实现 IMessageHandler<TInput, TOutput>：

• TInput：此执行程序接受的数据类型
• TOutput：此执行程序生成的数据类型
• HandleAsync：处理输入并返回输出的方法

.NET 工作流生成器模式

WorkflowBuilder 提供用于构造工作流的流畅 API。

• 构造函数：接收起始执行器
• AddEdge（）：在执行程序之间创建定向连接
• WithOutputFrom（）：指定执行程序生成工作流输出
• 生成（）：创建最终不可变工作流

.NET 事件类型

在执行期间，可以观察以下事件类型：

• ExecutorCompletedEvent - 执行程序完成处理时
• WorkflowOutputEvent - 包含最终工作流结果（用于流式执行）

运行 .NET 示例

1. 创建新的控制台应用程序
2. 安装Microsoft.Agents.AI.WorkflowsNuGet 包
3. 将上述步骤中的所有代码片段合并到 Program.cs
4. 运行应用程序

工作流将通过执行程序处理输入并显示结果。

完成 .NET 示例

有关完整的现成实现，请参阅 Agent Framework 存储库中的 01_ExecutorsAndEdges示例 。

此示例包括：

• 使用所有 using 语句和类结构的完整实现
• 解释工作流概念的其他注释
• 完成项目设置和配置

概述

在本教程中，你将创建包含两个执行器的工作流：

1. 大写执行程序 - 将输入文本转换为大写
2. 反向文本执行程序 - 反转文本并输出最终结果

工作流演示以下核心概念：

• @executor使用修饰器创建工作流节点
• 将执行器与 WorkflowBuilder 连接
• 在步骤间进行数据传递 ctx.send_message()
• 使用 ctx.yield_output() 生成最终输出
• 用于实时观察的事件流处理

先决条件

• Python 3.10 或更高版本
• 已安装 Agent Framework Core Python 包： pip install agent-framework-core
• 此基本示例不需要外部 AI 服务

分步实现

以下部分演示如何逐步生成顺序工作流。

步骤 1：导入所需的模块

首先，从代理框架导入必要的模块：

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

步骤 2：创建第一个执行程序

使用 @executor 修饰器创建将文本转换为大写的执行程序：

@executor(id="upper_case_executor")
async def to_upper_case(text: str, ctx: WorkflowContext[str]) -> None:
    """Transform the input to uppercase and forward it to the next step."""
    result = text.upper()

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

要点：

• @executor修饰器将此函数注册为工作流节点
• WorkflowContext[str] 指示此执行程序向下游发送字符串
• ctx.send_message() 将数据传递到下一步

步骤 3：创建第二个执行程序

创建一个执行程序来反转文本并生成最终输出：

@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] 指示这是终端执行程序
• ctx.yield_output() 提供最终工作流结果
• 工作流在进入空闲状态时完成

步骤 4：生成工作流

使用 WorkflowBuilder 连接执行程序。

workflow = (
    WorkflowBuilder()
    .add_edge(to_upper_case, reverse_text)
    .set_start_executor(to_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（）：完成并返回不可变工作流对象

运行示例

1. 将上述步骤中的所有代码片段合并到单个 Python 文件中
2. 将其另存为 sequential_workflow.py
3. 运行方式： python sequential_workflow.py

工作流将通过多个执行器处理输入“hello world”，并显示流数据事件。

完整的示例

有关完整的现成实现，请参阅 Agent Framework 存储库中的 sequential_streaming.py示例 。

此示例包括：

• 所有导入和文档环节的完整实施
• 解释工作流概念的其他注释
• 显示预期结果的示例输出