워크플로는 실행기와 에지를 방향성 그래프로 연결하고 실행을 관리합니다. 실행기 호출, 메시지 라우팅 및 이벤트 스트리밍을 조정합니다.
워크플로 빌드
워크플로는 워크플로 구조를 정의하기 위한 흐름 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()
이 패키지는 workflow 에지에 의해 실행기가 연결되는 그래프 기반 실행 모델을 제공합니다.
- 실행기 - 입력을 수신하고 출력을 생성하는 처리 장치
- Edge - 한 실행기의 출력을 다른 실행기의 입력에 연결합니다.
- 작성기 - 실행기 및 에지를 정의하여 워크플로 생성
- 실행 - 지정된 입력을 사용하여 워크플로 실행
import (
"github.com/microsoft/agent-framework-go/workflow"
"github.com/microsoft/agent-framework-go/workflow/inproc"
)
uppercase := workflow.NewExecutor("UppercaseExecutor", func(input string) string {
return strings.ToUpper(input)
}).Bind()
reverse := workflow.NewExecutor("ReverseExecutor", func(input string) string {
runes := []rune(input)
slices.Reverse(runes)
return string(runes)
}).Bind()
wf, err := workflow.NewBuilder(uppercase).
AddEdge(uppercase, reverse).
WithOutputFrom(reverse).
Build()
if err != nil {
return err
}
워크플로 실행
워크플로는 스트리밍 및 비 스트리밍 실행 모드를 모두 지원합니다.
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을 사용하세요:
stream, err := inproc.Default.RunStreaming(context.Background(), wf, "Hello, World!")
if err != nil {
return err
}
defer stream.Close(context.Background())
for evt, err := range stream.WatchStream(context.Background()) {
if err != nil {
return err
}
if output, ok := evt.(workflow.OutputEvent); ok {
fmt.Printf("Workflow completed: %v\n", output.Output)
}
}
워크플로 완료를 기다린 다음 수집된 이벤트를 검사하려는 경우에 사용합니다 Run .
run, err := inproc.Default.Run(context.Background(), wf, "Hello, World!")
if err != nil {
return err
}
for evt := range run.NewEvents() {
if output, ok := evt.(workflow.OutputEvent); ok {
fmt.Printf("Final result: %v\n", output.Output)
}
}
스트리밍이 아닌 실행에서 수집한 실행기 이벤트를 검사할 수도 있습니다.
for evt := range run.NewEvents() {
if evt, ok := evt.(workflow.ExecutorCompletedEvent); ok {
fmt.Printf("%s: %v\n", evt.ExecutorID, evt.Result)
}
}
Tip
실행 가능한 전체 샘플 은 워크플로 예제 를 참조하세요.
워크플로 유효성 검사
프레임워크는 워크플로를 빌드할 때 포괄적인 유효성 검사를 수행합니다.
- 형식 호환성: 메시지 형식이 연결된 실행기 간에 호환되는지 확인합니다.
- 그래프 연결: 시작 실행기에서 모든 실행기에 연결할 수 있는지 확인합니다.
- 실행기 바인딩: 모든 실행기가 올바르게 바인딩되고 인스턴스화되는지 확인합니다.
- 에지 유효성 검사: 중복된 에지 및 잘못된 연결 확인
실행 모델: 슈퍼스텝
프레임워크는 수정된 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 모델은 다음과 같은 중요한 보장을 제공합니다.
- 결정적 실행: 동일한 입력을 감안할 때 워크플로는 항상 동일한 순서로 실행됩니다.
- 신뢰할 수 있는 검사점: 내결함성을 위해 슈퍼스텝 경계에 상태를 저장할 수 있습니다.
- 더 간단한 추론: 슈퍼스텝 사이의 경합 조건 없음; 각각은 메시지의 일관된 보기를 확인합니다.
슈퍼스텝 모델로 작업하기
서로를 차단하지 않는 진정으로 독립적인 병렬 경로가 필요한 경우 순차적 단계를 단일 실행기로 통합합니다. 연쇄적으로 연결하는 대신, 해당 논리를 하나의 실행기로 통합하십시오. 그런 다음 두 병렬 경로는 단일 슈퍼스텝 내에서 실행됩니다.
다음 단계
관련 항목: