graflo.architecture.schema.context.subschema

Bounded, seeded schema slicing.

The deliverable of schema context is not a getter: it is a valid standalone :class:~graflo.architecture.schema.document.Schema small enough to hand an agent, paired with an explicit account of what was left out and how to get it.

Attributes

Classes

Functions:

protected_property_names(vertex)

Property names that must never be elided from vertex.

Identity-bearing fields: the primary identity, every secondary identity, and whatever feeds a hash or funnel digest. Dropping one of these yields a schema that still validates but is wrong — the agent would believe a type is addressable by fields it cannot actually be addressed by.

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

def protected_property_names(vertex: Vertex) -> set[str]:
    """Property names that must never be elided from *vertex*.

    Identity-bearing fields: the primary identity, every secondary identity, and
    whatever feeds a hash or funnel digest. Dropping one of these yields a schema
    that still validates but is wrong — the agent would believe a
    type is addressable by fields it cannot actually be addressed by.
    """
    protected: set[str] = set(vertex.identity)
    for entry in vertex.secondary_identities:
        protected.update(entry.fields)
    protected.update(vertex.digest_source_fields)
    return protected

subschema(schema, seeds, *, budget=None, max_hops=3, weights=None, direction=EdgeDirection.ANY, graph=None)

Slice schema down to a budgeted neighbourhood around seeds.

Seeds are always admitted, even when the budget cannot afford them: a slice that omits what the caller explicitly asked about answers a different question than the one posed. An edge seed admits the edge and both of its endpoints, which then count as vertex seeds.

Admission order is: seeds, the edges between seeds, then ranked non-seed vertices, each together with the edges joining it to what is already admitted, then any remaining edges among everything admitted. A vertex whose joining edges do not fit is not admitted: a type the slice cannot connect is noise. Trimming to the token ceiling runs the other way and spares the edges between seeds until every non-seed vertex is gone, so a tight budget never trades how the seeds relate for a type further away.

Parameters:

Name	Type	Description	Default
schema	Schema	Source schema. Never mutated.	required
seeds	Sequence[str | EdgeId]	Vertex types, or edge ids (source, target, relation), to centre the slice on. Must all be declared.	required
budget	Budget | None	Element and token ceilings. Defaults to :class:Budget.	None
max_hops	int	How far from a seed a type may be and still be a candidate.	3
weights	RankingWeights | None	Ranking weights for admission order.	None
direction	EdgeDirection	Orientation followed when expanding from seeds.	ANY
graph	SchemaGraph | None	Prebuilt index, if the caller already has one.	None

Returns:

Name	Type	Description
tuple	tuple[Schema, ElisionReport]	the slice, and an :class:ElisionReport describing what it omits.

Raises:

Type	Description
ValueError	if seeds is empty.
KeyError	if a seed is not declared in schema.

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

