diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index 9bec220e073..dc14d3904c0 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -33,9 +33,11 @@ from litellm.integrations.otel.model.baggage import promoted_metadata from litellm.integrations.otel.model.db_endpoint import db_span_attributes from litellm.integrations.otel.model.metadata import flatten_metadata from litellm.integrations.otel.model.semconv import LiteLLM, Metric +from litellm.integrations.otel.model.utils import as_str from litellm.integrations.otel.plumbing.otlp_tls import resolve_otlp_http_tls from litellm.integrations.otel.routing import routed_model_name, routing_decision_attributes from litellm.litellm_core_utils.core_helpers import iter_items, require_attributes +from litellm.litellm_core_utils.hidden_params import get_hidden_params from litellm.litellm_core_utils.internal_call_metadata import is_unbilled_non_inference_call_from_params from litellm.litellm_core_utils.safe_json_dumps import safe_dumps from litellm.litellm_core_utils.secret_redaction import redact_string @@ -140,6 +142,18 @@ def _response_field(response_obj: object, key: str) -> object: return require_attributes(response_obj, _SupportsGet, "get").get(key) +def _semconv_response_model( + response_obj: object, response_model: object, model_group: str | None, deployment_model: str | None +) -> object: + hidden_params: Final = get_hidden_params(response_obj) + provider_model: Final = hidden_params.get("provider_response_model") if hidden_params is not None else None + if isinstance(provider_model, str) and provider_model: + return provider_model + if response_model and response_model == model_group and deployment_model: + return deployment_model + return response_model + + def _billed_usage(response_obj: object, call_type: str | None, litellm_params: Mapping[str, object] | None) -> object: if not response_obj or is_unbilled_non_inference_call_from_params(call_type, litellm_params, response_obj): return None @@ -152,6 +166,7 @@ class _CacheHitView(TypedDict, total=False): _CACHE_HIT_VIEW: Final = TypeAdapter(_CacheHitView) +_OPAQUE_VALUE: Final = TypeAdapter(object) # Dedicated so a slow exporter shutdown cannot starve the shared logging executor. _PROVIDER_SHUTDOWN_EXECUTOR: Final = ThreadPoolExecutor(max_workers=4, thread_name_prefix="OtelProviderShutdown") @@ -2521,11 +2536,12 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): ############################################# # The name of the LLM a request is being made to - if kwargs.get("model"): + request_model: Final = as_str(_OPAQUE_VALUE.validate_python(kwargs.get("model"))) + if request_model: self.safe_set_attribute( span=span, key=SpanAttributes.LLM_REQUEST_MODEL.value, - value=kwargs.get("model"), + value=request_model, ) # The LLM request type @@ -2617,10 +2633,14 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): ) # The model used to generate the response. - response_model: Final = routed_model_name( - _response_field(response_obj, "model"), - standard_logging_payload.get("model_group"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # kwargs is untyped - decision, + model_group: Final = as_str(standard_logging_payload.get("model_group")) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # kwargs is untyped + routed_response_model: Final = routed_model_name( + _response_field(response_obj, "model"), model_group, decision + ) + response_model: Final = ( + _semconv_response_model(response_obj, routed_response_model, model_group, request_model) + if self._gen_ai_semconv_latest_experimental + else routed_response_model ) if response_model: self.safe_set_attribute( diff --git a/tests/unit/integrations/test_opentelemetry.py b/tests/unit/integrations/test_opentelemetry.py index 4ef95d3a79d..6f2424e74d2 100644 --- a/tests/unit/integrations/test_opentelemetry.py +++ b/tests/unit/integrations/test_opentelemetry.py @@ -57,7 +57,10 @@ from litellm.types.services import ServiceLoggerPayload, ServiceTypes from collections.abc import AsyncIterator, Callable, Mapping from litellm.constants import KUBERNETES_POD_ROUTING_KEY, LOGGING_WORKER_MAX_TIME_PER_COROUTINE from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER -from litellm.types.utils import ChatCompletionMessageToolCall, Function +from litellm.litellm_core_utils.hidden_params import set_hidden_params +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper +from litellm.types.utils import ChatCompletionMessageToolCall, Delta, Function, ModelResponseStream, StreamingChoices from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome exporter = InMemorySpanExporter() @@ -401,6 +404,121 @@ def test_v1_auto_router_response_model_names_the_routed_model(decision: object, assert (attributes["gen_ai.response.model"], attributes["gen_ai.usage.input_tokens"]) == (expected, 3) +def _response_with_provider_model(model: str, provider_model: str | None) -> litellm.ModelResponse: + response: Final = litellm.ModelResponse(model=model) + if provider_model: + set_hidden_params(response, {"provider_response_model": provider_model}) + return response + + +def _aliased_response(provider_model: str | None) -> litellm.ModelResponse: + return _response_with_provider_model("qwen-small", provider_model) + + +def _streamed_aliased_response(provider_model: str | None) -> litellm.ModelResponse: + wrapper: Final = CustomStreamWrapper( + completion_stream=None, + model="qwen-small", + logging_obj=Logging( + model="qwen-small", + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="acompletion", + start_time=1.0, + litellm_call_id="alias-call", + function_id="test", + ), + custom_llm_provider="hosted_vllm", + ) + chunks: Final = tuple( + wrapper.chunk_creator( + chunk=ModelResponseStream( + id="chunk-1", + model=provider_model, + choices=[StreamingChoices(finish_reason=finish_reason, index=0, delta=delta)], + ) + ) + for finish_reason, delta in ((None, Delta(content="hi")), ("stop", Delta())) + ) + assembled: Final = litellm.stream_chunk_builder(chunks=list(chunks)) + assert isinstance(assembled, litellm.ModelResponse) and assembled.model == "qwen-small" + return assembled + + +def _logged_model_names( + response_obj: object, latest_semconv: bool, model_group: str = "qwen-small", decision: object = None +) -> tuple[object, object, object]: + semconv: Final = {OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL} if latest_semconv else set[OTELSemconvCategory]() + exporter: Final = InMemorySpanExporter() + provider: Final = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + with patch.dict(os.environ, {"OTEL_SEMCONV_STABILITY_OPT_IN": ""}): + otel: Final = OpenTelemetry( + config=OpenTelemetryConfig(semconv_stability_opt_in=semconv), tracer_provider=provider + ) + with provider.get_tracer(__name__).start_as_current_span("inference") as span: + otel.set_attributes( + span, + { + "model": "qwen-dep", + "standard_logging_object": { + "id": "alias-call", + "call_type": "acompletion", + "model_group": model_group, + "metadata": {"routing_decision": decision}, + }, + }, + response_obj, + ) + (finished,) = exporter.get_finished_spans() + attributes: Final = finished.attributes + return ( + attributes.get("gen_ai.response.model"), + attributes.get("litellm.model_group"), + attributes.get("gen_ai.request.model"), + ) + + +@pytest.mark.parametrize( + ("make_response", "provider_model", "latest_semconv", "expected_response_model"), + [ + (_aliased_response, "Qwen3-0.6B-served", True, "Qwen3-0.6B-served"), + (_streamed_aliased_response, "Qwen3-0.6B-served", True, "Qwen3-0.6B-served"), + (_aliased_response, None, True, "qwen-dep"), + (_aliased_response, "Qwen3-0.6B-served", False, "qwen-small"), + (_streamed_aliased_response, "Qwen3-0.6B-served", False, "qwen-small"), + ], + ids=["semconv", "semconv-streamed", "semconv-deployment-fallback", "legacy", "legacy-streamed"], +) +def test_aliased_response_model_names_the_provider_model_only_under_latest_semconv( + make_response: Callable[[str | None], litellm.ModelResponse], + provider_model: str | None, + latest_semconv: bool, + expected_response_model: str, +) -> None: + assert _logged_model_names(make_response(provider_model), latest_semconv) == ( + expected_response_model, + "qwen-small", + "qwen-dep", + ) + + +@pytest.mark.parametrize( + ("provider_model", "expected_response_model"), + [("Qwen3-0.6B-served", "Qwen3-0.6B-served"), (None, "answer")], + ids=["provider-model", "routed-model"], +) +def test_latest_semconv_auto_router_response_model_prefers_provider_then_routed_model( + provider_model: str | None, expected_response_model: str +) -> None: + assert _logged_model_names( + _response_with_provider_model("auto-router", provider_model), + latest_semconv=True, + model_group="auto-router", + decision={"router_model_name": "auto-router", "routed_model": "answer"}, + ) == (expected_response_model, "auto-router", "qwen-dep") + + def test_v1_server_tool_calls_lead_the_completion_function_call_attributes() -> None: exporter: Final = InMemorySpanExporter() provider: Final = TracerProvider()