lpgcast.select

Stage A selection: the slice of one schema a text is about.

Two mechanisms, kept apart. Seeds come from retrieval: lexical lanes over the schema's element text, fused by rank. Structure comes from closure: graflo's subschema() grows the seeds into a valid standalone schema under a budget and never drops identity fields. A schema already under the budget is passed whole.

Lanes:

• exact — code-like names and synonyms (PLC, WorkOrder, an IRI) found verbatim, case-sensitively, in the text.
• lexical — names and synonyms found as phrases in the text after case-folding, camel/snake splitting and a light suffix stemmer, weighted by how few elements share the phrase; plus description words the text repeats.
• unit — a number followed by a token that is a property's declared unit (200 kV) seeds that property's owner.

A hit on a property seeds its owner — the vertex or edge — at half weight.

Attributes

DESCRIPTION_MIN_WORDS = 2 module-attribute

DESCRIPTION_WEIGHT = 0.25 module-attribute

LANES = ('exact', 'lexical', 'unit') module-attribute

Lane = Literal['exact', 'lexical', 'unit'] module-attribute

MAX_PHRASE_WORDS = 8 module-attribute

PROPERTY_HIT_WEIGHT = 0.5 module-attribute

Seed = str | EdgeId module-attribute

Classes

RetrievalMetrics

Bases: BaseModel

What retrieval found, before and after each cut.

fused == 0 means nothing matched; fused > 0 with no seeds would mean the gate rejected everything — kept apart because they have different fixes.

Source code in lpgcast/select.py

class RetrievalMetrics(BaseModel):
    """What retrieval found, before and after each cut.

    ``fused == 0`` means nothing matched; ``fused > 0`` with no seeds would mean
    the gate rejected everything — kept apart because they have different fixes.
    """

    lane_hits: dict[Lane, int] = Field(description="Distinct seeds per lane.")
    fused: int = Field(description="Distinct seeds over all lanes.")
    gated_out: int = Field(description="Seeds below the relevance gate.")
    capped: int = Field(description="Seeds over max_seeds.")
    seeds: int = Field(description="Seeds passed on.")
    schema_tokens: int = Field(description="Estimated size of the whole schema.")

Attributes

capped = Field(description='Seeds over max_seeds.') class-attribute instance-attribute

fused = Field(description='Distinct seeds over all lanes.') class-attribute instance-attribute

gated_out = Field(description='Seeds below the relevance gate.') class-attribute instance-attribute

lane_hits = Field(description='Distinct seeds per lane.') class-attribute instance-attribute

schema_tokens = Field(description='Estimated size of the whole schema.') class-attribute instance-attribute

seeds = Field(description='Seeds passed on.') class-attribute instance-attribute

SeedHit

Bases: BaseModel

A seed and how it was found.

Source code in lpgcast/select.py

class SeedHit(BaseModel):
    """A seed and how it was found."""

    vertex: str | None = Field(default=None, description="Seed vertex type.")
    edge_id: EdgeId | None = Field(default=None, description="Seed edge.")
    score: float = Field(description="Fused score.")
    ranks: dict[Lane, int] = Field(description="Rank within each lane that found it.")

    @property
    def seed(self) -> Seed:
        if self.edge_id is not None:
            return self.edge_id
        assert self.vertex is not None
        return self.vertex

Attributes

edge_id = Field(default=None, description='Seed edge.') class-attribute instance-attribute

ranks = Field(description='Rank within each lane that found it.') class-attribute instance-attribute

score = Field(description='Fused score.') class-attribute instance-attribute

seed property

vertex = Field(default=None, description='Seed vertex type.') class-attribute instance-attribute

Selection

Bases: BaseModel

The schema a cast runs against, and how it was chosen.

Source code in lpgcast/select.py

