Coverage for haystack/components/evaluators/faithfulness.py: 100%

50 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-12 09:42 +0000

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

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import math 

6from typing import Any 

7 

8from numpy import mean as np_mean 

9 

10from haystack import component, default_from_dict, default_to_dict, logging 

11from haystack.components.evaluators.llm_evaluator import LLMEvaluator 

12from haystack.components.generators.chat.types import ChatGenerator 

13from haystack.core.serialization import component_to_dict 

14from haystack.utils import deserialize_chatgenerator_inplace 

15 

16logger = logging.getLogger(__name__) 

17 

18# Default examples to include in the prompt if the user does not provide any examples 

19_DEFAULT_EXAMPLES = [ 

20 { 

21 "inputs": { 

22 "questions": "What is the capital of Germany and when was it founded?", 

23 "contexts": ["Berlin is the capital of Germany and was founded in 1244."], 

24 "predicted_answers": "The capital of Germany, Berlin, was founded in the 13th century.", 

25 }, 

26 "outputs": { 

27 "statements": ["Berlin is the capital of Germany.", "Berlin was founded in 1244."], 

28 "statement_scores": [1, 1], 

29 }, 

30 }, 

31 { 

32 "inputs": { 

33 "questions": "What is the capital of France?", 

34 "contexts": ["Berlin is the capital of Germany."], 

35 "predicted_answers": "Paris", 

36 }, 

37 "outputs": {"statements": ["Paris is the capital of France."], "statement_scores": [0]}, 

38 }, 

39 { 

40 "inputs": { 

41 "questions": "What is the capital of Italy?", 

42 "contexts": ["Rome is the capital of Italy."], 

43 "predicted_answers": "Rome is the capital of Italy with more than 4 million inhabitants.", 

44 }, 

45 "outputs": { 

46 "statements": ["Rome is the capital of Italy.", "Rome has more than 4 million inhabitants."], 

47 "statement_scores": [1, 0], 

48 }, 

49 }, 

50] 

51 

52 

53@component 

54class FaithfulnessEvaluator(LLMEvaluator): 

55 """ 

56 Evaluator that checks if a generated answer can be inferred from the provided contexts. 

57 

58 An LLM separates the answer into multiple statements and checks whether the statement can be inferred from the 

59 context or not. The final score for the full answer is a number from 0.0 to 1.0. It represents the proportion of 

60 statements that can be inferred from the provided contexts. 

61 

62 Usage example: 

63 ```python 

64 from haystack.components.evaluators import FaithfulnessEvaluator 

65 

66 questions = ["Who created the Python language?"] 

67 contexts = [ 

68 [( 

69 "Python, created by Guido van Rossum in the late 1980s, is a high-level general-purpose programming " 

70 "language. Its design philosophy emphasizes code readability, and its language constructs aim to help " 

71 "programmers write clear, logical code for both small and large-scale software projects." 

72 )], 

73 ] 

74 predicted_answers = [ 

75 "Python is a high-level general-purpose programming language that was created by George Lucas." 

76 ] 

77 evaluator = FaithfulnessEvaluator() 

78 result = evaluator.run(questions=questions, contexts=contexts, predicted_answers=predicted_answers) 

79 

80 print(result["individual_scores"]) 

81 # [0.5] 

82 print(result["score"]) 

83 # 0.5 

84 print(result["results"]) 

85 # [{'statements': ['Python is a high-level general-purpose programming language.', 

86 # 'Python was created by George Lucas.'], 'statement_scores': [1, 0], 'score': 0.5, 

87 # 'status': 'evaluated'}] 

88 ``` 

89 """ 

90 

91 def __init__( 

92 self, 

93 examples: list[dict[str, Any]] | None = None, 

94 progress_bar: bool = True, 

95 raise_on_failure: bool = True, 

96 chat_generator: ChatGenerator | None = None, 

97 ) -> None: 

98 """ 

99 Creates an instance of FaithfulnessEvaluator. 

100 

101 If no LLM is specified using the `chat_generator` parameter, the component will use OpenAI in JSON mode. 

102 

103 :param examples: 

104 Optional few-shot examples conforming to the expected input and output format of FaithfulnessEvaluator. 

105 Default examples will be used if none are provided. 

106 Each example must be a dictionary with keys "inputs" and "outputs". 

107 "inputs" must be a dictionary with keys "questions", "contexts", and "predicted_answers". 

108 "outputs" must be a dictionary with "statements" and "statement_scores". 

109 Expected format: 

110 ```python 

111 [{ 

112 "inputs": { 

113 "questions": "What is the capital of Italy?", "contexts": ["Rome is the capital of Italy."], 

114 "predicted_answers": "Rome is the capital of Italy with more than 4 million inhabitants.", 

115 }, 

116 "outputs": { 

117 "statements": ["Rome is the capital of Italy.", "Rome has more than 4 million inhabitants."], 

118 "statement_scores": [1, 0], 

119 }, 

120 }] 

121 ``` 

122 :param progress_bar: 

123 Whether to show a progress bar during the evaluation. 

124 :param raise_on_failure: 

125 If True, the component raises an exception when the chat generator call fails or its response isn't valid 

126 JSON with the expected keys. If False, it logs a warning and sets the result for that input to `None`. 

127 :param chat_generator: 

128 a ChatGenerator instance which represents the LLM. 

129 In order for the component to work, the LLM should be configured to return a JSON object. For example, 

130 when using the OpenAIChatGenerator, you should pass `{"response_format": {"type": "json_object"}}` in the 

131 `generation_kwargs`. 

132 """ 

