scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 353

Merge 443a353a72 into f96200bcea

Browse source
This commit is contained in:
Kolade Fajimi 2026-10-26 20:10:37 +00:00 • committed by GitHub
commit 39c17bdc58
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 385 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

47
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -16,7 +16,7 @@ from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, cast
from pydantic import TypeAdapter
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm import verbose_logger
@ -29,6 +29,7 @@ from litellm.types.llms.openai import (
ChatCompletionImageObject,
ChatCompletionReasoningItem,
ChatCompletionReasoningSummaryTextBlock,
ChatCompletionReasoningTextBlock,
ChatCompletionRedactedThinkingBlock,
ChatCompletionResponseMessage,
ChatCompletionTextObject,
@ -2257,6 +2258,31 @@ def reasoning_content_from_thinking_blocks(
ENCRYPTED_REASONING_SIGNATURE_PREFIX: Final = "litellm_encrypted_reasoning:"
REASONING_CONTENT_SIGNATURE_PREFIX: Final = "litellm_reasoning_content:"
class _ReasoningContentReplay(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
summary: tuple[str, ...]
content: tuple[str, ...]
encrypted_content: str | None = None
def reasoning_content_signature(
summary: tuple[str, ...], content: tuple[str, ...], encrypted_content: str | None
) -> str:
payload: Final = _ReasoningContentReplay(summary=summary, content=content, encrypted_content=encrypted_content)
return f"{REASONING_CONTENT_SIGNATURE_PREFIX}{payload.model_dump_json()}"
def _reasoning_content_from_signature(signature: object) -> _ReasoningContentReplay | None:
if not isinstance(signature, str) or not signature.startswith(REASONING_CONTENT_SIGNATURE_PREFIX):
return None
try:
return _ReasoningContentReplay.model_validate_json(signature.removeprefix(REASONING_CONTENT_SIGNATURE_PREFIX))
except ValidationError:
return None
def encrypted_reasoning_signature(encrypted_content: str) -> str:
@ -2270,11 +2296,16 @@ def encrypted_reasoning_signature(encrypted_content: str) -> str:
def _carries_encrypted_reasoning(signature: object) -> bool:
return isinstance(signature, str) and signature.startswith(ENCRYPTED_REASONING_SIGNATURE_PREFIX)
return isinstance(signature, str) and signature.startswith(
(ENCRYPTED_REASONING_SIGNATURE_PREFIX, REASONING_CONTENT_SIGNATURE_PREFIX)
)
def encrypted_content_from_signature(signature: object) -> str | None:
if not isinstance(signature, str) or not _carries_encrypted_reasoning(signature):
replay: Final = _reasoning_content_from_signature(signature)
if replay is not None:
return replay.encrypted_content
if not isinstance(signature, str) or not signature.startswith(ENCRYPTED_REASONING_SIGNATURE_PREFIX):
return None
return signature.removeprefix(ENCRYPTED_REASONING_SIGNATURE_PREFIX) or None
@ -2415,6 +2446,16 @@ def _reasoning_replay_group_key(indexed_block: tuple[int, Mapping[str, object]])
def _reasoning_item_from_block_group(group: tuple[Mapping[str, object], ...]) -> ChatCompletionReasoningItem | None:
replay: Final = _reasoning_content_from_signature(_encrypted_reasoning_field(group[0]))
if replay is not None:
return ChatCompletionReasoningItem(
type="reasoning",
summary=[
ChatCompletionReasoningSummaryTextBlock(type="summary_text", text=text) for text in replay.summary
],
content=[ChatCompletionReasoningTextBlock(type="reasoning_text", text=text) for text in replay.content],
**({"encrypted_content": replay.encrypted_content} if replay.encrypted_content is not None else {}),
)
summary: Final[list[ChatCompletionReasoningSummaryTextBlock]] = [ # mutable-ok: API message payload
ChatCompletionReasoningSummaryTextBlock(type="summary_text", text=text)
for block in group

80
litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py
View file

@ -11,9 +11,6 @@ from litellm import verbose_logger
from litellm._logging import redact_internal_details_from_client_message
from litellm._uuid import uuid
from litellm.exceptions import MidStreamFallbackError
from litellm.litellm_core_utils.prompt_templates.common_utils import (
encrypted_reasoning_signature,
)
from litellm.llms.anthropic.pass_through.messages.utils import (
INCOMPLETE_STREAM_ERROR_MESSAGE,
refusal_stop_details,
@ -144,6 +141,7 @@ class AnthropicResponsesStreamWrapper:
self._stream_failed = False
self._chunk_queue: deque[dict[str, object]] = deque()
self._refusal_text: str = ""
self._reasoning_text_parts_seen: frozenset[tuple[str | None, bool, int]] = frozenset()
self._sync_responses_iterator: Iterator[object] | None = None
def _make_message_start(self) -> dict[str, object]:
@ -187,13 +185,26 @@ class AnthropicResponsesStreamWrapper:
def _field(source: object, name: str) -> object:
return source.get(name) if isinstance(source, dict) else getattr(source, name, None)
def _emit_thinking_text(self, item_id: str | None, text: str) -> None:
existing_index: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
block_index: Final = (
existing_index
if existing_index >= 0
else self._open_block(item_id, {"type": "thinking", "thinking": "", "signature": ""})
)
self._chunk_queue.append(
{"type": "content_block_delta", "index": block_index, "delta": {"type": "thinking_delta", "thinking": text}}
)
def _close_reasoning_item(self, item: object, item_id: str | None) -> None:
reasoning_block: Final = LiteLLMAnthropicToResponsesAPIAdapter.thinking_block_from_reasoning_item(item)
existing_index: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
completed_text: Final = reasoning_block.get("thinking") if reasoning_block is not None else None
if existing_index < 0 and isinstance(completed_text, str) and completed_text:
self._emit_thinking_text(item_id, completed_text)
block_idx: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
encrypted_content: Final = self._field(item, "encrypted_content")
signature: Final = (
encrypted_reasoning_signature(encrypted_content)
if isinstance(encrypted_content, str) and encrypted_content
else None
reasoning_block.get("signature", reasoning_block.get("data")) if reasoning_block is not None else None
)
if block_idx < 0 and signature is None:
return
@ -217,9 +228,7 @@ class AnthropicResponsesStreamWrapper:
def _process_event(self, event: object) -> None:
"""Convert one Responses API event into zero or more Anthropic chunks queued for emission."""
event_type = getattr(event, "type", None)
if event_type is None and isinstance(event, dict):
event_type = event.get("type")
event_type: Final = self._field(event, "type")
if event_type is None:
return
@ -298,6 +307,10 @@ class AnthropicResponsesStreamWrapper:
if event_type == "response.reasoning_summary_part.added":
part_item_id: Final = self._field(event, "item_id")
if any(
item_id == part_item_id and not is_summary for item_id, is_summary, _ in self._reasoning_text_parts_seen
):
return
summary_index: Final = self._field(event, "summary_index")
part_block_idx: Final = (
self._item_id_to_block_index.get(part_item_id, -1) if isinstance(part_item_id, str) else -1
@ -316,25 +329,36 @@ class AnthropicResponsesStreamWrapper:
)
return
# ---- reasoning summary text delta ----
if event_type == "response.reasoning_summary_text.delta":
item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
if block_idx < 0:
if not delta:
return
block_idx = self._open_block(
item_id,
{"type": "thinking", "thinking": "", "signature": ""},
)
self._chunk_queue.append(
{
"type": "content_block_delta",
"index": block_idx,
"delta": {"type": "thinking_delta", "thinking": delta},
}
if event_type in (
"response.reasoning_summary_text.delta",
"response.reasoning_text.delta",
"response.reasoning_text.done",
):
reasoning_item_id: Final = self._field(event, "item_id")
reasoning_id: Final = reasoning_item_id if isinstance(reasoning_item_id, str) else None
is_summary: Final = event_type == "response.reasoning_summary_text.delta"
if any(
item_id == reasoning_id and summary != is_summary
for item_id, summary, _ in self._reasoning_text_parts_seen
):
return
content_index: Final = self._field(event, "summary_index" if is_summary else "content_index")
part_key: Final = (reasoning_id, is_summary, content_index if isinstance(content_index, int) else 0)
if event_type == "response.reasoning_text.done" and part_key in self._reasoning_text_parts_seen:
return
reasoning_text: Final = self._field(
event, "text" if event_type == "response.reasoning_text.done" else "delta"
)
if not isinstance(reasoning_text, str) or not reasoning_text:
return
if (
not is_summary
and part_key not in self._reasoning_text_parts_seen
and any(item_id == reasoning_id for item_id, _, _ in self._reasoning_text_parts_seen)
):
self._emit_thinking_text(reasoning_id, REASONING_SUMMARY_PART_SEPARATOR)
self._reasoning_text_parts_seen = self._reasoning_text_parts_seen | frozenset((part_key,))
self._emit_thinking_text(reasoning_id, reasoning_text)
return
# ---- function call arguments delta ----

70
litellm/llms/anthropic/pass_through/responses_adapters/transformation.py
View file

@ -10,10 +10,13 @@ from collections.abc import Iterable, Mapping, Sequence
from itertools import groupby
from typing import Any, Final, cast
from pydantic import BaseModel, ConfigDict
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_BOUNDARY,
TOOL_RESULT_IMAGE_PLACEHOLDER,
encrypted_reasoning_signature,
reasoning_content_signature,
responses_reasoning_items_from_thinking_blocks,
with_prompt_cache_breakpoint,
)
@ -55,6 +58,14 @@ REASONING_SUMMARY_PART_SEPARATOR: Final = "\n\n"
RESPONSES_INCLUDE_ENCRYPTED_REASONING: Final = "reasoning.encrypted_content"
class _ReasoningItem(BaseModel):
model_config = ConfigDict(frozen=True, from_attributes=True)
summary: tuple[object, ...] | None = None
content: tuple[object, ...] | None = None
encrypted_content: str | None = None
class LiteLLMAnthropicToResponsesAPIAdapter:
"""
Converts Anthropic /v1/messages requests to OpenAI Responses API format and
@ -247,10 +258,9 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
return str(getattr(part, "text", None) or "")
@classmethod
def _thinking_block_from_reasoning_item(
def thinking_block_from_reasoning_item(
cls,
summary: Iterable[object],
encrypted_content: object,
item: object,
) -> dict[str, object] | None: # mutable-ok: API message payload
"""The one Anthropic block for a Responses reasoning item.
@ -259,23 +269,37 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
turn replays the very item OpenAI produced; without it the signature stays empty,
since only Anthropic can sign a thinking block.
"""
text: Final = REASONING_SUMMARY_PART_SEPARATOR.join(
part_text for part in summary if (part_text := cls._summary_part_text(part))
reasoning: Final = _ReasoningItem.model_validate(item)
summary_texts: Final = tuple(
part_text for part in reasoning.summary or () if (part_text := cls._summary_part_text(part))
)
if not isinstance(encrypted_content, str) or not encrypted_content:
content_texts: Final = tuple(
part_text for part in reasoning.content or () if (part_text := cls._summary_part_text(part))
)
text: Final = REASONING_SUMMARY_PART_SEPARATOR.join(content_texts or summary_texts)
if content_texts:
content_signature: Final = reasoning_content_signature(
summary_texts, content_texts, reasoning.encrypted_content
)
return AnthropicResponseContentBlockThinking(
type="thinking", thinking=text, signature=content_signature
).model_dump()
if not reasoning.encrypted_content:
if not text:
return None
return AnthropicResponseContentBlockThinking(type="thinking", thinking=text, signature=None).model_dump()
signature: Final = encrypted_reasoning_signature(encrypted_content)
signature: Final = encrypted_reasoning_signature(reasoning.encrypted_content)
if not text:
return AnthropicResponseContentBlockRedactedThinking(type="redacted_thinking", data=signature).model_dump()
return AnthropicResponseContentBlockThinking(type="thinking", thinking=text, signature=signature).model_dump()
@staticmethod
def _assistant_block_group_key(indexed_block: tuple[int, Mapping[str, object]]) -> str:
"""Group a run of consecutive thinking blocks together; keep every other block alone."""
index, block = indexed_block
return "thinking" if block.get("type") in ("thinking", "redacted_thinking") else f"block:{index}"
block_type: Final = block.get("type")
if block_type in ("thinking", "redacted_thinking"):
return "thinking"
return "text" if block_type == "text" else f"block:{index}"
@classmethod
def _assistant_group_to_input_items(
@ -286,6 +310,14 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
if btype in ("thinking", "redacted_thinking"):
replayed: Final = responses_reasoning_items_from_thinking_blocks(group)
return tuple(dict(item) for item in replayed)
if btype == "text":
return (
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": block.get("text", "")} for block in group],
},
)
if btype == "tool_use":
return (
{
@ -448,19 +480,6 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
for _, group in groupby(enumerate(blocks), key=self._assistant_block_group_key)
for item in self._assistant_group_to_input_items(tuple(block for _, block in group))
)
asst_parts: list[dict[str, object]] = [ # mutable-ok: API message payload
{"type": "output_text", "text": block.get("text", "")}
for block in blocks
if block.get("type") == "text"
]
if asst_parts:
input_items.append(
{
"type": "message",
"role": "assistant",
"content": asst_parts,
}
)
return input_items
@ -741,7 +760,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
for item in response.output:
if isinstance(item, ResponseReasoningItem):
reasoning_block = self._thinking_block_from_reasoning_item(item.summary, item.encrypted_content)
reasoning_block = self.thinking_block_from_reasoning_item(item)
if reasoning_block is not None:
content.append(reasoning_block)
@ -793,10 +812,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
).model_dump()
)
elif item_type == "reasoning":
reasoning_block = self._thinking_block_from_reasoning_item(
cast(Iterable[object], item.get("summary") or ()), # cast-ok: untyped provider json
item.get("encrypted_content"),
)
reasoning_block = self.thinking_block_from_reasoning_item(item)
if reasoning_block is not None:
content.append(reasoning_block)
elif item_type == "function_call":

26
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -47857,6 +47857,13 @@
"ChatCompletionReasoningItem": {
"description": "Represents an OpenAI Responses API reasoning item for round-tripping in conversation history.",
"properties": {
"content": {
"items": {
"$ref": "#/components/schemas/ChatCompletionReasoningTextBlock"
},
"title": "Content",
"type": "array"
},
"encrypted_content": {
"anyOf": [
{
@ -47909,6 +47916,25 @@
"title": "ChatCompletionReasoningSummaryTextBlock",
"type": "object"
},
"ChatCompletionReasoningTextBlock": {
"properties": {
"text": {
"title": "Text",
"type": "string"
},
"type": {
"const": "reasoning_text",
"title": "Type",
"type": "string"
}
},
"required": [
"type",
"text"
],
"title": "ChatCompletionReasoningTextBlock",
"type": "object"
},
"ChatCompletionRedactedThinkingBlock": {
"properties": {
"cache_control": {

14
litellm/types/llms/openai.py
View file

@ -1,5 +1,5 @@
import builtins
from collections.abc import Iterable, Mapping
from collections.abc import Iterable, Mapping, Sequence
from enum import Enum
from os import PathLike
from typing import IO, TYPE_CHECKING, Any, Final, Generic, Literal, Optional, TypeAlias, Union
@ -711,13 +711,19 @@ class ChatCompletionReasoningSummaryTextBlock(TypedDict, total=False):
text: str
class ChatCompletionReasoningTextBlock(TypedDict):
type: ReadOnly[Literal["reasoning_text"]]
text: ReadOnly[str]
class ChatCompletionReasoningItem(TypedDict, total=False):
"""Represents an OpenAI Responses API reasoning item for round-tripping in conversation history."""
type: Required[Literal["reasoning"]]
id: str
encrypted_content: str | None
type: ReadOnly[Required[Literal["reasoning"]]]
id: ReadOnly[str]
encrypted_content: ReadOnly[str | None]
summary: ReadOnly[list[ChatCompletionReasoningSummaryTextBlock]]
content: ReadOnly[Sequence[ChatCompletionReasoningTextBlock]]
class WebSearchOptionsUserLocationApproximate(TypedDict, total=False):

30
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py
View file

@ -28,6 +28,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
merge_consecutive_system_messages,
native_redacted_thinking_data,
parse_tool_call_arguments,
reasoning_content_signature,
responses_reasoning_items_from_thinking_blocks,
salvage_concatenated_tool_arguments,
schema_grammar_branch_count,
@ -1994,6 +1995,35 @@ class TestEncryptedReasoningReplay:
def test_signature_round_trips_the_encrypted_content(self):
assert encrypted_content_from_signature(encrypted_reasoning_signature("gAAAA_bytes")) == "gAAAA_bytes"
def test_reasoning_content_signatures_keep_consecutive_items_separate(self) -> None:
first_signature: Final = reasoning_content_signature(("Brief summary",), ("First thought",), "opaque")
second_signature: Final = reasoning_content_signature((), ("Second thought",), None)
assert encrypted_content_from_signature(first_signature) == "opaque"
assert responses_reasoning_items_from_thinking_blocks(
(
{"type": "thinking", "thinking": "First thought", "signature": first_signature},
{"type": "thinking", "thinking": "Second thought", "signature": second_signature},
)
) == (
{
"type": "reasoning",
"summary": [{"type": "summary_text", "text": "Brief summary"}],
"content": [{"type": "reasoning_text", "text": "First thought"}],
"encrypted_content": "opaque",
},
{"type": "reasoning", "summary": [], "content": [{"type": "reasoning_text", "text": "Second thought"}]},
)
@pytest.mark.parametrize("payload", ["not-json", '{"summary":[],"content":[1]}'])
def test_malformed_reasoning_content_signature_keeps_readable_thinking(self, payload: str) -> None:
signature: Final = "litellm_reasoning_content:" + payload
assert encrypted_content_from_signature(signature) is None
assert responses_reasoning_items_from_thinking_blocks(
({"type": "thinking", "thinking": "Keep this thought", "signature": signature},)
) == ({"type": "reasoning", "summary": [{"type": "summary_text", "text": "Keep this thought"}]},)
@pytest.mark.parametrize(
"signature", [None, "", "ErcBCkgIValidAnthropicSignature", "litellm_encrypted_reasoning:", 7]
)

86
tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py
View file

@ -8,8 +8,11 @@ import json
import os
import sys
from types import SimpleNamespace
from typing import Final, Literal
import pytest
from anthropic.types import RawContentBlockDeltaEvent, RawMessageStreamEvent, SignatureDelta, ThinkingDelta
from pydantic import TypeAdapter
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
@ -25,6 +28,89 @@ from litellm.llms.anthropic.pass_through.responses_adapters.streaming_iterator i
from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse
@pytest.mark.asyncio
@pytest.mark.parametrize("delivery", ["delta", "done", "item"])
@pytest.mark.parametrize("encrypted_content", [None, "opaque-reasoning"])
@pytest.mark.parametrize("summary_order", ["absent", "first", "last"])
async def test_reasoning_text_reaches_client_and_survives_replay(
delivery: Literal["delta", "done", "item"],
encrypted_content: str | None,
summary_order: Literal["absent", "first", "last"],
) -> None:
from litellm.llms.anthropic.pass_through.responses_adapters.transformation import (
LiteLLMAnthropicToResponsesAPIAdapter,
)
from litellm.types.llms.anthropic import AnthropicMessagesRequest
reasoning: Final = {
"type": "reasoning",
"id": "rs_text",
"summary": []
if summary_order == "absent"
else [{"type": "summary_text", "text": "Tool"}, {"type": "summary_text", "text": "needed"}],
"content": [
{"type": "reasoning_text", "text": "Use a tool"},
{"type": "reasoning_text", "text": "Then answer"},
],
**({"encrypted_content": encrypted_content} if encrypted_content is not None else {}),
}
summary_events: Final = (
{"type": "response.reasoning_summary_text.delta", "item_id": "rs_text", "summary_index": 0, "delta": "Tool"},
{"type": "response.reasoning_summary_part.added", "item_id": "rs_text", "summary_index": 1},
{"type": "response.reasoning_summary_text.delta", "item_id": "rs_text", "summary_index": 1, "delta": "needed"},
)
events: Final = (
{"type": "response.created"},
{"type": "response.output_item.added", "item": {"type": "reasoning", "id": "rs_text"}},
*(summary_events if summary_order == "first" and delivery != "item" else ()),
*(
(
{"type": "response.reasoning_text.delta", "item_id": "rs_text", "content_index": 0, "delta": "Use "},
{"type": "response.reasoning_text.delta", "item_id": "rs_text", "content_index": 0, "delta": "a tool"},
)
if delivery == "delta"
else ()
),
*(
(
{"type": "response.reasoning_text.done", "item_id": "rs_text", "content_index": 0, "text": "Use a tool"},
{"type": "response.reasoning_text.done", "item_id": "rs_text", "content_index": 1, "text": "Then answer"},
)
if delivery != "item"
else ()
),
*(summary_events if summary_order == "last" and delivery != "item" else ()),
{"type": "response.output_item.done", "item": reasoning},
{"type": "response.completed"},
)
wrapper: Final = AnthropicResponsesStreamWrapper(responses_stream=iter(events), model="test-model")
chunks: Final = TypeAdapter(tuple[RawMessageStreamEvent, ...]).validate_python(tuple([c async for c in wrapper]))
thinking: Final = "".join(
event.delta.thinking
for event in chunks
if isinstance(event, RawContentBlockDeltaEvent) and isinstance(event.delta, ThinkingDelta)
)
signature: Final = "".join(
event.delta.signature
for event in chunks
if isinstance(event, RawContentBlockDeltaEvent) and isinstance(event.delta, SignatureDelta)
)
request: Final = AnthropicMessagesRequest(
model="test-model",
max_tokens=1024,
messages=[
{"role": "assistant", "content": [{"type": "thinking", "thinking": thinking, "signature": signature}]}
],
)
assert thinking == (
"Tool\n\nneeded" if summary_order == "first" and delivery != "item" else "Use a tool\n\nThen answer"
)
assert LiteLLMAnthropicToResponsesAPIAdapter().translate_messages_to_responses_input(request["messages"]) == [
{key: value for key, value in reasoning.items() if key != "id"}
]
def _process_all(events: list) -> list:
wrapper = AnthropicResponsesStreamWrapper(responses_stream=None, model="m")
for event in events:

79
tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py
View file

@ -33,6 +33,85 @@ from litellm.types.llms.anthropic import (
from litellm.types.llms.openai import ResponseAPIUsage
def test_parallel_tool_round_keeps_assistant_block_order() -> None:
request: Final = AnthropicMessagesRequest(
model="test-model",
max_tokens=1024,
messages=[
{"role": "user", "content": "Run both tools"},
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": "Use both tools"},
{"type": "text", "text": "Working"},
{"type": "tool_use", "id": "call_a", "name": "noop", "input": {}},
{"type": "tool_use", "id": "call_b", "name": "noop", "input": {}},
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "call_a", "content": "done a"},
{"type": "tool_result", "tool_use_id": "call_b", "content": "done b"},
],
},
],
)
assert _ADAPTER.translate_messages_to_responses_input(request["messages"]) == [
{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Run both tools"}]},
{"type": "reasoning", "summary": [{"type": "summary_text", "text": "Use both tools"}]},
{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Working"}]},
{"type": "function_call", "call_id": "call_a", "name": "noop", "arguments": "{}"},
{"type": "function_call", "call_id": "call_b", "name": "noop", "arguments": "{}"},
{"type": "function_call_output", "call_id": "call_a", "output": "done a"},
{"type": "function_call_output", "call_id": "call_b", "output": "done b"},
]
@pytest.mark.parametrize("encrypted_content", [None, "opaque-reasoning"])
@pytest.mark.parametrize("with_summary", [False, True])
@pytest.mark.parametrize("as_dict", [False, True])
def test_reasoning_content_survives_response_and_replay(
encrypted_content: str | None, with_summary: bool, as_dict: bool
) -> None:
from openai.types.responses import ResponseReasoningItem
from litellm.types.llms.openai import ResponsesAPIResponse
reasoning: Final = ResponseReasoningItem(
id="rs_content",
type="reasoning",
summary=[{"type": "summary_text", "text": "Brief summary"}] if with_summary else [],
content=[
{"type": "reasoning_text", "text": "First thought"},
{"type": "reasoning_text", "text": "Second thought"},
],
encrypted_content=encrypted_content,
)
response: Final = ResponsesAPIResponse(
id="resp_content",
created_at=1,
model="test-model",
object="response",
status="completed",
output=[reasoning.model_dump(exclude_none=True) if as_dict else reasoning],
parallel_tool_calls=True,
tool_choice="auto",
tools=[],
)
translated: Final = _ADAPTER.translate_response(response)
request: Final = AnthropicMessagesRequest(
model="test-model", max_tokens=1024, messages=[{"role": "assistant", "content": translated["content"]}]
)
assert translated["content"][0]["type"] == "thinking"
assert translated["content"][0]["thinking"] == "First thought\n\nSecond thought"
assert _ADAPTER.translate_messages_to_responses_input(request["messages"]) == [
{key: value for key, value in reasoning.model_dump(exclude_none=True).items() if key != "id"}
]
def _make_request(**overrides) -> AnthropicMessagesRequest:
base: dict = {
"model": "openai.gpt-5.1-codex",

2
tests/unit/llms/openai/responses/test_openai_count_tokens_transformation.py
View file

@ -591,12 +591,12 @@ def test_anthropic_messages_keep_every_block_the_chat_converter_drops() -> None:
{"type": "input_image", "image_url": f"data:image/png;base64,{PNG_BASE64}", "detail": "auto"},
],
},
{"type": "function_call", "call_id": "toolu_01", "name": "lookup_clause", "arguments": '{"id": "14"}'},
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Let me pull clause 14."}],
},
{"type": "function_call", "call_id": "toolu_01", "name": "lookup_clause", "arguments": '{"id": "14"}'},
{"type": "function_call_output", "call_id": "toolu_01", "output": "14. Indemnification."},
)

14
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -28872,6 +28872,8 @@ export interface components {
* @description Represents an OpenAI Responses API reasoning item for round-tripping in conversation history.
*/
ChatCompletionReasoningItem: {
/** Content */
content?: components["schemas"]["ChatCompletionReasoningTextBlock"][];
/** Encrypted Content */
encrypted_content?: string | null;
/** Id */
@ -28898,6 +28900,18 @@ export interface components {
} & {
[key: string]: unknown;
};
/** ChatCompletionReasoningTextBlock */
ChatCompletionReasoningTextBlock: {
/** Text */
text: string;
/**
* Type
* @constant
*/
type: "reasoning_text";
} & {
[key: string]: unknown;
};
/** ChatCompletionRedactedThinkingBlock */
ChatCompletionRedactedThinkingBlock: {
/** Cache Control */