JsonSchemaValidator

Validate the JSON content of ChatMessage objects against a specified JSON Schema.

Key Features

• Validates the last message in a list of ChatMessage objects against a JSON schema
• Routes valid messages to the validated output and invalid ones to validation_error
• Supports custom error message templates for formatting validation errors
• Enables LLM recovery loops in Haystack pipelines by feeding error messages back to the LLM
• Accepts a JSON schema at init time or at run time
• Compatible with OpenAI and other LLMs configured to return JSON output

Configuration

1. Drag the JsonSchemaValidator component onto the canvas from the Component Library.
2. Click the component to open the configuration panel.
3. Configure the parameters as needed. Provide a json_schema to validate against and optionally an error_template to customize validation error messages.

Connections

JsonSchemaValidator accepts a list of ChatMessage objects and an optional JSON schema and error template as input. It routes valid messages to validated and invalid messages to validation_error. It is typically used after a ChatGenerator in a recovery loop, where validation_error messages are fed back to the LLM via a BranchJoiner for correction.

Usage Example

from typing import List

from haystack import Pipeline
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.joiners import BranchJoiner
from haystack.components.validators import JsonSchemaValidator
from haystack import component
from haystack.dataclasses import ChatMessage

@component
class MessageProducer:

    @component.output_types(messages=List[ChatMessage])
    def run(self, messages: List[ChatMessage]) -> dict:
        return {"messages": messages}

p = Pipeline()
p.add_component("llm", OpenAIChatGenerator(model="gpt-4-1106-preview",
                                           generation_kwargs={"response_format": {"type": "json_object"}}))
p.add_component("schema_validator", JsonSchemaValidator())
p.add_component("joiner_for_llm", BranchJoiner(List[ChatMessage]))
p.add_component("message_producer", MessageProducer())

p.connect("message_producer.messages", "joiner_for_llm")
p.connect("joiner_for_llm", "llm")
p.connect("llm.replies", "schema_validator.messages")
p.connect("schema_validator.validation_error", "joiner_for_llm")

result = p.run(data={
    "message_producer": {
        "messages":[ChatMessage.from_user("Generate JSON for person with name 'John' and age 30")]},
        "schema_validator": {
            "json_schema": {
                "type": "object",
                "properties": {"name": {"type": "string"},
                "age": {"type": "integer"}
            }
        }
    }
})
print(result)
>> {'schema_validator': {'validated': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>,
_content=[TextContent(text="\n{\n  "name": "John",\n  "age": 30\n}")],
_name=None, _meta={'model': 'gpt-4-1106-preview', 'index': 0,
'finish_reason': 'stop', 'usage': {'completion_tokens': 17, 'prompt_tokens': 20, 'total_tokens': 37}})]}}

components:
  JsonSchemaValidator:
    type: components.validators.json_schema.JsonSchemaValidator
    init_parameters:

Parameters

Inputs

Parameter	Type	Default	Description
messages	List[ChatMessage]		A list of ChatMessage instances to be validated. The last message in this list is the one that is validated.
json_schema	Optional[Dict[str, Any]]	None	A dictionary representing the JSON schema against which the messages' content is validated. If not provided, the schema from the component init is used.
error_template	Optional[str]	None	A custom template string for formatting the error message in case of validation. If not provided, the error_template from the component init is used.

Outputs

Parameter	Type	Default	Description
validated	List[ChatMessage]		A dictionary with the following keys: - "validated": A list of messages if the last message is valid. - "validation_error": A list of messages if the last message is invalid.
validation_error	List[ChatMessage]		A dictionary with the following keys: - "validated": A list of messages if the last message is valid. - "validation_error": A list of messages if the last message is invalid.

Init Parameters

These are the parameters you can configure in Pipeline Builder:

Parameter	Type	Default	Description
json_schema	Optional[Dict[str, Any]]	None	A dictionary representing the JSON schema against which the messages' content is validated.
error_template	Optional[str]	None	A custom template string for formatting the error message in case of validation failure.

Run Method Parameters

These are the parameters you can configure for the component's run() method. This means you can pass these parameters at query time through the API, in Playground, or when running a job. For details, see Modify Pipeline Parameters at Query Time.

Parameter	Type	Default	Description
messages	List[ChatMessage]		A list of ChatMessage instances to be validated. The last message in this list is the one that is validated.
json_schema	Optional[Dict[str, Any]]	None	A dictionary representing the JSON schema against which the messages' content is validated. If not provided, the schema from the component init is used.
error_template	Optional[str]	None	A custom template string for formatting the error message in case of validation. If not provided, the error_template from the component init is used.