diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 7b0efa634e..7b65ec3b2a 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -2032,12 +2032,8 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse logger.warning(f"Skipping unknown content type or invalid content: {exc}") continue match content_type: - # mypy doesn't narrow type based on match/case, but we know these are FunctionCallContents - case "function_call" if message.contents and message.contents[-1].type == "function_call": - try: - message.contents[-1] += content - except (AdditionItemMismatch, ContentError): - message.contents.append(content) + case "function_call": + _merge_function_call_content(message, content) case "usage": if response.usage_details is None: response.usage_details = UsageDetails() @@ -2075,6 +2071,38 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse response.continuation_token = update.continuation_token +def _merge_function_call_content(message: Message, content: Content) -> None: + """Merge a streamed function_call chunk into the in-progress call it belongs to. + + Providers can stream multiple tool calls in parallel, so the next function_call + chunk is not necessarily a continuation of the most recently appended one; a chunk + with a call_id is matched against existing contents by that id first. Chunks with + no call_id (continuation deltas some providers only stamp on the first chunk) fall + back to merging with the trailing function_call item, preserving prior behavior. + """ + call_id = getattr(content, "call_id", None) + if call_id: + for index in range(len(message.contents) - 1, -1, -1): + existing = message.contents[index] + if existing.type == "function_call" and getattr(existing, "call_id", None) == call_id: + try: + message.contents[index] = existing + content + except (AdditionItemMismatch, ContentError): + break + return + # A tagged chunk that matches no in-progress call is a new call, not a + # continuation - an untagged trailing item would silently absorb it otherwise. + message.contents.append(content) + return + if message.contents and message.contents[-1].type == "function_call": + try: + message.contents[-1] += content + return + except (AdditionItemMismatch, ContentError): + pass + message.contents.append(content) + + def _coalesce_text_content(contents: list[Content], type_str: Literal["text", "text_reasoning"]) -> None: """Take any subsequence Text or TextReasoningContent items and coalesce them into a single item.""" if not contents: diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index f7669221bd..ff374a609a 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -1912,6 +1912,87 @@ def test_function_call_incompatible_ids_are_not_merged(): assert len(fcs) == 2 +def test_function_call_interleaved_parallel_streaming_merges_by_call_id(): + """Argument deltas for two parallel tool calls can interleave; each must land on its own call. + + This mirrors how the OpenAI Responses API streams parallel tool calls: every + ``response.function_call_arguments.delta`` event is tagged with the call's real + call_id (tracked per output_index), but deltas for different calls are not + guaranteed to arrive grouped together. Before this fix, only the trailing + content item was ever considered a merge target, so an out-of-turn delta for an + earlier call_id was appended as a stray duplicate instead of being folded into + its call, leaving both calls with incomplete, unparsable arguments. + """ + updates = [ + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments="")]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments="")]), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='{"location":')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='{"timezone":')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='"NYC"}')] + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='"EST"}')] + ), + ] + + resp = ChatResponse.from_updates(updates) + assert len(resp.messages) == 1 + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 2 + + by_call_id = {c.call_id: c for c in fcs} + assert by_call_id["call_1"].arguments == '{"location":"NYC"}' + assert by_call_id["call_2"].arguments == '{"timezone":"EST"}' + + +def test_function_call_merge_falls_back_to_trailing_item_without_call_id(): + """Continuation deltas some providers never re-stamp with a call_id still merge. + + Not every provider repeats the call_id on every streamed chunk (e.g. the OpenAI + Chat Completions API only sends it on the first delta for a tool call), so the + fallback of merging into the trailing function_call item must still hold for the + single-call-in-flight case. + """ + updates = [ + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="f", arguments="{")]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="", name="", arguments="}")]), + ] + + resp = ChatResponse.from_updates(updates) + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 1 + assert fcs[0].call_id == "call_1" + assert fcs[0].arguments == "{}" + + +def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call(): + """A tagged chunk with no matching in-progress call must not merge into an untagged one. + + Content.__add__ only rejects a merge when *both* sides carry a call_id and they + differ, so an untagged trailing item (call_id falsy) would otherwise silently + accept a chunk tagged with a brand-new call_id, adopting that id and + concatenating unrelated arguments. The trailing-item fallback must be reserved + for chunks that carry no call_id at all. + """ + untagged = Content("function_call", call_id=None, name="a", arguments="partial-a") + resp = ChatResponse.from_updates([ + ChatResponseUpdate(contents=[untagged]), + ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_new", name="b", arguments="{}")]), + ]) + + fcs = [c for c in resp.messages[0].contents if c.type == "function_call"] + assert len(fcs) == 2 + assert fcs[0].call_id is None + assert fcs[0].arguments == "partial-a" + assert fcs[1].call_id == "call_new" + assert fcs[1].arguments == "{}" + + # region Role & FinishReason basics