This commit is contained in:
dingdangmao 2026-10-20 03:06:43 +00:00 • committed by GitHub
commit 3b1e31e73d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 198 additions and 14 deletions

View file

@ -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", ""),

View file

@ -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(

View file

@ -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"