Human-in-the-Loop Agent

HITLAgent adds approval gates to the ReAct loop. Before every tool call, a review_fn is called with a HITLStep describing the proposed action. The reviewer returns a Verdict — approve, reject, or approve-with-override.

This is the recommended pattern for any agentic system that touches real-world state (databases, APIs, files, external services).

Usage

from brain.patterns import HITLAgent, HITLStep, Verdict

def my_reviewer(step: HITLStep) -> Verdict:
    print(f"\nAgent wants to: {step.action}({step.action_input})")
    choice = input("Allow? [y/n]: ").strip().lower()
    return Verdict(approved=choice == "y")

agent = HITLAgent(
    tools={"delete_record": db.delete, "send_email": mailer.send},
    review_fn=my_reviewer,
)
result = agent.run("Archive all records older than 90 days")
print(result.answer)
print(f"Approved {result.approved_steps}, rejected {result.rejected_steps}")

Automated policy reviewers

from brain.patterns import block_tools, block_all, allow_all

# Block specific destructive tools
agent = HITLAgent(tools={...}, review_fn=block_tools("delete_record", "truncate_table"))

# Stop immediately on first rejection
agent = HITLAgent(tools={...}, review_fn=my_policy, stop_on_rejection=True)

# Override the action input (e.g. sanitise a query)
def sanitise(step: HITLStep) -> Verdict:
    safe_input = step.action_input.replace("DROP", "")
    return Verdict(approved=True, override_input=safe_input)

API Reference

brain.patterns.human_in_loop.HITLAgent

HITLAgent(tools: dict[str, Tool] | None = None, review_fn: ReviewFn | None = None, provider: LLMProvider | None = None, max_steps: int = 8, stop_on_rejection: bool = False, system_prompt: str | None = None)

Bases: BasePattern

ReAct agent with human-in-the-loop approval gates.

Before every tool call the agent proposes, review_fn is called with a HITLStep describing the proposed action. If the reviewer returns Verdict(approved=False), the step is skipped, the agent is told the action was rejected, and the loop continues.

If stop_on_rejection=True (default: False), the first rejection terminates the run immediately.

Parameters:

Name	Type	Description	Default
tools	dict[str, Tool] | None	Mapping of tool name → callable(str) -> str.	None
review_fn	ReviewFn | None	Callable that receives a HITLStep and returns a Verdict. Defaults to allow_all (no-op gate — useful for testing).	None
provider	LLMProvider | None	LLMProvider instance; if None, uses LocalEchoProvider.	None
max_steps	int	Maximum iterations before giving up.	8
stop_on_rejection	bool	If True, the first rejection ends the run.	False
system_prompt	str | None	Override the default ReAct system prompt.	None

Source code in brain/patterns/human_in_loop.py

def __init__(
    self,
    tools: dict[str, Tool] | None = None,
    review_fn: ReviewFn | None = None,
    provider: LLMProvider | None = None,
    max_steps: int = 8,
    stop_on_rejection: bool = False,
    system_prompt: str | None = None,
) -> None:
    self.tools: dict[str, Tool] = tools or {}
    self.review_fn: ReviewFn = review_fn or allow_all
    self.max_steps = max_steps
    self.stop_on_rejection = stop_on_rejection
    self._system_prompt = system_prompt
    self._provider = provider

run

run(task: str, **kwargs: Any) -> HITLResult

Run the HITL agent.

Parameters:

Name	Type	Description	Default
task	str	The question or instruction for the agent.	required

Returns:

Type	Description
HITLResult	HITLResult (extends PatternResult) with approval/rejection counts.

Source code in brain/patterns/human_in_loop.py

