fix: merge indexed streaming deltas on first chunk

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
 

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
 

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",
+        }
+    ]