Keep OpenAI Responses tool-call arguments that arrive only in function_call_arguments.done or output_item.done (#10039)

This commit is contained in:
Douwe Maan
2026-10-09 15:36:42 +00:00
committed by GitHub
parent 49f3ef23c7
commit 858e67f679
2 changed files with 623 additions and 26 deletions
+81 -25
View File
@@ -4797,6 +4797,66 @@ class OpenAIResponsesStreamedResponse(StreamedResponse):
_phase_by_item: dict[str, Literal['commentary', 'final_answer']] = {}
mcp_list_tools_return_ids: set[str] = set()
pending_tool_search_call_ids: deque[str] = deque()
function_call_args: dict[str, str] = {}
finalized_function_calls: set[str] = set()
def handle_function_call_part(
item: responses.ResponseFunctionToolCall,
) -> ModelResponseStreamEvent:
item_id = item.id
if item_id is not None: # pragma: no branch
function_call_args[item_id] = item.arguments
# Preserve any namespace attached to a discovered deferred tool for replay.
provider_details: dict[str, Any] | None = None
if item.namespace:
provider_details = {'namespace': item.namespace}
return self._parts_manager.handle_tool_call_part(
vendor_part_id=item_id,
tool_name=item.name,
args=item.arguments,
tool_call_id=_response_tool_call_id(
item.call_id,
self.provider_response_id if self._tool_call_ids_are_response_scoped else None,
),
id=item_id,
provider_name=self.provider_name,
provider_details=provider_details,
)
def finalize_function_call(item_id: str, final_args: str) -> ModelResponseStreamEvent | None:
if item_id in finalized_function_calls or not final_args:
return None
existing_part = self._parts_manager.get_part_by_vendor_id(item_id)
if not isinstance(existing_part, ToolCallPart):
# A stream resumed after `output_item.added` has no complete part yet;
# `output_item.done` creates it from the full item.
return None
received_args = function_call_args.get(item_id, '')
if received_args == final_args:
finalized_function_calls.add(item_id)
return None
if final_args.startswith(received_args):
# The Codex backend can send arguments only in the done events.
event = self._parts_manager.handle_tool_call_delta(
vendor_part_id=item_id,
args=final_args[len(received_args) :],
)
else:
event = self._parts_manager.handle_tool_call_part(
vendor_part_id=item_id,
tool_name=existing_part.tool_name,
args=final_args,
tool_call_id=existing_part.tool_call_id,
id=existing_part.id,
provider_name=existing_part.provider_name,
provider_details=existing_part.provider_details,
)
function_call_args[item_id] = final_args
finalized_function_calls.add(item_id)
return event
if self._provider_timestamp is not None: # pragma: no branch
self.provider_details = {'timestamp': self._provider_timestamp}
@@ -4887,15 +4947,20 @@ class OpenAIResponsesStreamedResponse(StreamedResponse):
self.finish_reason = _RESPONSES_FINISH_REASON_MAP.get('failed')
elif isinstance(chunk, responses.ResponseFunctionCallArgumentsDeltaEvent):
maybe_event = self._parts_manager.handle_tool_call_delta(
vendor_part_id=chunk.item_id,
args=chunk.delta,
)
if maybe_event is not None: # pragma: no branch
yield maybe_event
# Without `output_item.added` (a resumed stream) the part has no tool name yet, so
# skip its deltas and let `output_item.done` create it; never mutate finalized args.
if chunk.item_id in function_call_args and chunk.item_id not in finalized_function_calls:
function_call_args[chunk.item_id] += chunk.delta
maybe_event = self._parts_manager.handle_tool_call_delta(
vendor_part_id=chunk.item_id,
args=chunk.delta,
)
if maybe_event is not None: # pragma: no branch
yield maybe_event
elif isinstance(chunk, responses.ResponseFunctionCallArgumentsDoneEvent):
pass # there's nothing we need to do here
if event := finalize_function_call(chunk.item_id, chunk.arguments):
yield event
elif isinstance(chunk, (responses.ResponseInProgressEvent, responses.ResponseQueuedEvent)):
self._usage += self._map_usage(chunk.response)
@@ -4913,23 +4978,7 @@ class OpenAIResponsesStreamedResponse(StreamedResponse):
elif isinstance(chunk, responses.ResponseOutputItemAddedEvent):
if isinstance(chunk.item, responses.ResponseFunctionToolCall):
# Preserve any `namespace` the Responses API attaches to a
# discovered deferred tool so it can be round-tripped on replay.
fn_provider_details: dict[str, Any] | None = None
if chunk.item.namespace:
fn_provider_details = {'namespace': chunk.item.namespace}
yield self._parts_manager.handle_tool_call_part(
vendor_part_id=chunk.item.id,
tool_name=chunk.item.name,
args=chunk.item.arguments,
tool_call_id=_response_tool_call_id(
chunk.item.call_id,
self.provider_response_id if self._tool_call_ids_are_response_scoped else None,
),
id=chunk.item.id,
provider_name=self.provider_name,
provider_details=fn_provider_details,
)
yield handle_function_call_part(chunk.item)
elif isinstance(chunk.item, responses.ResponseReasoningItem):
pass
elif isinstance(chunk.item, responses.ResponseOutputMessage):
@@ -5026,7 +5075,14 @@ class OpenAIResponsesStreamedResponse(StreamedResponse):
)
elif isinstance(chunk, responses.ResponseOutputItemDoneEvent):
if isinstance(chunk.item, responses.ResponseReasoningItem):
if isinstance(chunk.item, responses.ResponseFunctionToolCall) and (item_id := chunk.item.id):
# The Codex backend can send a call's arguments only in the done events, without deltas.
if not isinstance(self._parts_manager.get_part_by_vendor_id(item_id), ToolCallPart):
yield handle_function_call_part(chunk.item)
finalized_function_calls.add(item_id)
elif event := finalize_function_call(item_id, chunk.item.arguments):
yield event
elif isinstance(chunk.item, responses.ResponseReasoningItem):
if signature := chunk.item.encrypted_content: # pragma: no branch
# Add the signature to the part corresponding to the first summary/raw CoT
for event in self._parts_manager.handle_thinking_delta(
+542 -1
View File
@@ -63,7 +63,13 @@ from pydantic_ai.exceptions import (
ModelRetry,
SuspendedResponseExpired,
)
from pydantic_ai.messages import INVALID_JSON_KEY, ToolSearchCallPart, ToolSearchReturnPart, sanitize_messages
from pydantic_ai.messages import (
INVALID_JSON_KEY,
ModelResponseStreamEvent,
ToolSearchCallPart,
ToolSearchReturnPart,
sanitize_messages,
)
from pydantic_ai.models import ModelRequestContext, ModelRequestParameters
from pydantic_ai.models.fallback import FallbackModel
from pydantic_ai.models.test import TestModel
@@ -1329,6 +1335,541 @@ async def test_openai_responses_stream(allow_model_requests: None, openai_api_ke
assert output_text == snapshot(['The capital of France is Paris.'])
def _function_call_item(
item_id: str,
arguments: str,
*,
name: str = 'lookup',
namespace: str | None = None,
) -> resp.ResponseFunctionToolCall:
return resp.ResponseFunctionToolCall(
id=item_id,
arguments=arguments,
call_id=f'call_{item_id}',
name=name,
namespace=namespace,
status='completed',
type='function_call',
)
async def _collect_function_call_stream(
stream_events: list[resp.ResponseStreamEvent],
) -> tuple[list[ModelResponseStreamEvent], ModelResponse]:
response = response_message([])
stream = [
resp.ResponseCreatedEvent(response=response, type='response.created', sequence_number=0),
*(event.model_copy(update={'sequence_number': index}) for index, event in enumerate(stream_events, 1)),
resp.ResponseCompletedEvent(
response=response.model_copy(update={'status': 'completed'}),
type='response.completed',
sequence_number=len(stream_events) + 1,
),
]
mock_client = MockOpenAIResponses.create_mock_stream(stream)
model = OpenAIResponsesModel('gpt-4o', provider=OpenAIProvider(openai_client=mock_client))
async with model.request_stream(
[ModelRequest(parts=[UserPromptPart(content='call a tool')])],
model_settings=None,
model_request_parameters=ModelRequestParameters(),
) as streamed:
events = [event async for event in streamed]
return events, streamed.get()
@pytest.mark.parametrize('args_event', ['function_call_arguments.done', 'output_item.done'])
async def test_openai_responses_stream_function_call_args_from_done_event(
allow_model_requests: None, args_event: Literal['function_call_arguments.done', 'output_item.done']
):
"""Codex sequence (issue #9996): args arrive only in a done event, no deltas.
Mocked: the ChatGPT/Codex backend can't be recorded reliably.
"""
full_args = '{"city":"Paris"}'
added_item = _function_call_item('fc_1', '')
done_item = added_item.model_copy(update={'arguments': full_args})
stream_events: list[resp.ResponseStreamEvent] = [
resp.ResponseOutputItemAddedEvent(
item=added_item,
output_index=0,
type='response.output_item.added',
sequence_number=0,
)
]
if args_event == 'function_call_arguments.done':
stream_events.append(
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=full_args,
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
)
)
stream_events.append(
resp.ResponseOutputItemDoneEvent(
item=done_item,
output_index=0,
type='response.output_item.done',
sequence_number=0,
)
)
events, response = await _collect_function_call_stream(stream_events)
assert events == snapshot(
[
PartStartEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
),
),
PartDeltaEvent(
index=0,
delta=ToolCallPartDelta(args_delta='{"city":"Paris"}', tool_call_id='call_fc_1'),
),
PartEndEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
),
),
]
)
assert response.parts == snapshot(
[
ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
)
]
)
async def test_openai_responses_stream_function_call_complete_deltas_not_duplicated(allow_model_requests: None):
"""Normal OpenAI sequence: complete deltas then matching done events add nothing.
Mocked to pin the exact event list.
"""
full_args = '{"city":"Paris"}'
item = _function_call_item('fc_1', '')
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemAddedEvent(
item=item, output_index=0, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='{"city":',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='"Paris"}',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=full_args,
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
resp.ResponseOutputItemDoneEvent(
item=item.model_copy(update={'arguments': full_args}),
output_index=0,
type='response.output_item.done',
sequence_number=0,
),
]
)
assert [type(event).__name__ for event in events] == snapshot(
['PartStartEvent', 'PartDeltaEvent', 'PartDeltaEvent', 'PartEndEvent']
)
assert isinstance(response.parts[0], ToolCallPart)
assert response.parts[0].args == full_args
async def test_openai_responses_stream_parallel_function_call_args(allow_model_requests: None):
"""Codex parallel calls: one done-only and one delta-driven item, empty `completed.output`.
Mocked: Codex-only sequence.
"""
done_args = '{"city":"Paris"}'
delta_args = '{"city":"London"}'
done_item = _function_call_item('fc_done', '')
delta_item = _function_call_item('fc_delta', '')
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemAddedEvent(
item=done_item, output_index=0, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=done_args,
item_id='fc_done',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
resp.ResponseOutputItemAddedEvent(
item=delta_item, output_index=1, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='{"city":',
item_id='fc_delta',
output_index=1,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='"London"}',
item_id='fc_delta',
output_index=1,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=delta_args,
item_id='fc_delta',
output_index=1,
type='response.function_call_arguments.done',
sequence_number=0,
),
]
)
assert events == snapshot(
[
PartStartEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='',
tool_call_id='call_fc_done',
id='fc_done',
provider_name='openai',
),
),
PartDeltaEvent(
index=0,
delta=ToolCallPartDelta(args_delta='{"city":"Paris"}', tool_call_id='call_fc_done'),
),
PartEndEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_done',
id='fc_done',
provider_name='openai',
),
next_part_kind='tool-call',
),
PartStartEvent(
index=1,
part=ToolCallPart(
tool_name='lookup', args='', tool_call_id='call_fc_delta', id='fc_delta', provider_name='openai'
),
previous_part_kind='tool-call',
),
PartDeltaEvent(
index=1,
delta=ToolCallPartDelta(args_delta='{"city":', tool_call_id='call_fc_delta'),
),
PartDeltaEvent(
index=1,
delta=ToolCallPartDelta(args_delta='"London"}', tool_call_id='call_fc_delta'),
),
PartEndEvent(
index=1,
part=ToolCallPart(
tool_name='lookup',
args='{"city":"London"}',
tool_call_id='call_fc_delta',
id='fc_delta',
provider_name='openai',
),
),
]
)
assert response.parts == snapshot(
[
ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_done',
id='fc_done',
provider_name='openai',
),
ToolCallPart(
tool_name='lookup',
args='{"city":"London"}',
tool_call_id='call_fc_delta',
id='fc_delta',
provider_name='openai',
),
]
)
async def test_openai_responses_stream_function_call_done_appends_missing_suffix(allow_model_requests: None):
"""Defensive: partial deltas then a full done snapshot append only the suffix.
Mocked: no live API sends this reliably.
"""
full_args = '{"city":"Paris"}'
item = _function_call_item('fc_1', '')
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemAddedEvent(
item=item, output_index=0, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='{"city":',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=full_args,
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
]
)
deltas = [event for event in events if isinstance(event, PartDeltaEvent)]
assert deltas[-1] == snapshot(
PartDeltaEvent(index=0, delta=ToolCallPartDelta(args_delta='"Paris"}', tool_call_id='call_fc_1'))
)
assert isinstance(response.parts[0], ToolCallPart)
assert response.parts[0].args == full_args
async def test_openai_responses_stream_function_call_done_replaces_disagreeing_args(allow_model_requests: None):
"""Defensive: deltas that disagree with the done snapshot are replaced by it.
Mocked: no live API sends this reliably.
"""
item = _function_call_item('fc_1', '', name='lookup_weather', namespace='weather')
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemAddedEvent(
item=item, output_index=0, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='{"city":"Londo',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments='{"city":"Paris"}',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
]
)
starts = [event for event in events if isinstance(event, PartStartEvent)]
assert starts[-1] == snapshot(
PartStartEvent(
index=0,
part=ToolCallPart(
tool_name='lookup_weather',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
provider_details={'namespace': 'weather'},
),
previous_part_kind='tool-call',
)
)
assert response.parts == snapshot(
[
ToolCallPart(
tool_name='lookup_weather',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
provider_details={'namespace': 'weather'},
)
]
)
async def test_openai_responses_stream_function_call_empty_done_and_late_delta(allow_model_requests: None):
"""Defensive: an empty done snapshot keeps streamed args and a late delta is ignored.
Mocked: not reproducible live.
"""
item = _function_call_item('fc_1', '')
full_args = '{"city":"Paris"}'
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemAddedEvent(
item=item, output_index=0, type='response.output_item.added', sequence_number=0
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='{"city":',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments='',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=full_args,
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='ignored',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
]
)
assert [
event.delta.args_delta
for event in events
if isinstance(event, PartDeltaEvent) and isinstance(event.delta, ToolCallPartDelta)
] == snapshot(['{"city":', '"Paris"}'])
assert isinstance(response.parts[0], ToolCallPart)
assert response.parts[0].args == full_args
async def test_openai_responses_stream_function_call_output_done_without_added(allow_model_requests: None):
"""Defensive: a resumed stream with only `output_item.done` creates the call.
Mocked: resume points aren't recordable.
"""
events, response = await _collect_function_call_stream(
[
resp.ResponseOutputItemDoneEvent(
item=_function_call_item('fc_1', '{"city":"Paris"}'),
output_index=0,
type='response.output_item.done',
sequence_number=0,
)
]
)
assert events == snapshot(
[
PartStartEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
),
),
PartEndEvent(
index=0,
part=ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
),
),
]
)
assert response.parts == snapshot(
[
ToolCallPart(
tool_name='lookup',
args='{"city":"Paris"}',
tool_call_id='call_fc_1',
id='fc_1',
provider_name='openai',
)
]
)
async def test_openai_responses_stream_function_call_resumed_mid_deltas(allow_model_requests: None):
"""Defensive: a stream resumed mid-deltas takes the call from `output_item.done`.
Mocked: resume points aren't recordable.
"""
full_args = '{"city":"Paris"}'
_, response = await _collect_function_call_stream(
[
resp.ResponseFunctionCallArgumentsDeltaEvent(
delta='"Paris"}',
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.delta',
sequence_number=0,
),
resp.ResponseFunctionCallArgumentsDoneEvent(
arguments=full_args,
item_id='fc_1',
output_index=0,
type='response.function_call_arguments.done',
sequence_number=0,
),
resp.ResponseOutputItemDoneEvent(
item=_function_call_item('fc_1', full_args),
output_index=0,
type='response.output_item.done',
sequence_number=0,
),
]
)
assert response.parts == snapshot(
[
ToolCallPart(
tool_name='lookup', args='{"city":"Paris"}', tool_call_id='call_fc_1', id='fc_1', provider_name='openai'
)
]
)
async def test_openai_responses_moderation(allow_model_requests: None, openai_api_key: str):
"""Moderation results requested via `openai_moderation` are surfaced in `provider_details['moderation']`."""
model = OpenAIResponsesModel('gpt-5', provider=OpenAIProvider(api_key=openai_api_key))