휴먼인더루프 (HITL)

휴먼인더루프 (HITL) 흐름을 사용하면 사람이 민감한 도구 호출을 승인하거나 거부할 때까지 에이전트 실행을 일시 중지할 수 있습니다. 도구는 승인이 필요할 때 이를 선언하고, 실행 결과는 대기 중인 승인을 인터럽션(중단 처리)으로 노출하며, RunState를 사용하면 결정이 내려진 뒤 실행을 직렬화하고 재개할 수 있습니다.

승인이 필요한 도구 표시

항상 승인을 요구하려면 needs_approval를 True로 설정하거나, 호출마다 결정하는 비동기 함수를 제공하세요. 이 callable은 실행 컨텍스트, 파싱된 도구 매개변수, 도구 호출 ID를 받습니다.

from agents import Agent, Runner, function_tool


@function_tool(needs_approval=True)
async def cancel_order(order_id: int) -> str:
    return f"Cancelled order {order_id}"


async def requires_review(_ctx, params, _call_id) -> bool:
    return "refund" in params.get("subject", "").lower()


@function_tool(needs_approval=requires_review)
async def send_email(subject: str, body: str) -> str:
    return f"Sent '{subject}'"


agent = Agent(
    name="Support agent",
    instructions="Handle tickets and ask for approval when needed.",
    tools=[cancel_order, send_email],
)

needs_approval는 function_tool, Agent.as_tool, ShellTool, ApplyPatchTool에서 사용할 수 있습니다. 로컬 MCP 서버도 MCPServerStdio, MCPServerSse, MCPServerStreamableHttp의 require_approval를 통해 승인 기능을 지원합니다. 호스티드 MCP 서버는 tool_config={"require_approval": "always"} 및 선택적 on_approval_request 콜백과 함께 HostedMCPTool을 통해 승인을 지원합니다. Shell 및 apply_patch 도구는 인터럽션을 노출하지 않고 자동 승인 또는 자동 거부를 하려는 경우 on_approval 콜백을 받을 수 있습니다.

승인 흐름 동작 방식

모델이 도구 호출을 내보내면 runner가 needs_approval를 평가합니다.
해당 도구 호출에 대한 승인 결정이 이미 RunContextWrapper에 저장되어 있으면(예: always_approve=True에서) runner는 프롬프트 없이 진행합니다. 호출별 승인은 특정 호출 ID 범위로 제한됩니다. 향후 호출을 자동으로 허용하려면 always_approve=True를 사용하세요.
그렇지 않으면 실행이 일시 중지되고 RunResult.interruptions(또는 RunResultStreaming.interruptions)에 agent.name, name, arguments 등의 세부 정보를 포함한 ToolApprovalItem 항목이 들어갑니다.
result.to_state()로 결과를 RunState로 변환하고, state.approve(...) 또는 state.reject(...)를 호출한 뒤(선택적으로 always_approve 또는 always_reject 전달) Runner.run(agent, state) 또는 Runner.run_streamed(agent, state)로 재개합니다.
재개된 실행은 중단된 지점부터 이어서 진행되며, 새로운 승인이 필요하면 이 흐름으로 다시 진입합니다.

예: 일시 중지, 승인, 재개

아래 스니펫은 JavaScript HITL 가이드를 그대로 반영합니다. 도구가 승인을 필요로 하면 일시 중지하고, 상태를 디스크에 저장한 뒤, 다시 로드하여 결정을 수집한 후 재개합니다.

import asyncio
import json
from pathlib import Path

from agents import Agent, Runner, RunState, function_tool


async def needs_oakland_approval(_ctx, params, _call_id) -> bool:
    return "Oakland" in params.get("city", "")


@function_tool(needs_approval=needs_oakland_approval)
async def get_temperature(city: str) -> str:
    return f"The temperature in {city} is 20° Celsius"


agent = Agent(
    name="Weather assistant",
    instructions="Answer weather questions with the provided tools.",
    tools=[get_temperature],
)

STATE_PATH = Path(".cache/hitl_state.json")


def prompt_approval(tool_name: str, arguments: str | None) -> bool:
    answer = input(f"Approve {tool_name} with {arguments}? [y/N]: ").strip().lower()
    return answer in {"y", "yes"}


