워크플로는 실행기와 에지를 방향성 그래프로 연결하고 실행을 관리합니다. 실행기 호출, 메시지 라우팅 및 이벤트 스트리밍을 조정합니다.
워크플로 빌드
워크플로는 워크플로 구조를 정의하기 위한 흐름 API를 제공하는 클래스를 사용하여 WorkflowBuilder 생성됩니다.
using Microsoft.Agents.AI.Workflows;
var processor = new DataProcessor();
var validator = new Validator();
var formatter = new Formatter();
// Build workflow
WorkflowBuilder builder = new(processor); // Set starting executor
builder.AddEdge(processor, validator);
builder.AddEdge(validator, formatter);
var workflow = builder.Build();
워크플로는 다음 클래스를 WorkflowBuilder 사용하여 생성됩니다.
from agent_framework import WorkflowBuilder
processor = DataProcessor()
validator = Validator()
formatter = Formatter()
# Build workflow
builder = WorkflowBuilder(start_executor=processor)
builder.add_edge(processor, validator)
builder.add_edge(validator, formatter)
workflow = builder.build()
빌드 시간 유효성 검사 중에 핸들러는 list, dict, tuple, 또는 Sequence와 같은 베어 컨테이너 애너테이션을 사용할 수 있습니다. 예: list[str]와 같은 해당 매개변수화된 컨테이너를 사용하는 출력을 수락할 수 있습니다.
이 패키지는 workflow 에지에 의해 실행기가 연결되는 그래프 기반 실행 모델을 제공합니다.
- 실행기 - 입력을 수신하고 출력을 생성하는 처리 장치
- Edge - 한 실행기의 출력을 다른 실행기의 입력에 연결합니다.
- 작성기 - 실행기 및 에지를 정의하여 워크플로 생성
- 실행 - 지정된 입력을 사용하여 워크플로 실행
이 함수는 모델 또는 자격 증명을 요구하지 않고 텍스트 처리 워크플로를 빌드합니다.
import (
"slices"
"strings"
"github.com/microsoft/agent-framework-go/workflow"
)
func buildWorkflow() (*workflow.Workflow, error) {
uppercase := workflow.NewExecutor(
"UppercaseExecutor", strings.ToUpper,
).Bind()
reverse := workflow.NewExecutor(
"ReverseExecutor", func(input string) string {
runes := []rune(input)
slices.Reverse(runes)
return string(runes)
},
).Bind()
return workflow.NewBuilder(uppercase).
AddEdge(uppercase, reverse).
WithOutputFrom(reverse).
Build()
}
WithOutputFrom(reverse) 는 역방향 실행기 결과를 워크플로 출력으로 노출합니다. 에지는 출력 지정에 관계없이 대문자 결과를 역방향 실행기로 전달합니다.
buildWorkflow에 의해 반환된 워크플로를 아래 실행 함수 중 하나에 전달합니다. 입력 "Hello, World!"의 경우 둘 다 Workflow output: !DLROW ,OLLEH로 인쇄됩니다.
워크플로 실행
워크플로는 스트리밍 및 비 스트리밍 실행 모드를 모두 지원합니다.
using Microsoft.Agents.AI.Workflows;
// Streaming execution — get events as they happen
StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, inputMessage);
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
if (evt is ExecutorCompletedEvent executorComplete)
{
Console.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
}
if (evt is WorkflowOutputEvent outputEvt)
{
Console.WriteLine($"Workflow completed: {outputEvt.Data}");
}
}
// Non-streaming execution — wait for completion
Run result = await InProcessExecution.RunAsync(workflow, inputMessage);
foreach (WorkflowEvent evt in result.NewEvents)
{
if (evt is WorkflowOutputEvent outputEvt)
{
Console.WriteLine($"Final result: {outputEvt.Data}");
}
}
# Streaming execution — get events as they happen
async for event in workflow.run(input_message, stream=True):
if event.type == "output":
print(f"Workflow completed: {event.data}")
# Non-streaming execution — wait for completion
events = await workflow.run(input_message)
print(f"Final result: {events.get_outputs()}")
이벤트가 발생하는 대로 보려면 RunStreaming을 사용하세요. 취소가 워크플로에 도달하도록 호출자의 컨텍스트를 전달하고, 반복이 끝난 후 오류를 확인하고, 마쳤으면 run을 닫습니다.
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runStreaming(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
stream, err := inproc.Default.RunStreaming(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, stream.Close(context.WithoutCancel(ctx)))
}()
for evt, streamErr := range stream.WatchStream(ctx) {
if streamErr != nil {
return streamErr
}
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
return ctx.Err()
}
WatchStream 외부 요청이 보류 중인 동안 대기합니다. 인간의 입력이 필요한 워크플로의 경우, 요청 이벤트를 처리하거나 일시 중지 후 반환하도록 WatchUntilHalt을(를) 설정하세요.
워크플로 이벤트를 참조하세요.
다음 중단 지점으로 이동한 다음 수집된 이벤트를 검사하는 데 Run을(를) 사용하세요. 실행은 작업이 완료되었을 때뿐만 아니라 외부 입력을 기다리는 동안에도 중단될 수 있습니다.
import (
"context"
"errors"
"fmt"
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
func runToHalt(
ctx context.Context, wf *workflow.Workflow,
) (err error) {
run, err := inproc.Default.Run(ctx, wf, "Hello, World!")
if err != nil {
return err
}
defer func() {
err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
}()
for evt := range run.NewEvents() {
switch event := evt.(type) {
case workflow.OutputEvent:
fmt.Printf("Workflow output: %v\n", event.Output)
case workflow.ErrorEvent:
return event.Error
case workflow.ExecutorFailedEvent:
return fmt.Errorf("executor %q failed: %w",
event.ExecutorID, event.Error)
}
}
for evt := range run.OutgoingEvents() {
if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
}
}
return ctx.Err()
}
NewEvents() 호출할 때마다 북마크가 이동합니다. 두 번째 루프는 OutgoingEvents()을(를) 사용하여 NewEvents()(으)로 누적된 실행기 이벤트를 다시 소비하려고 시도하는 대신 해당 이벤트를 재생합니다.
이 기능들은 결과를 확인한 후 실행을 종료합니다. 애플리케이션이 추가 메시지를 보내야 하거나 일시 중지 후 다시 시작해야 하는 경우 실행을 계속 열어 두세요.
전체 소비자, 출력 분류 및 일시 중지 의미 체계에 대한 자세한 내용은 워크플로 이벤트를 참조하세요.
Tip
실행 가능한 전체 샘플 은 워크플로 예제 를 참조하세요.
워크플로 유효성 검사
워크플로 빌더는 실행 전에 그래프의 유효성을 검사합니다. 정확한 검사는 SDK 및 사용 가능한 형식 메타데이터에 따라 달라집니다.
- 형식 호환성: 메타데이터를 사용할 수 있는 연결된 실행기 간에 선언된 메시지 프로토콜을 확인합니다.
- 그래프 연결: 등록된 실행기가 시작 실행기에서 도달 가능한지 확인합니다.
- 실행기 바인딩: 필요한 실행기 등록이 있는지 확인합니다.
- 에지 유효성 검사: 작성기 규칙에 따라 잘못된 연결 및 중복 에지를 확인합니다.
Go Build()은(는) 해결되지 않은 바인딩, 도달할 수 없는 등록된 실행기, 중복된 무조건 에지 및 호환되지 않는 선언된 메시지 유형을 거부합니다. 조건부 경로 및 동적 프로토콜에는 여전히 런타임 테스트가 필요합니다.
유형 검증은 프로토콜을 검사하기 위해 임시 실행기 인스턴스를 생성할 수 있습니다. 팩터리 또는 프로토콜을 검사할 수 없는 경우 해당 호환성 검사를 건너뜁니다. 빌드가 성공한다고 해서 실행기 만들기 또는 모든 메시지 배달이 런타임에 성공한다고 보장하지는 않습니다. run 생성, 이벤트 반복 및 워크플로 실패 이벤트의 오류를 처리합니다.
BuildWithOptions(workflow.BuildOptions{DisableOrphanValidation: true}) 는 고아 실행기 검사만 사용하지 않도록 설정합니다. 다른 유효성 검사는 계속 실행됩니다.
실행 모델: 슈퍼스텝
프레임워크는 수정된 Pregel 실행 모델인 BSP(대량 동기 병렬) 접근 방식과 슈퍼스텝 기반 처리를 사용합니다.
슈퍼스텝 작동 방식
워크플로 실행은 개별 상위 단계로 구성됩니다. 각 슈퍼스텝:
- 이전 슈퍼스텝에서 보류 중인 모든 메시지를 수집합니다.
- 에지 정의에 따라 메시지를 대상 실행기에 라우팅
- 슈퍼스텝 내에서 모든 대상 실행기를 동시에 실행합니다.
- 진행하기 전에 모든 실행기가 완료되기를 기다립니다(동기화 장벽)
- 다음 슈퍼스텝을 위해 실행기에서 내보낸 새 메시지를 큐에 추가합니다.
Superstep N:
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Collect All │───▶│ Route Messages │───▶│ Execute All │
│ Pending │ │ Based on Type │ │ Target │
│ Messages │ │ & Conditions │ │ Executors │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
│ (barrier: wait for all)
┌─────────────────┐ ┌─────────────────┐ │
│ Start Next │◀───│ Emit Events & │◀────────────┘
│ Superstep │ │ New Messages │
└─────────────────┘ └─────────────────┘
동기화 장벽
가장 중요한 특징은 슈퍼스텝 간의 동기화 장벽입니다. 트리거된 실행기는 슈퍼스텝 내에서 동시에 실행할 수 있지만 워크플로는 해당 실행기들이 완료될 때까지 다음 슈퍼스텝으로 진행되지 않습니다.
이는 팬아웃 패턴에 영향을 미칩니다. 여러 경로로 팬아웃하는 경우, 즉, 하나는 실행기 체인을 사용하고 다른 하나는 단일 장기 실행기를 사용하는 경우, 장기 실행기가 완료될 때까지 연결된 경로는 진행할 수 없습니다.
왜 슈퍼스텝인가요?
BSP 모델은 다음과 같은 중요한 보장을 제공합니다.
- 정의된 단계 경계: 슈퍼스텝 중에 내보낸 메시지는 나중에 처리하기 위해 큐에 대기됩니다.
- 체크포인트 경계: 체크포인트 기능이 활성화된 워크플로는 슈퍼스텝 경계에서 실행 상태를 저장합니다.
- 버퍼링된 상태 업데이트: 대기 중인 워크플로 상태 변경 내용은 슈퍼스텝 간에 커밋됩니다.
이러한 보장은 한 슈퍼스텝 내에서 동시 실행기 간의 실행 또는 이벤트 순서를 보장하지 않습니다. 슈퍼스텝 장벽이 모든 경쟁 상태를 방지할 수 있다고 가정하지 말고 애플리케이션 소유 공유 상태에 대한 액세스를 동기화합니다.
슈퍼스텝 모델로 작업하기
서로를 차단하지 않는 진정으로 독립적인 병렬 경로가 필요한 경우 순차적 단계를 단일 실행기로 통합합니다. 연쇄적으로 연결하는 대신, 해당 논리를 하나의 실행기로 통합하십시오. 그런 다음 두 병렬 경로는 단일 슈퍼스텝 내에서 실행됩니다.
다음 단계
관련 항목: