scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 250

fix(responses): replay the original Responses item order, phase, and boundaries on the next chat turn (#48912)

Browse source 
* 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>
This commit is contained in:
simonp9 2026-10-29 00:56:02 -07:00 • committed by GitHub
parent 3f01341f0a
commit dff474e450
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 1882 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/completion_extras/litellm_responses_transformation/handler.py
View file

@ -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,

432
litellm/completion_extras/litellm_responses_transformation/history.py Normal file
View file

@ -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)

185
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -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):

10
litellm/litellm_core_utils/streaming_handler.py
View file

@ -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

4
litellm/tool_loop.py
View file

@ -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

214
tests/integration/_support/responses_replay.py Normal file
View file

@ -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"
)

233
tests/integration/providers/test_responses_bridge_history_replay_wire.py Normal file
View file

@ -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

15
tests/integration/providers/test_responses_bridge_incomplete.py
View file

@ -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")]

37
tests/integration/sdk/test_tool_loop_responses_history_replay_sdk.py Normal file
View file

@ -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

13
tests/integration/translation/chat_completions/bases/openai_responses.py
View file

@ -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},
},
}
],

12
tests/integration/translation/response_scenarios.py
View file

@ -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 {}
),
},
}
],

627
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_history.py Normal file
View file

@ -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"},
]

45
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
View file

@ -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"}],
},
]
},
},

32
tests/unit/litellm_core_utils/test_streaming_handler.py
View file

@ -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:

66
tests/unit/test_tool_loop.py
View file

@ -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"}'}],
},
]