Batching and Streaming

Batch Encoding

Use encode_batch when all states are available in memory.

This page fragment is generated from docs/snippets/hgraph_batch.py. It is intended to show the exact code and typical output for this version.

Commit: 390f8090430944521df7c322996af8aa89f78469

Program

from __future__ import annotations

import mifrost

from ._helpers import first_successor, load_problem, print_encoding_summary


def main() -> None:
    space, domain, _problem, state = load_problem("blocks", "smedium")
    _action, succ = first_successor(space, state)
    encoder = mifrost.HGraphEncoder(domain)
    encoding = encoder.encode_batch([state, succ])
    print_encoding_summary(
        encoding, label="HGraphEncoder.encode_batch([state, successor])"
    )


if __name__ == "__main__":
    main()

Output

== HGraphEncoder.encode_batch([state, successor]) ==
python: 3.12.13
platform: Linux-6.17.0-1018-azure-x86_64-with-glibc2.39
graph_kind: hetero
num_graphs: 2
num_nodes: 24
num_edges: 50
schema_fingerprint: 2869292577121628696
node_types_count: 28
edge_types_count: 54
node_types_head: ['[+]clear[g]', '[+]clear[g][sat]', '[+]handempty[g]', '[+]handempty[g][sat]', '[+]holding[g]', '[+]holding[g][sat]', '[+]on[g]', '[+]on[g][sat]', '[+]ontable[g]', '[+]ontable[g][sat]']
edge_types_head: ['[+]clear[g][sat]|0|_symbol_', '[+]clear[g]|0|_symbol_', '[+]holding[g][sat]|0|_symbol_', '[+]holding[g]|0|_symbol_', '[+]on[g][sat]|0|_symbol_', '[+]on[g][sat]|1|_symbol_', '[+]on[g]|0|_symbol_', '[+]on[g]|1|_symbol_', '[+]ontable[g][sat]|0|_symbol_', '[+]ontable[g]|0|_symbol_']
SNAPSHOT_JSON_BEGIN
{
  "edge_types_count": 54,
  "edge_types_head": [
    "[+]clear[g][sat]|0|_symbol_",
    "[+]clear[g]|0|_symbol_",
    "[+]holding[g][sat]|0|_symbol_",
    "[+]holding[g]|0|_symbol_",
    "[+]on[g][sat]|0|_symbol_",
    "[+]on[g][sat]|1|_symbol_",
    "[+]on[g]|0|_symbol_",
    "[+]on[g]|1|_symbol_",
    "[+]ontable[g][sat]|0|_symbol_",
    "[+]ontable[g]|0|_symbol_"
  ],
  "graph_kind": "hetero",
  "label": "HGraphEncoder.encode_batch([state, successor])",
  "node_types_count": 28,
  "node_types_head": [
    "[+]clear[g]",
    "[+]clear[g][sat]",
    "[+]handempty[g]",
    "[+]handempty[g][sat]",
    "[+]holding[g]",
    "[+]holding[g][sat]",
    "[+]on[g]",
    "[+]on[g][sat]",
    "[+]ontable[g]",
    "[+]ontable[g][sat]"
  ],
  "num_edges": 50,
  "num_graphs": 2,
  "num_nodes": 24,
  "platform": "Linux-6.17.0-1018-azure-x86_64-with-glibc2.39",
  "python": "3.12.13",
  "schema_fingerprint": 2869292577121628696
}
SNAPSHOT_JSON_END

Append-Only Stream

Use stream() when samples arrive incrementally and only append semantics are needed.

This page fragment is generated from docs/snippets/hgraph_stream_append_only.py. It is intended to show the exact code and typical output for this version.

Commit: 390f8090430944521df7c322996af8aa89f78469

Program

from __future__ import annotations

import mifrost

from ._helpers import first_successor, load_problem, print_encoding_summary


def main() -> None:
    space, domain, _problem, state = load_problem("blocks", "smedium")
    _action, succ = first_successor(space, state)

    encoder = mifrost.HGraphEncoder(domain)
    stream = encoder.stream()
    stream.append(state)
    stream.append(succ)
    encoding = stream.flush()

    print_encoding_summary(
        encoding, label="HGraphEncoder.stream().append(...).flush() (2 graphs)"
    )


if __name__ == "__main__":
    main()

Output

