사람 중간 개입(Human-in-the-Loop) — 워크플로를 멈추고 사람 입력 받기

사람 중간 개입(Human-in-the-Loop) — 워크플로를 멈추고 사람 입력 받기

검토나 승인처럼 사람의 판단이 필요한 작업은 워크플로가 알아서 통과하면 안 돼요. 사람 중간 개입(HITL)이 필요한 워크플로는 일시 정지하고, 호출자에게 어떤 입력이 필요한지 알리고, 호출자가 응답을 보내면 이어서 진행해야 해요. 워크플로는 이걸 평범한 이벤트로 지원합니다.

출처: 공식문서

가장 직접적인 패턴: 스텝 한 쌍

한 스텝은 InputRequiredEvent를 돌려줘서 입력을 요청하고, 다른 스텝은 HumanResponseEvent를 소비해요. 호출자는 스트림을 지켜보다가 응답을 같은 핸들러로 다시 보내면 됩니다.

from llama_index.core.workflow import Workflow, step
from llama_index.core.workflow.events import (
    StartEvent, StopEvent, InputRequiredEvent, HumanResponseEvent,
)

class NumberWorkflow(Workflow):
    @step
    async def ask(self, ev: StartEvent) -> InputRequiredEvent:
        return InputRequiredEvent(prefix="Enter a number: ")

    @step
    async def answer(self, ev: HumanResponseEvent) -> StopEvent:
        return StopEvent(result=ev.response)

workflow = NumberWorkflow()
handler = workflow.run()

async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        # 여기는 input()이 될 수도, 웹소켓 응답, 웹 폼 제출 등이 될 수도 있어요
        response = input(event.prefix)
        await handler.send_event(HumanResponseEvent(response=response))

final_result = await handler

워크플로는 HumanResponseEvent가 도착할 때까지 기다려요. 프롬프트나 응답에 더 많은 구조가 필요하면 두 이벤트를 모두 서브클래싱할 수 있어요.

사람 응답 사이에서 멈추고 다시 시작하기

웹 앱에서는 프롬프트를 보는 프로세스가 답을 받는 요청과 같지 않을 때가 많아요. 그럴 땐 프롬프트 이후의 컨텍스트를 스냅샷해 저장해 두고, 응답이 도착하면 복원하면 됩니다. 스냅샷 후 원래 핸들러를 취소해 두는 게 좋아요 — 실행을 다른 요청·프로세스에 의도적으로 넘기는 거라면, 메모리에 남은 실행이 복원된 실행에 전달될 답을 계속 기다리며 낭비되지 않게 말이죠.

handler = workflow.run()
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        await db.save("run-123", json.dumps(handler.ctx.to_dict()))
        await handler.cancel_run()
        break

# 나중에 사람 응답이 도착하면:
response = form_data["response"]
ctx_dict = json.loads(await db.load("run-123"))
restored_ctx = Context.from_dict(workflow, ctx_dict)
handler = workflow.run(ctx=restored_ctx)
await handler.send_event(HumanResponseEvent(response=response))
async for event in handler.stream_events():
    continue
final_result = await handler

wait_for_event로 한 스텝 안에서 기다리기

대안은 ctx.wait_for_event()를 써서 하나의 스텝 안에서 입력을 기다리는 방식이에요. waiter_id는 같은 스텝이 여러 번 기다려야 할 때 쓰고, requirements는 여러 웨이터가 같은 이벤트 타입을 소비할 때 올바른 웨이터에게 응답을 라우팅해야 하는 상황에 씁니다.

wait_for_event는 스텝이 트리거 이벤트나 매칭되는 대기 이벤트를 받을 때마다 그 앞의 모든 코드를 재실행해요. 스텝은 웨이터까지는 항상 최소 한 번 실행되고, 웨이터가 내부 예외를 던져 실행을 일시 중지합니다. 그래서 wait_for_event 호출 앞의 코드는 반복되어도 안전해야 해요. 이런 복잡함 때문에, 이벤트 기반의 분리된 스텝 방식이 일반적으로 권장돼요.

더 알아보기