Ignore leading whitespace when streaming text, fixing run_stream + Ollama + Qwen3 (#2294)

DouweM · KRRT7 · commit d8086c14a745 · 2025-07-24T15:34:36.000-07:00
diff --git a/pydantic_ai_slim/pydantic_ai/_parts_manager.py b/pydantic_ai_slim/pydantic_ai/_parts_manager.py
@@ -15,7 +15,7 @@
 
 from collections.abc import Hashable
 from dataclasses import dataclass, field, replace
-from typing import Any, Literal, Union, overload
+from typing import Any, Union
 
 from pydantic_ai._thinking_part import END_THINK_TAG, START_THINK_TAG
 from pydantic_ai.exceptions import UnexpectedModelBehavior
@@ -67,23 +67,6 @@ def get_parts(self) -> list[ModelResponsePart]:
         """
         return [p for p in self._parts if not isinstance(p, ToolCallPartDelta)]
 
-    @overload
-    def handle_text_delta(
-        self,
-        *,
-        vendor_part_id: VendorId | None,
-        content: str,
-    ) -> ModelResponseStreamEvent: ...
-
-    @overload
-    def handle_text_delta(
-        self,
-        *,
-        vendor_part_id: VendorId,
-        content: str,
-        extract_think_tags: Literal[True],
-    ) -> ModelResponseStreamEvent | None: ...
-
     def handle_text_delta(
         self,
         *,
@@ -105,7 +88,9 @@ def handle_text_delta(
             extract_think_tags: Whether to extract `<think>` tags from the text content and handle them as thinking parts.
 
         Returns:
-            A `PartStartEvent` if a new part was created, or a `PartDeltaEvent` if an existing part was updated.
+            - A `PartStartEvent` if a new part was created.
+            - A `PartDeltaEvent` if an existing part was updated.
+            - `None` if no new event is emitted (e.g., the first text part was all whitespace).
 
         Raises:
             UnexpectedModelBehavior: If attempting to apply text content to a part that is not a TextPart.
@@ -144,6 +129,12 @@ def handle_text_delta(
             return self.handle_thinking_delta(vendor_part_id=vendor_part_id, content='')
 
         if existing_text_part_and_index is None:
+            # If the first text delta is all whitespace, don't emit a new part yet.
+            # This is a workaround for models that emit `<think>\n</think>\n\n` ahead of tool calls (e.g. Ollama + Qwen3),
+            # which we don't want to end up treating as a final result.
+            if content.isspace():
+                return None
+
             # There is no existing text part that should be updated, so create a new one
             new_part_index = len(self._parts)
             part = TextPart(content=content)
diff --git a/pydantic_ai_slim/pydantic_ai/messages.py b/pydantic_ai_slim/pydantic_ai/messages.py
@@ -690,7 +690,7 @@ class ThinkingPart:
 
     def has_content(self) -> bool:
         """Return `True` if the thinking content is non-empty."""
-        return bool(self.content)  # pragma: no cover
+        return bool(self.content)
 
     __repr__ = _utils.dataclasses_no_defaults_repr
 
diff --git a/pydantic_ai_slim/pydantic_ai/models/anthropic.py b/pydantic_ai_slim/pydantic_ai/models/anthropic.py
@@ -470,7 +470,7 @@ class AnthropicStreamedResponse(StreamedResponse):
     _response: AsyncIterable[BetaRawMessageStreamEvent]
     _timestamp: datetime
 
-    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
+    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:  # noqa: C901
         current_block: BetaContentBlock | None = None
 
         async for event in self._response:
@@ -479,7 +479,11 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
             if isinstance(event, BetaRawContentBlockStartEvent):
                 current_block = event.content_block
                 if isinstance(current_block, BetaTextBlock) and current_block.text:
-                    yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=current_block.text)
+                    maybe_event = self._parts_manager.handle_text_delta(
+                        vendor_part_id='content', content=current_block.text
+                    )
+                    if maybe_event is not None:  # pragma: no branch
+                        yield maybe_event
                 elif isinstance(current_block, BetaThinkingBlock):
                     yield self._parts_manager.handle_thinking_delta(
                         vendor_part_id='thinking',
@@ -498,7 +502,11 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
 
             elif isinstance(event, BetaRawContentBlockDeltaEvent):
                 if isinstance(event.delta, BetaTextDelta):
-                    yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=event.delta.text)
+                    maybe_event = self._parts_manager.handle_text_delta(
+                        vendor_part_id='content', content=event.delta.text
+                    )
+                    if maybe_event is not None:  # pragma: no branch
+                        yield maybe_event
                 elif isinstance(event.delta, BetaThinkingDelta):
                     yield self._parts_manager.handle_thinking_delta(
                         vendor_part_id='thinking', content=event.delta.thinking
diff --git a/pydantic_ai_slim/pydantic_ai/models/bedrock.py b/pydantic_ai_slim/pydantic_ai/models/bedrock.py
@@ -572,7 +572,7 @@ class BedrockStreamedResponse(StreamedResponse):
     _event_stream: EventStream[ConverseStreamOutputTypeDef]
     _timestamp: datetime = field(default_factory=_utils.now_utc)
 
-    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
+    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:  # noqa: C901
         """Return an async iterator of [`ModelResponseStreamEvent`][pydantic_ai.messages.ModelResponseStreamEvent]s.
 
         This method should be implemented by subclasses to translate the vendor-specific stream of events into
@@ -618,7 +618,9 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                             UserWarning,
                         )
                 if 'text' in delta:
