graflo.architecture.schema.context

Bounded schema context: the schema graph as a navigable, budgetable object.

Answers "what can I ask?" about a schema without touching a database. Every export here is layer 2 — pure logical model, no db, no data_source, no embeddings, no tokenizer.

Eager re-exports (this is a package boundary, not a lazy façade).

Budget

Bases: ConfigBaseModel

Caller-requested ceilings on a schema context payload.

Source code in graflo/architecture/schema/context/budget.py

class Budget(ConfigBaseModel):
    """Caller-requested ceilings on a schema context payload."""

    max_elements: int | None = PydanticField(
        default=60,
        description="Maximum vertices + edges in the slice. None disables the cap.",
        ge=1,
    )
    max_tokens: int | None = PydanticField(
        default=4000,
        description="Maximum estimated tokens for the serialized slice. None disables the cap.",
        ge=1,
    )
    max_properties_per_vertex: int | None = PydanticField(
        default=None,
        description=(
            "Maximum properties retained per vertex. Identity-bearing fields are "
            "never counted against this and never dropped."
        ),
        ge=1,
    )

BudgetAccounting

Bases: ConfigBaseModel

What the budget actually cost, measured rather than assumed.

Source code in graflo/architecture/schema/context/budget.py

class BudgetAccounting(ConfigBaseModel):
    """What the budget actually cost, measured rather than assumed."""

    requested: Budget = PydanticField(..., description="The budget as asked for.")
    elements_used: int = PydanticField(
        ..., description="Vertices + edges admitted into the slice."
    )
    estimated_tokens: int = PydanticField(
        ..., description="Token estimate for the assembled slice."
    )
    serialized_chars: int = PydanticField(
        ...,
        description=(
            "Exact character count of the compact serialization. Lets a caller "
            "re-estimate with a real tokenizer without trusting CHARS_PER_TOKEN."
        ),
    )
    exhausted_by: Literal["elements", "tokens", "none"] = PydanticField(
        ..., description="Which ceiling stopped admission, if any."
    )

ElidedEdge

Bases: ConfigBaseModel

An edge left out of the slice.

Source code in graflo/architecture/schema/context/elision.py

class ElidedEdge(ConfigBaseModel):
    """An edge left out of the slice."""

    edge_id: EdgeId = PydanticField(..., description="(source, target, relation).")
    reason: EdgeElisionReason = PydanticField(..., description="Why it was dropped.")
    description: str | None = PydanticField(
        default=None, description="Authored description, kept for the same reason."
    )

ElidedVertex

Bases: ConfigBaseModel

A vertex type left out of the slice.

Source code in graflo/architecture/schema/context/elision.py

class ElidedVertex(ConfigBaseModel):
    """A vertex type left out of the slice."""

    name: str = PydanticField(..., description="Vertex type name.")
    reason: VertexElisionReason = PydanticField(..., description="Why it was dropped.")
    degree: int = PydanticField(..., description="Incident edges in the full schema.")
    hop_distance: int | None = PydanticField(
        default=None, description="Hops from the nearest seed; None when unreachable."
    )
    description: str | None = PydanticField(
        default=None,
        description="Authored description, kept because it may be the reason to drill in.",
    )
    drill_in: str = PydanticField(
        ..., description="Call that would bring this type into a slice."
    )

ElisionReport

Bases: ConfigBaseModel

Everything the slice does not contain, plus the budget that caused it.

Source code in graflo/architecture/schema/context/elision.py

class ElisionReport(ConfigBaseModel):
    """Everything the slice does not contain, plus the budget that caused it."""

    elided_vertices: list[ElidedVertex] = PydanticField(
        default_factory=list, description="Vertex types not in the slice."
    )
    elided_edges: list[ElidedEdge] = PydanticField(
        default_factory=list, description="Edges not in the slice."
    )
    elided_properties: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description="Vertex type -> property names dropped from a surviving type.",
    )
    budget: BudgetAccounting = PydanticField(
        ..., description="Measured cost of the slice."
    )

    @property
    def truncated(self) -> bool:
        """Whether anything at all was left out."""
        return bool(self.elided_vertices or self.elided_edges or self.elided_properties)

truncated property

Whether anything at all was left out.

EntryPoint

Bases: ConfigBaseModel

A vertex type an agent can look up directly.

