ModelIR API

ModelIR 表达模型 value、tensor type、显式数据流、operation、effect 和 dialect-owned semantic payload；不包含并行 placement、硬件吞吐或预测时间。

Hardware- and distribution-independent semantic model IR.

EMPTY_SEMANTIC module-attribute

EMPTY_SEMANTIC = EmptySemantic()

_MODEL_RESERVED module-attribute

_MODEL_RESERVED = frozenset({'model_spec', 'tp', 'pp', 'dp', 'rank', 'device', 'device_id', 'queue', 'kernel', 'implementation_id', 'start', 'start_time', 'end', 'end_time', 'duration', 'latency', 'memory_address', 'memory_offset'})

__all__ module-attribute

__all__ = ['ModelIR', 'ModelOperation', 'ModelValue', 'ValueRole']

FrozenDict

FrozenDict(source: Mapping[str, V] | None = None, *, items: Iterable[tuple[str, V]] | None = None)

Bases: Mapping[str, V], Generic[V]

A compact, hashable mapping with recursively frozen values.

Source code in src/blueprinting/schema/frozen.py

def __init__(
    self,
    source: Mapping[str, V] | None = None,
    *,
    items: Iterable[tuple[str, V]] | None = None,
) -> None:
    if source is not None and items is not None:
        raise TypeError("provide either source or items, not both")
    raw_items = source.items() if source is not None else (items or ())
    copied: dict[str, V] = {}
    for key, value in raw_items:
        if not isinstance(key, str):
            raise TypeError("FrozenDict keys must be strings")
        if key in copied:
            raise ValueError(f"duplicate FrozenDict key: {key!r}")
        copied[key] = cast(V, freeze(value))
    self._items: tuple[tuple[str, V], ...] = tuple(sorted(copied.items(), key=lambda pair: pair[0]))
    self._hash: int | None = None

__reduce__

__reduce__() -> tuple[type[FrozenDict[V]], tuple[dict[str, V]]]

Use the public constructor for process and UI cache round-trips.

Source code in src/blueprinting/schema/frozen.py

def __reduce__(self) -> tuple[type[FrozenDict[V]], tuple[dict[str, V]]]:
    """Use the public constructor for process and UI cache round-trips."""

    return FrozenDict, (dict(self._items),)

DiagnosticBag

DiagnosticBag()

Bases: DiagnosticBag

Compatibility builder that publishes VerificationReport.

Source code in src/blueprinting/schema/diagnostics.py

def __init__(self) -> None:
    self._items: list[Diagnostic] = []

VerificationReport dataclass

VerificationReport(diagnostics: tuple[Diagnostic, ...] = ())

Bases: DiagnosticSet

Backward-compatible name for the domain-free immutable diagnostics.

Lineage

Typed provenance from source entities to one lowering product.

NodeId dataclass

NodeId(value: str)

Bases: StableId

ValueId dataclass

ValueId(value: str)

Bases: StableId

ModelOperationSemantic

Bases: SemanticPayload

Dialect semantics attached to a ModelIR operation.

CanonicalIRMixin

Behavior shared by immutable canonical IR roots.

to_json

to_json() -> str

Serialize through a self-describing, digest-checked envelope.

Source code in src/blueprinting/synthesizer/stages/common.py

def to_json(self) -> str:
    """Serialize through a self-describing, digest-checked envelope."""

    self.require_valid()
    snapshot = IRSnapshot(
        schema_name=self.header.schema_name,
        schema_version=self.header.schema_version,
        producer_version=self.header.producer_version,
        feature_set=self.header.feature_set,
        content_digest=self.digest,
        payload=self,
    )
    return canonical_dumps(snapshot)

from_json classmethod

from_json(payload: str) -> Checked[IR]

Decode an exact-schema snapshot without exception control flow.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def from_json(cls: type[IR], payload: str) -> Checked[IR]:
    """Decode an exact-schema snapshot without exception control flow."""

    try:
        return Ok(cls._decode_json(payload))
    except IRVerificationError as error:
        return Err(DiagnosticSet(tuple(error.diagnostics)))
    except SerializationError as error:
        return Err(
            DiagnosticSet.of(
                Diagnostic("serialization.snapshot", str(error), ("snapshot",)),
            )
        )

require_from_json classmethod

require_from_json(payload: str) -> IR

