Regression in v1.0.3: UnexpectedModelBehavior crash during streaming with model reasoning

[Joakim-T]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

Related to issue #2687 and PR #2855

I've been digging into an issue I ran into after upgrading from (my very slightly modified) v1.0.2 to v1.0.3.

It seems there's a regression that causes a crash during streaming whenever the model includes a reasoning step in its response via the OpenAI Responses API.
The specific error I'm seeing is pydantic_ai.exceptions.UnexpectedModelBehavior: Cannot create a ThinkingPart with no content.

What's Happening

It looks like the crash is triggered by new logic introduced in v1.0.3 to handle the ResponseOutputItemDoneEvent for a ResponseReasoningItem.

Here's what I've traced so far:

1. In pydantic_ai/models/openai.py, inside the OpenAIResponsesStreamedResponse._get_event_iterator method, there's a new block for ResponseOutputItemDoneEvent

pydantic-ai/pydantic_ai_slim/pydantic_ai/models/openai.py

Lines 1372 to 1388 in bee76e6

	warnings.warn( # pragma: no cover
	f'Handling of this item type is not yet implemented. Please report on our GitHub: {chunk}',
	UserWarning,
	)
	elif isinstance(chunk, responses.ResponseOutputItemDoneEvent):
	if isinstance(chunk.item, responses.ResponseReasoningItem):
	# Add the signature to the part corresponding to the first summary item
	signature = chunk.item.encrypted_content
	yield self._parts_manager.handle_thinking_delta(
	vendor_part_id=f'{chunk.item.id}-0',
	id=chunk.item.id,
	signature=signature,
	provider_name=self.provider_name if signature else None,
	)
	pass

2. When this event is for a ResponseReasoningItem, the new code calls self._parts_manager.handle_thinking_delta, passing in a signature but leaving the content argument as None.