def run(self, task: str, **kwargs: Any) -> HITLResult:  # type: ignore[override]
    """Run the HITL agent.

    Args:
        task: The question or instruction for the agent.

    Returns:
        HITLResult (extends PatternResult) with approval/rejection counts.
    """
    provider = self._get_provider()
    tool_names = ", ".join(self.tools.keys()) if self.tools else "none"
    system = (self._system_prompt or _SYSTEM_PROMPT).format(tool_names=tool_names)

    messages: list[dict[str, Any]] = [
        {"role": "system", "content": system},
        {"role": "user", "content": task},
    ]
    steps: list[Step] = []
    approved_count = 0
    rejected_count = 0
    rejection_reasons: list[tuple[int, str]] = []

    for i in range(self.max_steps):
        try:
            llm_result = provider.generate(messages, tools=[], tool_choice="none")
        except Exception as exc:  # noqa: BLE001
            logger.warning("HITLAgent: provider error at step %d: %s", i, exc)
            return HITLResult(
                answer="",
                steps=steps,
                iterations=i,
                ok=False,
                error=str(exc),
                approved_steps=approved_count,
                rejected_steps=rejected_count,
                rejection_reasons=rejection_reasons,
            )

        text = llm_result.text.strip()
        thought, action, action_input, final = _parse(text)

        if final is not None:
            steps.append(Step(index=i, thought=thought, text=text))
            return HITLResult(
                answer=final,
                steps=steps,
                iterations=i + 1,
                approved_steps=approved_count,
                rejected_steps=rejected_count,
                rejection_reasons=rejection_reasons,
            )

        # No action — treat as final answer
        if not action:
            steps.append(Step(index=i, thought=thought, text=text))
            return HITLResult(
                answer=text,
                steps=steps,
                iterations=i + 1,
                approved_steps=approved_count,
                rejected_steps=rejected_count,
                rejection_reasons=rejection_reasons,
            )

        # Present the proposed step to the reviewer
        hitl_step = HITLStep(
            step_index=i,
            thought=thought,
            action=action,
            action_input=action_input,
            conversation_so_far=list(messages),
        )

        try:
            verdict = self.review_fn(hitl_step)
        except Exception as exc:  # noqa: BLE001
            logger.warning("HITLAgent: review_fn raised at step %d: %s", i, exc)
            verdict = Verdict(approved=False, reason=f"review_fn error: {exc}")

        if not verdict.approved:
            rejected_count += 1
            rejection_reasons.append((i, verdict.reason))
            logger.info("HITLAgent: step %d rejected: %s", i, verdict.reason)

            if self.stop_on_rejection:
                steps.append(
                    Step(
                        index=i,
                        thought=thought,
                        action=action,
                        action_input=action_input,
                        text=text,
                    )
                )
                return HITLResult(
                    answer="",
                    steps=steps,
                    iterations=i + 1,
                    ok=False,
                    error=f"Step {i} rejected by reviewer: {verdict.reason}",
                    approved_steps=approved_count,
                    rejected_steps=rejected_count,
                    rejection_reasons=rejection_reasons,
                    stopped_by_reviewer=True,
                )

            # Tell the agent the action was rejected and continue
            rejection_msg = (
                f"Action '{action}' was rejected by the safety reviewer"
                + (f": {verdict.reason}" if verdict.reason else "")
                + ". Please try a different approach."
            )
            messages.append({"role": "assistant", "content": text})
            messages.append({"role": "user", "content": f"Observation: {rejection_msg}"})
            steps.append(
                Step(
                    index=i,
                    thought=thought,
                    action=action,
                    action_input=action_input,
                    observation=rejection_msg,
                    text=text,
                )
            )
            continue

        # Approved — optionally override the input
        approved_count += 1
        effective_input = (
            verdict.override_input if verdict.override_input is not None else action_input
        )

        if action in self.tools:
            try:
                observation = str(self.tools[action](effective_input))
            except Exception as exc:  # noqa: BLE001
                observation = f"Tool error: {exc}"
        else:
            observation = f"Unknown tool '{action}'. Available: {tool_names}"

        step = Step(
            index=i,
            thought=thought,
            action=action,
            action_input=effective_input,
            observation=observation,
            text=text,
        )
        steps.append(step)

        messages.append({"role": "assistant", "content": text})
        messages.append({"role": "user", "content": f"Observation: {observation}"})

    return HITLResult(
        answer=steps[-1].text if steps else "",
        steps=steps,
        iterations=self.max_steps,
        ok=False,
        error="max_steps reached without Final Answer",
        approved_steps=approved_count,
        rejected_steps=rejected_count,
        rejection_reasons=rejection_reasons,
    )