Explicit exception adapter for trusted internal/replay boundaries.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def require_from_json(cls: type[IR], payload: str) -> IR:
    """Explicit exception adapter for trusted internal/replay boundaries."""

    return cls.from_json(payload).or_raise(
        lambda diagnostics: SerializationError("; ".join(item.render() for item in diagnostics.errors))
    )

load_migrated classmethod

load_migrated(payload: str, *, registry: Any = None) -> Checked[IR]

Load through an explicit registered migration path.

Source code in src/blueprinting/synthesizer/stages/common.py

@classmethod
def load_migrated(cls: type[IR], payload: str, *, registry: Any = None) -> Checked[IR]:
    """Load through an explicit registered migration path."""

    if registry is None:
        from ..schema_migration import DEFAULT_SCHEMA_MIGRATIONS

        registry = DEFAULT_SCHEMA_MIGRATIONS
    try:
        result = registry.migrate_json(
            payload,
            schema_name=cls.SCHEMA_NAME,
            target_version=cls.SCHEMA_VERSION,
        )
    except SerializationError as error:
        return Err(DiagnosticSet.of(Diagnostic("serialization.migration", str(error), ("snapshot",))))
    return cls.from_json(result.payload)

verify

verify() -> Checked[IR]

Return the immutable snapshot or all expected verifier failures.

Source code in src/blueprinting/synthesizer/stages/common.py

def verify(self: IR) -> Checked[IR]:
    """Return the immutable snapshot or all expected verifier failures."""

    return checked(self, self.diagnostics())

Effect

IRHeader

Version and provenance header embedded in every canonical IR.

OperationName

Structured operation identity; dialect is never inferred from a string.

SchemaVersion

Semantic version of one serialized IR schema.

TensorType

Target-neutral logical tensor type.

ValueRole

Bases: Enum

Semantic ownership role of a model-level SSA value.

ModelValue

One typed model-level SSA value with stable provenance.

ModelOperation

One target-neutral operation with explicit dataflow and effects.

ModelIR

Bases: CanonicalIRMixin

Explicit tensor SSA graph before distribution decisions.

enum

enum(tag: str) -> Callable[[type[T]], type[T]]

Register one closed enumeration through the public authoring surface.

Source code in src/blueprinting/schema/deriving.py

def enum(tag: str) -> Callable[[type[T]], type[T]]:
    """Register one closed enumeration through the public authoring surface."""

    if not isinstance(tag, str) or _WIRE_RE.fullmatch(tag) is None:
        raise TypeError("canonical enum tag must be a dotted lowercase wire identity")
    return enum_type(tag)

record

record(tag: str, *, order: bool = False) -> Callable[[type[T]], type[T]]

Derive an immutable canonical record from an annotated class declaration.

Source code in src/blueprinting/schema/deriving.py

@dataclass_transform(frozen_default=True)
def record(
    tag: str,
    *,
    order: bool = False,
) -> Callable[[type[T]], type[T]]:
    """Derive an immutable canonical record from an annotated class declaration."""

    if not isinstance(tag, str) or _WIRE_RE.fullmatch(tag) is None:
        raise TypeError("canonical record tag must be a dotted lowercase wire identity")

    def decorate(cls: type[T]) -> type[T]:
        if "__dataclass_fields__" in cls.__dict__:
            raise TypeError("@record derives its own frozen dataclass; do not combine it with @dataclass")
        _install_structural_post_init(cls)
        cls = dataclass(frozen=True, slots=True, order=order)(cls)
        return record_type(tag)(cls)

    return decorate

is_known_target_dialect

is_known_target_dialect(operation: OperationName) -> bool

Recognize built-in target dialects forbidden before target binding.

Source code in src/blueprinting/synthesizer/stages/common.py

def is_known_target_dialect(operation: OperationName) -> bool:
    """Recognize built-in target dialects forbidden before target binding."""

    return operation.dialect.lower() in _KNOWN_TARGET_DIALECTS

make_header

make_header(schema_name: str, schema_version: SchemaVersion, *, parent_digests: Iterable[str] = (), features: Iterable[str] = (), producer_version: str = '0.0.0') -> IRHeader

Source code in src/blueprinting/synthesizer/stages/common.py

