diff --git a/src/openai/lib/streaming/_assistants.py b/src/openai/lib/streaming/_assistants.py index 6efb3ca3f1..0069c56daf 100644 --- a/src/openai/lib/streaming/_assistants.py +++ b/src/openai/lib/streaming/_assistants.py @@ -977,15 +977,59 @@ def accumulate_event( return current_message_snapshot, new_content +def _normalize_indexed_list(items: list[object]) -> list[object]: + """Merge list entries that share the same integer ``index`` key. + + When the first delta chunk for a key carries multiple entries with the + same ``index`` (e.g. tool name **and** the start of arguments both at + index 0), those entries *must* be merged into a single logical entry. + Without this step the accumulator stores them as separate physical list + entries; the second entry is then orphaned because every subsequent delta + merges into ``acc_value[index]`` (the first one), leaving argument + fragments stranded and producing invalid final JSON. + """ + if not items: + return items + if not all(is_dict(item) for item in items): + return items + + # Count occurrences of each integer index. + index_counts: dict[int, int] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + index_counts[idx] = index_counts.get(idx, 0) + 1 + + # No duplicate indices -> nothing to normalise. + if not index_counts or max(index_counts.values()) <= 1: + return items + + # Merge entries that share the same index. + merged: dict[int, dict[object, object]] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + if idx in merged: + merged[idx] = accumulate_delta(merged[idx], dict(item)) + else: + merged[idx] = dict(item) + + # Rebuild the list ordered by index. + return [ + {**entry, "index": idx} + for idx, entry in sorted(merged.items()) + ] + + 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 + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue # the `index` property is used in arrays of objects so it should diff --git a/src/openai/lib/streaming/_deltas.py b/src/openai/lib/streaming/_deltas.py index a5e1317612..1e6a0f4603 100644 --- a/src/openai/lib/streaming/_deltas.py +++ b/src/openai/lib/streaming/_deltas.py @@ -3,15 +3,59 @@ from ..._utils import is_dict, is_list +def _normalize_indexed_list(items: list[object]) -> list[object]: + """Merge list entries that share the same integer ``index`` key. + + When the first delta chunk for a key carries multiple entries with the + same ``index`` (e.g. tool name **and** the start of arguments both at + index 0), those entries *must* be merged into a single logical entry. + Without this step the accumulator stores them as separate physical list + entries; the second entry is then orphaned because every subsequent delta + merges into ``acc_value[index]`` (the first one), leaving argument + fragments stranded and producing invalid final JSON. + """ + if not items: + return items + if not all(is_dict(item) for item in items): + return items + + # Count occurrences of each integer index. + index_counts: dict[int, int] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + index_counts[idx] = index_counts.get(idx, 0) + 1 + + # No duplicate indices -> nothing to normalise. + if not index_counts or max(index_counts.values()) <= 1: + return items + + # Merge entries that share the same index. + merged: dict[int, dict[object, object]] = {} + for item in items: + idx = item.get("index") + if isinstance(idx, int): + if idx in merged: + merged[idx] = accumulate_delta(merged[idx], dict(item)) + else: + merged[idx] = dict(item) + + # Rebuild the list ordered by index. + return [ + {**entry, "index": idx} + for idx, entry in sorted(merged.items()) + ] + + 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 + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue acc_value = acc[key] if acc_value is None: - acc[key] = delta_value + acc[key] = _normalize_indexed_list(delta_value) if is_list(delta_value) else delta_value continue # the `index` property is used in arrays of objects so it should diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py index 5f072cafbd..5f72bda6fd 100644 --- a/src/openai/lib/streaming/chat/_completions.py +++ b/src/openai/lib/streaming/chat/_completions.py @@ -22,7 +22,7 @@ FunctionToolCallArgumentsDoneEvent, FunctionToolCallArgumentsDeltaEvent, ) -from .._deltas import accumulate_delta +from .._deltas import _normalize_indexed_list, accumulate_delta from ...._types import Omit, IncEx, omit from ...._utils import is_given, consume_sync_iterator, consume_async_iterator from ...._compat import model_dump @@ -409,13 +409,17 @@ def _accumulate_chunk(self, chunk: ChatCompletionChunk) -> ParsedChatCompletionS elif TYPE_CHECKING: # type: ignore[unreachable] assert_never(prev_tool) except IndexError: + message = choice.delta.to_dict() + tool_calls = message.get("tool_calls") + if isinstance(tool_calls, list): + message["tool_calls"] = _normalize_indexed_list(tool_calls) choice_snapshot = cast( ParsedChoiceSnapshot, construct_type( type_=ParsedChoiceSnapshot, value={ **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": message, }, ), ) @@ -742,9 +746,13 @@ def _convert_initial_chunk_into_snapshot(chunk: ChatCompletionChunk) -> ParsedCh choices = cast("list[object]", data["choices"]) for choice in chunk.choices: + message = choice.delta.to_dict() + tool_calls = message.get("tool_calls") + if isinstance(tool_calls, list): + message["tool_calls"] = _normalize_indexed_list(tool_calls) choices[choice.index] = { **choice.model_dump(exclude_unset=True, exclude={"delta"}), - "message": choice.delta.to_dict(), + "message": message, } return cast(