async def main() -> None:
    result = await Runner.run(agent, "What is the temperature in Oakland?")

    while result.interruptions:
        # Persist the paused state.
        state = result.to_state()
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        STATE_PATH.write_text(state.to_string())

        # Load the state later (could be a different process).
        stored = json.loads(STATE_PATH.read_text())
        state = await RunState.from_json(agent, stored)

        for interruption in result.interruptions:
            approved = await asyncio.get_running_loop().run_in_executor(
                None, prompt_approval, interruption.name or "unknown_tool", interruption.arguments
            )
            if approved:
                state.approve(interruption, always_approve=False)
            else:
                state.reject(interruption)

        result = await Runner.run(agent, state)

    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

이 예제에서 prompt_approval는 input()을 사용하고 run_in_executor(...)로 실행되므로 동기식입니다. 승인 소스가 이미 비동기인 경우(예: HTTP 요청 또는 async 데이터베이스 쿼리) async def 함수를 사용하고 대신 직접 await할 수 있습니다.

승인을 기다리는 동안 출력을 스트리밍하려면 Runner.run_streamed를 호출하고 완료될 때까지 result.stream_events()를 소비한 다음, 위에 표시된 것과 동일한 result.to_state() 및 재개 단계를 따르세요.

리포지토리 패턴 및 코드 예제

스트리밍 승인: examples/agent_patterns/human_in_the_loop_stream.py는 stream_events()를 비운 뒤 대기 중인 도구 호출을 승인하고 Runner.run_streamed(agent, state)로 재개하는 방법을 보여줍니다.
도구로서의 에이전트 승인: Agent.as_tool(..., needs_approval=...)는 위임된 에이전트 작업에 검토가 필요할 때 동일한 인터럽션 흐름을 적용합니다.
Shell 및 apply_patch 도구: ShellTool과 ApplyPatchTool도 needs_approval를 지원합니다. state.approve(interruption, always_approve=True) 또는 state.reject(..., always_reject=True)를 사용해 향후 호출을 위한 결정을 캐시하세요. 자동 결정을 위해서는 on_approval를 제공하고(examples/tools/shell.py 참고), 수동 결정을 위해서는 인터럽션을 처리하세요(examples/tools/shell_human_in_the_loop.py 참고).
로컬 MCP 서버: MCPServerStdio / MCPServerSse / MCPServerStreamableHttp에서 require_approval를 사용해 MCP 도구 호출을 게이트하세요(examples/mcp/get_all_mcp_tools_example/main.py 및 examples/mcp/tool_filter_example/main.py 참고).
호스티드 MCP 서버: HostedMCPTool에서 require_approval를 "always"로 설정해 HITL을 강제하고, 선택적으로 on_approval_request를 제공해 자동 승인 또는 거부를 수행하세요(examples/hosted_mcp/human_in_the_loop.py 및 examples/hosted_mcp/on_approval.py 참고). 신뢰할 수 있는 서버에는 "never"를 사용하세요(examples/hosted_mcp/simple.py).
세션 및 메모리: Runner.run에 세션을 전달하면 승인과 대화 기록이 여러 턴에 걸쳐 유지됩니다. SQLite 및 OpenAI Conversations 세션 변형은 examples/memory/memory_session_hitl_example.py 및 examples/memory/openai_session_hitl_example.py에 있습니다.
Realtime agents: 실시간 데모는 RealtimeSession에서 approve_tool_call / reject_tool_call을 통해 도구 호출을 승인 또는 거부하는 WebSocket 메시지를 노출합니다(서버 측 핸들러는 examples/realtime/app/server.py 참고).

장시간 대기 승인

RunState는 내구성을 갖도록 설계되었습니다. state.to_json() 또는 state.to_string()을 사용해 대기 중인 작업을 데이터베이스나 큐에 저장하고, 나중에 RunState.from_json(...) 또는 RunState.from_string(...)으로 다시 생성하세요. 직렬화된 페이로드에 민감한 컨텍스트 데이터를 저장하고 싶지 않다면 context_override를 전달하세요.

대기 중 작업 버전 관리

승인이 한동안 대기할 수 있다면, 직렬화된 state와 함께 에이전트 정의 또는 SDK에 대한 버전 마커를 저장하세요. 그러면 역직렬화를 해당 코드 경로로 라우팅하여 모델, 프롬프트 또는 도구 정의가 변경될 때 비호환성을 피할 수 있습니다.