The single most useful fact about an unfamiliar graph: a type with a natural identity and a secondary index is one you can filter on cheaply, which is where a query should start.

Source code in graflo/architecture/schema/context/card.py

class EntryPoint(ConfigBaseModel):
    """A vertex type an agent can look up directly.

    The single most useful fact about an unfamiliar graph: a type with a natural
    identity *and* a secondary index is one you can filter on cheaply, which is
    where a query should start.
    """

    name: str = PydanticField(..., description="Vertex type name.")
    identity: list[str] = PydanticField(
        ..., description="Primary identity field names."
    )
    identity_mode: str = PydanticField(
        ..., description="natural / hash / assigned / blank."
    )
    secondary_identities: list[str] = PydanticField(
        default_factory=list, description="Declared secondary identity names."
    )
    indexed_fields: list[list[str]] = PydanticField(
        default_factory=list, description="Secondary index field-sets on this type."
    )
    description: str | None = PydanticField(
        default=None, description="Authored description, if any."
    )

RankingWeights

Bases: ConfigBaseModel

Relative weight of each local signal. Weights need not sum to 1.

Source code in graflo/architecture/schema/context/rank.py

class RankingWeights(ConfigBaseModel):
    """Relative weight of each local signal. Weights need not sum to 1."""

    hop_decay: float = PydanticField(
        default=0.55,
        description="Score multiplier per hop of distance from the nearest seed.",
        gt=0.0,
        le=1.0,
    )
    degree: float = PydanticField(
        default=0.20, description="Weight of normalized incident-edge count.", ge=0.0
    )
    identity: float = PydanticField(
        default=0.15, description="Weight of identity-mode strength.", ge=0.0
    )
    properties: float = PydanticField(
        default=0.10, description="Weight of log-scaled property count.", ge=0.0
    )
    indexed: float = PydanticField(
        default=0.10,
        description="Weight of secondary-index presence (cheap to filter on).",
        ge=0.0,
    )

SchemaCard

Bases: ConfigBaseModel

Bounded orientation summary of a whole schema.

Source code in graflo/architecture/schema/context/card.py

class SchemaCard(ConfigBaseModel):
    """Bounded orientation summary of a whole schema."""

    name: str = PydanticField(..., description="Schema name.")
    version: str | None = PydanticField(default=None, description="Schema version.")
    description: str | None = PydanticField(
        default=None, description="Authored schema description."
    )
    db_flavor: str = PydanticField(
        ..., description="Target database flavor from the db profile."
    )
    vertex_count: int = PydanticField(..., description="Declared vertex types.")
    edge_count: int = PydanticField(..., description="Declared edges.")
    total_property_count: int = PydanticField(
        ..., description="Declared properties across all vertex types."
    )
    hub_types: list[VertexSignals] = PydanticField(
        default_factory=list, description="Highest-ranked types, most central first."
    )
    entry_points: list[EntryPoint] = PydanticField(
        default_factory=list, description="Types that can be looked up directly."
    )
    identity_modes: dict[str, int] = PydanticField(
        default_factory=dict, description="Histogram of vertex identity modes."
    )
    isolated_types: list[str] = PydanticField(
        default_factory=list,
        description="Vertex types with no incident edge, truncated to ``max_names``.",
    )
    isolated_type_count: int = PydanticField(
        default=0, description="Total isolated types, including any not listed."
    )
    relation_vocabulary: list[str] = PydanticField(
        default_factory=list,
        description="Distinct edge relation names, truncated to ``max_names``.",
    )
    relation_count: int = PydanticField(
        default=0, description="Total distinct relations, including any not listed."
    )
    estimated_tokens: int = PydanticField(
        ..., description="Estimated token cost of this card."
    )

SchemaGraph

Read-only adjacency index over a :class:Schema's vertex types.

Built once per schema and never mutates it. Plain dicts throughout — no networkx, because this is layer 2 and the whole point is to stay free of heavyweight dependencies.

Source code in graflo/architecture/schema/context/graph.py