def subschema(
    schema: Schema,
    seeds: Sequence[str | EdgeId],
    *,
    budget: Budget | None = None,
    max_hops: int = 3,
    weights: RankingWeights | None = None,
    direction: EdgeDirection = EdgeDirection.ANY,
    graph: SchemaGraph | None = None,
) -> tuple[Schema, ElisionReport]:
    """Slice *schema* down to a budgeted neighbourhood around *seeds*.

    Seeds are always admitted, even when the budget cannot afford them: a slice
    that omits what the caller explicitly asked about answers a different
    question than the one posed. An edge seed admits the edge and both of its
    endpoints, which then count as vertex seeds.

    Admission order is: seeds, the edges between seeds, then ranked non-seed
    vertices, each together with the edges joining it to what is already
    admitted, then any remaining edges among everything admitted. A vertex
    whose joining edges do not fit is not admitted: a type the slice cannot
    connect is noise. Trimming to the token ceiling runs the other way and
    spares the edges between seeds until every non-seed vertex is gone, so a
    tight budget never trades how the seeds relate for a type further away.

    Args:
        schema: Source schema. Never mutated.
        seeds: Vertex types, or edge ids ``(source, target, relation)``, to
            centre the slice on. Must all be declared.
        budget: Element and token ceilings. Defaults to :class:`Budget`.
        max_hops: How far from a seed a type may be and still be a candidate.
        weights: Ranking weights for admission order.
        direction: Orientation followed when expanding from seeds.
        graph: Prebuilt index, if the caller already has one.

    Returns:
        tuple: the slice, and an :class:`ElisionReport` describing what it omits.

    Raises:
        ValueError: if *seeds* is empty.
        KeyError: if a seed is not declared in *schema*.
    """
    if not seeds:
        raise ValueError("subschema requires at least one seed vertex type or edge")

    graph = graph or SchemaGraph.from_schema(schema)
    budget = budget or Budget()
    declared_edges = set(graph.edge_ids)
    seed_vertices: list[str] = []
    seed_edges: list[EdgeId] = []
    for seed in seeds:
        if isinstance(seed, str):
            seed_vertices.append(seed)
            continue
        edge_id: EdgeId = (seed[0], seed[1], seed[2])
        if edge_id not in declared_edges:
            raise KeyError(f"Unknown seed edge {edge_id!r}")
        if edge_id not in seed_edges:
            seed_edges.append(edge_id)
        seed_vertices.extend(edge_id[:2])
    seed_list = list(dict.fromkeys(seed_vertices))
    for seed in seed_list:
        if seed not in graph.vertex_types:
            raise KeyError(
                f"Unknown seed vertex type {seed!r}; declared: {sorted(graph.vertex_types)}"
            )
    seed_set = set(seed_list)

    ranked = score_vertices(
        graph,
        seed_list,
        weights=weights,
        max_hops=max_hops,
        direction=direction,
    )
    signals_by_name = {signal.name: signal for signal in ranked}

    vertex_config = schema.core_schema.vertex_config
    vertex_cost = {
        name: estimate_tokens(vertex_config[name].to_minimal_canonical_dict())
        for name in graph.vertex_types
    }
    edge_cost = {
        edge_id: estimate_tokens(graph.edge(edge_id).to_minimal_canonical_dict())
        for edge_id in graph.edge_ids
    }

    # Fixed cost of the envelope (metadata + db profile). Counting only element
    # costs would let an assembled slice sail past ``max_tokens`` — the caller
    # would be handed a payload larger than the ceiling they set. The source
    # profile is an upper bound on the projected one, so this errs conservative.
    envelope_cost = estimate_tokens(
        {
            "metadata": schema.metadata.to_minimal_canonical_dict(),
            "db_profile": schema.db_profile.to_minimal_canonical_dict(),
        }
    )

    admitted: list[str] = list(seed_list)
    elements_used = len(admitted) + len(seed_edges)
    tokens_used = (
        envelope_cost
        + sum(vertex_cost[name] for name in admitted)
        + sum(edge_cost[edge_id] for edge_id in seed_edges)
    )
    exhausted: str = "none"
    seed_edge_set = set(seed_edges)
    rank_position = {signal.name: index for index, signal in enumerate(ranked)}

    def _edge_rank(edge_id: EdgeId) -> tuple[int, tuple[str, str, str]]:
        return (
            min(rank_position[edge_id[0]], rank_position[edge_id[1]]),
            edge_sort_key(edge_id),
        )

    between_seeds = sorted(
        (
            edge_id
            for edge_id in graph.edge_ids
            if edge_id[0] in seed_set
            and edge_id[1] in seed_set
            and edge_id not in seed_edge_set
        ),
        key=_edge_rank,
    )
    seed_pair_edges: list[EdgeId] = []
    for edge_id in between_seeds:
        if budget.max_elements is not None and elements_used + 1 > budget.max_elements:
            exhausted = "elements"
            break
        cost = edge_cost[edge_id]
        if budget.max_tokens is not None and tokens_used + cost > budget.max_tokens:
            exhausted = "tokens"
            break
        seed_pair_edges.append(edge_id)
        elements_used += 1
        tokens_used += cost

    admitted_set = set(admitted)
    taken_edges = seed_edge_set | set(seed_pair_edges)
    # Each admitted non-seed vertex with the edges that joined it.
    units: list[tuple[str, list[EdgeId]]] = []
    for signal in ranked:
        if signal.name in seed_set:
            continue
        # Unreachable within max_hops: not a budget decision, so it never sets
        # `exhausted` and never stops admission of reachable candidates.
        if signal.hop_distance is None:
            continue
        name = signal.name
        joining = sorted(
            (
                edge_id
                for edge_id in graph.edge_ids
                if edge_id not in taken_edges
                and name in edge_id[:2]
                and all(end == name or end in admitted_set for end in edge_id[:2])
            ),
            key=_edge_rank,
        )
        unit_elements = 1 + len(joining)
        unit_tokens = vertex_cost[name] + sum(edge_cost[e] for e in joining)
        if (
            budget.max_elements is not None
            and elements_used + unit_elements > budget.max_elements
        ):
            exhausted = "elements"
            break
        if (
            budget.max_tokens is not None
            and tokens_used + unit_tokens > budget.max_tokens
        ):
            exhausted = "tokens"
            break
        admitted.append(name)
        admitted_set.add(name)
        taken_edges.update(joining)
        units.append((name, joining))
        elements_used += unit_elements
        tokens_used += unit_tokens

    candidate_edges = sorted(
        (
            edge_id
            for edge_id in graph.edge_ids
            if edge_id[0] in admitted_set
            and edge_id[1] in admitted_set
            and edge_id not in taken_edges
        ),
        key=_edge_rank,
    )

    admitted_edges: list[EdgeId] = []
    for edge_id in candidate_edges:
        if budget.max_elements is not None and elements_used + 1 > budget.max_elements:
            exhausted = "elements"
            break
        cost = edge_cost[edge_id]
        if budget.max_tokens is not None and tokens_used + cost > budget.max_tokens:
            exhausted = "tokens"
            break
        admitted_edges.append(edge_id)
        elements_used += 1
        tokens_used += cost

    drop_properties = {}
    for name in admitted:
        dropped = _drop_properties_for(
            vertex_config[name], budget.max_properties_per_vertex
        )
        if dropped:
            drop_properties[name] = set(dropped)

    # Per-element costs omit the structural overhead of nesting them (keys,
    # separators, list wrappers), so the greedy pass can overshoot by a few
    # percent. Measure the assembled slice and trim until the ceiling actually
    # holds — a budget that is exceeded by the payload it produced is not a
    # budget. Least valuable first: edges admitted last, then lowest-ranked
    # non-seed vertices with the edges that joined them, then edges between
    # seeds.
    trimmable_edges = list(admitted_edges)
    trimmable_units = list(units)
    trimmable_seed_pairs = list(seed_pair_edges)
    while True:
        selection = select_induced(
            schema.core_schema,
            keep_vertices=set(seed_list) | {name for name, _ in trimmable_units},
            keep_edge_ids=seed_edge_set
            | set(trimmable_edges)
            | {edge_id for _, joined in trimmable_units for edge_id in joined}
            | set(trimmable_seed_pairs),
            connectivity="induced",
        )
        sliced = build_subschema(schema, selection, drop_properties=drop_properties)
        payload = sliced.to_minimal_canonical_dict()
        estimated = estimate_tokens(payload)
        if budget.max_tokens is None or estimated <= budget.max_tokens:
            break
        # Seeds — vertex and edge — are never trimmed: a slice that drops what the caller asked about
        # answers a different question. If the seeds alone blow the budget, the
        # overrun is reported rather than hidden.
        if trimmable_edges:
            trimmable_edges.pop()
        elif trimmable_units:
            trimmable_units.pop()
        elif trimmable_seed_pairs:
            trimmable_seed_pairs.pop()
        else:
            break
        exhausted = "tokens"

    accounting = BudgetAccounting(
        requested=budget,
        elements_used=len(selection.surviving_vertices)
        + len(selection.surviving_edge_ids),
        estimated_tokens=estimate_tokens(payload),
        serialized_chars=len(serialize_compact(payload)),
        exhausted_by=exhausted,
    )

    return sliced, _build_report(
        graph=graph,
        selection_vertices=selection.surviving_vertices,
        selection_edges=selection.surviving_edge_ids,
        signals_by_name=signals_by_name,
        drop_properties=drop_properties,
        accounting=accounting,
    )