fix(otel): serialize pydantic Message objects in langfuse_otel, weave_otel and langtrace (#46845)

langfuse_otel, weave_otel and langtrace now encode litellm.Message objects, and any value JSON cannot encode, when they write the prompt, so spans keep their token usage and model parameters. Under LITELLM_OTEL_V2 the payload builder dumps Message objects instead of dropping them
This commit is contained in:
taxfree 2026-11-11 07:50:39 +09:00 • committed by GitHub
parent 9ef5542339
commit bc3a69ce35
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 323 additions and 10 deletions

View file

@ -17,6 +17,7 @@ from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes impor
safe_set_attribute,
)
from litellm.litellm_core_utils.core_helpers import require_attributes
from litellm.litellm_core_utils.safe_json_dumps import jsonable_or_str
from litellm.types.llms.openai import HttpxBinaryResponseContent, ResponsesAPIResponse
from litellm.types.utils import (
EmbeddingResponse,
@ -103,7 +104,7 @@ class LangfuseLLMObsOTELAttributes(BaseLLMObsOTELAttributes):
prompt["tools"] = tools
input: Final = prompt
safe_set_attribute(span, "langfuse.observation.input", json.dumps(input))
safe_set_attribute(span, "langfuse.observation.input", json.dumps(input, default=jsonable_or_str))
@staticmethod
@override

View file

@ -1,6 +1,7 @@
import json
from typing import TYPE_CHECKING, Any, Final
from litellm.litellm_core_utils.safe_json_dumps import jsonable_or_str
from litellm.types.integrations.otel_span_attributes import SpanAttributes
if TYPE_CHECKING:
@ -60,7 +61,7 @@ class LangtraceAttributes:
if prompts:
span.add_event(
name="gen_ai.content.prompt",
attributes={SpanAttributes.LLM_PROMPTS.value: json.dumps(prompts)},
attributes={SpanAttributes.LLM_PROMPTS.value: json.dumps(prompts, default=jsonable_or_str)},
)
self.set_span_attributes(span, span_attributes)

View file

@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, ClassVar, Final, Literal, Protocol, cast
from urllib.parse import urlsplit
from openai.types.responses.response_function_web_search import ResponseFunctionWebSearch
from pydantic import TypeAdapter, ValidationError
from pydantic import BaseModel, TypeAdapter, ValidationError
from typing_extensions import ReadOnly, TypedDict
from litellm.integrations.otel.model.metadata import (
@ -519,7 +519,7 @@ class LLMCallSpanData:
promoted_metadata=promoted_metadata,
is_streaming=as_bool(payload.get("stream")),
tools=_extract_tools(params),
messages_in=_dicts(payload.get("messages")) if capture_content else (),
messages_in=_message_dicts(payload.get("messages")) if capture_content else (),
choices_out=choices_out if capture_content else (),
system_fingerprint=as_str(response.get("system_fingerprint")),
time_to_first_chunk_seconds=time_to_first_chunk_seconds,
@ -783,6 +783,27 @@ def _dicts(value: object) -> tuple[Mapping[str, object], ...]:
return tuple(item for item in value if isinstance(item, dict))
_OBJECTS: Final = TypeAdapter(tuple[object, ...])
_ANY_KEY_MAPPING: Final = TypeAdapter(Mapping[object, object])
def _message_dict(message: object) -> Mapping[str, object] | None:
if isinstance(message, dict):
return {str(key): value for key, value in _ANY_KEY_MAPPING.validate_python(message).items()}
if not isinstance(message, BaseModel):
return None
try:
return message.model_dump(mode="json", fallback=str)
except ValueError:
return None
def _message_dicts(value: object) -> tuple[Mapping[str, object], ...]:
if not isinstance(value, list):
return ()
return tuple(message for item in _OBJECTS.validate_python(value) if (message := _message_dict(item)) is not None)
def _finish_reasons(choices: tuple[Mapping[str, object], ...]) -> tuple[str, ...]:
"""Non-empty ``finish_reason`` of each response choice."""
return tuple(r for c in choices if (r := as_str(c.get("finish_reason"))))

View file

@ -21,7 +21,7 @@ from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes impor
safe_set_attribute,
)
from litellm.litellm_core_utils.core_helpers import has_attributes, require_attributes
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.litellm_core_utils.safe_json_dumps import jsonable_or_str, safe_dumps
from litellm.types.integrations.weave_otel import WeaveOtelConfig, WeaveSpanAttributes
from litellm.types.utils import StandardCallbackDynamicParams
@ -74,7 +74,11 @@ class WeaveLLMObsOTELAttributes(BaseLLMObsOTELAttributes):
prompt["functions"] = functions
if tools is not None:
prompt["tools"] = tools
safe_set_attribute(span, OpenInferenceSpanAttributes.INPUT_VALUE, json.dumps(prompt))
safe_set_attribute(
span,
OpenInferenceSpanAttributes.INPUT_VALUE,
json.dumps(prompt, default=jsonable_or_str),
)
def _set_weave_specific_attributes(span: Span, kwargs: Mapping[str, object], response_obj: object):

View file

@ -3,7 +3,8 @@ import math
from collections.abc import Callable
from typing import Any, Final
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter
from pydantic_core import to_jsonable_python
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
@ -113,3 +114,13 @@ def safe_dumps(
return json.dumps(
safe_json_structure(data, max_depth, value_transform, non_finite_as_null=non_finite_as_null), default=str
)
_JSONABLE: Final = TypeAdapter(object)
def jsonable_or_str(value: object) -> object:
try:
return _JSONABLE.validate_python(to_jsonable_python(value, fallback=str))
except ValueError:
return str(value)

View file

@ -15,6 +15,7 @@ import pytest
from litellm.integrations.otel import GenAIOperation
from litellm.integrations.otel.mappers import (
GenAIMapper,
LangfuseMapper,
LangtraceMapper,
OpenInferenceMapper,
@ -33,6 +34,7 @@ from litellm.integrations.otel.model.payloads import (
ToolDefinition,
)
from litellm.integrations.otel.model.trace_controls import TraceControls
from litellm.types.utils import Message
from tests.unit.integrations.otel.test_otel_v2_sources_of_truth import _responses_payload
@ -708,6 +710,78 @@ def test_langfuse_mapper_keeps_chat_output_when_no_embedding_summary():
assert json.loads(attrs["langfuse.observation.output"]) == [{"role": "assistant", "content": "Sunny."}]
LITELLM_USER_MESSAGE: Final = Message(role="user", content="weather in sf?")
NOT_JSON_ENCODABLE: Final = object()
@pytest.mark.parametrize(
("message", "expected_prompt"),
[
({"role": "user", "content": "weather in sf?"}, {"role": "user", "content": "weather in sf?"}),
(
{"role": "user", "content": "weather in sf?", 1: "one"},
{"role": "user", "content": "weather in sf?", "1": "one"},
),
(LITELLM_USER_MESSAGE, LITELLM_USER_MESSAGE.model_dump(mode="json")),
(
Message(role="user", content="weather in sf?", trace_handle=NOT_JSON_ENCODABLE),
{**LITELLM_USER_MESSAGE.model_dump(mode="json"), "trace_handle": str(NOT_JSON_ENCODABLE)},
),
],
ids=[
"dict_message",
"dict_message_with_int_key",
"litellm_message_object",
"litellm_message_with_value_json_cannot_encode",
],
)
def test_vendor_mappers_keep_the_prompt_of_litellm_message_objects(
message: Mapping[str, object] | Message, expected_prompt: Mapping[str, object]
) -> None:
payload: Final = {
"call_type": "acompletion",
"custom_llm_provider": "openai",
"model": "gpt-5.4-nano",
"messages": [message],
"response": {"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "Sunny."}}]},
}
data: Final = LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True)
assert {
"genai": json.loads(str(GenAIMapper().map(data)["gen_ai.input.messages"])),
"langfuse": json.loads(str(LangfuseMapper().map(data)["langfuse.observation.input"])),
"langtrace": json.loads(str(LangtraceMapper().map(data)["llm.prompts"])),
"openinference": json.loads(str(OpenInferenceMapper().map(data)["input.value"])),
} == {
"genai": [expected_prompt],
"langfuse": [expected_prompt],
"langtrace": [expected_prompt],
"openinference": [{"role": "user", "content": "weather in sf?"}],
}
def test_vendor_mappers_keep_the_span_when_a_litellm_message_cannot_be_dumped() -> None:
payload: Final = {
"call_type": "acompletion",
"custom_llm_provider": "openai",
"model": "gpt-5.4-nano",
"messages": [
{"role": "system", "content": "be brief"},
Message(role="user", content="weather in sf?", audio_bytes=b"\xff\xfe"),
],
"response": {"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "Sunny."}}]},
}
attrs: Final = LangfuseMapper().map(LLMCallSpanData.from_standard_logging_payload(payload, capture_content=True))
assert {
"input": json.loads(str(attrs["langfuse.observation.input"])),
"output": json.loads(str(attrs["langfuse.observation.output"])),
} == {
"input": [{"role": "system", "content": "be brief"}],
"output": [{"role": "assistant", "content": "Sunny."}],
}
def test_langfuse_mapper_renders_a_responses_api_call_from_the_standard_logging_payload():
payload = {
"call_type": "aresponses",

View file

@ -20,6 +20,7 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanE
from opentelemetry.trace import Span
import litellm
from litellm.integrations._types.open_inference import SpanAttributes
from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, REDACTED_BY_LITELLM
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
from litellm.integrations.langfuse.langfuse_otel_attributes import LangfuseLLMObsOTELAttributes
@ -28,7 +29,20 @@ from litellm.integrations.otel.model.destination import OtelDestination
from litellm.integrations.otel.plumbing.context import set_request_destinations
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams
from litellm.types.utils import Choices, Message, ModelResponse, StandardCallbackDynamicParams
CHAT_MESSAGE_DICTS: Final = ({"role": "user", "content": "hello"}, {"role": "assistant", "content": "hi back"})
CHAT_MESSAGE_OBJECTS: Final = tuple(
Message(role=message["role"], content=message["content"]) for message in CHAT_MESSAGE_DICTS
)
NOT_JSON_ENCODABLE: Final = object()
NON_UTF8_BYTES: Final = b"\x80"
GENERATION_USAGE_AND_PARAM_KEYS: Final = (
SpanAttributes.LLM_TOKEN_COUNT_PROMPT,
SpanAttributes.LLM_TOKEN_COUNT_COMPLETION,
SpanAttributes.LLM_TOKEN_COUNT_TOTAL,
SpanAttributes.LLM_INVOCATION_PARAMETERS,
)
class TestLangfuseOtelIntegration:
@ -140,6 +154,52 @@ class TestLangfuseOtelIntegration:
"langfuse.observation.type", "generation"
)
@pytest.mark.parametrize(
("messages", "expected_input"),
[
(CHAT_MESSAGE_DICTS, list(CHAT_MESSAGE_DICTS)),
(CHAT_MESSAGE_OBJECTS, [message.model_dump() for message in CHAT_MESSAGE_OBJECTS]),
(
(Message(role="user", content="hello", trace_handle=NOT_JSON_ENCODABLE),),
[{**CHAT_MESSAGE_OBJECTS[0].model_dump(), "trace_handle": str(NOT_JSON_ENCODABLE)}],
),
(
(Message(role="user", content="hello", audio_bytes=NON_UTF8_BYTES),),
[{**CHAT_MESSAGE_OBJECTS[0].model_dump(), "audio_bytes": str(NON_UTF8_BYTES)}],
),
],
ids=[
"dict_messages",
"litellm_message_objects",
"litellm_message_with_value_json_cannot_encode",
"litellm_message_with_non_utf8_bytes",
],
)
def test_generation_keeps_input_model_params_and_token_usage(
self, messages: Sequence[object], expected_input: Sequence[object]
) -> None:
span: Final = TracerProvider().get_tracer("langfuse-otel-tests").start_span("litellm_request")
response: Final = ModelResponse(
model="gpt-4o",
choices=[{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "done"}}],
usage={"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10},
)
LangfuseOtelLogger.set_langfuse_otel_attributes(
span,
{"messages": messages, "standard_logging_object": {"model_parameters": {"temperature": 0.2}}},
response,
)
attributes: Final = span.attributes or {}
assert json.loads(str(attributes["langfuse.observation.input"])) == expected_input
assert {key: attributes.get(key) for key in GENERATION_USAGE_AND_PARAM_KEYS} == {
SpanAttributes.LLM_TOKEN_COUNT_PROMPT: 7,
SpanAttributes.LLM_TOKEN_COUNT_COMPLETION: 3,
SpanAttributes.LLM_TOKEN_COUNT_TOTAL: 10,
SpanAttributes.LLM_INVOCATION_PARAMETERS: json.dumps({"temperature": 0.2}),
}
def test_set_langfuse_environment_attribute(self):
"""Test that Langfuse environment is set correctly when environment variable is present."""
mock_span = MagicMock()

View file

@ -0,0 +1,76 @@
import json
from collections.abc import Sequence
from typing import Final
import pytest
from opentelemetry.sdk.trace import TracerProvider
from litellm.integrations.langtrace import LangtraceAttributes
from litellm.types.integrations.otel_span_attributes import SpanAttributes
from litellm.types.utils import Message, ModelResponse
CHAT_MESSAGE_DICTS: Final = ({"role": "user", "content": "weather in SF?"}, {"role": "assistant", "content": "Sunny."})
CHAT_MESSAGE_OBJECTS: Final = tuple(
Message(role=message["role"], content=message["content"]) for message in CHAT_MESSAGE_DICTS
)
NOT_JSON_ENCODABLE: Final = object()
NON_UTF8_BYTES: Final = b"\x80"
@pytest.mark.parametrize(
("messages", "expected_prompts"),
[
(CHAT_MESSAGE_DICTS, list(CHAT_MESSAGE_DICTS)),
(CHAT_MESSAGE_OBJECTS, [message.model_dump() for message in CHAT_MESSAGE_OBJECTS]),
(
({"role": "user", "content": "before\x00after", "metadata": {1: "one"}},),
[{"role": "user", "content": "before\x00after", "metadata": {"1": "one"}}],
),
(
(Message(role="user", content="weather in SF?", trace_handle=NOT_JSON_ENCODABLE),),
[{**CHAT_MESSAGE_OBJECTS[0].model_dump(), "trace_handle": str(NOT_JSON_ENCODABLE)}],
),
(
(Message(role="user", content="weather in SF?", audio_bytes=NON_UTF8_BYTES),),
[str(Message(role="user", content="weather in SF?", audio_bytes=NON_UTF8_BYTES))],
),
],
ids=[
"dict_messages",
"litellm_message_objects",
"dict_message_with_nul_and_int_key",
"litellm_message_with_value_json_cannot_encode",
"litellm_message_with_non_utf8_bytes",
],
)
def test_langtrace_span_keeps_prompt_and_token_usage(
messages: Sequence[object], expected_prompts: Sequence[object]
) -> None:
span: Final = TracerProvider().get_tracer("langtrace-tests").start_span("litellm_request")
response: Final = ModelResponse(
model="gpt-4o",
choices=[{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "Still sunny."}}],
usage={"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10},
)
LangtraceAttributes().set_langtrace_attributes(
span, {"messages": messages, "litellm_params": {"custom_llm_provider": "openai"}}, response
)
assert [
json.loads(str((event.attributes or {})[SpanAttributes.LLM_PROMPTS.value]))
for event in span.events
if event.name == "gen_ai.content.prompt"
] == [expected_prompts]
assert {
key: (span.attributes or {}).get(key)
for key in (
SpanAttributes.LLM_USAGE_PROMPT_TOKENS.value,
SpanAttributes.LLM_USAGE_COMPLETION_TOKENS.value,
SpanAttributes.LLM_USAGE_TOTAL_TOKENS.value,
)
} == {
SpanAttributes.LLM_USAGE_PROMPT_TOKENS.value: 7,
SpanAttributes.LLM_USAGE_COMPLETION_TOKENS.value: 3,
SpanAttributes.LLM_USAGE_TOTAL_TOKENS.value: 10,
}

View file

@ -1,6 +1,6 @@
import json
import os
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from typing import Final
from unittest.mock import MagicMock, patch
@ -12,9 +12,10 @@ from litellm.integrations.weave.weave_otel import (
WeaveLLMObsOTELAttributes,
_set_weave_specific_attributes,
get_weave_otel_config,
set_weave_otel_attributes,
)
from litellm.types.integrations.weave_otel import WeaveOtelConfig, WeaveSpanAttributes
from litellm.types.utils import ModelResponse
from litellm.types.utils import Message, ModelResponse
def test_get_weave_otel_config():
@ -205,3 +206,67 @@ def test_weave_output_value_is_the_dumped_model_response() -> None:
output: Final = json.loads(str(span.attributes[OpenInferenceSpanAttributes.OUTPUT_VALUE]))
assert output["id"] == "chatcmpl-1"
assert output["choices"][0]["message"]["content"] == "Sunny."
NOT_JSON_ENCODABLE: Final = object()
NON_UTF8_BYTES: Final = b"\x80"
@pytest.mark.parametrize(
("messages", "expected_messages"),
[
(
(Message(role="user", content="weather in SF?"), Message(role="assistant", content="Sunny.")),
[
Message(role="user", content="weather in SF?").model_dump(),
Message(role="assistant", content="Sunny.").model_dump(),
],
),
(
(Message(role="user", content="weather in SF?", trace_handle=NOT_JSON_ENCODABLE),),
[{**Message(role="user", content="weather in SF?").model_dump(), "trace_handle": str(NOT_JSON_ENCODABLE)}],
),
(
(Message(role="user", content="weather in SF?", audio_bytes=NON_UTF8_BYTES),),
[str(Message(role="user", content="weather in SF?", audio_bytes=NON_UTF8_BYTES))],
),
],
ids=["litellm_messages", "litellm_message_with_value_json_cannot_encode", "litellm_message_with_non_utf8_bytes"],
)
def test_weave_span_keeps_input_and_token_usage_when_messages_are_litellm_messages(
messages: tuple[Message, ...], expected_messages: Sequence[object]
) -> None:
span: Final = _TRACER.start_span("litellm_request")
response: Final = ModelResponse(
model="gpt-4o",
choices=[{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "Still sunny."}}],
usage={"prompt_tokens": 7, "completion_tokens": 3, "total_tokens": 10},
)
set_weave_otel_attributes(span, {"messages": messages, "standard_logging_object": {}}, response)
attributes: Final = span.attributes or {}
assert json.loads(str(attributes[OpenInferenceSpanAttributes.INPUT_VALUE])) == {"messages": expected_messages}
assert {
key: attributes.get(key)
for key in (
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_PROMPT,
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_COMPLETION,
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_TOTAL,
)
} == {
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_PROMPT: 7,
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_COMPLETION: 3,
OpenInferenceSpanAttributes.LLM_TOKEN_COUNT_TOTAL: 10,
}
def test_weave_span_input_keeps_dict_message_content_exactly_as_sent() -> None:
span: Final = _TRACER.start_span("litellm_request")
messages: Final = ({"role": "user", "content": "before\x00after", "metadata": {1: "one"}},)
set_weave_otel_attributes(span, {"messages": messages, "standard_logging_object": {}}, ModelResponse(model="gpt-4o"))
assert json.loads(str((span.attributes or {})[OpenInferenceSpanAttributes.INPUT_VALUE])) == {
"messages": [{"role": "user", "content": "before\x00after", "metadata": {"1": "one"}}]
}