ModelIR API
ModelIR records model values, tensor types, explicit dataflow, operations,
effects, and dialect-owned semantic payloads. It excludes parallel placement,
hardware throughput, and predicted time.
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
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.
ModelOperationSemantic
Bases: SemanticPayload
Dialect semantics attached to a ModelIR operation.
CanonicalIRMixin
Behavior shared by immutable canonical IR roots.
to_json
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
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())
|
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(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)
|