创建简单的顺序工作流

概述

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

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

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

• 使用一个处理程序创建自定义执行程序
• 从函数创建自定义执行程序
• 使用WorkflowBuilder将执行器与边缘连接
• 通过顺序步骤处理数据
• 通过事件观察工作流执行

涵盖的概念

• 执行器
• Direct Edges
• 工作流生成器
• Events

先决条件

• .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;
    }
}

步骤 6：了解工作流输出

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

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

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

关键概念解析

执行程序接口

来自函数的执行器：

• 使用 BindExecutor() 从函数创建执行器

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

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

.NET 工作流生成器模式

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

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

.NET 事件类型

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

• ExecutorCompletedEvent - 执行程序完成处理时

完成 .NET 示例

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

此示例包括：

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

概述

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

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

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

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

涵盖的概念

• 执行器
• Direct Edges
• 工作流生成器
• Events

先决条件

• Python 3.10 或更高版本
• 已安装 Agent Framework Core 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修饰器将此函数注册为工作流节点
• 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 存储库中的 示例 。

此示例包括：

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