babylon.engine.adapters

Graph adapters for the Babylon simulation engine.

Slice 1.7: The Graph Bridge

This package contains adapters that implement GraphProtocol for different graph backends:

NetworkXAdapter: Reference implementation using NetworkX (Epoch 1-2) ColumnarAdapter: DuckDB + DuckPGQ implementation (Epoch 3, planned)

All Systems interact with the graph through GraphProtocol, never directly with the backend. This enables backend swapping without changing System code.

class babylon.engine.adapters.NetworkXAdapter[source]

Bases: AggregationMixin, QueryMixin

Reference implementation of GraphProtocol using NetworkX.

Wraps nx.DiGraph and provides all 18 GraphProtocol methods. Node types are stored as ‘_node_type’ attribute, edge types as ‘_edge_type’.

Inherits from mixins:

AggregationMixin: aggregate()
QueryMixin: query_nodes(), query_edges(), count_nodes(), count_edges()

Example

>>> adapter = NetworkXAdapter()
>>> adapter.add_node("C001", "social_class", wealth=100.0)
>>> node = adapter.get_node("C001")
>>> node.wealth
100.0

__init__()[source]

Initialize with empty DiGraph.

Return type:: None

add_edge(source, target, edge_type, weight=1.0, **attributes)[source]

Add directed edge with type and weight.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge category.
weight (float) – Generic weight (default 1.0).
**attributes (Any) – Type-specific attributes.

Return type:

None

add_node(node_id, node_type, **attributes)[source]

Add a node with type marker and attributes.

Parameters:

node_id (str) – Unique identifier for the node.
node_type (str) – Discriminator for polymorphism.
**attributes (Any) – Type-specific attributes to store.

Return type:

None

execute_traversal(query)[source]

Execute generic traversal query.

Parameters:: query (TraversalQuery) – TraversalQuery specifying the traversal.
Return type:: TraversalResult
Returns:: TraversalResult with nodes, edges, paths, or aggregates.
Raises:: ValueError – If query_type is not supported.

get_edge(source, target, edge_type)[source]

Retrieve specific edge by source, target, and type.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Return type:

GraphEdge | None

Returns:

GraphEdge model if found and type matches, None otherwise.

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

Retrieve a graph-level attribute.

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.

get_neighborhood(node_id, radius=1, edge_types=None, direction='out')[source]

Get all nodes within radius hops of the source node.

Parameters:

node_id (str) – Center node for neighborhood.
radius (int) – Maximum hop distance (1 = immediate neighbors).
edge_types (set[str] | None) – Filter to specific edge types (None = all).
direction (Literal['out', 'in', 'both']) – Which edges to follow: outgoing, incoming, or both.

Return type:

SubgraphView

Returns:

SubgraphView containing nodes in neighborhood.

Raises:

KeyError – If node does not exist.

get_node(node_id)[source]

Retrieve node by ID.

Parameters:: node_id (str) – The node identifier to look up.
Return type:: GraphNode | None
Returns:: GraphNode model if found, None otherwise.

remove_edge(source, target, edge_type)[source]

Remove specific edge.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Raises:

KeyError – If edge does not exist or type doesn’t match.

Return type:

None

remove_node(node_id)[source]

Remove node and all incident edges.

Parameters:: node_id (str) – The node identifier to remove.
Raises:: KeyError – If node does not exist.
Return type:: None

set_graph_attr(key, value)[source]

Set a graph-level attribute.

Parameters:

key (str) – Attribute name to set.
value (Any) – Value to store.

Return type:

None

shortest_path(source, target, edge_types=None, weight_attr=None)[source]

Find shortest path between two nodes.

Parameters:

source (str) – Start node ID.
target (str) – End node ID.
edge_types (set[str] | None) – Filter to specific edge types.
weight_attr (str | None) – Attribute to use as weight (None = hop count).

Return type:

list[str] | None

Returns:

List of node IDs in path, or None if no path exists.

property underlying_graph

Access the underlying NetworkX DiGraph.

Used by WorldState.from_graph() to unwrap the adapter back to a raw graph, and by topology_monitor for NetworkX-specific algorithms (copy-and-purge resilience testing) that have no protocol equivalent.

Returns:: The raw nx.DiGraph backing this adapter.

update_edge(source, target, edge_type, **attributes)[source]

Partial update of edge attributes.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.
**attributes (Any) – Attributes to update.

Raises:

KeyError – If edge does not exist or type doesn’t match.

Return type:

None

update_node(node_id, **attributes)[source]

Partial update of node attributes (merge, not replace).

Parameters:

node_id (str) – The node identifier to update.
**attributes (Any) – Attributes to update.

Raises:

KeyError – If node does not exist.

Return type:

None

classmethod wrap(graph)[source]

Wrap an existing nx.DiGraph in a protocol adapter.

Normalizes edge keys: raw graphs from WorldState.to_graph() store edge_type (from Pydantic model_dump()), but the adapter uses _edge_type. This method adds _edge_type alongside existing edge_type so protocol methods work correctly. Similarly normalizes node_type → _node_type for node type discrimination.

The adapter holds a reference to the graph (not a copy), so mutations via protocol methods flow through to the original graph.

Parameters:: graph – An existing NetworkX DiGraph to wrap.
Returns:: A NetworkXAdapter backed by the given graph.
Return type:: NetworkXAdapter

class babylon.engine.adapters.SubgraphView(subgraph)[source]

Bases: object

Read-only view of a subgraph.

Provides iteration over nodes in a subgraph returned by neighborhood queries. Wraps the underlying NetworkX subgraph to provide GraphNode iteration.

Parameters:: subgraph (nx.Graph[str])

_subgraph: The underlying NetworkX subgraph.

Example

>>> from babylon.engine.adapters.inmemory_adapter import NetworkXAdapter
>>> adapter = NetworkXAdapter()
>>> adapter.add_node("C001", "social_class", wealth=100.0)
>>> adapter.add_node("C002", "social_class", wealth=50.0)
>>> adapter.add_edge("C001", "C002", "SOLIDARITY")
>>> view = adapter.get_neighborhood("C001")
>>> list(view.nodes())
[GraphNode(...), GraphNode(...)]

__init__(subgraph)[source]

Initialize with a NetworkX subgraph.

Parameters:: subgraph – The NetworkX subgraph to wrap. Can be DiGraph or subgraph view.

edges()[source]

Iterate over edges in the subgraph.

Yields:: GraphEdge models for each edge in the subgraph.
Return type:: Iterator[GraphEdge]

nodes()[source]

Iterate over nodes in the subgraph.

Yields:: GraphNode models for each node in the subgraph.
Return type:: Iterator[GraphNode]

Modules

`aggregation_mixin`	AggregationMixin for graph aggregation operations.
`inmemory_adapter`	NetworkX-based in-memory graph adapter.
`query_mixin`	QueryMixin for graph query operations.
`subgraph_filter`	SubgraphFilterBuilder for constructing filtered subgraphs.
`subgraph_view`	SubgraphView for read-only graph iteration.