3. The problem is that in pydantic_ai/_parts_manager.py, the handle_thinking_delta function has a strict check. If it tries to create a new ThinkingPart (because one doesn't already exist for that vendor_part_id), it requires the content to not be None ([around line 210 in the 1.0.3 release]

pydantic-ai/pydantic_ai_slim/pydantic_ai/_parts_manager.py

Lines 217 to 224 in bee76e6

	)
	self._parts[part_index] = part_delta.apply(existing_thinking_part)
	return PartDeltaEvent(index=part_index, delta=part_delta)
	else:
	raise UnexpectedModelBehavior('Cannot update a ThinkingPart with no content or signature')
	def handle_tool_call_delta(
	self,

This mismatch leads to the UnexpectedModelBehavior exception, which stops the stream. This doesn't happen in v1.0.2 because that version doesn't have the code to handle the ResponseOutputItemDoneEvent, so the problematic call is never made.

Possible Fix

It seems like the fix might be to either:

• Make the validation in handle_thinking_delta a bit more flexible, allowing a ThinkingPart to be created with just a signature.
• Or, ensure that a ThinkingPart is always created with some initial content before the ResponseOutputItemDoneEvent is processed, so the call becomes an update rather than a creation.

Let me know if you need any more info.

Example Code

Python, Pydantic AI & LLM client version

pydantic-ai-slim==1.0.3
openai==1.107.1
ag-ui-protocol==0.1.8

[DouweM]

@Joakim-T Thanks for the report! Can you help me reproduce this? The test at

pydantic-ai/tests/models/test_openai_responses.py

Lines 1253 to 1298 in 99581d7

	async def test_openai_responses_thinking_part_iter(allow_model_requests: None, openai_api_key: str):
	provider = OpenAIProvider(api_key=openai_api_key)
	responses_model = OpenAIResponsesModel('o3-mini', provider=provider)
	settings = OpenAIResponsesModelSettings(openai_reasoning_effort='high', openai_reasoning_summary='detailed')
	agent = Agent(responses_model, model_settings=settings)
	async with agent.iter(user_prompt='How do I cross the street?') as agent_run:
	async for node in agent_run:
	if Agent.is_model_request_node(node) or Agent.is_call_tools_node(node):
	async with node.stream(agent_run.ctx) as request_stream:
	async for _ in request_stream:
	pass
	assert agent_run.result is not None
	assert agent_run.result.all_messages() == snapshot(
	[
	ModelRequest(
	parts=[
	UserPromptPart(
	content='How do I cross the street?',
	timestamp=IsDatetime(),
	)
	]
	),
	ModelResponse(
	parts=[
	ThinkingPart(
	content=IsStr(),
	id='rs_68bb4cc9730481a38d9aaf55dff8dd4a0b681c5350c0b73b',
	signature=IsStr(),
	provider_name='openai',
	),
	ThinkingPart(content=IsStr(), id='rs_68bb4cc9730481a38d9aaf55dff8dd4a0b681c5350c0b73b'),
	ThinkingPart(content=IsStr(), id='rs_68bb4cc9730481a38d9aaf55dff8dd4a0b681c5350c0b73b'),
	TextPart(content=IsStr()),
	],
	usage=RequestUsage(input_tokens=13, output_tokens=1733, details={'reasoning_tokens': 1280}),
	model_name='o3-mini-2025-01-31',
	timestamp=IsDatetime(),
	provider_name='openai',
	provider_details={'finish_reason': 'completed'},
	provider_response_id='resp_68bb4cbe885481a3a55b6a2e6d6aa5120b681c5350c0b73b',
	finish_reason='stop',
	),
	]
	)

works fine, even if I change it to use gpt-5.

Are you running against a non-OpenAI Responses API? Or the one from gpt-oss by any chance? It would be very useful to see the full API response.

With OpenAI Responses, any responses.ResponseOutputItemDoneEvent(responses.ResponseReasoningItem) that I've seen would have been preceded by a ResponseReasoningSummaryPartAddedEvent, so there would already be content:

pydantic-ai/pydantic_ai_slim/pydantic_ai/models/openai.py

Lines 1389 to 1394 in 99581d7

	elif isinstance(chunk, responses.ResponseReasoningSummaryPartAddedEvent):
	yield self._parts_manager.handle_thinking_delta(
	vendor_part_id=f'{chunk.item_id}-{chunk.summary_index}',
	content=chunk.part.text,
	id=chunk.item_id,
	)

I can still make this more robust based on the issues you've already identified, but it'd be great to have a reproducible example.

[Joakim-T]

@DouweM

Are you running against a non-OpenAI Responses API? Or the one from gpt-oss by any chance? It would be very useful to see the full API response.

Yes and i should have mentioned that.
I'm using GPT-5 in Azures new(ish) v1 API that is supposed to be 1:1 with OpenAI, or so they claim.
https://learn.microsoft.com/en-us/azure/ai-foundry/openai/api-version-lifecycle?tabs=key#api-evolution

Supports using OpenAI() client instead of AzureOpenAI()

I reverted to pydantic-ai-slim 1.0.2 with your TODO's uncommented, but i can reinstall 1.0.3 and get full output tomorrow.

[DouweM]

@Joakim-T Can you try 1.0.4 which I released earlier today?

I wonder if that API is including the (raw or summarized) CoT in reasoning_text parts rather than the reasoning_summary parts we currently support:

pydantic-ai/pydantic_ai_slim/pydantic_ai/models/openai.py

Line 874 in ec87153

# NOTE: We don't currently handle the raw CoT from gpt-oss `reasoning_text`: https://cookbook.openai.com/articles/gpt-oss/handle-raw-cot

[DouweM]

@Joakim-T This seems to be the relevant bit in their docs https://learn.microsoft.com/en-us/azure/ai-foundry/openai/how-to/reasoning?tabs=gpt-5%2Cpython%2Cpy#reasoning-summary:

Even when enabled, reasoning summaries are not guaranteed to be generated for every step/request. This is expected behavior.

I was always getting a summary from the regular OpenAI Responses API, but it's better to not rely on that.

[yf-yang]

v1.0.5 fixes it. Yes it is not reliable as it can be easily reproduced with "minimal" setting:

import asyncio

from dotenv import load_dotenv
from pydantic_ai import (
  Agent,
)
from pydantic_ai.models.openai import OpenAIResponsesModel, OpenAIResponsesModelSettings

load_dotenv()

agent = Agent(
  model=OpenAIResponsesModel("gpt-5-nano"),
  model_settings=OpenAIResponsesModelSettings(
    openai_reasoning_effort="minimal",
    openai_service_tier="flex",
    max_tokens=8192,
    timeout=15,
  ),
)


async def main():
  async with agent.iter("hello") as run:
    async for node in run:
      print(node)
      if Agent.is_model_request_node(node):
        async with node.stream(run.ctx) as response_stream:
          async for event in response_stream:
            print(event)


if __name__ == "__main__":
  asyncio.run(main())