diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 6efb3ca3f1..23d76b7856 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -980,12 +980,44 @@ def accumulate_event( 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 + # Normalize lists of dicts with index fields before storing + # to handle cases like tool_calls with duplicate indexes + # in the first chunk + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized = [] + index_map = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue # the `index` property is used in arrays of objects so it should @@ -1011,7 +1043,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> acc_value.extend(delta_value) continue - for delta_entry in delta_value: + # Normalize delta entries by index before merging. + # When multiple entries share the same index (e.g., tool_calls + # with duplicate indexes in the first chunk), they must be merged + # together first so that subsequent chunks merge into the correct + # logical position rather than stranding entries at duplicate indexes. + normalized_delta: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + if not is_dict(entry): + raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}") + + try: + idx = entry["index"] + except KeyError as exc: + raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc + + if not isinstance(idx, int): + raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}") + + if idx in index_map: + existing_pos = index_map[idx] + normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry) + else: + index_map[idx] = len(normalized_delta) + normalized_delta.append(entry) + + for delta_entry in normalized_delta: if not is_dict(delta_entry): raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}") diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..c50acd1d9b 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -6,12 +6,44 @@ 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 + # Normalize lists of dicts with index fields before storing + # to handle cases like tool_calls with duplicate indexes + # in the first chunk (see issue #3201) + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + # Normalize lists of dicts with index fields before storing + if is_list(delta_value) and all(is_dict(x) for x in delta_value): + normalized = [] + index_map = {} + for entry in delta_value: + idx = entry.get("index") + if idx is not None and isinstance(idx, int) and idx in index_map: + existing_pos = index_map[idx] + normalized[existing_pos] = accumulate_delta(normalized[existing_pos], entry) + else: + if idx is not None and isinstance(idx, int): + index_map[idx] = len(normalized) + normalized.append(entry) + acc[key] = normalized + else: + acc[key] = delta_value continue # the `index` property is used in arrays of objects so it should @@ -37,7 +69,33 @@ def accumulate_delta(acc: dict[object, object], delta: dict[object, object]) -> acc_value.extend(delta_value) continue - for delta_entry in delta_value: + # Normalize delta entries by index before merging. + # When multiple entries share the same index (e.g., tool_calls + # with duplicate indexes in the first chunk), they must be merged + # together first so that subsequent chunks merge into the correct + # logical position rather than stranding entries at duplicate indexes. + normalized_delta: list[object] = [] + index_map: dict[int, int] = {} + for entry in delta_value: + if not is_dict(entry): + raise TypeError(f"Unexpected list delta entry is not a dictionary: {entry}") + + try: + idx = entry["index"] + except KeyError as exc: + raise RuntimeError(f"Expected list delta entry to have an `index` key; {entry}") from exc + + if not isinstance(idx, int): + raise TypeError(f"Unexpected, list delta entry `index` value is not an integer; {idx}") + + if idx in index_map: + existing_pos = index_map[idx] + normalized_delta[existing_pos] = accumulate_delta(normalized_delta[existing_pos], entry) + else: + index_map[idx] = len(normalized_delta) + normalized_delta.append(entry) + + for delta_entry in normalized_delta: if not is_dict(delta_entry): raise TypeError(f"Unexpected list delta entry is not a dictionary: {delta_entry}") diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 5f072cafbd..d5a3cee360 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -744,7 +744,10 @@ 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": accumulate_delta( + cast("dict[object, object]", {}), + cast("dict[object, object]", choice.delta.to_dict()), + ), } return cast( diff --git a/tests/test_deltas.py b/tests/test_deltas.py new file mode 100644 index 0000000000..510d7ac586 --- /dev/null +++ b/tests/test_deltas.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from openai.lib.streaming._deltas import accumulate_delta + + +def test_duplicate_index_in_first_chunk() -> None: + """ + Regression test for issue #3201: + When the first chunk contains multiple tool_calls entries with the same + `index`, they should be merged into a single entry by logical index, + not stored as separate physical entries. + """ + chunk1 = { + "tool_calls": [ + { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "get_weather", "arguments": ""}, + }, + { + "index": 0, + "function": {"arguments": ' {"cit'}, + }, + ] + } + + chunk2 = { + "tool_calls": [ + { + "index": 0, + "function": {"arguments": 'y": "London"}'}, + } + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}" + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"cit' + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 1, f"Expected 1 tool call, got {len(acc['tool_calls'])}" + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + + +def test_multiple_tool_calls_different_indexes() -> None: + """Multiple tool calls with different indexes should remain separate.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}}, + {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' {"city": "London"}'}}, + {"index": 1, "function": {"arguments": ' {"tz": "UTC"}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 2 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 2 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + assert acc["tool_calls"][1]["function"]["arguments"] == ' {"tz": "UTC"}' + + +def test_sparse_indexes() -> None: + """Sparse indexes (index 2 arriving before index 1) should be handled correctly. + + Note: The accumulator inserts at the physical position matching the index, + so index 2 is stored at position 0 when the list is empty, then index 1 + is inserted at position 1 (append). This is existing behavior. + """ + chunk1 = { + "tool_calls": [ + {"index": 2, "id": "call_3", "type": "function", "function": {"name": "search", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 1, "id": "call_2", "type": "function", "function": {"name": "get_time", "arguments": ""}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["index"] == 2 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 2 + assert acc["tool_calls"][0]["index"] == 2 + assert acc["tool_calls"][1]["index"] == 1 + + +def test_normal_case_no_duplicates() -> None: + """Normal case with no duplicate indexes should work as before.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": ""}}, + ] + } + + chunk2 = { + "tool_calls": [ + {"index": 0, "function": {"arguments": ' {"city": "London"}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + + acc = accumulate_delta(acc, chunk2) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"city": "London"}' + + +def test_three_duplicate_indexes_in_first_chunk() -> None: + """Three entries with the same index in the first chunk should all be merged.""" + chunk1 = { + "tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", "function": {"name": "test", "arguments": ""}}, + {"index": 0, "function": {"arguments": ' {"a"'}}, + {"index": 0, "function": {"arguments": ': 1}'}}, + ] + } + + acc: dict = {} + acc = accumulate_delta(acc, chunk1) + assert len(acc["tool_calls"]) == 1 + assert acc["tool_calls"][0]["function"]["arguments"] == ' {"a": 1}'