fix(observe): avoid buffering uncaptured stream items

Author: hassiebp

langfuse/_client/observe.py

@@ -571,7 +571,8 @@ 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
 
@@ -644,7 +645,8 @@ 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
 

tests/unit/test_observe.py

@@ -30,6 +30,7 @@ def body():
     assert memory_exporter.get_finished_spans() == []
 
     assert list(generator) == ["item_0", "item_1"]
+    assert generator.items == []
 
     langfuse_memory_client.flush()
 
@@ -73,6 +74,7 @@ async def endpoint():
     assert memory_exporter.get_finished_spans() == []
 
     assert [item async for item in response.body_iterator] == ["chunk_0", "chunk_1"]
+    assert response.body_iterator.items == []
 
     langfuse_memory_client.flush()