logging | Kitaru

Structured metadata logging.

kitaru.log() attaches structured key-value metadata to the current checkpoint or execution. It is context-sensitive: inside a checkpoint it attaches to that checkpoint; inside a flow but outside a checkpoint it attaches to the execution.

Example::

from kitaru import checkpoint

@checkpoint def call_model(prompt: str) -> str: response = model.generate(prompt) kitaru.log( tokens=response.usage.total_tokens, cost=response.usage.cost, model=response.model, ) return response.text

func_resolve_log_target() -> tuple[RunMetadataResource, UUID | None]

Resolve the metadata target resource for kitaru.log().

Returns

tuple[zenml.models.v2.misc.run_metadata.RunMetadataResource, uuid.UUID | None]

funclog(**kwargs) -> None

Attach structured metadata to the current checkpoint or execution.

Standard keys include cost, tokens, latency, but arbitrary user-defined keys are accepted.

Notes

Values should be JSON-serializable. Metadata is persisted through ZenML's run-metadata APIs. Multiple calls in the same scope append metadata entries; repeated keys with dictionary values are merged on hydration, while repeated non-dictionary keys resolve to latest value.

paramkwargsAny

= {}

Returns

None