"""Validation module for knowledge graph consistency."""
import re
from typing import List, Optional, Set
from .interfaces import Entity, KnowledgeGraph, Relation
class ValidationError(Exception):
"""Base class for validation errors."""
pass
class EntityValidationError(ValidationError):
"""Raised when entity validation fails."""
pass
class RelationValidationError(ValidationError):
"""Raised when relation validation fails."""
pass
class KnowledgeGraphValidator:
"""Validator for ensuring knowledge graph consistency."""
# Constants for validation rules
ENTITY_NAME_PATTERN = r"^[a-z][a-z0-9-]{0,99}$"
MAX_OBSERVATION_LENGTH = 500
VALID_ENTITY_TYPES = {
"person",
"concept",
"project",
"document",
"tool",
"organization",
"location",
"event",
}
VALID_RELATION_TYPES = {
"knows",
"contains",
"uses",
"created",
"belongs-to",
"depends-on",
"related-to",
}
@classmethod
def validate_entity_name(cls, name: str) -> None:
"""Validate entity name follows naming convention.
Args:
name: Entity name to validate
Raises:
EntityValidationError: If name is invalid
"""
if not re.match(cls.ENTITY_NAME_PATTERN, name):
raise EntityValidationError(
f"Invalid entity name '{name}'. Must start with lowercase letter, "
"contain only lowercase letters, numbers and hyphens, "
"and be 1-100 characters long."
)
@classmethod
def validate_entity_type(cls, entity_type: str) -> None:
"""Validate entity type is from allowed set.
Args:
entity_type: Entity type to validate
Raises:
EntityValidationError: If type is invalid
"""
if entity_type not in cls.VALID_ENTITY_TYPES:
raise EntityValidationError(
f"Invalid entity type '{entity_type}'. Must be one of: "
f"{', '.join(sorted(cls.VALID_ENTITY_TYPES))}"
)
@classmethod
def validate_observations(cls, observations: List[str]) -> None:
"""Validate entity observations.
Args:
observations: List of observations to validate
Raises:
EntityValidationError: If any observation is invalid
"""
seen = set()
for obs in observations:
if not obs:
raise EntityValidationError("Empty observation")
if len(obs) > cls.MAX_OBSERVATION_LENGTH:
raise EntityValidationError(
f"Observation exceeds length of {cls.MAX_OBSERVATION_LENGTH} chars"
)
if obs in seen:
raise EntityValidationError(f"Duplicate observation: {obs}")
seen.add(obs)
@classmethod
def validate_entity(cls, entity: Entity) -> None:
"""Validate an entity.
Args:
entity: Entity to validate
Raises:
EntityValidationError: If entity is invalid
"""
cls.validate_entity_name(entity.name)
cls.validate_entity_type(entity.entityType)
cls.validate_observations(list(entity.observations))
@classmethod
def validate_relation_type(cls, relation_type: str) -> None:
"""Validate relation type is from allowed set.
Args:
relation_type: Relation type to validate
Raises:
RelationValidationError: If type is invalid
"""
if relation_type not in cls.VALID_RELATION_TYPES:
valid_types = ", ".join(sorted(cls.VALID_RELATION_TYPES))
raise RelationValidationError(
f"Invalid relation type '{relation_type}'. Valid types: {valid_types}"
)
@classmethod
def validate_relation(cls, relation: Relation) -> None:
"""Validate a relation.
Args:
relation: Relation to validate
Raises:
RelationValidationError: If relation is invalid
"""
if relation.from_ == relation.to:
raise RelationValidationError("Self-referential relations not allowed")
cls.validate_relation_type(relation.relationType)
@classmethod
def validate_no_cycles(
cls,
relations: List[Relation],
existing_relations: Optional[List[Relation]] = None,
) -> None:
"""Validate that relations don't create cycles.
Args:
relations: New relations to validate
existing_relations: Optional list of existing relations to check against
Raises:
RelationValidationError: If cycles are detected
"""
# Build adjacency list
graph: dict[str, Set[str]] = {}
all_relations = list(relations)
if existing_relations:
all_relations.extend(existing_relations)
for rel in all_relations:
if rel.from_ not in graph:
graph[rel.from_] = set()
graph[rel.from_].add(rel.to)
# Check for cycles using DFS
def has_cycle(node: str, visited: Set[str], path: Set[str]) -> bool:
visited.add(node)
path.add(node)
for neighbor in graph.get(node, set()):
if neighbor not in visited:
if has_cycle(neighbor, visited, path):
return True
elif neighbor in path:
return True
path.remove(node)
return False
visited: Set[str] = set()
path: Set[str] = set()
for node in graph:
if node not in visited:
if has_cycle(node, visited, path):
raise RelationValidationError(
"Circular dependency detected in relations"
)
@classmethod
def validate_graph(cls, graph: KnowledgeGraph) -> None:
"""Validate entire knowledge graph.
Args:
graph: Knowledge graph to validate
Raises:
ValidationError: If any validation fails
"""
# Validate all entities
entity_names = set()
for entity in graph.entities:
cls.validate_entity(entity)
if entity.name in entity_names:
raise EntityValidationError(f"Duplicate entity name: {entity.name}")
entity_names.add(entity.name)
# Validate all relations
for relation in graph.relations:
cls.validate_relation(relation)
if relation.from_ not in entity_names:
raise RelationValidationError(
f"Source entity '{relation.from_}' not found in graph"
)
if relation.to not in entity_names:
raise RelationValidationError(
f"Target entity '{relation.to}' not found in graph"
)
# Check for cycles
cls.validate_no_cycles(graph.relations)
@classmethod
def validate_batch_entities(
cls, entities: List[Entity], existing_names: Set[str]
) -> None:
"""Validate a batch of entities efficiently.
Args:
entities: List of entities to validate
existing_names: Set of existing entity names
Raises:
EntityValidationError: If validation fails
"""
if not entities:
raise EntityValidationError("Entity list cannot be empty")
# Check for duplicates within the batch
new_names = set()
for entity in entities:
if entity.name in new_names:
raise EntityValidationError(
f"Duplicate entity name in batch: {entity.name}"
)
new_names.add(entity.name)
# Check for conflicts with existing entities
conflicts = new_names.intersection(existing_names)
if conflicts:
raise EntityValidationError(
f"Entities already exist: {', '.join(conflicts)}"
)
# Validate all entities in one pass
for entity in entities:
cls.validate_entity(entity)
@classmethod
def validate_batch_relations(
cls,
relations: List[Relation],
existing_relations: List[Relation],
entity_names: Set[str],
) -> None:
"""Validate a batch of relations efficiently.
Args:
relations: List of relations to validate
existing_relations: List of existing relations
entity_names: Set of valid entity names
Raises:
RelationValidationError: If validation fails
"""
if not relations:
raise RelationValidationError("Relations list cannot be empty")
# Track relation keys to prevent duplicates
seen_relations: Set[tuple[str, str, str]] = set()
# Validate all relations in one pass
missing_entities = set()
for relation in relations:
# Basic validation
cls.validate_relation(relation)
# Check for duplicate relations
key = (relation.from_, relation.to, relation.relationType)
if key in seen_relations:
raise RelationValidationError(
f"Duplicate relation: {relation.from_} -> {relation.to}"
)
seen_relations.add(key)
# Collect missing entities
if relation.from_ not in entity_names:
missing_entities.add(relation.from_)
if relation.to not in entity_names:
missing_entities.add(relation.to)
# Report all missing entities at once
if missing_entities:
raise RelationValidationError(
f"Entities not found: {', '.join(missing_entities)}"
)
# Check for cycles including existing relations
cls.validate_no_cycles(relations, existing_relations)
@classmethod
def validate_batch_observations(
cls,
observations_map: dict[str, List[str]],
existing_entities: dict[str, Entity],
) -> None:
"""Validate a batch of observations efficiently.
Args:
observations_map: Dictionary mapping entity names to lists of observations
existing_entities: Dictionary of existing entities
Raises:
EntityValidationError: If validation fails
"""
if not observations_map:
raise EntityValidationError("Observations map cannot be empty")
# Check for missing entities first
missing_entities = [
name for name in observations_map if name not in existing_entities
]
if missing_entities:
raise EntityValidationError(
f"Entities not found: {', '.join(missing_entities)}"
)
# Validate all observations in one pass
for entity_name, observations in observations_map.items():
if not observations:
continue
# Validate observation format
cls.validate_observations(observations)
# Check for duplicates against existing observations
entity = existing_entities[entity_name]
existing_observations = set(entity.observations)
duplicates = [obs for obs in observations if obs in existing_observations]
if duplicates:
raise EntityValidationError(
f"Duplicate observations for {entity_name}: {', '.join(duplicates)}"
)