Human in the Loop

Human in the Loop (HITL)

Human-in-the-loop(HITL)을 사용하면 Agent의 tool 호출을 실행 전에 가로챌 수 있어요. 사람이 각 tool 호출의 파라미터를 실시간으로 확인(confirm), 거부(reject), 또는 **수정(modify)**할 수 있어요. 이메일 전송, 데이터베이스 수정, API 호출 같은 고위험 작업 — 사람이 먼저 행동을 검토하길 원하는 경우 — 에 유용해요.

출처: 문서

본문

HITL은 Agent의 일반 hook 메커니즘의 한 응용이에요. before_tool hook 포인트에 등록된 ConfirmationHook이 모델이 요청한 tool 호출을 실행 전에 가로채고, Agent의 State에서 대화를 다시 써서 그것들을 확인·수정·거부해요.

HITL 시스템은 다음 계층으로 구성돼요:

• ConfirmationHook — 대기 중인 tool 호출에 확인 전략을 적용하는 before_tool hook. confirmation_strategies 매핑은 단일 tool 이름, tool 이름 튜플, 또는 더 구체적인 항목이 없는 tool에 적용되는 와일드카드 "*"를 받아들여요.
• Strategy — tool이 호출되려 할 때 무엇을 할지 결정. 내장 BlockingConfirmationStrategy는 실행을 멈추고 사람에게 묻는다.
• Policy — 언제 물을지 결정. 내장 policy: AlwaysAskPolicy, NeverAskPolicy, AskOncePolicy.
• UI — 사람에게 묻는 데 사용되는 인터페이스. 내장 UI: RichConsoleUI(rich 필요)와 SimpleConsoleUI(stdlib만).

Agent가 tool을 호출하려 할 때 전략이 policy를 확인해요. policy가 묻기로 하면 UI가 tool 이름, 설명, 파라미터로 사람에게 프롬프트를 띄워요. 사람은 다음을 할 수 있어요:

• Confirm (y) — 그대로 실행
• Reject (n) — 실행 건너뛰고 거부 피드백을 LLM에 전달
• Modify (m) — 실행 전에 파라미터 편집

그러면 Agent는 사람의 결정에 따라 계속 진행해요.

info 전략은 모델이 tool 호출에 대해 만든 인자만 볼 수 있어요. tool의 inputs_from_state 매핑을 통해 State에서 주입된 값은 확인에 표시되는 대상에 포함되지 않아요 — 그 주입은 tool 실행 시점에 발생해요.

• 구성 대상: before_tool hook 포인트 아래 ConfirmationHook으로 등록된 Agent 구성 요소
• 주요 클래스: ConfirmationHook, BlockingConfirmationStrategy, AlwaysAskPolicy, AskOncePolicy, NeverAskPolicy, RichConsoleUI, SimpleConsoleUI
• Import path: haystack.hooks.human_in_the_loop
• 패키지명: haystack-ai

Usage ​

Basic setup ​

from typing import Annotated
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.hooks.human_in_the_loop import (
    AlwaysAskPolicy,
    BlockingConfirmationStrategy,
    ConfirmationHook,
    SimpleConsoleUI,
)
from haystack.tools import tool

@tool
def send_email(
    to: Annotated[str, "The recipient email address"],
    subject: Annotated[str, "The email subject line"],
    body: Annotated[str, "The email body"],
) -> str:
    """Send an email to a recipient."""
    return f"Email sent to {to}."

strategy = BlockingConfirmationStrategy(
    confirmation_policy=AlwaysAskPolicy(),
    confirmation_ui=SimpleConsoleUI(),
)
agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-5.4-mini"),
    tools=[send_email],
    hooks={
        "before_tool": [
            ConfirmationHook(confirmation_strategies={"send_email": strategy}),
        ],
    },
)
result = agent.run(
    messages=[ChatMessage.from_user("Send a welcome email to [email protected]")],
)

Agent가 send_email을 호출하면 터미널이 멈추고 이렇게 표시돼요:

