Coverage for haystack/hooks/human_in_the_loop/hooks.py: 97%
30 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from typing import Any
7from haystack.components.agents.state.state import State
8from haystack.components.agents.state.state_utils import replace_values
9from haystack.core.serialization import default_from_dict, default_to_dict
10from haystack.hooks.human_in_the_loop.strategies import (
11 _deserialize_confirmation_strategies,
12 _process_confirmation_strategies,
13 _process_confirmation_strategies_async,
14 _serialize_confirmation_strategies,
15)
16from haystack.hooks.human_in_the_loop.types import ConfirmationStrategy
19class ConfirmationHook:
20 """
21 A `before_tool` Agent hook that applies Human-in-the-Loop confirmation strategies to pending tool calls.
23 Register it on an `Agent` to confirm, modify, or reject tool calls before they run:
25 ```python
26 from haystack.components.agents import Agent
27 from haystack.components.generators.chat import OpenAIChatGenerator
28 from haystack.tools import tool
29 from haystack.hooks.human_in_the_loop import (
30 AlwaysAskPolicy,
31 BlockingConfirmationStrategy,
32 ConfirmationHook,
33 NeverAskPolicy,
34 RichConsoleUI,
35 SimpleConsoleUI,
36 )
38 @tool
39 def delete_file(path: str) -> str:
40 '''Delete the file at the given path.'''
41 return f"Deleted {path}."
43 hook = ConfirmationHook(
44 confirmation_strategies={
45 "delete_file": BlockingConfirmationStrategy(
46 confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI()
47 )
48 }
49 )
50 agent = Agent(chat_generator=OpenAIChatGenerator(), tools=[delete_file], hooks={"before_tool": [hook]})
51 ```
53 A key may be a single tool name, a tuple of tool names sharing one strategy, or the wildcard `"*"` which applies
54 to any tool without a more specific entry. More specific keys win, so you can set a default for all tools and
55 override individual ones:
57 ```python
58 hook = ConfirmationHook(
59 confirmation_strategies={
60 "delete_file": BlockingConfirmationStrategy(
61 confirmation_policy=AlwaysAskPolicy(), confirmation_ui=RichConsoleUI()
62 ),
63 "*": BlockingConfirmationStrategy(
64 confirmation_policy=NeverAskPolicy(), confirmation_ui=SimpleConsoleUI()
65 ),
66 }
67 )
68 ```
70 Request-scoped resources for the strategies (e.g. a WebSocket or queue) are passed per run via the Agent's
71 `hook_context` argument (`agent.run(messages=[...], hook_context={...})`) and read by the hook with
72 `state.data.get("hook_context")`.
74 This hook only makes sense at the `before_tool` hook point, where the pending tool calls exist (between the model
75 requesting tools and those tools running); the Agent enforces this and raises if it is registered elsewhere. Use a
76 single ConfirmationHook with one entry per tool (or per tuple of tools) in `confirmation_strategies` rather than
77 registering several hooks.
78 """
80 # Restrict this hook to the "before_tool" point; the Agent validates this at construction.
81 allowed_hook_points = ("before_tool",)
83 def __init__(self, confirmation_strategies: dict[str | tuple[str, ...], ConfirmationStrategy]) -> None:
84 """
85 Initialize the hook with its per-tool confirmation strategies.
87 :param confirmation_strategies: Mapping of tool name (or a tuple of tool names) to its `ConfirmationStrategy`.
88 The wildcard key `"*"` applies to any tool without a more specific entry.
89 """
90 self.confirmation_strategies = confirmation_strategies
92 def run(self, state: State) -> None:
93 """
94 Confirm the pending tool calls, rewriting the `messages` in `state` to reflect modifications and rejections.
96 :param state: The Agent's live `State`. Reads the available tools (`state.data.get("tools")`) and the per-run
97 context (`state.data.get("hook_context")`), and the pending tool calls from the last message; writes the
98 updated conversation back to `messages`. Reads go through `state.data` rather than `state.get`, which
99 deep-copies and would break non-copyable resources (e.g. a WebSocket or client) in `hook_context`.
100 """
101 messages = state.data.get("messages") or []
102 if not messages or not messages[-1].tool_calls:
103 return
104 new_chat_history = _process_confirmation_strategies(
105 confirmation_strategies=self.confirmation_strategies,
106 messages_with_tool_calls=[messages[-1]],
107 tools=state.data.get("tools") or [],
108 state=state,
109 confirmation_strategy_context=state.data.get("hook_context"),
110 )
111 state.set("messages", new_chat_history, handler_override=replace_values)
113 async def run_async(self, state: State) -> None:
114 """Async version of `run`."""
115 messages = state.data.get("messages") or []
116 if not messages or not messages[-1].tool_calls:
117 return
118 new_chat_history = await _process_confirmation_strategies_async(
119 confirmation_strategies=self.confirmation_strategies,
120 messages_with_tool_calls=[messages[-1]],
121 tools=state.data.get("tools") or [],
122 state=state,
123 confirmation_strategy_context=state.data.get("hook_context"),
124 )
125 state.set("messages", new_chat_history, handler_override=replace_values)
127 def to_dict(self) -> dict[str, Any]:
128 """Serialize the hook, including its confirmation strategies (tuple keys become JSON-array strings)."""
129 return default_to_dict(
130 self, confirmation_strategies=_serialize_confirmation_strategies(self.confirmation_strategies)
131 )
133 @classmethod
134 def from_dict(cls, data: dict[str, Any]) -> "ConfirmationHook":
135 """Deserialize the hook, reconstructing its confirmation strategies."""
136 init_params = data.get("init_parameters", {})
137 if init_params.get("confirmation_strategies") is not None:
138 init_params["confirmation_strategies"] = _deserialize_confirmation_strategies(
139 init_params["confirmation_strategies"]
140 )
141 return default_from_dict(cls, data)