diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index 9667eb7804c..fe25f4a307d 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -469,6 +469,15 @@ class ComputerSafetyCheck(TypedDict, total=False): ContentT = TypeVar("ContentT", bound="Content") _MODEL_OUTPUT_KIND_KEY = "model_output_kind" _MODEL_OUTPUT_REFUSAL = "refusal" +# Marks a streamed nested content item (e.g. a code interpreter call's `inputs`/`outputs` +# text) as a full resend of everything streamed so far, rather than an incremental delta +# to append. Some providers follow up a run of true deltas with one event that repeats the +# complete value (e.g. the OpenAI Responses API's `code_interpreter_call_code.done`, sent +# after a series of `.delta` events for the same call). Text content alone can't reliably +# tell that apart from a delta that merely happens to start with what came before - e.g. a +# delta of "(" followed by a delta of "()" is two chunks that combine into "(()", not a +# resend of "()" - so providers must set this explicitly on a full resend. +_CONTENT_ITEM_SNAPSHOT_KEY = "content_item_snapshot" # endregion @@ -2524,21 +2533,55 @@ def _content_items_text(items: Any) -> str | None: return "".join(text_parts) +def _is_content_item_snapshot(items: Any) -> bool: + """Whether a nested content list is tagged as a full resend, see `_CONTENT_ITEM_SNAPSHOT_KEY`.""" + if not isinstance(items, list): + return False + return any( + isinstance(item, Content) and item.additional_properties.get(_CONTENT_ITEM_SNAPSHOT_KEY) + for item in cast("list[object]", items) + ) + + +def _strip_snapshot_marker(items: Any) -> None: + """Remove the merge-only snapshot marker from a nested content list, in place. + + The marker is only an instruction for the merge step that reads it; stripping it + keeps it from persisting into finalized content that gets serialized into history + or passed to middleware. + """ + if isinstance(items, list): + for item in cast("list[object]", items): + if isinstance(item, Content): + item.additional_properties.pop(_CONTENT_ITEM_SNAPSHOT_KEY, None) + + +def _copy_without_snapshot_marker(items: Any) -> Any: + """Deep-copy a nested content list, stripping the merge-only snapshot marker.""" + copied = deepcopy(items) + _strip_snapshot_marker(copied) + return copied + + def _merge_content_item_lists(existing: Any, incoming: Any) -> Any: """Merge streamed nested content lists, replacing deltas with a later full value when present.""" if incoming is None: return existing - if existing is None: + # A snapshot can be the very first (or only) chunk seen for this call - not just one + # that arrives after prior deltas - so strip the marker here too, not only below. + if existing is None or _is_content_item_snapshot(incoming): + return _copy_without_snapshot_marker(incoming) + + # An empty list has no item to fold a delta into (and nothing to add from one), + # so hand back whichever side actually has content before indexing into it below. + if not existing: return deepcopy(incoming) + if not incoming: + return existing existing_text = _content_items_text(existing) incoming_text = _content_items_text(incoming) if existing_text is not None and incoming_text is not None: - if incoming_text.startswith(existing_text): - return deepcopy(incoming) - if existing_text.startswith(incoming_text): - return existing - existing_items = cast(list[Content], existing) merged = deepcopy(existing_items[0]) merged.text = existing_text + incoming_text @@ -2585,7 +2628,13 @@ def _coalesce_code_interpreter_content(contents: list[Content]) -> None: existing = seen.get(key) if existing is None: + # The call_id's first chunk can itself be a tagged snapshot (a call that + # never streams incremental deltas, just one done-style chunk), so this + # copy needs the same stripping `_merge_code_interpreter_content` does for + # later chunks. copied = deepcopy(content) + _strip_snapshot_marker(copied.inputs) + _strip_snapshot_marker(copied.outputs) seen[key] = copied coalesced_contents.append(copied) continue diff --git a/python/packages/core/tests/core/test_sessions.py b/python/packages/core/tests/core/test_sessions.py index 1cf5d5adcfc..666aecc0a39 100644 --- a/python/packages/core/tests/core/test_sessions.py +++ b/python/packages/core/tests/core/test_sessions.py @@ -47,6 +47,7 @@ is_local_history_conversation_id, ) from agent_framework._telemetry import FeatureIndex +from agent_framework._types import _CONTENT_ITEM_SNAPSHOT_KEY from agent_framework.exceptions import MiddlewareException from .test_filesystem import COLLIDING_IDENTIFIERS @@ -673,7 +674,12 @@ async def test_after_run_stores_coalesced_code_interpreter_chunks(self) -> None: contents=[ Content.from_code_interpreter_tool_call( call_id="ci_123", - inputs=[Content.from_text(text="import pandas as pd")], + inputs=[ + Content.from_text( + text="import pandas as pd", + additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True}, + ) + ], additional_properties={"sequence_number": 3}, ) ], diff --git a/python/packages/core/tests/core/test_types.py b/python/packages/core/tests/core/test_types.py index 82a58714acd..eebc8d9a8ed 100644 --- a/python/packages/core/tests/core/test_types.py +++ b/python/packages/core/tests/core/test_types.py @@ -42,6 +42,7 @@ GROUP_TOKEN_COUNT_KEY, ) from agent_framework._types import ( + _CONTENT_ITEM_SNAPSHOT_KEY, _MODEL_OUTPUT_KIND_KEY, _append_instructions, _get_data_bytes, @@ -2246,6 +2247,153 @@ def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call( assert fcs[1].arguments == "{}" +# region CodeInterpreter streaming merge + + +def test_code_interpreter_deltas_concatenate_even_when_one_starts_with_the_other(): + """Two genuine deltas must concatenate even if one happens to start with the other. + + The merge previously guessed "is this a full resend?" by checking whether one + chunk's text starts with the other's. That guess is wrong whenever two ordinary, + unrelated deltas simply happen to line up that way - e.g. a delta of "(" followed + by a delta of "()" is two characters streamed in order, not the second one + re-sending everything. Before the fix this dropped the first delta entirely and + left only "()". + """ + updates = [ + ChatResponseUpdate( + contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="(")])] + ), + ChatResponseUpdate( + contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="()")])] + ), + ] + + resp = ChatResponse.from_updates(updates) + calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"] + assert len(calls) == 1 + assert calls[0].inputs is not None + assert "".join(item.text or "" for item in calls[0].inputs) == "(()" + + +def test_code_interpreter_deltas_concatenate_in_reverse_prefix_order_too(): + """Same as above with the longer delta arriving first.""" + updates = [ + ChatResponseUpdate( + contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="()")])] + ), + ChatResponseUpdate( + contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="(")])] + ), + ] + + resp = ChatResponse.from_updates(updates) + calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"] + assert len(calls) == 1 + assert calls[0].inputs is not None + assert "".join(item.text or "" for item in calls[0].inputs) == "()(" + + +def test_code_interpreter_snapshot_chunk_replaces_accumulated_deltas(): + """A chunk explicitly tagged as a full resend replaces the accumulated deltas. + + Mirrors the OpenAI Responses API, whose `code_interpreter_call_code.done` event + repeats the complete code generated so far after a run of `.delta` events for the + same call; the client tags that one chunk so the merge can tell it apart from an + ordinary delta instead of guessing from the text. + """ + updates = [ + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="import")]) + ] + ), + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text=" pandas")]) + ] + ), + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call( + call_id="ci_1", + inputs=[ + Content.from_text( + text="import pandas as pd", + additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True}, + ) + ], + ) + ] + ), + ] + + resp = ChatResponse.from_updates(updates) + calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"] + assert len(calls) == 1 + assert calls[0].inputs is not None + assert "".join(item.text or "" for item in calls[0].inputs) == "import pandas as pd" + # The marker is a merge-time instruction only; it must not survive into the + # finalized content, where it would get serialized into history or seen by middleware. + assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties + + +def test_code_interpreter_snapshot_marker_stripped_as_the_first_chunk(): + """The marker must not leak even when the snapshot is the only chunk for a call. + + The first delta for a call takes a different code path (`existing` is still `None`, + so there's nothing to fold onto yet) than a later one, and that path has to strip the + marker too - a call that never streams incremental deltas at all, just one tagged + chunk, is exactly this case. + """ + updates = [ + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call( + call_id="ci_1", + inputs=[ + Content.from_text( + text="print('hi')", + additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True}, + ) + ], + ) + ] + ), + ] + + resp = ChatResponse.from_updates(updates) + calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"] + assert len(calls) == 1 + assert calls[0].inputs is not None + assert "".join(item.text or "" for item in calls[0].inputs) == "print('hi')" + assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties + + +def test_code_interpreter_delta_merges_onto_an_empty_placeholder(): + """A placeholder chunk with an empty `inputs`/`outputs` list must not blow up the merge. + + A provider can open a call with an empty list before any text streams in (an + `outputs=[]` placeholder is exactly what the OpenAI client sends when a call + starts). `existing` is then `[]`, not `None`, so the merge has to hand the + first real delta back rather than index into the empty list. + """ + updates = [ + ChatResponseUpdate(contents=[Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[])]), + ChatResponseUpdate( + contents=[ + Content.from_code_interpreter_tool_call(call_id="ci_1", inputs=[Content.from_text(text="import os")]) + ] + ), + ] + + resp = ChatResponse.from_updates(updates) + calls = [c for c in resp.messages[0].contents if c.type == "code_interpreter_tool_call"] + assert len(calls) == 1 + assert calls[0].inputs is not None + assert "".join(item.text or "" for item in calls[0].inputs) == "import os" + + # region Role & FinishReason basics diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index a42424457d8..510ad4f7a19 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -156,6 +156,7 @@ class _PromptCacheOptions(TypedDict, total=False): _MODEL_OUTPUT_KIND_KEY = "model_output_kind" _MODEL_OUTPUT_REFUSAL = "refusal" +_CONTENT_ITEM_SNAPSHOT_KEY = "content_item_snapshot" _TERMINAL_RESPONSE_STATUSES: Final[frozenset[str]] = frozenset({"completed", "failed", "incomplete", "cancelled"}) @@ -2520,6 +2521,59 @@ def _image_generation_item_to_contents(self, item: Any) -> list[Content]: ), ] + def _code_interpreter_item_to_contents(self, item: Any, *, mark_code_as_snapshot: bool) -> list[Content]: + """Convert a ``code_interpreter_call`` output item into framework ``Content`` objects. + + Used by the non-streaming parser and by both the streaming ``response.output_item.added`` + and ``response.output_item.done`` handlers. ``outputs`` (logs/images) are only ever + populated on the completed item - at ``added`` the call has not run yet - so there is + nothing to mark there; callers pass ``mark_code_as_snapshot=True`` only for a completed + item's code, so the streaming merge layer replaces (rather than concatenates onto) any + `.delta`/`.done` code chunks already accumulated for this call instead of duplicating them. + """ + call_id = getattr(item, "call_id", None) or getattr(item, "id", None) + outputs: list[Content] = [] + if item_outputs := getattr(item, "outputs", None): + for code_output in item_outputs: + if getattr(code_output, "type", None) == "logs": + outputs.append( + Content.from_text( + text=code_output.logs, + raw_representation=code_output, + ) + ) + elif getattr(code_output, "type", None) == "image": + outputs.append( + Content.from_uri( + uri=code_output.url, + raw_representation=code_output, + media_type="image", + ) + ) + contents: list[Content] = [] + if code := getattr(item, "code", None): + contents.append( + Content.from_code_interpreter_tool_call( + call_id=call_id, + inputs=[ + Content.from_text( + text=code, + raw_representation=item, + additional_properties={_CONTENT_ITEM_SNAPSHOT_KEY: True} if mark_code_as_snapshot else None, + ) + ], + raw_representation=item, + ) + ) + contents.append( + Content.from_code_interpreter_tool_result( + call_id=call_id, + outputs=outputs, + raw_representation=item, + ) + ) + return contents + def _shell_item_to_contents(self, item: Any, local_shell_tool_name: str | None) -> list[Content]: """Convert a shell output item into framework ``Content`` objects. @@ -3441,40 +3495,9 @@ def _parse_response_from_openai( ) ) case "code_interpreter_call": # ResponseOutputCodeInterpreterCall - call_id = getattr(item, "call_id", None) or getattr(item, "id", None) - outputs: list[Content] = [] - if item_outputs := getattr(item, "outputs", None): - for code_output in item_outputs: - if getattr(code_output, "type", None) == "logs": - outputs.append( - Content.from_text( - text=code_output.logs, - raw_representation=code_output, - ) - ) - elif getattr(code_output, "type", None) == "image": - outputs.append( - Content.from_uri( - uri=code_output.url, - raw_representation=code_output, - media_type="image", - ) - ) - if code := getattr(item, "code", None): - contents.append( - Content.from_code_interpreter_tool_call( - call_id=call_id, - inputs=[Content.from_text(text=code, raw_representation=item)], - raw_representation=item, - ) - ) - contents.append( - Content.from_code_interpreter_tool_result( - call_id=call_id, - outputs=outputs, - raw_representation=item, - ) - ) + # Non-streaming: the whole item is already complete, so there is nothing to + # merge against and no need to tag the code as a snapshot. + contents.extend(self._code_interpreter_item_to_contents(item, mark_code_as_snapshot=False)) case "function_call": # ResponseOutputFunctionCall contents.append( Content.from_function_call( @@ -3887,7 +3910,11 @@ def parse_annotations( Content.from_text( text=event.code, raw_representation=event, - additional_properties=ci_additional_properties, + # The done event repeats the complete code generated so far + # rather than a further delta, so it must replace (not be + # concatenated onto) any `.delta` events already accumulated + # for this call. + additional_properties={**ci_additional_properties, _CONTENT_ITEM_SNAPSHOT_KEY: True}, ) ], raw_representation=event, @@ -3967,44 +3994,10 @@ def parse_annotations( ) # Result deferred to response.output_item.done case "code_interpreter_call": # ResponseOutputCodeInterpreterCall - call_id = getattr(event_item, "call_id", None) or getattr(event_item, "id", None) - outputs: list[Content] = [] - if hasattr(event_item, "outputs") and event_item.outputs: - for code_output in event_item.outputs: - if getattr(code_output, "type", None) == "logs": - outputs.append( - Content.from_text( - text=cast(Any, code_output).logs, - raw_representation=code_output, - ) - ) - elif getattr(code_output, "type", None) == "image": - outputs.append( - Content.from_uri( - uri=cast(Any, code_output).url, - raw_representation=code_output, - media_type="image", - ) - ) - if hasattr(event_item, "code") and event_item.code: - contents.append( - Content.from_code_interpreter_tool_call( - call_id=call_id, - inputs=[ - Content.from_text( - text=event_item.code, - raw_representation=event_item, - ) - ], - raw_representation=event_item, - ) - ) - contents.append( - Content.from_code_interpreter_tool_result( - call_id=call_id, - outputs=outputs, - raw_representation=event_item, - ) + # The call has not run yet at `added` time, so any code seen here is an + # opening fragment deltas build on, not a complete value to snapshot. + contents.extend( + self._code_interpreter_item_to_contents(event_item, mark_code_as_snapshot=False) ) case ( "shell_call" @@ -4202,6 +4195,12 @@ def parse_annotations( seen_function_call_output_ids, done_item ): contents.append(self._parse_function_call_output_content(output_item, done_call_id)) + elif getattr(done_item, "type", None) == "code_interpreter_call": + # The completed item is the only place `outputs` (logs/images) are ever + # delivered during streaming - there is no incremental outputs event - and + # its `code` is the authoritative final value, so it is tagged as a snapshot + # even though `.code_interpreter_call_code.done` (above) usually already set it. + contents.extend(self._code_interpreter_item_to_contents(done_item, mark_code_as_snapshot=True)) elif getattr(done_item, "type", None) == _AZURE_AI_SEARCH_CALL_OUTPUT_TYPE: pass case _: diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 99e1d75c979..a31687c8236 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -75,7 +75,7 @@ from pytest import param from agent_framework_openai import OpenAIChatClient, OpenAIChatOptions, RawOpenAIChatClient -from agent_framework_openai._chat_client import OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY +from agent_framework_openai._chat_client import _CONTENT_ITEM_SNAPSHOT_KEY, OPENAI_LOCAL_SHELL_CALL_ITEM_ID_KEY from agent_framework_openai._exceptions import OpenAIContentFilterException _OPENAI_HTTPX = cast(Any, import_module(DefaultAsyncHttpxClient.__mro__[1].__module__.partition(".")[0])) @@ -8211,6 +8211,121 @@ def test_parse_chunk_from_openai_code_interpreter_done() -> None: assert result.contents[0].additional_properties["output_index"] == 0 assert result.contents[0].additional_properties["sequence_number"] == 5 assert result.contents[0].additional_properties["item_id"] == "ci_456" + # The done event repeats the full code rather than a further delta, so the merge + # layer needs it tagged as a snapshot to replace (not concatenate onto) accumulated + # `.delta` chunks for the same call. + assert result.contents[0].inputs[0].additional_properties[_CONTENT_ITEM_SNAPSHOT_KEY] is True + + +def test_parse_chunk_from_openai_code_interpreter_output_item_done() -> None: + """Test _parse_chunk_from_openai with a completed code_interpreter_call output item. + + `response.output_item.done` is the only place `outputs` (logs/images) are ever delivered + for a streamed code interpreter call - there is no incremental outputs event - so this + completed item must be parsed, and its `code` must be tagged as a snapshot like + `code_interpreter_call_code.done` is, since both can carry the same final code. + """ + client = OpenAIChatClient(model="test-model", api_key="test-key") + chat_options: dict[str, Any] = {} + function_call_ids: dict[int, tuple[str, str]] = {} + + mock_log_output = MagicMock() + mock_log_output.type = "logs" + mock_log_output.logs = "hello" + + mock_done_event = MagicMock() + mock_done_event.type = "response.output_item.done" + mock_done_item = MagicMock() + mock_done_item.type = "code_interpreter_call" + mock_done_item.call_id = "ci_789" + mock_done_item.id = "ci_789" + mock_done_item.code = "print('hello')" + mock_done_item.outputs = [mock_log_output] + mock_done_event.item = mock_done_item + + result = client._parse_chunk_from_openai(mock_done_event, chat_options, function_call_ids) + call_contents = [c for c in result.contents if c.type == "code_interpreter_tool_call"] + result_contents = [c for c in result.contents if c.type == "code_interpreter_tool_result"] + assert len(call_contents) == 1 + assert call_contents[0].call_id == "ci_789" + assert call_contents[0].inputs + assert call_contents[0].inputs[0].text == "print('hello')" + assert call_contents[0].inputs[0].additional_properties[_CONTENT_ITEM_SNAPSHOT_KEY] is True + assert len(result_contents) == 1 + assert result_contents[0].call_id == "ci_789" + assert result_contents[0].outputs + assert any(out.type == "text" and out.text == "hello" for out in result_contents[0].outputs) + + +def test_code_interpreter_streaming_sequence_does_not_duplicate_code_and_captures_outputs() -> None: + """A realistic added/delta/done/output_item.done sequence must not duplicate code. + + `code_interpreter_call_code.done` and `response.output_item.done` can both carry the + same final code for one call; without both being tagged as snapshots, the second one + to arrive would concatenate onto the first instead of being recognized as a repeat of + the same authoritative value. + """ + client = OpenAIChatClient(model="test-model", api_key="test-key") + chat_options: dict[str, Any] = {} + function_call_ids: dict[int, tuple[str, str]] = {} + updates = [] + + added_event = MagicMock() + added_event.type = "response.output_item.added" + added_item = MagicMock() + added_item.type = "code_interpreter_call" + added_item.call_id = "ci_1" + added_item.id = "ci_1" + added_item.code = None + added_item.outputs = None + added_event.item = added_item + updates.append(client._parse_chunk_from_openai(added_event, chat_options, function_call_ids)) + + for index, delta in enumerate(["import pandas", " as pd"]): + delta_event = MagicMock() + delta_event.type = "response.code_interpreter_call_code.delta" + delta_event.item_id = "ci_1" + delta_event.delta = delta + delta_event.output_index = 0 + delta_event.sequence_number = index + 1 + delta_event.call_id = None + delta_event.id = None + updates.append(client._parse_chunk_from_openai(delta_event, chat_options, function_call_ids)) + + done_code_event = MagicMock() + done_code_event.type = "response.code_interpreter_call_code.done" + done_code_event.item_id = "ci_1" + done_code_event.code = "import pandas as pd" + done_code_event.output_index = 0 + done_code_event.sequence_number = 3 + done_code_event.call_id = None + done_code_event.id = None + updates.append(client._parse_chunk_from_openai(done_code_event, chat_options, function_call_ids)) + + mock_log_output = MagicMock() + mock_log_output.type = "logs" + mock_log_output.logs = "pandas 2.1.0" + done_item_event = MagicMock() + done_item_event.type = "response.output_item.done" + done_item = MagicMock() + done_item.type = "code_interpreter_call" + done_item.call_id = "ci_1" + done_item.id = "ci_1" + done_item.code = "import pandas as pd" + done_item.outputs = [mock_log_output] + done_item_event.item = done_item + updates.append(client._parse_chunk_from_openai(done_item_event, chat_options, function_call_ids)) + + response = ChatResponse.from_updates(updates) + calls = [c for c in response.messages[0].contents if c.type == "code_interpreter_tool_call"] + results = [c for c in response.messages[0].contents if c.type == "code_interpreter_tool_result"] + assert len(calls) == 1 + assert calls[0].inputs + assert "".join(item.text or "" for item in calls[0].inputs) == "import pandas as pd" + assert _CONTENT_ITEM_SNAPSHOT_KEY not in calls[0].inputs[0].additional_properties + assert len(results) == 1 + assert results[0].outputs + assert any(out.type == "text" and out.text == "pandas 2.1.0" for out in results[0].outputs) def test_parse_chunk_from_openai_reasoning() -> None: