Artifact and Serialization API Reference

ArtifactIdentity

async_batch_llm.ArtifactIdentity dataclass

ArtifactIdentity(provider: str | None = None, model: str | None = None, prompt_version: str | None = None, parser_version: str | None = None, application_version: str | None = None, extra: Mapping[str, JSONValue] = dict())

Caller-supplied provenance used to decide whether replay is compatible.

ResumePolicy

async_batch_llm.ResumePolicy

Bases: str, Enum

Which compatible terminal artifact records may bypass provider work.

ArtifactStore

async_batch_llm.ArtifactStore

Bases: Protocol

Provider-neutral asynchronous checkpoint/replay store.

append async

append(work_item: LLMWorkItem[Any, Any, Any], prepared_item: Any, result: WorkItemResult[Any, Any]) -> None

Durably append one newly executed terminal result.

Source code in src/async_batch_llm/artifacts.py

async def append(
    self,
    work_item: LLMWorkItem[Any, Any, Any],
    prepared_item: Any,
    result: WorkItemResult[Any, Any],
) -> None:
    """Durably append one newly executed terminal result."""

close async

close() -> None

Flush and close the store; repeated calls are safe.

Source code in src/async_batch_llm/artifacts.py

async def close(self) -> None:
    """Flush and close the store; repeated calls are safe."""

iter_results

iter_results(*, successes_only: bool = False) -> AsyncIterator[WorkItemResult]

Iterate stored results without starting a processor.

Source code in src/async_batch_llm/artifacts.py

def iter_results(self, *, successes_only: bool = False) -> AsyncIterator[WorkItemResult]:
    """Iterate stored results without starting a processor."""

lookup async

lookup(work_item: LLMWorkItem[Any, Any, Any], prepared_item: Any, policy: ResumePolicy) -> WorkItemResult[Any, Any] | None

Return the newest compatible reusable result, if any.

Source code in src/async_batch_llm/artifacts.py

async def lookup(
    self,
    work_item: LLMWorkItem[Any, Any, Any],
    prepared_item: Any,
    policy: ResumePolicy,
) -> WorkItemResult[Any, Any] | None:
    """Return the newest compatible reusable result, if any."""

prepare_item async

prepare_item(work_item: LLMWorkItem[Any, Any, Any]) -> Any

Prepare the run and validate/fingerprint an item before execution.

Source code in src/async_batch_llm/artifacts.py

async def prepare_item(self, work_item: LLMWorkItem[Any, Any, Any]) -> Any:
    """Prepare the run and validate/fingerprint an item before execution."""

JsonlArtifactStore

async_batch_llm.JsonlArtifactStore

JsonlArtifactStore(path: str | Path, *, identity: ArtifactIdentity | None = None, user_metadata: Mapping[str, JSONValue] | None = None, include_output: bool = True, include_metadata: bool = True, include_prompt: bool = False, include_context: bool = False, context_in_identity: bool = True, encoder: ValueEncoder | None = None, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None, context_fingerprinter: ContextFingerprinter | None = None, cost_calculator: CostCalculator | None = None, fsync: bool = False)

Append-only version-1 JSONL artifact store.

Prompt and context text are excluded by default; their SHA-256 hashes are always recorded for compatibility. Output and metadata are included by default because successful replay requires output. Set include_output false for audit-only artifacts; those successful records are not replayable.

Source code in src/async_batch_llm/artifacts.py

