Coverage for haystack/hooks/human_in_the_loop/user_interfaces.py: 98%
96 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
5import json
6from threading import Lock
7from typing import Any
9from haystack.core.serialization import default_to_dict
10from haystack.hooks.human_in_the_loop import ConfirmationUIResult
11from haystack.hooks.human_in_the_loop.types import ConfirmationUI
12from haystack.lazy_imports import LazyImport
14with LazyImport(message="Run 'pip install rich'") as rich_import:
15 from rich.console import Console
16 from rich.panel import Panel
17 from rich.prompt import Prompt
19_ui_lock = Lock()
22class RichConsoleUI(ConfirmationUI):
23 """Rich console interface for user interaction."""
25 def __init__(self, console: "Console | None" = None) -> None:
26 """Creates an instance of RichConsoleUI."""
27 rich_import.check()
28 self.console = console or Console()
30 def get_user_confirmation(
31 self, tool_name: str, tool_description: str, tool_params: dict[str, Any]
32 ) -> ConfirmationUIResult:
33 """
34 Get user confirmation for tool execution via rich console prompts.
36 :param tool_name: The name of the tool to be executed.
37 :param tool_description: The description of the tool.
38 :param tool_params: The parameters to be passed to the tool.
39 :returns: ConfirmationUIResult based on user input.
40 """
41 with _ui_lock:
42 self._display_tool_info(tool_name, tool_description, tool_params)
43 # If wrong input is provided, Prompt.ask will re-prompt
44 choice = Prompt.ask("\nYour choice", choices=["y", "n", "m"], default="y", console=self.console)
45 return self._process_choice(choice, tool_params)
47 def _display_tool_info(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> None:
48 """
49 Display tool information and parameters in a rich panel.
51 :param tool_name: The name of the tool to be executed.
52 :param tool_description: The description of the tool.
53 :param tool_params: The parameters to be passed to the tool.
54 """
55 lines = [
56 f"[bold yellow]Tool:[/bold yellow] {tool_name}",
57 f"[bold yellow]Description:[/bold yellow] {tool_description}",
58 "\n[bold yellow]Arguments:[/bold yellow]",
59 ]
61 if tool_params:
62 for k, v in tool_params.items():
63 lines.append(f"[cyan]{k}:[/cyan] {v}")
64 else:
65 lines.append(" (No arguments)")
67 self.console.print(Panel("\n".join(lines), title="🔧 Tool Execution Request", title_align="left"))
69 def _process_choice(self, choice: str, tool_params: dict[str, Any]) -> ConfirmationUIResult:
70 """
71 Process the user's choice and return the corresponding ConfirmationUIResult.
73 :param choice: The user's choice ('y', 'n', or 'm').
74 :param tool_params: The original tool parameters.
75 :returns:
76 ConfirmationUIResult based on user input.
77 """
78 if choice == "y":
79 return ConfirmationUIResult(action="confirm")
80 if choice == "m":
81 return self._modify_params(tool_params)
82 # reject
83 feedback = Prompt.ask("Feedback message (optional)", default="", console=self.console)
84 return ConfirmationUIResult(action="reject", feedback=feedback or None)
86 def _modify_params(self, tool_params: dict[str, Any]) -> ConfirmationUIResult:
87 """
88 Prompt the user to modify tool parameters.
90 :param tool_params: The original tool parameters.
91 :returns:
92 ConfirmationUIResult with modified parameters.
93 """
94 new_params: dict[str, Any] = {}
95 for k, v in tool_params.items():
96 # We don't JSON dump strings to avoid users needing to input extra quotes
97 default_val = json.dumps(v) if not isinstance(v, str) else v
98 while True:
99 new_val = Prompt.ask(f"Modify '{k}'", default=default_val, console=self.console)
100 try:
101 if isinstance(v, str):
102 # Always treat input as string
103 new_params[k] = new_val
104 else:
105 # Parse JSON for all non-string types
106 new_params[k] = json.loads(new_val)
107 break
108 except json.JSONDecodeError:
109 self.console.print("[red]❌ Invalid JSON, please try again.[/red]")
111 return ConfirmationUIResult(action="modify", new_tool_params=new_params)
113 def to_dict(self) -> dict[str, Any]:
114 """
115 Serializes the RichConsoleConfirmationUI to a dictionary.
117 :returns:
118 Dictionary with serialized data.
119 """
120 # Note: Console object is not serializable; we store None
121 return default_to_dict(self, console=None)
124class SimpleConsoleUI(ConfirmationUI):
125 """Simple console interface using standard input/output."""
127 def get_user_confirmation(
128 self, tool_name: str, tool_description: str, tool_params: dict[str, Any]
129 ) -> ConfirmationUIResult:
130 """
131 Get user confirmation for tool execution via simple console prompts.
133 :param tool_name: The name of the tool to be executed.
134 :param tool_description: The description of the tool.
135 :param tool_params: The parameters to be passed to the tool.
136 """
137 with _ui_lock:
138 self._display_tool_info(tool_name, tool_description, tool_params)
139 valid_choices = {"y", "yes", "n", "no", "m", "modify"}
140 while True:
141 choice = input("Confirm execution? (y=confirm / n=reject / m=modify): ").strip().lower()
142 if choice in valid_choices:
143 break
144 print("Invalid input. Please enter 'y', 'n', or 'm'.")
145 return self._process_choice(choice, tool_params)
147 def _display_tool_info(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> None:
148 """
149 Display tool information and parameters in the console.
151 :param tool_name: The name of the tool to be executed.
152 :param tool_description: The description of the tool.
153 :param tool_params: The parameters to be passed to the tool.
154 """
155 print("\n--- Tool Execution Request ---")
156 print(f"Tool: {tool_name}")
157 print(f"Description: {tool_description}")
158 print("Arguments:")
159 if tool_params:
160 for k, v in tool_params.items():
161 print(f" {k}: {v}")
162 else:
163 print(" (No arguments)")
164 print("-" * 30)
166 def _process_choice(self, choice: str, tool_params: dict[str, Any]) -> ConfirmationUIResult:
167 """
168 Process the user's choice and return the corresponding ConfirmationUIResult.
170 :param choice: The user's choice ('y', 'n', or 'm').
171 :param tool_params: The original tool parameters.
172 :returns:
173 ConfirmationUIResult based on user input.
174 """
175 if choice in ("y", "yes"):
176 return ConfirmationUIResult(action="confirm")
177 if choice in ("m", "modify"):
178 return self._modify_params(tool_params)
179 # reject
180 feedback = input("Feedback message (optional): ").strip()
181 return ConfirmationUIResult(action="reject", feedback=feedback or None)
183 def _modify_params(self, tool_params: dict[str, Any]) -> ConfirmationUIResult:
184 """
185 Prompt the user to modify tool parameters.
187 :param tool_params: The original tool parameters.
188 :returns:
189 ConfirmationUIResult with modified parameters.
190 """
191 new_params: dict[str, Any] = {}
193 if not tool_params:
194 print("No parameters to modify, skipping modification.")
195 return ConfirmationUIResult(action="modify", new_tool_params=new_params)
197 for k, v in tool_params.items():
198 # We don't JSON dump strings to avoid users needing to input extra quotes
199 default_val = json.dumps(v) if not isinstance(v, str) else v
200 while True:
201 new_val = input(f"Modify '{k}' (current: {default_val}): ").strip() or default_val
202 try:
203 if isinstance(v, str):
204 # Always treat input as string
205 new_params[k] = new_val
206 else:
207 # Parse JSON for all non-string types
208 new_params[k] = json.loads(new_val)
209 break
210 except json.JSONDecodeError:
211 print("❌ Invalid JSON, please try again.")
213 return ConfirmationUIResult(action="modify", new_tool_params=new_params)