def make_header(
    schema_name: str,
    schema_version: SchemaVersion,
    *,
    parent_digests: Iterable[str] = (),
    features: Iterable[str] = (),
    producer_version: str = "0.0.0",
) -> IRHeader:
    return IRHeader(
        schema_name=schema_name,
        schema_version=schema_version,
        producer_version=producer_version,
        feature_set=REQUIRED_IR_FEATURES | frozenset(features),
        parent_digests=tuple(parent_digests),
    )

reject_reserved_attributes

reject_reserved_attributes(bag: DiagnosticBag, attributes: FrozenDict, reserved: frozenset[str], *path: str) -> None

Reject semantic fields smuggled through extension dictionaries.

Source code in src/blueprinting/synthesizer/stages/common.py

def reject_reserved_attributes(
    bag: DiagnosticBag,
    attributes: FrozenDict,
    reserved: frozenset[str],
    *path: str,
) -> None:
    """Reject semantic fields smuggled through extension dictionaries."""

    def walk(value: Any, location: tuple[str, ...]) -> None:
        if isinstance(value, FrozenDict):
            for key, item in value.items():
                if key.lower() in reserved:
                    bag.error(
                        "attribute.reserved",
                        f"{key!r} is a semantic field and is illegal in this IR layer",
                        *(location + (key,)),
                    )
                walk(item, location + (key,))
        elif isinstance(value, (tuple, frozenset)):
            values = value if isinstance(value, tuple) else tuple(sorted(value, key=repr))
            for index, item in enumerate(values):
                walk(item, location + (str(index),))

    walk(attributes, tuple(path))

verify_known_references

verify_known_references(bag: DiagnosticBag, references: Iterable[StableId], known: Iterable[StableId], *path: str) -> None

Source code in src/blueprinting/synthesizer/stages/common.py

def verify_known_references(
    bag: DiagnosticBag,
    references: Iterable[StableId],
    known: Iterable[StableId],
    *path: str,
) -> None:
    known_set = set(known)
    for reference in references:
        if reference not in known_set:
            bag.error("reference.unknown", f"unknown reference {reference}", *path)

verify_ordered_dag

verify_ordered_dag(bag: DiagnosticBag, entities: Sequence[Entity], id_of: Callable[[Entity], StableId], dependencies_of: Callable[[Entity], Iterable[StableId]], path: str) -> None

Verify references and require canonical topological sequence order.

Source code in src/blueprinting/synthesizer/stages/common.py

def verify_ordered_dag(
    bag: DiagnosticBag,
    entities: Sequence[Entity],
    id_of: Callable[[Entity], StableId],
    dependencies_of: Callable[[Entity], Iterable[StableId]],
    path: str,
) -> None:
    """Verify references and require canonical topological sequence order."""

    identifiers = {id_of(entity) for entity in entities}
    seen = set()
    for index, entity in enumerate(entities):
        identifier = id_of(entity)
        dependencies = tuple(dependencies_of(entity))
        if len(set(dependencies)) != len(dependencies):
            bag.error("dag.duplicate_dependency", "dependencies must be unique", path, str(index), "dependencies")
        if identifier in dependencies:
            bag.error("dag.self_dependency", f"{identifier} depends on itself", path, str(index), "dependencies")
        for dependency in dependencies:
            if dependency not in identifiers:
                bag.error(
                    "reference.unknown",
                    f"dependency {dependency} is not declared",
                    path,
                    str(index),
                    "dependencies",
                )
            elif dependency not in seen:
                bag.error(
                    "dag.not_topological",
                    f"dependency {dependency} must precede {identifier}",
                    path,
                    str(index),
                    "dependencies",
                    hint="store canonical DAG nodes in topological order",
                )
        seen.add(identifier)

verify_unique_ids

verify_unique_ids(bag: DiagnosticBag, entities: Sequence[Entity], id_of: Callable[[Entity], StableId], path: str) -> None

Source code in src/blueprinting/synthesizer/stages/common.py

def verify_unique_ids(
    bag: DiagnosticBag,
    entities: Sequence[Entity],
    id_of: Callable[[Entity], StableId],
    path: str,
) -> None:
    seen = set()
    for index, entity in enumerate(entities):
        identifier = id_of(entity)
        if identifier in seen:
            bag.error("id.duplicate", f"duplicate identifier {identifier}", path, str(index), "id")
        seen.add(identifier)