Coverage for haystack/hooks/human_in_the_loop/policies.py: 100%

17 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 

4 

5from typing import Any 

6 

7from haystack.hooks.human_in_the_loop import ConfirmationUIResult 

8from haystack.hooks.human_in_the_loop.types import ConfirmationPolicy 

9 

10 

11class AlwaysAskPolicy(ConfirmationPolicy): 

12 """Always ask for confirmation.""" 

13 

14 def should_ask(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> bool: # noqa: ARG002 

15 """ 

16 Always ask for confirmation before executing the tool. 

17 

18 :param tool_name: The name of the tool to be executed. 

19 :param tool_description: The description of the tool. 

20 :param tool_params: The parameters to be passed to the tool. 

21 :returns: Always returns True, indicating confirmation is needed. 

22 """ 

23 return True 

24 

25 

26class NeverAskPolicy(ConfirmationPolicy): 

27 """Never ask for confirmation.""" 

28 

29 def should_ask(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> bool: # noqa: ARG002 

30 """ 

31 Never ask for confirmation, always proceed with tool execution. 

32 

33 :param tool_name: The name of the tool to be executed. 

34 :param tool_description: The description of the tool. 

35 :param tool_params: The parameters to be passed to the tool. 

36 :returns: Always returns False, indicating no confirmation is needed. 

37 """ 

38 return False 

39 

40 

41class AskOncePolicy(ConfirmationPolicy): 

42 """Ask only once per tool with specific parameters.""" 

43 

44 def __init__(self) -> None: 

45 """Creates an instance of AskOncePolicy.""" 

46 self._asked_tools: dict[str, Any] = {} 

47 

48 def should_ask(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> bool: # noqa: ARG002 

49 """ 

50 Ask for confirmation only once per tool with specific parameters. 

51 

52 :param tool_name: The name of the tool to be executed. 

53 :param tool_description: The description of the tool. 

54 :param tool_params: The parameters to be passed to the tool. 

55 :returns: True if confirmation is needed, False if already asked with the same parameters. 

56 """ 

57 # Don't ask again if we've already asked for this tool with the same parameters 

58 return not (tool_name in self._asked_tools and self._asked_tools[tool_name] == tool_params) 

59 

60 def update_after_confirmation( 

61 self, 

62 tool_name: str, 

63 tool_description: str, # noqa: ARG002 

64 tool_params: dict[str, Any], 

65 confirmation_result: ConfirmationUIResult, 

66 ) -> None: 

67 """ 

68 Store the tool and parameters if the action was "confirm" to avoid asking again. 

69 

70 This method updates the internal state to remember that the user has already confirmed the execution of the 

71 tool with the given parameters. 

72 

73 :param tool_name: The name of the tool that was executed. 

74 :param tool_description: The description of the tool. 

75 :param tool_params: The parameters that were passed to the tool. 

76 :param confirmation_result: The result from the confirmation UI. 

77 """ 

78 if confirmation_result.action == "confirm": 

79 self._asked_tools[tool_name] = tool_params