class Selection(BaseModel):
    """The schema a cast runs against, and how it was chosen."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    slice: Schema = Field(description="A valid standalone schema.")
    mode: Literal["whole", "slice"] = Field(
        description="``whole``: the schema fit the budget and was passed as is."
    )
    seeds: list[SeedHit] = Field(description="Seeds, best first.")
    elision: ElisionReport | None = Field(
        default=None, description="What closure left out; None when whole."
    )
    metrics: RetrievalMetrics

Attributes

elision = Field(default=None, description='What closure left out; None when whole.') class-attribute instance-attribute

metrics instance-attribute

mode = Field(description='``whole``: the schema fit the budget and was passed as is.') class-attribute instance-attribute

model_config = ConfigDict(arbitrary_types_allowed=True) class-attribute instance-attribute

seeds = Field(description='Seeds, best first.') class-attribute instance-attribute

slice = Field(description='A valid standalone schema.') class-attribute instance-attribute

SelectionConfig

Bases: BaseModel

Tuning for stage A selection.

Source code in lpgcast/select.py

class SelectionConfig(BaseModel):
    """Tuning for stage A selection."""

    model_config = ConfigDict(extra="forbid")

    max_tokens: int | None = Field(
        default=4000, ge=1, description="Token budget of the slice."
    )
    max_elements: int | None = Field(
        default=60, ge=1, description="Vertex and edge budget of the slice."
    )
    lane_depth: int = Field(
        default=20, ge=1, description="Hits kept per lane before fusion."
    )
    max_seeds: int = Field(default=12, ge=1, description="Seeds passed to closure.")
    rank_constant: float = Field(
        default=0.0,
        ge=0.0,
        description=(
            "Added to each rank before the reciprocal. Raising it lets "
            "agreement across lanes outweigh position within one."
        ),
    )
    min_relative_score: float = Field(
        default=0.2,
        gt=0.0,
        le=1.0,
        description=(
            "A seed must reach this fraction of the score a rank-1 hit in one "
            "lane earns. Relative, so it means the same for any schema."
        ),
    )
    lane_weights: dict[Lane, float] = Field(
        default_factory=lambda: dict.fromkeys(LANES, 1.0),
        description="Weight of each lane in fusion.",
    )

    @property
    def budget(self) -> Budget:
        return Budget(max_tokens=self.max_tokens, max_elements=self.max_elements)

Attributes

budget property

lane_depth = Field(default=20, ge=1, description='Hits kept per lane before fusion.') class-attribute instance-attribute

lane_weights = Field(default_factory=lambda: dict.fromkeys(LANES, 1.0), description='Weight of each lane in fusion.') class-attribute instance-attribute

max_elements = Field(default=60, ge=1, description='Vertex and edge budget of the slice.') class-attribute instance-attribute

max_seeds = Field(default=12, ge=1, description='Seeds passed to closure.') class-attribute instance-attribute

max_tokens = Field(default=4000, ge=1, description='Token budget of the slice.') class-attribute instance-attribute

min_relative_score = Field(default=0.2, gt=0.0, le=1.0, description='A seed must reach this fraction of the score a rank-1 hit in one lane earns. Relative, so it means the same for any schema.') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

rank_constant = Field(default=0.0, ge=0.0, description='Added to each rank before the reciprocal. Raising it lets agreement across lanes outweigh position within one.') class-attribute instance-attribute

SelectionError

Bases: RuntimeError

No slice could be chosen.

Attributes:

Name	Type	Description
reason		empty_text or no_match (the schema is over budget and nothing in the text matched it).
metrics		Retrieval metrics, when retrieval ran.

Source code in lpgcast/select.py

class SelectionError(RuntimeError):
    """No slice could be chosen.

    Attributes:
        reason: ``empty_text`` or ``no_match`` (the schema is over budget and
            nothing in the text matched it).
        metrics: Retrieval metrics, when retrieval ran.
    """

    def __init__(
        self,
        message: str,
        *,
        reason: Literal["empty_text", "no_match"],
        metrics: RetrievalMetrics | None = None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.metrics = metrics

Attributes

metrics = metrics instance-attribute

reason = reason instance-attribute

Methods:

__init__(message, *, reason, metrics=None)

Source code in lpgcast/select.py

def __init__(
    self,
    message: str,
    *,
    reason: Literal["empty_text", "no_match"],
    metrics: RetrievalMetrics | None = None,
) -> None:
    super().__init__(message)
    self.reason = reason
    self.metrics = metrics

Functions:

number_adjacent_tokens(text)

Tokens that directly follow a number: candidate unit mentions.

Source code in lpgcast/select.py

def number_adjacent_tokens(text: str) -> set[str]:
    """Tokens that directly follow a number: candidate unit mentions."""
    return {
        match.group("token")
        for match in _NUMBER_UNIT.finditer(text)
        if match.group("token").lower() not in _STOPWORDS
    }

retrieve(schema, text, config=None)

Rank the schema's elements against text; return gated, capped seeds.

Source code in lpgcast/select.py

def retrieve(
    schema: Schema, text: str, config: SelectionConfig | None = None
) -> tuple[list[SeedHit], RetrievalMetrics]:
    """Rank the schema's elements against *text*; return gated, capped seeds."""
    config = config or SelectionConfig()
    rows = list(iter_elements(schema))
    lanes: dict[Lane, list[Seed]] = {
        "exact": _rank(_exact_lane(rows, text), config.lane_depth),
        "lexical": _rank(
            _lexical_lane(rows, text, _descriptions(schema)), config.lane_depth
        ),
        "unit": _rank(_unit_lane(rows, text), config.lane_depth),
    }

    fused: dict[Seed, float] = defaultdict(float)
    ranks: dict[Seed, dict[Lane, int]] = defaultdict(dict)
    for lane, ranked in lanes.items():
        weight = config.lane_weights.get(lane, 0.0)
        for rank, seed in enumerate(ranked, start=1):
            fused[seed] += weight / (config.rank_constant + rank)
            ranks[seed][lane] = rank

    top_single = max(config.lane_weights.values(), default=1.0) / (
        config.rank_constant + 1
    )
    gate = config.min_relative_score * top_single
    ordered = sorted(fused, key=lambda seed: (-fused[seed], _seed_sort_key(seed)))
    passed = [seed for seed in ordered if fused[seed] >= gate]
    kept = passed[: config.max_seeds]

    hits = [
        SeedHit(
            vertex=seed if isinstance(seed, str) else None,
            edge_id=None if isinstance(seed, str) else seed,
            score=fused[seed],
            ranks=ranks[seed],
        )
        for seed in kept
    ]
    metrics = RetrievalMetrics(
        lane_hits={lane: len(ranked) for lane, ranked in lanes.items()},
        fused=len(fused),
        gated_out=len(ordered) - len(passed),
        capped=len(passed) - len(kept),
        seeds=len(kept),
        schema_tokens=estimate_tokens(schema.to_minimal_canonical_dict()),
    )
    return hits, metrics