class SchemaGraph:
    """Read-only adjacency index over a :class:`Schema`'s vertex types.

    Built once per schema and never mutates it. Plain dicts throughout — no
    networkx, because this is layer 2 and the whole point is to stay free of
    heavyweight dependencies.
    """

    def __init__(self, schema: Schema) -> None:
        self._schema = schema
        core = schema.core_schema
        self._vertex_types = frozenset(core.vertex_config.vertex_set)
        self._out: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
        self._in: dict[str, list[EdgeId]] = {name: [] for name in self._vertex_types}
        self._edges: dict[EdgeId, Edge] = {}

        for edge in core.edge_config.edges:
            edge_id = edge.edge_id
            self._edges[edge_id] = edge
            source, target, _relation = edge_id
            if source in self._out:
                self._out[source].append(edge_id)
            if target in self._in:
                self._in[target].append(edge_id)

        for adjacency in (self._out, self._in):
            for edge_ids in adjacency.values():
                edge_ids.sort(key=edge_sort_key)

    @classmethod
    def from_schema(cls, schema: Schema) -> SchemaGraph:
        """Build an index for *schema*."""
        return cls(schema)

    @property
    def schema(self) -> Schema:
        """The indexed schema. Treat as read-only."""
        return self._schema

    @property
    def vertex_types(self) -> frozenset[str]:
        """Every declared vertex type name."""
        return self._vertex_types

    @property
    def edge_ids(self) -> list[EdgeId]:
        """Every declared edge id, in deterministic order."""
        return sorted(self._edges, key=edge_sort_key)

    def edge(self, edge_id: EdgeId) -> Edge:
        """Return the declared edge for *edge_id*."""
        return self._edges[edge_id]

    def out_edges(self, vertex_type: str) -> list[EdgeId]:
        """Edges whose source is *vertex_type*."""
        return list(self._out.get(vertex_type, []))

    def in_edges(self, vertex_type: str) -> list[EdgeId]:
        """Edges whose target is *vertex_type*."""
        return list(self._in.get(vertex_type, []))

    def degree(self, vertex_type: str) -> int:
        """Total incident edge count (out + in), counting self-loops twice."""
        return len(self._out.get(vertex_type, [])) + len(self._in.get(vertex_type, []))

    def isolated_types(self) -> list[str]:
        """Vertex types with no incident edge at all."""
        return sorted(name for name in self._vertex_types if self.degree(name) == 0)

    def relation_vocabulary(self) -> list[str]:
        """Distinct non-null relation names across all edges."""
        return sorted(
            {
                relation
                for _source, _target, relation in self._edges
                if relation is not None
            }
        )

    def _traversable(
        self,
        edge_id: EdgeId,
        anchor: str,
        direction: EdgeDirection,
    ) -> str | None:
        """Return the far endpoint when *edge_id* may be followed from *anchor*.

        An edge declared ``directed=False`` is traversable both ways regardless of
        the requested direction — the same rule
        :func:`~graflo.db.edge_direction_support.default_direction_for_edge`
        applies on the instance plane.
        """
        source, target, _relation = edge_id
        undirected = not self._edges[edge_id].directed
        effective = EdgeDirection.ANY if undirected else direction

        forward = source == anchor and effective in (
            EdgeDirection.OUT,
            EdgeDirection.ANY,
        )
        backward = target == anchor and effective in (
            EdgeDirection.IN,
            EdgeDirection.ANY,
        )
        if forward:
            return target
        if backward:
            return source
        return None

    def _incident(self, vertex_type: str) -> list[EdgeId]:
        """Every edge touching *vertex_type*, deduplicated (self-loops appear once)."""
        seen: set[EdgeId] = set()
        incident: list[EdgeId] = []
        for edge_id in self._out.get(vertex_type, []) + self._in.get(vertex_type, []):
            if edge_id in seen:
                continue
            seen.add(edge_id)
            incident.append(edge_id)
        return sorted(incident, key=edge_sort_key)

    def schema_neighbors(
        self,
        vertex_type: str,
        *,
        hops: int = 1,
        direction: EdgeDirection = EdgeDirection.ANY,
        edge_relations: set[str | None] | None = None,
    ) -> SchemaNeighborhood:
        """Vertex types adjacent to *vertex_type* within *hops*.

        Args:
            vertex_type: Seed vertex type. Must be declared.
            hops: Maximum hop distance. ``0`` returns just the seed.
            direction: Orientation followed from each frontier vertex. Defaults to
                :attr:`EdgeDirection.ANY` — deliberately unlike
                ``Connection.fetch_edges``, which defaults to ``OUT``. "What is
                adjacent to ``person`` in the schema" almost never means "only
                where person is the source"; an agent asking that wants the whole
                local shape. Edges declared ``directed=False`` are followed both
                ways whatever is requested here.
            edge_relations: Restrict traversal to these relation names (``None`` is
                a valid member, matching edges with no relation).

        Returns:
            SchemaNeighborhood: distances per reachable type and the edges used.

        Raises:
            KeyError: if *vertex_type* is not declared in the schema.
        """
        if vertex_type not in self._vertex_types:
            raise KeyError(
                f"Unknown vertex type {vertex_type!r}; declared: {sorted(self._vertex_types)}"
            )
        if hops < 0:
            raise ValueError(f"hops must be >= 0, got {hops}")

        distances: dict[str, int] = {vertex_type: 0}
        used: set[EdgeId] = set()
        frontier: deque[tuple[str, int]] = deque([(vertex_type, 0)])

        while frontier:
            current, depth = frontier.popleft()
            if depth >= hops:
                continue
            for edge_id in self._incident(current):
                if edge_relations is not None and edge_id[2] not in edge_relations:
                    continue
                far = self._traversable(edge_id, current, direction)
                if far is None:
                    continue
                used.add(edge_id)
                if far not in distances:
                    distances[far] = depth + 1
                    frontier.append((far, depth + 1))

        return SchemaNeighborhood(
            origin=vertex_type,
            hops=hops,
            direction=direction,
            distances=distances,
            edges=sorted(used, key=edge_sort_key),
        )

    def relations_between(
        self,
        a: str,
        b: str,
        *,
        max_len: int = 3,
        max_paths: int = 20,
        direction: EdgeDirection = EdgeDirection.ANY,
    ) -> list[SchemaPath]:
        """Simple paths from vertex type *a* to *b*, shortest first.

        Bounded breadth-first enumeration: no vertex repeats within a path, so
        cycles terminate. Results are ordered by ``(length, edge ids)`` and are
        therefore reproducible run to run.

        Args:
            a: Source vertex type.
            b: Target vertex type.
            max_len: Maximum hops per path.
            max_paths: Maximum number of paths returned.
            direction: Orientation followed from each frontier vertex.

        Returns:
            list[SchemaPath]: paths found, possibly empty.

        Raises:
            KeyError: if either endpoint is not declared in the schema.
        """
        for name in (a, b):
            if name not in self._vertex_types:
                raise KeyError(
                    f"Unknown vertex type {name!r}; declared: {sorted(self._vertex_types)}"
                )
        if max_len < 1 or max_paths < 1:
            return []

        found: list[SchemaPath] = []
        queue: deque[tuple[str, list[str], list[EdgeId]]] = deque([(a, [a], [])])

        while queue and len(found) < max_paths:
            current, vertices, edges = queue.popleft()
            if len(edges) >= max_len:
                continue
            for edge_id in self._incident(current):
                far = self._traversable(edge_id, current, direction)
                if far is None:
                    continue
                # Paths stay simple, except that reaching the target closes the
                # walk — which is what makes ``relations_between(a, a)`` return
                # self-loops and cycles rather than nothing.
                if far in vertices and far != b:
                    continue
                next_vertices = [*vertices, far]
                next_edges = [*edges, edge_id]
                if far == b:
                    found.append(SchemaPath(vertices=next_vertices, edges=next_edges))
                    if len(found) >= max_paths:
                        break
                else:
                    queue.append((far, next_vertices, next_edges))

        found.sort(
            key=lambda path: (path.length, [edge_sort_key(e) for e in path.edges])
        )
        return found[:max_paths]

edge_ids property

Every declared edge id, in deterministic order.

schema property

The indexed schema. Treat as read-only.

vertex_types property

Every declared vertex type name.

degree(vertex_type)

Total incident edge count (out + in), counting self-loops twice.

Source code in graflo/architecture/schema/context/graph.py

def degree(self, vertex_type: str) -> int:
    """Total incident edge count (out + in), counting self-loops twice."""
    return len(self._out.get(vertex_type, [])) + len(self._in.get(vertex_type, []))

edge(edge_id)

Return the declared edge for edge_id.

Source code in graflo/architecture/schema/context/graph.py

def edge(self, edge_id: EdgeId) -> Edge:
    """Return the declared edge for *edge_id*."""
    return self._edges[edge_id]

from_schema(schema) classmethod

Build an index for schema.

Source code in graflo/architecture/schema/context/graph.py

@classmethod
def from_schema(cls, schema: Schema) -> SchemaGraph:
    """Build an index for *schema*."""
    return cls(schema)

in_edges(vertex_type)

Edges whose target is vertex_type.

Source code in graflo/architecture/schema/context/graph.py

def in_edges(self, vertex_type: str) -> list[EdgeId]:
    """Edges whose target is *vertex_type*."""
    return list(self._in.get(vertex_type, []))

isolated_types()

