feat: Add Human in the Loop utilities for tool execution in Agent

[sjrl]

Related Issues

• fixes https://github.com/deepset-ai/haystack-private/issues/186

Proposed Changes:

This PR introduces a human-in-the-loop (HITL) mechanism for tool execution in Agent, allowing users to confirm, reject, or modify tool invocations interactively.

TODOs

• Create a HiTL example that uses Haystack's BreakPoint feature
• Add to_dict and from_dict to Agent
• Add unit and integration tests
  • Double check tools that expect no params work as expected
• Update README with new experiment

Key Changes

• haystack_experimental/components/agents/human_in_the_loop/

  • dataclasses.py

    • Adds ConfirmationUIResult and ToolExecutionDecision dataclasses to standardize user feedback and tool execution decisions

  • policies.py

    • Implements ConfirmationPolicy base class and concrete policies: AlwaysAskPolicy, NeverAskPolicy, and AskOncePolicy

  • user_interfaces.py

    • Provides ConfirmationUI base class and two implementations: RichConsoleUI (using Rich library) and SimpleConsoleUI (using standard input), supporting interactive user feedback and parameter modification.
    • These UI's are thread locked so in the case of parallel tool execution that the confirmations are still run sequentially. Look at hitl_multi_agent_example.py to see a case where this occurs.

  • protocol.py

    • Defines the ConfirmationStrategy protocol. I'm not 100% if this should be a protocol or a base class.

  • strategies.py

    • Introduces HumanInTheLoopStrategy to orchestrate the confirmation policy and the confirmation UI, returning a ToolExecutionDecision.

• haystack_experimental/components/tools/tool_invoker.py

  • Extends ToolInvoker to support per-tool confirmation strategies.
  • Integrates HITL logic before tool execution, handling all user feedback and parameter modification before calling tools.

• haystack_experimental/components/agents/agent.py

  • Modified the init method of Agent use the new experimental ToolInvoker and added the confirmation_strategies explicitly.

How did you test it?

Provided an example script called hitl_intro_example.py which shows how to use these new utility functions to convert existing tools into ones that ask for confirmation. I've also reproduced it here

Single Agent Example highlighting different Confirmation Policies and Confirmation UIs

# SPDX-FileCopyrightText: 2022-present deepset GmbH <[email protected]>
#
# SPDX-License-Identifier: Apache-2.0

from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import create_tool_from_function
from rich.console import Console

from haystack_experimental.components.agents.agent import Agent
from haystack_experimental.components.agents.human_in_the_loop.policies import (
    AlwaysAskPolicy,
    AskOncePolicy,
    NeverAskPolicy,
)
from haystack_experimental.components.agents.human_in_the_loop.user_interfaces import (
    RichConsoleUI,
    SimpleConsoleUI,
)
from haystack_experimental.components.agents.human_in_the_loop.strategies import HumanInTheLoopStrategy


def addition(a: float, b: float) -> float:
    """
    A simple addition function.

    :param a: First float.
    :param b: Second float.
    :returns:
        Sum of a and b.
    """
    return a + b


addition_tool = create_tool_from_function(
    function=addition,
    name="addition",
    description="Add two floats together.",
)


def get_bank_balance(account_id: str) -> str:
    """
    Simulate fetching a bank balance for a given account ID.

    :param account_id: The ID of the bank account.
    :returns:
        A string representing the bank balance.
    """
    return f"Balance for account {account_id} is $1,234.56"


balance_tool = create_tool_from_function(
    function=get_bank_balance,
    name="get_bank_balance",
    description="Get the bank balance for a given account ID.",
)


def get_phone_number(name: str) -> str:
    """
    Simulate fetching a phone number for a given name.

    :param name: The name of the person.
    :returns:
        A string representing the phone number.
    """
    return f"The phone number for {name} is (123) 456-7890"


phone_tool = create_tool_from_function(
    function=get_phone_number,
    name="get_phone_number",
    description="Get the phone number for a given name.",
)

# Define shared console
cons = Console()

# Define Main Agent with multiple tools and different confirmation strategies
agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4.1"),
    tools=[balance_tool, addition_tool, phone_tool],
    system_prompt="You are a helpful financial assistant. Use the provided tool to get bank balances when needed.",
    confirmation_strategies={
        balance_tool.name: HumanInTheLoopStrategy(
            confirmation_policy=AlwaysAskPolicy(), confirmation_ui=RichConsoleUI(console=cons)
        ),
        addition_tool.name: HumanInTheLoopStrategy(
            confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI()
        ),
        phone_tool.name: HumanInTheLoopStrategy(
            confirmation_policy=AskOncePolicy(), confirmation_ui=SimpleConsoleUI()
        ),
    },
)

