다음을 통해 공유

간단한 순차 워크플로 만들기

이 자습서에서는 에이전트 프레임워크 워크플로를 사용하여 간단한 순차 워크플로를 만드는 방법을 보여 줍니다.

순차 워크플로는 복잡한 AI 에이전트 시스템을 구축하는 기초입니다. 이 자습서에서는 각 단계가 데이터를 처리하고 다음 단계로 전달하는 간단한 2단계 워크플로를 만드는 방법을 보여 줍니다.

개요

이 자습서에서는 다음 두 개의 실행기를 사용하여 워크플로를 만듭니다.

  1. 대문자 실행기 - 입력 텍스트를 대문자로 변환
  2. 역방향 텍스트 실행기 - 텍스트를 반대로 바꾸고 최종 결과를 출력합니다.

워크플로는 다음과 같은 핵심 개념을 보여 줍니다.

  • 하나의 처리기를 사용하여 사용자 지정 실행기 만들기
  • 함수에서 사용자 지정 실행기 만들기
  • 실행기와 에지를 연결하는 데 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;
    }
}

6단계: 워크플로 출력 이해

워크플로를 실행하면 다음과 같은 출력이 표시됩니다.

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

"Hello, World!" 입력이 먼저 대문자로 변환된 다음("HELLO, WORLD!"), 역순으로 변환됩니다("!DLROW ,OLLEH").

주요 개념 설명

실행기 인터페이스

함수의 실행기:

  • 함수에서 실행기를 만드는 데 사용 BindExecutor()

실행기(Executors)는 다음을 구현합니다.Executor<TInput, TOutput>

  • TInput: 이 실행기에서 허용하는 데이터 형식
  • TOutput: 이 실행기가 생성하는 데이터 형식
  • HandleAsync: 입력을 처리하고 출력을 반환하는 메서드입니다.

.NET 워크플로 작성기 패턴

워크플로 WorkflowBuilder 를 생성하기 위한 흐름 API를 제공합니다.

  • 생성자: 시작 실행기를 받습니다.
  • AddEdge(): 실행기 간에 직접 연결을 만듭니다.
  • WithOutputFrom(): 워크플로 출력을 생성하는 실행기를 지정합니다.
  • 빌드(): 변경할 수 없는 최종 워크플로를 만듭니다.

.NET 이벤트 유형

실행하는 동안 다음 이벤트 유형을 관찰할 수 있습니다.

  • ExecutorCompletedEvent - 실행기가 처리를 완료하는 경우

전체 .NET 예제

실행 준비가 완료된 완전한 구현은 Agent Framework 리포지토리의 01_ExecutorsAndEdges 샘플을 참조하세요.

이 샘플에는 다음이 포함됩니다.

  • 모든 using 문과 클래스 구조를 포함한 전체 구현
  • 워크플로 개념을 설명하는 추가 설명
  • 프로젝트 설정 및 구성 완료

개요

이 자습서에서는 다음 두 개의 실행기를 사용하여 워크플로를 만듭니다.

  1. 대문자 실행기 - 입력 텍스트를 대문자로 변환
  2. 역방향 텍스트 실행기 - 텍스트를 반대로 바꾸고 최종 결과를 출력합니다.

워크플로는 다음과 같은 핵심 개념을 보여 줍니다.

  • 작업 단위(실행기 노드)를 정의하는 두 가지 방법:
    1. @handler으로 표시된 비동기 메서드를 사용하여 Executor을 서브클래싱하는 사용자 지정 클래스
    2. 로 데코레이팅된 독립 실행형 비동기 함수 @executor
  • 다음을 사용하여 실행기 연결 WorkflowBuilder
  • 다음을 사용하여 단계 간에 데이터 전달 ctx.send_message()
  • ctx.yield_output()를 사용하여 최종 출력을 생성합니다.
  • 실시간 관찰을 위한 스트리밍 이벤트

다루는 개념

필수 조건

  • 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 사용하면 필요한 경우 수명 주기 후크를 사용하여 명명된 노드를 정의할 수 있습니다.
  • 데코레이터는 @handler 작업을 수행하는 비동기 메서드를 표시합니다.
  • 처리기 서명은 계약을 따릅니다.
    • 첫 번째 매개 변수는 이 노드에 대한 형식화된 입력입니다(여기: text: 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

주요 개념 설명

실행기를 정의하는 두 가지 방법

  1. 사용자 지정 클래스(서브클래싱 Executor): 수명 주기 후크 또는 복잡한 상태가 필요한 경우에 가장 적합합니다. 데코레이터를 사용하여 비동기 메서드를 정의합니다 @handler .
  2. 함수 기반(@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