From 8a006bb32c326efb2ea3eb5a325955e69e5985e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=AF=E5=9F=BA=E9=AD=81?= <1412414664@qq.com> Date: Tue, 23 Jun 2026 11:36:24 +0800 Subject: [PATCH 1/3] fix: merge indexed streaming deltas on first chunk --- src/openai/lib/streaming/_assistants.py | 24 ++++++++--- src/openai/lib/streaming/_deltas.py | 24 ++++++++--- tests/lib/test_streaming_deltas.py | 56 +++++++++++++++++++++++++ 3 files changed, 94 insertions(+), 10 deletions(-) create mode 100644 tests/lib/test_streaming_deltas.py diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 6efb3ca3f1..bb32989141 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -977,16 +977,28 @@ def accumulate_event( return current_message_snapshot, new_content +def _has_indexed_entries(value: object) -> bool: + return is_list(value) and any(is_dict(entry) and "index" in entry for entry in value) + + def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: for key, delta_value in delta.items(): if key not in acc: - acc[key] = delta_value - continue + if _has_indexed_entries(delta_value): + acc[key] = [] + else: + acc[key] = delta_value + continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value - continue + if _has_indexed_entries(delta_value): + acc_value = [] + else: + acc[key] = delta_value + continue + else: + acc_value = acc[key] # the `index` property is used in arrays of objects so it should # not be accumulated like other values e.g. @@ -1007,7 +1019,9 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> elif is_list(acc_value) and is_list(delta_value): # for lists of non-dictionary items we'll only ever get new entries # in the array, existing entries will never be changed - if all(isinstance(x, (str, int, float)) for x in acc_value): + if all(isinstance(x, (str, int, float)) for x in acc_value) and all( + isinstance(x, (str, int, float)) for x in delta_value + ): acc_value.extend(delta_value) continue diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..b291b0016d 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -3,16 +3,28 @@ from ..._utils import is_dict, is_list +def _has_indexed_entries(value: object) -> bool: + return is_list(value) and any(is_dict(entry) and "index" in entry for entry in value) + + def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: for key, delta_value in delta.items(): if key not in acc: - acc[key] = delta_value - continue + if _has_indexed_entries(delta_value): + acc[key] = [] + else: + acc[key] = delta_value + continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value - continue + if _has_indexed_entries(delta_value): + acc_value = [] + else: + acc[key] = delta_value + continue + else: + acc_value = acc[key] # the `index` property is used in arrays of objects so it should # not be accumulated like other values e.g. @@ -33,7 +45,9 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> elif is_list(acc_value) and is_list(delta_value): # for lists of non-dictionary items we'll only ever get new entries # in the array, existing entries will never be changed - if all(isinstance(x, (str, int, float)) for x in acc_value): + if all(isinstance(x, (str, int, float)) for x in acc_value) and all( + isinstance(x, (str, int, float)) for x in delta_value + ): acc_value.extend(delta_value) continue diff --git a/tests/lib/test_streaming_deltas.py b/tests/lib/test_streaming_deltas.py new file mode 100644 index 0000000000..547820a84e --- /dev/null +++ b/tests/lib/test_streaming_deltas.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from collections.abc import Callable + +import pytest + +from openai.lib.streaming._deltas import accumulate_delta +from openai.lib.streaming._assistants import accumulate_delta as accumulate_assistant_delta + + +@pytest.mark.parametrize("accumulator", [accumulate_delta, accumulate_assistant_delta]) +def test_accumulate_delta_merges_duplicate_index_entries_in_initial_list( + accumulator: Callable[[dict[object, object], dict[object, object]], dict[object, object]], +) -> None: + acc: dict[object, object] = {} + + accumulator( + acc, + { + "tool_calls": [ + { + "index": 0, + "id": "call_abc", + "function": {"name": "list_files"}, + "type": "function", + }, + { + "index": 0, + "function": {"arguments": '{"path"'}, + }, + ] + }, + ) + accumulator( + acc, + { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": ': "."}'}, + } + ] + }, + ) + + assert acc["tool_calls"] == [ + { + "index": 0, + "id": "call_abc", + "function": { + "name": "list_files", + "arguments": '{"path": "."}', + }, + "type": "function", + } + ] From f1525081a937bbfb05b1862c354136d094dca2e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=AF=E5=9F=BA=E9=AD=81?= <1412414664@qq.com> Date: Tue, 23 Jun 2026 17:14:33 +0800 Subject: [PATCH 2/3] fix initial duplicate tool call deltas --- src/openai/lib/streaming/chat/_completions.py | 12 ++++--- tests/lib/chat/test_completions_streaming.py | 32 +++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 5f072cafbd..ceca2b6f10 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -39,7 +39,7 @@ from ....types.chat import ChatCompletionChunk, ParsedChatCompletion, ChatCompletionToolUnionParam from ...._exceptions import LengthFinishReasonError, ContentFilterFinishReasonError from ....types.chat.chat_completion import ChoiceLogprobs -from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk +from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk, ChoiceDelta from ....types.chat.completion_create_params import ResponseFormat as ResponseFormatParam @@ -393,7 +393,7 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS ), ), ), - cast("dict[object, object]", choice.delta.to_dict()), + _convert_delta_to_message(choice.delta), ), ), ) @@ -415,7 +415,7 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS type_=ParsedChoiceSnapshot, value={ **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": _convert_delta_to_message(choice.delta), }, ), ) @@ -744,7 +744,7 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh for choice in chunk.choices: choices[choice.index] = { **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": _convert_delta_to_message(choice.delta), } return cast( @@ -760,6 +760,10 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh ) +def _convert_delta_to_message(delta: ChoiceDelta) -> dict[object, object]: + return accumulate_delta({}, cast("dict[object, object]", delta.to_dict())) + + def _is_valid_chat_completion_chunk_weak(sse_event: ChatCompletionChunk) -> bool: # Although the _raw_stream is always supposed to contain only objects adhering to ChatCompletionChunk schema, # this is broken by the Azure OpenAI in case of Asynchronous Filter enabled. diff --git a/tests/lib/chat/test_completions_streaming.py b/tests/lib/chat/test_completions_streaming.py index 598a41ee2b..d9a460901d 100644 --- a/tests/lib/chat/test_completions_streaming.py +++ b/tests/lib/chat/test_completions_streaming.py @@ -725,6 +725,38 @@ class GetWeatherArgs(BaseModel): ) +@pytest.mark.respx(base_url=base_url) +def test_duplicate_tool_call_index_in_initial_chunk(client: OpenAI, respx_mock: MockRouter) -> None: + listener = _make_stream_snapshot_request( + lambda c: c.chat.completions.stream( + model="gpt-4o-2024-08-06", + messages=[{"role": "user", "content": "List files"}], + tools=[ + { + "type": "function", + "function": { + "name": "list_files", + "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}, + }, + } + ], + ), + content_snapshot=snapshot( + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_abc","type":"function","function":{"name":"list_files"}},{"index":0,"function":{"arguments":"{\\"path\\""}}]}}]}\n\n' + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":": \\".\\"}"}}]}}]}\n\n' + b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n' + b"data: [DONE]\n\n" + ), + mock_client=client, + respx_mock=respx_mock, + ) + + final_choice = listener.stream.get_final_completion().choices[0] + assert final_choice.message.tool_calls is not None + assert len(final_choice.message.tool_calls) == 1 + assert final_choice.message.tool_calls[0].function.arguments == '{"path": "."}' + + @pytest.mark.respx(base_url=base_url) def test_parse_multiple_pydantic_tools(client: OpenAI, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: class GetWeatherArgs(BaseModel): From ac1c3c445261639676a69a956c1a22153ee16c7a Mon Sep 17 00:00:00 2001 From: Marcus Wood Date: Thu, 10 Sep 2026 17:02:27 +0000 Subject: [PATCH 3/3] fix(streaming): share delta accumulation and cover initialization paths --- src/openai/lib/streaming/_assistants.py | 78 +--------- src/openai/lib/streaming/_deltas.py | 18 +-- src/openai/lib/streaming/chat/_completions.py | 12 +- tests/lib/chat/test_completions_streaming.py | 32 ---- tests/lib/chat/test_tool_call_deltas.py | 139 ++++++++++++++++++ tests/lib/test_assistant_tool_call_deltas.py | 94 ++++++++++++ tests/lib/test_streaming_deltas.py | 91 +++++++----- 7 files changed, 302 insertions(+), 162 deletions(-) create mode 100644 tests/lib/chat/test_tool_call_deltas.py create mode 100644 tests/lib/test_assistant_tool_call_deltas.py diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index cf94e17fa8..1cdca954a5 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -5,7 +5,8 @@ from typing import TYPE_CHECKING, Any, Generic, TypeVar, Callable, Iterable, Iterator, cast from typing_extensions import Awaitable, AsyncIterable, AsyncIterator, assert_never -from ..._utils import is_dict, is_list, consume_sync_iterator, consume_async_iterator +from ._deltas import accumulate_delta as accumulate_delta +from ..._utils import consume_sync_iterator, consume_async_iterator from ..._compat import model_dump from ..._httpx2 import timeout_exceptions from ..._models import construct_type @@ -978,78 +979,3 @@ def accumulate_event( ) return current_message_snapshot, new_content - - -def _has_indexed_entries(value: object) -> bool: - return is_list(value) and any(is_dict(entry) and "index" in entry for entry in value) - - -def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: - for key, delta_value in delta.items(): - if key not in acc: - if _has_indexed_entries(delta_value): - acc[key] = [] - else: - acc[key] = delta_value - continue - - acc_value = acc[key] - if acc_value is None: - if _has_indexed_entries(delta_value): - acc_value = [] - else: - acc[key] = delta_value - continue - else: - acc_value = acc[key] - - # the `index` property is used in arrays of objects so it should - # not be accumulated like other values e.g. - # [{'foo': 'bar', 'index': 0}] - # - # the same applies to `type` properties as they're used for - # discriminated unions - if key == "index" or key == "type": - acc[key] = delta_value - continue - - if isinstance(acc_value, str) and isinstance(delta_value, str): - acc_value += delta_value - elif isinstance(acc_value, (int, float)) and isinstance(delta_value, (int, float)): - acc_value += delta_value - elif is_dict(acc_value) and is_dict(delta_value): - acc_value = accumulate_delta(acc_value, delta_value) - elif is_list(acc_value) and is_list(delta_value): - # for lists of non-dictionary items we'll only ever get new entries - # in the array, existing entries will never be changed - if all(isinstance(x, (str, int, float)) for x in acc_value) and all( - isinstance(x, (str, int, float)) for x in delta_value - ): - acc_value.extend(delta_value) - continue - - for delta_entry in delta_value: - if not is_dict(delta_entry): - raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}") - - try: - index = delta_entry["index"] - except KeyError as exc: - raise RuntimeError(f"Expected list delta entry to have an `index` key; {delta_entry}") from exc - - if not isinstance(index, int): - raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {index}") - - try: - acc_entry = acc_value[index] - except IndexError: - acc_value.insert(index, delta_entry) - else: - if not is_dict(acc_entry): - raise TypeError("not handled yet") - - acc_value[index] = accumulate_delta(acc_entry, delta_entry) - - acc[key] = acc_value - - return acc diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index b291b0016d..dc80325eb4 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -9,22 +9,16 @@ def _has_indexed_entries(value: object) -> bool: def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> dict[object, object]: for key, delta_value in delta.items(): - if key not in acc: + if key not in acc or acc[key] is None: if _has_indexed_entries(delta_value): + # Even the first indexed list must be merged: one chunk can + # contain several fragments for the same index. acc[key] = [] else: acc[key] = delta_value continue acc_value = acc[key] - if acc_value is None: - if _has_indexed_entries(delta_value): - acc_value = [] - else: - acc[key] = delta_value - continue - else: - acc_value = acc[key] # the `index` property is used in arrays of objects so it should # not be accumulated like other values e.g. @@ -45,8 +39,10 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> elif is_list(acc_value) and is_list(delta_value): # for lists of non-dictionary items we'll only ever get new entries # in the array, existing entries will never be changed - if all(isinstance(x, (str, int, float)) for x in acc_value) and all( - isinstance(x, (str, int, float)) for x in delta_value + # An empty accumulator cannot tell us whether this is an indexed + # list. Inspect the delta before taking the append-only path. + if all(isinstance(x, (str, int, float)) for x in acc_value) and ( + acc_value or not _has_indexed_entries(delta_value) ): acc_value.extend(delta_value) continue diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 8bab9e95cc..cf3ecfe213 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -39,7 +39,7 @@ from ....types.chat import ChatCompletionChunk, ParsedChatCompletion, ChatCompletionToolUnionParam from ...._exceptions import LengthFinishReasonError, ContentFilterFinishReasonError from ....types.chat.chat_completion import ChoiceLogprobs -from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk, ChoiceDelta +from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk from ....types.chat.completion_create_params import ResponseFormat as ResponseFormatParam @@ -393,7 +393,7 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS ), ), ), - _convert_delta_to_message(choice.delta), + cast("dict[object, object]", choice.delta.to_dict()), ), ), ) @@ -415,7 +415,7 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS type_=ParsedChoiceSnapshot, value={ **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": _convert_delta_to_message(choice.delta), + "message": accumulate_delta({}, cast("dict[object, object]", choice.delta.to_dict())), }, ), ) @@ -745,7 +745,7 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh for choice in chunk.choices: choices[choice.index] = { **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": _convert_delta_to_message(choice.delta), + "message": accumulate_delta({}, cast("dict[object, object]", choice.delta.to_dict())), } return cast( @@ -761,10 +761,6 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh ) -def _convert_delta_to_message(delta: ChoiceDelta) -> dict[object, object]: - return accumulate_delta({}, cast("dict[object, object]", delta.to_dict())) - - def _is_valid_chat_completion_chunk_weak(sse_event: ChatCompletionChunk) -> bool: # Although the _raw_stream is always supposed to contain only objects adhering to ChatCompletionChunk schema, # this is broken by the Azure OpenAI in case of Asynchronous Filter enabled. diff --git a/tests/lib/chat/test_completions_streaming.py b/tests/lib/chat/test_completions_streaming.py index 9525480474..25d2942639 100644 --- a/tests/lib/chat/test_completions_streaming.py +++ b/tests/lib/chat/test_completions_streaming.py @@ -727,38 +727,6 @@ class GetWeatherArgs(BaseModel): ) -@pytest.mark.respx2(base_url=base_url) -def test_duplicate_tool_call_index_in_initial_chunk(client: OpenAI, respx2_mock: MockRouter) -> None: - listener = _make_stream_snapshot_request( - lambda c: c.chat.completions.stream( - model="gpt-4o-2024-08-06", - messages=[{"role": "user", "content": "List files"}], - tools=[ - { - "type": "function", - "function": { - "name": "list_files", - "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}, - }, - } - ], - ), - content_snapshot=snapshot( - b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_abc","type":"function","function":{"name":"list_files"}},{"index":0,"function":{"arguments":"{\\"path\\""}}]}}]}\n\n' - b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":": \\".\\"}"}}]}}]}\n\n' - b'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1727346161,"model":"gpt-4o-2024-08-06","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n' - b"data: [DONE]\n\n" - ), - mock_client=client, - respx2_mock=respx2_mock, - ) - - final_choice = listener.stream.get_final_completion().choices[0] - assert final_choice.message.tool_calls is not None - assert len(final_choice.message.tool_calls) == 1 - assert final_choice.message.tool_calls[0].function.arguments == '{"path": "."}' - - @pytest.mark.respx2(base_url=base_url) def test_parse_multiple_pydantic_tools( client: OpenAI, respx2_mock: MockRouter, monkeypatch: pytest.MonkeyPatch diff --git a/tests/lib/chat/test_tool_call_deltas.py b/tests/lib/chat/test_tool_call_deltas.py new file mode 100644 index 0000000000..73478ed942 --- /dev/null +++ b/tests/lib/chat/test_tool_call_deltas.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import json +from typing import Any + +import httpx2 +import pytest +from pydantic import BaseModel + +import openai +from openai import OpenAI, AsyncOpenAI +from tests.respx2 import MockRouter +from openai.types.chat import ParsedChatCompletion +from openai.lib.streaming.chat import ChatCompletionStreamEvent + +from ...conftest import base_url + + +class ListFilesArgs(BaseModel): + path: str + + +@pytest.fixture(params=["initial", "after_role", "empty", "new_choice", "existing_tool", "parallel"]) +def tool_call_chunks(request: pytest.FixtureRequest) -> list[dict[str, Any]]: + scenario = request.param + chunks: list[dict[str, Any]] = [] + choice_index = 1 if scenario == "new_choice" else 0 + tool_index = 1 if scenario in ("existing_tool", "parallel") else 0 + + def chunk(delta: dict[str, Any], index: int = choice_index, finish: str | None = None) -> dict[str, Any]: + return { + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "created": 0, + "model": "gpt-4o-mini", + "choices": [{"index": index, "delta": delta, "finish_reason": finish}], + } + + if scenario in ("after_role", "empty", "new_choice"): + chunks.append(chunk({"role": "assistant", **({"tool_calls": []} if scenario == "empty" else {})}, index=0)) + + other_tool = {"index": 0, "id": "call_other", "type": "function", "function": {"name": "other", "arguments": "{}"}} + if scenario == "existing_tool": + chunks.append(chunk({"role": "assistant", "tool_calls": [other_tool]})) + + # Match the report: the name and first argument fragment share an index + # within one chunk. Also split a later argument delta into two entries. + first_tools = [ + {"index": tool_index, "id": "call_files", "type": "function", "function": {"name": "list_files"}}, + {"index": tool_index, "function": {"arguments": '{"path"'}}, + ] + if scenario == "parallel": + first_tools.insert(0, other_tool) + chunks.extend( + [ + chunk({"role": "assistant", "tool_calls": first_tools}), + chunk( + { + "tool_calls": [ + {"index": tool_index, "function": {"arguments": ": "}}, + {"index": tool_index, "function": {"arguments": '"."}'}}, + ] + } + ), + chunk({}, finish="tool_calls"), + ] + ) + return chunks + + +def mock_tool_stream(router: MockRouter, chunks: list[dict[str, Any]]) -> None: + body = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + "data: [DONE]\n\n" + router.post("/chat/completions").mock( + return_value=httpx2.Response(200, text=body, headers={"content-type": "text/event-stream"}) + ) + + +def assert_tool_stream( + completion: ParsedChatCompletion[None], + events: list[ChatCompletionStreamEvent[None]], + chunks: list[dict[str, Any]], +) -> None: + calls = completion.choices[-1].message.tool_calls + assert calls is not None + expected_count = ( + 2 + if any( + tool.get("id") == "call_other" + for chunk in chunks + for choice in chunk["choices"] + for tool in choice["delta"].get("tool_calls", []) + ) + else 1 + ) + assert len(calls) == expected_count + assert calls[-1].id == "call_files" + assert calls[-1].function.name == "list_files" + assert calls[-1].function.arguments == '{"path": "."}' + assert calls[-1].function.parsed_arguments == ListFilesArgs(path=".") + if expected_count == 2: + assert calls[0].id == "call_other" + assert calls[0].function.arguments == "{}" + + deltas = [ + event for event in events if event.type == "tool_calls.function.arguments.delta" and event.name == "list_files" + ] + assert [event.arguments_delta for event in deltas] == ["", '{"path"', ": ", '"."}'] + done = [ + event for event in events if event.type == "tool_calls.function.arguments.done" and event.name == "list_files" + ] + assert len(done) == 1 + assert done[0].arguments == '{"path": "."}' + assert done[0].parsed_arguments == ListFilesArgs(path=".") + # Snapshot normalization must not change the raw chunks exposed to callers. + assert [event.chunk.to_dict() for event in events if event.type == "chunk"] == chunks + + +@pytest.mark.respx2(base_url=base_url) +def test_duplicate_tool_call_deltas( + client: OpenAI, respx2_mock: MockRouter, tool_call_chunks: list[dict[str, Any]] +) -> None: + mock_tool_stream(respx2_mock, tool_call_chunks) + with client.chat.completions.stream( + model="gpt-4o-mini", messages=[], tools=[openai.pydantic_function_tool(ListFilesArgs, name="list_files")] + ) as stream: + events = list(stream) + assert_tool_stream(stream.get_final_completion(), events, tool_call_chunks) + + +@pytest.mark.respx2(base_url=base_url) +async def test_async_duplicate_tool_call_deltas( + async_client: AsyncOpenAI, respx2_mock: MockRouter, tool_call_chunks: list[dict[str, Any]] +) -> None: + mock_tool_stream(respx2_mock, tool_call_chunks) + async with async_client.chat.completions.stream( + model="gpt-4o-mini", messages=[], tools=[openai.pydantic_function_tool(ListFilesArgs, name="list_files")] + ) as stream: + events = [event async for event in stream] + assert_tool_stream(await stream.get_final_completion(), events, tool_call_chunks) diff --git a/tests/lib/test_assistant_tool_call_deltas.py b/tests/lib/test_assistant_tool_call_deltas.py new file mode 100644 index 0000000000..647bfd381b --- /dev/null +++ b/tests/lib/test_assistant_tool_call_deltas.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import json + +import httpx2 +import pytest + +from openai import OpenAI, AsyncOpenAI +from tests.respx2 import MockRouter +from openai.types.beta.threads.runs import RunStep + +from ..conftest import base_url + + +@pytest.fixture(params=[False, True], ids=["empty", "snapshot-without-index"]) +def assistant_tool_stream(request: pytest.FixtureRequest, respx2_mock: MockRouter) -> None: + tool = {"id": "call_files", "type": "function", "function": {"name": "list_files", "arguments": ""}} + step = { + "id": "step_test", + "object": "thread.run.step", + "assistant_id": "asst_test", + "thread_id": "thread_test", + "run_id": "run_test", + "created_at": 0, + "status": "in_progress", + "type": "tool_calls", + "step_details": {"type": "tool_calls", "tool_calls": [tool] if request.param else []}, + } + fragments = [ + {"index": 0, "type": "function", "function": {"arguments": '{"'}}, + {"index": 0, "type": "function", "function": {"arguments": 'path"'}}, + ] + if not request.param: + fragments.insert(0, {"index": 0, **tool}) + events = [ + ("thread.run.step.created", step), + ( + "thread.run.step.delta", + { + "id": "step_test", + "object": "thread.run.step.delta", + "delta": { + "step_details": {"type": "tool_calls", "tool_calls": fragments}, + }, + }, + ), + ( + "thread.run.step.delta", + { + "id": "step_test", + "object": "thread.run.step.delta", + "delta": { + "step_details": { + "type": "tool_calls", + "tool_calls": [ + {"index": 0, "type": "function", "function": {"arguments": ': "."}'}}, + ], + }, + }, + }, + ), + ] + body = "".join(f"event: {name}\ndata: {json.dumps(data)}\n\n" for name, data in events) + "data: [DONE]\n\n" + respx2_mock.post("/threads/thread_test/runs").mock( + return_value=httpx2.Response(200, text=body, headers={"content-type": "text/event-stream"}) + ) + + +def assert_tool_arguments(steps: list[RunStep]) -> None: + assert len(steps) == 1 + details = steps[0].step_details + assert details.type == "tool_calls" + assert len(details.tool_calls) == 1 + tool = details.tool_calls[0] + assert tool.type == "function" + assert tool.id == "call_files" + assert tool.function.name == "list_files" + assert json.loads(tool.function.arguments) == {"path": "."} + + +@pytest.mark.respx2(base_url=base_url) +@pytest.mark.usefixtures("assistant_tool_stream") +@pytest.mark.filterwarnings("ignore:The Assistants API is deprecated:DeprecationWarning") +def test_assistant_tool_call_deltas(client: OpenAI) -> None: + with client.beta.threads.runs.stream(thread_id="thread_test", assistant_id="asst_test") as stream: # pyright: ignore[reportDeprecated] + assert_tool_arguments(stream.get_final_run_steps()) + + +@pytest.mark.respx2(base_url=base_url) +@pytest.mark.usefixtures("assistant_tool_stream") +@pytest.mark.filterwarnings("ignore:The Assistants API is deprecated:DeprecationWarning") +async def test_async_assistant_tool_call_deltas(async_client: AsyncOpenAI) -> None: + async with async_client.beta.threads.runs.stream(thread_id="thread_test", assistant_id="asst_test") as stream: # pyright: ignore[reportDeprecated] + assert_tool_arguments(await stream.get_final_run_steps()) diff --git a/tests/lib/test_streaming_deltas.py b/tests/lib/test_streaming_deltas.py index 547820a84e..e563968436 100644 --- a/tests/lib/test_streaming_deltas.py +++ b/tests/lib/test_streaming_deltas.py @@ -1,56 +1,77 @@ from __future__ import annotations -from collections.abc import Callable +from copy import deepcopy import pytest from openai.lib.streaming._deltas import accumulate_delta -from openai.lib.streaming._assistants import accumulate_delta as accumulate_assistant_delta -@pytest.mark.parametrize("accumulator", [accumulate_delta, accumulate_assistant_delta]) -def test_accumulate_delta_merges_duplicate_index_entries_in_initial_list( - accumulator: Callable[[dict[object, object], dict[object, object]], dict[object, object]], -) -> None: - acc: dict[object, object] = {} - - accumulator( - acc, - { - "tool_calls": [ - { - "index": 0, - "id": "call_abc", - "function": {"name": "list_files"}, - "type": "function", - }, - { - "index": 0, - "function": {"arguments": '{"path"'}, - }, - ] - }, - ) - accumulator( +@pytest.mark.parametrize("initial", [{}, {"tool_calls": None}, {"tool_calls": []}]) +def test_duplicate_indexes_in_initial_list(initial: dict[object, object]) -> None: + acc = deepcopy(initial) + accumulate_delta( acc, { "tool_calls": [ - { - "index": 0, - "function": {"arguments": ': "."}'}, - } + {"index": 0, "id": "call_abc", "type": "function", "function": {"name": "list_files"}}, + {"index": 0, "function": {"arguments": '{"path"'}}, ] }, ) - + accumulate_delta(acc, {"tool_calls": [{"index": 0, "function": {"arguments": ': "."}'}}]}) assert acc["tool_calls"] == [ { "index": 0, "id": "call_abc", - "function": { - "name": "list_files", - "arguments": '{"path": "."}', - }, "type": "function", + "function": {"name": "list_files", "arguments": '{"path": "."}'}, } ] + + +@pytest.mark.parametrize( + "initial,delta,expected", + [ + ({}, {"value": [1, "a"]}, {"value": [1, "a"]}), + ({"value": None}, {"value": [1, "a"]}, {"value": [1, "a"]}), + ({"value": []}, {"value": [1, "a"]}, {"value": [1, "a"]}), + ({"value": [1]}, {"value": [2, "a"]}, {"value": [1, 2, "a"]}), + ({"value": []}, {"value": [{"text": "a"}]}, {"value": [{"text": "a"}]}), + ({}, {"value": [{"text": "a"}]}, {"value": [{"text": "a"}]}), + ({"value": [1]}, {"value": []}, {"value": [1]}), + # Full Assistants snapshots omit the delta-only index field. + ( + {"value": [{"text": "a"}]}, + {"value": [{"index": 0, "text": "b"}]}, + {"value": [{"index": 0, "text": "ab"}]}, + ), + ( + {"index": 1, "type": "text", "text": {"value": "a"}, "count": 2}, + {"index": 1, "type": "text", "text": {"value": "b"}, "count": 3}, + {"index": 1, "type": "text", "text": {"value": "ab"}, "count": 5}, + ), + ( + {"function": {"name": "list_", "arguments": "{"}}, + {"function": {"name": "files", "arguments": "}"}}, + {"function": {"name": "list_files", "arguments": "{}"}}, + ), + ], +) +def test_existing_delta_semantics( + initial: dict[object, object], delta: dict[object, object], expected: dict[object, object] +) -> None: + assert accumulate_delta(deepcopy(initial), deepcopy(delta)) == expected + + +@pytest.mark.parametrize( + "entry,error,match", + [ + ("invalid", TypeError, "not a dictionary"), + ({"text": "invalid"}, RuntimeError, "an `index` key"), + ({"index": "0"}, TypeError, "not an integer"), + ], +) +def test_invalid_indexed_delta(entry: object, error: type[Exception], match: str) -> None: + with pytest.raises(error, match=match): + accumulate_delta({"content": [{"index": 0, "text": "a"}]}, {"content": [entry]})