이 페이지에서는 Microsoft 에이전트 프레임워크 워크플로 시스템의 HITL(휴먼 인 더 루프) 상호 작용에 대한 개요를 제공합니다. HITL은 워크플로의 요청 및 응답 처리 메커니즘을 통해 수행되며, 이를 통해 실행자는 외부 시스템(예: 인간 연산자)에 요청을 보내고 워크플로 실행을 계속하기 전에 응답을 기다릴 수 있습니다.
개요
워크플로의 실행기는 워크플로 외부로 요청을 보내고 응답을 기다릴 수 있습니다. 이는 실행기가 휴먼 인 더 루프 상호 작용 또는 기타 비동기 작업과 같은 외부 시스템과 상호 작용해야 하는 시나리오에 유용합니다.
인간 운영자에게 숫자를 추측하도록 요청하고 실행기를 사용하여 추측이 올바른지 여부를 판단하는 워크플로를 빌드해 보겠습니다.
워크플로에서 요청 및 응답 처리 사용
요청 및 응답은 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 는 워크플로와 외부 세계 간의 형식화된 요청/응답 채널을 정의합니다. 실행기가 요청 포트에 도달하면 워크플로가 일시 중지되고 외부 요청 이벤트를 내보낸다. 외부 응답이 제공되면 워크플로가 다시 시작됩니다.
요청 및 응답 처리
요청을 RequestPort 받으면 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;
}
}
팁 (조언)
전체 실행 가능한 프로젝트에 대한 전체 샘플을 참조하세요.
실행기는 별도의 구성 요소 없이 요청을 직접 보낼 수 있습니다. 실행기가 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)
팁 (조언)
전체 실행 가능한 파일은 이 전체 샘플을 참조하세요.
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)
}
}
팁 (조언)
전체 실행 가능한 파일은 휴먼 인 더 루프 샘플을 참조하세요.
휴먼 인 더 루프와 에이전트 오케스트레이션
위에서 설명한 패턴은 RequestPort 사용자 지정 실행기 및 WorkflowBuilder와 함께 작동합니다.
에이전트 오케스트레이션(예: 순차, 동시 또는 그룹 채팅 워크플로)을 사용하는 경우 휴먼 인더 루프 요청/응답 메커니즘을 통해 도구 승인이 수행됩니다.
에이전트는 실행 전에 사용자 승인이 필요한 도구를 사용할 수 있습니다. 에이전트가 승인이 필요한 도구를 호출하려고 하면 워크플로가 일시 중지되고 `RequestInfoEvent` 패턴과 마찬가지로 `RequestPort`를 내보내지만, 이벤트 페이로드에는 사용자 지정 요청 타입 대신 `ToolApprovalRequestContent`(C# 및 Go) 또는 `Content`가 포함된 `type == "function_approval_request"`(Python)가 들어 있습니다.
에이전트가 사용자로부터 추가 정보를 수집하고 계속하기 전에 반복해야 하는 대화형 시나리오의 경우 도구 호출만 승인하거나 거부하는 대신 는 핸드오프 오케스트레이션을 사용합니다. 핸드오프는 기본적으로 대화형입니다. 에이전트가 다른 에이전트에 전달하지 않고 응답하는 경우 컨트롤은 다음 입력을 위해 사용자에게 반환되므로 오케스트레이션 내에서 앞뒤로 멀티 턴이 가능합니다. 순차, 동시 및 그룹 채팅 오케스트레이션은 자유 형식의 사용자 입력을 받기 위해 자체적으로 일시 중지되지 않으므로, 단계 사이에 이러한 제어가 필요할 경우 사용자 지정 RequestPort 워크플로에서 WorkflowBuilder와 함께 사용하세요.
팁 (조언)
코드가 포함된 전체 예제는 다음을 참조하세요.
검사점 및 요청
검사점에 대한 자세한 내용은 검사점을 참조하세요.
검사점을 만들 때 보류 중인 요청도 검사점 상태의 일부로 저장됩니다. 검사점에서 복원하면 보류 중인 모든 요청이 RequestInfoEvent 개체로 다시 내보내져서 이를 캡처하고 응답할 수 있습니다. 체크포인트에서 다시 시작하고 checkpoint_id와 responses를 모두 workflow.run(...)에 전달하여 동일한 호출에서 응답을 제공할 수도 있습니다.
복원한 후 다시 내보내는 요청 이벤트를 수신 대기하고 언어에 대해 이전에 표시된 것과 동일한 응답 메커니즘을 통해 응답합니다.