diff --git a/src/openai/lib/_parsing/_responses.py b/src/openai/lib/_parsing/_responses.py index c607587ec1..81e6b2b983 100644 --- a/src/openai/lib/_parsing/_responses.py +++ b/src/openai/lib/_parsing/_responses.py @@ -58,7 +58,7 @@ def parse_response( ) -> ParsedResponse[TextFormatT]: output_list: List[ParsedResponseOutputItem[TextFormatT]] = [] - for output in response.output: + for output in response.output or []: if output.type == "message": content_list: List[ParsedContent[TextFormatT]] = [] for item in output.content: diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py index 6975a9260d..735238493e 100644 --- a/src/openai/lib/streaming/responses/_responses.py +++ b/src/openai/lib/streaming/responses/_responses.py @@ -15,6 +15,7 @@ ) from ...._types import Omit, omit from ...._utils import is_given, consume_sync_iterator, consume_async_iterator +from ...._compat import model_copy from ...._models import build, construct_type_unchecked from ...._streaming import Stream, AsyncStream from ....types.responses import ParsedResponse, ResponseStreamEvent as RawResponseStreamEvent @@ -25,6 +26,7 @@ ParsedResponseOutputMessage, ParsedResponseFunctionToolCall, ) +from ....types.responses.response_output_item import ResponseOutputItem class ResponseStream(Generic[TextFormatT]): @@ -240,6 +242,7 @@ def __init__( ) -> None: self.__current_snapshot: ParsedResponseSnapshot | None = None self._completed_response: ParsedResponse[TextFormatT] | None = None + self._completed_output: dict[int, ResponseOutputItem] = {} self._input_tools = [tool for tool in input_tools] if is_given(input_tools) else [] self._text_format = text_format self._rich_text_format: type | Omit = text_format if inspect.isclass(text_format) else omit @@ -356,10 +359,17 @@ def accumulate_event(self, event: RawResponseStreamEvent) -> ParsedResponseSnaps output = snapshot.output[event.output_index] if output.type == "function_call": output.arguments += event.delta + elif event.type == "response.output_item.done": + self._completed_output[event.output_index] = event.item elif event.type == "response.completed": + response = event.response + if getattr(response, "output", None) is None: + # Recover finalized items; deltas can omit final status, annotations, or refusals. + response = model_copy(response) + response.output = [self._completed_output[index] for index in sorted(self._completed_output)] self._completed_response = parse_response( text_format=self._text_format, - response=event.response, + response=response, input_tools=self._input_tools, ) diff --git a/tests/lib/responses/test_null_output.py b/tests/lib/responses/test_null_output.py new file mode 100644 index 0000000000..4c782de03a --- /dev/null +++ b/tests/lib/responses/test_null_output.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import json + +import httpx2 +import pytest +from pydantic import BaseModel + +from openai import OpenAI, AsyncOpenAI +from openai.types.responses import ToolParam + + +class Answer(BaseModel): + answer: int + + +@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"]) +@pytest.mark.parametrize("terminal_output", ["null", "missing", "empty", "present"]) +@pytest.mark.parametrize("has_items", [True, False], ids=["with-items", "without-items"]) +async def test_stream_recovers_finalized_output(sync: bool, terminal_output: str, has_items: bool) -> None: + items: list[dict[str, object]] = ( + [ + { + "id": "msg_test", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": '{"answer": 4}', + "annotations": [ + { + "type": "url_citation", + "url": "https://example.com", + "title": "Example", + "start_index": 0, + "end_index": 1, + } + ], + }, + {"type": "refusal", "refusal": "Final refusal"}, + ], + }, + { + "id": "fc_test", + "type": "function_call", + "call_id": "call_test", + "name": "lookup", + "arguments": '{"answer": 4}', + "status": "completed", + }, + ] + if has_items + else [] + ) + response: dict[str, object] = {"id": "resp_test", "status": "in_progress", "output": []} + events: list[dict[str, object]] = [{"type": "response.created", "response": response}] + for index, item in enumerate(items): + added = {**item, "status": "in_progress"} + if item["type"] == "message": + added["content"] = [] + else: + added["arguments"] = "" + events.append({"type": "response.output_item.added", "output_index": index, "item": added}) + # Final items carry data absent from the live snapshot. Preserve output_index order. + for index in reversed(range(len(items))): + events.append({"type": "response.output_item.done", "output_index": index, "item": items[index]}) + completed: dict[str, object] = { + "id": "resp_test", + "status": "completed", + "usage": {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3}, + } + if terminal_output != "missing": + # A supplied list must win, even when it differs from the streamed items. + completed["output"] = {"null": None, "empty": [], "present": items[:1]}[terminal_output] + events.append({"type": "response.completed", "response": completed}) + body = "".join( + f"data: {json.dumps({**event, 'sequence_number': index})}\n\n" for index, event in enumerate(events) + ).encode() + transport = httpx2.MockTransport( + lambda _request: httpx2.Response(200, content=body, headers={"content-type": "text/event-stream"}) + ) + tools: list[ToolParam] = [{"type": "function", "name": "lookup", "parameters": {}, "strict": True}] + if sync: + with OpenAI(api_key="fake-test-key", http_client=httpx2.Client(transport=transport)) as client: + with client.responses.stream(model="test-model", input="test", text_format=Answer, tools=tools) as stream: + emitted = list(stream) + final = stream.get_final_response() + else: + async with AsyncOpenAI( + api_key="fake-test-key", http_client=httpx2.AsyncClient(transport=transport) + ) as async_client: + async with async_client.responses.stream( + model="test-model", input="test", text_format=Answer, tools=tools + ) as async_stream: + emitted = [event async for event in async_stream] + final = await async_stream.get_final_response() + + last_event = emitted[-1] + assert last_event.type == "response.completed" + assert last_event.response == final + assert final.id == "resp_test" + assert final.status == "completed" + assert final.usage is not None and final.usage.total_tokens == 3 + if not has_items or terminal_output == "empty": + assert final.output == [] + return + + assert len(final.output) == (1 if terminal_output == "present" else 2) + message = final.output[0] + assert message.type == "message" and message.status == "completed" + assert message.id == "msg_test" + assert final.output_parsed == Answer(answer=4) + text = message.content[0] + assert text.type == "output_text" + citation = text.annotations[0] + assert citation.type == "url_citation" and citation.url == "https://example.com" + refusal = message.content[1] + assert refusal.type == "refusal" and refusal.refusal == "Final refusal" + if terminal_output != "present": + tool = final.output[1] + assert tool.type == "function_call" and tool.status == "completed" + assert tool.id == "fc_test" + assert tool.parsed_arguments == {"answer": 4} diff --git a/tests/lib/responses/test_responses.py b/tests/lib/responses/test_responses.py index c970c53a77..390681e8fb 100644 --- a/tests/lib/responses/test_responses.py +++ b/tests/lib/responses/test_responses.py @@ -103,3 +103,11 @@ def test_parse_method_definition_in_sync(sync: bool, client: OpenAI, async_clien checking_client.responses.parse, exclude_params={"tools"}, ) + + +def test_parse_response_with_null_output() -> None: + response = construct_type_unchecked(type_=Response, value={"output": None}) + + parsed = parse_response(text_format=omit, input_tools=omit, response=response) + + assert parsed.output == []