aiq.profiler.decorators.function_tracking#

Functions#

_serialize_data(→ Any)

Convert obj into a structure that can be passed to json.dumps(...).

_prepare_serialized_args_kwargs(→ tuple[list[Any], ...)

Serialize args and kwargs before calling the wrapped function.

push_intermediate_step(→ None)

Push an intermediate step to the AIQ Toolkit Event Stream.

track_function([func, metadata])

Decorator that can wrap any type of function (sync, async, generator,

Module Contents#

_serialize_data(obj: Any) → Any#

Convert obj into a structure that can be passed to json.dumps(...).

_prepare_serialized_args_kwargs(
*args,
**kwargs,
) → tuple[list[Any], dict[str, Any]]#

Serialize args and kwargs before calling the wrapped function.

push_intermediate_step(
step_manager: aiq.builder.intermediate_step_manager.IntermediateStepManager,
identifier: str,
function_name: str,
event_type: aiq.data_models.intermediate_step.IntermediateStepType,
args: Any = None,
kwargs: Any = None,
output: Any = None,
metadata: dict[str, Any] | None = None,
) → None#

Push an intermediate step to the AIQ Toolkit Event Stream.

track_function(
func: Any = None,
*,
metadata: dict[str, Any] | None = None,
)#

Decorator that can wrap any type of function (sync, async, generator, async generator) and executes “tracking logic” around it.

  • If the function is async, it will be wrapped in an async function.

  • If the function is a generator, it will be wrapped in a generator function.

  • If the function is an async generator, it will be wrapped in an async generator function.

  • If the function is sync, it will be wrapped in a sync function.