== HGraphEncoder.stream().append(...).flush() (2 graphs) ==
python: 3.12.13
platform: Linux-6.17.0-1018-azure-x86_64-with-glibc2.39
graph_kind: hetero
num_graphs: 2
num_nodes: 24
num_edges: 50
schema_fingerprint: 2869292577121628696
node_types_count: 28
edge_types_count: 54
node_types_head: ['[+]clear[g]', '[+]clear[g][sat]', '[+]handempty[g]', '[+]handempty[g][sat]', '[+]holding[g]', '[+]holding[g][sat]', '[+]on[g]', '[+]on[g][sat]', '[+]ontable[g]', '[+]ontable[g][sat]']
edge_types_head: ['[+]clear[g][sat]|0|_symbol_', '[+]clear[g]|0|_symbol_', '[+]holding[g][sat]|0|_symbol_', '[+]holding[g]|0|_symbol_', '[+]on[g][sat]|0|_symbol_', '[+]on[g][sat]|1|_symbol_', '[+]on[g]|0|_symbol_', '[+]on[g]|1|_symbol_', '[+]ontable[g][sat]|0|_symbol_', '[+]ontable[g]|0|_symbol_']
SNAPSHOT_JSON_BEGIN
{
  "edge_types_count": 54,
  "edge_types_head": [
    "[+]clear[g][sat]|0|_symbol_",
    "[+]clear[g]|0|_symbol_",
    "[+]holding[g][sat]|0|_symbol_",
    "[+]holding[g]|0|_symbol_",
    "[+]on[g][sat]|0|_symbol_",
    "[+]on[g][sat]|1|_symbol_",
    "[+]on[g]|0|_symbol_",
    "[+]on[g]|1|_symbol_",
    "[+]ontable[g][sat]|0|_symbol_",
    "[+]ontable[g]|0|_symbol_"
  ],
  "graph_kind": "hetero",
  "label": "HGraphEncoder.stream().append(...).flush() (2 graphs)",
  "node_types_count": 28,
  "node_types_head": [
    "[+]clear[g]",
    "[+]clear[g][sat]",
    "[+]handempty[g]",
    "[+]handempty[g][sat]",
    "[+]holding[g]",
    "[+]holding[g][sat]",
    "[+]on[g]",
    "[+]on[g][sat]",
    "[+]ontable[g]",
    "[+]ontable[g][sat]"
  ],
  "num_edges": 50,
  "num_graphs": 2,
  "num_nodes": 24,
  "platform": "Linux-6.17.0-1018-azure-x86_64-with-glibc2.39",
  "python": "3.12.13",
  "schema_fingerprint": 2869292577121628696
}
SNAPSHOT_JSON_END

Flat append-only support:

FlatRelationEncoder.stream()
This stream accepts the same main-lane payloads as FlatRelationEncoder.encode(...): goals, actions, subgoal_layers, history_subgoals, history_max_steps

Mutable Stream

Use mutable_stream() when you need stable IDs and update/remove semantics.

This page fragment is generated from docs/snippets/hgraph_stream_mutable.py. It is intended to show the exact code and typical output for this version.

Commit: 390f8090430944521df7c322996af8aa89f78469

Program

from __future__ import annotations

import mifrost

from ._helpers import first_successor, load_problem, print_encoding_summary


def main() -> None:
    space, domain, _problem, state = load_problem("blocks", "smedium")
    _action, succ = first_successor(space, state)

    encoder = mifrost.HGraphEncoder(domain)
    mstream = encoder.mutable_stream()
    sid0 = mstream.append(state)
    sid1 = mstream.append(succ)
    mstream.update(sid0, succ)
    mstream.remove(sid1)
    encoding = mstream.flush()

    print_encoding_summary(
        encoding,
        label="HGraphEncoder.mutable_stream() append/update/remove/flush (1 graph)",
    )


if __name__ == "__main__":
    main()

Output

