Merge branch 'main' into fix/control-flow-exception-check-in-error-handlers

Author: vishnumishra

.github/workflows/dependabot-merge.yml

@@ -15,7 +15,7 @@ jobs:
     steps:
       - name: Dependabot metadata
         id: metadata
-        uses: dependabot/fetch-metadata@ffa630c65fa7e0ecfa0625b5ceda64399aea1b36 # v3
+        uses: dependabot/fetch-metadata@ffa630c65fa7e0ecfa0625b5ceda64399aea1b36 # v3.0.0
         with:
           github-token: "${{ secrets.GITHUB_TOKEN }}"
       - name: Enable auto-merge for Dependabot PRs

.github/workflows/zizmor.yml

@@ -19,7 +19,6 @@ jobs:
     if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository
     runs-on: ubuntu-latest
     permissions:
-      security-events: write
       contents: read
     steps:
       - name: Checkout
@@ -29,4 +28,7 @@ jobs:
       - name: Run zizmor
         uses: zizmorcore/zizmor-action@b1d7e1fb5de872772f31590499237e7cce841e8e # v0.5.3
         with:
-          advanced-security: true
+          # Using false as a code scanning ruleset would block the release
+          # workflow which creates a new commit and pushes directly to main.
+          advanced-security: false
+          min-severity: medium

langfuse/_client/observe.py

@@ -290,42 +290,15 @@ async def async_wrapper(*args: Tuple[Any], **kwargs: Dict[str, Any]) -> Any:
 
                     try:
                         result = await func(*args, **kwargs)
-
-                        if capture_output is True:
-                            if inspect.isgenerator(result):
-                                is_return_type_generator = True
-
-                                return self._wrap_sync_generator_result(
-                                    langfuse_span_or_generation,
-                                    result,
-                                    transform_to_string,
-                                )
-
-                            if inspect.isasyncgen(result):
-                                is_return_type_generator = True
-
-                                return self._wrap_async_generator_result(
-                                    langfuse_span_or_generation,
-                                    result,
-                                    transform_to_string,
-                                )
-
-                            # handle starlette.StreamingResponse
-                            if type(result).__name__ == "StreamingResponse" and hasattr(
-                                result, "body_iterator"
-                            ):
-                                is_return_type_generator = True
-
-                                result.body_iterator = (
-                                    self._wrap_async_generator_result(
-                                        langfuse_span_or_generation,
-                                        result.body_iterator,
-                                        transform_to_string,
-                                    )
-                                )
-
-                            langfuse_span_or_generation.update(output=result)
-
+                        (
+                            is_return_type_generator,
+                            result,
+                        ) = self._handle_observe_result(
+                            langfuse_span_or_generation,
+                            result,
+                            capture_output=capture_output,
+                            transform_to_string=transform_to_string,
+                        )
                         return result
                     except (Exception, asyncio.CancelledError) as e:
                         langfuse_span_or_generation.update(
@@ -408,42 +381,15 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
 
                     try:
                         result = func(*args, **kwargs)
-
-                        if capture_output is True:
-                            if inspect.isgenerator(result):
-                                is_return_type_generator = True
-
-                                return self._wrap_sync_generator_result(
-                                    langfuse_span_or_generation,
-                                    result,
-                                    transform_to_string,
-                                )
-
-                            if inspect.isasyncgen(result):
-                                is_return_type_generator = True
-
-                                return self._wrap_async_generator_result(
-                                    langfuse_span_or_generation,
-                                    result,
-                                    transform_to_string,
-                                )
-
-                            # handle starlette.StreamingResponse
-                            if type(result).__name__ == "StreamingResponse" and hasattr(
-                                result, "body_iterator"
-                            ):
-                                is_return_type_generator = True
-
-                                result.body_iterator = (
-                                    self._wrap_async_generator_result(
-                                        langfuse_span_or_generation,
-                                        result.body_iterator,
-                                        transform_to_string,
-                                    )
-                                )
-
-                            langfuse_span_or_generation.update(output=result)
-
+                        (
+                            is_return_type_generator,
+                            result,
+                        ) = self._handle_observe_result(
+                            langfuse_span_or_generation,
+                            result,
+                            capture_output=capture_output,
+                            transform_to_string=transform_to_string,
+                        )
                         return result
                     except (Exception, asyncio.CancelledError) as e:
                         langfuse_span_or_generation.update(
@@ -493,6 +439,7 @@ def _wrap_sync_generator_result(
             LangfuseGuardrail,
         ],
         generator: Generator,
+        capture_output: bool,
         transform_to_string: Optional[Callable[[Iterable], str]] = None,
     ) -> Any:
         preserved_context = contextvars.copy_context()
@@ -501,6 +448,7 @@ def _wrap_sync_generator_result(
             generator,
             preserved_context,
             langfuse_span_or_generation,
+            capture_output,
             transform_to_string,
         )
 
@@ -518,6 +466,7 @@ def _wrap_async_generator_result(
             LangfuseGuardrail,
         ],
         generator: AsyncGenerator,
+        capture_output: bool,
         transform_to_string: Optional[Callable[[Iterable], str]] = None,
     ) -> Any:
         preserved_context = contextvars.copy_context()
