mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-13 08:52:02 +00:00
fix(responses): replay the original Responses item order, phase, and boundaries on the next chat turn
This commit is contained in:
mateo-berri
parent
d84a2d4452
commit
c8d7cdad97
7 changed files with 1036 additions and 33 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,392 @@
|
|||
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 _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 _PhasedMessage(_Frozen):
|
||||
phase: Phase | None = None
|
||||
provider_specific_fields: _PhaseFields | None = None
|
||||
|
||||
|
||||
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 _chat_message(message: object) -> _ChatMessage | None:
|
||||
try:
|
||||
return _ChatMessage.model_validate(_as_json(message))
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def explicit_phase(message: object) -> Phase | None:
|
||||
try:
|
||||
phased: Final = _PhasedMessage.model_validate(_as_json(message))
|
||||
except ValidationError:
|
||||
return None
|
||||
fields: Final = phased.provider_specific_fields
|
||||
return phased.phase or (fields.phase if fields else None)
|
||||
|
||||
|
||||
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()).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 _last_phase(items: Sequence[_OutputItem | None]) -> Phase | None:
|
||||
messages: Final = tuple(item for item in items if isinstance(item, _OutputMessage) and item.phase is not None)
|
||||
return messages[-1].phase if messages 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) -> Mapping[str, object]:
|
||||
parsed: Final = tuple(map(_parsed_output_item, output_items))
|
||||
phase: Final = _last_phase(parsed)
|
||||
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 _covers(entries: Sequence[_IndexedEntry], count: int) -> bool:
|
||||
return sorted(entry.index for entry in entries) == list(range(count))
|
||||
|
||||
|
||||
def _message_texts(entries: Sequence[_MessageEntry], content: str) -> tuple[tuple[str, ...], ...] | None:
|
||||
lengths: Final = tuple(chain.from_iterable(entry.parts for entry in entries))
|
||||
if sum(lengths) != len(content):
|
||||
return None
|
||||
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 _resolve(
|
||||
message: _ChatMessage, entries: Sequence[_MessageEntry | _IndexedEntry], replay_tool_search: bool
|
||||
) -> ReplayedTurn | None:
|
||||
messages: Final = tuple(entry for entry in entries if isinstance(entry, _MessageEntry))
|
||||
indexed: Final = tuple(entry for entry in entries if isinstance(entry, _IndexedEntry))
|
||||
reasoning: Final = tuple(entry for entry in indexed if entry.type == "reasoning")
|
||||
calls: Final = tuple(entry for entry in indexed if entry.type in ("function_call", "custom_tool_call"))
|
||||
searches: Final = tuple(entry for entry in indexed if entry.type in ("tool_search_call", "tool_search_output"))
|
||||
tool_searches: Final = _tool_search_entries(message)
|
||||
texts: Final = _message_texts(messages, message.content or "")
|
||||
if (
|
||||
texts is None
|
||||
or (searches and not replay_tool_search)
|
||||
or not _covers(reasoning, len(message.reasoning_items or ()))
|
||||
or not _covers(calls, len(message.tool_calls or ()))
|
||||
or not _covers(searches, len(tool_searches))
|
||||
or any(tool_searches[entry.index].get("type") != entry.type for entry in searches)
|
||||
):
|
||||
return None
|
||||
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_searches,
|
||||
)
|
||||
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
|
||||
return _resolve(chat, stored.items, replay_tool_search)
|
||||
|
|
@ -25,6 +25,14 @@ 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,
|
||||
explicit_phase,
|
||||
output_message_text,
|
||||
reasoning_input_item,
|
||||
replay_fields,
|
||||
replay_items,
|
||||
)
|
||||
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 +498,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 +517,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 +534,49 @@ def _fold_choices_into_one(
|
|||
)
|
||||
|
||||
|
||||
def _with_replay_fields(choice: "Choices", output_items: Sequence[object], *, completed: bool) -> "Choices":
|
||||
fields: Final = replay_fields(output_items, choice.message, completed=completed)
|
||||
if not fields:
|
||||
return choice
|
||||
message: Final = choice.message.model_copy(
|
||||
update={"provider_specific_fields": {**(choice.message.provider_specific_fields or {}), **fields}}
|
||||
)
|
||||
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
|
||||
try:
|
||||
items: Final = _STREAMED_OUTPUT_ITEMS.validate_python(output_payload)
|
||||
except ValidationError:
|
||||
return dict(tool_search_fields) if tool_search_fields else None
|
||||
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)}
|
||||
return fields or None
|
||||
|
||||
|
||||
def _assistant_phase_field(msg: object, replay_openai_history: bool) -> Mapping[str, object]:
|
||||
phase: Final = explicit_phase(msg) if replay_openai_history else None
|
||||
return {PHASE_FIELD: phase} if phase else {}
|
||||
|
||||
|
||||
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 +718,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 +847,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 +870,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 +902,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 +970,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
"content": self._convert_content_to_responses_format(
|
||||
content, "assistant", drop_params=drop_params
|
||||
),
|
||||
**_assistant_phase_field(msg, replay_openai_history),
|
||||
}
|
||||
)
|
||||
for tool_call in tool_calls:
|
||||
|
|
@ -955,6 +1011,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"),
|
||||
}
|
||||
)
|
||||
elif role == "assistant":
|
||||
|
|
@ -1081,6 +1138,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 +1223,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 +1234,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 +1258,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 +1267,17 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
|
|||
|
||||
for item in output_items:
|
||||
if isinstance(item, ResponseReasoningItem):
|
||||
pending_reasoning_item = _build_reasoning_item(
|
||||
built_reasoning_item = _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 pending_reasoning_item["summary"] if s.get("text"))
|
||||
pending_reasoning_items = (*pending_reasoning_items, built_reasoning_item)
|
||||
reasoning_content = " ".join(
|
||||
s["text"]
|
||||
for s in chain.from_iterable(built["summary"] for built in pending_reasoning_items)
|
||||
if s.get("text")
|
||||
)
|
||||
|
||||
elif isinstance(item, ResponseOutputMessage):
|
||||
for content in item.content:
|
||||
|
|
@ -1227,7 +1294,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 +1307,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 +1367,15 @@ 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,
|
||||
)
|
||||
return [_with_replay_fields(folded, output_items, completed=completed) if replay_openai_history else folded]
|
||||
|
||||
@staticmethod
|
||||
def _build_empty_choice(
|
||||
|
|
@ -1443,6 +1509,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 +1577,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]:
|
||||
|
|
@ -1898,8 +1974,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
|
||||
|
|
@ -1945,6 +2023,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.
|
||||
|
|
@ -1952,6 +2032,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
|
||||
|
|
@ -2162,6 +2244,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"):
|
||||
|
|
@ -2177,8 +2263,15 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
|
|||
delta=Delta(
|
||||
content="",
|
||||
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,
|
||||
|
|
@ -2224,7 +2317,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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,410 @@
|
|||
import json
|
||||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
|
||||
import litellm
|
||||
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",
|
||||
) -> dict[str, object]:
|
||||
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 httpx.Response(200, json=_response(output, status))
|
||||
)
|
||||
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,
|
||||
)
|
||||
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}"
|
||||
) -> 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",
|
||||
)
|
||||
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"
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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"},
|
||||
]
|
||||
|
|
@ -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_the_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])[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
|
||||
|
|
|
|||
|
|
@ -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"}'}],
|
||||
},
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue