diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index 90392d9a45d..af3df171deb 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -1,7 +1,7 @@ import os import threading from collections import OrderedDict -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from datetime import datetime @@ -2791,7 +2791,9 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): primitive_value: Final = self._cast_as_primitive_value_type(value) span.set_attribute(key, primitive_value) - def _transform_messages_to_otel_semantic_conventions(self, messages: list[dict] | str) -> list[dict]: + def _transform_messages_to_otel_semantic_conventions( + self, messages: Sequence[dict[str, object]] | str + ) -> list[dict[str, object]]: # mutable-ok: match mixin base return """ Transforms LiteLLM/OpenAI style messages into OTEL GenAI 1.38 compliant format. OTEL expects a 'parts' array instead of a single 'content' string. @@ -2807,25 +2809,61 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): transformed: Final = [] for msg in messages: + # Normalize pydantic / LiteLLM wrapper objects (e.g. a ModelResponse + # Message) to something we can read with .get(); "tool_calls" in msg + # is False for those objects, which silently dropped tool calls. + normalized = self._to_dict(msg) + if normalized is None: + continue + msg = cast(dict[str, object], normalized) # cast-ok: _to_dict guarantees a dict-like payload + role = msg.get("role", "user") content = msg.get("content", "") - parts = [] + parts: list[dict[str, object]] = [] # mutable-ok: local accumulator returned read-only if isinstance(content, str): parts.append({"type": "text", "content": content}) elif isinstance(content, list): # Handle multi-modal content if necessary - for part in content: + for part in cast(list[object], content): # cast-ok: content already narrowed to list if isinstance(part, dict): parts.append(part) else: parts.append({"type": "text", "content": str(part)}) - transformed_msg = {"role": role, "parts": parts} + # Record tool calls as GenAI tool_call parts (OpenLLMetry-compatible) + # so id / name / arguments are not dropped from the trace. + raw_tool_calls = msg.get("tool_calls") + tool_calls: Sequence[object] = raw_tool_calls if isinstance(raw_tool_calls, list) else () + for tc in tool_calls: + if isinstance(tc, dict): + tc_dict = cast(dict[str, object], tc) # cast-ok: tc narrowed to dict + tc_id = tc_dict.get("id") + fn = tc_dict.get("function") + if isinstance(fn, dict): + fn_dict = cast(dict[str, object], fn) # cast-ok: fn narrowed to dict + fn_name = fn_dict.get("name") + fn_arguments = fn_dict.get("arguments") + else: + fn_name = None + fn_arguments = None + else: + # pydantic tool-call objects (e.g. ChatCompletionMessageToolCall): + # read fields via literal getattr; arbitrary objects degrade to + # None exactly like the old .get()/getattr pair. + tc_id = cast(object, getattr(tc, "id", None)) # cast-ok: getattr returns object + fn = cast(object, getattr(tc, "function", None)) # cast-ok: getattr returns object + if fn is None: + fn_name = None + fn_arguments = None + else: + fn_name = cast(object, getattr(fn, "name", None)) # cast-ok: getattr returns object + fn_arguments = cast(object, getattr(fn, "arguments", None)) # cast-ok: getattr returns object + parts.append({"type": "tool_call", "id": tc_id, "name": fn_name, "arguments": fn_arguments}) + + transformed_msg: dict[str, object] = {"role": role, "parts": parts} # mutable-ok: key-assigned below if "id" in msg: transformed_msg["id"] = msg["id"] - if "tool_calls" in msg: - transformed_msg["tool_calls"] = msg["tool_calls"] if "tool_call_id" in msg: transformed_msg["tool_call_id"] = msg["tool_call_id"] transformed.append(transformed_msg) @@ -2895,13 +2933,19 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): item = self._to_dict(raw_item) if item is None: continue + item = cast(dict[str, object], item) # cast-ok: _to_dict guarantees a dict-like payload if item.get("type") == "message": role = item.get("role", "assistant") - parts = [] - for raw_content in item.get("content", []): + parts: list[dict[str, object]] = [] # mutable-ok: local accumulator returned read-only + raw_content_list = item.get("content", []) + content_items = cast( # cast-ok: narrowed to list above + Sequence[object], raw_content_list if isinstance(raw_content_list, list) else () + ) + for raw_content in content_items: content = self._to_dict(raw_content) if content is None: continue + content = cast(dict[str, object], content) # cast-ok: _to_dict guarantees a dict-like payload if content.get("type") == "output_text": text = content.get("text", "") if text: @@ -2910,7 +2954,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): transformed.append({"role": role, "parts": parts}) elif item.get("type") == "function_call": # Surface tool calls from Responses API output - part: dict = { + part: dict[str, object] = { # mutable-ok: assembled then key-assigned before append "type": "tool_call", "name": item.get("name", ""), "arguments": item.get("arguments", ""), diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index d505192184e..b7d7665fe33 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -369,6 +369,33 @@ class ChunkProcessor: def _get_role_from_chunks(chunks: Sequence["_BaseChunk"]) -> str: return ChunkProcessor._role_of_choice(next((c["choices"][0] for c in chunks if c.get("choices")), None)) + @staticmethod + def _read_delta(choice: StreamingChoices | Mapping[str, object]) -> object: + """Return a streaming choice's delta payload. + + ``StreamingChoices`` (an ``openai`` ``BaseModel`` subclass) declares + ``delta``; raw dict chunks carry the same data under a ``delta`` key. + """ + return choice.get("delta") if isinstance(choice, Mapping) else choice.delta + + @staticmethod + def _read_message(choice: StreamingChoices | Mapping[str, object]) -> object: + """Return a streaming choice's message payload. + + Only dict chunks carry ``message``: ``StreamingChoices.__init__`` pops + it (Perplexity compat), so non-dict choices always read as ``None``. + """ + return choice.get("message") if isinstance(choice, Mapping) else None + + @staticmethod + def _read_tool_calls(payload: object) -> object: + """Return a delta/message payload's ``tool_calls``, dict or object form.""" + if isinstance(payload, dict): + return cast(dict[str, object], payload).get("tool_calls") # cast-ok: payload narrowed to dict + if isinstance(payload, Delta): + return payload.tool_calls + return None + @staticmethod def _role_of_choice(choice: object) -> str: match choice: @@ -414,15 +441,35 @@ class ChunkProcessor: role: Final = ChunkProcessor._get_role_from_chunks(chunks) finish_reason = "stop" + saw_tool_call = False # rebind-ok: flipped to True inside the loop when a tool-call delta or message is seen for chunk in chunks: if "choices" in chunk and len(chunk["choices"]) > 0: + choice = chunk["choices"][0] + delta = ChunkProcessor._read_delta(choice) + if delta is not None: + delta_tool_calls = ChunkProcessor._read_tool_calls(delta) + if delta_tool_calls: + saw_tool_call = True + message = ChunkProcessor._read_message(choice) + if message is not None: + msg_tool_calls = ChunkProcessor._read_tool_calls(message) + if msg_tool_calls: + saw_tool_call = True chunk_finish_reason = None - if hasattr(chunk["choices"][0], "finish_reason"): - chunk_finish_reason = chunk["choices"][0].finish_reason - elif "finish_reason" in chunk["choices"][0]: - chunk_finish_reason = chunk["choices"][0]["finish_reason"] + if hasattr(choice, "finish_reason"): + # StreamingChoices is a dynamic OpenAIObject: the attribute may + # or may not exist at runtime, so read via literal getattr. + chunk_finish_reason = cast(str, getattr(choice, "finish_reason", None)) # cast-ok: dynamic attr + elif "finish_reason" in choice: + chunk_finish_reason = cast(str, choice["finish_reason"]) # cast-ok: key checked if chunk_finish_reason is not None: finish_reason = chunk_finish_reason + if finish_reason == "stop" and saw_tool_call: + # A tool-calling stream whose chunks never surfaced an explicit + # finish_reason must aggregate to "tool_calls" (OpenAI spec), + # not the default "stop" - otherwise callbacks (e.g. OpenTelemetry) + # record a wrong finish reason for single-chunk streams. (litellm#45796) + finish_reason = "tool_calls" # rebind-ok: aggregated finish reason for tool-calling streams (litellm#45796) # Initialize the response dictionary response = ModelResponse( diff --git a/tests/unit/integrations/test_otel_tool_call_parts.py b/tests/unit/integrations/test_otel_tool_call_parts.py new file mode 100644 index 00000000000..52bed152314 --- /dev/null +++ b/tests/unit/integrations/test_otel_tool_call_parts.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""Regression tests for #45796: OTEL tool-call parts + streamed finish reason. + +Covered behavior: +- ``_transform_messages_to_otel_semantic_conventions`` emits GenAI + ``{"type": "tool_call", "id", "name", "arguments"}`` parts for assistant + messages carrying ``tool_calls`` — for plain dicts and for Pydantic + message objects (which used to be skipped because ``"tool_calls" in msg`` + is False for non-dict objects). +- Plain string ``content`` still emits a text part (no regression). +- ``ChunkProcessor.build_base_response`` aggregates a stream whose chunks + carried tool-call deltas but no explicit ``finish_reason`` to + ``"tool_calls"`` instead of the default ``"stop"``. +""" +from pydantic import BaseModel + +from litellm.integrations.opentelemetry import OpenTelemetry +from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor + + +class _FakeAssistantMessage(BaseModel): + role: str = "assistant" + content: str | None = None + tool_calls: list | None = None + + +class TestOtelToolCallParts: + def test_dict_message_with_tool_calls_emits_genai_tool_call_parts(self): + msg = { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "SF"}'}, + } + ], + } + out = OpenTelemetry()._transform_messages_to_otel_semantic_conventions([msg])[0] + assert out["parts"] == [ + {"type": "tool_call", "id": "call_1", "name": "get_weather", "arguments": '{"city": "SF"}'} + ] + assert "tool_calls" not in out + + def test_pydantic_message_with_tool_calls_emits_genai_tool_call_parts(self): + msg = _FakeAssistantMessage( + tool_calls=[ + { + "id": "call_2", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ] + ) + out = OpenTelemetry()._transform_messages_to_otel_semantic_conventions([msg])[0] + assert out["parts"] == [ + {"type": "tool_call", "id": "call_2", "name": "lookup", "arguments": "{}"} + ] + + def test_string_content_still_emits_text_part(self): + out = OpenTelemetry()._transform_messages_to_otel_semantic_conventions( + [{"role": "assistant", "content": "hello"}] + )[0] + assert out["parts"] == [{"type": "text", "content": "hello"}] + + def test_streamed_tool_call_without_explicit_finish_reason_aggregates_to_tool_calls(self): + chunks = [ + { + "object": "chat.completion.chunk", + "created": 0, + "model": "mock", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": None}}], + }, + { + "choices": [ + { + "index": 0, + "delta": { + "tool_calls": [ + { + "id": "c1", + "type": "function", + "function": {"name": "f", "arguments": "{}"}, + } + ] + }, + } + ] + }, + ] + resp = ChunkProcessor(chunks).build_base_response(chunks) + assert resp.choices[0].finish_reason == "tool_calls"