lpgcast.resolve

Resolve identities and assemble the graph container of one episode.

Identities are computed before edges are built, because graflo's writer fills digest and minted keys on vertex documents only, never on the endpoint documents inside edges.

Identity mode	Key
natural	the identity properties the text stated
hash / funnel	the digest of its source properties
assigned, blank	a UUID derived from the schema, the episode, the type and the ref

Derived UUIDs make a replay byte-identical; a fresh model call that names its refs differently still mints new vertices for assigned and blank types.

A mention (mention: true) is never written as a vertex. Edges reach it by its primary identity, or — when the text gave only a secondary identity — are held as pending until the write looks the primary identity up.

Attributes

Endpoint = EdgeEndpoint module-attribute

MINT_NAMESPACE = uuid.UUID('6f1c3a52-6a51-4f8e-9e0e-2b8c1d7a4e10') module-attribute

Classes

Resolution

Bases: BaseModel

One episode's graph, ready to write.

Source code in lpgcast/resolve.py

class Resolution(BaseModel):
    """One episode's graph, ready to write."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    container: GraphContainer
    pending: list[PendingEdge] = Field(default_factory=list)
    groundings: list[Grounding] = Field(default_factory=list)

Attributes

container instance-attribute

groundings = Field(default_factory=list) class-attribute instance-attribute

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

pending = Field(default_factory=list) class-attribute instance-attribute

Functions:

merge_containers(containers)

One container holding every vertex and edge of containers, in order.

Source code in lpgcast/resolve.py

def merge_containers(containers: list[GraphContainer]) -> GraphContainer:
    """One container holding every vertex and edge of *containers*, in order."""
    vertices: dict[str, list[Any]] = {}
    edges: dict[EdgeId, list[Any]] = {}
    for container in containers:
        for name, docs in container.vertices.items():
            vertices.setdefault(name, []).extend(docs)
        for edge_id, triples in container.edges.items():
            edges.setdefault(edge_id, []).extend(triples)
    return GraphContainer(vertices=vertices, edges=edges)

mint_id(schema, source_id, vertex_type, ref)

The UUID of an assigned or blank vertex the model introduced as ref.

Source code in lpgcast/resolve.py

def mint_id(schema: Schema, source_id: str, vertex_type: str, ref: str) -> str:
    """The UUID of an assigned or blank vertex the model introduced as *ref*."""
    name = "\x1f".join((schema_ref(schema), source_id, vertex_type, ref))
    return str(uuid.uuid5(MINT_NAMESPACE, name))

resolve(validation, slice_, episode, *, extractor, model=None)

Build the episode's container, pending edges and groundings.

Source code in lpgcast/resolve.py

def resolve(
    validation: Validation,
    slice_: Schema,
    episode: Episode,
    *,
    extractor: str,
    model: str | None = None,
) -> Resolution:
    """Build the episode's container, pending edges and groundings."""
    vertex_config = slice_.core_schema.vertex_config
    ref = schema_ref(slice_)
    vertices: dict[str, list[dict[str, Any]]] = {}
    edges: dict[
        EdgeId, list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]
    ] = {}
    pending: list[PendingEdge] = []
    groundings: list[Grounding] = []
    endpoints: dict[str, Endpoint] = {}

    for valid in validation.vertices:
        vertex = vertex_config[valid.type]
        endpoint = _endpoint(vertex, valid, slice_, episode.source_id)
        endpoints[valid.ref] = endpoint
        if not valid.mention:
            assert endpoint.identity is not None
            vertices.setdefault(valid.type, []).append(
                {**valid.props, **endpoint.identity}
            )
        target = {"vertex": valid.type, "identity": endpoint.identity or endpoint.match}
        groundings.append(
            _grounding(
                episode,
                valid.span,
                extractor=extractor,
                model=model,
                ref=ref,
                kind="vertex",
                **target,
            )
        )
        for name, span in valid.prop_spans.items():
            groundings.append(
                _grounding(
                    episode,
                    span,
                    extractor=extractor,
                    model=model,
                    ref=ref,
                    kind="property",
                    property=name,
                    **target,
                )
            )

    for valid in validation.edges:
        source = endpoints[valid.source_ref]
        target = endpoints[valid.target_ref]
        if source.identity is not None and target.identity is not None:
            edges.setdefault(valid.edge_id, []).append(
                (dict(source.identity), dict(target.identity), dict(valid.props))
            )
        else:
            pending.append(
                PendingEdge(
                    edge_id=valid.edge_id,
                    source=source,
                    target=target,
                    props=valid.props,
                )
            )
        edge_target = {
            "edge": valid.edge_id,
            "source_identity": source.identity or source.match,
            "target_identity": target.identity or target.match,
        }
        groundings.append(
            _grounding(
                episode,
                valid.span,
                extractor=extractor,
                model=model,
                ref=ref,
                kind="edge",
                **edge_target,
            )
        )
        for name, span in valid.prop_spans.items():
            groundings.append(
                _grounding(
                    episode,
                    span,
                    extractor=extractor,
                    model=model,
                    ref=ref,
                    kind="property",
                    property=name,
                    **edge_target,
                )
            )

    return Resolution(
        container=GraphContainer(vertices=vertices, edges=edges),
        pending=pending,
        groundings=groundings,
    )

schema_ref(schema)

<name>@<version> of a schema.

Source code in lpgcast/resolve.py

def schema_ref(schema: Schema) -> str:
    """``<name>@<version>`` of a schema."""
    return f"{schema.metadata.name}@{schema.metadata.version or ''}"