Merge branch 'main' into main

Author: nimarb

langfuse/_client/client.py

@@ -2730,6 +2730,7 @@ async def process_item(item: ExperimentItem) -> ExperimentItemResult:
                         comment=evaluation.comment,
                         metadata=evaluation.metadata,
                         data_type=evaluation.data_type,  # type: ignore
+                        config_id=evaluation.config_id,
                     )
 
             except Exception as e:
@@ -2856,9 +2857,11 @@ async def _process_experiment_item(
                             self.create_score(
                                 trace_id=trace_id,
                                 name=evaluation.name,
-                                value=evaluation.value or -1,
+                                value=evaluation.value,  # type: ignore
                                 comment=evaluation.comment,
                                 metadata=evaluation.metadata,
+                                config_id=evaluation.config_id,
+                                data_type=evaluation.data_type,  # type: ignore
                             )
 
                     except Exception as e:

langfuse/_client/observe.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import inspect
 import logging
 import os
@@ -10,6 +11,7 @@
     Dict,
     Generator,
     Iterable,
+    List,
     Optional,
     Tuple,
     TypeVar,
@@ -21,25 +23,24 @@
 from opentelemetry.util._decorator import _AgnosticContextManager
 from typing_extensions import ParamSpec
 
-from langfuse._client.environment_variables import (
-    LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED,
-)
-
 from langfuse._client.constants import (
     ObservationTypeLiteralNoEvent,
     get_observation_types_list,
 )
+from langfuse._client.environment_variables import (
+    LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED,
+)
 from langfuse._client.get_client import _set_current_public_key, get_client
 from langfuse._client.span import (
-    LangfuseGeneration,
-    LangfuseSpan,
     LangfuseAgent,
-    LangfuseTool,
     LangfuseChain,
-    LangfuseRetriever,
-    LangfuseEvaluator,
     LangfuseEmbedding,
+    LangfuseEvaluator,
+    LangfuseGeneration,
     LangfuseGuardrail,
+    LangfuseRetriever,
+    LangfuseSpan,
+    LangfuseTool,
 )
 from langfuse.types import TraceContext
 
@@ -468,29 +469,54 @@ def _wrap_sync_generator_result(
         generator: Generator,
         transform_to_string: Optional[Callable[[Iterable], str]] = None,
     ) -> Any:
-        items = []
+        preserved_context = contextvars.copy_context()
 
-        try:
-            for item in generator:
-                items.append(item)
+        return _ContextPreservedSyncGeneratorWrapper(
+            generator,
+            preserved_context,
+            langfuse_span_or_generation,
+            transform_to_string,
+        )
+
+    def _wrap_async_generator_result(
+        self,
+        langfuse_span_or_generation: Union[
+            LangfuseSpan,
+            LangfuseGeneration,
+            LangfuseAgent,
+            LangfuseTool,
+            LangfuseChain,
+            LangfuseRetriever,
+            LangfuseEvaluator,
+            LangfuseEmbedding,
+            LangfuseGuardrail,
+        ],
+        generator: AsyncGenerator,
+        transform_to_string: Optional[Callable[[Iterable], str]] = None,
+    ) -> Any:
+        preserved_context = contextvars.copy_context()
 
-                yield item
+        return _ContextPreservedAsyncGeneratorWrapper(
+            generator,
+            preserved_context,
+            langfuse_span_or_generation,
+            transform_to_string,
+        )
 
-        finally:
-            output: Any = items
 
-            if transform_to_string is not None:
-                output = transform_to_string(items)
+_decorator = LangfuseDecorator()
+
+observe = _decorator.observe
 
-            elif all(isinstance(item, str) for item in items):
-                output = "".join(items)
 
-            langfuse_span_or_generation.update(output=output)
-            langfuse_span_or_generation.end()
+class _ContextPreservedSyncGeneratorWrapper:
+    """Sync generator wrapper that ensures each iteration runs in preserved context."""
 
-    async def _wrap_async_generator_result(
+    def __init__(
         self,
-        langfuse_span_or_generation: Union[
+        generator: Generator,
+        context: contextvars.Context,
+        span: Union[
             LangfuseSpan,
             LangfuseGeneration,
             LangfuseAgent,
@@ -501,30 +527,105 @@ async def _wrap_async_generator_result(
             LangfuseEmbedding,
             LangfuseGuardrail,
         ],
-        generator: AsyncGenerator,
-        transform_to_string: Optional[Callable[[Iterable], str]] = None,
-    ) -> AsyncGenerator:
-        items = []
+        transform_fn: Optional[Callable[[Iterable], str]],
+    ) -> None:
+        self.generator = generator
+        self.context = context
+        self.items: List[Any] = []
+        self.span = span
+        self.transform_fn = transform_fn
+
+    def __iter__(self) -> "_ContextPreservedSyncGeneratorWrapper":
+        return self
+
+    def __next__(self) -> Any:
+        try:
+            # Run the generator's __next__ in the preserved context
+            item = self.context.run(next, self.generator)
+            self.items.append(item)
+
+            return item
+
+        except StopIteration:
+            # Handle output and span cleanup when generator is exhausted
+            output: Any = self.items
+
+            if self.transform_fn is not None:
+                output = self.transform_fn(self.items)
+
+            elif all(isinstance(item, str) for item in self.items):
+                output = "".join(self.items)
+
+            self.span.update(output=output).end()
+
+            raise  # Re-raise StopIteration
+
+        except Exception as e:
+            self.span.update(level="ERROR", status_message=str(e)).end()
 
+            raise
+
+
+class _ContextPreservedAsyncGeneratorWrapper:
+    """Async generator wrapper that ensures each iteration runs in preserved context."""
+
+    def __init__(
+        self,
+        generator: AsyncGenerator,
+        context: contextvars.Context,
+        span: Union[
+            LangfuseSpan,
+            LangfuseGeneration,
+            LangfuseAgent,
+            LangfuseTool,
+            LangfuseChain,
+            LangfuseRetriever,
+            LangfuseEvaluator,
+            LangfuseEmbedding,
+            LangfuseGuardrail,
+        ],
+        transform_fn: Optional[Callable[[Iterable], str]],
+    ) -> None:
+        self.generator = generator
+        self.context = context
+        self.items: List[Any] = []
+        self.span = span
+        self.transform_fn = transform_fn
+
+    def __aiter__(self) -> "_ContextPreservedAsyncGeneratorWrapper":
+        return self
+
+    async def __anext__(self) -> Any:
         try:
-            async for item in generator:
-                items.append(item)
+            # Run the generator's __anext__ in the preserved context
+            try:
+                # Python 3.10+ approach with context parameter
+                item = await asyncio.create_task(
+                    self.generator.__anext__(),  # type: ignore
+                    context=self.context,
+                )  # type: ignore
+            except TypeError:
+                # Python < 3.10 fallback - context parameter not supported
+                item = await self.generator.__anext__()
 
-                yield item
+            self.items.append(item)
 
-        finally:
-            output: Any = items
+            return item
 
-            if transform_to_string is not None:
-                output = transform_to_string(items)
+        except StopAsyncIteration:
+            # Handle output and span cleanup when generator is exhausted
+            output: Any = self.items
 
-            elif all(isinstance(item, str) for item in items):
-                output = "".join(items)
+            if self.transform_fn is not None:
+                output = self.transform_fn(self.items)
 
-            langfuse_span_or_generation.update(output=output)
-            langfuse_span_or_generation.end()
+            elif all(isinstance(item, str) for item in self.items):
+                output = "".join(self.items)
 
+            self.span.update(output=output).end()
 
-_decorator = LangfuseDecorator()
+            raise  # Re-raise StopAsyncIteration
+        except Exception as e:
+            self.span.update(level="ERROR", status_message=str(e)).end()
 
-observe = _decorator.observe
+            raise

langfuse/_client/span.py

@@ -31,6 +31,7 @@
 
 from opentelemetry import trace as otel_trace_api
 from opentelemetry.util._decorator import _AgnosticContextManager
+from opentelemetry.trace.status import Status, StatusCode
 
 from langfuse.model import PromptClient
 
@@ -188,6 +189,8 @@ def __init__(
             self._otel_span.set_attributes(
                 {k: v for k, v in attributes.items() if v is not None}
             )
+            # Set OTEL span status if level is ERROR
+            self._set_otel_span_status_if_error(level=level, status_message=status_message)
 
     def end(self, *, end_time: Optional[int] = None) -> "LangfuseObservationWrapper":
         """End the span, marking it as completed.
@@ -540,6 +543,28 @@ def _process_media_in_attribute(
 
         return data
 
+    def _set_otel_span_status_if_error(
+            self, *, level: Optional[SpanLevel] = None, status_message: Optional[str] = None
+    ) -> None:
+        """Set OpenTelemetry span status to ERROR if level is ERROR.
+
+        This method sets the underlying OpenTelemetry span status to ERROR when the
+        Langfuse observation level is set to ERROR, ensuring consistency between
+        Langfuse and OpenTelemetry error states.
+
+        Args:
+            level: The span level to check
+            status_message: Optional status message to include as description
+        """
+        if level == "ERROR" and self._otel_span.is_recording():
+            try:
+                self._otel_span.set_status(
+                    Status(StatusCode.ERROR, description=status_message)
+                )
+            except Exception:
+                # Silently ignore any errors when setting OTEL status to avoid existing flow disruptions
+                pass
+
     def update(
         self,
         *,
@@ -636,6 +661,8 @@ def update(
             )
 
         self._otel_span.set_attributes(attributes=attributes)
+        # Set OTEL span status if level is ERROR
+        self._set_otel_span_status_if_error(level=level, status_message=status_message)
 
         return self
 

langfuse/api/__init__.py

@@ -200,6 +200,7 @@
     UpdateGenerationBody,
     UpdateGenerationEvent,
     UpdateObservationEvent,
+    UpdateScoreConfigRequest,
     UpdateSpanBody,
     UpdateSpanEvent,
     UpsertLlmConnectionRequest,
@@ -433,6 +434,7 @@
     "UpdateGenerationBody",
     "UpdateGenerationEvent",
     "UpdateObservationEvent",
+    "UpdateScoreConfigRequest",
     "UpdateSpanBody",
     "UpdateSpanEvent",
     "UpsertLlmConnectionRequest",