logging | Kitaru Reference

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

funclog_to_execution(run_id, *, _client=None, **kwargs) -> None

Attach structured metadata to a specific execution.

This helper is for SDK observation paths such as FlowHandle.wait() that need to write pipeline-run metadata after user code has finished and there is no active flow context anymore.

paramrun_idstr

param_clientAny | None

= None

paramkwargsAny

= {}

Returns

None

funclog_to_checkpoint(step_id, *, _client=None, **kwargs) -> None

Attach structured metadata to a specific checkpoint step run.

This helper is for code paths that need to publish metadata by checkpoint step ID when there is no active checkpoint context.

paramstep_idstr

param_clientAny | None

= None

paramkwargsAny

= {}

Returns

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