Vertex types with no incident edge at all.

Source code in graflo/architecture/schema/context/graph.py

def isolated_types(self) -> list[str]:
    """Vertex types with no incident edge at all."""
    return sorted(name for name in self._vertex_types if self.degree(name) == 0)

out_edges(vertex_type)

Edges whose source is vertex_type.

Source code in graflo/architecture/schema/context/graph.py

def out_edges(self, vertex_type: str) -> list[EdgeId]:
    """Edges whose source is *vertex_type*."""
    return list(self._out.get(vertex_type, []))

relation_vocabulary()

Distinct non-null relation names across all edges.

Source code in graflo/architecture/schema/context/graph.py

def relation_vocabulary(self) -> list[str]:
    """Distinct non-null relation names across all edges."""
    return sorted(
        {
            relation
            for _source, _target, relation in self._edges
            if relation is not None
        }
    )

relations_between(a, b, *, max_len=3, max_paths=20, direction=EdgeDirection.ANY)

Simple paths from vertex type a to b, shortest first.

Bounded breadth-first enumeration: no vertex repeats within a path, so cycles terminate. Results are ordered by (length, edge ids) and are therefore reproducible run to run.

Parameters:

Name	Type	Description	Default
a	str	Source vertex type.	required
b	str	Target vertex type.	required
max_len	int	Maximum hops per path.	3
max_paths	int	Maximum number of paths returned.	20
direction	EdgeDirection	Orientation followed from each frontier vertex.	ANY

Returns:

Type	Description
list[SchemaPath]	list[SchemaPath]: paths found, possibly empty.

Raises:

Type	Description
KeyError	if either endpoint is not declared in the schema.

Source code in graflo/architecture/schema/context/graph.py

def relations_between(
    self,
    a: str,
    b: str,
    *,
    max_len: int = 3,
    max_paths: int = 20,
    direction: EdgeDirection = EdgeDirection.ANY,
) -> list[SchemaPath]:
    """Simple paths from vertex type *a* to *b*, shortest first.

    Bounded breadth-first enumeration: no vertex repeats within a path, so
    cycles terminate. Results are ordered by ``(length, edge ids)`` and are
    therefore reproducible run to run.

    Args:
        a: Source vertex type.
        b: Target vertex type.
        max_len: Maximum hops per path.
        max_paths: Maximum number of paths returned.
        direction: Orientation followed from each frontier vertex.

    Returns:
        list[SchemaPath]: paths found, possibly empty.

    Raises:
        KeyError: if either endpoint is not declared in the schema.
    """
    for name in (a, b):
        if name not in self._vertex_types:
            raise KeyError(
                f"Unknown vertex type {name!r}; declared: {sorted(self._vertex_types)}"
            )
    if max_len < 1 or max_paths < 1:
        return []

    found: list[SchemaPath] = []
    queue: deque[tuple[str, list[str], list[EdgeId]]] = deque([(a, [a], [])])

    while queue and len(found) < max_paths:
        current, vertices, edges = queue.popleft()
        if len(edges) >= max_len:
            continue
        for edge_id in self._incident(current):
            far = self._traversable(edge_id, current, direction)
            if far is None:
                continue
            # Paths stay simple, except that reaching the target closes the
            # walk — which is what makes ``relations_between(a, a)`` return
            # self-loops and cycles rather than nothing.
            if far in vertices and far != b:
                continue
            next_vertices = [*vertices, far]
            next_edges = [*edges, edge_id]
            if far == b:
                found.append(SchemaPath(vertices=next_vertices, edges=next_edges))
                if len(found) >= max_paths:
                    break
            else:
                queue.append((far, next_vertices, next_edges))

    found.sort(
        key=lambda path: (path.length, [edge_sort_key(e) for e in path.edges])
    )
    return found[:max_paths]