def __init__(
    self,
    path: str | Path,
    *,
    identity: ArtifactIdentity | None = None,
    user_metadata: Mapping[str, JSONValue] | None = None,
    include_output: bool = True,
    include_metadata: bool = True,
    include_prompt: bool = False,
    include_context: bool = False,
    context_in_identity: bool = True,
    encoder: ValueEncoder | None = None,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
    context_fingerprinter: ContextFingerprinter | None = None,
    cost_calculator: CostCalculator | None = None,
    fsync: bool = False,
) -> None:
    self.path = Path(path)
    self.identity = identity
    self.user_metadata = user_metadata or {}
    self.include_output = include_output
    self.include_metadata = include_metadata
    self.include_prompt = include_prompt
    self.include_context = include_context
    self.context_in_identity = context_in_identity
    self.encoder = encoder
    self.output_decoder = output_decoder
    self.context_decoder = context_decoder
    self.context_fingerprinter = context_fingerprinter
    self.cost_calculator = cost_calculator
    self.fsync = fsync

    # With no explicit identity, resolution is deferred to the first
    # prepare_item() call, which infers provider/model from the item's
    # strategy. Until then the
    # fingerprint is unset and lookup/append refuse to run.
    self._identity_value: dict[str, JSONValue] | None = None
    self.identity_fingerprint: str | None = None
    if identity is not None:
        self._identity_value = _identity_mapping(identity)
        self.identity_fingerprint = _sha256(_identity_mapping(identity, for_fingerprint=True))
    try:
        metadata_value = to_json_value(self.user_metadata, path="$.user_metadata")
    except ResultSerializationError as exc:
        raise ArtifactSerializationError(str(exc)) from exc
    if not isinstance(metadata_value, dict):
        raise ArtifactSerializationError("user_metadata must serialize to an object")
    self._user_metadata_value = metadata_value

    self._lock = asyncio.Lock()
    self._prepared = False
    self._closed = False
    self._handle: TextIO | None = None
    self._records: list[dict[str, Any]] = []
    self._latest_replayable: dict[_ReplayKey, dict[str, Any]] = {}
    self._latest_success: dict[_ReplayKey, dict[str, Any]] = {}
    self._next_sequence = 0
    self._detached_io_tasks: set[asyncio.Task[Any]] = set()
    self._detached_io_errors: list[Exception] = []

prepare_item async

prepare_item(work_item: LLMWorkItem[Any, Any, Any]) -> _ItemFingerprint

Create/validate the artifact and fingerprint input before provider work.

Source code in src/async_batch_llm/artifacts.py

async def prepare_item(self, work_item: LLMWorkItem[Any, Any, Any]) -> _ItemFingerprint:
    """Create/validate the artifact and fingerprint input before provider work."""
    self._resolve_identity_from(work_item.strategy)
    await self._prepare()
    return self._fingerprint_item(work_item)

read_results classmethod

read_results(path: str | Path, *, successes_only: bool = False, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> BatchResult[Any, Any]

Read stored results without opening a writer or calling a provider.

Source code in src/async_batch_llm/artifacts.py

@classmethod
def read_results(
    cls,
    path: str | Path,
    *,
    successes_only: bool = False,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
) -> BatchResult[Any, Any]:
    """Read stored results without opening a writer or calling a provider."""
    records, _ = _read_artifact_records(Path(path), allow_create=False)
    results: list[WorkItemResult[Any, Any]] = []
    for record in records:
        if successes_only and not record.get("success"):
            continue
        try:
            results.append(
                work_item_result_from_dict(
                    record["result"],
                    output_decoder=output_decoder,
                    context_decoder=context_decoder,
                )
            )
        except (KeyError, ResultSerializationError) as exc:
            raise ArtifactFormatError(f"Malformed stored result: {exc}") from exc
    return BatchResult(results=results, termination=BatchTermination())

ResultSerializationError

async_batch_llm.ResultSerializationError

Bases: ValueError

A result value or serialized payload is not safely supported.

Artifact errors

async_batch_llm.ArtifactError

Bases: RuntimeError

Base class for artifact preparation, format, and persistence failures.

async_batch_llm.ArtifactSerializationError

Bases: ArtifactError

An artifact identity/input/result could not be canonically serialized.

async_batch_llm.ArtifactIOError

Bases: ArtifactError

An artifact could not be read, written, flushed, or closed.

async_batch_llm.ArtifactFormatError

Bases: ArtifactError

An artifact is malformed or uses an unsupported schema version.