artifacts
Artifact helpers for explicit named artifacts.
kitaru.save() persists a named artifact inside a checkpoint.
kitaru.load() retrieves a named artifact from a previous execution.
Both are valid only inside a checkpoint.
Example::
from kitaru import checkpoint
@checkpoint def research(topic: str) -> str: context = gather_sources(topic) kitaru.save("sources", context, type="context") return summarize(context)
In a later execution:
@checkpoint def refine(exec_id: str) -> str: old_sources = kitaru.load(exec_id, "sources") return improve(old_sources)
func_require_checkpoint_scope(api_name) -> tuple[UUID, UUID]Validate that an API call happens inside a checkpoint scope.
paramapi_namestrCalling Kitaru API name.
Returns
tupleTuple of current execution UUID and checkpoint UUID.
func_normalize_artifact_type(artifact_type) -> strValidate and normalize Kitaru artifact type labels.
paramartifact_typestrArtifact type supplied by the user.
Returns
strNormalized artifact type.
func_normalize_step_name(step_name) -> strNormalize ZenML step names back to user-facing checkpoint names.
paramstep_namestrReturns
strfunc_matches_requested_name(*, step_name, artifact, requested_name) -> boolCheck whether an artifact matches a Kitaru load() name lookup.
paramstep_namestrparamartifactArtifactVersionResponseparamrequested_namestrReturns
boolfunc_collect_named_artifact_matches(*, run, requested_name) -> list[_ArtifactMatch]Collect candidate artifacts in a run for a given Kitaru lookup name.
paramrunPipelineRunResponseparamrequested_namestrReturns
list[kitaru.artifacts._ArtifactMatch]func_format_match(match) -> strFormat a match for ambiguity error messages.
parammatch_ArtifactMatchReturns
strfuncsave(name, value, *, type='output', tags=None) -> NonePersist a named artifact inside the current checkpoint.
paramnamestrArtifact name (unique within the checkpoint).
paramvalueAnyThe value to persist. Must be serializable.
paramtypestr= 'output'Artifact type for categorization (one of "prompt",
"response", "context", "input", "output",
"blob").
paramtagslist[str] | None= NoneOptional tags for filtering and discovery.
Returns
Nonefuncload(exec_id, name) -> AnyLoad a named artifact from a previous execution.
paramexec_idstrThe execution ID to load from.
paramnamestrThe artifact name to retrieve.
Returns
typing.AnyThe materialized artifact value.