AdjacencySpec
dataclass
¶
AdjacencySpec(
nodes: tuple[str, ...],
node_widths: tuple[int, ...],
input_nodes: tuple[str, ...],
output_nodes: tuple[str, ...],
hidden_nodes: tuple[str, ...],
source_index: tuple[int, ...],
target_index: tuple[int, ...],
input_index: tuple[int, ...],
output_index: tuple[int, ...],
)
Frozen blueprint from parse_adjacency.
Packed wiring for one knowledge-primed network: every node
owns a contiguous block of units (one unit unless
parse_adjacency(widths=) says otherwise) of one
alphabetical state vector, and every edge is packed unit
pairs, so cycles and self-loops are ordinary. Structure
only — no nn.Module, no parameters, no stored square — and
the update is yours. Input nodes have no incoming edges, so
writing the inputs into the state each step is required.
LayeredSpec is the depth-ranked alternative.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
tuple[str, ...]
|
Every node name, alphabetical. This is the node order of
the state vector and of the rows and columns of
|
required |
|
tuple[int, ...]
|
Units per entry of |
required |
|
tuple[str, ...]
|
In-degree 0 names, alphabetical. This is the column order
|
required |
|
tuple[str, ...]
|
Out-degree 0 names, alphabetical. |
required |
|
tuple[str, ...]
|
Names that are neither input nor output, alphabetical. Empty when every node is an input or an output. |
required |
|
tuple[int, ...]
|
Source unit of each live unit pair. A named edge
|
required |
|
tuple[int, ...]
|
Target unit of each live unit pair. A dense square would
have |
required |
|
tuple[int, ...]
|
The unit of each |
required |
|
tuple[int, ...]
|
Every unit of each |
required |
See Also
parse_adjacency : Builds this spec from a source /
target edgelist; the only supported constructor.
LayeredSpec : Depth-ranked sibling layout, one incoming packed
hop per layer, rejecting cycles and self-loops.
PackedLinear.from_adjacency : One weight per edge of this
spec, with no (n, n).
PackedMultiheadAttention.from_adjacency : Scores only those
same packed pairs, for an attention update instead of a
linear one.
align_inputs : Column index that puts named features onto
input_nodes.
Notes
Fields cannot be reassigned and sequences are tuples, so the
structure itself is fixed. There is no mask field and no
densifying mask property. to_mask() allocates a
fresh dense square on every call; mutating that tensor does
not change this spec. MaskedLinear(spec.to_mask())
clones the square into a non-persistent buffer, so a layer
built earlier keeps its own connectivity.
align_inputs returns len(input_index) positions (one
per input node), which is not the state width. Scatter the
gathered columns into the state_dim-wide state vector
with input_index. Input rows of
to_mask() are all zeros, so under the degree-aware init
of MaskedLinear and PackedLinear those units stay
zero: writing the inputs in is required, not cosmetic.
Depth does not exist in this layout, so there is no
layer_nodes, layer_dims, hops, or skips, and
gather_hop_inputs does not accept this spec. An edge
that would span layers is already an ordinary index pair.
This is not a one-layer LayeredSpec; the layout is your
choice, and a DAG is valid input to either parser.
to_dict() with from_dict() and fingerprint are
the supported interchange; they round-trip through
parse_adjacency, cycle edges, self-loops, and widths
included. to_edgelist() round-trips at width 1. Pickle
and torch.save of the
dataclass are not. edge_location finds packed slots of
a named edge, and edge_index those of several edges, in
order; neither is a constraint. A hard freeze of those
slots is torch.where inside constraint=.
slot_table goes the other way and names the edge of
every slot.
node_table counts each node's edges and the nodes
that reach it or that it reaches.
Examples:
An input feeding a two-node feedback core plus one output:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> spec.nodes
('a', 'b', 'x', 'y')
>>> spec.input_nodes, spec.output_nodes
(('x',), ('y',))
>>> spec.hidden_nodes
('a', 'b')
>>> spec.input_index, spec.output_index
((2,), (3,))
>>> spec.source_index
(0, 0, 1, 2)
>>> spec.target_index
(1, 3, 0, 0)
>>> spec.edge_location("a", "b")
(0,)
>>> spec.state_dim
4
>>> tuple(spec.to_mask().shape)
(4, 4)
>>> spec.to_mask()[0].tolist()
[0.0, 1.0, 1.0, 0.0]
Attributes¶
state_dim
property
¶
state_dim: int
Number of units in the state vector, sum(node_widths).
This is the in_features / out_features of a
PackedLinear built on this spec and the side of
to_mask(). It equals len(nodes) when every node
has width 1.
Returns:
| Type | Description |
|---|---|
int
|
Number of units in the state vector. |
fingerprint
property
¶
fingerprint: str
SHA-256 hex digest of the canonical to_dict() JSON.
The payload is json.dumps(self.to_dict(),
sort_keys=True, separators=(",", ":"),
ensure_ascii=False) encoded as UTF-8. The result is
64 lowercase hex characters. It is not Python
hash().
Returns:
| Type | Description |
|---|---|
str
|
Hex digest of the tagged spec dict. |
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> len(spec.fingerprint)
64
>>> (
... spec.fingerprint
... == kpnn2.parse_adjacency(spec.to_edgelist()).fingerprint
... )
True
Methods:¶
node_units ¶
node_units(name: object) -> slice
Return the unit slice of one named node in the state vector.
name is matched after str(...), same as parse.
Always a slice, including at width 1. Index as
state[..., units]. LayeredSpec.node_units also
returns the depth; this layout has none.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
str
|
Node name. Non-strings are converted with
|
required |
Returns:
| Type | Description |
|---|---|
slice
|
Contiguous units of that node. Length is its width. |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(
... edgelist,
... widths={"b": 3},
... )
>>> spec.node_units("b")
slice(1, 4, None)
>>> spec.node_units("x")
slice(4, 5, None)
unit_index ¶
unit_index(names: object) -> tuple[int, ...]
Return every state unit of the named nodes, in order.
Each named node contributes all of its units, ascending,
in names order. This is the rule that builds
input_index and output_index, so
spec.unit_index(spec.input_nodes) == spec.input_index
(same for outputs). Use it for nodes you name yourself,
such as readouts that are not in output_nodes or data
written onto hidden nodes. At width 1, position i
belongs to names[i]; a node of width k contributes
k entries. node_units is the slice of one node.
LayeredSpec.unit_index is the same rule on one layer
and also takes layer=; this layout has none.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
sequence of str
|
Node names in the order to index: a list, a tuple, a
one-dimensional numpy array, a |
required |
Returns:
| Type | Description |
|---|---|
tuple of int
|
State units of the named nodes, in order. |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
This is identity into the state axis: it takes no tensor
and no data. Keep it as a buffer, as input_index is,
with torch.as_tensor(index, dtype=torch.int64). The
dtype is needed because torch.as_tensor(()) is
float32 and fails as an index.
Reading through a repeated unit is safe. Writing through
one is your concern: index_copy through repeated units
is ambiguous, while index_add sums.
Examples:
At width 1, each name gives its position in nodes:
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> spec.nodes
('a', 'b', 'x', 'y')
>>> spec.unit_index(["b", "a"])
(1, 0)
A node of width 3 gives its whole block:
>>> wide = kpnn2.parse_adjacency(
... edgelist,
... widths={"b": 3},
... )
>>> wide.unit_index(["b", "a"])
(1, 2, 3, 0)
>>> wide.unit_index(wide.input_nodes) == wide.input_index
True
As a buffer, it reads those units from a state tensor:
>>> index = torch.as_tensor(
... wide.unit_index(["b"]),
... dtype=torch.int64,
... )
>>> state = torch.arange(6.0).unsqueeze(0)
>>> state[..., index].tolist()
[[1.0, 2.0, 3.0]]
to_edgelist ¶
to_edgelist() -> pd.DataFrame
Return this spec's edges as a two-column table.
Columns are exactly source then target. Rows
follow the packed indices in canonical order: sorted
lexicographically by (source, target), one row per
original edge, names as strings, including cycle edges
and self-loops. Extra columns from the DataFrame that
was parsed are not reproduced.
parse_adjacency on this table reconstructs the same
node lists, packed indices, and input/output indices
when every node has width 1. Widths live on the spec
dict, not the edgelist: round-trip a wide spec with
from_dict or parse_adjacency(..., widths=).
Returns:
| Type | Description |
|---|---|
DataFrame
|
One row per original edge. |
Examples:
A cycle comes back as sorted pairs:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> table = spec.to_edgelist()
>>> list(table.columns)
['source', 'target']
>>> table["source"].tolist()
['a', 'a', 'b', 'x']
>>> table["target"].tolist()
['b', 'y', 'a', 'a']
to_mask ¶
to_mask() -> Tensor
Allocate a dense float32 square from the packed edges.
Shape is (state_dim, state_dim). The result starts at
zeros;
each live edge sets 1.0 at
[target_index[i], source_index[i]]. Every call
returns a fresh tensor. Mutating it does not change this
spec or the next to_mask() call.
Returns:
| Type | Description |
|---|---|
Tensor
|
New dense connectivity square. This allocates. |
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> mask = spec.to_mask()
>>> tuple(mask.shape)
(4, 4)
>>> mask[0, 1].item(), mask[1, 0].item()
(1.0, 1.0)
edge_location ¶
edge_location(
source: object,
target: object,
) -> tuple[int, ...]
Return packed weight slots of one named edge.
source and target are matched after str(...),
same as parse. Indices address source_index /
target_index and PackedLinear.weight built from
those arrays. This is identity into the packed arrays,
not a constraint. A named edge A -> B is the full
(k_B, k_A) block of unit pairs; at width 1 one pair.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
str
|
Source node name. Non-strings are converted with
|
required |
|
str
|
Target node name. Non-strings are converted with
|
required |
Returns:
| Type | Description |
|---|---|
tuple of int
|
Packed indices of the named edge, in stored order
(canonical lexicographic by
|
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If the pair is missing, a name is empty, or a name
is not a node. The message names the pair as
|
Notes
There is no hop index. Width greater than 1 does not
change the named edge; it returns one packed index per
unit pair of the block. Mixed signs and frozen values are
caller PyTorch on these slots. A hard freeze is
torch.where inside constraint=, not a
gradient hook.
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> spec.edge_location("a", "b")
(0,)
>>> spec.edge_location("x", "a")
(3,)
edge_index ¶
edge_index(
sources: object,
targets: object,
) -> tuple[int, ...]
Return every packed slot of the named edges, in order.
Edge i is sources[i] -> targets[i]. Each edge
contributes all of its slots, in the order
edge_location gives them, in edge order. The slots
index source_index / target_index and the
weight of a PackedLinear built from this spec, so
a value you keep per edge on your own table, such as a
sign, a prior flag, or an edge type, lands on its slots
without a loop, and a per-slot tensor such as a trained
weight reads back in your table's order. At width 1,
position i belongs to edge i; an edge between
nodes of widths k_source and k_target contributes
k_source * k_target entries. edge_location gives
the slots of one edge. LayeredSpec.edge_index is the
same rule on one hop and also takes hop=.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
sequence of str
|
Source node name of each edge, in the order to index:
a list, a tuple, a one-dimensional numpy array, a
|
required |
|
sequence of str
|
Target node name of each edge, same length and order
as |
required |
Returns:
| Type | Description |
|---|---|
tuple of int
|
Packed slots of the named edges, in order. |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
This is identity into the packed slots: it takes no
tensor and no values. Index with
torch.as_tensor(index, dtype=torch.int64). The
dtype is needed because torch.as_tensor(()) is
float32 and fails as an index.
With a wide node, one value per edge has to be repeated
across that edge's slots, len(spec.edge_location(source,
target)) of them. A repeated edge repeats its slots.
Reading through a repeated slot is safe. Writing through
one is your concern: index_copy through repeated slots
is ambiguous, while index_add sums.
Examples:
At width 1, each edge gives its row in to_edgelist():
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> spec.to_edgelist().values.tolist()
[['a', 'b'], ['a', 'y'], ['b', 'a'], ['x', 'a']]
>>> spec.edge_index(["x", "b"], ["a", "a"])
(3, 2)
A sign kept on your own table lands on its slots:
>>> signs = pd.DataFrame(
... {
... "source": ["b", "a"],
... "target": ["a", "b"],
... "sign": [-1.0, 1.0],
... }
... )
>>> index = torch.as_tensor(
... spec.edge_index(signs["source"], signs["target"]),
... dtype=torch.int64,
... )
>>> sign = torch.zeros(len(spec.source_index))
>>> sign[index] = torch.tensor(signs["sign"].tolist())
>>> sign.tolist()
[1.0, 0.0, -1.0, 0.0]
An edge into a node of width 3 gives its whole block:
>>> wide = kpnn2.parse_adjacency(
... edgelist,
... widths={"b": 3},
... )
>>> wide.edge_index(["x", "a"], ["a", "b"])
(7, 0, 1, 2)
slot_table ¶
slot_table() -> pd.DataFrame
Return the named edge of every packed slot, in slot order.
One row per packed slot, in the order of source_index
and of the weight of PackedLinear.from_adjacency
on this spec, so a per-slot tensor such as a trained
weight gets its edge names by assignment as a numpy
array: table["weight"] = weight.detach().numpy().
At width 1 a slot is a named edge, and the rows are the
rows of to_edgelist(). An edge between nodes of
widths k_source and k_target fills
k_source * k_target rows, one per unit pair, all with
the same names. edge_index goes the other way, from
named edges to their slots.
Returns:
| Type | Description |
|---|---|
DataFrame
|
Index |
See Also
edge_index : Slots of named edges, in order. to_edgelist : One row per named edge.
Notes
The table is computed on each call and is a new DataFrame
each time, so changing it does not change this spec. The
units of slot i are source_index[i] and
target_index[i]. To get one value per named edge on a
wide spec, group the rows by ["source", "target"].
Examples:
Four edges, one slot each:
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> table = spec.slot_table()
>>> table
source target
slot
0 a b
1 a y
2 b a
3 x a
A per-slot tensor, here standing in for
layer.effective_weight().detach(), names its values:
>>> weight = torch.tensor([0.5, -1.0, 2.0, 0.25])
>>> table["weight"] = weight.numpy()
>>> table.loc[table["target"] == "a"]
source target weight
slot
2 b a 2.00
3 x a 0.25
With b three units wide, a -> b and b -> a
fill three rows each:
>>> wide = kpnn2.parse_adjacency(
... edgelist,
... widths={"b": 3},
... )
>>> wide.slot_table().groupby(["source", "target"]).size()
source target
a b 3
y 1
b a 3
x a 1
dtype: int64
node_table ¶
node_table() -> pd.DataFrame
Return one row of graph counts per node.
Each row says where a node sits in the graph: how many
units it owns, how many edges enter and leave it, and how
many nodes reach it or are reached from it. A node with
many connections tends to get large sums and large scores
only because it is large, so these counts are what such a
node is scaled by: a widths value set from its number
of parents, a division of its input by its in-degree in
forward(), or a check of whether node scores follow
connectivity. The table holds the raw counts; the formula
stays yours.
Counts are over named edges in the direction of computation, so widths never change them. An input that feeds a node, directly or through other nodes, is one of its ancestors. A self-loop adds 1 to both degrees of its node. A node is never counted among its own ancestors or descendants, even on a cycle; the other nodes of its cycle are both. The table is computed on each call and is a new DataFrame each time, so changing it does not change this spec.
Returns:
| Type | Description |
|---|---|
DataFrame
|
Index
|
See Also
to_edgelist : The named edges the counts are taken over. node_units : One node's unit slice on the state vector.
Notes
in_degree == 0 selects exactly input_nodes, and
out_degree == 0 exactly output_nodes, self-loops
included. Rows follow the state vector, so
numpy.repeat(table[column], table["width"]) gives one
value per state unit, state_dim long. When counts
should set widths, read them from a first parse
without widths: they count named edges, so the second
parse has the same counts.
Examples:
A feedback edge b -> a and a self-loop on b:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a", "b"],
... "target": ["a", "b", "a", "y", "b"],
... }
... )
>>> table = kpnn2.parse_adjacency(edgelist).node_table()
>>> table[["width", "in_degree", "out_degree"]]
width in_degree out_degree
node
a 1 2 2
b 1 2 2
x 1 0 1
y 1 1 0
>>> table[["n_ancestors", "n_descendants", "n_input_ancestors"]]
n_ancestors n_descendants n_input_ancestors
node
a 2 2 1
b 2 2 1
x 0 3 0
y 3 0 1
A wider node keeps its counts, and repeating a column by
width gives one value per state unit:
>>> import numpy as np
>>> wide = kpnn2.parse_adjacency(
... edgelist,
... widths={"a": 2},
... ).node_table()
>>> np.repeat(wide["in_degree"], wide["width"]).tolist()
[2, 2, 2, 0, 1]
to_dict ¶
to_dict() -> dict
Return this spec as a JSON-safe tagged dict.
Keys are kpnn2_spec (integer 1), layout
("adjacency"), and edges (list of
[source, target] lists in the same order as
to_edgelist() rows, including cycle edges and
self-loops). When any node has width other than 1, a
"widths" object of those names is included; all-1
specs omit it. The returned dict is new on every call.
Returns:
| Type | Description |
|---|---|
dict
|
Tagged edge list plus layout. |
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> payload = spec.to_dict()
>>> payload["kpnn2_spec"]
1
>>> payload["layout"]
'adjacency'
>>> payload["edges"]
[['a', 'b'], ['a', 'y'], ['b', 'a'], ['x', 'a']]
from_dict
classmethod
¶
from_dict(payload: dict) -> AdjacencySpec
Rebuild an AdjacencySpec from to_dict() output.
Calls parse_adjacency on a DataFrame built from
payload["edges"], passing payload["widths"] when
present. Packed indices are not assembled by hand. Extra
unknown keys are ignored, including a stray "ranks".
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
dict
|
A dict with |
required |
Returns:
| Type | Description |
|---|---|
AdjacencySpec
|
The parsed spec. |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Warns:
| Type | Description |
|---|---|
UserWarning
|
If an edge name is a float or a |
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x", "a", "b", "a"],
... "target": ["a", "b", "a", "y"],
... }
... )
>>> spec = kpnn2.parse_adjacency(edgelist)
>>> roundtrip = kpnn2.AdjacencySpec.from_dict(spec.to_dict())
>>> roundtrip.nodes == spec.nodes
True