schema_neighbors(vertex_type, *, hops=1, direction=EdgeDirection.ANY, edge_relations=None)

Vertex types adjacent to vertex_type within hops.

Parameters:

Name	Type	Description	Default
vertex_type	str	Seed vertex type. Must be declared.	required
hops	int	Maximum hop distance. 0 returns just the seed.	1
direction	EdgeDirection	Orientation followed from each frontier vertex. Defaults to :attr:EdgeDirection.ANY — deliberately unlike Connection.fetch_edges, which defaults to OUT. "What is adjacent to person in the schema" almost never means "only where person is the source"; an agent asking that wants the whole local shape. Edges declared directed=False are followed both ways whatever is requested here.	ANY
edge_relations	set[str | None] | None	Restrict traversal to these relation names (None is a valid member, matching edges with no relation).	None

Returns:

Name	Type	Description
SchemaNeighborhood	SchemaNeighborhood	distances per reachable type and the edges used.

Raises:

Type	Description
KeyError	if vertex_type is not declared in the schema.

Source code in graflo/architecture/schema/context/graph.py

def schema_neighbors(
    self,
    vertex_type: str,
    *,
    hops: int = 1,
    direction: EdgeDirection = EdgeDirection.ANY,
    edge_relations: set[str | None] | None = None,
) -> SchemaNeighborhood:
    """Vertex types adjacent to *vertex_type* within *hops*.

    Args:
        vertex_type: Seed vertex type. Must be declared.
        hops: Maximum hop distance. ``0`` returns just the seed.
        direction: Orientation followed from each frontier vertex. Defaults to
            :attr:`EdgeDirection.ANY` — deliberately unlike
            ``Connection.fetch_edges``, which defaults to ``OUT``. "What is
            adjacent to ``person`` in the schema" almost never means "only
            where person is the source"; an agent asking that wants the whole
            local shape. Edges declared ``directed=False`` are followed both
            ways whatever is requested here.
        edge_relations: Restrict traversal to these relation names (``None`` is
            a valid member, matching edges with no relation).

    Returns:
        SchemaNeighborhood: distances per reachable type and the edges used.

    Raises:
        KeyError: if *vertex_type* is not declared in the schema.
    """
    if vertex_type not in self._vertex_types:
        raise KeyError(
            f"Unknown vertex type {vertex_type!r}; declared: {sorted(self._vertex_types)}"
        )
    if hops < 0:
        raise ValueError(f"hops must be >= 0, got {hops}")

    distances: dict[str, int] = {vertex_type: 0}
    used: set[EdgeId] = set()
    frontier: deque[tuple[str, int]] = deque([(vertex_type, 0)])

    while frontier:
        current, depth = frontier.popleft()
        if depth >= hops:
            continue
        for edge_id in self._incident(current):
            if edge_relations is not None and edge_id[2] not in edge_relations:
                continue
            far = self._traversable(edge_id, current, direction)
            if far is None:
                continue
            used.add(edge_id)
            if far not in distances:
                distances[far] = depth + 1
                frontier.append((far, depth + 1))

    return SchemaNeighborhood(
        origin=vertex_type,
        hops=hops,
        direction=direction,
        distances=distances,
        edges=sorted(used, key=edge_sort_key),
    )

SchemaNeighborhood

Bases: ConfigBaseModel

Vertex types reachable from a seed within a hop bound.

Source code in graflo/architecture/schema/context/graph.py

class SchemaNeighborhood(ConfigBaseModel):
    """Vertex types reachable from a seed within a hop bound."""

    origin: str = PydanticField(..., description="Vertex type the walk started from.")
    hops: int = PydanticField(..., description="Hop bound the walk honoured.")
    direction: EdgeDirection = PydanticField(
        ..., description="Orientation followed from each frontier vertex."
    )
    distances: dict[str, int] = PydanticField(
        ...,
        description="Reachable vertex type -> hop distance from origin (origin itself is 0).",
    )
    edges: list[EdgeId] = PydanticField(
        ..., description="Edges traversed to reach the neighbourhood, deduplicated."
    )

    @property
    def vertex_types(self) -> list[str]:
        """Reachable vertex types, nearest first then alphabetical."""
        return sorted(self.distances, key=lambda name: (self.distances[name], name))

