babylon.models.graph

Graph abstraction layer data models.

Slice 1.7: The Graph Bridge

These Pydantic models define the type-safe boundary between Systems and the graph backend. All models are frozen (immutable) for thread safety.

Models:

EdgeFilter: Filter specification for edge traversal NodeFilter: Filter specification for node inclusion GraphNode: Frozen Pydantic model for node representation GraphEdge: Frozen Pydantic model for edge representation TraversalQuery: Generic traversal query specification TraversalResult: Result of traversal query execution

Design Principles:

Backend-agnostic: Works with NetworkX now, DuckDB later
Set-oriented: Think in tables, not just objects (DuckDB-ready)
Minimal but complete: Just enough to cover all System needs
Lazy evaluation: Return iterators/generators where possible

Classes

`EdgeFilter`(**data)	Filter specification for edge traversal.
`GraphEdge`(**data)	Type-safe graph edge representation.
`GraphNode`(**data)	Type-safe graph node representation.
`NodeFilter`(**data)	Filter specification for node inclusion.
`TraversalQuery`(**data)	Generic traversal query specification.
`TraversalResult`(**data)	Result of traversal query execution.

class babylon.models.graph.EdgeFilter(**data)[source]

Bases: BaseModel

Filter specification for edge traversal.

Used to select which edges to traverse during graph operations. All fields are optional - None means no filter (match all).

Parameters:

edge_types (set[str] | None)
min_weight (float | None)
max_weight (float | None)

edge_types: Set of edge types to include (None = all types).

min_weight: Minimum weight threshold (inclusive).

max_weight: Maximum weight threshold (inclusive).

Example

>>> f = EdgeFilter(edge_types={"SOLIDARITY"}, min_weight=0.5)
>>> f.edge_types
{'SOLIDARITY'}

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

edge_types: set[str] | None

min_weight: float | None

max_weight: float | None

class babylon.models.graph.NodeFilter(**data)[source]

Bases: BaseModel

Filter specification for node inclusion.

Used to select which nodes to include during graph operations. All fields are optional - None means no filter (match all).

Parameters:

node_types (set[str] | None)
attribute_predicates (dict[str, Any] | None)

node_types: Set of node types to include (None = all types).

attribute_predicates: Dict of attribute equality filters.

Example

>>> f = NodeFilter(node_types={"social_class"})
>>> f.node_types
{'social_class'}

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

node_types: set[str] | None

attribute_predicates: dict[str, Any] | None

class babylon.models.graph.GraphNode(**data)[source]

Bases: BaseModel

Type-safe graph node representation.

Frozen Pydantic model for node data at the protocol boundary. Adapters convert to/from their internal representation.

Parameters:

id (Annotated[str, MinLen(min_length=1)])
node_type (Annotated[str, MinLen(min_length=1)])
attributes (dict[str, Any])

id: Unique node identifier.

node_type: Discriminator for polymorphism (e.g., ‘social_class’, ‘territory’).

attributes: Type-specific attributes stored as dict.

Example

>>> node = GraphNode(id="C001", node_type="social_class", attributes={"wealth": 100.0})
>>> node.wealth
100.0

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

id: str

node_type: str

attributes: dict[str, Any]

get_attr(key, default=None)[source]

Get attribute with default.

Parameters:

key (str) – Attribute name to retrieve.
default (Any) – Value to return if attribute not present.

Return type:

Any

Returns:

The attribute value or default.

property wealth: float

Convenience accessor for wealth attribute.

Returns:: Wealth value or 0.0 if not present.

class babylon.models.graph.GraphEdge(**data)[source]

Bases: BaseModel

Type-safe graph edge representation.

Frozen Pydantic model for edge data at the protocol boundary. Adapters convert to/from their internal representation.

Parameters:

source_id (Annotated[str, MinLen(min_length=1)])
target_id (Annotated[str, MinLen(min_length=1)])
edge_type (Annotated[str, MinLen(min_length=1)])
weight (float)
attributes (dict[str, Any])

source_id: Origin node ID.

target_id: Destination node ID.

edge_type: Edge category (e.g., ‘SOLIDARITY’, ‘EXPLOITATION’).

weight: Generic weight (default 1.0).

attributes: Type-specific attributes stored as dict.

Example

>>> edge = GraphEdge(source_id="C001", target_id="C002", edge_type="SOLIDARITY")
>>> edge.weight
1.0

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source_id: str

target_id: str

edge_type: str

weight: float

attributes: dict[str, Any]

property tension: float

Convenience accessor for tension attribute.

Returns:: Tension value or 0.0 if not present.

property value_flow: float

Convenience accessor for value_flow attribute.

Returns:: Value flow or 0.0 if not present.

class babylon.models.graph.TraversalQuery(**data)[source]

Bases: BaseModel

Generic traversal query specification.

Specifies what traversal to execute and what to collect. This is the generic hook for complex operations like percolation analysis, pathfinding, and component detection.

Parameters:

query_type (Literal['bfs', 'dfs', 'shortest_path', 'connected_components', 'percolation', 'reachability'])
start_nodes (list[str] | None)
target_nodes (list[str] | None)
edge_filter (EdgeFilter | None)
node_filter (NodeFilter | None)
max_depth (int | None)
collect (list[str])

query_type: Type of traversal algorithm to use.

start_nodes: Starting node IDs (None = all nodes).

target_nodes: Target node IDs (for shortest_path, reachability).

edge_filter: Filter which edges to traverse.

node_filter: Filter which nodes to include.

max_depth: Maximum traversal depth.

collect: What to include in result.

Example

>>> query = TraversalQuery(query_type="bfs", start_nodes=["C001"])
>>> query.query_type
'bfs'

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

query_type: Literal['bfs', 'dfs', 'shortest_path', 'connected_components', 'percolation', 'reachability']

start_nodes: list[str] | None

target_nodes: list[str] | None

edge_filter: EdgeFilter | None

node_filter: NodeFilter | None

max_depth: int | None

collect: list[str]

class babylon.models.graph.TraversalResult(**data)[source]

Bases: BaseModel

Result of traversal query execution.

Contains all data collected during traversal based on query.collect.

Parameters:

nodes (list[str])
edges (list[tuple[str, str, str]])
paths (list[list[str]])
components (list[list[str]])
component_sizes (list[int])
metadata (dict[str, Any])

nodes: Node IDs in traversal result.

edges: Edge tuples (source, target, type) in result.

paths: Paths found (for shortest_path queries).

components: Connected components (for component queries).

component_sizes: Sizes of connected components (sorted descending).

metadata: Query-specific metadata.

Example

>>> result = TraversalResult(component_sizes=[10, 5, 3])
>>> result.largest_component_size
10

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodes: list[str]

edges: list[tuple[str, str, str]]

paths: list[list[str]]

components: list[list[str]]

component_sizes: list[int]

metadata: dict[str, Any]

property largest_component_size: int

Size of the largest (giant) component.

Returns:: Size of first component or 0 if no components.

property percolation_ratio: float

Fraction of nodes in the largest component.

This is the key metric for phase transition detection. A ratio near 1.0 indicates full percolation (connectivity).

Returns:: Ratio of largest_component_size to total nodes, or 0.0 if empty.