0x38A

상위: Agno

요약

Agno의 HITL(Human-in-the-Loop)은 에이전트 실행을 일시 중지하고 사람의 승인/입력을 대기한 후 재개하는 패턴이다. Agent 수준(도구 단위)과 Workflow 수준(스텝 단위) 모두에서 지원한다.

Agent HITL — 5가지 패턴

패턴	`@tool` 데코레이터	설명
User Confirmation	`requires_confirmation=True`	도구 실행 전 명시적 승인/거부
User Input	`requires_user_input=True`	도구 파라미터를 사용자로부터 수집
Dynamic User Input	—	에이전트가 실행 중 필요에 따라 동적으로 정보 요청
External Tool Execution	`external_execution=True`	에이전트 외부에서 도구 실행 (보안 강화)
Approval/Audit	—	관리자 리뷰 + 영속적 감사 추적

상호 배타: requires_confirmation, requires_user_input, external_execution은 하나의 도구에 동시 사용 불가. 하나만 선택해야 한다.

@tool 데코레이터

from agno.tools import tool

@tool(requires_confirmation=True)
def delete_records(table: str, count: int) -> str:
    """Delete records from database."""
    db.delete(table, count)
    return f"Deleted {count} records from {table}"

Pause/Resume 패턴

# 1. 실행 → 일시 중지
output = agent.run("Delete all user records")

# 2. 요구사항 확인 및 처리
if output.is_paused:
    for req in output.active_requirements:
        if req.needs_confirmation:
            approval = input("Approve? (y/n): ")
            if approval == "y":
                req.confirm()
            else:
                req.reject()
        elif req.needs_user_input:
            value = input(f"Provide {req.parameter_name}: ")
            req.provide_input(value)
        elif req.is_external_tool_execution:
            result = external_system.execute(req.tool_call)
            req.provide_result(result)

# 3. 재개
output = agent.continue_run(run_response=output)

스트리밍 + HITL

for event in agent.run("Do something sensitive", stream=True):
    if event.is_paused:
        for req in event.requirements:
            req.confirm()
        for resumed_event in agent.continue_run(
            run_id=event.run_id,
            requirements=event.requirements,
            stream=True,
        ):
            print(resumed_event.content)
    else:
        print(event.content)

Workflow HITL — 스텝 단위 제어

스텝 단위로 일시 중지 → 사용자 확인/입력 → 재개가 가능하다.

3가지 패턴

패턴	설정	설명
Confirmation	`requires_confirmation=True`	스텝 실행 전 승인/거부
User Input	`requires_user_input=True`	사용자 파라미터 수집 (`user_input_schema`)
Error Handling	`on_error=OnError.pause`	에러 시 일시 중지 → retry/skip

Error Handling

from agno.workflow import Workflow, OnError
from agno.workflow.step import Step

workflow = Workflow(
    steps=[
        Step(
            name="fetch_data",
            executor=unreliable_api_call,
            on_error=OnError.pause,
        ),
        Step(name="process", executor=process_data),
    ],
)

run_output = workflow.run("Fetch and process")
while run_output.is_paused:
    for req in run_output.steps_with_errors:
        print(f"Step '{req.step_name}' failed: {req.error_message}")
        choice = input("Retry or skip? (r/s): ")
        if choice == "r":
            req.retry()
        else:
            req.skip()
    run_output = workflow.continue_run(run_output)

User Input (Schema 기반)

from agno.workflow.step import Step
from agno.workflow.types import UserInputField

Step(
    name="process",
    executor=process_with_params,
    requires_user_input=True,
    user_input_message="Configure processing:",
    user_input_schema=[
        UserInputField(
            name="threshold", field_type="float",
            description="Threshold (0.0-1.0)", required=True,
        ),
        UserInputField(
            name="mode", field_type="str",
            description="Mode: fast/accurate", required=True,
        ),
    ],
)

AgentOS에서의 HITL

프로덕션 환경에서는 HTTP API로 동작한다:

# 1. 에이전트 실행 → status: "paused" 반환
POST /agents/my-agent/run
{"message": "Delete records", "session_id": "abc"}

# 2. 응답에서 tool_calls_requiring_confirmation 확인

# 3. 승인 후 재개
POST /agents/my-agent/continue
{"run_id": "xyz", "confirmed_tool_calls": [...]}

Agno HITL

요약