-                    yield self._parts_manager.handle_text_delta(vendor_part_id=index, content=delta['text'])
+                    maybe_event = self._parts_manager.handle_text_delta(vendor_part_id=index, content=delta['text'])
+                    if maybe_event is not None:
+                        yield maybe_event
                 if 'toolUse' in delta:
                     tool_use = delta['toolUse']
                     maybe_event = self._parts_manager.handle_tool_call_delta(
diff --git a/pydantic_ai_slim/pydantic_ai/models/function.py b/pydantic_ai_slim/pydantic_ai/models/function.py
@@ -264,7 +264,9 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
             if isinstance(item, str):
                 response_tokens = _estimate_string_tokens(item)
                 self._usage += usage.Usage(response_tokens=response_tokens, total_tokens=response_tokens)
-                yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=item)
+                maybe_event = self._parts_manager.handle_text_delta(vendor_part_id='content', content=item)
+                if maybe_event is not None:  # pragma: no branch
+                    yield maybe_event
             elif isinstance(item, dict) and item:
                 for dtc_index, delta in item.items():
                     if isinstance(delta, DeltaThinkingPart):
@@ -286,7 +288,7 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                             args=delta.json_args,
                             tool_call_id=delta.tool_call_id,
                         )
-                        if maybe_event is not None:
+                        if maybe_event is not None:  # pragma: no branch
                             yield maybe_event
                     else:
                         assert_never(delta)
diff --git a/pydantic_ai_slim/pydantic_ai/models/gemini.py b/pydantic_ai_slim/pydantic_ai/models/gemini.py
@@ -438,7 +438,11 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                 if 'text' in gemini_part:
                     # Using vendor_part_id=None means we can produce multiple text parts if their deltas are sprinkled
                     # amongst the tool call deltas
-                    yield self._parts_manager.handle_text_delta(vendor_part_id=None, content=gemini_part['text'])
+                    maybe_event = self._parts_manager.handle_text_delta(
+                        vendor_part_id=None, content=gemini_part['text']
+                    )
+                    if maybe_event is not None:  # pragma: no branch
+                        yield maybe_event
 
                 elif 'function_call' in gemini_part:
                     # Here, we assume all function_call parts are complete and don't have deltas.
diff --git a/pydantic_ai_slim/pydantic_ai/models/google.py b/pydantic_ai_slim/pydantic_ai/models/google.py
@@ -458,7 +458,9 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                     if part.thought:
                         yield self._parts_manager.handle_thinking_delta(vendor_part_id='thinking', content=part.text)
                     else:
