feat(tracing)!: send observation metadata as one JSON attribute

langfuse/_client/attributes.py

@@ -11,18 +11,32 @@
 """
 
 import json
+import math
 from datetime import datetime
-from typing import Any, Dict, Literal, Optional, Union
+from typing import Any, Dict, Literal, Optional, Tuple, TypeVar, Union
 
 from langfuse._client.constants import (
     ObservationTypeGenerationLike,
     ObservationTypeSpanLike,
 )
 from langfuse._utils.serializer import EventSerializer
 from langfuse.api import MapValue
+from langfuse.logger import langfuse_logger
 from langfuse.model import PromptClient
 from langfuse.types import SpanLevel
 
+MAX_OBSERVATION_METADATA_KEYS = 128
+"""Maximum number of top-level keys allowed in observation metadata."""
+
+
+class ObservationMetadataKeyLimitError(ValueError):
+    """Raised when observation metadata has more than 128 top-level keys."""
+
+
+_FAILED_TO_SERIALIZE = "<failed to serialize>"
+_T = TypeVar("_T")
+_COMPACT_SEPARATORS = (",", ":")
+
 
 class LangfuseOtelSpanAttributes:
     # Langfuse-Trace attributes
@@ -107,7 +121,9 @@ def create_span_attributes(
         LangfuseOtelSpanAttributes.VERSION: version,
         LangfuseOtelSpanAttributes.OBSERVATION_INPUT: _serialize(input),
         LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT: _serialize(output),
-        **_flatten_and_serialize_metadata(metadata, "observation"),
+        LangfuseOtelSpanAttributes.OBSERVATION_METADATA: merge_observation_metadata(
+            None, serialize_observation_metadata(metadata)
+        )[0],
     }
 
     return {k: v for k, v in attributes.items() if v is not None}
@@ -152,7 +168,9 @@ def create_generation_attributes(
         LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS: _serialize(
             model_parameters
         ),
-        **_flatten_and_serialize_metadata(metadata, "observation"),
+        LangfuseOtelSpanAttributes.OBSERVATION_METADATA: merge_observation_metadata(
+            None, serialize_observation_metadata(metadata)
+        )[0],
     }
 
     return {k: v for k, v in attributes.items() if v is not None}
@@ -191,25 +209,122 @@ def flatten_value(path: str, value: Any) -> None:
     return flattened_metadata
 
 
-def _flatten_and_serialize_metadata(
-    metadata: Any, type: Literal["observation", "trace"]
-) -> dict:
-    prefix = (
-        LangfuseOtelSpanAttributes.OBSERVATION_METADATA
-        if type == "observation"
-        else LangfuseOtelSpanAttributes.TRACE_METADATA
-    )
+def serialize_observation_metadata(
+    metadata: Any,
+) -> Union[None, str, Dict[str, str]]:
+    """Serialize observation metadata for merge_observation_metadata.
+
+    Values of dict metadata are serialized one by one, so a value that fails to
+    serialize becomes `"<failed to serialize>"` instead of dropping all metadata.
+    Keys with `None` values are skipped.
+
+    Returns:
+        None if there is no metadata, the serialized value for non-dict
+        metadata, or the serialized values keyed by top-level key.
+    """
+    if metadata is None:
+        return None
+
+    if not isinstance(metadata, dict):
+        return _serialize_metadata_value(metadata, top_level=True)
+
+    return {
+        str(key): _serialize_metadata_value(value)
+        for key, value in metadata.items()
+        if value is not None
+    }
+
+
+def merge_observation_metadata(
+    previous: Optional[Dict[str, str]],
+    serialized: Union[None, str, Dict[str, str]],
+) -> Tuple[Optional[str], Optional[Dict[str, str]]]:
+    """Merge serialized metadata into earlier metadata of the same observation.
+
+    Observation metadata is written as one JSON object to the
+    `langfuse.observation.metadata` attribute. Top-level keys of `serialized`
+    overwrite keys in `previous`. Non-dict metadata replaces earlier metadata.
+
+    Args:
+        previous: Encoded `"key":value` JSON members from earlier updates, keyed
+            by top-level key
+        serialized: Output of serialize_observation_metadata for this update
 
-    metadata_attributes: Dict[str, Union[str, int, None]] = {}
+    Returns:
+        The attribute value to write (None if there is nothing to write) and the
+        encoded members to keep for the next update (None if there are none).
 
