Coverage for haystack/components/evaluators/sas_evaluator.py: 56%

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 numpy import mean as np_mean 

8 

9from haystack import component, default_from_dict, default_to_dict 

10from haystack.lazy_imports import LazyImport 

11from haystack.utils import ComponentDevice, expit 

12from haystack.utils.auth import Secret 

13 

14with LazyImport(message="Run 'pip install \"sentence-transformers>=5.0.0\"'") as sas_import: 

15 from sentence_transformers import CrossEncoder, SentenceTransformer, util 

16 from transformers import AutoConfig 

17 

18 

19@component 

20class SASEvaluator: 

21 """ 

22 SASEvaluator computes the Semantic Answer Similarity (SAS) between a list of predictions and a one of ground truths. 

23 

24 It's usually used in Retrieval Augmented Generation (RAG) pipelines to evaluate the quality of the generated 

25 answers. The SAS is computed using a pre-trained model from the Hugging Face model hub. The model can be either a 

26 Bi-Encoder or a Cross-Encoder. The choice of the model is based on the `model` parameter. 

27 

28 Usage example: 

29 ```python 

30 from haystack.components.evaluators.sas_evaluator import SASEvaluator 

31 

32 evaluator = SASEvaluator(model="cross-encoder/ms-marco-MiniLM-L-6-v2") 

33 ground_truths = [ 

34 "A construction budget of US $2.3 billion", 

35 "The Eiffel Tower, completed in 1889, symbolizes Paris's cultural magnificence.", 

36 "The Meiji Restoration in 1868 transformed Japan into a modernized world power.", 

37 ] 

38 predictions = [ 

39 "A construction budget of US $2.3 billion", 

40 "The Eiffel Tower, completed in 1889, symbolizes Paris's cultural magnificence.", 

41 "The Meiji Restoration in 1868 transformed Japan into a modernized world power.", 

42 ] 

43 result = evaluator.run( 

44 ground_truth_answers=ground_truths, predicted_answers=predictions 

45 ) 

46 

47 print(result["score"]) 

48 # 0.9999673763910929 

49 

50 print(result["individual_scores"]) 

51 # [0.9999765157699585, 0.999968409538269, 0.9999572038650513] 

52 ``` 

53 """ 

54 

55 def __init__( 

56 self, 

57 model: str = "sentence-transformers/paraphrase-multilingual-mpnet-base-v2", 

58 batch_size: int = 32, 

59 device: ComponentDevice | None = None, 

60 token: Secret = Secret.from_env_var(["HF_API_TOKEN", "HF_TOKEN"], strict=False), 

61 ) -> None: 

62 """ 

63 Creates a new instance of SASEvaluator. 

64 

65 :param model: 

66 SentenceTransformers semantic textual similarity model, should be path or string pointing to a downloadable 

67 model. 

68 :param batch_size: 

69 Number of prediction-label pairs to encode at once. 

70 :param device: 

71 The device on which the model is loaded. If `None`, the default device is automatically selected. 

72 :param token: 

73 The Hugging Face token for HTTP bearer authorization. 

74 You can find your HF token in your [account settings](https://huggingface.co/settings/tokens) 

75 """ 

76 sas_import.check() 

77 

78 self._model = model 

79 self._batch_size = batch_size 

80 self._device = device 

81 self._token = token 

82 self._similarity_model: SentenceTransformer | CrossEncoder | None = None 

83 

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

85 """ 

86 Serialize this component to a dictionary. 

87 

88 :returns: 

89 The serialized component as a dictionary. 

90 """ 

91 return default_to_dict( 

92 self, model=self._model, batch_size=self._batch_size, device=self._device, token=self._token 

93 ) 

94 

95 @classmethod 

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

97 """ 

98 Deserialize this component from a dictionary. 

99 

100 :param data: 

101 The dictionary representation of this component. 

102 :returns: 

103 The deserialized component instance. 

104 """ 

105 return default_from_dict(cls, data) 

106 

107 def warm_up(self) -> None: 

108 """ 

109 Initializes the component. 

110 """ 

111 if self._similarity_model: 

112 return 

113 

114 token = self._token.resolve_value() if self._token else None 

115 config = AutoConfig.from_pretrained(self._model, use_auth_token=token) 

116 cross_encoder_used = False 

117 if config.architectures: 

118 cross_encoder_used = any(arch.endswith("ForSequenceClassification") for arch in config.architectures) 

119 device = ComponentDevice.resolve_device(self._device).to_torch_str() 

120 # Based on the Model string we can load either Bi-Encoders or Cross Encoders. 

121 # Similarity computation changes for both approaches 

122 if cross_encoder_used: 

123 self._similarity_model = CrossEncoder(self._model, device=device, token=token) 

124 else: 

125 self._similarity_model = SentenceTransformer(self._model, device=device, token=token) 

126 

127 @component.output_types(score=float, individual_scores=list[float]) 

128 def run(self, ground_truth_answers: list[str], predicted_answers: list[str]) -> dict[str, float | list[float]]: 

129 """ 

130 SASEvaluator component run method. 

131 

132 Run the SASEvaluator to compute the Semantic Answer Similarity (SAS) between a list of predicted answers 

133 and a list of ground truth answers. Both must be list of strings of same length. 

134 

135 :param ground_truth_answers: 

136 A list of expected answers for each question. 

137 :param predicted_answers: 

138 A list of generated answers for each question. 

139 :returns: 

140 A dictionary with the following outputs: 

141 - `score`: Mean SAS score over all the predictions/ground-truth pairs. 

142 - `individual_scores`: A list of similarity scores for each prediction/ground-truth pair. 

143 """ 

144 if len(ground_truth_answers) != len(predicted_answers): 

145 raise ValueError("The number of predictions and labels must be the same.") 

146 

147 if any(answer is None for answer in predicted_answers): 

148 raise ValueError("Predicted answers must not contain None values.") 

149 

150 if len(predicted_answers) == 0: 

151 return {"score": 0.0, "individual_scores": [0.0]} 

152 

153 if not self._similarity_model: 

154 self.warm_up() 

155 

156 if isinstance(self._similarity_model, CrossEncoder): 

157 # For Cross Encoders we create a list of pairs of predictions and labels 

158 sentence_pairs = list(zip(predicted_answers, ground_truth_answers, strict=True)) 

159 similarity_scores = self._similarity_model.predict( 

160 sentence_pairs, batch_size=self._batch_size, convert_to_numpy=True 

161 ) 

162 

163 # All Cross Encoders do not return a set of logits scores that are normalized 

164 # We normalize scores if they are larger than 1 

165 if (similarity_scores > 1).any(): 

166 similarity_scores = expit(similarity_scores) 

167 

168 # Convert scores to list of floats from numpy array 

169 similarity_scores = similarity_scores.tolist() 

170 

171 elif isinstance(self._similarity_model, SentenceTransformer): 

172 # For Bi-encoders we create embeddings separately for predictions and labels 

173 predictions_embeddings = self._similarity_model.encode( 

174 predicted_answers, batch_size=self._batch_size, convert_to_tensor=True 

175 ) 

176 label_embeddings = self._similarity_model.encode( 

177 ground_truth_answers, batch_size=self._batch_size, convert_to_tensor=True 

178 ) 

179 

180 # Compute cosine-similarities 

181 similarity_scores = [ 

182 float(util.cos_sim(pred_embedding, label_embedding).cpu().squeeze().numpy()) 

183 for pred_embedding, label_embedding in zip(predictions_embeddings, label_embeddings, strict=True) 

184 ] 

185 

186 sas_score = np_mean(similarity_scores) 

187 

188 return {"score": sas_score, "individual_scores": similarity_scores}