src.asqi.utils¶

Utility functions for ASQI library and test containers.

Functions¶

get_openai_tracking_kwargs(→ Dict[str, Any])

Convert ASQI metadata into kwargs that can be splatted into OpenAI/LiteLLM calls.

Module Contents¶

src.asqi.utils.get_openai_tracking_kwargs(metadata: Dict[str, Any] | asqi.schemas.ExecutionMetadata | None = None) → Dict[str, Any]¶

Convert ASQI metadata into kwargs that can be splatted into OpenAI/LiteLLM calls.

This function is designed to be used in test containers to convert the metadata structure passed from the workflow into OpenAI/LiteLLM client parameters.

Accepts either a dict or ExecutionMetadata Pydantic model for type safety.

Expected ASQI metadata format (from workflow): {

“user_id”: “<optional>”, # Top-level from metadata_config “custom_field”: “<optional>”, # Other top-level from metadata_config “tags”: { # Workflow tracking + metadata_config[“tags”]

“job_id”: “…”, “job_type”: “…”, “parent_id”: “…”, “experiment_id”: “…”, # Example custom tag

}

}

Output (OpenAI/LiteLLM client kwargs): {

“user”: “<user_id>”, # From metadata[“user_id”] “extra_body”: {

“metadata”: {
“tags”: [“k:v”, …], # From metadata[“tags”] “custom_field”: “…” # Other metadata keys

}

}

}

Args:

metadata: Optional metadata dictionary from ASQI workflow

Returns:

Dictionary of kwargs to pass to OpenAI/LiteLLM client methods

Example:

>>> metadata = {
...     "user_id": "user123",
...     "custom_field": "experiment_A",
...     "tags": {"job_id": "test-001", "job_type": "test"}
... }
>>> kwargs = get_openai_tracking_kwargs(metadata)
>>> client.chat.completions.create(
...     model="gpt-4",
...     messages=[...],
...     **kwargs
... )