def _get_embeddings_attributes_from_wrapper()

in instrumentation/elastic-opentelemetry-instrumentation-openai/src/opentelemetry/instrumentation/openai/helpers.py [0:0]

• 20 lines of code
• 7 McCabe index (conditional complexity)

def _get_embeddings_attributes_from_wrapper(instance, kwargs) -> Attributes:
    # we import this here to avoid races with other instrumentations
    try:
        # available since 1.13.4
        from openai import NotGiven
    except ImportError:
        NotGiven = None

    def _is_set(value):
        if NotGiven is not None:
            return value is not None and not isinstance(value, NotGiven)
        return value is not None

    span_attributes = {
        GEN_AI_OPERATION_NAME: "embeddings",
        GEN_AI_SYSTEM: "openai",
    }

    if _is_set(request_model := kwargs.get("model")):
        span_attributes[GEN_AI_REQUEST_MODEL] = request_model

    if client := getattr(instance, "_client", None):
        span_attributes.update(_attributes_from_client(client))

    if _is_set(encoding_format := kwargs.get("encoding_format")):
        span_attributes[GEN_AI_REQUEST_ENCODING_FORMATS] = [encoding_format]

    return span_attributes