@@ -526,9 +475,61 @@ def _wrap_async_generator_result(
             generator,
             preserved_context,
             langfuse_span_or_generation,
+            capture_output,
             transform_to_string,
         )
 
+    def _handle_observe_result(
+        self,
+        langfuse_span_or_generation: Union[
+            LangfuseSpan,
+            LangfuseGeneration,
+            LangfuseAgent,
+            LangfuseTool,
+            LangfuseChain,
+            LangfuseRetriever,
+            LangfuseEvaluator,
+            LangfuseEmbedding,
+            LangfuseGuardrail,
+        ],
+        result: Any,
+        *,
+        capture_output: bool,
+        transform_to_string: Optional[Callable[[Iterable], str]] = None,
+    ) -> Tuple[bool, Any]:
+        if inspect.isgenerator(result):
+            return True, self._wrap_sync_generator_result(
+                langfuse_span_or_generation,
+                result,
+                capture_output,
+                transform_to_string,
+            )
+
+        if inspect.isasyncgen(result):
+            return True, self._wrap_async_generator_result(
+                langfuse_span_or_generation,
+                result,
+                capture_output,
+                transform_to_string,
+            )
+
+        # handle starlette.StreamingResponse
+        if type(result).__name__ == "StreamingResponse" and hasattr(
+            result, "body_iterator"
+        ):
+            result.body_iterator = self._wrap_async_generator_result(
+                langfuse_span_or_generation,
+                result.body_iterator,
+                capture_output,
+                transform_to_string,
+            )
+            return True, result
+
+        if capture_output is True:
+            langfuse_span_or_generation.update(output=result)
+
+        return False, result
+
 
 _decorator = LangfuseDecorator()
 
@@ -553,12 +554,14 @@ def __init__(
             LangfuseEmbedding,
             LangfuseGuardrail,
         ],
+        capture_output: bool,
         transform_fn: Optional[Callable[[Iterable], str]],
     ) -> None:
         self.generator = generator
         self.context = context
         self.items: List[Any] = []
         self.span = span
+        self.capture_output = capture_output
         self.transform_fn = transform_fn
 
     def __iter__(self) -> "_ContextPreservedSyncGeneratorWrapper":
@@ -568,21 +571,25 @@ 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)
+            if self.capture_output:
+                self.items.append(item)
 
             return item
 
         except StopIteration:
             # Handle output and span cleanup when generator is exhausted
-            output: Any = self.items
+            if self.capture_output:
+                output: Any = self.items
+
+                if self.transform_fn is not None:
+                    output = self.transform_fn(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)
 
-            elif all(isinstance(item, str) for item in self.items):
-                output = "".join(self.items)
+                self.span.update(output=output)
 
-            self.span.update(output=output).end()
+            self.span.end()
 
             raise  # Re-raise StopIteration
 
@@ -612,12 +619,14 @@ def __init__(
             LangfuseEmbedding,
             LangfuseGuardrail,
         ],
+        capture_output: bool,
         transform_fn: Optional[Callable[[Iterable], str]],
     ) -> None:
         self.generator = generator
         self.context = context
         self.items: List[Any] = []
         self.span = span
+        self.capture_output = capture_output
         self.transform_fn = transform_fn
 
     def __aiter__(self) -> "_ContextPreservedAsyncGeneratorWrapper":
@@ -636,21 +645,25 @@ async def __anext__(self) -> Any:
                 # Python < 3.10 fallback - context parameter not supported
                 item = await self.generator.__anext__()
 
-            self.items.append(item)
+            if self.capture_output:
+                self.items.append(item)
 
             return item
 
         except StopAsyncIteration:
             # Handle output and span cleanup when generator is exhausted
-            output: Any = self.items
+            if self.capture_output:
+                output: Any = self.items
+
+                if self.transform_fn is not None:
+                    output = self.transform_fn(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)
 
-            elif all(isinstance(item, str) for item in self.items):
-                output = "".join(self.items)
+                self.span.update(output=output)
 
-            self.span.update(output=output).end()
+            self.span.end()
 
             raise  # Re-raise StopAsyncIteration
         except (Exception, asyncio.CancelledError) as e:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langfuse"
-version = "4.3.1"
+version = "4.4.0b1"
 description = "A client library for accessing langfuse"
 readme = "README.md"
 authors = [{ name = "langfuse", email = "developers@langfuse.com" }]