== HGraphEncoder.mutable_stream() append/update/remove/flush (1 graph) ==
python: 3.12.13
platform: Linux-6.17.0-1018-azure-x86_64-with-glibc2.39
graph_kind: hetero
num_graphs: 1
num_nodes: 12
num_edges: 24
schema_fingerprint: 2869292577121628696
node_types_count: 28
edge_types_count: 54
node_types_head: ['[+]clear[g]', '[+]clear[g][sat]', '[+]handempty[g]', '[+]handempty[g][sat]', '[+]holding[g]', '[+]holding[g][sat]', '[+]on[g]', '[+]on[g][sat]', '[+]ontable[g]', '[+]ontable[g][sat]']
edge_types_head: ['[+]clear[g][sat]|0|_symbol_', '[+]clear[g]|0|_symbol_', '[+]holding[g][sat]|0|_symbol_', '[+]holding[g]|0|_symbol_', '[+]on[g][sat]|0|_symbol_', '[+]on[g][sat]|1|_symbol_', '[+]on[g]|0|_symbol_', '[+]on[g]|1|_symbol_', '[+]ontable[g][sat]|0|_symbol_', '[+]ontable[g]|0|_symbol_']
SNAPSHOT_JSON_BEGIN
{
  "edge_types_count": 54,
  "edge_types_head": [
    "[+]clear[g][sat]|0|_symbol_",
    "[+]clear[g]|0|_symbol_",
    "[+]holding[g][sat]|0|_symbol_",
    "[+]holding[g]|0|_symbol_",
    "[+]on[g][sat]|0|_symbol_",
    "[+]on[g][sat]|1|_symbol_",
    "[+]on[g]|0|_symbol_",
    "[+]on[g]|1|_symbol_",
    "[+]ontable[g][sat]|0|_symbol_",
    "[+]ontable[g]|0|_symbol_"
  ],
  "graph_kind": "hetero",
  "label": "HGraphEncoder.mutable_stream() append/update/remove/flush (1 graph)",
  "node_types_count": 28,
  "node_types_head": [
    "[+]clear[g]",
    "[+]clear[g][sat]",
    "[+]handempty[g]",
    "[+]handempty[g][sat]",
    "[+]holding[g]",
    "[+]holding[g][sat]",
    "[+]on[g]",
    "[+]on[g][sat]",
    "[+]ontable[g]",
    "[+]ontable[g][sat]"
  ],
  "num_edges": 24,
  "num_graphs": 1,
  "num_nodes": 12,
  "platform": "Linux-6.17.0-1018-azure-x86_64-with-glibc2.39",
  "python": "3.12.13",
  "schema_fingerprint": 2869292577121628696
}
SNAPSHOT_JSON_END

Flat mutable support:

FlatRelationEncoder.mutable_stream()
FlatHorizonEncoder.stream()
FlatTransitionEncoder.stream()
FlatTransitionEffectsEncoder.stream()
Flat transition streams are wrapper-level mutable streams built on top of flat horizon streaming.

Semantics

encode_batch(states, *, ...) treats states as the only batch axis.
For batch-capable kwargs, pass either:
one shared payload (reused for all states), or
a per-state sequence with len(...) == len(states) and optional None entries.
Encoder-specific batch kwargs follow the same pattern:
transition encoders: successors=successor, successors=[succ0, succ1], successors=BatchParam.shared(successor), or successors=BatchParam.separate([succ0, succ1])
horizon encoder: dags=dag, dags=[dag0, None], dags=BatchParam.shared(dag), or dags=BatchParam.separate([dag0, None]) where each DAG leaf may be either TransitionDAG or rustworkx.PyDiGraph
High-level encode_batch(...) performs Python-side wrapper/adapter conversion, then executes C++-backed batch parsing.
Low-level _core._parse_* batch helpers remain strict advanced-only and reject adapter-backed objects.
Example:
shared: encoder.encode_batch(states, goals=goals, actions=actions)
per-state: encoder.encode_batch(states, goals=[g0, g1], actions=[[a0], None])
transition shared successor: encoder.encode_batch(states, successors=BatchParam.shared(succ))
horizon per-entry DAGs: encoder.encode_batch(states, dags=BatchParam.separate([dag0, None]))
rustworkx support is optional and stays on the Python side: mifrost.transition_dag_from_rustworkx(...) performs the same conversion explicitly, and direct PyDiGraph inputs are normalized to TransitionDAG before they reach C++.
Rustworkx candidate metadata import:
node payload candidate_id (mapping key or attribute) maps to TransitionDAG node candidate IDs.
partial candidate IDs fail by default; set fallback_missing_candidate_id_to_node_index=True on explicit conversion to fill gaps.
The interop path assumes the PyDiGraph is already the correct single-root horizon DAG/tree. mifrost does not run a semantic DAG verification pass during conversion.
After conversion, stream and non-stream horizon paths both validate that the DAG root matches the provided root state.
Horizon/HGraph target rows expose aligned identity metadata: target_positions[i], target_indices[i], and target_candidate_ids[i] refer to the same target row.
Append-only stream intentionally does not provide update/remove.
Mutable stream provides ID-based mutation and merges cached entries at flush.
Both stream variants return native BatchEncoding via flush().
Flat flush_pyg() returns FlatRelationData, not generic Data.