-                        yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=part.text)
+                        maybe_event = self._parts_manager.handle_text_delta(vendor_part_id='content', content=part.text)
+                        if maybe_event is not None:  # pragma: no branch
+                            yield maybe_event
                 elif part.function_call:
                     maybe_event = self._parts_manager.handle_tool_call_delta(
                         vendor_part_id=uuid4(),
diff --git a/pydantic_ai_slim/pydantic_ai/models/mistral.py b/pydantic_ai_slim/pydantic_ai/models/mistral.py
@@ -601,7 +601,9 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                             tool_call_id=maybe_tool_call_part.tool_call_id,
                         )
                 else:
-                    yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=text)
+                    maybe_event = self._parts_manager.handle_text_delta(vendor_part_id='content', content=text)
+                    if maybe_event is not None:  # pragma: no branch
+                        yield maybe_event
 
             # Handle the explicit tool calls
             for index, dtc in enumerate(choice.delta.tool_calls or []):
diff --git a/pydantic_ai_slim/pydantic_ai/models/openai.py b/pydantic_ai_slim/pydantic_ai/models/openai.py
@@ -1144,7 +1144,11 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                 )
 
             elif isinstance(chunk, responses.ResponseTextDeltaEvent):
-                yield self._parts_manager.handle_text_delta(vendor_part_id=chunk.content_index, content=chunk.delta)
+                maybe_event = self._parts_manager.handle_text_delta(
+                    vendor_part_id=chunk.content_index, content=chunk.delta
+                )
+                if maybe_event is not None:  # pragma: no branch
+                    yield maybe_event
 
             elif isinstance(chunk, responses.ResponseTextDoneEvent):
                 pass  # there's nothing we need to do here
diff --git a/pydantic_ai_slim/pydantic_ai/models/test.py b/pydantic_ai_slim/pydantic_ai/models/test.py
@@ -269,10 +269,14 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                     mid = len(text) // 2
                     words = [text[:mid], text[mid:]]
                 self._usage += _get_string_usage('')
-                yield self._parts_manager.handle_text_delta(vendor_part_id=i, content='')
+                maybe_event = self._parts_manager.handle_text_delta(vendor_part_id=i, content='')
+                if maybe_event is not None:  # pragma: no branch
+                    yield maybe_event
                 for word in words:
                     self._usage += _get_string_usage(word)
-                    yield self._parts_manager.handle_text_delta(vendor_part_id=i, content=word)
+                    maybe_event = self._parts_manager.handle_text_delta(vendor_part_id=i, content=word)
+                    if maybe_event is not None:  # pragma: no branch
+                        yield maybe_event
             elif isinstance(part, ToolCallPart):
                 yield self._parts_manager.handle_tool_call_part(
                     vendor_part_id=i, tool_name=part.tool_name, args=part.args, tool_call_id=part.tool_call_id
diff --git a/tests/models/test_bedrock.py b/tests/models/test_bedrock.py
@@ -1127,15 +1127,9 @@ async def test_bedrock_model_thinking_part_stream(allow_model_requests: None, be
             PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' all together.\n')),
             PartStartEvent(
                 index=1,
-                part=TextPart(
-                    content="""\
-
-
-"""
-                ),
+                part=TextPart(content='Crossing the'),
             ),
             FinalResultEvent(tool_name=None, tool_call_id=None),
-            PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='Crossing the')),
             PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' street safely involves')),
             PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' careful')),
             PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' observation')),
diff --git a/tests/models/test_groq.py b/tests/models/test_groq.py
@@ -934,17 +934,16 @@ async def test_groq_model_thinking_part_iter(allow_model_requests: None, groq_ap
                 7: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' come')),
                 8: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' up')),
                 9: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' with')),
-                589: PartStartEvent(index=1, part=TextPart(content='\n\n')),
+                589: PartStartEvent(index=1, part=TextPart(content='**')),
                 590: FinalResultEvent(tool_name=None, tool_call_id=None),
-                591: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='**')),
-                592: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='Ur')),
-                593: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ugu')),
-                594: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ayan')),
-                595: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' Alf')),
-                596: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='aj')),
-                597: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ores')),
-                598: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' Recipe')),
+                591: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='Ur')),
+                592: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ugu')),
+                593: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ayan')),
+                594: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' Alf')),
+                595: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='aj')),
+                596: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ores')),
+                597: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' Recipe')),
             },