vertex_types property

Reachable vertex types, nearest first then alphabetical.

SchemaPath

Bases: ConfigBaseModel

One path between two vertex types, as an alternating vertex/edge walk.

Source code in graflo/architecture/schema/context/graph.py

class SchemaPath(ConfigBaseModel):
    """One path between two vertex types, as an alternating vertex/edge walk."""

    vertices: list[str] = PydanticField(
        ..., description="Vertex types visited, from source to target inclusive."
    )
    edges: list[EdgeId] = PydanticField(
        ..., description="Edges traversed, one fewer than ``vertices``."
    )

    @property
    def length(self) -> int:
        """Number of hops (edges) in this path."""
        return len(self.edges)

length property

Number of hops (edges) in this path.

VertexSignals

Bases: ConfigBaseModel

Per-vertex-type ranking inputs and the score derived from them.

Source code in graflo/architecture/schema/context/rank.py

class VertexSignals(ConfigBaseModel):
    """Per-vertex-type ranking inputs and the score derived from them."""

    name: str = PydanticField(..., description="Vertex type name.")
    hop_distance: int | None = PydanticField(
        default=None,
        description="Hops from the nearest seed; None when unreachable.",
    )
    degree: int = PydanticField(..., description="Incident edges (out + in).")
    identity_mode: str = PydanticField(
        ..., description="One of natural / hash / assigned / blank."
    )
    property_count: int = PydanticField(..., description="Declared property count.")
    has_secondary_index: bool = PydanticField(
        ..., description="Whether db_profile declares a secondary index for this type."
    )
    score: float = PydanticField(..., description="Composite rank; higher is better.")

build_card(schema, *, top_n=10, max_names=25, graph=None)

Summarize schema for an agent's first contact with it.

Every list on the card is bounded, with a count reported alongside. A card whose size grows with the schema is not a card — it is the problem this wave exists to solve, wearing a summary's clothes.

Parameters:

Name	Type	Description	Default
schema	Schema	Schema to summarize. Never mutated.	required
top_n	int	How many hub types and entry points to list.	10
max_names	int	How many isolated types and relation names to list.	25
graph	SchemaGraph | None	Prebuilt index, if the caller already has one.	None

Source code in graflo/architecture/schema/context/card.py

def build_card(
    schema: Schema,
    *,
    top_n: int = 10,
    max_names: int = 25,
    graph: SchemaGraph | None = None,
) -> SchemaCard:
    """Summarize *schema* for an agent's first contact with it.

    Every list on the card is bounded, with a count reported alongside. A card
    whose size grows with the schema is not a card — it is the problem this wave
    exists to solve, wearing a summary's clothes.

    Args:
        schema: Schema to summarize. Never mutated.
        top_n: How many hub types and entry points to list.
        max_names: How many isolated types and relation names to list.
        graph: Prebuilt index, if the caller already has one.
    """
    graph = graph or SchemaGraph.from_schema(schema)
    vertex_config = schema.core_schema.vertex_config
    db_profile = schema.db_profile

    ranked = score_vertices(graph)
    isolated = graph.isolated_types()
    relations = graph.relation_vocabulary()
    identity_modes = Counter(
        vertex_config[name].identity_mode for name in graph.vertex_types
    )

    entry_points: list[EntryPoint] = []
    for signal in ranked:
        if len(entry_points) >= top_n:
            break
        vertex = vertex_config[signal.name]
        indexes = db_profile.vertex_secondary_indexes(signal.name)
        # A blank type has no natural key and nothing to filter on: it is not an
        # entry point, whatever its centrality.
        if vertex.identity_mode == "blank" and not indexes:
            continue
        if not vertex.identity and not indexes:
            continue
        entry_points.append(
            EntryPoint(
                name=signal.name,
                identity=list(vertex.identity),
                identity_mode=vertex.identity_mode,
                secondary_identities=vertex.secondary_identity_names,
                indexed_fields=[list(index.fields) for index in indexes],
                description=vertex.description,
            )
        )

    card = SchemaCard(
        name=schema.metadata.name,
        version=schema.metadata.version,
        description=schema.metadata.description,
        db_flavor=str(db_profile.db_flavor),
        vertex_count=len(graph.vertex_types),
        edge_count=len(graph.edge_ids),
        total_property_count=sum(
            len(vertex_config[name].property_names) for name in graph.vertex_types
        ),
        hub_types=ranked[:top_n],
        entry_points=entry_points,
        identity_modes=dict(sorted(identity_modes.items())),
        isolated_types=isolated[:max_names],
        isolated_type_count=len(isolated),
        relation_vocabulary=relations[:max_names],
        relation_count=len(relations),
        estimated_tokens=0,
    )
    card.estimated_tokens = estimate_tokens(card.to_minimal_canonical_dict())
    return card