# Call bank tool with confirmation (Always Ask) using RichConsoleUI
result = agent.run([ChatMessage.from_user("What's the balance of account 56789?")])
last_message = result["last_message"]
cons.print(f"\n[bold green]Agent Result:[/bold green] {last_message.text}")

# Call addition tool with confirmation (Never Ask)
result = agent.run([ChatMessage.from_user("What is 5.5 + 3.2?")])
last_message = result["last_message"]
print(f"\nAgent Result: {last_message.text}")

# Call phone tool with confirmation (Ask Once) using SimpleConsoleUI
result = agent.run([ChatMessage.from_user("What is the phone number of Alice?")])
last_message = result["last_message"]
print(f"\nAgent Result: {last_message.text}")

# Call phone tool again to see that it doesn't ask for confirmation the second time
result = agent.run([ChatMessage.from_user("What is the phone number of Alice?")])
last_message = result["last_message"]
print(f"\nAgent Result: {last_message.text}")

Multi-Agent Example: Tools within Sub-Agents use HiTL Confirmation

# SPDX-FileCopyrightText: 2022-present deepset GmbH <[email protected]>
#
# SPDX-License-Identifier: Apache-2.0

from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import ComponentTool, create_tool_from_function
from rich.console import Console

from haystack_experimental.components.agents.agent import Agent
from haystack_experimental.components.agents.human_in_the_loop.policies import AlwaysAskPolicy
from haystack_experimental.components.agents.human_in_the_loop.strategies import HumanInTheLoopStrategy
from haystack_experimental.components.agents.human_in_the_loop.user_interfaces import RichConsoleUI


def addition(a: float, b: float) -> float:
    """
    A simple addition function.

    :param a: First float.
    :param b: Second float.
    :returns:
        Sum of a and b.
    """
    return a + b


addition_tool = create_tool_from_function(
    function=addition,
    name="addition",
    description="Add two floats together.",
)


def get_bank_balance(account_id: str) -> str:
    """
    Simulate fetching a bank balance for a given account ID.

    :param account_id: The ID of the bank account.
    :returns:
        A string representing the bank balance.
    """
    return f"Balance for account {account_id} is $1,234.56"


balance_tool = create_tool_from_function(
    function=get_bank_balance,
    name="get_bank_balance",
    description="Get the bank balance for a given account ID.",
)

# Define shared console for all UIs
cons = Console()

# Define Bank Sub-Agent
bank_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4.1"),
    tools=[balance_tool],
    system_prompt="You are a helpful financial assistant. Use the provided tool to get bank balances when needed.",
    confirmation_strategies={
        balance_tool.name: HumanInTheLoopStrategy(
            confirmation_policy=AlwaysAskPolicy(), confirmation_ui=RichConsoleUI(console=cons)
        ),
    },
)
bank_agent_tool = ComponentTool(
    component=bank_agent,
    name="bank_agent_tool",
    description="A bank agent that can get bank balances.",
)

# Define Math Sub-Agent
math_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4.1"),
    tools=[addition_tool],
    system_prompt="You are a helpful math assistant. Use the provided tool to perform addition when needed.",
    confirmation_strategies={
        addition_tool.name: HumanInTheLoopStrategy(
            # We use AlwaysAskPolicy here for demonstration; in real scenarios, you might choose NeverAskPolicy
            confirmation_policy=AlwaysAskPolicy(),
            confirmation_ui=RichConsoleUI(console=cons),
        ),
    },
)
math_agent_tool = ComponentTool(
    component=math_agent,
    name="math_agent_tool",
    description="A math agent that can perform addition.",
)

# Define Main Agent with Sub-Agents as tools
planner_agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4.1"),
    tools=[bank_agent_tool, math_agent_tool],
    system_prompt="""You are a master agent that can delegate tasks to sub-agents based on the user's request.
Available sub-agents:
- bank_agent_tool: A bank agent that can get bank balances.
- math_agent_tool: A math agent that can perform addition.
Use the appropriate sub-agent to handle the user's request.
""",
)

