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

Merge pull request #19636 from BerriAI/litellm_langfuse_callback

Browse Source 
Add litellm_callback_logging_failures_metric for Langfuse, Langfuse Otel and other Otel providers
This commit is contained in:
Sameer Kankute
2026-01-28 18:02:17 +05:30
committed by GitHub
parent 0fadcbb21f 11b4c456b4
commit f5e5569e40
4 changed files with 185 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/my-website/docs/proxy/prometheus.md
+6 -1
View File
@@ -130,7 +130,12 @@ Monitor failures while shipping logs to downstream callbacks like `s3_v3` cold s
| Metric Name | Description |
|----------------------|--------------------------------------|
| `litellm_callback_logging_failures_metric` | Total number of failed attempts to emit logs to a configured callback. Labels: `"callback_name"`. Use this to alert on callback delivery issues such as repeated failures when writing to `s3_v3`. |
| `litellm_callback_logging_failures_metric` | Total number of failed attempts to emit logs to a configured callback. Labels: `"callback_name"`. Use this to alert on callback delivery issues such as repeated failures when writing to `s3_v3`, `langfuse`, or `langfuse_otel` and other otel providers |
**Supported Callbacks:**
- `S3Logger` - S3 v2 cold storage failures
- `langfuse` - Langfuse logging failures
- `otel` - OpenTelemetry logging failures
## LLM Provider Metrics
litellm/integrations/langfuse/langfuse_prompt_management.py
+54 -38
View File
@@ -300,43 +300,59 @@ class LangfusePromptManagement(LangFuseLogger, PromptManagementBase, CustomLogge
)
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
standard_callback_dynamic_params = kwargs.get(
"standard_callback_dynamic_params"
)
langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request(
globalLangfuseLogger=self,
standard_callback_dynamic_params=standard_callback_dynamic_params,
in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache,
)
langfuse_logger_to_use.log_event_on_langfuse(
kwargs=kwargs,
response_obj=response_obj,
start_time=start_time,
end_time=end_time,
user_id=kwargs.get("user", None),
)
try:
standard_callback_dynamic_params = kwargs.get(
"standard_callback_dynamic_params"
)
langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request(
globalLangfuseLogger=self,
standard_callback_dynamic_params=standard_callback_dynamic_params,
in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache,
)
langfuse_logger_to_use.log_event_on_langfuse(
kwargs=kwargs,
response_obj=response_obj,
start_time=start_time,
end_time=end_time,
user_id=kwargs.get("user", None),
)
except Exception as e:
from litellm._logging import verbose_logger
verbose_logger.exception(
f"Langfuse Layer Error - Exception occurred while logging success event: {str(e)}"
)
self.handle_callback_failure(callback_name="langfuse")
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
standard_callback_dynamic_params = kwargs.get(
"standard_callback_dynamic_params"
)
langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request(
globalLangfuseLogger=self,
standard_callback_dynamic_params=standard_callback_dynamic_params,
in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache,
)
standard_logging_object = cast(
Optional[StandardLoggingPayload],
kwargs.get("standard_logging_object", None),
)
if standard_logging_object is None:
return
langfuse_logger_to_use.log_event_on_langfuse(
start_time=start_time,
end_time=end_time,
response_obj=None,
user_id=kwargs.get("user", None),
status_message=standard_logging_object["error_str"],
level="ERROR",
kwargs=kwargs,
)
try:
standard_callback_dynamic_params = kwargs.get(
"standard_callback_dynamic_params"
)
langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request(
globalLangfuseLogger=self,
standard_callback_dynamic_params=standard_callback_dynamic_params,
in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache,
)
standard_logging_object = cast(
Optional[StandardLoggingPayload],
kwargs.get("standard_logging_object", None),
)
if standard_logging_object is None:
return
langfuse_logger_to_use.log_event_on_langfuse(
start_time=start_time,
end_time=end_time,
response_obj=None,
user_id=kwargs.get("user", None),
status_message=standard_logging_object["error_str"],
level="ERROR",
kwargs=kwargs,
)
except Exception as e:
from litellm._logging import verbose_logger
verbose_logger.exception(
f"Langfuse Layer Error - Exception occurred while logging failure event: {str(e)}"
)
self.handle_callback_failure(callback_name="langfuse")
litellm/integrations/opentelemetry.py
+7 -2
View File
@@ -995,9 +995,13 @@ class OpenTelemetry(CustomLogger):
from opentelemetry._logs import SeverityNumber, get_logger, get_logger_provider
try:
from opentelemetry.sdk._logs import LogRecord as SdkLogRecord # type: ignore[attr-defined] # OTEL < 1.39.0
from opentelemetry.sdk._logs import (
LogRecord as SdkLogRecord, # type: ignore[attr-defined] # OTEL < 1.39.0
)
except ImportError:
from opentelemetry.sdk._logs._internal import LogRecord as SdkLogRecord # OTEL >= 1.39.0
from opentelemetry.sdk._logs._internal import (
LogRecord as SdkLogRecord, # OTEL >= 1.39.0
)
otel_logger = get_logger(LITELLM_LOGGER_NAME)
@@ -1618,6 +1622,7 @@ class OpenTelemetry(CustomLogger):
)
except Exception as e:
self.handle_callback_failure(callback_name= self.callback_name)
verbose_logger.exception(
"OpenTelemetry logging error in set_attributes %s", str(e)
)
tests/enterprise/litellm_enterprise/integrations/test_prometheus.py
+118
View File
@@ -937,6 +937,124 @@ def test_callback_failure_metric_different_callbacks(prometheus_logger):
)
@pytest.mark.asyncio
async def test_langfuse_callback_failure_metric(prometheus_logger):
"""
Test that Langfuse callback failures are properly tracked in Prometheus metrics.
This test verifies that when Langfuse logging fails, the
litellm_callback_logging_failures_metric is incremented with callback_name="langfuse".
"""
from unittest.mock import MagicMock, patch
from litellm.integrations.langfuse.langfuse_prompt_management import (
LangfusePromptManagement,
)
# Get initial value
initial_value = 0
try:
initial_value = prometheus_logger.litellm_callback_logging_failures_metric.labels(
callback_name="langfuse"
)._value.get()
except Exception:
initial_value = 0
# Create Langfuse logger with mocked initialization
with patch("litellm.integrations.langfuse.langfuse_prompt_management.langfuse_client_init"):
langfuse_logger = LangfusePromptManagement()
# Mock the log_event_on_langfuse to raise an exception
with patch(
"litellm.integrations.langfuse.langfuse_prompt_management.LangFuseHandler.get_langfuse_logger_for_request"
) as mock_get_logger:
mock_logger = MagicMock()
mock_logger.log_event_on_langfuse.side_effect = Exception("Langfuse API error")
mock_get_logger.return_value = mock_logger
# Mock handle_callback_failure to track calls
with patch.object(prometheus_logger, "increment_callback_logging_failure") as mock_increment:
# Inject prometheus logger into the langfuse logger
langfuse_logger.handle_callback_failure = lambda callback_name: mock_increment(
callback_name=callback_name
)
# Call async_log_success_event - should catch exception and increment metric
await langfuse_logger.async_log_success_event(
kwargs={},
response_obj={},
start_time=None,
end_time=None,
)
# Verify that increment was called with correct callback name
mock_increment.assert_called_once_with(callback_name="langfuse")
print("✓ Langfuse callback failure metric test passed")
@pytest.mark.asyncio
async def test_langfuse_otel_callback_failure_metric(prometheus_logger):
"""
Test that Langfuse OTEL callback failures are properly tracked in Prometheus metrics.
This test verifies that when Langfuse OTEL logging fails, the
litellm_callback_logging_failures_metric is incremented with callback_name="langfuse_otel".
"""
from unittest.mock import MagicMock, patch
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
# Get initial value
initial_value = 0
try:
initial_value = prometheus_logger.litellm_callback_logging_failures_metric.labels(
callback_name="langfuse_otel"
)._value.get()
except Exception:
initial_value = 0
# Create Langfuse OTEL logger with mocked initialization
with patch("litellm.integrations.opentelemetry.OpenTelemetry.__init__", return_value=None):
langfuse_otel_logger = LangfuseOtelLogger(callback_name="langfuse_otel")
langfuse_otel_logger.callback_name = "langfuse_otel"
# Mock handle_callback_failure to track calls
with patch.object(prometheus_logger, "increment_callback_logging_failure") as mock_increment:
# Inject prometheus logger into the langfuse otel logger
langfuse_otel_logger.handle_callback_failure = lambda callback_name: mock_increment(
callback_name=callback_name
)
# Test that the OpenTelemetry base class set_attributes exception handler works
# This is where langfuse_otel failures are caught and tracked
with patch.object(langfuse_otel_logger, "set_attributes") as mock_set_attributes:
# Simulate the exception handling in set_attributes
def set_attributes_with_error(*args, **kwargs):
# This simulates what happens in the real set_attributes method
try:
raise Exception("Attribute error")
except Exception as e:
langfuse_otel_logger.handle_callback_failure(callback_name=langfuse_otel_logger.callback_name)
mock_set_attributes.side_effect = set_attributes_with_error
# Call set_attributes
try:
langfuse_otel_logger.set_attributes(
span=MagicMock(),
kwargs={},
response_obj={}
)
except Exception:
pass
# Verify that increment was called with correct callback name
mock_increment.assert_called_with(callback_name="langfuse_otel")
print("✓ Langfuse OTEL callback failure metric test passed")
# ==============================================================================
# END CALLBACK FAILURE METRICS TESTS
# ==============================================================================