--- Tool Execution Request ---
Tool: send_email
Description: Send an email to a recipient.
Arguments:
  to: [email protected]
  subject: Welcome!
  body: Hi Alice, welcome aboard!
------------------------------
Confirm execution? (y=confirm / n=reject / m=modify):

Using RichConsoleUI ​

RichConsoleUI는 rich 라이브러리를 사용해 스타일된 터미널 프롬프트를 제공해요:

pip install rich

from haystack.hooks.human_in_the_loop import RichConsoleUI

strategy = BlockingConfirmationStrategy(
    confirmation_policy=AlwaysAskPolicy(),
    confirmation_ui=RichConsoleUI(),
)

Applying strategies to multiple tools ​

tool마다 다른 전략을 구성하거나, tuple 키로 tool 그룹에 전략 하나를 공유하거나, 와일드카드 "*"로 모든 tool(더 구체적인 항목이 없는 tool)에 기본값을 설정할 수 있어요:

@tool
def delete_record(record_id: Annotated[str, "The ID of the record to delete"]) -> str:
    """Delete a record from the database."""
    return f"Record {record_id} deleted."

@tool
def update_record(
    record_id: Annotated[str, "The ID of the record to update"],
    data: Annotated[str, "The new data as a JSON string"],
) -> str:
    """Update a record in the database."""
    return f"Record {record_id} updated."

@tool
def search(query: Annotated[str, "The search query"]) -> str:
    """Search the knowledge base."""
    return f"Results for: {query}"

ask_strategy = BlockingConfirmationStrategy(
    confirmation_policy=AlwaysAskPolicy(),
    confirmation_ui=SimpleConsoleUI(),
)
confirmation_hook = ConfirmationHook(
    confirmation_strategies={
        # Share one strategy across multiple sensitive tools using a tuple key
        ("send_email", "delete_record", "update_record"): ask_strategy,
        # search has no strategy - always executes without asking
    },
)
agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-5.4-mini"),
    tools=[send_email, delete_record, update_record, search],
    hooks={"before_tool": [confirmation_hook]},
)

Customizing feedback messages ​

tool 호출이 거부되거나 수정되면 BlockingConfirmationStrategy는 무슨 일이 일어났는지 설명하는 메시지를 LLM에 보내요. 세 개의 선택적 템플릿 파라미터가 이 메시지들을 제어해요 — 각각 합리적인 기본값이 있으므로 다른 표현을 원할 때만 설정하면 돼요:

• reject_template: 사용자가 tool 호출을 거부할 때 LLM에 전송. {tool_name} 플레이스홀더를 포함해야 함. 기본값: "Tool execution for '{tool_name}' was rejected by the user."
• modify_template: 사용자가 파라미터를 수정할 때 전송. {tool_name}과 {final_tool_params} 플레이스홀더를 포함해야 함. 기본값: "The parameters for tool '{tool_name}' were updated by the user to:\n{final_tool_params}"
• user_feedback_template: 사용자의 선택적 자유 텍스트 피드백을 두 메시지 중 하나에 추가. {feedback} 플레이스홀더를 포함해야 함. 기본값: "With user feedback: {feedback}"

strategy = BlockingConfirmationStrategy(
    confirmation_policy=AlwaysAskPolicy(),
    confirmation_ui=SimpleConsoleUI(),
    reject_template="Skipping '{tool_name}' — rejected by operator.",
    modify_template="Updated parameters for '{tool_name}': {final_tool_params}",
    user_feedback_template="Reason: {feedback}",
)

Policies ​

Policy는 언제 사람에게 물을지 제어해요.

Custom policy ​

haystack.hooks.human_in_the_loop.types에서 ConfirmationPolicy를 서브클래싱해 자신만의 policy를 구현할 수 있어요:

from haystack.hooks.human_in_the_loop.types import (
    ConfirmationPolicy,
    ConfirmationUIResult,
)
from typing import Any

class AskForSensitiveParamsPolicy(ConfirmationPolicy):
    """Only ask when the 'to' parameter looks like an external email domain."""
    def should_ask(
        self,
        tool_name: str,
        tool_description: str,
        tool_params: dict[str, Any],
    ) -> bool:
        to = tool_params.get("to", "")
        return not to.endswith("@mycompany.com")

