agents.rag.db_rag.graph_db.models¶

Pydantic models for structured outputs in the Graph DB RAG Agent.

This module defines the structured output models used by various LLM engines in the Graph DB RAG workflow. These models ensure type safety and validation for LLM responses.

Example

Using the models for structured LLM outputs:

>>> from haive.agents.rag.db_rag.graph_db.models import CypherQueryOutput
>>>
>>> # Create a Cypher query output
>>> cypher_output = CypherQueryOutput(
...     query="MATCH (m:Movie) WHERE m.year = $year RETURN m.title",
...     parameters={"year": 2023}
... )
>>> print(cypher_output.query)
MATCH (m:Movie) WHERE m.year = $year RETURN m.title

Classes¶

`Config`	Top-level configuration class for Graph DB RAG models.
`CypherQueryOutput`	Structured output for Cypher query generation.
`GuardrailsOutput`	Output for domain relevance checking.
`PropertyFilter`	Represents a filter condition on a node property in a Cypher query.
`ValidateCypherOutput`	Validation result for a Cypher query.

Functions¶

`validate_cypher_syntax`(query)	Validate Cypher query syntax.
`validate_decision`(decision, allowed_values)	Validate decision against allowed values.
`validate_filter_type`(filter_type)	Validate filter type for property filtering.

Module Contents¶

class agents.rag.db_rag.graph_db.models.Config(/, **data)¶

Bases: pydantic.BaseModel

Top-level configuration class for Graph DB RAG models.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

class agents.rag.db_rag.graph_db.models.CypherQueryOutput(/, **data)¶

Bases: pydantic.BaseModel

Structured output for Cypher query generation.

This model ensures that generated Cypher queries are properly formatted and optionally include parameters for parameterized queries.

query¶: The generated Cypher query string. Must start with a valid Cypher keyword (MATCH, CREATE, etc.).

parameters¶: Optional dictionary of query parameters for parameterized queries. Keys are parameter names (without $), values are the parameter values.

Example

>>> # Simple query without parameters
>>> output = CypherQueryOutput(
...     query="MATCH (m:Movie) RETURN m.title LIMIT 10"
... )

>>> # Parameterized query
>>> output = CypherQueryOutput(
...     query="MATCH (m:Movie) WHERE m.year = $year RETURN m.title",
...     parameters={"year": 2023}
... )

Raises:: ValueError – If the query doesn’t start with a valid Cypher keyword.
Parameters:: data (Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod validate_cypher_syntax(query)¶

Validate that the query starts with a valid Cypher keyword.

Parameters:: query (str) – The Cypher query string to validate.
Returns:: The validated query string.
Return type:: str
Raises:: ValueError – If the query doesn’t start with a valid keyword.

class agents.rag.db_rag.graph_db.models.GuardrailsOutput(/, **data)¶

Bases: pydantic.BaseModel

Output for domain relevance checking.

This model represents the decision on whether a query is relevant to the configured domain. It supports multiple categories within a domain.

Parameters:: data (Any)

decision¶: The routing decision. Either “end” (not relevant) or one of the allowed categories (relevant to that category).

allowed_categories¶: List of valid categories for the domain. The “end” option is always implicitly included.

Example

>>> # Query about movies in a movie domain
>>> output = GuardrailsOutput(
...     decision="movie",
...     allowed_categories=["movie", "actor", "director"]
... )

>>> # Query not relevant to the domain
>>> output = GuardrailsOutput(
...     decision="end",
...     allowed_categories=["movie", "actor", "director"]
... )

Note

The validate_decision method should be called after instantiation to ensure the decision is valid.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

class Config¶: Pydantic model configuration.

validate_decision()¶

Validate that the decision is within allowed values.

Raises:: ValueError – If the decision is not ‘end’ or in allowed_categories.
Return type:: None

Example

>>> output = GuardrailsOutput(decision="movie")
>>> output.validate_decision()  # No error

>>> output = GuardrailsOutput(decision="invalid")
>>> output.validate_decision()  # Raises ValueError

class agents.rag.db_rag.graph_db.models.PropertyFilter(/, **data)¶

Bases: pydantic.BaseModel

Represents a filter condition on a node property in a Cypher query.

This model captures property-based filtering conditions that appear in WHERE clauses or inline property matches in Cypher queries.

node_label¶: The Neo4j label of the node being filtered (e.g., “Movie”, “Person”).

property_key¶: The property name being filtered (e.g., “title”, “year”).

property_value¶: The value to match against. Can be string, number, or boolean.

filter_type¶: The comparison operator used. Defaults to equality.

Example

>>> # Filter for movies released after 2020
>>> filter = PropertyFilter(
...     node_label="Movie",
...     property_key="year",
...     property_value=2020,
...     filter_type=">"
... )

>>> # Filter for person named "Keanu Reeves"
>>> filter = PropertyFilter(
...     node_label="Person",
...     property_key="name",
...     property_value="Keanu Reeves"
... )

Raises:: ValueError – If filter_type is not one of the valid operators.
Parameters:: data (Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod validate_filter_type(v)¶

Validate that the filter type is a supported operator.

Parameters:: v – The filter type value to validate.
Returns:: The validated filter type.
Return type:: str
Raises:: ValueError – If the filter type is not supported.

class agents.rag.db_rag.graph_db.models.ValidateCypherOutput(/, **data)¶

Bases: pydantic.BaseModel

Validation result for a Cypher query.

This model captures the results of validating a Cypher query against the database schema, including any errors found and filters detected.

Parameters:: data (Any)

is_valid¶: Whether the Cypher query is valid and can be executed.

errors¶: List of syntax or semantic errors found. Each error should explain what’s wrong and potentially how to fix it.

filters¶: List of property filters detected in the query. Useful for understanding what the query is filtering on.

Example

>>> # Valid query result
>>> result = ValidateCypherOutput(
...     is_valid=True,
...     errors=[],
...     filters=[PropertyFilter(
...         node_label="Movie",
...         property_key="year",
...         property_value=2023
...     )]
... )

>>> # Invalid query result
>>> result = ValidateCypherOutput(
...     is_valid=False,
...     errors=[
...         "Label 'Film' does not exist in schema. Did you mean 'Movie'?",
...         "Property 'release_date' does not exist for Movie. Use 'year' instead."
...     ]
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

agents.rag.db_rag.graph_db.models.validate_cypher_syntax(query)¶

Validate Cypher query syntax.

Parameters:: query (str) – Cypher query string to validate
Returns:: True if syntax is valid, False otherwise
Return type:: bool

agents.rag.db_rag.graph_db.models.validate_decision(decision, allowed_values)¶

Validate decision against allowed values.

Parameters:

decision (str) – Decision value to validate
allowed_values (list[str]) – List of allowed decision values

Returns:

True if decision is valid, False otherwise

Return type:

bool

agents.rag.db_rag.graph_db.models.validate_filter_type(filter_type)¶

Validate filter type for property filtering.

Parameters:: filter_type (str) – Filter type to validate
Returns:: True if filter type is valid, False otherwise
Return type:: bool