Coverage for haystack/hooks/human_in_the_loop/types/protocol.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any, Protocol 

6 

7from haystack.core.serialization import default_from_dict, default_to_dict 

8from haystack.hooks.human_in_the_loop.dataclasses import ConfirmationUIResult, ToolExecutionDecision 

9 

10 

11class ConfirmationUI(Protocol): 

12 """Base class for confirmation UIs.""" 

13 

14 def get_user_confirmation( 

15 self, tool_name: str, tool_description: str, tool_params: dict[str, Any] 

16 ) -> ConfirmationUIResult: 

17 """Get user confirmation for tool execution.""" 

18 ... 

19 

20 def to_dict(self) -> dict[str, Any]: 

21 """Serialize the UI to a dictionary.""" 

22 return default_to_dict(self) 

23 

24 @classmethod 

25 def from_dict(cls, data: dict[str, Any]) -> "ConfirmationUI": 

26 """Deserialize the ConfirmationUI from a dictionary.""" 

27 return default_from_dict(cls, data) 

28 

29 

30class ConfirmationPolicy(Protocol): 

31 """Base class for confirmation policies.""" 

32 

33 def should_ask(self, tool_name: str, tool_description: str, tool_params: dict[str, Any]) -> bool: 

34 """Determine whether to ask for confirmation.""" 

35 ... 

36 

37 def update_after_confirmation( 

38 self, 

39 tool_name: str, # noqa: ARG002 

40 tool_description: str, # noqa: ARG002 

41 tool_params: dict[str, Any], # noqa: ARG002 

42 confirmation_result: ConfirmationUIResult, # noqa: ARG002 

43 ) -> None: 

44 """Update the policy based on the confirmation UI result.""" 

45 return 

46 

47 def to_dict(self) -> dict[str, Any]: 

48 """Serialize the policy to a dictionary.""" 

49 return default_to_dict(self) 

50 

51 @classmethod 

52 def from_dict(cls, data: dict[str, Any]) -> "ConfirmationPolicy": 

53 """Deserialize the policy from a dictionary.""" 

54 return default_from_dict(cls, data) 

55 

56 

57class ConfirmationStrategy(Protocol): 

58 def run( 

59 self, 

60 *, 

61 tool_name: str, 

62 tool_description: str, 

63 tool_params: dict[str, Any], 

64 tool_call_id: str | None = None, 

65 confirmation_strategy_context: dict[str, Any] | None = None, 

66 ) -> ToolExecutionDecision: 

67 """ 

68 Run the confirmation strategy for a given tool and its parameters. 

69 

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

71 :param tool_description: The description of the tool. 

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

73 :param tool_call_id: Optional unique identifier for the tool call. This can be used to track and correlate 

74 the decision with a specific tool invocation. 

75 :param confirmation_strategy_context: Optional context dictionary for passing request-scoped resources 

76 (e.g., WebSocket connections, async queues) in web/server environments. 

77 

78 :returns: 

79 The result of the confirmation strategy (e.g., tool output, rejection message, etc.). 

80 """ 

81 ... 

82 

83 async def run_async( 

84 self, 

85 *, 

86 tool_name: str, 

87 tool_description: str, 

88 tool_params: dict[str, Any], 

89 tool_call_id: str | None = None, 

90 confirmation_strategy_context: dict[str, Any] | None = None, 

91 ) -> ToolExecutionDecision: 

92 """ 

93 Async version of run. Run the confirmation strategy for a given tool and its parameters. 

94 

95 Default implementation calls the sync run() method. Override for true async behavior. 

96 

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

98 :param tool_description: The description of the tool. 

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

100 :param tool_call_id: Optional unique identifier for the tool call. This can be used to track and correlate 

101 the decision with a specific tool invocation. 

102 :param confirmation_strategy_context: Optional context dictionary for passing request-scoped resources 

103 (e.g., WebSocket connections, async queues) in web/server environments. 

104 

105 :returns: 

106 The result of the confirmation strategy (e.g., tool output, rejection message, etc.). 

107 """ 

108 ... 

109 

110 def to_dict(self) -> dict[str, Any]: 

111 """Serialize the strategy to a dictionary.""" 

112 ... 

113 

114 @classmethod 

115 def from_dict(cls, data: dict[str, Any]) -> "ConfirmationStrategy": 

116 """Deserialize the strategy from a dictionary.""" 

117 ...