-m rest of the implementation

Author: pakrym-oai

examples/responses/background.py

@@ -0,0 +1,46 @@
+from typing import List
+
+import rich
+from pydantic import BaseModel
+
+from openai import OpenAI
+
+
+class Step(BaseModel):
+    explanation: str
+    output: str
+
+
+class MathResponse(BaseModel):
+    steps: List[Step]
+    final_answer: str
+
+
+client = OpenAI()
+id = None
+
+with client.responses.create(
+    input="solve 8x + 31 = 2",
+    model="gpt-4o-2024-08-06",
+    background=True,
+    stream=True,
+) as stream:
+    for event in stream:
+        if event.type == "response.created":
+            id = event.response.id
+        if "output_text" in event.type:
+            rich.print(event)
+        if event.sequence_number == 10:
+            break
+
+print("Interrupted. Continuing...")
+
+assert id is not None
+with client.responses.retrieve(
+    response_id=id,
+    stream=True,
+    starting_after=10,
+) as stream:
+    for event in stream:
+        if "output_text" in event.type:
+            rich.print(event)

examples/responses/background_async.py

@@ -0,0 +1,52 @@
+import asyncio
+from typing import List
+
+import rich
+from pydantic import BaseModel
+
+from openai._client import AsyncOpenAI
+
+
+class Step(BaseModel):
+    explanation: str
+    output: str
+
+
+class MathResponse(BaseModel):
+    steps: List[Step]
+    final_answer: str
+
+
+async def main() -> None:
+    client = AsyncOpenAI()
+    id = None
+
+    async with await client.responses.create(
+        input="solve 8x + 31 = 2",
+        model="gpt-4o-2024-08-06",
+        background=True,
+        stream=True,
+    ) as stream:
+        async for event in stream:
+            if event.type == "response.created":
+                id = event.response.id
+            if "output_text" in event.type:
+                rich.print(event)
+            if event.sequence_number == 10:
+                break
+
+    print("Interrupted. Continuing...")
+
+    assert id is not None
+    async with await client.responses.retrieve(
+        response_id=id,
+        stream=True,
+        starting_after=10,
+    ) as stream:
+        async for event in stream:
+            if "output_text" in event.type:
+                rich.print(event)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/responses/background_streaming.py

@@ -32,6 +32,8 @@ class MathResponse(BaseModel):
         if event.sequence_number == 10:
             break
 
+print("Interrupted. Continuing...")
+
 assert id is not None
 with client.responses.stream(
     response_id=id,

examples/responses/background_streaming_async.py

@@ -0,0 +1,53 @@
+import asyncio
+from typing import List
+
+import rich
+from pydantic import BaseModel
+
+from openai import AsyncOpenAI
+
+
+class Step(BaseModel):
+    explanation: str
+    output: str
+
+
+class MathResponse(BaseModel):
+    steps: List[Step]
+    final_answer: str
+
+
+async def main() -> None:
+    client = AsyncOpenAI()
+    id = None
+    async with client.responses.stream(
+        input="solve 8x + 31 = 2",
+        model="gpt-4o-2024-08-06",
+        text_format=MathResponse,
+        background=True,
+    ) as stream:
+        async for event in stream:
+            if event.type == "response.created":
+                id = event.response.id
+            if "output_text" in event.type:
+                rich.print(event)
+            if event.sequence_number == 10:
+                break
+
+    print("Interrupted. Continuing...")
+
+    assert id is not None
+    async with client.responses.stream(
+        response_id=id,
+        starting_after=10,
+        text_format=MathResponse,
+    ) as stream:
+        async for event in stream:
+            if "output_text" in event.type:
+                rich.print(event)
+
+        rich.print(stream.get_final_response())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/openai/resources/responses/responses.py

@@ -870,19 +870,17 @@ def stream(
             if not is_given(response_id):
                 raise ValueError("id must be provided when streaming an existing response")
 
-            # TODO: get rid of the cast
-            r = self.retrieve(
-                response_id,
-                stream=True,
-                include=include or [],
-                extra_headers=extra_headers,
-                extra_query=extra_query,
-                extra_body=extra_body,
-                starting_after=NOT_GIVEN,
-                timeout=timeout,
-            )
             return ResponseStreamManager(
-                lambda: cast(Stream[ResponseStreamEvent], r),
+                lambda: self.retrieve(
+                    response_id=response_id,
+                    stream=True,
+                    include=include or [],
+                    extra_headers=extra_headers,
+                    extra_query=extra_query,
+                    extra_body=extra_body,
+                    starting_after=NOT_GIVEN,
+                    timeout=timeout,
+                ),
                 text_format=text_format,
                 input_tools=tools,
                 starting_after=starting_after if is_given(starting_after) else None,
@@ -1000,7 +998,7 @@ def retrieve(
         extra_query: Query | None = None,
         extra_body: Body | None = None,
         timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
-    ) -> Response | Stream[ResponseStreamEvent]: ...
+    ) -> Stream[ResponseStreamEvent]: ...
 
     @overload
     def retrieve(
@@ -1039,6 +1037,16 @@ def retrieve(
           include: Additional fields to include in the response. See the `include` parameter for
               Response creation above for more information.
 
+          stream: If set to true, the model response data will be streamed to the client using
+              [server-sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Event_stream_format).
+              See the
+              [Streaming section below](https://platform.openai.com/docs/api-reference/responses-streaming)
+              for more information.
+
+          starting_after: When retrieving a background response, this parameter can be used to start
+            replaying after an event with the given sequence number. Must be used in conjunction with
+            the `stream` parameter set to `true`.
+
           extra_headers: Send extra headers
 
           extra_query: Add additional query parameters to the request
@@ -1919,7 +1927,7 @@ def stream(
 
         tools = _make_tools(tools)
         if len(new_response_args_names) > 0:
-            if not is_given(input):
+            if isinstance(input, NotGiven):
                 raise ValueError("input must be provided when creating a new response")
 
             if not is_given(model):
@@ -1966,7 +1974,7 @@ def stream(
                 starting_after=None,
             )
         else:
-            if not is_given(response_id):
+            if isinstance(response_id, NotGiven):
                 raise ValueError("response_id must be provided when streaming an existing response")
 
             api_request = self.retrieve(
@@ -2097,7 +2105,7 @@ async def retrieve(
         extra_query: Query | None = None,
         extra_body: Body | None = None,
         timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
-    ) -> Response | AsyncStream[ResponseStreamEvent]: ...
+    ) -> AsyncStream[ResponseStreamEvent]: ...
 
     @overload
     async def retrieve(
@@ -2136,6 +2144,10 @@ async def retrieve(
           include: Additional fields to include in the response. See the `include` parameter for
               Response creation above for more information.
 
+          stream:
+          starting_after: When retrieving a background response, this parameter can be used to start
+          replaying after an event with the given sequence number. Must be used in
+
           extra_headers: Send extra headers
 
           extra_query: Add additional query parameters to the request