agents.rag.db_rag.graph_db.config¶

Configuration module for the Graph Database RAG Agent.

This module provides configuration classes for connecting to Neo4j databases and configuring the Graph DB RAG Agent with appropriate engines, schemas, and domain-specific settings.

Example

Basic configuration setup:

>>> from haive.agents.rag.db_rag.graph_db.config import GraphDBRAGConfig
>>>
>>> # Create config with default settings
>>> config = GraphDBRAGConfig(
...     domain_name="movies",
...     domain_categories=["movie", "actor", "director"]
... )
>>>
>>> # Create agent with config
>>> agent = GraphDBRAGAgent(config)
Environment Variables:

The following environment variables are used for Neo4j connection:

  • NEO4J_URI: The URI of the Neo4j database (e.g., “bolt://localhost:7687”)

  • NEO4J_USER: Username for Neo4j authentication

  • NEO4J_PASSWORD: Password for Neo4j authentication

  • NEO4J_DATABASE: Database name (optional, defaults to “neo4j”)

Attributes¶

GraphDBAgentConfig

Alias for backward compatibility with older code.

Classes¶

ExampleConfig

Configuration for few-shot examples used in Cypher generation.

GraphDBConfig

Configuration for connecting to a Neo4j graph database.

GraphDBRAGConfig

Main configuration for the Graph Database RAG Agent.

Functions¶

get_graph_db([uri, user, password, database])

Get a Neo4j database connection.

get_graph_db_schema([db_connection])

Get the schema from a Neo4j database.

validate_engines(engines)

Validate that all required engines are properly configured.

Module Contents¶

class agents.rag.db_rag.graph_db.config.ExampleConfig(/, **data)¶

Bases: pydantic.BaseModel

Configuration for few-shot examples used in Cypher generation.

This class manages example queries that help the LLM learn the mapping between natural language questions and Cypher queries for a specific domain.

Parameters:

data (Any)

examples_path¶

Path to JSON file containing examples. The file should contain a list of dicts with “question” and “query” keys.

examples¶

Direct list of examples if not using a file. Each example should have “question” (natural language) and “query” (Cypher) keys.

k¶

Number of examples to retrieve for few-shot prompting. More examples can improve accuracy but increase prompt length.

Example

>>> # Using a file
>>> example_config = ExampleConfig(
...     examples_path="examples/movie_queries.json",
...     k=3
... )
>>>
>>> # Using direct examples
>>> example_config = ExampleConfig(
...     examples=[
...         {
...             "question": "Who directed Inception?",
...             "query": "MATCH (p:Person)-[:DIRECTED]->(m:Movie {title: 'Inception'}) RETURN p.name"
...         },
...         {
...             "question": "What movies did Tom Hanks act in?",
...             "query": "MATCH (p:Person {name: 'Tom Hanks'})-[:ACTED_IN]->(m:Movie) RETURN m.title"
...         }
...     ],
...     k=2
... )

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 agents.rag.db_rag.graph_db.config.GraphDBConfig(/, **data)¶

Bases: pydantic.BaseModel

Configuration for connecting to a Neo4j graph database.

This class manages Neo4j connection parameters and provides methods for establishing connections and retrieving schema information.

Parameters:

data (Any)

graph_db_uri¶

Neo4j connection URI. Defaults to NEO4J_URI env var. Format: “bolt://host:port” or “neo4j://host:port”

graph_db_user¶

Username for authentication. Defaults to NEO4J_USER env var.

graph_db_password¶

Password for authentication. Defaults to NEO4J_PASSWORD env var.

graph_db_database¶

Database name. Defaults to NEO4J_DATABASE env var or “neo4j”.

enhanced_schema¶

Whether to use enhanced schema scanning for better property and relationship detection.

Example

>>> # Using environment variables
>>> db_config = GraphDBConfig()
>>>
>>> # Explicit configuration
>>> db_config = GraphDBConfig(
...     graph_db_uri="bolt://localhost:7687",
...     graph_db_user="neo4j",
...     graph_db_password="password",
...     graph_db_database="movies"
... )
>>>
>>> # Connect to database
>>> graph = db_config.get_graph_db()
>>> if graph:
...     schema = db_config.get_graph_db_schema()

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.

get_graph_db()¶

Create and return a Neo4jGraph connection object.

Establishes a secure connection to the Neo4j database with proper timeout, sanitization, and schema refresh settings.

Returns:

Connected graph object if successful. None: If connection fails.

Return type:

Neo4jGraph

Example

>>> config = GraphDBConfig()
>>> graph = config.get_graph_db()
>>> if graph:
...     print("âś… Connected to Neo4j")
... else:
...     print("❌ Connection failed")
get_graph_db_schema()¶

Retrieve the graph schema from the Neo4j database.

Gets the complete schema including node labels, relationship types, and properties. Uses enhanced schema if enabled for more detailed information.

Returns:

Schema dictionary with structure:
{

“node_props”: {label: [properties]}, “rel_props”: {type: [properties]}, “relationships”: [relationship_structures]

}

None: If connection fails or schema cannot be retrieved.

Return type:

dict

Example

>>> config = GraphDBConfig(enhanced_schema=True)
>>> schema = config.get_graph_db_schema()
>>> if schema:
...     print(f"Node labels: {list(schema['node_props'].keys())}")
...     print(f"Relationship types: {list(schema['rel_props'].keys())}")
class agents.rag.db_rag.graph_db.config.GraphDBRAGConfig¶

Bases: haive.core.engine.agent.agent.AgentConfig

Main configuration for the Graph Database RAG Agent.

This class provides comprehensive configuration for the Graph DB RAG Agent, including LLM engines, domain settings, database connection, and schemas.

engines¶

Dictionary of AugLLMConfig engines for different workflow steps: - “guardrails”: Checks domain relevance - “text2cypher”: Converts natural language to Cypher - “validate_cypher”: Validates generated Cypher - “correct_cypher”: Fixes Cypher errors - “generate_final_answer”: Creates natural language response

domain_name¶

The domain this agent specializes in (e.g., “movies”, “healthcare”). Used for guardrails and example selection.

domain_categories¶

Valid categories within the domain for fine-grained routing.

example_config¶

Configuration for few-shot examples.

state_schema¶

Pydantic model for workflow state management.

graph_db_config¶

Neo4j connection configuration.

input_schema¶

Schema for agent input validation.

output_schema¶

Schema for agent output structure.

domain_examples¶

Domain-specific examples for different categories.

Example

>>> # Configure a movie domain agent
>>> config = GraphDBRAGConfig(
...     domain_name="movies",
...     domain_categories=["movie", "actor", "director", "genre"],
...     example_config=ExampleConfig(
...         examples_path="examples/movie_queries.json",
...         k=3
...     ),
...     graph_db_config=GraphDBConfig(
...         graph_db_uri="bolt://localhost:7687",
...         graph_db_user="neo4j",
...         graph_db_password="password"
...     )
... )
>>>
>>> # Create agent
>>> agent = GraphDBRAGAgent(config)

Note

All required engines must be present in the engines dictionary. The validator will check for their presence and raise an error if any are missing.

classmethod validate_engines(engines)¶

Validate that all required engines are present.

Checks for the presence of all required engine configurations and handles potential naming mismatches (e.g., “generate_cypher” vs “text2cypher”).

Parameters:

engines (dict[str, haive.core.engine.aug_llm.AugLLMConfig]) – Dictionary of engine configurations.

Returns:

Validated engines dictionary.

Return type:

Dict[str, AugLLMConfig]

Raises:

ValueError – If any required engine is missing.

agents.rag.db_rag.graph_db.config.get_graph_db(uri=None, user=None, password=None, database='neo4j')¶

Get a Neo4j database connection.

Parameters:

  • uri (str) – Neo4j URI (defaults to NEO4J_URI env var)

  • user (str) – Username (defaults to NEO4J_USER env var)

  • password (str) – Password (defaults to NEO4J_PASSWORD env var)

  • database (str) – Database name (defaults to “neo4j”)

Returns:

Mock database connection (placeholder implementation)

agents.rag.db_rag.graph_db.config.get_graph_db_schema(db_connection=None)¶

Get the schema from a Neo4j database.

Parameters:

db_connection – Database connection object

Returns:

Dictionary representing the database schema

Return type:

dict

agents.rag.db_rag.graph_db.config.validate_engines(engines)¶

Validate that all required engines are properly configured.

Parameters:

engines (dict) – Dictionary of engine configurations

Returns:

True if all engines are valid, False otherwise

Return type:

bool

agents.rag.db_rag.graph_db.config.GraphDBAgentConfig¶

Alias for backward compatibility with older code.