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

Code before:
            ),
        }

    def _get_response_id_and_trim(
        self, messages: list[ModelMessage]
    ) -> tuple[list[ModelMessage], str | NotGiven | None]:
        # If the message history contains only openai responses,
        # we can limit the history to the most recent ModelRequest.
        # The provider_response_id from the latest ModelResponse is
        # then passed as previous_response_id to preserve context.
        response_id = NOT_GIVEN
        latest_model_response: list[ModelMessage] = []
        for m in messages:
            # Openai may return a dated model_name that differs from self.model_name
            # (e.g., "gpt-5" vs "gpt-5-2025-08-07").
            if isinstance(m, ModelResponse) and (self.model_name in m.model_name):  # type: ignore
                response_id = m.provider_response_id
            elif isinstance(m, ModelRequest):
                latest_model_response = [m]
            else:
                # Mixed model responses invalidate response_id,
                # so the history is kept intact.
                response_id = NOT_GIVEN
                break
        if response_id:
            messages = latest_model_response
        return messages, response_id

    async def _map_messages(
        self, messages: list[ModelMessage]
    ) -> tuple[str | NotGiven, list[responses.ResponseInputItemParam]]:
        """Just maps a `pydantic_ai.Message` to a `openai.types.responses.ResponseInputParam`."""
        openai_messages: list[responses.ResponseInputItemParam] = []
        for message in messages:
            if isinstance(message, ModelRequest):
                for part in message.parts:
                    if isinstance(part, SystemPromptPart):
                        openai_messages.append(responses.EasyInputMessageParam(role='system', content=part.content))
                    elif isinstance(part, UserPromptPart):
                        openai_messages.append(await self._map_user_prompt(part))
                    elif isinstance(part, ToolReturnPart):
                        openai_messages.append(
                            FunctionCallOutput(
                                type='function_call_output',
                                call_id=_guard_tool_call_id(t=part),
                                output=part.model_response_str(),
                            )
                        )
                    elif isinstance(part, RetryPromptPart):
                        # TODO(Marcelo): How do we test this conditional branch?
                        if part.tool_name is None:  # pragma: no cover

Review comment (suggestion):
`latest_model_response` doesn't need to be a list right? We can make this `messages = [latest_model_response]`

Code after:
            openai_messages.insert(0, responses.EasyInputMessageParam(role='system', content=instructions))
            instructions = NOT_GIVEN

        if verbosity := model_settings.get('openai_text_verbosity'):
            text = text or {}
            text['verbosity'] = verbosity

        profile = OpenAIModelProfile.from_profile(self.profile)
        unsupported_model_settings = profile.openai_unsupported_model_settings
        for setting in unsupported_model_settings:
            model_settings.pop(setting, None)

        include: list[responses.ResponseIncludable] | None = None
        if profile.openai_supports_encrypted_reasoning_content:
            include = ['reasoning.encrypted_content']

        try:
            extra_headers = model_settings.get('extra_headers', {})
            extra_headers.setdefault('User-Agent', get_user_agent())
            return await self.client.responses.create(
                input=openai_messages,
                model=self._model_name,
                instructions=instructions,
                parallel_tool_calls=model_settings.get('parallel_tool_calls', NOT_GIVEN),
                tools=tools or NOT_GIVEN,
                tool_choice=tool_choice or NOT_GIVEN,
                max_output_tokens=model_settings.get('max_tokens', NOT_GIVEN),
                stream=stream,
                temperature=model_settings.get('temperature', NOT_GIVEN),
                top_p=model_settings.get('top_p', NOT_GIVEN),
                truncation=model_settings.get('openai_truncation', NOT_GIVEN),
                timeout=model_settings.get('timeout', NOT_GIVEN),
                service_tier=model_settings.get('openai_service_tier', NOT_GIVEN),
                previous_response_id=previous_response_id,
                reasoning=reasoning,
                user=model_settings.get('openai_user', NOT_GIVEN),
                text=text or NOT_GIVEN,
                include=include or NOT_GIVEN,
                extra_headers=extra_headers,
                extra_body=model_settings.get('extra_body'),
            )
        except APIStatusError as e:
            if (status_code := e.status_code) >= 400:
                raise ModelHTTPError(status_code=status_code, model_name=self.model_name, body=e.body) from e
            raise  # pragma: lax no cover

    def _get_reasoning(self, model_settings: OpenAIResponsesModelSettings) -> Reasoning | NotGiven:
        reasoning_effort = model_settings.get('openai_reasoning_effort', None)
        reasoning_summary = model_settings.get('openai_reasoning_summary', None)
        reasoning_generate_summary = model_settings.get('openai_reasoning_generate_summary', None)