상태 저장 policy라면 update_after_confirmation도 구현하세요. 사용자가 응답한 뒤 호출되며 전체 ConfirmationUIResult를 받아 결과에 따라 내부 상태를 업데이트할 수 있어요. 다음 policy는 tool 이름당 한 번 묻고 사용자가 이미 확인한 tool은 다시 묻지 않아요:

from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy
from haystack.hooks.human_in_the_loop import ConfirmationUIResult
from typing import Any

class AskOncePerToolPolicy(ConfirmationPolicy):
    """Ask once per tool name, regardless of parameters. Skip on repeat confirmed calls."""
    def __init__(self) -> None:
        self._confirmed_tools: set[str] = set()

    def should_ask(
        self,
        tool_name: str,
        tool_description: str,
        tool_params: dict[str, Any],
    ) -> bool:
        return tool_name not in self._confirmed_tools

    def update_after_confirmation(
        self,
        tool_name: str,
        tool_description: str,
        tool_params: dict[str, Any],
        confirmation_result: ConfirmationUIResult,
    ) -> None:
        if confirmation_result.action == "confirm":
            self._confirmed_tools.add(tool_name)

Dataclasses ​

ConfirmationUIResult ​

사람이 응답한 뒤 UI가 반환해요.

ToolExecutionDecision ​

전략이 Agent에게 반환해요.

Example: HITL with Hayhooks and Open WebUI ​

hitl-hayhooks-redis-openwebui 저장소는 Hayhooks로 서빙되는 Haystack Agent를 Open WebUI에 렌더링된 승인 다이얼로그와 함께 사용하는 전체 프로덕션 스타일 HITL 설정을 보여줘요.

핵심 패턴은 실행 시점에 요청별 리소스(Redis 클라이언트와 비동기 이벤트 큐)를 받는 커스텀 RedisConfirmationStrategy예요. 그런 리소스는 일반 hook_context run 인자(agent.run(messages=[...], hook_context={"redis": client}))로 전달하세요. ConfirmationHook은 state.data["hook_context"]로 이 dict를 읽고(state.get은 클라이언트·큐 같은 라이브 리소스에서 실패하는 깊은 복사를 반환 — Hooks 참조), 각 전략의 run()에 confirmation_strategy_context 키워드 인자로 전달해요. 이것이 커스텀 전략이 Redis 클라이언트와 이벤트 큐를 받는 방식이에요:

• tool 호출이 실행되려 하면 전략이 tool_call_start SSE 이벤트를 내보내고 승인 결정을 기다리며 Redis BLPOP에서 블록해요.
• Open WebUI Pipe 함수가 SSE 이벤트를 받고, 사용자에게 확인 다이얼로그를 보여준 뒤 approved 또는 rejected를 LPUSH로 Redis에 써요.
• Redis가 풀리면 전략이 ToolExecutionDecision을 반환하고 Agent가 계속돼요.

SimpleConsoleUI와 RichConsoleUI가 적합하지 않은 웹이나 서버 환경에서 non-blocking HITL이 필요하다면 좋은 참고 자료예요.

Custom UI ​

haystack.hooks.human_in_the_loop.types에서 ConfirmationUI를 구현해 자신만의 인터페이스 — 예를 들어 웹 기반 승인 큐 — 를 만들 수 있어요:

from haystack.hooks.human_in_the_loop.types import ConfirmationUI
from haystack.hooks.human_in_the_loop import ConfirmationUIResult
from typing import Any

class WebhookApprovalUI(ConfirmationUI):
    """Sends a webhook and waits for an async approval response."""
    def get_user_confirmation(
        self,
        tool_name: str,
        tool_description: str,
        tool_params: dict[str, Any],
    ) -> ConfirmationUIResult:
        # Send approval request to your system and wait for response
        response = send_approval_request_and_wait(tool_name, tool_params)
        return ConfirmationUIResult(
            action=response["action"],
            feedback=response.get("feedback"),
        )