+    Raises:
+        ObservationMetadataKeyLimitError: If the merged metadata has more than
+            MAX_OBSERVATION_METADATA_KEYS top-level keys.
+    """
+    if serialized is None:
+        return None, previous
+
+    if isinstance(serialized, str):
+        return serialized, None
+
+    # Encode only the new members, so an update doesn't re-encode every earlier key
+    merged = {
+        **(previous or {}),
+        **{key: f"{json.dumps(key)}:{value}" for key, value in serialized.items()},
+    }
+
+    if len(merged) > MAX_OBSERVATION_METADATA_KEYS:
+        message = (
+            f"Observation metadata has {len(merged)} keys, which exceeds the "
+            f"maximum of {MAX_OBSERVATION_METADATA_KEYS}."
+        )
+        raise ObservationMetadataKeyLimitError(message)
+
+    if not merged:
+        return None, None
+
+    attribute_value = "{" + ",".join(merged.values()) + "}"
+
+    return attribute_value, merged
+
+
+def drop_metadata_over_key_limit(metadata: _T) -> Optional[_T]:
+    """Drop dict metadata with more than MAX_OBSERVATION_METADATA_KEYS keys.
+
+    For integrations: the SDK raises on too many metadata keys, but raising inside
+    instrumentation would break or lose the user's call. Integrations drop the
+    metadata with a warning instead. Keys with `None` values are not counted.
+    """
     if not isinstance(metadata, dict):
-        metadata_attributes[prefix] = _serialize(metadata)
-    else:
-        for key, value in metadata.items():
-            metadata_attributes[f"{prefix}.{key}"] = (
-                value
-                if isinstance(value, str) or isinstance(value, int)
-                else _serialize(value)
-            )
-
-    return metadata_attributes
+        return metadata
+
+    key_count = sum(1 for value in metadata.values() if value is not None)
+
+    if key_count > MAX_OBSERVATION_METADATA_KEYS:
+        langfuse_logger.warning(
+            "Dropping observation metadata: it has %s keys, which exceeds the "
+            "maximum of %s.",
+            key_count,
+            MAX_OBSERVATION_METADATA_KEYS,
+        )
+        return None
+
+    return metadata
+
+
+def _serialize_metadata_value(value: Any, *, top_level: bool = False) -> str:
+    if top_level and isinstance(value, str):
+        return value
+
+    # Plain json.dumps encodes these JSON-native scalars exactly like the
+    # EventSerializer call below, at a fraction of the cost (no encoder per call)
+    value_type = type(value)
+    if (
+        value_type is str
+        or value_type is bool
+        or (value_type is int and EventSerializer.is_js_safe_integer(value))
+        or (value_type is float and math.isfinite(value))
+    ):
+        return json.dumps(value)
+
+    try:
+        return json.dumps(value, cls=EventSerializer, separators=_COMPACT_SEPARATORS)
+    except Exception:
+        return _FAILED_TO_SERIALIZE if top_level else json.dumps(_FAILED_TO_SERIALIZE)

langfuse/_client/client.py

@@ -45,6 +45,7 @@
     LangfuseOtelSpanAttributes,
     _flatten_and_serialize_metadata_values,
     _serialize,
+    drop_metadata_over_key_limit,
 )
 from langfuse._client.constants import (
     LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT,
@@ -1434,7 +1435,10 @@ def update_current_generation(
             name: The generation name
             input: Updated input data for the model
             output: Output from the model (e.g., completions)
-            metadata: Additional metadata to associate with the generation
+            metadata: Additional metadata to associate with the generation. Merges
+                into earlier metadata by top-level key; keys set to None keep
+                their earlier values. At most 128 top-level keys are allowed,
+                more raise a ValueError.
             version: Version identifier for the model or component
             level: Importance level of the generation (info, warning, error)
             status_message: Optional status message for the generation
@@ -1474,10 +1478,8 @@ def update_current_generation(
                 otel_span=current_otel_span, langfuse_client=self
             )
 
-            if name:
-                current_otel_span.update_name(name)
-
             generation.update(
+                name=name,
                 input=input,
                 output=output,
                 metadata=metadata,
@@ -1513,7 +1515,10 @@ def update_current_span(
             name: The span name
             input: Updated input data for the operation
             output: Output data from the operation
-            metadata: Additional metadata to associate with the span
+            metadata: Additional metadata to associate with the span. Merges
+                into earlier metadata by top-level key; keys set to None keep
+                their earlier values. At most 128 top-level keys are allowed,
+                more raise a ValueError.
             version: Version identifier for the code or component
             level: Importance level of the span (info, warning, error)
             status_message: Optional status message for the span
@@ -1553,10 +1558,8 @@ def update_current_span(
                 release=self._release,
             )
 
-            if name:
-                current_otel_span.update_name(name)
-
             span.update(
+                name=name,
                 input=input,
                 output=output,
                 metadata=metadata,
@@ -2992,6 +2995,11 @@ async def _process_experiment_item(
                         {"dataset_id": dataset_id, "dataset_item_id": dataset_item_id}
                     )
 
+                # Don't fail the item when its metadata exceeds the key limit
+                item_observation_metadata = drop_metadata_over_key_limit(
+                    final_observation_metadata
+                )
+
                 experiment_item_id = (
                     dataset_item_id or get_sha256_hash_hex(_serialize(input_data))[:16]
                 )
@@ -3012,7 +3020,7 @@ async def _process_experiment_item(
                     name="experiment-item-task",
                     as_type="span",
                     input=input_data,
-                    metadata=final_observation_metadata,
+                    metadata=item_observation_metadata,
                 ) as task_span:
                     task_span._otel_span.set_attributes(experiment_span_attributes)
 
@@ -3081,7 +3089,7 @@ async def _process_experiment_item(
                 span.update(
                     input=input_data,
                     output=output,
-                    metadata=final_observation_metadata,
+                    metadata=item_observation_metadata,
                 )
 
             except Exception as e: