tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-05 12:23:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(langfuse_otel): ignore service logs and fix callback shadowing (#19298)

Browse Source 
* fix(langfuse_otel): ignore service logs and fix callback shadowing

* add test cases for service logger
This commit is contained in:
Harshit Jain
2026-01-19 05:53:47 -08:00
committed by GitHub
parent 07fbd77c91
commit fe92f4af9c
3 changed files with 164 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/_service_logger.py
+23 -16
View File
@@ -145,16 +145,19 @@ class ServiceLogging(CustomLogger):
event_metadata=event_metadata,
)
elif callback == "otel" or isinstance(callback, OpenTelemetry):
from litellm.proxy.proxy_server import open_telemetry_logger
_otel_logger_to_use: Optional[OpenTelemetry] = None
if isinstance(callback, OpenTelemetry):
_otel_logger_to_use = callback
else:
from litellm.proxy.proxy_server import open_telemetry_logger
await self.init_otel_logger_if_none()
if open_telemetry_logger is not None and isinstance(
open_telemetry_logger, OpenTelemetry
):
_otel_logger_to_use = open_telemetry_logger
if (
parent_otel_span is not None
and open_telemetry_logger is not None
and isinstance(open_telemetry_logger, OpenTelemetry)
):
await self.otel_logger.async_service_success_hook(
if _otel_logger_to_use is not None and parent_otel_span is not None:
await _otel_logger_to_use.async_service_success_hook(
payload=payload,
parent_otel_span=parent_otel_span,
start_time=start_time,
@@ -253,20 +256,24 @@ class ServiceLogging(CustomLogger):
event_metadata=event_metadata,
)
elif callback == "otel" or isinstance(callback, OpenTelemetry):
from litellm.proxy.proxy_server import open_telemetry_logger
_otel_logger_to_use: Optional[OpenTelemetry] = None
if isinstance(callback, OpenTelemetry):
_otel_logger_to_use = callback
else:
from litellm.proxy.proxy_server import open_telemetry_logger
await self.init_otel_logger_if_none()
if open_telemetry_logger is not None and isinstance(
open_telemetry_logger, OpenTelemetry
):
_otel_logger_to_use = open_telemetry_logger
if not isinstance(error, str):
error = str(error)
if (
parent_otel_span is not None
and open_telemetry_logger is not None
and isinstance(open_telemetry_logger, OpenTelemetry)
):
await self.otel_logger.async_service_success_hook(
if _otel_logger_to_use is not None and parent_otel_span is not None:
await _otel_logger_to_use.async_service_failure_hook(
payload=payload,
error=error,
parent_otel_span=parent_otel_span,
start_time=start_time,
end_time=end_time,
litellm/integrations/langfuse/langfuse_otel.py
+56 -11
View File
@@ -156,7 +156,11 @@ class LangfuseOtelLogger(OpenTelemetry):
"arguments": arguments_obj,
}
transformed_tool_calls.append(langfuse_tool_call)
safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(transformed_tool_calls))
safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_OUTPUT.value,
safe_dumps(transformed_tool_calls),
)
else:
output_data = {}
if message.get("role"):
@@ -164,7 +168,11 @@ class LangfuseOtelLogger(OpenTelemetry):
if message.get("content") is not None:
output_data["content"] = message.get("content")
if output_data:
safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(output_data))
safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_OUTPUT.value,
safe_dumps(output_data),
)
output = response_obj.get("output", [])
if output:
@@ -175,15 +183,28 @@ class LangfuseOtelLogger(OpenTelemetry):
if item_type == "reasoning" and hasattr(item, "summary"):
for summary in item.summary:
if hasattr(summary, "text"):
output_items_data.append({"role": "reasoning_summary", "content": summary.text})
output_items_data.append(
{
"role": "reasoning_summary",
"content": summary.text,
}
)
elif item_type == "message":
output_items_data.append({
"role": getattr(item, "role", "assistant"),
"content": getattr(getattr(item, "content", [{}])[0], "text", "")
})
output_items_data.append(
{
"role": getattr(item, "role", "assistant"),
"content": getattr(
getattr(item, "content", [{}])[0], "text", ""
),
}
)
elif item_type == "function_call":
arguments_str = getattr(item, "arguments", "{}")
arguments_obj = json.loads(arguments_str) if isinstance(arguments_str, str) else arguments_str
arguments_obj = (
json.loads(arguments_str)
if isinstance(arguments_str, str)
else arguments_str
)
langfuse_tool_call = {
"id": getattr(item, "id", ""),
"name": getattr(item, "name", ""),
@@ -193,7 +214,11 @@ class LangfuseOtelLogger(OpenTelemetry):
}
output_items_data.append(langfuse_tool_call)
if output_items_data:
safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_OUTPUT.value, safe_dumps(output_items_data))
safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_OUTPUT.value,
safe_dumps(output_items_data),
)
@staticmethod
def _set_langfuse_specific_attributes(span: Span, kwargs, response_obj):
@@ -210,14 +235,22 @@ class LangfuseOtelLogger(OpenTelemetry):
langfuse_environment = os.environ.get("LANGFUSE_TRACING_ENVIRONMENT")
if langfuse_environment:
safe_set_attribute(span, LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value, langfuse_environment)
safe_set_attribute(
span,
LangfuseSpanAttributes.LANGFUSE_ENVIRONMENT.value,
langfuse_environment,
)
metadata = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)
LangfuseOtelLogger._set_metadata_attributes(span=span, metadata=metadata)
messages = kwargs.get("messages")
if messages:
safe_set_attribute(span, LangfuseSpanAttributes.OBSERVATION_INPUT.value, safe_dumps(messages))
safe_set_attribute(
span,
LangfuseSpanAttributes.OBSERVATION_INPUT.value,
safe_dumps(messages),
)
LangfuseOtelLogger._set_observation_output(span=span, response_obj=response_obj)
@@ -319,3 +352,15 @@ class LangfuseOtelLogger(OpenTelemetry):
dynamic_headers["Authorization"] = auth_header
return dynamic_headers
async def async_service_success_hook(self, *args, **kwargs):
"""
Langfuse should not receive service success logs.
"""
pass
async def async_service_failure_hook(self, *args, **kwargs):
"""
Langfuse should not receive service failure logs.
"""
pass
tests/test_service_logger_otel.py
+85
View File
@@ -0,0 +1,85 @@
import os
import sys
import unittest
from unittest.mock import patch, AsyncMock, MagicMock
# Add the project root to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
import litellm
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
from litellm.integrations.opentelemetry import OpenTelemetry
from litellm.types.services import ServiceTypes
from litellm._service_logger import ServiceLogging
class TestServiceLoggerOTEL(unittest.IsolatedAsyncioTestCase):
def setUp(self):
# Reset callbacks before each test
litellm.service_callback = []
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-123"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-123"
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_tracing")
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_metrics")
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_logs")
async def test_langfuse_otel_ignores_service_logs(
self, mock_logs, mock_metrics, mock_tracing
):
"""
Test that LangfuseOtelLogger overrides the service logging hooks with 'pass'.
"""
logger = LangfuseOtelLogger()
# Verify hooks are overriden
self.assertEqual(
logger.async_service_success_hook.__qualname__,
"LangfuseOtelLogger.async_service_success_hook",
)
self.assertEqual(
logger.async_service_failure_hook.__qualname__,
"LangfuseOtelLogger.async_service_failure_hook",
)
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_tracing")
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_metrics")
@patch("litellm.integrations.opentelemetry.OpenTelemetry._init_logs")
async def test_service_logging_shadowing_fix(
self, mock_logs, mock_metrics, mock_tracing
):
"""
Test the architectural fix: multiple OTEL loggers should receive logs independently.
"""
# 1. Initialize two loggers
langfuse_logger = LangfuseOtelLogger()
otel_logger = OpenTelemetry()
# 2. Setup service_callback list
litellm.service_callback = [langfuse_logger, otel_logger]
service_logging = ServiceLogging()
# 3. Mock the base OpenTelemetry hook
with patch.object(
OpenTelemetry, "async_service_success_hook", new_callable=AsyncMock
) as mock_base_hook:
# Trigger a service event
await service_logging.async_service_success_hook(
service=ServiceTypes.DB,
call_type="success",
duration=0.1,
parent_otel_span=MagicMock(),
start_time=0.0,
end_time=1.0,
)
# The architectural fix ensures we call each correctly.
self.assertEqual(
mock_base_hook.call_count,
1,
"Generic OTEL logger should have received the log exactly once.",
)
if __name__ == "__main__":
unittest.main()