feat(otel): cap default exporter batches at 64 MiB

langfuse/_client/client.py

@@ -268,7 +268,7 @@ def mask_otel_spans(
         additional_headers (Optional[Dict[str, str]]): Additional headers to include in all API requests and in the default OTLPSpanExporter requests. These headers will be merged with default headers. Note: If httpx_client is provided, additional_headers must be set directly on your custom httpx_client as well. If `span_exporter` is provided, these headers are not wired into that exporter and must be configured on the exporter instance directly.
         tracer_provider(Optional[TracerProvider]): OpenTelemetry TracerProvider to use for Langfuse. This can be useful to set to have disconnected tracing between Langfuse and other OpenTelemetry-span emitting libraries. Note: To track active spans, the context is still shared between TracerProviders. This may lead to broken trace trees.
         id_generator (Optional[IdGenerator]): OpenTelemetry ID generator to use when Langfuse creates its own TracerProvider. If omitted, the OpenTelemetry SDK default is used. If `tracer_provider` is provided, or an OpenTelemetry TracerProvider is already registered globally, configure the ID generator on that provider instead.
-        span_exporter (Optional[SpanExporter]): Custom OpenTelemetry span exporter for the Langfuse span processor. If omitted, Langfuse creates an OTLPSpanExporter pointed at the Langfuse OTLP endpoint. If provided, Langfuse does not wire `base_url`, exporter headers, exporter auth, or exporter timeout into it. Configure endpoint, headers, and timeout on the exporter instance directly. If you are sending spans to Langfuse v4 or using Langfuse Cloud Fast Preview, include `x-langfuse-ingestion-version=4` on the exporter to enable real time processing of exported spans.
+        span_exporter (Optional[SpanExporter]): Custom OpenTelemetry span exporter for the Langfuse span processor. If omitted, Langfuse creates an OTLPSpanExporter pointed at the Langfuse OTLP endpoint. If provided, Langfuse does not wire `base_url`, exporter headers, exporter auth, exporter timeout, or the `LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES` request size limit into it. Configure endpoint, headers, and timeout on the exporter instance directly. If you are sending spans to Langfuse v4 or using Langfuse Cloud Fast Preview, include `x-langfuse-ingestion-version=4` on the exporter to enable real time processing of exported spans.
 
     Example:
         ```python

langfuse/_client/environment_variables.py

@@ -62,6 +62,17 @@
 **Default value:** ``/api/public/otel/v1/traces``
 """
 
+LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES = "LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES"
+"""
+.. envvar:: LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES
+
+Maximum serialized size in bytes (before compression) of a span batch exported by the
+default OTLP exporter. Oversized batches are dropped before sending and logged as a
+failed export. Must be a positive integer. Custom span exporters are not affected.
+
+**Default value:** ``67108864`` (64 MiB)
+"""
+
 LANGFUSE_DEBUG = "LANGFUSE_DEBUG"
 """
 .. envvar:: LANGFUSE_DEBUG

langfuse/_client/span_exporter.py

@@ -9,14 +9,14 @@
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 from opentelemetry.sdk.util import BoundedList
 from opentelemetry.trace import format_span_id, format_trace_id
-from opentelemetry.util.types import AttributeValue
 
 from langfuse._client.attributes import LangfuseOtelSpanAttributes
 from langfuse._task_manager.media_manager import MediaManager
 from langfuse._utils.serializer import EventSerializer
 from langfuse.logger import langfuse_logger
 from langfuse.media import LangfuseMedia
 from langfuse.types import (
+    AttributeValue,
     MaskOtelSpansFunction,
     MaskOtelSpansParams,
     MaskOtelSpansResult,

langfuse/_client/span_processor.py

@@ -28,6 +28,7 @@
 from langfuse._client.environment_variables import (
     LANGFUSE_FLUSH_AT,
     LANGFUSE_FLUSH_INTERVAL,
+    LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
     LANGFUSE_OTEL_TRACES_EXPORT_PATH,
 )
 from langfuse._client.propagation import (
@@ -47,6 +48,23 @@
 from langfuse.types import MaskOtelSpansFunction
 
 
+def _resolve_max_batch_size_bytes() -> Optional[int]:
+    """Return the configured batch byte limit, or None to keep the exporter default."""
+    raw_value = os.environ.get(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, "").strip()
+    if not raw_value:
+        return None
+
+    if raw_value.isascii() and raw_value.isdigit() and int(raw_value) > 0:
+        return int(raw_value)
+
+    langfuse_logger.warning(
+        "Invalid %s=%r. Expected a positive integer. Using the default limit.",
+        LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
+        raw_value,
+    )
+    return None
+
+
 class LangfuseSpanProcessor(BatchSpanProcessor):
     """OpenTelemetry span processor that exports spans to the Langfuse API.
 
@@ -127,6 +145,7 @@ def __init__(
                 endpoint=endpoint,
                 headers=headers,
                 timeout=timeout,
+                max_request_size=_resolve_max_batch_size_bytes(),
             )
 
         if media_manager is not None or mask_otel_spans is not None:

langfuse/types.py

@@ -28,10 +28,9 @@ def my_evaluator(*, output: str, **kwargs) -> Evaluation:
     Protocol,
     Sequence,
     TypedDict,
+    Union,
 )
 
-from opentelemetry.util.types import AttributeValue
-
 try:
     from typing import NotRequired  # type: ignore
 except ImportError:
@@ -40,6 +39,20 @@ def my_evaluator(*, output: str, **kwargs) -> Evaluation:
 
 from langfuse.api import MediaContentType
 
+# Span attribute values accepted by the OpenTelemetry trace API. OpenTelemetry 1.45
+# widened ``opentelemetry.util.types.AttributeValue`` to the recursive ``AnyValue``,
+# which mypy cannot use as a type alias.
+AttributeValue = Union[
+    str,
+    bool,
+    int,
+    float,
+    Sequence[str],
+    Sequence[bool],
+    Sequence[int],
+    Sequence[float],
+]
+
 SpanLevel = Literal["DEBUG", "DEFAULT", "WARNING", "ERROR"]
 
 ScoreDataType = Literal["NUMERIC", "CATEGORICAL", "BOOLEAN", "TEXT", "CORRECTION"]

pyproject.toml

@@ -29,9 +29,9 @@ dependencies = [
     "backoff>=1.10.0",
     "wrapt>=1.14,<3",
     "packaging>=23.2,<27.0",
-    "opentelemetry-api>=1.33.1,<2",
-    "opentelemetry-sdk>=1.33.1,<2",
-    "opentelemetry-exporter-otlp-proto-http>=1.33.1,<2",
+    "opentelemetry-api>=1.45.0,<2",
+    "opentelemetry-sdk>=1.45.0,<2",
+    "opentelemetry-exporter-otlp-proto-http>=1.45.0,<2",
     "typing-extensions>=4.12.2,<5",
 ]
 
@@ -71,6 +71,8 @@ build-backend = "uv_build"
 [tool.uv]
 # Basic protection against supply chain attacks
 exclude-newer = "7 days"
+# OpenTelemetry 1.45 ships the OTLP request size cap the SDK relies on
+exclude-newer-package = { "opentelemetry-api" = "3 days", "opentelemetry-sdk" = "3 days", "opentelemetry-semantic-conventions" = "3 days", "opentelemetry-proto" = "3 days", "opentelemetry-exporter-otlp-proto-common" = "3 days", "opentelemetry-exporter-otlp-proto-http" = "3 days", "opentelemetry-exporter-otlp-common" = "3 days", "opentelemetry-exporter-http-transport" = "3 days", "opentelemetry-instrumentation" = "3 days", "opentelemetry-instrumentation-threading" = "3 days" }
 
 [tool.uv.build-backend]
 module-root = ""

tests/live_provider/test_langchain.py

@@ -180,7 +180,7 @@ def test_callback_simple_openai():
 
         # Initialize components
         handler = CallbackHandler()
-        llm = OpenAI()
+        llm = ChatOpenAI(model="gpt-4o-mini")
         text = (
             "What would be a good company name for a company that makes colorful socks?"
         )
@@ -222,7 +222,7 @@ def test_callback_multiple_invocations_on_different_traces():
         test_name_2 = f"Test Multiple Invocations 2 {create_uuid()}"
 
         # Setup components
-        llm = OpenAI()
+        llm = ChatOpenAI(model="gpt-4o-mini")
         text = (
             "What would be a good company name for a company that makes colorful socks?"
         )
@@ -257,6 +257,10 @@ def test_callback_multiple_invocations_on_different_traces():
         assert generation.output != ""
 
 
+@pytest.mark.skip(
+    reason="OpenAI retired gpt-3.5-turbo-instruct, the legacy completions model "
+    "this test depends on"
+)
 def test_openai_instruct_usage():
     langfuse = Langfuse()
 

tests/live_provider/test_openai.py

@@ -10,6 +10,11 @@
 
 langfuse: Langfuse | None = None
 
+requires_legacy_completion_model = pytest.mark.skip(
+    reason="OpenAI retired gpt-3.5-turbo-instruct, the legacy completions model "
+    "this test depends on"
+)
+
 
 @pytest.fixture(scope="module")
 def openai():
@@ -405,6 +410,7 @@ def test_openai_chat_completion_with_seed(openai):
     assert len(completion.choices) != 0
 
 
+@requires_legacy_completion_model
 def test_openai_completion(openai):
     generation_name = create_uuid()
     completion = openai.OpenAI().completions.create(
@@ -445,6 +451,7 @@ def test_openai_completion(openai):
     assert generation.data[0].output == "2\n\n1 + 2 = 3\n\n2 + 3 = "
 
 
+@requires_legacy_completion_model
 def test_openai_completion_stream(openai):
     generation_name = create_uuid()
     completion = openai.OpenAI().completions.create(
@@ -588,6 +595,7 @@ def test_openai_completion_stream_fail(openai):
     openai.api_key = os.environ["OPENAI_API_KEY"]
 
 
+@requires_legacy_completion_model
 def test_openai_completion_with_langfuse_prompt(openai):
     generation_name = create_uuid()
     langfuse = Langfuse()

tests/unit/test_additional_headers_simple.py

@@ -180,16 +180,16 @@ def test_span_processor_has_additional_headers_in_otel_exporter(self):
         exporter = processor.span_exporter
 
         # Verify additional headers are in the exporter's headers
-        assert exporter._headers["X-Custom-Trace-Header"] == "trace-value"
-        assert exporter._headers["X-Override-Default"] == "override-value"
+        assert exporter._client._headers["x-custom-trace-header"] == "trace-value"
+        assert exporter._client._headers["x-override-default"] == "override-value"
 
         # Verify default headers are still present
-        assert "Authorization" in exporter._headers
-        assert "x-langfuse-sdk-name" in exporter._headers
-        assert "x-langfuse-public-key" in exporter._headers
+        assert "authorization" in exporter._client._headers
+        assert "x-langfuse-sdk-name" in exporter._client._headers
+        assert "x-langfuse-public-key" in exporter._client._headers
 
         # Check that our override worked
-        assert exporter._headers["X-Override-Default"] == "override-value"
+        assert exporter._client._headers["x-override-default"] == "override-value"
 
     def test_span_processor_none_additional_headers_works(self):
         """Test that span processor works with None additional headers."""
@@ -207,9 +207,9 @@ def test_span_processor_none_additional_headers_works(self):
         exporter = processor.span_exporter
 
         # Verify default headers are present
-        assert "Authorization" in exporter._headers
-        assert "x-langfuse-sdk-name" in exporter._headers
-        assert "x-langfuse-public-key" in exporter._headers
+        assert "authorization" in exporter._client._headers
+        assert "x-langfuse-sdk-name" in exporter._client._headers
+        assert "x-langfuse-public-key" in exporter._client._headers
 
     def test_span_processor_uses_custom_span_exporter_when_provided(self):
         """Test that a custom exporter bypasses the default OTLP exporter construction."""