이 페이지에서는 Microsoft 에이전트 프레임워크 워크플로 시스템의 HITL(휴먼 인 더 루프) 상호 작용에 대한 개요를 제공합니다. HITL은 워크플로의 요청 및 응답 처리 메커니즘을 통해 수행되며, 이를 통해 실행자는 외부 시스템(예: 인간 연산자)에 요청을 보내고 워크플로 실행을 계속하기 전에 응답을 기다릴 수 있습니다.
Overview
워크플로의 실행기는 워크플로 외부로 요청을 보내고 응답을 기다릴 수 있습니다. 이는 실행기가 휴먼 인 더 루프 상호 작용 또는 기타 비동기 작업과 같은 외부 시스템과 상호 작용해야 하는 시나리오에 유용합니다.
인간 운영자에게 숫자를 추측하도록 요청하고 실행기를 사용하여 추측이 올바른지 여부를 판단하는 워크플로를 빌드해 보겠습니다.
워크플로에서 요청 및 응답 처리 사용
요청 및 응답은 RequestPort라는 특수 형식을 통해 처리됩니다.
A RequestPort 는 실행기가 요청을 보내고 응답을 받을 수 있도록 하는 통신 채널입니다. 실행기가 RequestPort에 메시지를 보낼 때, 요청 포트는 요청의 세부 정보가 포함된 RequestInfoEvent를 생성한다. 외부 시스템은 이러한 이벤트를 수신 대기하고, 요청을 처리하고, 워크플로에 응답을 다시 보낼 수 있습니다. 프레임워크는 원래 요청에 따라 자동으로 응답을 적절한 실행기로 다시 라우팅합니다.
// Create a request port that receives requests of type NumberSignal and responses of type int.
var numberRequestPort = RequestPort.Create<NumberSignal, int>("GuessNumber");
워크플로에 입력 포트를 추가합니다.
JudgeExecutor judgeExecutor = new(42);
var workflow = new WorkflowBuilder(numberRequestPort)
.AddEdge(numberRequestPort, judgeExecutor)
.AddEdge(judgeExecutor, numberRequestPort)
.WithOutputFrom(judgeExecutor)
.Build();
정의 JudgeExecutor 에는 대상 번호가 필요하며 추측이 올바른지 여부를 판단할 수 있어야 합니다. 올바르지 않을 경우, RequestPort을 통해 새 추측을 재요청합니다.
internal enum NumberSignal
{
Init,
Above,
Below,
}
internal sealed class JudgeExecutor() : Executor<int>("Judge")
{
private readonly int _targetNumber;
private int _tries;
public JudgeExecutor(int targetNumber) : this()
{
this._targetNumber = targetNumber;
}
public override async ValueTask HandleAsync(int message, IWorkflowContext context, CancellationToken cancellationToken = default)
{
this._tries++;
if (message == this._targetNumber)
{
await context.YieldOutputAsync($"{this._targetNumber} found in {this._tries} tries!", cancellationToken);
}
else if (message < this._targetNumber)
{
await context.SendMessageAsync(NumberSignal.Below, cancellationToken: cancellationToken);
}
else
{
await context.SendMessageAsync(NumberSignal.Above, cancellationToken: cancellationToken);
}
}
}
Python에서 실행기는 데코레이터를 사용하여 ctx.request_info() 요청을 보내고 응답을 @response_handler 처리합니다.
인간 운영자에게 숫자를 추측하도록 요청하고 실행기를 사용하여 추측이 올바른지 여부를 판단하는 워크플로를 빌드해 보겠습니다.
워크플로에서 요청 및 응답 처리 사용
from dataclasses import dataclass
from agent_framework import (
Executor,
WorkflowBuilder,
WorkflowContext,
handler,
response_handler,
)
@dataclass
class NumberSignal:
hint: str # "init", "above", or "below"
class JudgeExecutor(Executor):
def __init__(self, target_number: int):
super().__init__(id="judge")
self._target_number = target_number
self._tries = 0
@handler
async def handle_guess(self, guess: int, ctx: WorkflowContext[int, str]) -> None:
self._tries += 1
if guess == self._target_number:
await ctx.yield_output(f"{self._target_number} found in {self._tries} tries!")
elif guess < self._target_number:
await ctx.request_info(request_data=NumberSignal(hint="below"), response_type=int)
else:
await ctx.request_info(request_data=NumberSignal(hint="above"), response_type=int)
@response_handler
async def on_human_response(
self,
original_request: NumberSignal,
response: int,
ctx: WorkflowContext[int, str],
) -> None:
await self.handle_guess(response, ctx)
judge = JudgeExecutor(target_number=42)
workflow = WorkflowBuilder(start_executor=judge).build()
데코레이터는 @response_handler 지정된 요청 및 응답 형식에 대한 응답을 처리하는 메서드를 자동으로 등록합니다. 프레임워크는 original_request 및 response 매개 변수의 형식 주석에 따라 들어오는 응답을 올바른 처리기에 매칭합니다.
워크플로는 실행을 일시 중지하고 외부 입력을 기다리는 휴먼 인 더 루프 패턴을 RequestPort지원합니다.
approvalPort := workflow.RequestPort{
ID: "ApprovalPort",
Request: reflect.TypeFor[string](),
Response: reflect.TypeFor[bool](),
}
approval := approvalPort.Bind()
finalize := workflow.NewExecutor("FinalizeExecutor", func(approved bool) string {
if approved {
return "Request approved by the human reviewer"
}
return "Request rejected by the human reviewer"
}).Bind()
wf, err := workflow.NewBuilder(approval).
AddEdge(approval, finalize).
WithOutputFrom(finalize).
Build()
A RequestPort 는 워크플로와 외부 세계 간의 형식화된 요청/응답 채널을 정의합니다. 실행기가 요청 포트에 도달하면 워크플로가 일시 중지되고 외부 요청 이벤트를 내보낸다. 외부 응답이 제공되면 워크플로가 다시 시작됩니다.
요청 및 응답 처리
ARequestPort는 요청을 받으면 RequestInfoEvent를 내보낸다. 이러한 이벤트를 구독하여 워크플로에서 들어오는 요청을 처리할 수 있습니다. 외부 시스템에서 응답을 받으면 응답 메커니즘을 사용하여 워크플로로 다시 보냅니다. 프레임워크는 원래 요청을 보낸 실행기로 응답을 자동으로 라우팅합니다.
await using StreamingRun handle = await InProcessExecution.RunStreamingAsync(workflow, NumberSignal.Init);
await foreach (WorkflowEvent evt in handle.WatchStreamAsync())
{
switch (evt)
{
case RequestInfoEvent requestInputEvt:
// Handle `RequestInfoEvent` from the workflow
int guess = ...; // Get the guess from the human operator or any external system
await handle.SendResponseAsync(requestInputEvt.Request.CreateResponse(guess));
break;
case WorkflowOutputEvent outputEvt:
// The workflow has yielded output
Console.WriteLine($"Workflow completed with result: {outputEvt.Data}");
return;
}
}
Tip
전체 실행 가능한 프로젝트에 대한 전체 샘플을 참조하세요.
실행기는 별도의 구성 요소 없이 요청을 직접 보낼 수 있습니다. 실행기가 ctx.request_info()를 호출하면, 워크플로는 WorkflowEvent를 포함한 type == "request_info"을 발생시킨다. 이러한 이벤트를 구독하여 워크플로에서 들어오는 요청을 처리할 수 있습니다. 외부 시스템에서 응답을 받으면 응답 메커니즘을 사용하여 워크플로로 다시 보냅니다. 프레임워크는 자동으로 응답을 실행자의 @response_handler 메서드로 라우팅합니다.
from collections.abc import AsyncIterable
from agent_framework import WorkflowEvent
async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str, int] | None:
"""Process events from the workflow stream to capture requests."""
requests: list[tuple[str, NumberSignal]] = []
async for event in stream:
if event.type == "request_info":
requests.append((event.request_id, event.data))
# Handle any pending human feedback requests.
if requests:
responses: dict[str, int] = {}
for request_id, request in requests:
guess = ... # Get the guess from the human operator or any external system.
responses[request_id] = guess
return responses
return None
# Initiate the first run of the workflow with an initial guess.
# Runs are not isolated; state is preserved across multiple calls to run.
stream = workflow.run(25, stream=True)
pending_responses = await process_event_stream(stream)
while pending_responses is not None:
# Run the workflow until there is no more human feedback to provide,
# in which case this workflow completes.
stream = workflow.run(stream=True, responses=pending_responses)
pending_responses = await process_event_stream(stream)
Tip
전체 실행 가능한 파일은 이 전체 샘플을 참조하세요.
workflow.RequestInfoEvent수신 대기하고, 요청에서 응답을 만들고, 해당 응답으로 실행을 다시 시작합니다.
run, err := inproc.Default.Run(ctx, wf, "Approve deployment to production?")
if err != nil {
return err
}
var request *workflow.ExternalRequest
for evt := range run.NewEvents() {
if requestEvent, ok := evt.(workflow.RequestInfoEvent); ok {
request = requestEvent.Request
break
}
}
response, err := request.CreateResponse(true)
if err != nil {
return err
}
if _, err := run.Resume(ctx, response); err != nil {
return err
}
for evt := range run.NewEvents() {
if output, ok := evt.(workflow.OutputEvent); ok {
fmt.Println(output.Output)
}
}
Tip
전체 실행 가능한 파일은 휴먼 인 더 루프 샘플을 참조하세요.
휴먼 인 더 루프와 에이전트 오케스트레이션
위에서 설명한 패턴은 RequestPort 사용자 지정 실행기 및 WorkflowBuilder와 함께 작동합니다.
에이전트 오케스트레이션(예: 순차, 동시 또는 그룹 채팅 워크플로)을 사용하는 경우 휴먼 인더 루프 요청/응답 메커니즘을 통해 도구 승인이 수행됩니다.
에이전트는 실행 전에 사용자 승인이 필요한 도구를 사용할 수 있습니다. 에이전트가 승인이 필요한 도구를 호출하려고 하면 워크플로가 일시 중지되고 `RequestInfoEvent` 패턴과 마찬가지로 `RequestPort`를 내보내지만, 이벤트 페이로드에는 사용자 지정 요청 타입 대신 `ToolApprovalRequestContent`(C# 및 Go) 또는 `Content`가 포함된 `type == "function_approval_request"`(Python)가 들어 있습니다.
에이전트가 사용자로부터 추가 정보를 수집하고 계속하기 전에 반복해야 하는 대화형 시나리오의 경우 도구 호출만 승인하거나 거부하는 대신 는 핸드오프 오케스트레이션을 사용합니다. 핸드오프는 기본적으로 대화형입니다. 에이전트가 다른 에이전트에 전달하지 않고 응답하는 경우 컨트롤은 다음 입력을 위해 사용자에게 반환되므로 오케스트레이션 내에서 앞뒤로 멀티 턴이 가능합니다. 순차, 동시 및 그룹 채팅 오케스트레이션은 자유 형식의 사용자 입력을 받기 위해 자체적으로 일시 중지되지 않으므로, 단계 사이에 이러한 제어가 필요할 경우 사용자 지정 RequestPort 워크플로에서 WorkflowBuilder와 함께 사용하세요.
Tip
코드가 포함된 전체 예제는 다음을 참조하세요.
검사점 및 요청
검사점에 대한 자세한 내용은 검사점을 참조하세요.
검사점을 만들 때 보류 중인 요청도 검사점 상태의 일부로 저장됩니다. 검사점에서 복원하면 보류 중인 모든 요청이 RequestInfoEvent 개체로 다시 내보내져서 이를 캡처하고 응답할 수 있습니다. 체크포인트에서 다시 시작하고 checkpoint_id와 responses를 모두 workflow.run(...)에 전달하여 동일한 호출에서 응답을 제공할 수도 있습니다.
복원한 후 다시 내보내는 요청 이벤트를 수신 대기하고 언어에 대해 이전에 표시된 것과 동일한 응답 메커니즘을 통해 응답합니다.