Coverage for haystack/components/validators/json_schema.py: 90%
73 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
5import json
6from typing import Any
8from jsonschema import ValidationError, validate
10from haystack import component
11from haystack.dataclasses import ChatMessage
14def is_valid_json(s: str) -> bool:
15 """
16 Check if the provided string is a valid JSON.
18 :param s: The string to be checked.
19 :returns: `True` if the string is a valid JSON; otherwise, `False`.
20 """
21 try:
22 json.loads(s)
23 except ValueError:
24 return False
25 return True
28@component
29class JsonSchemaValidator:
30 """
31 Validates JSON content of `ChatMessage` against a specified [JSON Schema](https://json-schema.org/).
33 If JSON content of a message conforms to the provided schema, the message is passed along the "validated" output.
34 If the JSON content does not conform to the schema, the message is passed along the "validation_error" output.
35 In the latter case, the error message is constructed using the provided `error_template` or a default template.
36 These error ChatMessages can be used by LLMs in Haystack 2.x recovery loops.
38 Usage example:
40 ```python
41 from haystack import Pipeline
42 from haystack.components.generators.chat import OpenAIResponsesChatGenerator
43 from haystack.components.joiners import BranchJoiner
44 from haystack.components.validators import JsonSchemaValidator
45 from haystack import component
46 from haystack.dataclasses import ChatMessage
49 @component
50 class MessageProducer:
52 @component.output_types(messages=list[ChatMessage])
53 def run(self, messages: list[ChatMessage]) -> dict:
54 return {"messages": messages}
57 p = Pipeline()
58 p.add_component(
59 "llm", OpenAIResponsesChatGenerator(generation_kwargs={"text": {"format": {"type": "json_object"}}})
60 )
61 p.add_component("schema_validator", JsonSchemaValidator())
62 p.add_component("joiner_for_llm", BranchJoiner(list[ChatMessage]))
63 p.add_component("message_producer", MessageProducer())
65 p.connect("message_producer.messages", "joiner_for_llm")
66 p.connect("joiner_for_llm", "llm")
67 p.connect("llm.replies", "schema_validator.messages")
68 p.connect("schema_validator.validation_error", "joiner_for_llm")
70 result = p.run(data={
71 "message_producer": {
72 "messages":[ChatMessage.from_user("Generate JSON for person with name 'John' and age 30")]},
73 "schema_validator": {
74 "json_schema": {
75 "type": "object",
76 "properties": {"name": {"type": "string"},
77 "age": {"type": "integer"}
78 }
79 }
80 }
81 })
82 print(result)
83 # >> {'schema_validator': {'validated': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
84 # _content=[TextContent(text="\\n{\\n "name": "John",\\n "age": 30\\n}")],
85 # _name=None, _meta={'index': 0, 'finish_reason': 'stop', 'usage': {'completion_tokens': 17, 'prompt_tokens': 20,
86 # 'total_tokens': 37}})]}}
87 ```
88 """
90 # Default error description template
91 default_error_template = (
92 "The following generated JSON does not conform to the provided schema.\n"
93 "Generated JSON: {failing_json}\n"
94 "Error details:\n- Message: {error_message}\n"
95 "- Error Path in JSON: {error_path}\n"
96 "- Schema Path: {error_schema_path}\n"
97 "Please match the following schema:\n"
98 "{json_schema}\n"
99 "and provide the corrected JSON content ONLY. Please do not output anything else than the raw corrected "
100 "JSON string, this is the most important part of the task. Don't use any markdown and don't add any comment."
101 )
103 def __init__(self, json_schema: dict[str, Any] | None = None, error_template: str | None = None) -> None:
104 """
105 Initialize the JsonSchemaValidator component.
107 :param json_schema: A dictionary representing the [JSON schema](https://json-schema.org/) against which
108 the messages' content is validated.
109 :param error_template: A custom template string for formatting the error message in case of validation failure.
110 """
111 self.json_schema = json_schema
112 self.error_template = error_template
114 @component.output_types(validated=list[ChatMessage], validation_error=list[ChatMessage])
115 def run(
116 self, messages: list[ChatMessage], json_schema: dict[str, Any] | None = None, error_template: str | None = None
117 ) -> dict[str, list[ChatMessage]]:
118 """
119 Validates the last of the provided messages against the specified json schema.
121 If it does, the message is passed along the "validated" output. If it does not, the message is passed along
122 the "validation_error" output.
124 :param messages: A list of ChatMessage instances to be validated. The last message in this list is the one
125 that is validated.
126 :param json_schema: A dictionary representing the [JSON schema](https://json-schema.org/)
127 against which the messages' content is validated. If not provided, the schema from the component init
128 is used.
129 :param error_template: A custom template string for formatting the error message in case of validation. If not
130 provided, the `error_template` from the component init is used.
131 :return: A dictionary with the following keys:
132 - "validated": A list of messages if the last message is valid.
133 - "validation_error": A list of messages if the last message is invalid.
134 :raises ValueError: If `messages` is empty, if the last message has no text content, or if no JSON schema is
135 provided either in the `run` method or in the component init.
136 """
137 if not messages:
138 raise ValueError("The provided list of messages is empty.")
140 last_message = messages[-1]
141 if last_message.text is None:
142 raise ValueError(f"The provided ChatMessage has no text. ChatMessage: {last_message}")
143 if not is_valid_json(last_message.text):
144 return {
145 "validation_error": [
146 ChatMessage.from_user(
147 f"The message '{last_message.text}' is not a valid JSON object. "
148 f"Please provide only a valid JSON object in string format."
149 f"Don't use any markdown and don't add any comment."
150 )
151 ]
152 }
154 last_message_content = json.loads(last_message.text)
155 json_schema = self.json_schema if json_schema is None else json_schema
156 error_template = error_template or self.error_template or self.default_error_template
158 if json_schema is None:
159 raise ValueError("Provide a JSON schema for validation either in the run method or in the component init.")
160 # fc payload is json object but subtree `parameters` is string - we need to convert to json object
161 # we need complete json to validate it against schema
162 last_message_json = self._recursive_json_to_object(last_message_content)
163 using_openai_schema: bool = self._is_openai_function_calling_schema(json_schema)
164 if using_openai_schema:
165 validation_schema = json_schema["parameters"]
166 else:
167 validation_schema = json_schema
168 try:
169 last_message_json = [last_message_json] if not isinstance(last_message_json, list) else last_message_json
170 for content in last_message_json:
171 if using_openai_schema:
172 validate(instance=content["function"]["arguments"], schema=validation_schema)
173 else:
174 validate(instance=content, schema=validation_schema)
176 return {"validated": [last_message]}
177 except ValidationError as e:
178 error_path = " -> ".join(map(str, e.absolute_path)) if e.absolute_path else "N/A"
179 error_schema_path = " -> ".join(map(str, e.absolute_schema_path)) if e.absolute_schema_path else "N/A"
181 error_template = error_template or self.default_error_template
183 recovery_prompt = self._construct_error_recovery_message(
184 error_template, str(e), error_path, error_schema_path, validation_schema, failing_json=last_message.text
185 )
186 return {"validation_error": [ChatMessage.from_user(recovery_prompt)]}
188 def _construct_error_recovery_message(
189 self,
190 error_template: str,
191 error_message: str,
192 error_path: str,
193 error_schema_path: str,
194 json_schema: dict[str, Any],
195 failing_json: str,
196 ) -> str:
197 """
198 Constructs an error recovery message using a specified template or the default one if none is provided.
200 :param error_template: A custom template string for formatting the error message in case of validation failure.
201 :param error_message: The error message returned by the JSON schema validator.
202 :param error_path: The path in the JSON content where the error occurred.
203 :param error_schema_path: The path in the JSON schema where the error occurred.
204 :param json_schema: The JSON schema against which the content is validated.
205 :param failing_json: The generated invalid JSON string.
206 """
207 error_template = error_template or self.default_error_template
209 return error_template.format(
210 error_message=error_message,
211 error_path=error_path,
212 error_schema_path=error_schema_path,
213 json_schema=json_schema,
214 failing_json=failing_json,
215 )
217 def _is_openai_function_calling_schema(self, json_schema: dict[str, Any]) -> bool:
218 """
219 Checks if the provided schema is a valid OpenAI function calling schema.
221 :param json_schema: The JSON schema to check
222 :return: `True` if the schema is a valid OpenAI function calling schema; otherwise, `False`.
223 """
224 return all(key in json_schema for key in ["name", "description", "parameters"])
226 def _recursive_json_to_object(self, data: Any) -> Any:
227 """
228 Convert any string values that are valid JSON objects into dictionary objects.
230 Returns a new data structure.
232 :param data: The data structure to be traversed.
233 :return: A new data structure with JSON strings converted to dictionary objects.
234 """
235 if isinstance(data, list):
236 return [self._recursive_json_to_object(item) for item in data]
238 if isinstance(data, dict):
239 new_dict = {}
240 for key, value in data.items():
241 if isinstance(value, str):
242 try:
243 json_value = json.loads(value)
244 if isinstance(json_value, (dict, list)):
245 new_dict[key] = self._recursive_json_to_object(json_value)
246 else:
247 new_dict[key] = value # Preserve the original string value
248 except json.JSONDecodeError:
249 new_dict[key] = value
250 elif isinstance(value, dict):
251 new_dict[key] = self._recursive_json_to_object(value)
252 else:
253 new_dict[key] = value
254 return new_dict
256 # If it's neither a list nor a dictionary, return the value directly
257 return data