From dff474e450b9f8cf94d3fad508ab6c789528a63e Mon Sep 17 00:00:00 2001 From: simonp9 <349053457+simonp9@users.noreply.github.com> Date: Thu, 29 Oct 2026 00:56:02 -0700 Subject: [PATCH] fix(responses): replay the original Responses item order, phase, and boundaries on the next chat turn (#48912) * fix(responses): replay the original Responses item order, phase, and boundaries on the next chat turn * fix(responses): keep tool-call-only stream content empty and drop unreachable replay guards * fix(responses): reject an assistant phase OpenAI does not define unless drop_params is set * fix(responses): keep other providers' reasoning and stream output unchanged and fall back on records that do not fit * fix(responses): label a structured output turn with its returned message's phase and keep lone surrogates from failing the checksum * fix(responses): leave a turn without a phase when its content message cannot be read * test(responses): add black box tests for chat turn history replay * test(responses): expect the replay record on bridged chat replies * test(responses): expect the replay record in bridged chat translation cases * fix(streaming): keep the served service tier on the finish chunk the stream adds * test(responses): read the expected chat message without JSON validation so the replay matcher stays --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> --- .../handler.py | 2 + .../history.py | 432 ++++++++++++ .../transformation.py | 185 +++++- .../litellm_core_utils/streaming_handler.py | 10 + litellm/tool_loop.py | 4 +- .../integration/_support/responses_replay.py | 214 ++++++ ...st_responses_bridge_history_replay_wire.py | 233 +++++++ .../test_responses_bridge_incomplete.py | 15 +- ..._tool_loop_responses_history_replay_sdk.py | 37 ++ .../bases/openai_responses.py | 13 +- .../translation/response_scenarios.py | 12 +- ...itellm_responses_transformation_history.py | 627 ++++++++++++++++++ ...responses_transformation_transformation.py | 45 +- .../test_streaming_handler.py | 32 + tests/unit/test_tool_loop.py | 66 ++ 15 files changed, 1882 insertions(+), 45 deletions(-) create mode 100644 litellm/completion_extras/litellm_responses_transformation/history.py create mode 100644 tests/integration/_support/responses_replay.py create mode 100644 tests/integration/providers/test_responses_bridge_history_replay_wire.py create mode 100644 tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py create mode 100644 tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index 3f7017eccc2..76d17ea9693 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -260,6 +260,7 @@ class ResponsesToCompletionBridgeHandler: structured_output=is_structured_output_request( cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict ), + replays_openai_history=custom_llm_provider == "openai", ) streamwrapper: Final = CustomStreamWrapper( completion_stream=completion_stream, @@ -363,6 +364,7 @@ class ResponsesToCompletionBridgeHandler: structured_output=is_structured_output_request( cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict ), + replays_openai_history=custom_llm_provider == "openai", ) streamwrapper: Final = CustomStreamWrapper( completion_stream=completion_stream, diff --git a/litellm/completion_extras/litellm_responses_transformation/history.py b/litellm/completion_extras/litellm_responses_transformation/history.py new file mode 100644 index 00000000000..4435de64cde --- /dev/null +++ b/litellm/completion_extras/litellm_responses_transformation/history.py @@ -0,0 +1,432 @@ +import hashlib +import json +from collections.abc import Iterator, Mapping, Sequence +from dataclasses import dataclass +from itertools import accumulate, chain, pairwise +from types import MappingProxyType +from typing import Annotated, Final, Literal, TypeAlias + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError + +REPLAY_FIELD: Final = "openai_responses_replay" +PHASE_FIELD: Final = "phase" + +Phase: TypeAlias = Literal["commentary", "final_answer"] + + +class _Frozen(BaseModel): + model_config = ConfigDict(frozen=True) + + +class _OutputText(_Frozen): + type: Literal["output_text"] + text: str + + +class _OutputMessage(_Frozen): + type: Literal["message"] + role: Literal["assistant"] + content: tuple[_OutputText, ...] + phase: Phase | None = None + + +class _OutputReasoning(_Frozen): + type: Literal["reasoning"] + id: str + + +class _OutputToolCall(_Frozen): + type: Literal["function_call", "custom_tool_call"] + call_id: str + id: str | None = None + + +class _OutputToolSearch(_Frozen): + type: Literal["tool_search_call", "tool_search_output"] + + +_OutputItem: TypeAlias = Annotated[ + _OutputMessage | _OutputReasoning | _OutputToolCall | _OutputToolSearch, Field(discriminator="type") +] +_OUTPUT_ITEM: Final[TypeAdapter[_OutputItem]] = TypeAdapter(_OutputItem) + + +class _AnyMessage(_Frozen): + type: Literal["message"] + + +class _MessageEntry(_Frozen): + type: Literal["message"] + parts: tuple[Annotated[int, Field(ge=0)], ...] + phase: Phase | None = None + + +class _IndexedEntry(_Frozen): + type: Literal["reasoning", "function_call", "custom_tool_call", "tool_search_call", "tool_search_output"] + index: Annotated[int, Field(ge=0)] + + +_Entry: TypeAlias = Annotated[_MessageEntry | _IndexedEntry, Field(discriminator="type")] + + +class _Replay(_Frozen): + version: Literal[1] + items: tuple[_Entry, ...] + checksum: str + + +class _Namespaced(_Frozen): + namespace: str | None = None + + +class _ToolFunction(_Frozen): + name: str = "" + arguments: str = "" + provider_specific_fields: _Namespaced | None = None + + +class _ToolCall(_Frozen): + id: str + function: _ToolFunction + provider_specific_fields: _Namespaced | None = None + + +class _SummaryText(_Frozen): + type: str = "summary_text" + text: str = "" + + +class _Reasoning(_Frozen): + id: str | None = None + encrypted_content: str | None = None + summary: tuple[_SummaryText, ...] | None = None + + +class _ProviderFields(_Frozen): + phase: Phase | None = None + tool_search_items: tuple[Mapping[str, object], ...] | None = None + openai_responses_replay: object = None + + +class _ChatMessage(_Frozen): + role: str | None = None + content: str | None = None + tool_calls: tuple[_ToolCall, ...] | None = None + reasoning_items: tuple[_Reasoning, ...] | None = None + phase: Phase | None = None + provider_specific_fields: _ProviderFields | None = None + + +class _PhaseFields(_Frozen): + phase: Phase | None = None + + +class _RequestedPhaseFields(_Frozen): + phase: object = None + + +class _RequestedPhase(_Frozen): + phase: object = None + provider_specific_fields: _RequestedPhaseFields | None = None + + +@dataclass(frozen=True, slots=True) +class UnknownPhase: + value: object + + +_PHASE: Final[TypeAdapter[Phase]] = TypeAdapter(Phase) + + +class _RetainedFields(_PhaseFields): + openai_responses_replay: object = None + + +@dataclass(frozen=True, slots=True) +class ReplayedTurn: + items: tuple[Mapping[str, object], ...] + custom_call_ids: frozenset[str] + + +def _as_json(value: object) -> object: + return value.model_dump() if isinstance(value, BaseModel) else value + + +def _parsed_output_item(item: object) -> _OutputItem | None: + try: + return _OUTPUT_ITEM.validate_python(_as_json(item)) + except ValidationError: + return None + + +def _is_message(item: object) -> bool: + try: + _AnyMessage.model_validate(_as_json(item)) + except ValidationError: + return False + return True + + +def _chat_message(message: object) -> _ChatMessage | None: + try: + return _ChatMessage.model_validate(_as_json(message)) + except ValidationError: + return None + + +def requested_phase(message: object) -> Phase | UnknownPhase | None: + try: + requested: Final = _RequestedPhase.model_validate(_as_json(message)) + except ValidationError: + return None + fields: Final = requested.provider_specific_fields + value: Final = requested.phase or (fields.phase if fields else None) + if value is None: + return None + try: + return _PHASE.validate_python(value) + except ValidationError: + return UnknownPhase(value) + + +def reasoning_input_item( + item_id: str | None, summary: Sequence[object] | None, encrypted_content: str | None +) -> Mapping[str, object]: + return { + "type": "reasoning", + **({"id": item_id} if item_id else {}), + "summary": list(summary or ()), + **({"encrypted_content": encrypted_content} if encrypted_content else {}), + } + + +def retained_replay_fields(provider_specific_fields: object) -> Mapping[str, object] | None: + try: + retained: Final = _RetainedFields.model_validate(provider_specific_fields) + except ValidationError: + return None + return retained.model_dump(exclude_none=True) or None + + +def output_message_text(output_items: Sequence[object], *, first_message_only: bool) -> str: + parsed: Final = map(_parsed_output_item, output_items) + messages: Final = tuple(item for item in parsed if isinstance(item, _OutputMessage)) + return "".join(_texts(messages[:1] if first_message_only else messages)) + + +def _tool_search_entries(message: _ChatMessage) -> tuple[Mapping[str, object], ...]: + fields: Final = message.provider_specific_fields + items: Final = (fields.tool_search_items if fields else None) or () + return tuple(item for item in items if item.get("type") in ("tool_search_call", "tool_search_output")) + + +def _checksum(message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedEntry]) -> str: + fields: Final = message.provider_specific_fields + bound: Final = { + "content": message.content or "", + "tool_calls": [[call.id, call.function.name, call.function.arguments] for call in message.tool_calls or ()], + "reasoning": [ + [item.id, item.encrypted_content, [part.text for part in item.summary or ()]] + for item in message.reasoning_items or () + ], + "tool_search_items": list((fields.tool_search_items if fields else None) or ()), + "phase": message.phase or (fields.phase if fields else None), + "items": [entry.model_dump(mode="json", exclude_none=True) for entry in entries], + } + encoded: Final = json.dumps(bound, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return hashlib.sha256(encoded.encode("utf-8", "surrogatepass")).hexdigest() + + +def _kind(item: _OutputItem) -> str: + match item: + case _OutputToolCall(): + return "tool_call" + case _OutputToolSearch(): + return "tool_search" + case _OutputMessage() | _OutputReasoning(): + return item.type + + +def _entry(items: Sequence[_OutputItem], position: int) -> _MessageEntry | _IndexedEntry: + item: Final = items[position] + if isinstance(item, _OutputMessage): + return _MessageEntry(type="message", parts=tuple(len(part.text) for part in item.content), phase=item.phase) + ordinal: Final = sum(1 for earlier in items[:position] if _kind(earlier) == _kind(item)) + return _IndexedEntry(type=item.type, index=ordinal) + + +def _texts(items: Sequence[_OutputItem]) -> Iterator[str]: + for item in items: + if isinstance(item, _OutputMessage): + yield from (part.text for part in item.content) + + +def _matches(items: Sequence[_OutputItem], message: _ChatMessage) -> bool: + reasoning_ids: Final = tuple(item.id for item in items if isinstance(item, _OutputReasoning)) + calls: Final = tuple(item for item in items if isinstance(item, _OutputToolCall)) + tool_calls: Final = message.tool_calls or () + searches: Final = sum(1 for item in items if isinstance(item, _OutputToolSearch)) + return ( + "".join(_texts(items)) == (message.content or "") + and reasoning_ids == tuple(item.id for item in message.reasoning_items or ()) + and len(calls) == len(tool_calls) + and all(chat.id in (call.call_id, call.id) for call, chat in zip(calls, tool_calls)) + and searches == len(_tool_search_entries(message)) + ) + + +def _content_phase( + output_items: Sequence[object], parsed: Sequence[_OutputItem | None], *, first_message_only: bool +) -> Phase | None: + messages: Final = tuple(item for raw, item in zip(output_items, parsed) if _is_message(raw)) + if not messages: + return None + content_message: Final = messages[0] if first_message_only else messages[-1] + return content_message.phase if isinstance(content_message, _OutputMessage) else None + + +def _with_phase(message: _ChatMessage, phase: Phase | None) -> _ChatMessage: + if phase is None: + return message + fields: Final = (message.provider_specific_fields or _ProviderFields()).model_copy(update={PHASE_FIELD: phase}) + return message.model_copy(update={"provider_specific_fields": fields}) + + +def replay_fields( + output_items: Sequence[object], message: object, *, completed: bool, first_message_only: bool +) -> Mapping[str, object]: + parsed: Final = tuple(map(_parsed_output_item, output_items)) + phase: Final = _content_phase(output_items, parsed, first_message_only=first_message_only) + phase_fields: Final[Mapping[str, object]] = MappingProxyType({PHASE_FIELD: phase} if phase else {}) + items: Final = tuple(item for item in parsed if item is not None) + parsed_message: Final = _chat_message(message) + chat: Final = _with_phase(parsed_message, phase) if parsed_message is not None else None + if not completed or not items or len(items) != len(parsed) or chat is None or not _matches(items, chat): + return phase_fields + entries: Final = tuple(_entry(items, position) for position in range(len(items))) + replay: Final = _Replay(version=1, items=entries, checksum=_checksum(chat, entries)) + return MappingProxyType({**phase_fields, REPLAY_FIELD: replay.model_dump(mode="json", exclude_none=True)}) + + +def _stored_replay(message: _ChatMessage) -> _Replay | None: + fields: Final = message.provider_specific_fields + if fields is None or fields.openai_responses_replay is None: + return None + try: + return _Replay.model_validate(fields.openai_responses_replay) + except ValidationError: + return None + + +def _message_texts(entries: Sequence[_MessageEntry], content: str) -> tuple[tuple[str, ...], ...]: + lengths: Final = tuple(chain.from_iterable(entry.parts for entry in entries)) + bounds: Final = tuple(accumulate(lengths, initial=0)) + parts: Final = tuple(content[start:end] for start, end in pairwise(bounds)) + groups: Final = tuple(accumulate((len(entry.parts) for entry in entries), initial=0)) + return tuple(parts[start:end] for start, end in pairwise(groups)) + + +def _message_item(texts: Sequence[str], phase: Phase | None) -> Mapping[str, object]: + return { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": text} for text in texts], + **({PHASE_FIELD: phase} if phase else {}), + } + + +def _function_call_item(call: _ToolCall) -> Mapping[str, object]: + namespaced: Final = call.provider_specific_fields or call.function.provider_specific_fields + namespace: Final = namespaced.namespace if namespaced else None + return { + "type": "function_call", + "call_id": call.id, + "name": call.function.name, + "arguments": call.function.arguments, + **({"namespace": namespace} if namespace else {}), + } + + +def _custom_tool_call_item(call: _ToolCall) -> Mapping[str, object]: + return { + "type": "custom_tool_call", + "call_id": call.id, + "name": call.function.name, + "input": call.function.arguments, + } + + +def _reasoning_item(item: _Reasoning) -> Mapping[str, object]: + summary: Final = tuple(part.model_dump() for part in item.summary or ()) + return reasoning_input_item(item.id, summary, item.encrypted_content) + + +@dataclass(frozen=True, slots=True) +class _Resolved: + message: _ChatMessage + texts: Mapping[int, tuple[str, ...]] + tool_searches: tuple[Mapping[str, object], ...] + + +def _input_item(resolved: _Resolved, position: int, entry: _MessageEntry | _IndexedEntry) -> Mapping[str, object]: + if isinstance(entry, _MessageEntry): + return _message_item(resolved.texts[position], entry.phase) + tool_calls: Final = resolved.message.tool_calls or () + match entry.type: + case "reasoning": + return _reasoning_item((resolved.message.reasoning_items or ())[entry.index]) + case "function_call": + return _function_call_item(tool_calls[entry.index]) + case "custom_tool_call": + return _custom_tool_call_item(tool_calls[entry.index]) + case "tool_search_call" | "tool_search_output": + return dict(resolved.tool_searches[entry.index]) + + +def _fits(message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedEntry]) -> bool: + tool_calls: Final = len(message.tool_calls or ()) + tool_searches: Final = len(_tool_search_entries(message)) + sizes: Final = MappingProxyType( + { + "reasoning": len(message.reasoning_items or ()), + "function_call": tool_calls, + "custom_tool_call": tool_calls, + "tool_search_call": tool_searches, + "tool_search_output": tool_searches, + } + ) + parts: Final = chain.from_iterable(entry.parts for entry in entries if isinstance(entry, _MessageEntry)) + indexed: Final = (entry for entry in entries if isinstance(entry, _IndexedEntry)) + return sum(parts) == len(message.content or "") and all(entry.index < sizes[entry.type] for entry in indexed) + + +def _resolve( + message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedEntry], replay_tool_search: bool +) -> ReplayedTurn | None: + indexed: Final = tuple(entry for entry in entries if isinstance(entry, _IndexedEntry)) + if not replay_tool_search and any(entry.type in ("tool_search_call", "tool_search_output") for entry in indexed): + return None + messages: Final = tuple(entry for entry in entries if isinstance(entry, _MessageEntry)) + calls: Final = tuple(entry for entry in indexed if entry.type in ("function_call", "custom_tool_call")) + texts: Final = _message_texts(messages, message.content or "") + positions: Final = tuple(position for position, entry in enumerate(entries) if isinstance(entry, _MessageEntry)) + resolved: Final = _Resolved( + message=message, + texts=MappingProxyType(dict(zip(positions, texts, strict=True))), + tool_searches=_tool_search_entries(message), + ) + tool_calls: Final = message.tool_calls or () + return ReplayedTurn( + items=tuple(_input_item(resolved, position, entry) for position, entry in enumerate(entries)), + custom_call_ids=frozenset(tool_calls[entry.index].id for entry in calls if entry.type == "custom_tool_call"), + ) + + +def replay_items(message: object, *, replay_tool_search: bool) -> ReplayedTurn | None: + chat: Final = _chat_message(message) + stored: Final = _stored_replay(chat) if chat is not None and chat.role == "assistant" else None + if chat is None or stored is None or stored.checksum != _checksum(chat, stored.items): + return None + if not _fits(chat, stored.items): + return None + return _resolve(chat, stored.items, replay_tool_search) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index ee3639ac387..6173794a210 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -24,6 +24,15 @@ from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError import litellm from litellm import ModelResponse from litellm._logging import verbose_logger +from litellm.completion_extras.litellm_responses_transformation.history import ( + PHASE_FIELD, + UnknownPhase, + output_message_text, + reasoning_input_item, + replay_fields, + replay_items, + requested_phase, +) from litellm.integrations.anthropic_cache_control_hook import supports_openai_prompt_cache_breakpoint from litellm.litellm_core_utils.hidden_params import get_hidden_params, get_or_create_hidden_params from litellm.litellm_core_utils.prompt_templates.common_utils import ( @@ -490,7 +499,7 @@ def _fold_choices_into_one( choices: Sequence["Choices"], tool_calls: Sequence[Mapping[str, object]], trailing_reasoning_content: str | None, - trailing_reasoning_item: _BuiltReasoningItem | None, + trailing_reasoning_items: Sequence[_BuiltReasoningItem], tool_search_items: Sequence[Mapping[str, object]], structured_output: bool = False, ) -> "Choices": @@ -509,7 +518,7 @@ def _fold_choices_into_one( reasoning_contents: Final = (*(_message_reasoning_content(m) for m in messages), trailing_reasoning_content) reasoning_items: Final = ( *chain.from_iterable(_message_reasoning_items(m) for m in messages), - *(() if trailing_reasoning_item is None else (trailing_reasoning_item,)), + *trailing_reasoning_items, ) return Choices( message=Message( @@ -526,6 +535,61 @@ def _fold_choices_into_one( ) +def _with_replay_fields( + choice: "Choices", output_items: Sequence[object], *, completed: bool, structured_output: bool +) -> "Choices": + fields: Final = { + **(choice.message.provider_specific_fields or {}), + **replay_fields(output_items, choice.message, completed=completed, first_message_only=structured_output), + } + message: Final = choice.message.model_copy(update={"provider_specific_fields": fields or None}) + return choice.model_copy(update={"message": message}) + + +_STREAMED_OUTPUT_ITEMS: Final[TypeAdapter[tuple[Mapping[str, object], ...]]] = TypeAdapter( + tuple[Mapping[str, object], ...] +) + + +def completed_stream_history( + output_payload: object, + tool_search_fields: Mapping[str, object] | None, + *, + completed: bool, + structured_output: bool, +) -> dict[str, object] | None: # mutable-ok: Delta.provider_specific_fields is typed as a dict + items: Final = _STREAMED_OUTPUT_ITEMS.validate_python(output_payload) + calls: Final = tuple(item for item in items if item.get("type") in ("function_call", "custom_tool_call")) + streamed_message: Final = { + "role": "assistant", + "content": output_message_text(items, first_message_only=structured_output), + "tool_calls": [tool_call_dict_from_output_item(call, index) for index, call in enumerate(calls)], + "reasoning_items": list(_reasoning_items_from_output_items(items)), + "provider_specific_fields": tool_search_fields, + } + fields: Final = { + **(tool_search_fields or {}), + **replay_fields(items, streamed_message, completed=completed, first_message_only=structured_output), + } + return fields or None + + +def _assistant_phase_field(msg: object, replay_openai_history: bool, drop_params: bool) -> Mapping[str, object]: + match requested_phase(msg) if replay_openai_history else None: + case None: + return {} + case UnknownPhase(value=value) if drop_params: + verbose_logger.debug("Chat provider: dropping assistant phase=%r under drop_params", value) + return {} + case UnknownPhase(value=value): + raise litellm.UnsupportedParamsError( + message=f"phase must be 'commentary' or 'final_answer', got {value!r}. " + "To drop it, set `litellm.drop_params=True` or pass `drop_params=True`." + ) + case phase: + return {PHASE_FIELD: phase} + + def _responses_event_type(chunk: Mapping[str, object]) -> object: event_type: Final = chunk.get("type") return event_type.value if isinstance(event_type, ResponsesAPIStreamEvents) else event_type @@ -667,15 +731,7 @@ def _reasoning_item_to_response_input( An item without an id is sent without one: the Responses API accepts that and verifies the encrypted content on its own, while it rejects any id it did not mint. """ - item_id: Final = r_item.get("id") - encrypted_content: Final = r_item.get("encrypted_content") - return { - "type": "reasoning", - **({"id": item_id} if item_id else {}), - # summary is always required by the Responses API, even when empty - "summary": r_item.get("summary") or [], - **({"encrypted_content": encrypted_content} if encrypted_content else {}), - } + return dict(reasoning_input_item(r_item.get("id"), r_item.get("summary"), r_item.get("encrypted_content"))) def _requested_defer_loading(tool: Mapping[str, object], function_tool: Mapping[str, object]) -> object: @@ -804,12 +860,14 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): keep_prompt_cache_breakpoints: bool = False, replay_tool_search: bool = False, include_namespace: bool = False, + replay_openai_history: bool = False, ) -> tuple[list[object], str | None]: converted_input_items, instructions = self._convert_chat_completion_messages_to_responses_input( messages, drop_params=drop_params, replay_tool_search=replay_tool_search, include_namespace=include_namespace, + replay_openai_history=replay_openai_history, ) return ( converted_input_items @@ -825,10 +883,18 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): drop_params: bool = False, replay_tool_search: bool = False, include_namespace: bool = False, + replay_openai_history: bool = False, ) -> tuple[list[object], str | None]: input_items: Final[list[object]] = [] instructions: str | None = None - custom_tool_call_ids: Final = frozenset( + replayed_turns: Final = tuple( + replay_items(msg, replay_tool_search=replay_tool_search) if replay_openai_history else None + for msg in messages + ) + replayed_custom_call_ids: Final = frozenset( + chain.from_iterable(turn.custom_call_ids for turn in replayed_turns if turn is not None) + ) + custom_tool_call_ids: Final = replayed_custom_call_ids | frozenset( tool_call["id"] for msg in messages if msg.get("role") == "assistant" and isinstance(msg.get("tool_calls"), list) @@ -849,7 +915,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): tool_calls = msg.get("tool_calls") tool_call_id = msg.get("tool_call_id") - if role == "system": + if (replayed_turn := replayed_turns[index]) is not None: + input_items.extend(replayed_turn.items) + elif role == "system": # Extract system message as instructions if isinstance(content, str) and index < leading_system_count: if instructions: @@ -915,6 +983,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): "content": self._convert_content_to_responses_format( content, "assistant", drop_params=drop_params ), + **_assistant_phase_field(msg, replay_openai_history, drop_params), } ) for tool_call in tool_calls: @@ -955,6 +1024,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): "content": self._convert_content_to_responses_format( content, cast(str, role), drop_params=drop_params ), + **_assistant_phase_field(msg, replay_openai_history and role == "assistant", drop_params), } ) elif role == "assistant": @@ -1081,6 +1151,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): keep_prompt_cache_breakpoints=supports_prompt_cache_breakpoint, replay_tool_search=_requests_tool_search(cast(object, optional_params)), # cast-ok: upcast include_namespace=True, + replay_openai_history=provider == "openai", ) # OpenAI's Responses API rejects an empty input. For a system-only # request, carry the system message as a system-role input item instead @@ -1165,6 +1236,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): output_items: Sequence[object], handle_raw_dict_callback: Callable[..., tuple["Choices | None", int]] | None = None, structured_output: bool = False, + replay_openai_history: bool = False, + completed: bool = False, ) -> list["Choices"]: """ Convert Responses API output items to chat completion choices. @@ -1174,6 +1247,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): handle_raw_dict_callback: Optional callback for handling raw dict items structured_output: The chat request carried a json response_format, so only the first message item becomes content + replay_openai_history: Record the output's item sequence for the next turn + completed: The response finished, so its item sequence is whole Returns: List of Choices objects @@ -1196,7 +1271,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): choices: Final[list[Choices]] = [] index = 0 reasoning_content: str | None = None - pending_reasoning_item: _BuiltReasoningItem | None = None + pending_reasoning_items: tuple[_BuiltReasoningItem, ...] = () # Collect all tool calls to put them in a single choice # (Chat Completions API expects all tool calls in one message) @@ -1205,12 +1280,19 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): for item in output_items: if isinstance(item, ResponseReasoningItem): - pending_reasoning_item = _build_reasoning_item( - item_id=item.id, - encrypted_content=getattr(item, "encrypted_content", None), - summary_raw=item.summary, + pending_reasoning_items = ( + *(pending_reasoning_items if replay_openai_history else ()), + _build_reasoning_item( + item_id=item.id, + encrypted_content=getattr(item, "encrypted_content", None), + summary_raw=item.summary, + ), + ) + reasoning_content = " ".join( + s["text"] + for s in chain.from_iterable(built["summary"] for built in pending_reasoning_items) + if s.get("text") ) - reasoning_content = " ".join(s["text"] for s in pending_reasoning_item["summary"] if s.get("text")) elif isinstance(item, ResponseOutputMessage): for content in item.content: @@ -1227,7 +1309,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): annotations=annotations, reasoning_items=cast( list[ChatCompletionReasoningItem] | None, - ([pending_reasoning_item] if pending_reasoning_item is not None else None), + (list(pending_reasoning_items) or None), ), ) @@ -1240,7 +1322,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): ) reasoning_content = None # flush - pending_reasoning_item = None # flush + pending_reasoning_items = () # flush index += 1 elif isinstance(item, ResponseFunctionToolCall): @@ -1300,16 +1382,17 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): output_items, "stop", tool_search_items=tool_search_items ) ] - return [ - _fold_choices_into_one( - choices=choices, - tool_calls=accumulated_tool_calls, - trailing_reasoning_content=reasoning_content, - trailing_reasoning_item=pending_reasoning_item, - tool_search_items=tool_search_items, - structured_output=structured_output, - ) - ] + folded: Final = _fold_choices_into_one( + choices=choices, + tool_calls=accumulated_tool_calls, + trailing_reasoning_content=reasoning_content, + trailing_reasoning_items=pending_reasoning_items, + tool_search_items=tool_search_items, + structured_output=structured_output, + ) + if not replay_openai_history: + return [folded] + return [_with_replay_fields(folded, output_items, completed=completed, structured_output=structured_output)] @staticmethod def _build_empty_choice( @@ -1443,6 +1526,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): structured_output=is_structured_output_request( cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict ), + replay_openai_history=_served_provider( + cast("Mapping[str, object]", litellm_params) # cast-ok: litellm_params is an untyped dict + ) + == "openai", + completed=raw_response.status == "completed", ) response_is_incomplete: Final = raw_response.status == "incomplete" or ( @@ -1506,9 +1594,14 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): sync_stream: bool, json_mode: bool | None = False, structured_output: bool = False, + replays_openai_history: bool = False, ) -> BaseModelResponseIterator: return OpenAiResponsesToChatCompletionStreamIterator( - streaming_response, sync_stream, json_mode, structured_output=structured_output + streaming_response, + sync_stream, + json_mode, + structured_output=structured_output, + replays_openai_history=replays_openai_history, ) def _convert_content_str_to_input_text(self, content: str, role: str) -> dict[str, object]: @@ -1890,8 +1983,10 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): sync_stream: bool, json_mode: bool | None = False, structured_output: bool = False, + replays_openai_history: bool = False, ): super().__init__(streaming_response, sync_stream, json_mode) + self._replays_openai_history: Final = replays_openai_history self._chat_completion_id: str | None = None self._served_service_tier: str | None = None self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state @@ -1937,6 +2032,8 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): def translate_responses_chunk_to_openai_stream( parsed_chunk: dict | BaseModel, tool_call_index_map: dict[int, int] | None = None, # mutable-ok: per-stream state, remapped in place + replays_openai_history: bool = False, + structured_output: bool = False, ) -> "ModelResponseStream": """ Translate a Responses API streaming chunk to OpenAI chat completion streaming format. @@ -1944,6 +2041,8 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): Args: parsed_chunk: Dict containing the Responses API event chunk tool_call_index_map: Per-stream output_index -> sequential tool_call index map + replays_openai_history: Record the completed output's item sequence for the next turn + structured_output: Only the first message item became content Returns: ModelResponseStream: OpenAI-formatted streaming chunk @@ -2154,6 +2253,10 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): terminal_reasoning_items_typed: Final = _as_chat_reasoning_items( _reasoning_items_from_output_items(output_items) ) + output_payload: Final = cast(object, output_items) # cast-ok: untyped + tool_search_fields: Final = _tool_search_provider_fields( + _tool_search_items_from_output_items(output_payload) + ) usage = None if response_data.get("usage"): @@ -2167,10 +2270,17 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): StreamingChoices( index=0, delta=Delta( - content="", + content=None if replays_openai_history else "", reasoning_items=terminal_reasoning_items_typed, - provider_specific_fields=_tool_search_provider_fields( - _tool_search_items_from_output_items(cast(object, output_items)) # cast-ok: untyped + provider_specific_fields=( + completed_stream_history( + output_payload, + tool_search_fields, + completed=event_type == "response.completed", + structured_output=structured_output, + ) + if replays_openai_history + else tool_search_fields ), ), finish_reason=finish_reason, @@ -2216,7 +2326,10 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): self._empty_content_chunk() if self._is_text_delta_after_structured_answer(chunk) else OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream( - chunk, tool_call_index_map=self._tool_call_index_map + chunk, + tool_call_index_map=self._tool_call_index_map, + replays_openai_history=self._replays_openai_history, + structured_output=self._structured_output, ) ) if self._structured_output and _is_message_item_done(chunk): diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index f2f06d801a8..5542249d2c7 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1938,6 +1938,16 @@ class CustomStreamWrapper: model_response: Final = self.model_response_creator( hidden_params=None if self.received_model_id is None else {"received_model_id": self.received_model_id} ) + served_tier: Final = next( + ( + tier + for chunk in reversed(self.chunks) + if isinstance(tier := getattr(chunk, "service_tier", None), str) and tier + ), + None, + ) + if served_tier is not None: + setattr(model_response, "service_tier", served_tier) # noqa: B010 # pydantic extra, not a declared field _finish_reason: Final = self.received_finish_reason or self.intermittent_finish_reason if _finish_reason is not None: model_response.choices[0].finish_reason = _finish_reason diff --git a/litellm/tool_loop.py b/litellm/tool_loop.py index aaa0d62eea7..d0b2850bb5b 100644 --- a/litellm/tool_loop.py +++ b/litellm/tool_loop.py @@ -63,6 +63,8 @@ def _assistant_tool_call(tool_call: ChatCompletionMessageToolCall) -> ChatComple def _assistant_message( message: Message, tool_calls: tuple[ChatCompletionMessageToolCall, ...], replays_provider_specific_fields: bool ) -> ChatCompletionAssistantMessage: + from litellm.completion_extras.litellm_responses_transformation.history import retained_replay_fields + return cast( # cast-ok: dict literal with the optional message fields spread in only when set "ChatCompletionAssistantMessage", { @@ -78,7 +80,7 @@ def _assistant_message( "provider_specific_fields", getattr(message, "provider_specific_fields", None) if replays_provider_specific_fields - else None, + else retained_replay_fields(getattr(message, "provider_specific_fields", None)), ), ) if value is not None diff --git a/tests/integration/_support/responses_replay.py b/tests/integration/_support/responses_replay.py new file mode 100644 index 00000000000..c4e73ffbf2b --- /dev/null +++ b/tests/integration/_support/responses_replay.py @@ -0,0 +1,214 @@ +import json +import uuid +from collections.abc import Callable, Sequence +from itertools import chain +from typing import Final + +from integration._support.client import list_value, object_value +from integration._support.wire import Reply, Request, Wire +from pydantic import JsonValue + +MODEL: Final = "gpt-6.1-sol" +PLAN: Final = "I am starting by choosing which city to check." +PICK: Final = "My pick is Tokyo, so I will check its weather." +PATCH_NOTE: Final = "I will create hello.txt containing hi." +ANSWER: Final = "It is raining in Tokyo." + +WEATHER_TOOL: Final[dict[str, JsonValue]] = { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get the weather for a city.", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + }, +} +PATCH_TOOL: Final[dict[str, JsonValue]] = { + "type": "custom", + "custom": {"name": "apply_patch", "description": "Apply a V4A patch to the workspace"}, +} + + +def _message(item_id: str, text: str, phase: str) -> dict[str, JsonValue]: + return { + "type": "message", + "id": item_id, + "status": "completed", + "role": "assistant", + "phase": phase, + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + + +def _reasoning(item_id: str, summary: str) -> dict[str, JsonValue]: + return { + "type": "reasoning", + "id": item_id, + "summary": [{"type": "summary_text", "text": summary}], + "encrypted_content": f"enc_{item_id}", + } + + +_WEATHER_CALL: Final[dict[str, JsonValue]] = { + "type": "function_call", + "id": "fc_weather", + "call_id": "call_weather", + "name": "get_weather", + "arguments": '{"city":"Tokyo"}', + "status": "completed", +} +WEATHER_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _message("msg_plan", PLAN, "commentary"), + _reasoning("rs_pick", "Tokyo is often rainy this time of year."), + _message("msg_pick", PICK, "commentary"), + _WEATHER_CALL, +) +TWO_THOUGHTS_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _reasoning("rs_scan", "Paris and Tokyo are the two candidates."), + _reasoning("rs_pick", "Tokyo is often rainy this time of year."), + _message("msg_pick", PICK, "commentary"), + _WEATHER_CALL, +) +CALL_ONLY_TURN: Final[tuple[dict[str, JsonValue], ...]] = (_WEATHER_CALL,) +PATCH_TURN: Final[tuple[dict[str, JsonValue], ...]] = ( + _message("msg_patch", PATCH_NOTE, "commentary"), + { + "type": "custom_tool_call", + "id": "ctc_patch", + "call_id": "call_patch", + "name": "apply_patch", + "input": "*** Begin Patch\n*** Add File: hello.txt\n+hi\n*** End Patch", + "status": "completed", + }, +) +WEATHER_REPLAY: Final = ( + ("message", "assistant", "commentary", PLAN), + ("reasoning", "rs_pick"), + ("message", "assistant", "commentary", PICK), + ("function_call", "call_weather", "get_weather"), + ("function_call_output", "call_weather"), +) +TWO_THOUGHTS_REPLAY: Final = ( + ("reasoning", "rs_scan"), + ("reasoning", "rs_pick"), + ("message", "assistant", "commentary", PICK), + ("function_call", "call_weather", "get_weather"), + ("function_call_output", "call_weather"), +) + + +def _response(identity: str, output: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + return { + "id": f"resp_{identity}", + "object": "response", + "created_at": 1789788253, + "status": "completed", + "model": MODEL, + "output": list(output), + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + } + + +def _event(payload: dict[str, JsonValue]) -> bytes: + return f"event: {payload['type']}\ndata: {json.dumps(payload)}\n\n".encode() + + +def _opened(item: dict[str, JsonValue]) -> dict[str, JsonValue]: + match item["type"]: + case "message": + return {**item, "status": "in_progress", "content": []} + case "function_call": + return {**item, "status": "in_progress", "arguments": ""} + case _: + return item + + +def _deltas(index: int, item: dict[str, JsonValue]) -> tuple[dict[str, JsonValue], ...]: + match item["type"]: + case "message": + text: Final = object_value(list_value(item["content"])[0])["text"] + return ( + { + "type": "response.output_text.delta", + "item_id": item["id"], + "output_index": index, + "content_index": 0, + "delta": text, + }, + ) + case "function_call": + return ( + { + "type": "response.function_call_arguments.delta", + "item_id": item["id"], + "output_index": index, + "delta": item["arguments"], + }, + ) + case _: + return () + + +def _item_events(index: int, item: dict[str, JsonValue]) -> tuple[bytes, ...]: + return ( + _event({"type": "response.output_item.added", "output_index": index, "item": _opened(item)}), + *(_event(delta) for delta in _deltas(index, item)), + _event({"type": "response.output_item.done", "output_index": index, "item": item}), + ) + + +def _stream(identity: str, output: Sequence[dict[str, JsonValue]]) -> tuple[bytes, ...]: + response: Final = _response(identity, output) + return ( + _event({"type": "response.created", "response": {**response, "status": "in_progress", "output": []}}), + *chain.from_iterable(_item_events(index, item) for index, item in enumerate(output)), + _event({"type": "response.completed", "response": response}), + ) + + +def _answered(body: dict[str, JsonValue]) -> bool: + return any( + object_value(item).get("type") in {"function_call_output", "custom_tool_call_output"} + for item in list_value(body["input"]) + ) + + +def responses_upstream(turn_one: Sequence[dict[str, JsonValue]]) -> Callable[[Request], Reply]: + def respond(request: Request) -> Reply: + if request.method == "GET": + return Reply(body=b'{"object":"list","data":[]}') + assert request.target == "/responses", request.target + body: Final = object_value(json.loads(request.body)) + identity: Final = uuid.uuid4().hex + output: Final = (_message("msg_answer", ANSWER, "final_answer"),) if _answered(body) else turn_one + if body.get("stream"): + return Reply(content_type="text/event-stream", chunks=_stream(identity, output)) + return Reply(body=json.dumps(_response(identity, output)).encode()) + + return respond + + +def _text(content: JsonValue) -> JsonValue: + if isinstance(content, str): + return content + return "".join(str(object_value(part).get("text")) for part in list_value(content)) + + +def _shape(item: JsonValue) -> tuple[JsonValue, ...]: + fields: Final = object_value(item) + match fields.get("type"): + case "message": + return ("message", fields.get("role"), fields.get("phase"), _text(fields.get("content"))) + case "reasoning": + return ("reasoning", fields.get("id")) + case "function_call" | "custom_tool_call" as kind: + return (kind, fields.get("call_id"), fields.get("name")) + case kind: + return (kind, fields.get("call_id")) + + +def posted_inputs(wire: Wire) -> tuple[tuple[tuple[JsonValue, ...], ...], ...]: + return tuple( + tuple(_shape(item) for item in list_value(object_value(json.loads(request.body))["input"])) + for request in wire.drain() + if request.method == "POST" + ) diff --git a/tests/integration/providers/test_responses_bridge_history_replay_wire.py b/tests/integration/providers/test_responses_bridge_history_replay_wire.py new file mode 100644 index 00000000000..f85dca67be4 --- /dev/null +++ b/tests/integration/providers/test_responses_bridge_history_replay_wire.py @@ -0,0 +1,233 @@ +import json +import uuid +from collections.abc import Sequence +from functools import reduce +from itertools import chain +from typing import Final + +import httpx +import pytest +from integration._support.client import Gateway, list_value, object_value +from integration._support.responses_replay import ( + CALL_ONLY_TURN, + MODEL, + PATCH_NOTE, + PATCH_TOOL, + PATCH_TURN, + PICK, + PLAN, + TWO_THOUGHTS_REPLAY, + TWO_THOUGHTS_TURN, + WEATHER_REPLAY, + WEATHER_TOOL, + WEATHER_TURN, + posted_inputs, + responses_upstream, +) +from integration._support.wire import wire_server +from pydantic import JsonValue + +_API_KEY: Final = "synthetic-openai-key" + + +def _chunks(response: httpx.Response) -> list[dict[str, JsonValue]]: + return [ + object_value(json.loads(line[6:])) + for line in response.iter_lines() + if line.startswith("data: ") and line != "data: [DONE]" + ] + + +def _tool_call(index: JsonValue, calls: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + parts: Final = [call for call in calls if call["index"] == index] + functions: Final = [object_value(part["function"]) for part in parts] + return { + "id": next(part["id"] for part in parts if part.get("id")), + "type": "function", + "function": { + "name": next(function["name"] for function in functions if function.get("name")), + "arguments": "".join(str(function.get("arguments") or "") for function in functions), + }, + } + + +def _deltas(chunks: Sequence[dict[str, JsonValue]]) -> list[dict[str, JsonValue]]: + return [ + object_value(object_value(choice)["delta"]) + for choice in chain.from_iterable(list_value(chunk["choices"]) for chunk in chunks) + ] + + +def _assembled(chunks: Sequence[dict[str, JsonValue]]) -> dict[str, JsonValue]: + deltas: Final = _deltas(chunks) + calls: Final = [object_value(call) for call in chain.from_iterable(d.get("tool_calls") or [] for d in deltas)] + fields: Final = reduce( + lambda merged, delta: {**merged, **object_value(delta.get("provider_specific_fields") or {})}, deltas, {} + ) + return { + "role": "assistant", + "content": "".join(str(delta["content"]) for delta in deltas if delta.get("content")) or None, + "tool_calls": [_tool_call(index, calls) for index in sorted({int(str(call["index"])) for call in calls})], + "reasoning_items": list(chain.from_iterable(list_value(d.get("reasoning_items") or []) for d in deltas)), + "provider_specific_fields": fields, + } + + +def _first_turn( + gateway: Gateway, model: str, prompt: str, tools: Sequence[dict[str, JsonValue]], *, stream: bool +) -> dict[str, JsonValue]: + response: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": list(tools), + "stream": stream, + "cache": {"no-cache": True}, + }, + ) + assert response.status_code == 200, response.text + if stream: + return _assembled(_chunks(response)) + return object_value(object_value(list_value(response.json()["choices"])[0])["message"]) + + +def _next_turn( + gateway: Gateway, + model: str, + prompt: str, + tools: Sequence[dict[str, JsonValue]], + message: dict[str, JsonValue], + **extra: JsonValue, +) -> httpx.Response: + call_id: Final = object_value(list_value(message["tool_calls"])[0])["id"] + return gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [ + {"role": "user", "content": prompt}, + message, + {"role": "tool", "tool_call_id": call_id, "content": "Rain, 18C"}, + ], + "tools": list(tools), + "cache": {"no-cache": True}, + **extra, + }, + ) + + +def _with_phase(message: dict[str, JsonValue], phase: str) -> dict[str, JsonValue]: + fields: Final = object_value(message.get("provider_specific_fields") or {}) + return {**message, "provider_specific_fields": {**fields, "phase": phase}} + + +def _assistant_messages(items: Sequence[tuple[JsonValue, ...]]) -> list[tuple[JsonValue, ...]]: + return [item for item in items if item[:2] == ("message", "assistant")] + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_next_turn_replays_responses_items_in_the_order_openai_produced(gateway: Gateway, stream: bool) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=stream) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert message["content"] == PLAN + PICK, message + assert object_value(message["provider_specific_fields"])["phase"] == "commentary", message + assert inputs[1] == (("message", "user", None, prompt), *WEATHER_REPLAY), inputs + + +def test_chat_next_turn_sends_the_phase_the_app_set_on_the_assistant_message(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], _with_phase(message, "final_answer")) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert [item[2] for item in _assistant_messages(inputs[1])] == ["final_answer"], inputs + + +def test_chat_next_turn_rejects_an_undefined_phase_unless_drop_params_is_set(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _with_phase(_first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False), "analysis") + rejected: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + dropped: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message, drop_params=True) + inputs: Final = posted_inputs(wire) + assert rejected.status_code == 400, rejected.text + assert "'analysis'" in rejected.text, rejected.text + assert dropped.status_code == 200, dropped.text + assert len(inputs) == 2, inputs + assert [item[2] for item in _assistant_messages(inputs[1])] == [None], inputs + + +def test_chat_next_turn_sends_the_text_the_app_edited_instead_of_the_recorded_items(gateway: Gateway) -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + edited_text: Final = "I picked Tokyo." + with wire_server(responses_upstream(WEATHER_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], {**message, "content": edited_text}) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert _assistant_messages(inputs[1]) == [("message", "assistant", "commentary", edited_text)], inputs + + +def test_chat_next_turn_replays_a_custom_tool_call_as_a_custom_tool_call(gateway: Gateway) -> None: + prompt: Final = f"Create hello.txt containing hi {uuid.uuid4().hex}" + with wire_server(responses_upstream(PATCH_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [PATCH_TOOL], stream=False) + response: Final = _next_turn(gateway, model, prompt, [PATCH_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert inputs[1] == ( + ("message", "user", None, prompt), + ("message", "assistant", "commentary", PATCH_NOTE), + ("custom_tool_call", "call_patch", "apply_patch"), + ("custom_tool_call_output", "call_patch"), + ), inputs + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_reply_keeps_every_reasoning_item_and_the_next_turn_replays_each_one( + gateway: Gateway, stream: bool +) -> None: + prompt: Final = f"Weigh two cities, then check whether it is raining in your pick {uuid.uuid4().hex}" + with wire_server(responses_upstream(TWO_THOUGHTS_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=stream) + response: Final = _next_turn(gateway, model, prompt, [WEATHER_TOOL], message) + inputs: Final = posted_inputs(wire) + assert response.status_code == 200, response.text + assert [object_value(item)["id"] for item in list_value(message["reasoning_items"])] == ["rs_scan", "rs_pick"] + assert inputs[1] == (("message", "user", None, prompt), *TWO_THOUGHTS_REPLAY), inputs + + +def test_chat_stream_of_a_tool_call_only_turn_carries_null_content_like_the_json_reply(gateway: Gateway) -> None: + prompt: Final = f"Check whether it is raining in Tokyo {uuid.uuid4().hex}" + with wire_server(responses_upstream(CALL_ONLY_TURN)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"openai/responses/{MODEL}", api_base=wire.url, api_key=_API_KEY) + message: Final = _first_turn(gateway, model, prompt, [WEATHER_TOOL], stream=False) + streamed: Final = gateway.request( + "POST", + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "tools": [WEATHER_TOOL], + "stream": True, + "cache": {"no-cache": True}, + }, + ) + assert streamed.status_code == 200, streamed.text + deltas: Final = _deltas(_chunks(streamed)) + assert message["content"] is None, message + assert [delta["content"] for delta in deltas if delta.get("content") is not None] == [], deltas diff --git a/tests/integration/providers/test_responses_bridge_incomplete.py b/tests/integration/providers/test_responses_bridge_incomplete.py index e3978b4debd..396930e9971 100644 --- a/tests/integration/providers/test_responses_bridge_incomplete.py +++ b/tests/integration/providers/test_responses_bridge_incomplete.py @@ -1,6 +1,7 @@ import json import uuid from typing import Final +from unittest.mock import ANY import httpx import pytest @@ -288,6 +289,7 @@ def test_chat_over_responses_deployment_merges_message_and_function_call(gateway "index": 0, } ], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -397,6 +399,7 @@ def test_chat_over_responses_deployment_keeps_reasoning_with_merged_tool_call(ga "index": 0, } ], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -480,6 +483,7 @@ def test_chat_over_responses_deployment_returns_tool_call_only_reply_as_one_choi "index": 0, } ], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -570,6 +574,7 @@ def test_chat_over_responses_deployment_merges_function_call_followed_by_message "index": 0, } ], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -671,6 +676,7 @@ def test_chat_over_responses_deployment_folds_two_messages_and_function_call_int "index": 0, } ], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -729,6 +735,7 @@ def test_chat_over_responses_deployment_folds_commentary_and_final_answer_into_o "message": { "role": "assistant", "content": 'Let me answer that.{"city": "Paris", "summary": "Sunny, 21C"}', + "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY}, }, } ], response.text @@ -843,6 +850,7 @@ def test_chat_over_responses_deployment_folds_reasoning_two_messages_and_functio "reasoning_content": "Checking the forecast.", "reasoning_items": [_REASONING_ITEM], "tool_calls": _TWO_CALLS_CHAT_SHAPE[:1], + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -884,6 +892,7 @@ def test_chat_over_responses_deployment_folds_message_and_two_function_calls_int "role": "assistant", "content": "Checking both.", "tool_calls": _TWO_CALLS_CHAT_SHAPE, + "provider_specific_fields": {"openai_responses_replay": ANY}, }, } ], response.text @@ -1353,7 +1362,11 @@ def test_chat_over_responses_deployment_with_json_schema_returns_first_of_two_fi { "finish_reason": "stop", "index": 0, - "message": {"role": "assistant", "content": '{"city":"Paris"}'}, + "message": { + "role": "assistant", + "content": '{"city":"Paris"}', + "provider_specific_fields": {"phase": "final_answer"}, + }, } ], response.text assert [(request.method, request.target) for request in wire.drain()] == [("POST", "/responses")] diff --git a/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py b/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py new file mode 100644 index 00000000000..0c58545182b --- /dev/null +++ b/tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py @@ -0,0 +1,37 @@ +import uuid +from typing import Final, cast + +from integration._support.responses_replay import ( + ANSWER, + MODEL, + WEATHER_REPLAY, + WEATHER_TOOL, + WEATHER_TURN, + posted_inputs, + responses_upstream, +) +from integration._support.wire import wire_server + +import litellm +from litellm.types.llms.openai import ChatCompletionToolMessage, ChatCompletionToolParam +from litellm.types.utils import ChatCompletionMessageToolCall + + +def _rain(call: ChatCompletionMessageToolCall) -> ChatCompletionToolMessage: + return {"role": "tool", "tool_call_id": call.id, "content": "Rain, 18C"} + + +def test_run_tool_loop_replays_responses_items_in_the_order_openai_produced() -> None: + prompt: Final = f"Pick a city and check whether it is raining there {uuid.uuid4().hex}" + with wire_server(responses_upstream(WEATHER_TURN)) as wire: + content: Final = litellm.run_tool_loop( + model=f"openai/responses/{MODEL}", + messages=[{"role": "user", "content": prompt}], + tools=[cast(ChatCompletionToolParam, WEATHER_TOOL)], + execute_tool=_rain, + api_base=wire.url, + api_key="synthetic-openai-key", + ) + inputs: Final = posted_inputs(wire) + assert content == ANSWER + assert inputs[1] == (("message", "user", None, prompt), *WEATHER_REPLAY), inputs diff --git a/tests/integration/translation/chat_completions/bases/openai_responses.py b/tests/integration/translation/chat_completions/bases/openai_responses.py index d086a56a221..f0fb0af6131 100644 --- a/tests/integration/translation/chat_completions/bases/openai_responses.py +++ b/tests/integration/translation/chat_completions/bases/openai_responses.py @@ -97,7 +97,17 @@ GPT_5_6_SOL_TEST_CASE: Final = TranslationTestCase( "created": ANY, "model": "openai/responses/gpt-5.6-sol", "object": "chat.completion", - "choices": [{"finish_reason": "stop", "index": 0, "message": {"content": "Hello!", "role": "assistant"}}], + "choices": [ + { + "finish_reason": "stop", + "index": 0, + "message": { + "content": "Hello!", + "role": "assistant", + "provider_specific_fields": {"phase": "final_answer", "openai_responses_replay": ANY}, + }, + } + ], "usage": { "completion_tokens": 6, "prompt_tokens": 19, @@ -252,6 +262,7 @@ GPT_6_1_SOL_TEST_CASE: Final = TranslationTestCase( "summary": [], } ], + "provider_specific_fields": {"phase": "commentary", "openai_responses_replay": ANY}, }, } ], diff --git a/tests/integration/translation/response_scenarios.py b/tests/integration/translation/response_scenarios.py index b49d2b71441..7168ad0ade5 100644 --- a/tests/integration/translation/response_scenarios.py +++ b/tests/integration/translation/response_scenarios.py @@ -57,7 +57,10 @@ def two_messages_before_tools(case: TranslationTestCase) -> TranslationTestCase: def _messages_then_tools_response(case: TranslationTestCase) -> Mapping[str, JsonValue]: expected: Final = case.expected_litellm_response if case.litellm_endpoint == "/v1/chat/completions": - choice: Final = items(expected["choices"])[0] + choices: Final = expected["choices"] + assert isinstance(choices, list) and isinstance(choices[0], dict), choices + message: Final = choices[0]["message"] + assert isinstance(message, dict), message return { **expected, "choices": [ @@ -67,7 +70,12 @@ def _messages_then_tools_response(case: TranslationTestCase) -> Mapping[str, Jso "message": { "role": "assistant", "content": "".join(COMMENTARY), - "tool_calls": object_value(choice["message"])["tool_calls"], + "tool_calls": message["tool_calls"], + **( + {"provider_specific_fields": message["provider_specific_fields"]} + if "provider_specific_fields" in message + else {} + ), }, } ], diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py new file mode 100644 index 00000000000..a2efbfef5dc --- /dev/null +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py @@ -0,0 +1,627 @@ +import json +from collections.abc import Callable, Iterator, Mapping, Sequence +from typing import Final + +import httpx +import pytest +import respx + +import litellm +from litellm.completion_extras.litellm_responses_transformation import history +from litellm.exceptions import MidStreamFallbackError + +OPENAI_RESPONSES_URL: Final = "https://api.openai.com/v1/responses" +AZURE_API_BASE: Final = "https://litellm-history-test.openai.azure.com" +MODEL: Final = "gpt-5.6" +TOOLS: Final = ( + { + "type": "function", + "function": { + "name": "get_weather", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, + }, + }, +) +USER: Final = {"role": "user", "content": "Weather in Paris? Narrate as you go."} + +COMMENTARY: Final = { + "type": "message", + "id": "msg_commentary", + "status": "completed", + "role": "assistant", + "phase": "commentary", + "content": [{"type": "output_text", "text": "Checking the weather.", "annotations": []}], +} +FIRST_REASONING: Final = { + "type": "reasoning", + "id": "rs_first", + "summary": [{"type": "summary_text", "text": "Paris is likely wet."}], + "encrypted_content": "enc_first", +} +SECOND_REASONING: Final = {"type": "reasoning", "id": "rs_second", "summary": [], "encrypted_content": "enc_second"} +WEATHER_CALL: Final = { + "type": "function_call", + "id": "fc_weather", + "call_id": "call_weather_paris", + "name": "get_weather", + "arguments": '{"city":"Paris"}', + "status": "completed", +} +FINAL_ANSWER: Final = { + "type": "message", + "id": "msg_final", + "status": "completed", + "role": "assistant", + "phase": "final_answer", + "content": [ + {"type": "output_text", "text": "Paris: ", "annotations": []}, + {"type": "output_text", "text": "light rain.", "annotations": []}, + ], +} +OUTPUT: Final = (COMMENTARY, FIRST_REASONING, SECOND_REASONING, WEATHER_CALL, FINAL_ANSWER) +REPLAYED: Final = ( + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Checking the weather."}], + "phase": "commentary", + }, + { + "type": "reasoning", + "id": "rs_first", + "summary": [{"type": "summary_text", "text": "Paris is likely wet."}], + "encrypted_content": "enc_first", + }, + {"type": "reasoning", "id": "rs_second", "summary": [], "encrypted_content": "enc_second"}, + {"type": "function_call", "call_id": "call_weather_paris", "name": "get_weather", "arguments": '{"city":"Paris"}'}, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Paris: "}, {"type": "output_text", "text": "light rain."}], + "phase": "final_answer", + }, +) +TOOL_RESULT: Final = {"role": "tool", "tool_call_id": "call_weather_paris", "content": "light rain"} +FOLLOW_UP: Final = { + "type": "message", + "id": "msg_follow_up", + "status": "completed", + "role": "assistant", + "phase": "final_answer", + "content": [{"type": "output_text", "text": "Done.", "annotations": []}], +} + + +def _response(output: Sequence[Mapping[str, object]], status: str = "completed") -> dict[str, object]: + return { + "id": "resp_history", + "object": "response", + "created_at": 1789788253, + "status": status, + "model": MODEL, + "output": list(output), + "incomplete_details": {"reason": "max_output_tokens"} if status == "incomplete" else None, + "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, + } + + +def _added_item(item: Mapping[str, object]) -> Mapping[str, object]: + match item.get("type"): + case "message": + return {**item, "status": "in_progress", "content": []} + case "function_call": + return {**item, "status": "in_progress", "arguments": ""} + case "custom_tool_call": + return {**item, "status": "in_progress", "input": ""} + case _: + return item + + +def _delta_events(index: int, item: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: + match item.get("type"): + case "message": + parts: Final = item["content"] + assert isinstance(parts, list) + return tuple( + { + "type": "response.output_text.delta", + "item_id": item["id"], + "output_index": index, + "content_index": part_index, + "delta": part["text"], + } + for part_index, part in enumerate(parts) + ) + case "function_call": + return ( + { + "type": "response.function_call_arguments.delta", + "item_id": item["id"], + "output_index": index, + "delta": item["arguments"], + }, + ) + case "custom_tool_call": + return ( + { + "type": "response.custom_tool_call_input.delta", + "item_id": item["id"], + "output_index": index, + "delta": item["input"], + }, + ) + case _: + return () + + +def _item_events(index: int, item: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: + return ( + {"type": "response.output_item.added", "output_index": index, "item": _added_item(item)}, + *_delta_events(index, item), + {"type": "response.output_item.done", "output_index": index, "item": item}, + ) + + +def _sse(output: Sequence[Mapping[str, object]], terminal_event: str | None = "response.completed") -> httpx.Response: + item_events: Final = tuple(event for index, item in enumerate(output) for event in _item_events(index, item)) + terminal: Final = ( + () + if terminal_event is None + else ( + { + "type": terminal_event, + "response": _response(output, "completed" if terminal_event == "response.completed" else "incomplete"), + }, + ) + ) + events: Final = ( + {"type": "response.created", "response": {**_response((), "in_progress"), "usage": None}}, + *item_events, + *terminal, + ) + body: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) + + +def _assistant_turn( + output: Sequence[Mapping[str, object]], + *, + stream: bool, + model: str = f"openai/responses/{MODEL}", + terminal_event: str | None = "response.completed", + status: str = "completed", + response_format: Mapping[str, object] | None = None, +) -> dict[str, object]: + reply: Final = httpx.Response( + 200, content=json.dumps(_response(output, status)), headers={"content-type": "application/json"} + ) + with respx.mock(assert_all_called=True) as router: + router.post(url__startswith=OPENAI_RESPONSES_URL if model.startswith("openai/") else AZURE_API_BASE).mock( + return_value=_sse(output, terminal_event) if stream else reply + ) + response: Final = litellm.completion( + model=model, + messages=[USER], + tools=list(TOOLS), + api_key="sk-test", + api_base=None if model.startswith("openai/") else AZURE_API_BASE, + api_version=None if model.startswith("openai/") else "2025-04-01-preview", + stream=stream, + response_format=response_format, + ) + message: Final = ( + (litellm.stream_chunk_builder(list(response), messages=[USER]) if stream else response).choices[0].message + ) + return json.loads(json.dumps(message.model_dump())) + + +def _next_turn_input( + assistant: Mapping[str, object], + model: str = f"openai/responses/{MODEL}", + tools: Sequence[Mapping[str, object]] = TOOLS, + drop_params: bool = False, +) -> list[Mapping[str, object]]: + with respx.mock(assert_all_called=True) as router: + upstream: Final = router.post( + url__startswith=OPENAI_RESPONSES_URL if model.startswith("openai/") else AZURE_API_BASE + ).mock(return_value=httpx.Response(200, json=_response((FOLLOW_UP,)))) + litellm.completion( + model=model, + messages=[USER, dict(assistant), TOOL_RESULT], + tools=list(tools), + api_key="sk-test", + api_base=None if model.startswith("openai/") else AZURE_API_BASE, + api_version=None if model.startswith("openai/") else "2025-04-01-preview", + drop_params=drop_params, + ) + return json.loads(upstream.calls[0].request.content)["input"] + + +def _replayed_fields(message: Mapping[str, object]) -> Mapping[str, object]: + fields: Final = message.get("provider_specific_fields") + return fields if isinstance(fields, Mapping) else {} + + +def _types(items: Sequence[Mapping[str, object]]) -> list[object]: + return [item.get("type") for item in items] + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_next_turn_replays_every_item_phase_and_boundary_once_in_order(stream: bool) -> None: + assistant: Final = _assistant_turn(OUTPUT, stream=stream) + + sent: Final = _next_turn_input(assistant) + + assert assistant["content"] == "Checking the weather.Paris: light rain." + assert _replayed_fields(assistant)["phase"] == "final_answer" + assert sent[1:-1] == list(REPLAYED) + assert sent[-1]["type"] == "function_call_output" + assert sent[-1]["call_id"] == "call_weather_paris" + + +def _edit_content(message: Mapping[str, object]) -> Mapping[str, object]: + return {**message, "content": "Checking the weather. Paris: heavy rain."} + + +def _edit_phase(message: Mapping[str, object]) -> Mapping[str, object]: + return {**message, "provider_specific_fields": {**_replayed_fields(message), "phase": "commentary"}} + + +def _edit_arguments(message: Mapping[str, object]) -> Mapping[str, object]: + tool_calls: Final = message["tool_calls"] + assert isinstance(tool_calls, list) + edited_call: Final = {**tool_calls[0], "function": {**tool_calls[0]["function"], "arguments": '{"city":"Lyon"}'}} + return {**message, "tool_calls": [edited_call]} + + +@pytest.mark.parametrize( + ("edit", "expected_text", "expected_phase", "expected_arguments"), + [ + (_edit_content, "Checking the weather. Paris: heavy rain.", "final_answer", '{"city":"Paris"}'), + (_edit_phase, "Checking the weather.Paris: light rain.", "commentary", '{"city":"Paris"}'), + (_edit_arguments, "Checking the weather.Paris: light rain.", "final_answer", '{"city":"Lyon"}'), + ], + ids=["content", "phase", "arguments"], +) +def test_an_edited_chat_field_wins_over_the_recorded_items( + edit: Callable[[Mapping[str, object]], Mapping[str, object]], + expected_text: str, + expected_phase: str, + expected_arguments: str, +) -> None: + sent: Final = _next_turn_input(edit(_assistant_turn(OUTPUT, stream=False))) + + assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"] + assert sent[3]["content"] == [{"type": "output_text", "text": expected_text}] + assert sent[3]["phase"] == expected_phase + assert sent[4]["arguments"] == expected_arguments + + +def _without_replay(fields: Mapping[str, object]) -> Mapping[str, object]: + return {key: value for key, value in fields.items() if key != "openai_responses_replay"} + + +def _with_replay(fields: Mapping[str, object], replay: object) -> Mapping[str, object]: + return {**fields, "openai_responses_replay": replay} + + +def _replay_of(fields: Mapping[str, object]) -> Mapping[str, object]: + replay: Final = fields["openai_responses_replay"] + assert isinstance(replay, Mapping) + return replay + + +@pytest.mark.parametrize( + "corrupt", + [ + _without_replay, + lambda fields: _with_replay(fields, {**_replay_of(fields), "checksum": "0" * 64}), + lambda fields: _with_replay(fields, {**_replay_of(fields), "version": 2}), + lambda fields: _with_replay(fields, {**_replay_of(fields), "items": [{"type": "message", "parts": [-1]}]}), + lambda fields: _with_replay(fields, "not a replay"), + ], + ids=["missing", "checksum", "version", "items", "not-a-mapping"], +) +def test_missing_or_corrupt_replay_metadata_falls_back_to_the_chat_fields( + corrupt: Callable[[Mapping[str, object]], Mapping[str, object]], +) -> None: + assistant: Final = _assistant_turn(OUTPUT, stream=False) + + sent: Final = _next_turn_input({**assistant, "provider_specific_fields": corrupt(_replayed_fields(assistant))}) + + assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"] + assert sent[3]["content"] == [{"type": "output_text", "text": "Checking the weather.Paris: light rain."}] + assert sent[3]["phase"] == "final_answer" + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_an_unphased_answer_after_commentary_is_never_labeled_commentary(stream: bool) -> None: + unphased_answer: Final = {key: value for key, value in FINAL_ANSWER.items() if key != "phase"} + assistant: Final = _assistant_turn((COMMENTARY, WEATHER_CALL, unphased_answer), stream=stream) + + sent: Final = _next_turn_input(_edit_content(assistant)) + + assert "phase" not in _replayed_fields(assistant) + assert _types(sent[1:-1]) == ["message", "function_call"] + assert "phase" not in sent[1] + + +def _with_text(item: Mapping[str, object], text: str) -> Mapping[str, object]: + return {**item, "content": [{"type": "output_text", "text": text, "annotations": []}]} + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_a_structured_output_turn_carries_the_phase_of_the_message_it_returns(stream: bool) -> None: + output: Final = (_with_text(COMMENTARY, '{"city":"Paris"}'), _with_text(FINAL_ANSWER, '{"city":"Lyon"}')) + + assistant: Final = _assistant_turn(output, stream=stream, response_format={"type": "json_object"}) + + assert assistant["content"] == '{"city":"Paris"}' + assert _replayed_fields(assistant)["phase"] == "commentary" + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_a_lone_surrogate_in_the_reply_still_gets_a_record(stream: bool) -> None: + assistant: Final = _assistant_turn((_with_text(COMMENTARY, "Checking \ud83d"), WEATHER_CALL), stream=stream) + + assert "openai_responses_replay" in _replayed_fields(assistant) + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_a_last_message_the_bridge_cannot_read_leaves_the_turn_without_a_phase(stream: bool) -> None: + unreadable: Final = {**_with_text(FINAL_ANSWER, "It is raining in Paris."), "phase": "summary"} + assistant: Final = _assistant_turn((COMMENTARY, WEATHER_CALL, unreadable), stream=stream) + + assert "phase" not in _replayed_fields(assistant) + + +def test_a_phase_openai_does_not_define_is_dropped_under_drop_params() -> None: + assistant: Final = _assistant_turn(OUTPUT, stream=False) + + sent: Final = _next_turn_input({**assistant, "provider_specific_fields": {"phase": "analysis"}}, drop_params=True) + + assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"] + assert "phase" not in sent[3] + + +def test_a_phase_openai_does_not_define_is_rejected_without_drop_params() -> None: + assistant: Final = _assistant_turn(OUTPUT, stream=False) + + with respx.mock(assert_all_called=False) as router: + upstream: Final = router.post(url__startswith=OPENAI_RESPONSES_URL).mock( + return_value=httpx.Response(200, json=_response((FOLLOW_UP,))) + ) + with pytest.raises(litellm.BadRequestError, match="'analysis'"): + litellm.completion( + model=f"openai/responses/{MODEL}", + messages=[USER, {**assistant, "phase": "analysis"}, TOOL_RESULT], + tools=list(TOOLS), + api_key="sk-test", + ) + + assert not upstream.called + + +@pytest.mark.parametrize("fields", [None, ["not", "a", "mapping"]], ids=["no-fields", "malformed-fields"]) +def test_an_assistant_message_the_app_wrote_replays_without_a_phase(fields: object) -> None: + sent: Final = _next_turn_input( + { + "role": "assistant", + "content": "Checking Paris.", + "tool_calls": [ + { + "id": "call_weather_paris", + "type": "function", + "function": {"name": "get_weather", "arguments": '{"city": "Paris"}'}, + } + ], + "provider_specific_fields": fields, + } + ) + + assert _types(sent[1:-1]) == ["message", "function_call"] + assert "phase" not in sent[1] + + +SEARCH_CALL: Final = { + "type": "tool_search_call", + "id": "tsc_weather", + "arguments": {"paths": ["get_weather"]}, + "status": "completed", +} +SEARCH_OUTPUT: Final = { + "type": "tool_search_output", + "id": "tso_weather", + "status": "completed", + "tools": [{"name": "get_weather"}], +} + + +@pytest.mark.parametrize( + ("tools", "expected"), + [ + ((*TOOLS, {"type": "tool_search"}), ["tool_search_call", "tool_search_output", *_types(REPLAYED)]), + (TOOLS, ["reasoning", "reasoning", "message", "function_call"]), + ], + ids=["searching", "not-searching"], +) +def test_tool_search_items_replay_only_when_the_next_request_searches( + tools: Sequence[Mapping[str, object]], expected: Sequence[str] +) -> None: + assistant: Final = _assistant_turn((SEARCH_CALL, SEARCH_OUTPUT, *OUTPUT), stream=False) + + sent: Final = _next_turn_input(assistant, tools=tools) + + assert _types(sent[1:-1]) == list(expected) + + +WEB_SEARCH: Final = {"type": "web_search_call", "id": "ws_1", "status": "completed", "action": {"type": "search"}} + + +@pytest.mark.parametrize( + ("output", "stream", "terminal_event", "status"), + [ + (OUTPUT, False, "response.completed", "incomplete"), + (OUTPUT, True, "response.incomplete", "incomplete"), + ((WEB_SEARCH, *OUTPUT), False, "response.completed", "completed"), + ((WEB_SEARCH, *OUTPUT), True, "response.completed", "completed"), + ], + ids=["incomplete-json", "incomplete-stream", "unsupported-item-json", "unsupported-item-stream"], +) +def test_an_incomplete_or_unsupported_output_records_no_replay( + output: Sequence[Mapping[str, object]], stream: bool, terminal_event: str, status: str +) -> None: + assistant: Final = _assistant_turn(output, stream=stream, terminal_event=terminal_event, status=status) + + sent: Final = _next_turn_input(assistant) + + assert "openai_responses_replay" not in _replayed_fields(assistant) + assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"] + + +def _chunks_before_error(stream: Iterator[object]) -> Iterator[object]: + try: + yield from stream + except MidStreamFallbackError: + return + + +def test_an_interrupted_stream_records_no_replay() -> None: + with respx.mock(assert_all_called=True) as router: + router.post(OPENAI_RESPONSES_URL).mock(return_value=_sse(OUTPUT, terminal_event=None)) + stream: Final = litellm.completion( + model=f"openai/responses/{MODEL}", messages=[USER], tools=list(TOOLS), api_key="sk-test", stream=True + ) + received: Final = tuple(_chunks_before_error(iter(stream))) + assistant: Final = litellm.stream_chunk_builder(list(received), messages=[USER]).choices[0].message.model_dump() + + assert assistant["content"] == "Checking the weather.Paris: light rain." + assert _replayed_fields(assistant) == {} + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_a_non_openai_route_records_no_phase_or_replay(stream: bool) -> None: + assistant: Final = _assistant_turn(OUTPUT, stream=stream, model=f"azure/responses/{MODEL}") + + assert _replayed_fields(assistant) == {} + + +def test_a_non_openai_route_ignores_a_recorded_replay() -> None: + sent: Final = _next_turn_input(_assistant_turn(OUTPUT, stream=False), model=f"azure/responses/{MODEL}") + + assert _types(sent[1:-1]) == ["reasoning", "reasoning", "message", "function_call"] + assert all("phase" not in item for item in sent) + + +SECOND_THOUGHT: Final = {**SECOND_REASONING, "summary": [{"type": "summary_text", "text": "Lyon is likely dry."}]} +LYON_CALL: Final = {**WEATHER_CALL, "id": "fc_lyon", "call_id": "call_weather_lyon", "arguments": '{"city":"Lyon"}'} + + +@pytest.mark.parametrize( + ("model", "expected_ids", "expected_text"), + [ + (f"openai/responses/{MODEL}", ["rs_first", "rs_second"], "Paris is likely wet. Lyon is likely dry."), + (f"azure/responses/{MODEL}", ["rs_second"], "Lyon is likely dry."), + ], + ids=["openai", "azure"], +) +def test_only_the_openai_route_keeps_every_reasoning_item_of_a_tool_call_turn( + model: str, expected_ids: list[str], expected_text: str +) -> None: + assistant: Final = _assistant_turn( + (FIRST_REASONING, WEATHER_CALL, SECOND_THOUGHT, LYON_CALL), stream=False, model=model + ) + + reasoning_items: Final = assistant["reasoning_items"] + assert isinstance(reasoning_items, list) + assert [item["id"] for item in reasoning_items] == expected_ids + assert assistant["reasoning_content"] == expected_text + + +@pytest.mark.parametrize( + ("model", "expected_content"), + [(f"openai/responses/{MODEL}", None), (f"azure/responses/{MODEL}", "")], + ids=["openai", "azure"], +) +def test_only_the_openai_route_ends_a_streamed_tool_call_turn_with_null_content( + model: str, expected_content: str | None +) -> None: + assistant: Final = _assistant_turn((FIRST_REASONING, WEATHER_CALL), stream=True, model=model) + + assert assistant["content"] == expected_content + + +def _resealed(message: Mapping[str, object], items: Sequence[Mapping[str, object]]) -> Mapping[str, object]: + chat: Final = history._chat_message(message) + assert chat is not None + entries: Final = history._Replay.model_validate({"version": 1, "items": items, "checksum": ""}).items + replay: Final = {"version": 1, "items": list(items), "checksum": history._checksum(chat, entries)} + return {**message, "provider_specific_fields": _with_replay(_replayed_fields(message), replay)} + + +@pytest.mark.parametrize( + "items", + [ + ( + {"type": "message", "parts": [21]}, + {"type": "function_call", "index": 1}, + {"type": "message", "parts": [7, 11]}, + ), + ( + {"type": "message", "parts": [5]}, + {"type": "function_call", "index": 0}, + {"type": "message", "parts": [7, 11]}, + ), + ], + ids=["tool-call-index", "text-lengths"], +) +def test_a_resealed_record_that_does_not_fit_the_chat_fields_falls_back(items: Sequence[Mapping[str, object]]) -> None: + assistant: Final = _assistant_turn((COMMENTARY, WEATHER_CALL, FINAL_ANSWER), stream=False) + + sent: Final = _next_turn_input(_resealed(assistant, items)) + + assert _types(sent[1:-1]) == ["message", "function_call"] + assert sent[1]["content"] == [{"type": "output_text", "text": "Checking the weather.Paris: light rain."}] + + +PATCH_CALL: Final = { + "type": "custom_tool_call", + "id": "ctc_patch", + "call_id": "call_patch_readme", + "name": "apply_patch", + "input": "*** Begin Patch\n*** End Patch", + "status": "completed", +} + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_a_custom_tool_call_replays_with_its_custom_output(stream: bool) -> None: + assistant: Final = _assistant_turn((COMMENTARY, PATCH_CALL), stream=stream) + + with respx.mock(assert_all_called=True) as router: + upstream: Final = router.post(OPENAI_RESPONSES_URL).mock( + return_value=httpx.Response(200, json=_response((FOLLOW_UP,))) + ) + litellm.completion( + model=f"openai/responses/{MODEL}", + messages=[USER, assistant, {"role": "tool", "tool_call_id": "call_patch_readme", "content": "applied"}], + tools=[{"type": "custom", "custom": {"name": "apply_patch"}}], + api_key="sk-test", + ) + sent: Final = json.loads(upstream.calls[0].request.content)["input"] + + assert sent[1:] == [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Checking the weather."}], + "phase": "commentary", + }, + { + "type": "custom_tool_call", + "call_id": "call_patch_readme", + "name": "apply_patch", + "input": PATCH_CALL["input"], + }, + {"type": "custom_tool_call_output", "call_id": "call_patch_readme", "output": "applied"}, + ] diff --git a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py index 46f1dbf2e7a..d59f59703ad 100644 --- a/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py +++ b/tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py @@ -670,6 +670,41 @@ def test_streamed_tool_search_turn_replays_with_its_reasoning_item(): assert request["input"][5]["namespace"] == "get_weather" +@pytest.mark.parametrize( + "answer_item", + [ + ResponseFunctionToolCall( + type="function_call", id="fc_1", call_id="call_weather", name="get_weather", arguments="{}" + ), + ResponseOutputMessage( + type="message", + id="msg_1", + role="assistant", + status="completed", + content=[ResponseOutputText(type="output_text", text="Rain.", annotations=[])], + ), + ], + ids=["tool-call", "message"], +) +def test_every_reasoning_item_before_an_answer_reaches_an_openai_chat_message(answer_item: object) -> None: + reasoning: Final = tuple( + ResponseReasoningItem( + type="reasoning", + id=f"rs_{name}", + summary=[Summary(type="summary_text", text=f"{name.title()}.")], + encrypted_content=f"enc_{name}", + ) + for name in ("first", "second") + ) + + message: Final = LiteLLMResponsesTransformationHandler._convert_response_output_to_choices( + [*reasoning, answer_item], replay_openai_history=True + )[0].message + + assert [item["id"] for item in message.reasoning_items] == ["rs_first", "rs_second"] + assert message.reasoning_content == "First. Second." + + def test_transform_response_with_reasoning_and_output(): """Test transform_response handles ResponsesAPIResponse with reasoning items and output messages.""" from unittest.mock import Mock @@ -1270,9 +1305,7 @@ def test_transform_request_forwards_defer_loading_only_beside_tool_search( [{"type": "tool_search", "execution": "client"}, {"type": "tool_search"}], ids=["client-only", "hosted-and-client"], ) -def test_transform_request_drops_defer_loading_beside_client_tool_search( - deferred_tool, drop_params, tool_search_tool -): +def test_transform_request_drops_defer_loading_beside_client_tool_search(deferred_tool, drop_params, tool_search_tool): tool_search_tools: Final = ( [tool_search_tool] if tool_search_tool.get("execution") == "client" @@ -4396,7 +4429,11 @@ def test_tool_search_items_replay_only_to_a_request_that_carries_tool_search(too "provider_specific_fields": { "tool_search_items": [ {"type": "tool_search_call", "id": "tsc_1", "arguments": {"paths": ["get_weather"]}}, - {"type": "tool_search_output", "id": "tso_1", "tools": [{"type": "function", "name": "get_weather"}]}, + { + "type": "tool_search_output", + "id": "tso_1", + "tools": [{"type": "function", "name": "get_weather"}], + }, ] }, }, diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index b580f526a0f..f4502937fce 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -7820,6 +7820,38 @@ async def test_openai_stream_relays_the_served_service_tier_on_every_chunk_inclu assert relayed[-1]["usage"]["total_tokens"] == 11 +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_stream_relays_the_served_service_tier_on_the_finish_chunk_it_adds(logging_obj: Logging, sync_mode: bool): + from litellm.utils import ModelResponseListIterator + + def _chunk(content: str, finish_reason: str | None) -> ModelResponseStream: + return ModelResponseStream( + id="chatcmpl-tier", + created=1742056047, + model="gpt-4.1-mini", + choices=[StreamingChoices(finish_reason=finish_reason, index=0, delta=Delta(content=content))], + service_tier="priority", + ) + + logging_obj.update_environment_variables( + model="gpt-4.1-mini", optional_params={}, litellm_params={}, custom_llm_provider="openai" + ) + wrapper = CustomStreamWrapper( + completion_stream=ModelResponseListIterator(model_responses=[_chunk("Hi", None), _chunk("!", "stop")]), + model="gpt-4.1-mini", + custom_llm_provider="openai", + logging_obj=logging_obj, + ) + + relayed = ( + [chunk.model_dump() for chunk in wrapper] if sync_mode else [chunk.model_dump() async for chunk in wrapper] + ) + + assert [chunk["choices"][0]["finish_reason"] for chunk in relayed] == [None, None, "stop"], relayed + assert [chunk.get("service_tier") for chunk in relayed] == ["priority"] * len(relayed), relayed + + def _last_chunk_carries_finish_reason_wrapper( logging_obj: Logging, finish_reason: str, sync_stream: bool ) -> CustomStreamWrapper: diff --git a/tests/unit/test_tool_loop.py b/tests/unit/test_tool_loop.py index e3ebfe7258f..de8f62803cc 100644 --- a/tests/unit/test_tool_loop.py +++ b/tests/unit/test_tool_loop.py @@ -606,3 +606,69 @@ def test_client_tool_turn_is_replayed_without_its_provider_specific_fields(respx replayed_assistant: Final = _request_bodies(respx_mock)[1]["messages"][1] assert replayed_assistant["tool_calls"][0]["id"] == "call_1" assert "provider_specific_fields" not in replayed_assistant + + +def test_responses_bridge_replays_commentary_reasoning_and_tool_call_in_order(respx_mock: respx.MockRouter) -> None: + round_one: Final = _responses_payload( + "resp_1", + [ + { + "type": "message", + "id": "msg_commentary", + "status": "completed", + "role": "assistant", + "phase": "commentary", + "content": [{"type": "output_text", "text": "Checking Paris.", "annotations": []}], + }, + {"type": "reasoning", "id": "rs_abc123", "summary": [], "encrypted_content": "enc_xyz"}, + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_weather_paris", + "name": "get_weather", + "arguments": '{"city": "Paris"}', + "status": "completed", + }, + ], + ) + round_two: Final = _responses_payload( + "resp_2", + [ + { + "type": "message", + "id": "msg_1", + "status": "completed", + "role": "assistant", + "phase": "final_answer", + "content": [{"type": "output_text", "text": "Paris is 72F", "annotations": []}], + } + ], + ) + respx_mock.post("https://api.openai.com/v1/responses").mock( + side_effect=[httpx.Response(200, json=round_one), httpx.Response(200, json=round_two)] + ) + + answer: Final = litellm.run_tool_loop( + model="openai/responses/gpt-5.6", + messages=[{"role": "user", "content": "weather in Paris?"}], + tools=WEATHER_TOOLS, + execute_tool=_tool_result, + api_key="sk-test", + ) + + assert answer == "Paris is 72F" + assert _request_bodies(respx_mock)[1]["input"][1:] == [ + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Checking Paris."}], + "phase": "commentary", + }, + {"type": "reasoning", "id": "rs_abc123", "summary": [], "encrypted_content": "enc_xyz"}, + {"type": "function_call", "call_id": "call_weather_paris", "name": "get_weather", "arguments": '{"city": "Paris"}'}, + { + "type": "function_call_output", + "call_id": "call_weather_paris", + "output": [{"type": "input_text", "text": '{"temp": "72F"}'}], + }, + ]