diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..dc23b85faf 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -6,6 +6,12 @@ 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: + # When the first chunk contains a list with multiple entries at the + # same index (e.g. from speculative decoding), storing it directly + # would leave duplicate entries that later merges can't fix. (#3201) + # Coalesce duplicate-index entries before storing. + if is_list(delta_value) and len(delta_value) > 1: + delta_value = _coalesce_list_by_index(delta_value) acc[key] = delta_value continue @@ -49,16 +55,54 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> 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") + # Merge by logical index, not physical position. (#3201) + # When the first chunk contains multiple entries with the same + # index (e.g. from speculative decoding), the physical position + # does not match the logical index. Find the existing entry by + # its index field and merge into it. + found = False + for i, existing in enumerate(acc_value): + if is_dict(existing) and existing.get("index") == index: + acc_value[i] = accumulate_delta(existing, delta_entry) + found = True + break - acc_value[index] = accumulate_delta(acc_entry, delta_entry) + if not found: + # Ensure the list is large enough + while len(acc_value) <= index: + acc_value.append({}) + acc_value[index] = delta_entry acc[key] = acc_value return acc + + +def _coalesce_list_by_index(lst: list[object]) -> list[object]: + """Merge list entries that share the same ``index`` field into a single entry. + + When the first streamed chunk contains multiple entries with the same + ``index`` (e.g. from speculative decoding), storing the list directly would + leave duplicate entries. This function coalesces them by merging entries + with the same index using :func:`accumulate_delta`, so the snapshot starts + in a clean state. (#3201) + """ + result: list[object] = [] + for entry in lst: + if not is_dict(entry): + result.append(entry) + continue + index = entry.get("index") + if not isinstance(index, int): + result.append(entry) + continue + # Find an existing entry with the same index + found = False + for i, existing in enumerate(result): + if is_dict(existing) and existing.get("index") == index: + result[i] = accumulate_delta(existing, entry) + found = True + break + if not found: + result.append(entry) + return result diff --git a/tests/lib/streaming/test_deltas.py b/tests/lib/streaming/test_deltas.py new file mode 100644 index 0000000000..1d613a374f --- /dev/null +++ b/tests/lib/streaming/test_deltas.py @@ -0,0 +1,116 @@ +"""Tests for the streaming delta accumulator.""" + +from __future__ import annotations + +from openai.lib.streaming._deltas import accumulate_delta + + +class TestAccumulateDelta: + """Tests for accumulate_delta — regression for #3201.""" + + def test_duplicate_index_first_chunk_merges(self) -> None: + """First chunk with two entries at the same index should merge into one.""" + acc: dict[object, object] = {} + delta = { + "tool_calls": [ + { + "index": 0, + "id": "call_abc", + "function": {"name": "list_files"}, + "type": "function", + }, + { + "index": 0, + "function": {"arguments": ' {"'}, + }, + ] + } + result = accumulate_delta(acc, delta) + calls = result["tool_calls"] + assert isinstance(calls, list) + # Should be a single entry at index 0, not two + assert len(calls) == 1 + assert calls[0]["index"] == 0 + assert calls[0]["id"] == "call_abc" + assert calls[0]["function"]["name"] == "list_files" + assert calls[0]["function"]["arguments"] == ' {"' + + def test_duplicate_index_subsequent_chunk_merges(self) -> None: + """Subsequent chunk with same index should merge into existing entry.""" + acc: dict[object, object] = { + "tool_calls": [ + { + "index": 0, + "id": "call_abc", + "function": {"name": "list_files", "arguments": ' {"'}, + "type": "function", + } + ] + } + delta = { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": 'path": "."}'}, + } + ] + } + result = accumulate_delta(acc, delta) + calls = result["tool_calls"] + assert len(calls) == 1 + assert calls[0]["function"]["arguments"] == ' {"path": "."}' + + def test_different_indexes_accumulate_separately(self) -> None: + """Entries with different indexes should accumulate separately.""" + acc: dict[object, object] = {} + delta1 = { + "tool_calls": [ + {"index": 0, "id": "call_a", "function": {"name": "tool_a"}, "type": "function"}, + ] + } + delta2 = { + "tool_calls": [ + {"index": 1, "id": "call_b", "function": {"name": "tool_b"}, "type": "function"}, + ] + } + result = accumulate_delta(acc, delta1) + result = accumulate_delta(result, delta2) + calls = result["tool_calls"] + assert len(calls) == 2 + assert calls[0]["index"] == 0 + assert calls[1]["index"] == 1 + + def test_string_accumulation_unchanged(self) -> None: + """Basic string accumulation should still work.""" + acc: dict[object, object] = {"content": "hello"} + delta = {"content": " world"} + result = accumulate_delta(acc, delta) + assert result["content"] == "hello world" + + def test_duplicate_index_first_chunk_then_subsequent_merge(self) -> None: + """Full round-trip: first chunk with duplicate indexes, then subsequent chunk merges correctly.""" + acc: dict[object, object] = {} + # First chunk: two entries at index 0 + delta1 = { + "tool_calls": [ + {"index": 0, "id": "call_abc", "function": {"name": "list_files"}, "type": "function"}, + {"index": 0, "function": {"arguments": ' {"'}}, + ] + } + result = accumulate_delta(acc, delta1) + calls = result["tool_calls"] + assert len(calls) == 1, f"Expected 1 entry after coalescing, got {len(calls)}" + assert calls[0]["function"]["arguments"] == ' {"' + + # Second chunk: more arguments for index 0 + delta2 = { + "tool_calls": [ + {"index": 0, "function": {"arguments": 'path": "."}'}}, + ] + } + result = accumulate_delta(result, delta2) + calls = result["tool_calls"] + assert len(calls) == 1 + assert calls[0]["function"]["arguments"] == ' {"path": "."}' + assert calls[0]["id"] == "call_abc" + assert calls[0]["function"]["name"] == "list_files" \ No newline at end of file