Serialization
Deterministic, version-tagged JSON for the checkpoint-safe models (AgentState, ExecutionPlan,
ModelCard, RunResult). AgentGraph is deliberately not supported — its nodes carry live,
non-serialisable compute callables.
to_json
to_json(model: BaseModel) -> str
Serialise model to a deterministic, version-tagged JSON envelope (spec 08 §6).
Keys are sorted (recursively, at every nesting level), separators are fixed (",",
":"), and the output is UTF-8 with no ASCII-escaping — the same model always produces
byte-identical output. Timestamps come out ISO-8601 with an explicit UTC offset and
microsecond precision (pydantic's own JSON mode); floats keep their repr round-trip
fidelity.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
BaseModel
|
An instance of a supported model — currently :class: |
required |
Returns:
| Type | Description |
|---|---|
str
|
The JSON envelope: |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If |
Example
from datetime import datetime, timezone from korchestrator.models.state import AgentState from korchestrator.serializers import to_json state = AgentState( ... run_id="r1", objective="summarize the quarterly report", ... transaction_time=datetime(2026, 7, 23, tzinfo=timezone.utc), ... ) payload = to_json(state) to_json(state) == payload # deterministic: same object, same bytes, every time True
from_json
from_json(payload: str, model_cls: type[T]) -> T
Deserialise payload into model_cls, applying migrations as needed (spec 08 §6.5).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payload
|
str
|
A JSON envelope produced by :func: |
required |
model_cls
|
type[T]
|
The expected model type. |
required |
Returns:
| Type | Description |
|---|---|
T
|
The reconstructed, validated instance. |
Raises:
| Type | Description |
|---|---|
ValidationError
|
If |
Example
from datetime import datetime, timezone from korchestrator.models.state import AgentState from korchestrator.serializers import from_json, to_json state = AgentState( ... run_id="r1", objective="summarize the quarterly report", ... transaction_time=datetime(2026, 7, 23, tzinfo=timezone.utc), ... ) from_json(to_json(state), AgentState) == state True