brain.patterns.human_in_loop.HITLResult dataclass

HITLResult(answer: str, steps: list[Step] = list(), iterations: int = 0, ok: bool = True, error: str | None = None, metadata: dict[str, Any] = dict(), approved_steps: int = 0, rejected_steps: int = 0, rejection_reasons: list[tuple[int, str]] = list(), stopped_by_reviewer: bool = False)

Bases: PatternResult

PatternResult extended with HITL-specific metrics.

Attributes:

Name	Type	Description
approved_steps	int	Number of steps the reviewer approved.
rejected_steps	int	Number of steps the reviewer rejected.
rejection_reasons	list[tuple[int, str]]	List of (step_index, reason) tuples for rejections.
stopped_by_reviewer	bool	True if the run stopped because of a rejection.

brain.patterns.human_in_loop.HITLStep dataclass

HITLStep(step_index: int, thought: str, action: str, action_input: str, conversation_so_far: list[dict[str, Any]] = list())

A proposed step surfaced to the human reviewer.

Attributes:

Name	Type	Description
step_index	int	Which iteration this is (0-based).
thought	str	The agent's reasoning text.
action	str	The tool name the agent wants to call.
action_input	str	The argument the agent wants to pass.
conversation_so_far	list[dict[str, Any]]	The full message history up to this point.

brain.patterns.human_in_loop.Verdict dataclass

Verdict(approved: bool, override_input: str | None = None, reason: str = '')

A reviewer's decision on a proposed agent step.

Attributes:

Name	Type	Description
approved	bool	Whether the step is allowed to proceed.
override_input	str | None	If set, replace the agent's action_input with this string before calling the tool. Implies approved=True.
reason	str	Optional human-readable explanation for the decision.

brain.patterns.human_in_loop.allow_all

allow_all(step: HITLStep) -> Verdict

Reviewer that approves every step. Useful for testing.

Source code in brain/patterns/human_in_loop.py

def allow_all(step: HITLStep) -> Verdict:
    """Reviewer that approves every step. Useful for testing."""
    return Verdict(approved=True)

brain.patterns.human_in_loop.block_all

block_all(step: HITLStep) -> Verdict

Reviewer that rejects every step. Useful for testing rejection paths.

Source code in brain/patterns/human_in_loop.py

def block_all(step: HITLStep) -> Verdict:
    """Reviewer that rejects every step. Useful for testing rejection paths."""
    return Verdict(approved=False, reason="block_all policy")

brain.patterns.human_in_loop.block_tools

block_tools(*tool_names: str) -> ReviewFn

Return a reviewer that rejects specific tool names and approves the rest.

Parameters:

Name	Type	Description	Default
*tool_names	str	Tool names to block.	()

Returns:

Type	Description
ReviewFn	A ReviewFn callable.

Source code in brain/patterns/human_in_loop.py

def block_tools(*tool_names: str) -> ReviewFn:
    """Return a reviewer that rejects specific tool names and approves the rest.

    Args:
        *tool_names: Tool names to block.

    Returns:
        A ReviewFn callable.
    """
    blocked = frozenset(tool_names)

    def _reviewer(step: HITLStep) -> Verdict:
        if step.action in blocked:
            return Verdict(approved=False, reason=f"tool '{step.action}' is blocked")
        return Verdict(approved=True)

    return _reviewer

When to Use

Situation	Recommendation
Agent touches real-world state	HITLAgent
Compliance / audit requirements	HITLAgent
Automated safety policy	HITLAgent + block_tools
No human in the loop needed	ReActAgent or StreamingReActAgent
HITL + streaming UI	Combine HITLAgent with a custom review_fn that yields events