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

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

순차 워크플로는 복잡한 AI 에이전트 시스템을 구축하는 기초입니다. 이 자습서에서는 각 단계가 데이터를 처리하고 다음 단계로 전달하는 간단한 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 문과 클래스 구조를 포함한 전체 구현
워크플로 개념을 설명하는 추가 설명
프로젝트 설정 및 구성 완료

개요

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

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

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

데코레이터를 @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 이 함수를 워크플로 노드로 등록합니다.
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단계: 워크플로 실행

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

주요 개념 설명

실행기 인터페이스

.NET 워크플로 작성기 패턴

.NET 이벤트 유형

전체 .NET 예제

개요

다루는 개념

필수 조건

단계별 구현

1단계: 필수 모듈 가져오기

2단계: 첫 번째 실행기 만들기

3단계: 두 번째 실행기 만들기

4단계: 워크플로 빌드

5단계: 스트리밍을 사용하여 워크플로 실행

6단계: 출력 이해

주요 개념 설명

워크플로 컨텍스트 형식

이벤트 유형

Python 워크플로 작성기 패턴

완성된 예시

다음 단계

피드백

추가 리소스