fix(openai): don't consume iterator arguments before calling OpenAI

langfuse/openai.py

@@ -27,6 +27,7 @@
 import os
 import types
 from collections import defaultdict
+from collections.abc import Iterator
 from dataclasses import dataclass
 from datetime import datetime
 from inspect import isawaitable, isclass
@@ -311,7 +312,12 @@ def __init__(
         self.args["trace_id"] = trace_id
         self.args["parent_observation_id"] = parent_observation_id
 
-        self.kwargs = kwargs
+        # Materialize one-shot iterators (e.g. a generator of messages) so that
+        # reading them for tracing does not leave OpenAI with an empty argument.
+        self.kwargs = {
+            key: list(value) if isinstance(value, Iterator) else value
+            for key, value in kwargs.items()
+        }
 
     def get_langfuse_args(self) -> Any:
         return {**self.args, **self.kwargs}

tests/unit/test_openai.py

@@ -1,8 +1,10 @@
 import asyncio
 import base64
+import json
 from types import SimpleNamespace
 from unittest.mock import patch
 
+import httpx
 import pytest
 from openai.types.chat import ChatCompletion
 from openai.types.responses import ParsedResponseOutputMessage, ParsedResponseOutputText
@@ -372,6 +374,43 @@ def test_chat_completion_exports_generation_span(
     }
 
 
+def test_chat_completion_does_not_consume_messages_iterator(
+    langfuse_memory_client, get_span, json_attr
+):
+    openai_client = lf_openai.OpenAI(api_key="test")
+    response = ChatCompletion.model_validate(
+        {
+            "id": "chatcmpl-1",
+            "object": "chat.completion",
+            "created": 1,
+            "model": "gpt-4o-mini",
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {"role": "assistant", "content": "hi"},
+                    "finish_reason": "stop",
+                }
+            ],
+        }
+    )
+    messages = [{"role": "user", "content": "hello"}]
+
+    with patch.object(
+        openai_client.chat.completions, "_post", return_value=response
+    ) as post:
+        openai_client.chat.completions.create(
+            name="unit-openai-iterator",
+            model="gpt-4o-mini",
+            messages=(message for message in messages),
+        )
+
+    assert post.call_args.kwargs["body"]["messages"] == messages
+
+    langfuse_memory_client.flush()
+    span = get_span("unit-openai-iterator")
+    assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_INPUT) == messages
+
+
 def test_chat_completion_with_none_choices_does_not_crash(
     langfuse_memory_client, get_span, json_attr
 ):
@@ -1531,3 +1570,44 @@ def test_with_raw_response_streaming_passes_through_untraced(
         span.name != "OpenAI-generation"
         for span in memory_exporter.get_finished_spans()
     )
+
+
+def test_responses_create_does_not_consume_input_iterator(
+    langfuse_memory_client, get_span, json_attr
+):
+    sent = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        sent.append(json.loads(request.content)["input"])
+        return httpx.Response(
+            200,
+            json={
+                "id": "resp_1",
+                "object": "response",
+                "created_at": 1,
+                "model": "gpt-4o-mini",
+                "status": "completed",
+                "output": [],
+                "parallel_tool_calls": False,
+                "tool_choice": "auto",
+                "tools": [],
+            },
+        )
+
+    openai_client = lf_openai.OpenAI(
+        api_key="test",
+        http_client=httpx.Client(transport=httpx.MockTransport(handler)),
+    )
+    messages = [{"role": "user", "content": "hello"}]
+
+    openai_client.responses.create(
+        name="unit-responses-iterator",
+        model="gpt-4o-mini",
+        input=(message for message in messages),
+    )
+
+    assert sent == [messages]
+
+    langfuse_memory_client.flush()
+    span = get_span("unit-responses-iterator")
+    assert json_attr(span, LangfuseOtelSpanAttributes.OBSERVATION_INPUT) == messages