133 self.instructions = ( 

134 "Your task is to judge the faithfulness or groundedness of statements based " 

135 "on context information. First, please extract statements from a provided " 

136 "predicted answer to a question. Second, calculate a faithfulness score for each " 

137 "statement made in the predicted answer. The score is 1 if the statement can be " 

138 "inferred from the provided context or 0 if it cannot be inferred." 

139 ) 

140 self.inputs = [("questions", list[str]), ("contexts", list[list[str]]), ("predicted_answers", list[str])] 

141 self.outputs = ["statements", "statement_scores"] 

142 self.examples = examples or _DEFAULT_EXAMPLES 

143 

144 super(FaithfulnessEvaluator, self).__init__( # noqa: UP008 

145 instructions=self.instructions, 

146 inputs=self.inputs, 

147 outputs=self.outputs, 

148 examples=self.examples, 

149 chat_generator=chat_generator, 

150 raise_on_failure=raise_on_failure, 

151 progress_bar=progress_bar, 

152 ) 

153 

154 @component.output_types( 

155 individual_scores=list[float], score=float, results=list[dict[str, Any]], meta=list[dict[str, Any]] | None 

156 ) 

157 def run(self, **inputs: Any) -> dict[str, Any]: 

158 """ 

159 Run the LLM evaluator. 

160 

161 :param questions: 

162 A list of questions. 

163 :param contexts: 

164 A nested list of contexts that correspond to the questions. 

165 :param predicted_answers: 

166 A list of predicted answers. 

167 :returns: 

168 A dictionary with the following outputs: 

169 - `score`: Mean faithfulness score over all the provided input answers. 

170 - `individual_scores`: A list of faithfulness scores for each input answer. 

171 - `results`: A list of dictionaries with `statements`, `statement_scores`, `score`, and `status` for 

172 each input answer. `status` is `evaluated` for valid results and `error` for failed evaluations. 

173 """ 

174 result = super(FaithfulnessEvaluator, self).run(**inputs) # noqa: UP008 

175 # Post-process the raw results to calculate relevance metrics and scores 

176 return self._postprocess_results(result) 

177 

178 @component.output_types( 

179 individual_scores=list[float], score=float, results=list[dict[str, Any]], meta=list[dict[str, Any]] | None 

180 ) 

181 async def run_async(self, **inputs: Any) -> dict[str, Any]: 

182 """ 

183 Run the LLM evaluator asynchronously. 

184 

185 :param questions: 

186 A list of questions. 

187 :param contexts: 

188 A nested list of contexts that correspond to the questions. 

189 :param predicted_answers: 

190 A list of predicted answers. 

191 :returns: 

192 A dictionary with the following outputs: 

193 - `score`: Mean faithfulness score over all the provided input answers. 

194 - `individual_scores`: A list of faithfulness scores for each input answer. 

195 - `results`: A list of dictionaries with `statements`, `statement_scores`, `score`, and `status` for 

196 each input answer. `status` is `evaluated` for valid results and `error` for failed evaluations. 

197 """ 

198 result = await super(FaithfulnessEvaluator, self).run_async(**inputs) # noqa: UP008 

199 # Post-process the raw results to calculate relevance metrics and scores 

200 return self._postprocess_results(result) 

201 

202 def _postprocess_results(self, result: dict[str, Any]) -> dict[str, Any]: 

203 """ 

204 Post-processes raw LLM evaluator outputs to compute faithfulness scores. 

205 

206 Calculates statement-level score averages, computes the overall mean faithfulness 

207 score across successful queries, and updates the result payload. 

208 

209 :param result: 

210 The raw evaluation dictionary from the base LLM evaluator. 

211 :returns: 

212 The updated dictionary containing final scores and tracking metrics. 

213 """ 

214 

215 # calculate average statement faithfulness score per query 

216 for idx, res in enumerate(result["results"]): 

217 if res is None: 

218 result["results"][idx] = { 

219 "statements": [], 

220 "statement_scores": [], 

221 "score": float("nan"), 

222 "status": "error", 

223 } 

224 continue 

225 res["status"] = "evaluated" 

226 if not res["statements"]: 

227 res["score"] = 0 

228 else: 

229 res["score"] = np_mean(res["statement_scores"]) 

230 

231 # calculate average answer faithfulness score over all queries 

232 scores = [res["score"] for res in result["results"]] 

233 valid_scores = [s for s in scores if not math.isnan(s)] 

234 skipped = len(scores) - len(valid_scores) 

235 if skipped: 

236 logger.warning("{skipped} query(s) failed and were excluded from the score.", skipped=skipped) 

237 result["score"] = np_mean(valid_scores) if valid_scores else float("nan") 

238 result["individual_scores"] = scores 

239 

240 return result 

241 

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

243 """ 

244 Serialize this component to a dictionary. 

245 

246 :returns: 

247 A dictionary with serialized data. 

248 """ 

249 return default_to_dict( 

250 self, 

251 chat_generator=component_to_dict(obj=self._chat_generator, name="chat_generator"), 

252 examples=self.examples, 

253 progress_bar=self.progress_bar, 

254 raise_on_failure=self.raise_on_failure, 

255 ) 

256 

257 @classmethod 

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

259 """ 

260 Deserialize this component from a dictionary. 

261 

262 :param data: 

263 The dictionary representation of this component. 

264 :returns: 

265 The deserialized component instance. 

266 """ 

267 if data["init_parameters"].get("chat_generator"): 

268 deserialize_chatgenerator_inplace(data["init_parameters"], key="chat_generator") 

269 return default_from_dict(cls, data)