estimate_tokens(payload)

Estimate token count for payload.

Runs over the compact serialization — estimating over a pretty-printed or defaults-included dump overcounts by a factor of two or more, which would make every budget silently pessimistic.

Source code in graflo/architecture/schema/context/budget.py

def estimate_tokens(payload: Any) -> int:
    """Estimate token count for *payload*.

    Runs over the compact serialization — estimating over a pretty-printed or
    defaults-included dump overcounts by a factor of two or more, which would make
    every budget silently pessimistic.
    """
    return math.ceil(len(serialize_compact(payload)) / CHARS_PER_TOKEN)

score_vertices(graph, seeds=(), *, weights=None, max_hops=3, direction=EdgeDirection.ANY)

Rank every vertex type in graph, highest score first.

With no seeds, ranking is seed-independent (structure only) and answers "what are the important types here" — which is what the orientation card needs. With seeds, hop distance dominates and answers "what is near what I asked about".

Ties break by vertex name ascending. This is not cosmetic: without a total order the elision report is not reproducible across runs, and the budget tests become flaky.

Source code in graflo/architecture/schema/context/rank.py

def score_vertices(
    graph: SchemaGraph,
    seeds: Sequence[str] = (),
    *,
    weights: RankingWeights | None = None,
    max_hops: int = 3,
    direction: EdgeDirection = EdgeDirection.ANY,
) -> list[VertexSignals]:
    """Rank every vertex type in *graph*, highest score first.

    With no *seeds*, ranking is seed-independent (structure only) and answers
    "what are the important types here" — which is what the orientation card
    needs. With seeds, hop distance dominates and answers "what is near what I
    asked about".

    Ties break by vertex name ascending. This is not cosmetic: without a total
    order the elision report is not reproducible across runs, and the budget tests
    become flaky.
    """
    weights = weights or RankingWeights()
    schema = graph.schema
    vertex_config = schema.core_schema.vertex_config
    db_profile = schema.db_profile

    distances: dict[str, int] = {}
    for seed in seeds:
        neighborhood = graph.schema_neighbors(seed, hops=max_hops, direction=direction)
        for name, distance in neighborhood.distances.items():
            if name not in distances or distance < distances[name]:
                distances[name] = distance

    degrees = {name: graph.degree(name) for name in graph.vertex_types}
    max_degree = max(degrees.values(), default=0)
    property_counts = {
        name: len(vertex_config[name].property_names) for name in graph.vertex_types
    }
    max_properties = max(property_counts.values(), default=0)

    signals: list[VertexSignals] = []
    for name in sorted(graph.vertex_types):
        vertex = vertex_config[name]
        hop_distance = distances.get(name) if seeds else None
        degree = degrees[name]
        property_count = property_counts[name]
        has_index = bool(db_profile.vertex_secondary_indexes(name))

        structural = (
            weights.degree * (degree / max_degree if max_degree else 0.0)
            + weights.identity * IDENTITY_MODE_STRENGTH.get(vertex.identity_mode, 0.5)
            + weights.properties
            * (
                math.log1p(property_count) / math.log1p(max_properties)
                if max_properties
                else 0.0
            )
            + weights.indexed * (1.0 if has_index else 0.0)
        )
        if not seeds:
            score = structural
        elif hop_distance is None:
            score = 0.0
        else:
            score = (weights.hop_decay**hop_distance) + structural

        signals.append(
            VertexSignals(
                name=name,
                hop_distance=hop_distance,
                degree=degree,
                identity_mode=vertex.identity_mode,
                property_count=property_count,
                has_secondary_index=has_index,
                score=score,
            )
        )

    signals.sort(key=lambda item: (-item.score, item.name))
    return signals