diff --git a/litellm/integrations/langfuse/langfuse_prompt_management.py b/litellm/integrations/langfuse/langfuse_prompt_management.py index 2c393019e17..ec3156ac56f 100644 --- a/litellm/integrations/langfuse/langfuse_prompt_management.py +++ b/litellm/integrations/langfuse/langfuse_prompt_management.py @@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.prompt_management_base import PromptManagementClient from litellm.litellm_core_utils.asyncify import run_async_function +from litellm.llms.custom_httpx.http_handler import HTTPHandler from litellm.types.integrations.langfuse import LangfuseLoggedEvent from litellm.types.llms.openai import AllMessageValues, ChatCompletionSystemMessage from litellm.types.prompts.init_prompts import PromptSpec @@ -92,20 +93,7 @@ def langfuse_client_init( raise_if_unsupported_langfuse_version(installed_langfuse_version()) warn_if_upstream_langfuse_configured() - import httpx - - import litellm - - from ...llms.custom_httpx.http_handler import get_ssl_configuration - - httpx_client: Final = ( - create_mock_langfuse_client() - if should_use_langfuse_mock() - else httpx.Client( - verify=get_ssl_configuration(), - cert=os.getenv("SSL_CERTIFICATE", litellm.ssl_certificate), - ) - ) + httpx_client: Final = create_mock_langfuse_client() if should_use_langfuse_mock() else HTTPHandler().client return build_langfuse_client( public_key=public_key, secret_key=secret_key, diff --git a/litellm/integrations/langfuse/langfuse_sdk.py b/litellm/integrations/langfuse/langfuse_sdk.py index a719e4950c8..3bdb318e512 100644 --- a/litellm/integrations/langfuse/langfuse_sdk.py +++ b/litellm/integrations/langfuse/langfuse_sdk.py @@ -21,6 +21,7 @@ from langfuse import LangfuseOtelSpanAttributes from langfuse.api import LangfuseAPI, Prompt, Prompt_Chat from langfuse.model import BasePromptClient, ChatPromptClient, PromptClient, TextPromptClient from opentelemetry.context import Context +from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult @@ -28,18 +29,18 @@ from opentelemetry.sdk.trace.id_generator import RandomIdGenerator from opentelemetry.sdk.trace.sampling import ALWAYS_ON, Decision, Sampler, SamplingResult from opentelemetry.trace import Link, NonRecordingSpan, Span, SpanContext, SpanKind, TraceFlags, Tracer, TraceState from opentelemetry.util.types import Attributes, AttributeValue -from requests import PreparedRequest, RequestException, Response, Session -from requests.adapters import HTTPAdapter +import litellm from litellm._logging import verbose_logger from litellm.litellm_core_utils.safe_json_dumps import safe_dumps +from litellm.llms.custom_httpx.http_handler import HTTPHandler, _get_httpx_client __all__ = ( "DiscardingSpanExporter", "LangfuseApiClient", "LangfuseObservation", + "LangfuseSpanExporter", "LangfuseTracing", - "RetryingSpanExporter", "TraceIdHashSampler", "acquire_langfuse_tracing", "build_langfuse_client", @@ -461,90 +462,78 @@ class DiscardingSpanExporter(SpanExporter): return True -@dataclass(frozen=True, slots=True) -class RetryingSpanExporter(SpanExporter): - """Retry a batch whose HTTP round trip raised, as the v2 ingestion consumer did. +_RETRYABLE_EXPORT_STATUSES: Final = frozenset({408, 429, 500, 502, 503, 504}) +_ExportOutcome = Literal["delivered", "retry", "rejected"] - The OTLP http exporter only retries 429 and 5xx responses; a connect or - read timeout propagates, and ``BatchSpanProcessor`` drops the whole batch - on any exception. A destination that stalls for a few seconds therefore - lost every observation in flight, where v2 backed off three times first. + +@dataclass(frozen=True, slots=True) +class LangfuseSpanExporter(SpanExporter): + """OTLP/HTTP protobuf export through litellm's own HTTP handler. + + The handler carries litellm's TLS material (``ssl_verify``, CA bundle, client certificate) exactly + as v2's injected httpx client did. A connect or read failure and a retryable status are re-sent after + each delay, matching the v2 ingestion consumer; ``BatchSpanProcessor`` would otherwise drop the whole + batch on the first exception. """ - exporter: SpanExporter + handler: HTTPHandler + endpoint: str + headers: Mapping[str, str] + timeout: float delays: Sequence[float] = (1.0, 2.0, 4.0) def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + body: Final = encode_spans(spans).SerializeToString() for delay in self.delays: - try: - return self.exporter.export(spans) - except RequestException as error: - verbose_logger.warning("Langfuse export raised %s, retrying in %ss", error, delay) - sleep(delay) + outcome: _ExportOutcome = self._post(body) + if outcome != "retry": + return SpanExportResult.SUCCESS if outcome == "delivered" else SpanExportResult.FAILURE + verbose_logger.warning("Langfuse export to %s failed, retrying in %ss", self.endpoint, delay) + sleep(delay) + last: Final = self._post(body) + if last == "retry": + verbose_logger.error("Langfuse export to %s failed after %d retries", self.endpoint, len(self.delays)) + return SpanExportResult.SUCCESS if last == "delivered" else SpanExportResult.FAILURE + + def _post(self, body: bytes) -> _ExportOutcome: try: - return self.exporter.export(spans) - except RequestException as error: - verbose_logger.error("Langfuse export failed after %d retries: %s", len(self.delays), error) - return SpanExportResult.FAILURE + self.handler.post(self.endpoint, data=body, headers=dict(self.headers), timeout=self.timeout) + except httpx.HTTPStatusError as error: + status: Final = error.response.status_code + if status in _RETRYABLE_EXPORT_STATUSES: + return "retry" + verbose_logger.error("Langfuse rejected an export to %s with HTTP %d", self.endpoint, status) + return "rejected" + except (httpx.TransportError, litellm.Timeout) as error: + verbose_logger.warning("Langfuse export to %s raised %s", self.endpoint, error) + return "retry" + return "delivered" def shutdown(self) -> None: - self.exporter.shutdown() + return None def force_flush(self, timeout_millis: int = 30_000) -> bool: - return self.exporter.force_flush(timeout_millis) + return True -class _UnverifiedTlsAdapter(HTTPAdapter): - """Honour ``ssl_verify=False``: the exporter passes ``verify`` per request, which outranks ``Session.verify``.""" - - def send( # pyright: ignore[reportIncompatibleMethodOverride] # the stub types verify as bool | str, the base accepts both - self, - request: PreparedRequest, - stream: bool = False, - timeout: float | tuple[float, float] | tuple[float, None] | None = None, - verify: bool | str = True, - cert: str | tuple[str, str] | None = None, - proxies: Mapping[str, str] | None = None, - ) -> Response: - return super().send(request, stream=stream, timeout=timeout, verify=False, cert=cert, proxies=proxies) - - -def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> RetryingSpanExporter: - """Build the OTLP export channel with litellm's TLS material and v2's retry behaviour. - - v2 ingested through the injected httpx client, which carried litellm's CA - bundle and client certificate. Endpoint, headers and timeout mirror the SDK's - own span processor so the server treats the spans as v4 SDK traffic. - """ - from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter - - import litellm - from litellm.llms.custom_httpx.http_handler import get_ssl_verify - - ssl_verify: Final = get_ssl_verify() - ca_bundle: Final = ssl_verify if isinstance(ssl_verify, str) and os.path.exists(ssl_verify) else None - configured_certificate: Final = os.getenv("SSL_CERTIFICATE") or litellm.ssl_certificate - client_certificate: Final = configured_certificate if isinstance(configured_certificate, str) else None +def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) -> LangfuseSpanExporter: + """Endpoint, headers and timeout mirror the SDK's span processor so the server treats the spans as v4 SDK traffic.""" export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH") or "/api/public/otel/v1/traces" - endpoint: Final = f"{base_url.rstrip('/')}/{export_path.lstrip('/')}" encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii") - session: Final = Session() - if ssl_verify is False: - session.mount("https://", _UnverifiedTlsAdapter()) - exporter: Final = OTLPSpanExporter( - session=session, - endpoint=endpoint, - headers={ # mutable-ok: the exporter copies these into its session headers - "Authorization": "Basic " + encoded_auth, - "x-langfuse-sdk-name": "python", - "x-langfuse-sdk-version": version("langfuse"), - "x-langfuse-public-key": public_key, - }, - timeout=int(os.getenv("LANGFUSE_TIMEOUT", "5")), - certificate_file=ca_bundle, - client_certificate_file=client_certificate, + return LangfuseSpanExporter( + handler=_get_httpx_client(), + endpoint=f"{base_url.rstrip('/')}/{export_path.lstrip('/')}", + headers=MappingProxyType( + { + "Authorization": "Basic " + encoded_auth, + "Content-Type": "application/x-protobuf", + "x-langfuse-sdk-name": "python", + "x-langfuse-sdk-version": version("langfuse"), + "x-langfuse-public-key": public_key, + } + ), + timeout=float(os.getenv("LANGFUSE_TIMEOUT", "5")), ) - return RetryingSpanExporter(exporter) def _resource(*, environment: str | None, release: str | None) -> Resource: diff --git a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py index d167794b569..7e05ff0053b 100644 --- a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py +++ b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py @@ -29,8 +29,8 @@ from litellm.integrations.langfuse.langfuse import ( ) from litellm.integrations.langfuse.langfuse_sdk import ( DiscardingSpanExporter, + LangfuseSpanExporter, LangfuseTracing, - RetryingSpanExporter, _build_span_exporter, acquire_langfuse_tracing, build_langfuse_client, @@ -843,28 +843,117 @@ def test_rest_client_does_not_take_over_the_process_tracer_provider(): assert otel_trace.get_tracer_provider() is provider_before -def test_ssl_exporter_carries_litellm_tls_material(monkeypatch, tmp_path): - """The channel is litellm's own OTLP client, so litellm's CA bundle must be rebuilt onto it.""" - import litellm +def _finished_span(): + provider = TracerProvider() + span = provider.get_tracer("t").start_span("generation") + span.end() + return span - for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE", "LANGFUSE_TIMEOUT"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setattr(litellm, "ssl_verify", True) - monkeypatch.setattr(litellm, "ssl_certificate", None) - default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter - assert default._certificate_file is True - assert default._timeout == 5 - ca_path = tmp_path / "private-ca.pem" - ca_path.write_text("dummy") - monkeypatch.setattr(litellm, "ssl_verify", str(ca_path)) +def _exporter_over(responses, *, delays=(0.5, 1.5), timeout=5.0): + """A LangfuseSpanExporter whose litellm HTTPHandler talks to a scripted transport instead of the network.""" + from litellm.llms.custom_httpx.http_handler import HTTPHandler + + seen = [] + script = list(responses) + + def transport(request: httpx.Request) -> httpx.Response: + seen.append(request) + step = script.pop(0) + if isinstance(step, Exception): + raise step + return httpx.Response(step, request=request) + + handler = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(transport))) + exporter = LangfuseSpanExporter( + handler=handler, + endpoint="https://lf.internal.example/api/public/otel/v1/traces", + headers=MappingProxyType({"Authorization": "Basic cGs6c2s=", "Content-Type": "application/x-protobuf"}), + timeout=timeout, + delays=delays, + ) + return exporter, seen + + +def test_exporter_posts_the_otlp_batch_through_litellm_http_handler(monkeypatch): + """Traces travel through litellm's own handler, so litellm's TLS and proxy settings apply to them.""" + from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest + + slept = [] + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) + exporter, seen = _exporter_over([200]) + span = _finished_span() + + assert exporter.export((span,)) is SpanExportResult.SUCCESS + + (request,) = seen + assert request.method == "POST" + assert str(request.url) == "https://lf.internal.example/api/public/otel/v1/traces" + assert request.headers["Authorization"] == "Basic cGs6c2s=" + assert request.headers["Content-Type"] == "application/x-protobuf" + decoded = ExportTraceServiceRequest() + decoded.ParseFromString(request.content) + exported = decoded.resource_spans[0].scope_spans[0].spans[0] + assert exported.name == "generation" + assert exported.span_id == span.context.span_id.to_bytes(8, "big") + assert slept == [] + + +@pytest.mark.parametrize( + "failure", + [httpx.ReadTimeout("stalled"), httpx.ConnectError("refused"), 503, 429], + ids=["read-timeout", "connect-error", "http-503", "http-429"], +) +def test_exporter_retries_a_failed_round_trip_and_then_succeeds(monkeypatch, failure): + """A stalled or restarting destination used to drop the batch outright; v2 backed off and re-sent it.""" + slept = [] + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) + exporter, seen = _exporter_over([failure, failure, 200], delays=(0.5, 1.5, 2.5)) + + assert exporter.export((_finished_span(),)) is SpanExportResult.SUCCESS + assert len(seen) == 3 + assert len({request.content for request in seen}) == 1 + assert slept == [0.5, 1.5] + + +def test_exporter_gives_up_after_the_last_delay(monkeypatch): + slept = [] + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) + exporter, seen = _exporter_over([httpx.ConnectError("refused")] * 3, delays=(1.0, 2.0)) + + assert exporter.export((_finished_span(),)) is SpanExportResult.FAILURE + assert len(seen) == 3 + assert slept == [1.0, 2.0] + + +@pytest.mark.parametrize("status", [400, 401, 403, 404]) +def test_exporter_does_not_retry_a_rejected_batch(monkeypatch, status): + """Bad credentials or a bad payload will not get better on the next attempt, so retrying only delays the flush.""" + slept = [] + monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) + exporter, seen = _exporter_over([status, 200]) + + assert exporter.export((_finished_span(),)) is SpanExportResult.FAILURE + assert len(seen) == 1 + assert slept == [] + + +def test_built_exporter_uses_the_shared_litellm_handler_and_langfuse_headers(monkeypatch): + """No private requests session or TLS adapter: the channel is the same handler the rest of litellm uses.""" + from litellm.llms.custom_httpx.http_handler import _get_httpx_client + + monkeypatch.delenv("LANGFUSE_TIMEOUT", raising=False) + default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") + assert default.handler is _get_httpx_client() + assert default.timeout == 5 + monkeypatch.setenv("LANGFUSE_TIMEOUT", "20") - exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter - assert exporter._endpoint == "https://lf.internal.example/api/public/otel/v1/traces" - assert exporter._certificate_file == str(ca_path) - assert exporter._timeout == 20 - assert exporter._headers["x-langfuse-public-key"] == "pk" - assert exporter._headers["x-langfuse-sdk-version"] == installed_langfuse_version() + exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") + assert exporter.endpoint == "https://lf.internal.example/api/public/otel/v1/traces" + assert exporter.timeout == 20 + assert exporter.headers["Authorization"] == "Basic " + b64encode(b"pk:sk").decode() + assert exporter.headers["x-langfuse-public-key"] == "pk" + assert exporter.headers["x-langfuse-sdk-version"] == installed_langfuse_version() @pytest.mark.parametrize( @@ -902,122 +991,6 @@ def test_export_endpoint_never_doubles_the_slash_or_leaves_the_configured_host( else: monkeypatch.setenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH", export_path) - exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url=base_url).exporter + exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url=base_url) - assert exporter._endpoint == expected - - -def test_retrying_exporter_retries_a_raised_export_and_then_succeeds(monkeypatch): - """A read timeout used to drop the batch outright; v2 backed off and re-sent it.""" - from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult - from requests import ReadTimeout - - attempts = [] - slept = [] - - class Flaky(SpanExporter): - def export(self, spans): - attempts.append(spans) - if len(attempts) < 3: - raise ReadTimeout("destination stalled") - return SpanExportResult.SUCCESS - - monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) - result = RetryingSpanExporter(Flaky(), delays=(0.5, 1.5, 2.5)).export(("span",)) - - assert result is SpanExportResult.SUCCESS - assert attempts == [("span",)] * 3 - assert slept == [0.5, 1.5] - - -def test_retrying_exporter_gives_up_after_the_last_delay(monkeypatch): - from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult - from requests import ConnectionError as RequestsConnectionError - - attempts = [] - slept = [] - - class Down(SpanExporter): - def export(self, spans): - attempts.append(spans) - raise RequestsConnectionError("refused") - - monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append) - result = RetryingSpanExporter(Down(), delays=(1.0, 2.0)).export(("span",)) - - assert result is SpanExportResult.FAILURE - assert len(attempts) == 3 - assert slept == [1.0, 2.0] - - -@pytest.mark.parametrize("switch", ["attribute", "env"]) -def test_ssl_exporter_disables_verification_when_litellm_does(monkeypatch, switch): - """v2 exported through the httpx client, so ``ssl_verify=False`` reached ingestion; the OTLP channel must match.""" - from requests.adapters import HTTPAdapter - - import litellm - - for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setattr(litellm, "ssl_certificate", None) - if switch == "attribute": - monkeypatch.setattr(litellm, "ssl_verify", False) - else: - monkeypatch.setattr(litellm, "ssl_verify", True) - monkeypatch.setenv("SSL_VERIFY", "False") - - exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter - - sent = [] - - def send(self, request, **kwargs): - sent.append((request.url, kwargs["verify"])) - raise ConnectionError("stop before the network") - - monkeypatch.setattr(HTTPAdapter, "send", send) - with pytest.raises(ConnectionError): - exporter._export(b"payload") - assert sent[0] == ("https://lf.internal.example/api/public/otel/v1/traces", False) - - -def test_ssl_exporter_verifies_by_default(monkeypatch): - from requests.adapters import HTTPAdapter - - import litellm - - for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setattr(litellm, "ssl_certificate", None) - monkeypatch.setattr(litellm, "ssl_verify", True) - exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter - - sent = [] - - def send(self, request, **kwargs): - sent.append(kwargs["verify"]) - raise ConnectionError("stop before the network") - - monkeypatch.setattr(HTTPAdapter, "send", send) - with pytest.raises(ConnectionError): - exporter._export(b"payload") - assert sent == [True] - - -@pytest.mark.parametrize("with_client_certificate", [False, True]) -def test_ssl_exporter_falls_back_to_default_ca_when_the_bundle_path_is_missing( - monkeypatch, tmp_path, with_client_certificate -): - """The httpx client ignores a CA path that does not exist; handing it to requests would fail every export.""" - import litellm - - for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"): - monkeypatch.delenv(name, raising=False) - monkeypatch.setattr(litellm, "ssl_verify", True) - monkeypatch.setenv("SSL_VERIFY", str(tmp_path / "missing-ca.pem")) - client_cert = tmp_path / "client.pem" - client_cert.write_text("dummy") - monkeypatch.setattr(litellm, "ssl_certificate", str(client_cert) if with_client_certificate else None) - - exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter - assert exporter._certificate_file is True - assert exporter._client_cert == (str(client_cert) if with_client_certificate else None) + assert exporter.endpoint == expected