# Make bank balance request to planner agent
result = planner_agent.run([ChatMessage.from_user("What's the balance of account 56789?")])
last_message = result["last_message"]
cons.print(f"\n[bold green]Agent Result:[/bold green] {last_message.text}")

# Make addition request to planner agent
result = planner_agent.run([ChatMessage.from_user("What is 5.5 + 3.2?")])
last_message = result["last_message"]
print(f"\nAgent Result: {last_message.text}")

# Make bank balance request and addition request to planner agent
# NOTE: This will try and invoke both sub-agents in parallel requiring a thread-safe UI
result = planner_agent.run([ChatMessage.from_user("What's the balance of account 56789 and what is 5.5 + 3.2?")])
last_message = result["last_message"]
print(f"\nAgent Result: {last_message.text}")

Notes for the reviewer

@julian-risch this now follows more closely your idea of passing the tool confirmation strategies to the Agent (or ToolInvoker) rather than applying it at the tool level. I think I like this design better since it has better separation and it makes the various policies easy to understand.

Checklist

• I have read the contributors guidelines and the code of conduct
• I have updated the related issue with new insights and changes
• I added unit tests and updated the docstrings
• I've used one of the conventional commit types for my PR title: fix:, feat:, build:, chore:, ci:, docs:, style:, refactor:, perf:, test:.
• I documented my code
• I ran pre-commit hooks and fixed any issue

[coveralls]

Pull Request Test Coverage Report for Build 18090924811

Details

• 0 of 0 changed or added relevant lines in 0 files are covered.
• No unchanged relevant lines lost coverage.
• Overall coverage decreased (-0.7%) to 66.742%

---

Totals
Change from base Build 18081831894:	-0.7%
Covered Lines:	885
Relevant Lines:	1326

---

💛 - Coveralls

[anakin87]

Some high-level comments:

• Skimming other frameworks, it doesn't seem like there's a single right way to do HITL (Tool level or Agent level?). If this approach also helps with serialization headaches, it could be a good direction.

• I see a design tension between giving users freedom to provide their own ConfirmationStrategy (there is a protocol for this and ConfirmationResult.action is a string) and requiring a specific stricter strategy (the current implementation of _handle_confirmation_strategies only expects confirm/reject/modify). If we want to let users free to implement their strategies, I would suggest delegating most of the logic in _handle_confirmation_strategies to the ConfirmationStrategy implementation. (Otherwise, we could be stricter in defining ConfirmationResult and removing the protocol...)

• about names: maybe we can rename UserInterface → ConfirmationUI but if we go with the approach proposed in this PR, I recommend checking in with Bilge.

[anakin87]

#371 should fix the failing docs workflow

[anakin87]

I ran the examples. They work properly and confirm that implementation is solid.

In a future notebook, as a user, I would also expect an introduction to the classes introduced, their meaning, and customization options.

[anakin87]

Protocols

I see that in this PR we are using a protocol (ConfirmationPolicy) and ConfirmationUI, which looks like a protocol (or an abstract class), since it cannot be used in isolation.

For consistency, I would use protocols. They can also have default implementations for some methods and other classes can inherit from them to use the default implementations. See Python docs.

EDIT: we should verify if this idea can work well...

Code organization

If we make both ConfirmationPolicy and ConfirmationUI two protocols, I would reorganize the code as follows:

human_in_the_loop/
├── __init__.py                    
├── dataclasses.py             
├── types.py                        # Protocols
├── policies.py                     # Policy implementations
├── strategies.py                   # Strategy implementations
└── user_interfaces.py              # UI implementations

[sjrl]

Protocols

I see that in this PR we are using a protocol (ConfirmationPolicy) and ConfirmationUI, which looks like a protocol (or an abstract class), since it cannot be used in isolation.

For consistency, I would use protocols. They can also have default implementations for some methods and other classes can inherit from them to use the default implementations. See Python docs.

EDIT: we should verify if this idea can work well...

Code organization

If we make both ConfirmationPolicy and ConfirmationUI two protocols, I would reorganize the code as follows:

human_in_the_loop/
├── __init__.py                    
├── dataclasses.py             
├── types.py                        # Protocols
├── policies.py                     # Policy implementations
├── strategies.py                   # Strategy implementations
└── user_interfaces.py              # UI implementations

This idea seems to work. I'll switch to this structure for now and keep testing.