Repository: pydantic/pydantic-ai
File: pydantic_ai_slim/pydantic_ai/result.py (Python)

Code before:
        if self._cached_output is not None:
            yield self._cached_output
            return

        last_response: _messages.ModelResponse | None = None
        async for response in self.stream_responses(debounce_by=debounce_by):
            if self._raw_stream_response.final_result_event is None or (
                last_response and response.parts == last_response.parts
            ):
                continue
            last_response = response

            try:
                yield await self.validate_response_output(response, allow_partial=True)
            except ValidationError:
                pass

        # Yield final validation and cache the result
        response = self.response
        if self._raw_stream_response.final_result_event is not None and (
            not last_response or response.parts != last_response.parts
        ):
            # Final validation (allow_partial=False)
            final_output = await self.validate_response_output(response, allow_partial=False)
            self._cached_output = final_output
            yield final_output

    async def stream_responses(self, *, debounce_by: float | None = 0.1) -> AsyncIterator[_messages.ModelResponse]:
        """Asynchronously stream the (unvalidated) model responses for the agent."""
        # if the message currently has any parts with content, yield before streaming
        msg = self.response
        for part in msg.parts:
            if part.has_content():
                yield msg
                break

        async with _utils.group_by_temporal(self, debounce_by) as group_iter:
            async for _items in group_iter:
                yield self.response  # current state of the response

    async def stream_text(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> AsyncIterator[str]:
        """Stream the text result as an async iterable.

        !!! note
            Result validators will NOT be called on the text result if `delta=True`.

        Args:
            delta: if `True`, yield each chunk of text as it is received, if `False` (default), yield the full text
                up to the current point.
            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
                Debouncing is particularly important for long structured responses to reduce the overhead of

Review comment (performance):
I'd suggest we make this:

```python
            self._cached_output = await self.validate_response_output(response, allow_partial=False)
            yield final_output
```

but I'm also worried that we're passing a reference to the original object, so if the user manipulates it they'll mutate the cached value! can you add a test for that?

Code after:
        if self._cached_output is not None:
            yield deepcopy(self._cached_output)
            return

        last_response: _messages.ModelResponse | None = None
        async for response in self.stream_responses(debounce_by=debounce_by):
            if self._raw_stream_response.final_result_event is None or (
                last_response and response.parts == last_response.parts
            ):
                continue
            last_response = response

            try:
                yield await self.validate_response_output(response, allow_partial=True)
            except ValidationError:
                pass

        # Yield final validation and cache the result
        response = self.response
        if self._raw_stream_response.final_result_event is not None and (
            not last_response or response.parts != last_response.parts
        ):
            # Final validation (allow_partial=False)
            self._cached_output = await self.validate_response_output(response, allow_partial=False)
            yield deepcopy(self._cached_output)

    async def stream_responses(self, *, debounce_by: float | None = 0.1) -> AsyncIterator[_messages.ModelResponse]:
        """Asynchronously stream the (unvalidated) model responses for the agent."""
        # if the message currently has any parts with content, yield before streaming
        msg = self.response
        for part in msg.parts:
            if part.has_content():
                yield msg
                break

        async with _utils.group_by_temporal(self, debounce_by) as group_iter:
            async for _items in group_iter:
                yield self.response  # current state of the response

    async def stream_text(self, *, delta: bool = False, debounce_by: float | None = 0.1) -> AsyncIterator[str]:
        """Stream the text result as an async iterable.

        !!! note
            Result validators will NOT be called on the text result if `delta=True`.

        Args:
            delta: if `True`, yield each chunk of text as it is received, if `False` (default), yield the full text
                up to the current point.
            debounce_by: by how much (if at all) to debounce/group the response chunks by. `None` means no debouncing.
                Debouncing is particularly important for long structured responses to reduce the overhead of
                performing validation as each token is received.