graflo.architecture.pipeline.runtime.assemble

Assembly phase for turning extracted observations into graph edges.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

assemble_edges(*, ctx, vertex_config, edge_config, infer_edges, infer_edge_only=None, infer_edge_except=None, target_db_flavor=None, edge_derivation=None, inverse_pairs=None)

Assemble all edge documents after extraction finishes.

Parameters:

Name	Type	Description	Default
inverse_pairs	Mapping[str, str] | None	{relation: declared inverse} for paired relations, symmetric ones excluded. Computed from edge_config when omitted; a caller assembling many documents passes it once.	None

Source code in graflo/architecture/pipeline/runtime/assemble.py

def assemble_edges(
    *,
    ctx: AssemblyContext,
    vertex_config: VertexConfig,
    edge_config: EdgeConfig,
    infer_edges: bool,
    infer_edge_only: set[EdgeId] | None = None,
    infer_edge_except: set[EdgeId] | None = None,
    target_db_flavor: DBType | None = None,
    edge_derivation: EdgeDerivationRegistry | None = None,
    inverse_pairs: Mapping[str, str] | None = None,
) -> None:
    """Assemble all edge documents after extraction finishes.

    Args:
        inverse_pairs: ``{relation: declared inverse}`` for paired relations,
            symmetric ones excluded. Computed from ``edge_config`` when omitted;
            a caller assembling many documents passes it once.
    """
    if infer_edge_only is None:
        infer_edge_only = set()
    if infer_edge_except is None:
        infer_edge_except = set()
    if inverse_pairs is None and any(
        intent.derivation is not None and intent.derivation.emit_inverse
        for intent in ctx.edge_intents
    ):
        inverse_pairs = inverse_map(edge_config.inverses)

    # Pairs an explicit edge actor already produced -- mirrored ones included.
    # Inference skips them so it does not duplicate authored edges — the same
    # intent as the documented per-resource auto-exclusion, applied at
    # (source, target) granularity.
    explicit_pairs: set[tuple[str, str]] = set()

    for intent in ctx.edge_intents:
        edge = intent.edge
        relation_input = _resolved_relation_input_field(
            edge,
            derivation=intent.derivation,
            target_db_flavor=target_db_flavor,
        )
        explicit_pairs |= _emit_edge_documents(
            ctx=ctx,
            vertex_config=vertex_config,
            edge=edge,
            lindex=intent.location,
            relation_input_field=relation_input,
            derivation=intent.derivation,
            edge_derivation=edge_derivation,
            edge_config=edge_config,
            inverse_pairs=inverse_pairs,
        )
    ctx.extraction.edge_intents = []

    if not infer_edges:
        return

    populated = {v for v, dd in ctx.acc_vertex.items() if any(dd.values())}
    for edge_id, edge in edge_config.items():
        s, t, _ = edge_id
        if (s, t) in explicit_pairs or s not in populated or t not in populated:
            continue
        if not _is_inference_allowed(
            edge_id,
            infer_edge_only=infer_edge_only,
            infer_edge_except=infer_edge_except,
        ):
            continue
        relation_input = _resolved_relation_input_field(
            edge,
            derivation=None,
            target_db_flavor=target_db_flavor,
        )
        # Deliberately does not record (s, t): `edge_config.items()` yields each
        # edge_id once, so recording the pair could only ever suppress a *different*
        # relation between the same two vertex types. Two declared relations on one
        # pair are both inferable, and dropping one of them by iteration order was
        # silent edge loss.
        _emit_edge_documents(
            ctx=ctx,
            vertex_config=vertex_config,
            edge=edge,
            lindex=None,
            relation_input_field=relation_input,
            derivation=None,
            edge_derivation=edge_derivation,
        )

mirrored_edge_id(edge_id, *, inverse_pairs, edge_config)

The declared inverse edge of edge_id, or None when there is nothing to write.

(s, t, a) mirrors to (t, s, b) when a has a declared inverse b and that inverse edge (or a relation-less template between the same types) is declared. The second condition is what makes a pair materialized: a pair that is only declared, or that the database maintains, has no edge to write into, and an undeclared edge would be dropped at the writer without a word.

Source code in graflo/architecture/pipeline/runtime/assemble.py

def mirrored_edge_id(
    edge_id: EdgeId,
    *,
    inverse_pairs: Mapping[str, str],
    edge_config: EdgeConfig,
) -> EdgeId | None:
    """The declared inverse edge of ``edge_id``, or None when there is nothing to write.

    ``(s, t, a)`` mirrors to ``(t, s, b)`` when ``a`` has a declared inverse ``b``
    **and** that inverse edge (or a relation-less template between the same
    types) is declared. The second condition is what makes a pair
    *materialized*: a pair that is only declared, or that the database
    maintains, has no edge to write into, and an undeclared edge would be
    dropped at the writer without a word.
    """
    source, target, relation = edge_id
    if relation is None:
        return None
    inverse = inverse_pairs.get(relation)
    if inverse is None or inverse == relation:
        return None
    if (target, source, inverse) in edge_config or (
        target,
        source,
        None,
    ) in edge_config:
        return target, source, inverse
    return None