select(schema, text, config=None)

Choose the schema a cast of text runs against.

Raises:

Type	Description
SelectionError	the text is empty, or the schema is over budget and nothing in the text matched it.

Source code in lpgcast/select.py

def select(
    schema: Schema, text: str, config: SelectionConfig | None = None
) -> Selection:
    """Choose the schema a cast of *text* runs against.

    Raises:
        SelectionError: the text is empty, or the schema is over budget and
            nothing in the text matched it.
    """
    config = config or SelectionConfig()
    if not text.strip():
        raise SelectionError("The text is empty.", reason="empty_text")
    hits, metrics = retrieve(schema, text, config)
    budget = config.budget
    if _fits(schema, metrics, budget):
        return Selection(slice=schema, mode="whole", seeds=hits, metrics=metrics)
    if not hits:
        raise SelectionError(
            "Nothing in the text matched the schema, and the schema is too large "
            f"to pass whole (~{metrics.schema_tokens} tokens).",
            reason="no_match",
            metrics=metrics,
        )
    sliced, elision = subschema(schema, [hit.seed for hit in hits], budget=budget)
    return Selection(
        slice=sliced, mode="slice", seeds=hits, elision=elision, metrics=metrics
    )

words(text)

Case-folded, camel- and snake-split, stemmed words.

Source code in lpgcast/select.py

def words(text: str) -> list[str]:
    """Case-folded, camel- and snake-split, stemmed words."""
    out: list[str] = []
    for token in _WORD.findall(text):
        for part in _CAMEL.split(token):
            if part:
                out.append(_stem(part.lower()))
    return out