LayeredSpec
dataclass
¶
LayeredSpec(
input_nodes: tuple[str, ...],
output_nodes: tuple[str, ...],
hidden_nodes: tuple[str, ...],
layer_nodes: tuple[tuple[str, ...], ...],
layer_dims: tuple[int, ...],
layer_widths: tuple[tuple[int, ...], ...],
hops: tuple[Hop, ...],
skips: tuple[Skip, ...],
)
Frozen blueprint from parse_layered.
Depth-ranked wiring for one knowledge-primed network: every
node sits at a layer, and one Hop per layer after the
first holds every edge entering it, skips included, as packed
indices. Build one PackedLinear.from_hop(hop) per hop, and
use the name tuples to label tensors. It is structure only —
no nn.Module, no parameters — and there is no stored mask.
AdjacencySpec is the shared-state packed alternative.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
tuple[str, ...]
|
In-degree 0 names, alphabetical. This is the feature-axis
order |
required |
|
tuple[str, ...]
|
Out-degree 0 names, alphabetical. A terminal node below
maximum depth belongs here too, so this is not the same
tuple as |
required |
|
tuple[str, ...]
|
Names that are neither input nor output, alphabetical. |
required |
|
tuple[tuple[str, ...], ...]
|
|
required |
|
tuple[int, ...]
|
Unit count of each layer: |
required |
|
tuple[tuple[int, ...], ...]
|
|
required |
|
tuple[Hop, ...]
|
One hop per layer after the first:
|
required |
|
tuple[Skip, ...]
|
Original edges with depth gap greater than 1, as metadata.
Each one is already a block of packed unit pairs in
|
required |
See Also
parse_layered : Builds this spec from a source / target
edgelist.
AdjacencySpec : Packed sibling layout, for cycles, self-loops,
or one shared state vector instead of depths.
align_inputs : Column index that puts named features in
input_nodes order; length is layer_dims[0].
gather_hop_inputs : Assembles one hop's input from the layer
tensors produced so far.
scatter_hop_outputs : Splits a transposed hop's output back
onto those source layers.
PackedLinear.from_hop : Builds one layer from hops[i].
PackedLinear.transpose : Tied decode of that hop.
MaskedLinear : Consumes hops[i].to_mask() as one layer.
Notes
Fields cannot be reassigned and sequences are tuples, so the
structure itself is fixed. There is no stored mask tensor and
no densifying mask property. to_mask() on each hop
allocates a fresh dense rectangle; mutating that tensor does
not change this spec. MaskedLinear(hop.to_mask()) clones
the rectangle into a non-persistent buffer, so a layer built
earlier keeps its own connectivity.
Because a hop carries every parent of its target, the
per-row degree PackedLinear and MaskedLinear
initialize from is the real fan-in of that unit, skips
included.
to_edgelist(), to_dict() with from_dict(), and
fingerprint are the supported interchange. Widths and
ranks live on to_dict(), not on the edgelist: reparse of
to_edgelist() without widths= / ranks= is
width 1 and longest-path. to_dict() omits "ranks"
when compacted layers equal longest-path on the same edges.
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 of one hop, 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 of every
hop. node_units and
hop_units map a named node to its contiguous unit slice
on a layer tensor or a hop source axis. unit_index gives
every unit of several named nodes of one layer, in order, as
one index. node_table
counts each node's edges and the nodes that reach it or
that it reaches.
Examples:
Inspect layers, a hop, and a skip after parsing:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H", "A"],
... "target": ["H", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> spec.layer_nodes
(('A',), ('H',), ('C',))
>>> spec.hops[0].source_layers, spec.hops[0].to_mask().tolist()
((0,), [[1.0]])
>>> spec.hops[1].source_layers, spec.hops[1].to_mask().tolist()
((0, 1), [[1.0, 1.0]])
>>> spec.skips[0].source, spec.skips[0].target
('A', 'C')
>>> spec.skips[0].source_layer, spec.skips[0].target_layer
(0, 2)
>>> spec.edge_location("A", "C")
(1, (0,))
Attributes¶
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": ["A", "H"],
... "target": ["H", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> len(spec.fingerprint)
64
>>> (
... spec.fingerprint
... == kpnn2.parse_layered(spec.to_edgelist()).fingerprint
... )
True
Methods:¶
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 hop edges in canonical order: sorted
lexicographically by (source, target), one row per
original edge, names as strings. Extra columns from the
DataFrame that was parsed are not reproduced.
parse_layered on this table reconstructs the same
node lists and named edges when every node has width 1
and ranking is longest-path. Packed hop indices and
layer_dims match in that case. Otherwise pass
widths= and/or ranks=, or use from_dict(),
which reads those keys from the tagged dict. Skip tuple
order follows these sorted rows rather than the original
parse input order; the skip set matches.
Returns:
| Type | Description |
|---|---|
DataFrame
|
One row per original edge. |
Examples:
Unsorted input comes back sorted:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["H", "A", "A"],
... "target": ["C", "C", "H"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> table = spec.to_edgelist()
>>> list(table.columns)
['source', 'target']
>>> table["source"].tolist()
['A', 'A', 'H']
>>> table["target"].tolist()
['C', 'H', 'C']
edge_location ¶
edge_location(
source: object,
target: object,
) -> tuple[int, tuple[int, ...]]
Return packed weight slots of one named edge.
source and target are matched after str(...),
same as parse. This is identity into the hop that
carries the edge, not a constraint.
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:
| Name | Type | Description |
|---|---|---|
hop_index |
int
|
Index |
packed_indices |
tuple of int
|
Indices into |
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
Width greater than 1 does not change the named edge. It
returns several packed indices, one 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:
A chain plus a skip. The skip is on the later hop:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H", "A"],
... "target": ["H", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> spec.edge_location("A", "H")
(0, (0,))
>>> spec.edge_location("A", "C")
(1, (0,))
edge_index ¶
edge_index(
sources: object,
targets: object,
*,
hop: Hop,
) -> tuple[int, ...]
Return every packed slot of the named edges of one hop, in order.
Edge i is sources[i] -> targets[i], and every edge
must enter hop's target layer. Each edge contributes
all of its slots, in the order edge_location gives
them, in edge order. The slots index hop.source_index
/ hop.target_index and the weight of
PackedLinear.from_hop(hop), 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 and the hop of
one edge. AdjacencySpec.edge_index is the same rule
on the state vector, which has no hops.
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 |
|
Hop
|
A hop from |
required |
Returns:
| Type | Description |
|---|---|
tuple of int
|
Packed slots of the named edges in |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
This is identity into one hop's 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.
Edges of several hops need one call per hop, because each
hop has its own slots. edge_location gives the hop of
an edge: the one into its target's layer.
With a wide node, one value per edge has to be repeated
across that edge's slots, len(spec.edge_location(source,
target)[1]) 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:
The hop into C carries the adjacent edge H -> C
and the skip A -> C:
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H", "A"],
... "target": ["H", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> spec.edge_index(["H", "A"], ["C", "C"], hop=spec.hops[1])
(1, 0)
A sign kept on your own table lands on its slots:
>>> signs = pd.DataFrame(
... {
... "source": ["H", "A"],
... "target": ["C", "C"],
... "sign": [-1.0, 1.0],
... }
... )
>>> hop = spec.hops[1]
>>> index = torch.as_tensor(
... spec.edge_index(
... signs["source"],
... signs["target"],
... hop=hop,
... ),
... dtype=torch.int64,
... )
>>> sign = torch.zeros(len(hop.source_index))
>>> sign[index] = torch.tensor(signs["sign"].tolist())
>>> sign.tolist()
[1.0, -1.0]
An edge out of a node of width 3 gives its whole block:
>>> wide = kpnn2.parse_layered(
... edgelist,
... widths={"H": 3},
... )
>>> wide.edge_index(["H", "A"], ["C", "C"], hop=wide.hops[1])
(1, 2, 3, 0)
slot_table ¶
slot_table() -> pd.DataFrame
Return the named edge of every packed slot, hop by hop.
One row per packed slot of every hop. The rows of hop
i are in the order of hops[i].source_index and of
the weight of PackedLinear.from_hop(hops[i]), and
the hops follow hops. Per-slot tensors such as trained
weights, one per hop joined with torch.cat in hop
order, therefore get their edge names by assignment as a
numpy array. At width 1 a slot is a named edge,
and the rows of one hop are that hop's 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 of one hop to their slots.
Returns:
| Type | Description |
|---|---|
DataFrame
|
Index levels |
See Also
edge_index : Slots of named edges of one hop, in order. edge_location : Hop and slots of one named edge. 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.
table.loc[i] is hop i, indexed by slot. The units
of a row are hops[hop].source_index[slot] on the hop's
source axis and hops[hop].target_index[slot] on its
target layer. To get one value per named edge on a wide
spec, group the rows by ["source", "target"].
Examples:
A chain plus a skip. The hop into C holds both of its
edges:
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H", "A"],
... "target": ["H", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> table = spec.slot_table()
>>> table
source target
hop slot
0 0 A H
1 0 A C
1 H C
Per-hop tensors, here standing in for each layer's
effective_weight().detach(), name their values once
joined in hop order:
>>> weights = [torch.tensor([0.5]), torch.tensor([-1.0, 2.0])]
>>> table["weight"] = torch.cat(weights).numpy()
>>> table.loc[1]
source target weight
slot
0 A C -1.0
1 H C 2.0
node_units ¶
node_units(name: object) -> tuple[int, slice]
Return the layer and unit slice of one named node.
name is matched after str(...), same as parse.
The slice indexes the last axis of that layer's tensor
(saved[layer]). It is always a slice, including at
width 1.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
str
|
Node name. Non-strings are converted with
|
required |
Returns:
| Name | Type | Description |
|---|---|---|
layer |
int
|
Depth of the node: index into |
units |
slice
|
Contiguous columns on that layer's last axis. Length is the node's width. |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
This is identity into the unit axis, not a dropout
module and not a head helper. Index as
saved[layer][..., units].
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H"],
... "target": ["H", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(
... edgelist,
... widths={"H": 3},
... )
>>> layer, units = spec.node_units("H")
>>> layer, units.start, units.stop
(1, 0, 3)
hop_units ¶
hop_units(hop: Hop, name: object) -> slice
Return the unit slice of one named node on a hop axis.
The axis is the concatenated source of hop, the
same tensor gather_hop_inputs returns.
name is matched after str(...), same as parse.
hop must compare equal to one entry of hops.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
Hop
|
A hop from |
required |
|
str
|
Node name. Non-strings are converted with
|
required |
Returns:
| Type | Description |
|---|---|
slice
|
Contiguous columns on |
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
column_offsets locates a whole source layer on this
axis. This method locates one named node, widths
included. A target-layer name is not on the source
axis.
Examples:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["A", "H", "A"],
... "target": ["H", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> hop = spec.hops[1]
>>> spec.hop_units(hop, "A").start
0
>>> spec.hop_units(hop, "H").start
1
unit_index ¶
unit_index(names: object, *, layer: int) -> tuple[int, ...]
Return every unit of the named nodes of one layer, in order.
The units index the last axis of that layer's tensor
(saved[layer]). Each named node contributes all of its
units, ascending, in names order. Use it for nodes you
name yourself within a layer, such as readouts, a loss
against measured values, 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.
AdjacencySpec.unit_index is the same rule on the
state vector, which has no layers.
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 |
|
int
|
0-based index into |
required |
Returns:
| Type | Description |
|---|---|
tuple of int
|
Units of the named nodes on that layer, in order.
|
Raises:
| Type | Description |
|---|---|
Kpnn2Error
|
If |
Notes
This is identity into a layer's unit axis: it takes no
tensor and no data. Keep it as a buffer with
torch.as_tensor(index, dtype=torch.int64). The
dtype is needed because torch.as_tensor(()) is
float32 and fails as an index.
Names in several layers need one call per layer, because
no single tensor holds their units. node_units gives
the layer of each node.
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 the layer:
>>> import pandas as pd
>>> import torch
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["a", "b", "H", "K"],
... "target": ["H", "K", "C", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> spec.layer_nodes[1]
('H', 'K')
>>> spec.unit_index(["K", "H"], layer=1)
(1, 0)
A node of width 3 gives its whole block, and every node of a layer gives every unit of that layer:
>>> wide = kpnn2.parse_layered(
... edgelist,
... widths={"H": 3},
... )
>>> wide.unit_index(["K", "H"], layer=1)
(3, 0, 1, 2)
>>> wide.unit_index(wide.layer_nodes[1], layer=1) == tuple(
... range(wide.layer_dims[1])
... )
True
As a buffer, it reads those units from that layer's tensor:
>>> index = torch.as_tensor(
... wide.unit_index(["K"], layer=1),
... dtype=torch.int64,
... )
>>> saved = {1: torch.arange(4.0).unsqueeze(0)}
>>> saved[1][..., index].tolist()
[[3.0]]
node_table ¶
node_table() -> pd.DataFrame
Return one row of graph counts per node.
Each row says where a node sits in the graph: its layer,
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. 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 its layer.
Notes
in_degree == 0 selects exactly input_nodes, and
out_degree == 0 exactly output_nodes. Rows follow
unit order, so
numpy.repeat(table[column], table["width"]) gives one
value per unit, layer after layer; filter
table[table["layer"] == i] first for the units of
layer i. 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:
Three inputs feed two hidden nodes, which feed y;
x3 also feeds y directly:
>>> import pandas as pd
>>> import kpnn2
>>> edgelist = pd.DataFrame(
... {
... "source": ["x1", "x2", "x2", "x3", "h1", "h2", "x3"],
... "target": ["h1", "h1", "h2", "h2", "y", "y", "y"],
... }
... )
>>> table = kpnn2.parse_layered(edgelist).node_table()
>>> table[["layer", "width", "in_degree", "out_degree"]]
layer width in_degree out_degree
node
x1 0 1 0 1
x2 0 1 0 2
x3 0 1 0 2
h1 1 1 2 1
h2 1 1 2 1
y 2 1 3 0
>>> table[["n_ancestors", "n_descendants", "n_input_ancestors"]]
n_ancestors n_descendants n_input_ancestors
node
x1 0 2 0
x2 0 3 0
x3 0 2 0
h1 2 1 2
h2 2 1 2
y 5 0 3
A wider node keeps its counts, and repeating a column by
width gives one value per unit:
>>> import numpy as np
>>> wide = kpnn2.parse_layered(
... edgelist,
... widths={"h1": 2},
... ).node_table()
>>> wide.loc["h1", ["width", "in_degree"]].tolist()
[2, 2]
>>> np.repeat(wide["in_degree"], wide["width"]).tolist()
[0, 0, 0, 2, 2, 2, 3]
to_dict ¶
to_dict() -> dict
Return this spec as a JSON-safe tagged dict.
Keys are kpnn2_spec (integer 1), layout
("layered"), and edges (list of
[source, target] lists in the same order as
to_edgelist() rows). When any node has width other
than 1, a "widths" object of those names is included.
All-1 graphs omit "widths". When compacted layers
differ from longest-path on the same edges, a "ranks"
object maps every node to its compacted 0-based layer
index. A ranks= parse that matches longest-path
omits "ranks". 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": ["A", "H"],
... "target": ["H", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> payload = spec.to_dict()
>>> payload["kpnn2_spec"]
1
>>> payload["layout"]
'layered'
>>> payload["edges"]
[['A', 'H'], ['H', 'C']]
from_dict
classmethod
¶
from_dict(payload: dict) -> LayeredSpec
Rebuild a LayeredSpec from to_dict() output.
Calls parse_layered on a DataFrame built from
payload["edges"], passing payload["widths"] and
payload["ranks"] when present. Hops and packed
indices are not assembled by hand. Extra unknown keys
are ignored. An absent or empty "widths" object is
width 1. Absent "ranks" is longest-path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
|
dict
|
A dict with |
required |
Returns:
| Type | Description |
|---|---|
LayeredSpec
|
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": ["A", "H"],
... "target": ["H", "C"],
... }
... )
>>> spec = kpnn2.parse_layered(edgelist)
>>> roundtrip = kpnn2.LayeredSpec.from_dict(spec.to_dict())
>>> roundtrip.layer_nodes == spec.layer_nodes
True