-            length=997,
+            length=996,
         )
     )
diff --git a/tests/models/test_huggingface.py b/tests/models/test_huggingface.py
@@ -1031,17 +1031,16 @@ async def test_hf_model_thinking_part_iter(allow_model_requests: None, huggingfa
                 5: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' user')),
                 6: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' is')),
                 7: PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=' asking')),
-                413: PartStartEvent(index=1, part=TextPart(content='\n\n')),
+                413: PartStartEvent(index=1, part=TextPart(content='Cross')),
                 414: FinalResultEvent(tool_name=None, tool_call_id=None),
-                415: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='Cross')),
-                416: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ing')),
-                417: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' the')),
-                418: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' street')),
-                419: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' safely')),
-                420: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' requires')),
-                421: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' attent')),
-                422: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='iveness')),
+                415: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='ing')),
+                416: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' the')),
+                417: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' street')),
+                418: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' safely')),
+                419: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' requires')),
+                420: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta=' attent')),
+                421: PartDeltaEvent(index=1, delta=TextPartDelta(content_delta='iveness')),
             },
-            length=1060,
+            length=1059,
         )
     )
diff --git a/tests/models/test_instrumented.py b/tests/models/test_instrumented.py
@@ -96,8 +96,12 @@ async def request_stream(
 class MyResponseStream(StreamedResponse):
     async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
         self._usage = Usage(request_tokens=300, response_tokens=400)
-        yield self._parts_manager.handle_text_delta(vendor_part_id=0, content='text1')
-        yield self._parts_manager.handle_text_delta(vendor_part_id=0, content='text2')
+        maybe_event = self._parts_manager.handle_text_delta(vendor_part_id=0, content='text1')
+        if maybe_event is not None:  # pragma: no branch
+            yield maybe_event
+        maybe_event = self._parts_manager.handle_text_delta(vendor_part_id=0, content='text2')
+        if maybe_event is not None:  # pragma: no branch
+            yield maybe_event
 
     @property
     def model_name(self) -> str:
diff --git a/tests/models/test_openai.py b/tests/models/test_openai.py
@@ -611,6 +611,38 @@ async def test_stream_tool_call_with_empty_text(allow_model_requests: None):
     assert await result.get_output() == snapshot({'first': 'One', 'second': 'Two'})
 
 
+async def test_stream_text_empty_think_tag_and_text_before_tool_call(allow_model_requests: None):
+    # Ollama + Qwen3 will emit `<think>\n</think>\n\n` ahead of tool calls,
+    # which we don't want to end up treating as a final result.
+    stream = [
+        text_chunk('<think>'),
+        text_chunk('\n'),
+        text_chunk('</think>'),
+        text_chunk('\n\n'),
+        struc_chunk('final_result', None),
+        struc_chunk(None, '{"first": "One'),
+        struc_chunk(None, '", "second": "Two"'),
+        struc_chunk(None, '}'),
+        chunk([]),
+    ]
+    mock_client = MockOpenAI.create_mock_stream(stream)
+    m = OpenAIModel('gpt-4o', provider=OpenAIProvider(openai_client=mock_client))
+    agent = Agent(m, output_type=[str, MyTypedDict])
+
+    async with agent.run_stream('') as result:
+        assert not result.is_complete
+        assert [c async for c in result.stream(debounce_by=None)] == snapshot(
+            [
+                {},
+                {'first': 'One'},
+                {'first': 'One', 'second': 'Two'},
+                {'first': 'One', 'second': 'Two'},
+                {'first': 'One', 'second': 'Two'},
+            ]
+        )
+    assert await result.get_output() == snapshot({'first': 'One', 'second': 'Two'})
+
+
 async def test_no_content(allow_model_requests: None):
     stream = [chunk([ChoiceDelta()]), chunk([ChoiceDelta()])]
     mock_client = MockOpenAI.create_mock_stream(stream)
diff --git a/tests/test_a2a.py b/tests/test_a2a.py
