From e5d5e842e824afbfc21428fa8472cfc3babca479 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 11 Jun 2026 20:42:54 +0000 Subject: [PATCH] feat(translation): response and stream paths, async engine, and the flag-gated completion() seam Phase 3 of the anthropic port (M3): - ir.py gains the response IR (ChatResponse, ResponseUsage, finish reasons) and stream-event union; providers/anthropic parse_response/parse_stream reproduce v1 exactly for the v2-reachable surface (tool-name reverse mapping recomputed from the request, json_tool_call rewrite, usage cache folding, redacted thinking) and fail loudly on block types a v2-sent request cannot trigger - inbound/openai_chat serialize_response/serialize_stream mirror v1's ModelResponse / CustomStreamWrapper shapes chunk-for-chunk (role on first chunk, tool-index counting, thinking provider fields, finish chunk) - engine/: pipeline.py (prepare -> send split so the fallback decision happens before any I/O; async-first, sync is one asyncio.run wrapper at the seam), http.py (injected HttpPort + ExecuteError values), stream.py (the ONE accumulator: provider lines -> IR events -> chunk bodies) - litellm/translation_seam.py (outside the package) adapts deps from litellm ambient state, ModelResponse/ModelResponseStream envelopes, and owns the completion() fork; litellm.translation_v2_providers allowlist global seeded from LITELLM_TRANSLATION_V2_PROVIDERS (off by default, yaml- configurable via the litellm_settings setattr fallback) - main.py anthropic branch forks to the seam; streaming and modify_params traffic stay on v1 (documented follow-ups); once sent, provider errors raise the v1 exception contract, never a silent re-send - differential gates: responses (4 full-cycle shapes incl. sanitized tool names and json_tool_call), streams (3 SSE replays vs the real CustomStreamWrapper), seam tests over respx (serve, fallback, 429 contract, async path); DIFFERENTIAL_REPORT.md committed as the merge artifact: 53 rows, 0 divergent --- litellm/__init__.py | 17 +- litellm/main.py | 20 + litellm/translation/deps.py | 7 +- litellm/translation/engine/http.py | 78 ++++ litellm/translation/engine/pipeline.py | 132 ++++++- litellm/translation/engine/stream.py | 57 +++ .../inbound/openai_chat/response.py | 143 +++++++ .../translation/inbound/openai_chat/stream.py | 150 ++++++++ litellm/translation/ir.py | 158 ++++++++ .../providers/anthropic/response.py | 224 +++++++++++ .../translation/providers/anthropic/stream.py | 200 ++++++++++ litellm/translation_seam.py | 356 ++++++++++++++++++ .../translation/DIFFERENTIAL_REPORT.md | 77 ++++ tests/test_litellm/translation/conftest.py | 5 +- .../generate_differential_report.py | 83 ++++ .../test_differential_anthropic_response.py | 238 ++++++++++++ .../test_differential_anthropic_stream.py | 245 ++++++++++++ tests/test_litellm/translation/test_seam.py | 139 +++++++ 18 files changed, 2308 insertions(+), 21 deletions(-) create mode 100644 litellm/translation/engine/http.py create mode 100644 litellm/translation/engine/stream.py create mode 100644 litellm/translation/inbound/openai_chat/response.py create mode 100644 litellm/translation/inbound/openai_chat/stream.py create mode 100644 litellm/translation/providers/anthropic/response.py create mode 100644 litellm/translation/providers/anthropic/stream.py create mode 100644 litellm/translation_seam.py create mode 100644 tests/test_litellm/translation/DIFFERENTIAL_REPORT.md create mode 100644 tests/test_litellm/translation/generate_differential_report.py create mode 100644 tests/test_litellm/translation/test_differential_anthropic_response.py create mode 100644 tests/test_litellm/translation/test_differential_anthropic_stream.py create mode 100644 tests/test_litellm/translation/test_seam.py diff --git a/litellm/__init__.py b/litellm/__init__.py index e6c30e1228..e489d8425f 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -234,6 +234,13 @@ use_chat_completions_url_for_anthropic_messages: bool = bool( route_all_chat_openai_to_responses: bool = ( os.getenv("LITELLM_ROUTE_ALL_CHAT_OPENAI_TO_RESPONSES", "false").lower() == "true" ) # When True, routes all OpenAI /chat/completions requests through the Responses API bridge +translation_v2_providers: List[str] = [ + entry.strip() + for entry in os.getenv("LITELLM_TRANSLATION_V2_PROVIDERS", "").split(",") + if entry.strip() +] # Per-provider opt-in allowlist for the translation v2 rewrite (litellm/translation). +# Off by default; e.g. "anthropic" or "anthropic,bedrock_converse". Any request whose +# shape is outside v2's proven surface transparently falls back to v1. # When True, Gemini/Vertex Live setup is deferred until client `session.update`. # Default False preserves historical behavior (auto-send setup on connect). gemini_live_defer_setup: bool = ( @@ -1356,16 +1363,6 @@ from .interactions.agents.main import ( alist_versions as alist_agent_versions, list_versions as list_agent_versions, ) -from .skills.main import ( - create_skill, - acreate_skill, - list_skills, - alist_skills, - get_skill, - aget_skill, - delete_skill, - adelete_skill, -) from .containers.main import * from .ocr.main import * from .rag.main import * diff --git a/litellm/main.py b/litellm/main.py index 02609217dd..c838421b90 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -2949,6 +2949,26 @@ def completion( # type: ignore # noqa: PLR0915 "LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX is set, skipping /v1/messages suffix" ) + if litellm.translation_v2_providers: + from litellm import translation_seam + + _v2_response = translation_seam.try_completion_v2( + model=model, + messages=messages, + optional_param_args=optional_param_args, + non_default_params=non_default_params, + api_key=api_key, + api_base=api_base, + timeout=timeout, + stream=stream, + acompletion=acompletion, + logging_obj=logging, + model_response=model_response, + request_drop_params=kwargs.get("drop_params"), + ) + if _v2_response is not None: + return _v2_response + response = anthropic_chat_completions.completion( model=model, messages=messages, diff --git a/litellm/translation/deps.py b/litellm/translation/deps.py index b371f4a21a..f6a9a5db05 100644 --- a/litellm/translation/deps.py +++ b/litellm/translation/deps.py @@ -9,8 +9,8 @@ free of upward imports. from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass -from typing import Callable, Optional @dataclass(frozen=True) @@ -34,9 +34,10 @@ class TranslationDeps: the contract. """ - max_tokens_for_model: Callable[[str], Optional[int]] + max_tokens_for_model: Callable[[str], int | None] supports_capability: Callable[[str, str], bool] - capability_flag: Callable[[str, str], Optional[bool]] + capability_flag: Callable[[str, str], bool | None] + count_response_tokens: Callable[[str], int] drop_params: bool drop_params_global: bool modify_params: bool diff --git a/litellm/translation/engine/http.py b/litellm/translation/engine/http.py new file mode 100644 index 0000000000..8745811a31 --- /dev/null +++ b/litellm/translation/engine/http.py @@ -0,0 +1,78 @@ +"""The injected HTTP port: the only I/O boundary in the package. + +The package never creates a client; the seam passes an object satisfying +``HttpPort`` (functional core, imperative shell). A non-2xx response is a +value (``ProviderHttpError``) the seam converts into the provider exception +contract, never an exception inside the package. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Literal, Protocol + +from expression import case, tag, tagged_union +from typing_extensions import assert_never + +from ..errors import TranslationError +from ..ir import PlainJson + + +@dataclass(frozen=True) +class Endpoint: + """Where and how to reach the provider; built by the seam (URL suffixing, + auth headers, beta headers are v1-side concerns).""" + + url: str + headers: Mapping[str, str] + timeout_seconds: float + + +@dataclass(frozen=True) +class HttpResponse: + status_code: int + body: PlainJson + text: str + headers: Mapping[str, str] + + +class HttpPort(Protocol): + async def post_json( + self, endpoint: Endpoint, body: Mapping[str, PlainJson] + ) -> HttpResponse: ... + + +@dataclass(frozen=True) +class ProviderHttpError: + """A non-2xx provider response; carries everything the seam needs to + raise the same exception v1 would.""" + + status_code: int + text: str + headers: Mapping[str, str] + + +@tagged_union(frozen=True) +class ExecuteError: + tag: Literal["translation", "provider_http"] = tag() + + translation: TranslationError = case() + provider_http: ProviderHttpError = case() + + @staticmethod + def of_translation(value: TranslationError) -> ExecuteError: + return ExecuteError(translation=value) + + @staticmethod + def of_provider_http(value: ProviderHttpError) -> ExecuteError: + return ExecuteError(provider_http=value) + + @property + def summary(self) -> str: + match self.tag: + case "translation": + return self.translation.summary + case "provider_http": + return f"provider returned HTTP {self.provider_http.status_code}" + assert_never(self.tag) diff --git a/litellm/translation/engine/pipeline.py b/litellm/translation/engine/pipeline.py index ce55d941e6..75e3ddf9bd 100644 --- a/litellm/translation/engine/pipeline.py +++ b/litellm/translation/engine/pipeline.py @@ -1,26 +1,36 @@ -"""Composition and the public translate entry point. +"""Async-first composition and the public translation entry points. -``translate_chat_request`` composes the inbound parse with the provider -serialize under injected deps; the whole pipeline returns one ``Result`` and -never raises. Any error means "outside v2's proven surface" and the dispatch -seam falls back to v1, so no request ever loses a feature silently. +``translate_chat_request`` is the pure request transform (parse -> serialize) +the differential gate runs. ``execute_chat_request`` is the full async +pipeline: translate, send through the injected HTTP port (popping the +``json_mode`` transform-seam marker exactly like v1's HTTP handler), parse +the provider response, and serialize the outbound body. Everything returns +one ``Result`` and never raises; sync callers get the one wrapper the seam +provides (v1's completion() already runs on an executor thread). """ from __future__ import annotations from collections.abc import Callable, Mapping +from dataclasses import dataclass from types import MappingProxyType -from expression import Error, Result +from expression import Error, Ok, Result from ..deps import TranslationDeps from ..dispatch import Provider from ..errors import TranslateResult, TranslationError from ..inbound.openai_chat import parse_request -from ..ir import Body, ChatRequest +from ..inbound.openai_chat.response import serialize_response +from ..ir import Body, ChatRequest, ChatResponse, PlainJson from ..providers.anthropic import serialize_request +from ..providers.anthropic.response import parse_response +from .http import Endpoint, ExecuteError, HttpPort, ProviderHttpError _Serializer = Callable[[ChatRequest, TranslationDeps], Result[Body, TranslationError]] +_ResponseParser = Callable[ + [PlainJson, ChatRequest], Result[ChatResponse, TranslationError] +] _SERIALIZERS: Mapping[Provider, _Serializer] = MappingProxyType( { @@ -28,6 +38,12 @@ _SERIALIZERS: Mapping[Provider, _Serializer] = MappingProxyType( } ) +_RESPONSE_PARSERS: Mapping[Provider, _ResponseParser] = MappingProxyType( + { + "anthropic": parse_response, + } +) + def translate_chat_request( raw: Mapping[str, object], provider: Provider, deps: TranslationDeps @@ -40,3 +56,105 @@ def translate_chat_request( ) ) return parse_request(raw).bind(lambda request: serializer(request, deps)) + + +def translate_chat_response( + raw_response: PlainJson, + request: ChatRequest, + provider: Provider, + deps: TranslationDeps, +) -> TranslateResult: + parser = _RESPONSE_PARSERS.get(provider) + if parser is None: + return Error( + TranslationError.of_unsupported( + f"provider {provider!r} has no v2 response parser yet" + ) + ) + return parser(raw_response, request).map( + lambda response: serialize_response(response, deps) + ) + + +@dataclass(frozen=True) +class PreparedRequest: + """A request that passed the fail-closed translation; from here on the + seam is committed to v2 (an HTTP or response failure surfaces as the + provider error contract, never a silent re-send through v1).""" + + request: ChatRequest + body: Body + + +def prepare_chat_request( + raw: Mapping[str, object], provider: Provider, deps: TranslationDeps +) -> Result[PreparedRequest, TranslationError]: + serializer = _SERIALIZERS.get(provider) + if serializer is None or provider not in _RESPONSE_PARSERS: + return Error( + TranslationError.of_unsupported( + f"provider {provider!r} is not fully ported to v2 yet" + ) + ) + match parse_request(raw): + case Result(tag="ok", ok=request): + pass + case Result(error=parse_err): + return Error(parse_err) + return serializer(request, deps).map( + lambda body: PreparedRequest(request=request, body=body) + ) + + +def wire_body(prepared: PreparedRequest) -> Body: + """v1's HTTP handler pops json_mode from optional_params before the wire; + the marker exists only at the transform seam.""" + return {key: value for key, value in prepared.body.items() if key != "json_mode"} + + +async def send_prepared( + prepared: PreparedRequest, + provider: Provider, + deps: TranslationDeps, + http: HttpPort, + endpoint: Endpoint, +) -> Result[Body, ExecuteError]: + parser = _RESPONSE_PARSERS.get(provider) + if parser is None: + return Error( + ExecuteError.of_translation( + TranslationError.of_unsupported( + f"provider {provider!r} has no v2 response parser yet" + ) + ) + ) + response = await http.post_json(endpoint, wire_body(prepared)) + if response.status_code < 200 or response.status_code >= 300: + return Error( + ExecuteError.of_provider_http( + ProviderHttpError( + status_code=response.status_code, + text=response.text, + headers=response.headers, + ) + ) + ) + match parser(response.body, prepared.request): + case Result(tag="ok", ok=chat_response): + return Ok(serialize_response(chat_response, deps)) + case Result(error=response_err): + return Error(ExecuteError.of_translation(response_err)) + + +async def execute_chat_request( + raw: Mapping[str, object], + provider: Provider, + deps: TranslationDeps, + http: HttpPort, + endpoint: Endpoint, +) -> Result[Body, ExecuteError]: + match prepare_chat_request(raw, provider, deps): + case Result(tag="ok", ok=prepared): + return await send_prepared(prepared, provider, deps, http, endpoint) + case Result(error=err): + return Error(ExecuteError.of_translation(err)) diff --git a/litellm/translation/engine/stream.py b/litellm/translation/engine/stream.py new file mode 100644 index 0000000000..ead2798989 --- /dev/null +++ b/litellm/translation/engine/stream.py @@ -0,0 +1,57 @@ +"""The ONE stream accumulator: provider wire lines -> IR events -> chunks. + +Async-first: ``chunk_stream`` is the production shape (an async line stream +in, OpenAI chunk bodies out); ``fold_lines`` is the synchronous fold over a +recorded stream that the differential tests replay. Both compose an injected +provider line parser with the inbound chunk fold, so adding a provider or an +inbound schema never touches this file. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable, Iterable + +from expression import Error, Ok, Result +from expression.collections import Block + +from ..errors import TranslationError +from ..inbound.openai_chat.stream import StreamState, initial_state, step +from ..ir import Body, StreamEvent + +ParseLine = Callable[[str], Result[StreamEvent | None, TranslationError]] + + +def fold_lines( + lines: Iterable[str], parse_line: ParseLine +) -> Result[Block[Body], TranslationError]: + state: StreamState = initial_state() + chunks: list[Body] = [] + for line in lines: + match parse_line(line): + case Result(tag="ok", ok=event): + if event is None: + continue + state, emitted = step(state, event) + chunks.extend(emitted) # nosemgrep: translation-no-mutation + case Result(error=err): + return Error(err) + return Ok(Block.of_seq(chunks)) + + +async def chunk_stream( + lines: AsyncIterator[str], parse_line: ParseLine +) -> AsyncIterator[Result[Body, TranslationError]]: + """Async-first form: yields one Result per outbound chunk; the first + error ends the stream (the seam surfaces it as a provider error).""" + state: StreamState = initial_state() + async for line in lines: + match parse_line(line): + case Result(tag="ok", ok=event): + if event is None: + continue + state, emitted = step(state, event) + for chunk in emitted: + yield Ok(chunk) + case Result(error=err): + yield Error(err) + return diff --git a/litellm/translation/inbound/openai_chat/response.py b/litellm/translation/inbound/openai_chat/response.py new file mode 100644 index 0000000000..9a1eb3e85e --- /dev/null +++ b/litellm/translation/inbound/openai_chat/response.py @@ -0,0 +1,143 @@ +"""IR ``ChatResponse`` -> OpenAI chat-completion response body. + +Emits the plain dict the seam feeds into ``ModelResponse`` (which owns the +ambient envelope: chatcmpl id, created timestamp). Field shapes mirror what +v1's ``transform_parsed_response`` builds, including the usage detail +wrappers and the always-present ``provider_specific_fields`` keys. +""" + +from __future__ import annotations + +import json + +from expression import Option +from expression.collections import Block + +from ...deps import TranslationDeps +from ...ir import Body, ChatResponse, ContentBlock, PlainJson, ResponseUsage + + +def serialize_response(response: ChatResponse, deps: TranslationDeps) -> Body: + text = "".join(block.text.text for block in response.content if block.tag == "text") + tool_calls = _tool_calls(response.content) + thinking_blocks = _thinking_blocks(response.content) + reasoning: str | None = None + if thinking_blocks is not None: + reasoning = "".join( + block.thinking.thinking + for block in response.content + if block.tag == "thinking" + ) + message: dict[str, PlainJson] + if response.synthesized_json_content: + # v1's json-mode replacement is a bare Message(content=...): no + # provider fields, no reasoning, no thinking blocks. + message = {"role": "assistant", "content": text or None} + else: + message = { + "role": "assistant", + "content": text or None, + "tool_calls": tool_calls, + "reasoning_content": reasoning, + "thinking_blocks": thinking_blocks, + "provider_specific_fields": { + "citations": None, + "thinking_blocks": thinking_blocks, + }, + } + return { + "object": "chat.completion", + "model": response.model, + "choices": [ + { + "index": 0, + "finish_reason": response.finish, + "message": message, + } + ], + "usage": _usage_json(response.usage, reasoning, deps), + } + + +def _tool_calls(content: Block[ContentBlock]) -> PlainJson: + calls: list[PlainJson] = [ + { + "id": block.tool_use.id, + "type": "function", + "function": { + "name": block.tool_use.name, + "arguments": json.dumps(block.tool_use.arguments.value), + }, + "index": index, + } + for index, block in enumerate(content) + if block.tag == "tool_use" + ] + return calls or None + + +def _thinking_blocks(content: Block[ContentBlock]) -> PlainJson: + blocks = [ + _thinking_block_json(block) + for block in content + if block.tag in ("thinking", "redacted_thinking") + ] + return blocks or None + + +def _thinking_block_json(block: ContentBlock) -> PlainJson: + if block.tag == "redacted_thinking": + return {"type": "redacted_thinking", "data": block.redacted_thinking.data} + thinking = block.thinking + base: dict[str, PlainJson] = {"type": "thinking", "thinking": thinking.thinking} + match thinking.signature: + case Option(tag="some", some=signature): + return {**base, "signature": signature} + case _: + return base + + +def _usage_json( + usage: ResponseUsage, reasoning: str | None, deps: TranslationDeps +) -> PlainJson: + """v1 ``calculate_usage``: cache tokens fold into prompt_tokens, the raw + input count lands in details.text_tokens, and reasoning tokens are + estimated by the injected token counter, capped at completion tokens.""" + prompt_tokens = ( + usage.input_tokens + + usage.cache_creation_input_tokens + + usage.cache_read_input_tokens + ) + completion_tokens = usage.output_tokens + estimated = deps.count_response_tokens(reasoning) if reasoning else 0 + reasoning_tokens = min(estimated, completion_tokens) + creation_details: PlainJson = None + match usage.cache_creation: + case Option(tag="some", some=details): + creation_details = { + "ephemeral_5m_input_tokens": details.five_minute.default_value(None), + "ephemeral_1h_input_tokens": details.one_hour.default_value(None), + } + case _: + creation_details = None + return { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + "prompt_tokens_details": { + "cached_tokens": usage.cache_read_input_tokens, + "cache_creation_tokens": usage.cache_creation_input_tokens, + "cache_creation_token_details": creation_details, + "text_tokens": usage.input_tokens, + }, + "completion_tokens_details": { + "reasoning_tokens": max(0, reasoning_tokens), + "text_tokens": ( + completion_tokens - reasoning_tokens + if reasoning_tokens > 0 + else completion_tokens + ), + }, + "cache_creation_input_tokens": usage.cache_creation_input_tokens, + "cache_read_input_tokens": usage.cache_read_input_tokens, + } diff --git a/litellm/translation/inbound/openai_chat/stream.py b/litellm/translation/inbound/openai_chat/stream.py new file mode 100644 index 0000000000..a24dbb80df --- /dev/null +++ b/litellm/translation/inbound/openai_chat/stream.py @@ -0,0 +1,150 @@ +"""IR stream events -> OpenAI chat-completion chunk bodies. + +A pure fold: ``step(state, event)`` returns the next state plus zero or more +chunk bodies. The shapes mirror what v1's CustomStreamWrapper emits for the +anthropic iterator: the first content-bearing chunk carries +``role: "assistant"``, tool chunks carry ``content: ""`` beside the tool +delta, tool indices count tool blocks (not content blocks), thinking deltas +ride ``reasoning_content`` + ``thinking_blocks`` + provider fields, and the +finish chunk is an empty delta with the mapped finish reason. The seam wraps +each body in ``ModelResponseStream`` (ambient id/created). +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from typing_extensions import assert_never + +from ...ir import Body, PlainJson, StreamEvent + + +@dataclass(frozen=True) +class StreamState: + model: str + sent_role: bool + tool_index: int + + +def initial_state() -> StreamState: + return StreamState(model="", sent_role=False, tool_index=-1) + + +_StepResult = tuple[StreamState, tuple[Body, ...]] + + +def step(state: StreamState, event: StreamEvent) -> _StepResult: + match event.tag: + case "start": + return ( + StreamState( + model=event.start.model, + sent_role=state.sent_role, + tool_index=state.tool_index, + ), + (), + ) + case "text_delta": + return _emit(state, {"content": event.text_delta.text}) + case "tool_use_start": + started = StreamState( + model=state.model, + sent_role=state.sent_role, + tool_index=state.tool_index + 1, + ) + return _emit( + started, + { + "content": "", + "tool_calls": [ + { + "id": event.tool_use_start.id, + "type": "function", + "function": { + "name": event.tool_use_start.name, + "arguments": "", + }, + "index": started.tool_index, + } + ], + }, + ) + case "tool_args_delta": + return _emit( + state, + { + "content": "", + "tool_calls": [ + { + "id": None, + "type": "function", + "function": { + "name": None, + "arguments": event.tool_args_delta.partial_json, + }, + "index": max(state.tool_index, 0), + } + ], + }, + ) + case "thinking_delta": + return _emit( + state, + _thinking_delta_body(event.thinking_delta.thinking, signature=""), + ) + case "signature_delta": + return _emit( + state, + _thinking_delta_body("", signature=event.signature_delta.signature), + ) + case "finish": + chunk: Body = { + "model": state.model, + "object": "chat.completion.chunk", + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": event.finish.finish, + } + ], + } + return state, (chunk,) + case "stop": + return state, () + assert_never(event.tag) + + +def _thinking_delta_body(thinking: str, signature: str) -> dict[str, PlainJson]: + block: PlainJson = { + "type": "thinking", + "thinking": thinking, + "signature": signature, + } + return { + "content": "", + "reasoning_content": thinking, + "thinking_blocks": [block], + "provider_specific_fields": {"thinking_blocks": [block]}, + } + + +def _emit(state: StreamState, delta: dict[str, PlainJson]) -> _StepResult: + role: str | None = None if state.sent_role else "assistant" + # v1's wrapper always sets provider_specific_fields (None when absent) on + # content-bearing deltas; the finish chunk's empty delta never has it. + body: Body = { + "model": state.model, + "object": "chat.completion.chunk", + "choices": [ + { + "index": 0, + "delta": {"role": role, "provider_specific_fields": None, **delta}, + "finish_reason": None, + } + ], + } + next_state = StreamState( + model=state.model, sent_role=True, tool_index=state.tool_index + ) + return next_state, (body,) diff --git a/litellm/translation/ir.py b/litellm/translation/ir.py index 4339121cc2..883a0afb87 100644 --- a/litellm/translation/ir.py +++ b/litellm/translation/ir.py @@ -357,3 +357,161 @@ def has_tool_blocks(messages: Block[Message]) -> bool: for message in messages for block in message.content ) + + +# -------------------------------------------------------------------------- +# Response IR: what a provider's parse_response yields and an inbound +# serialize_response consumes. Content reuses ContentBlock (text, tool_use, +# thinking, redacted_thinking are the cases a v2-sent request can produce; +# anything else fails parse loudly because the fail-closed request surface +# cannot trigger it). +# -------------------------------------------------------------------------- + +FinishReason = Literal["stop", "length", "tool_calls", "content_filter"] + + +@dataclass(frozen=True) +class CacheCreationDetails: + five_minute: Option[int] + one_hour: Option[int] + + +@dataclass(frozen=True) +class ResponseUsage: + """Provider-reported token counts, provider-neutral.""" + + input_tokens: int + output_tokens: int + cache_creation_input_tokens: int + cache_read_input_tokens: int + cache_creation: Option[CacheCreationDetails] + + +@dataclass(frozen=True) +class ChatResponse: + id: str + model: str + content: Block[ContentBlock] + finish: FinishReason + usage: ResponseUsage + synthesized_json_content: bool + """True when the provider rewrote a forced json_tool_call into plain + content (v1 then emits a bare message: no provider fields, no thinking).""" + + +# -------------------------------------------------------------------------- +# Stream IR: provider stream parsers map wire events onto these; the inbound +# stream serializer folds them into outbound chunks. One event per provider +# wire event that carries information; keep-alives map to nothing. +# -------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class StreamStart: + id: str + model: str + usage: ResponseUsage + + +@dataclass(frozen=True) +class TextDelta: + index: int + text: str + + +@dataclass(frozen=True) +class ToolUseStart: + index: int + id: str + name: str + + +@dataclass(frozen=True) +class ToolArgsDelta: + index: int + partial_json: str + + +@dataclass(frozen=True) +class ThinkingDelta: + index: int + thinking: str + + +@dataclass(frozen=True) +class SignatureDelta: + index: int + signature: str + + +@dataclass(frozen=True) +class StreamFinish: + """Anthropic message_delta: the final stop reason plus output usage.""" + + finish: FinishReason + output_tokens: int + + +@tagged_union(frozen=True) +class StreamEvent: + tag: Literal[ + "start", + "text_delta", + "tool_use_start", + "tool_args_delta", + "thinking_delta", + "signature_delta", + "finish", + "stop", + ] = tag() + + start: StreamStart = case() + text_delta: TextDelta = case() + tool_use_start: ToolUseStart = case() + tool_args_delta: ToolArgsDelta = case() + thinking_delta: ThinkingDelta = case() + signature_delta: SignatureDelta = case() + finish: StreamFinish = case() + stop: Unit = case() + + @staticmethod + def of_start(value: StreamStart) -> StreamEvent: + return _stream_start(value) + + @staticmethod + def of_text_delta(value: TextDelta) -> StreamEvent: + return _stream_text_delta(value) + + @staticmethod + def of_tool_use_start(value: ToolUseStart) -> StreamEvent: + return _stream_tool_use_start(value) + + @staticmethod + def of_tool_args_delta(value: ToolArgsDelta) -> StreamEvent: + return _stream_tool_args_delta(value) + + @staticmethod + def of_thinking_delta(value: ThinkingDelta) -> StreamEvent: + return _stream_thinking_delta(value) + + @staticmethod + def of_signature_delta(value: SignatureDelta) -> StreamEvent: + return _stream_signature_delta(value) + + @staticmethod + def of_finish(value: StreamFinish) -> StreamEvent: + return _stream_finish(value) + + @staticmethod + def of_stop() -> StreamEvent: + return _stream_stop(UNIT) + + +_stream_start = _case_maker(StreamEvent, "start") +_stream_text_delta = _case_maker(StreamEvent, "text_delta") +_stream_tool_use_start = _case_maker(StreamEvent, "tool_use_start") +_stream_tool_args_delta = _case_maker(StreamEvent, "tool_args_delta") +_stream_thinking_delta = _case_maker(StreamEvent, "thinking_delta") +_stream_signature_delta = _case_maker(StreamEvent, "signature_delta") +_stream_finish = _case_maker(StreamEvent, "finish") +_stream_stop = _case_maker(StreamEvent, "stop") diff --git a/litellm/translation/providers/anthropic/response.py b/litellm/translation/providers/anthropic/response.py new file mode 100644 index 0000000000..98ae9a7447 --- /dev/null +++ b/litellm/translation/providers/anthropic/response.py @@ -0,0 +1,224 @@ +"""Anthropic ``/v1/messages`` response JSON -> IR ``ChatResponse``. + +The envelope is lenient (unknown metadata keys are ignored, exactly like v1) +but content blocks are fail-closed: a block type the v2 request surface +cannot trigger (server tools, citations, compaction) is a loud error value, +never silently dropped. The json_tool_call rewrite (structured outputs on the +json-tool strategy) happens here because the tool is an anthropic-side +artifact the caller never asked for by name. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from types import MappingProxyType + +from expression import Error, Nothing, Ok, Option, Result, Some +from expression.collections import Block + +from ...errors import BoundaryError, TranslationError +from ...ir import ( + CacheCreationDetails, + ChatRequest, + ChatResponse, + ContentBlock, + FinishReason, + JsonBlob, + PlainJson, + RedactedThinking, + ResponseUsage, + Text, + Thinking, + ToolUse, +) +from . import params as p +from .tools import request_name_maps + +_ParseResult = Result[ChatResponse, TranslationError] + +# v1 map_finish_reason, anthropic rows; unknown reasons default to "stop" +# with a warning in v1, so the same default applies here. +FINISH_MAP: Mapping[str, FinishReason] = MappingProxyType( + { + "stop_sequence": "stop", + "end_turn": "stop", + "max_tokens": "length", + "tool_use": "tool_calls", + "refusal": "content_filter", + "compaction": "length", + } +) + + +def parse_response(raw: PlainJson, request: ChatRequest) -> _ParseResult: + if not isinstance(raw, dict): + return Error(_boundary("response body is not a JSON object")) + if "error" in raw: + return Error(_boundary(f"provider error payload: {raw['error']!r}")) + content = raw.get("content") + if not isinstance(content, list): + return Error(_boundary("response 'content' is not an array")) + blocks: list[ContentBlock] = [] + for block in content: + parsed = _parse_block(block, request) + if isinstance(parsed, TranslationError): + return Error(parsed) + blocks.append(parsed) # nosemgrep: translation-no-mutation + stop_reason = raw.get("stop_reason") + finish = ( + FINISH_MAP.get(stop_reason, "stop") if isinstance(stop_reason, str) else "stop" + ) + usage = parse_usage(raw.get("usage")) + if isinstance(usage, TranslationError): + return Error(usage) + model = raw.get("model") + response_id = raw.get("id") + response = ChatResponse( + id=response_id if isinstance(response_id, str) else "", + model=model if isinstance(model, str) else request.model, + content=Block.of_seq(blocks), + finish=finish, + usage=usage, + synthesized_json_content=False, + ) + return Ok(_resolve_json_tool(response, request)) + + +def _boundary(reason: str) -> TranslationError: + return TranslationError.of_boundary(BoundaryError.of(Block.of_seq([reason]))) + + +def _parse_block( + block: PlainJson, request: ChatRequest +) -> ContentBlock | TranslationError: + if not isinstance(block, dict): + return _boundary("response content block is not an object") + if block.get("citations") is not None: + return TranslationError.of_unsupported( + "response citations blocks; unreachable for v2-sent requests" + ) + kind = block.get("type") + if kind == "text": + text = block.get("text") + if not isinstance(text, str): + return _boundary("text block is missing 'text'") + return ContentBlock.of_text(Text(text=text, cache=Nothing)) + if kind == "tool_use": + return _parse_tool_use(block, request) + if kind == "thinking": + thinking = block.get("thinking") + signature = block.get("signature") + return ContentBlock.of_thinking( + Thinking( + thinking=thinking if isinstance(thinking, str) else "", + signature=Some(signature) if isinstance(signature, str) else Nothing, + cache=Nothing, + ) + ) + if kind == "redacted_thinking": + data = block.get("data") + return ContentBlock.of_redacted_thinking( + RedactedThinking(data=data if isinstance(data, str) else "") + ) + return TranslationError.of_unsupported( + f"response content block type {kind!r}; unreachable for v2-sent requests" + ) + + +def _parse_tool_use( + block: dict[str, PlainJson], request: ChatRequest +) -> ContentBlock | TranslationError: + if "caller" in block: + return TranslationError.of_unsupported( + "programmatic tool calling (caller); unreachable for v2-sent requests" + ) + identifier = block.get("id") + name = block.get("name") + if not isinstance(identifier, str) or not isinstance(name, str): + return _boundary("tool_use block is missing 'id'/'name'") + _, reverse = request_name_maps(request.tools) + return ContentBlock.of_tool_use( + ToolUse( + id=identifier, + name=reverse.get(name, name), + arguments=JsonBlob(value=block.get("input")), + cache=Nothing, + ) + ) + + +def parse_usage(raw: PlainJson) -> ResponseUsage | TranslationError: + if not isinstance(raw, dict): + return _boundary("response 'usage' is not an object") + creation = raw.get("cache_creation") + details: Option[CacheCreationDetails] = Nothing + if isinstance(creation, dict): + five = creation.get("ephemeral_5m_input_tokens") + hour = creation.get("ephemeral_1h_input_tokens") + details = Some( + CacheCreationDetails( + five_minute=Some(five) if isinstance(five, int) else Nothing, + one_hour=Some(hour) if isinstance(hour, int) else Nothing, + ) + ) + return ResponseUsage( + input_tokens=_int_of(raw.get("input_tokens")), + output_tokens=_int_of(raw.get("output_tokens")), + cache_creation_input_tokens=_int_of(raw.get("cache_creation_input_tokens")), + cache_read_input_tokens=_int_of(raw.get("cache_read_input_tokens")), + cache_creation=details, + ) + + +def _int_of(value: PlainJson) -> int: + """v1 tolerates explicit nulls and non-numerics in usage counts.""" + if isinstance(value, bool): + return 0 + if isinstance(value, (int, float)): + return int(value) + return 0 + + +def _uses_json_tool(request: ChatRequest) -> bool: + match request.response_format: + case Option(tag="some", some=response_format): + return response_format.tag == "json_schema" and not p.uses_output_format( + request.model + ) + case _: + return False + + +def _resolve_json_tool(response: ChatResponse, request: ChatRequest) -> ChatResponse: + """v1 ``_resolve_json_mode_non_streaming``: when structured outputs ride + the json_tool_call strategy, the forced tool call comes back as plain + content and the stop reason is rewritten to a clean stop. The mixed + user-tools case is unreachable (the request side refuses it).""" + if not _uses_json_tool(request): + return response + tool_uses = [ + block.tool_use for block in response.content if block.tag == "tool_use" + ] + json_calls = [tool for tool in tool_uses if tool.name == "json_tool_call"] + if not json_calls or len(json_calls) != len(tool_uses): + return response + arguments = json_calls[0].arguments.value + if arguments is None: + return response + payload: PlainJson = arguments + if isinstance(arguments, dict) and arguments.get("values") is not None: + payload = arguments["values"] + # v1 replaces the whole message with the JSON content (any text or + # thinking blocks in the raw response are discarded with it). + kept = Block.of_seq( + [ContentBlock.of_text(Text(text=json.dumps(payload), cache=Nothing))] + ) + return ChatResponse( + id=response.id, + model=response.model, + content=kept, + finish="stop", + usage=response.usage, + synthesized_json_content=True, + ) diff --git a/litellm/translation/providers/anthropic/stream.py b/litellm/translation/providers/anthropic/stream.py new file mode 100644 index 0000000000..97ea48030c --- /dev/null +++ b/litellm/translation/providers/anthropic/stream.py @@ -0,0 +1,200 @@ +"""Anthropic SSE stream -> IR stream events. + +One SSE ``data:`` payload maps to at most one ``StreamEvent``; keep-alives +and block-stop bookkeeping map to none. Event types the v2 request surface +cannot trigger (server tools, citations) are loud error values. Tool names +are reverse-mapped through the per-request map, mirroring the non-streaming +path. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping + +from expression import Error, Ok, Result +from expression.collections import Block + +from ...errors import BoundaryError, TranslationError +from ...ir import ( + ChatRequest, + FinishReason, + PlainJson, + SignatureDelta, + StreamEvent, + StreamFinish, + StreamStart, + TextDelta, + ThinkingDelta, + ToolArgsDelta, + ToolUseStart, +) +from .response import FINISH_MAP, parse_usage +from .tools import request_name_maps + +_EventResult = Result[StreamEvent | None, TranslationError] + + +def reverse_names(request: ChatRequest) -> Mapping[str, str]: + _, reverse = request_name_maps(request.tools) + return reverse + + +def parse_sse_line(line: str, reverse: Mapping[str, str]) -> _EventResult: + """One raw SSE line -> at most one event. Non-data lines are framing.""" + stripped = line.strip() + if not stripped.startswith("data:"): + return Ok(None) + payload = stripped[len("data:") :].strip() + try: + event: PlainJson = json.loads(payload) + except ValueError: + return Error(_boundary(f"stream payload is not JSON: {payload[:120]!r}")) + return parse_event(event, reverse) + + +def parse_event(event: PlainJson, reverse: Mapping[str, str]) -> _EventResult: + if not isinstance(event, dict): + return Error(_boundary("stream event is not an object")) + kind = event.get("type") + if kind in ("ping", "content_block_stop"): + return Ok(None) + if kind == "message_start": + return _start_event(event) + if kind == "content_block_start": + return _block_start_event(event, reverse) + if kind == "content_block_delta": + return _delta_event(event) + if kind == "message_delta": + return _finish_event(event) + if kind == "message_stop": + return Ok(StreamEvent.of_stop()) + if kind == "error": + return Error(_boundary(f"provider stream error: {event.get('error')!r}")) + return Error( + TranslationError.of_unsupported( + f"stream event type {kind!r}; unreachable for v2-sent requests" + ) + ) + + +def _boundary(reason: str) -> TranslationError: + return TranslationError.of_boundary(BoundaryError.of(Block.of_seq([reason]))) + + +def _start_event(event: dict[str, PlainJson]) -> _EventResult: + message = event.get("message") + if not isinstance(message, dict): + return Error(_boundary("message_start is missing 'message'")) + usage = parse_usage(message.get("usage") or {}) + if isinstance(usage, TranslationError): + return Error(usage) + identifier = message.get("id") + model = message.get("model") + return Ok( + StreamEvent.of_start( + StreamStart( + id=identifier if isinstance(identifier, str) else "", + model=model if isinstance(model, str) else "", + usage=usage, + ) + ) + ) + + +def _block_start_event( + event: dict[str, PlainJson], reverse: Mapping[str, str] +) -> _EventResult: + block = event.get("content_block") + index = event.get("index") + if not isinstance(block, dict) or not isinstance(index, int): + return Error(_boundary("content_block_start is malformed")) + kind = block.get("type") + if kind == "text": + text = block.get("text") + if isinstance(text, str) and text: + return Ok(StreamEvent.of_text_delta(TextDelta(index=index, text=text))) + return Ok(None) + if kind == "thinking": + return Ok(None) # thinking starts empty; deltas carry the content + if kind == "tool_use": + identifier = block.get("id") + name = block.get("name") + if not isinstance(identifier, str) or not isinstance(name, str): + return Error(_boundary("tool_use block start is missing 'id'/'name'")) + return Ok( + StreamEvent.of_tool_use_start( + ToolUseStart(index=index, id=identifier, name=reverse.get(name, name)) + ) + ) + return Error( + TranslationError.of_unsupported( + f"stream content block type {kind!r}; unreachable for v2-sent requests" + ) + ) + + +def _delta_event(event: dict[str, PlainJson]) -> _EventResult: + delta = event.get("delta") + index = event.get("index") + if not isinstance(delta, dict) or not isinstance(index, int): + return Error(_boundary("content_block_delta is malformed")) + kind = delta.get("type") + if kind == "text_delta": + text = delta.get("text") + return Ok( + StreamEvent.of_text_delta( + TextDelta(index=index, text=text if isinstance(text, str) else "") + ) + ) + if kind == "input_json_delta": + partial = delta.get("partial_json") + return Ok( + StreamEvent.of_tool_args_delta( + ToolArgsDelta( + index=index, + partial_json=partial if isinstance(partial, str) else "", + ) + ) + ) + if kind == "thinking_delta": + thinking = delta.get("thinking") + return Ok( + StreamEvent.of_thinking_delta( + ThinkingDelta( + index=index, + thinking=thinking if isinstance(thinking, str) else "", + ) + ) + ) + if kind == "signature_delta": + signature = delta.get("signature") + return Ok( + StreamEvent.of_signature_delta( + SignatureDelta( + index=index, + signature=signature if isinstance(signature, str) else "", + ) + ) + ) + return Error( + TranslationError.of_unsupported( + f"stream delta type {kind!r}; unreachable for v2-sent requests" + ) + ) + + +def _finish_event(event: dict[str, PlainJson]) -> _EventResult: + delta = event.get("delta") + usage = event.get("usage") + stop_reason = delta.get("stop_reason") if isinstance(delta, dict) else None + finish: FinishReason = ( + FINISH_MAP.get(stop_reason, "stop") if isinstance(stop_reason, str) else "stop" + ) + output_tokens = 0 + if isinstance(usage, dict): + raw = usage.get("output_tokens") + output_tokens = int(raw) if isinstance(raw, (int, float)) else 0 + return Ok( + StreamEvent.of_finish(StreamFinish(finish=finish, output_tokens=output_tokens)) + ) diff --git a/litellm/translation_seam.py b/litellm/translation_seam.py new file mode 100644 index 0000000000..aaf03f8f4a --- /dev/null +++ b/litellm/translation_seam.py @@ -0,0 +1,356 @@ +"""The v1-side adapter for translation v2 (lives OUTSIDE litellm/translation). + +This module is the one place that speaks both languages: it reads litellm +ambient state (model map helpers, drop_params/modify_params, the per-provider +allowlist flag) into the package's injected ``TranslationDeps`` value, adapts +v2's plain response bodies onto ``ModelResponse``, and owns the +``completion()`` fork. The translation package never imports the v1 stack; +this seam imports both, in the allowed direction. + +Flag precedent (dossier section 7): ``litellm.translation_v2_providers`` is a +module global seeded from the ``LITELLM_TRANSLATION_V2_PROVIDERS`` +comma-separated env var, and the proxy's ``litellm_settings`` generic setattr +fallback makes it yaml-configurable with zero extra plumbing. Off by default. +""" + +from __future__ import annotations + +import json +from typing import Any, Dict, FrozenSet, Optional, cast + +import litellm +from litellm.llms.anthropic.common_utils import AnthropicModelInfo +from litellm.types.utils import ( + CacheCreationTokenDetails, + CompletionTokensDetailsWrapper, + ModelResponse, + PromptTokensDetailsWrapper, + Usage, +) + +from litellm.translation import TranslationDeps +from litellm.translation.ir import Body + + +def enabled_providers() -> FrozenSet[str]: + """The per-provider opt-in allowlist, read at call time so proxy yaml and + runtime changes apply without restart.""" + configured = getattr(litellm, "translation_v2_providers", None) or [] + return frozenset(str(entry).strip() for entry in configured if str(entry).strip()) + + +def _max_tokens_for_model(model: str) -> Optional[int]: + try: + return litellm.utils.get_max_tokens(model) + except Exception: + return None + + +def _count_response_tokens(text: str) -> int: + from litellm.utils import token_counter + + return token_counter(text=text, count_response_tokens=True) + + +def build_translation_deps( + request_drop_params: Optional[bool] = None, +) -> TranslationDeps: + drop_params_global = litellm.drop_params is True + return TranslationDeps( + max_tokens_for_model=_max_tokens_for_model, + supports_capability=AnthropicModelInfo._supports_model_capability, + capability_flag=AnthropicModelInfo._get_model_capability, + count_response_tokens=_count_response_tokens, + drop_params=drop_params_global or request_drop_params is True, + drop_params_global=drop_params_global, + modify_params=litellm.modify_params is True, + ) + + +def to_model_response( + body: Body, model_response: Optional[ModelResponse] = None +) -> ModelResponse: + """Adapt a v2 response body onto litellm's ModelResponse envelope. + + Mirrors v1's ``transform_parsed_response`` assembly exactly (assign into + the pre-allocated response's first choice, stamp created/model, setattr a + real ``Usage``), so the serialized shape is identical to v1's. The + envelope (chatcmpl id, created timestamp) stays litellm-ambient. + """ + import time + + from litellm.types.utils import Message + + response = model_response if model_response is not None else ModelResponse() + choices = body.get("choices") + first = choices[0] if isinstance(choices, list) and choices else {} + message_payload = first.get("message") if isinstance(first, dict) else {} + if isinstance(message_payload, dict): + response.choices[0].message = Message(**cast(Dict[str, Any], message_payload)) + finish = first.get("finish_reason") if isinstance(first, dict) else None + response.choices[0].finish_reason = cast( + Any, finish if isinstance(finish, str) else "stop" + ) + usage_payload = body.get("usage") + usage = _build_usage(usage_payload) if isinstance(usage_payload, dict) else Usage() + setattr(response, "usage", usage) + response.created = int(time.time()) + model = body.get("model") + if isinstance(model, str): + response.model = model + return response + + +def _build_usage(payload: dict) -> Usage: + """Construct Usage with the exact kwarg set v1's calculate_usage passes, + because nested (extra-attribute) serialization only includes explicitly + set fields; a kwargs-splat would drop the always-set None fields v1 emits + (server_tool_use, inference_geo, speed, wrapper audio/image/video).""" + prompt_details = payload.get("prompt_tokens_details") or {} + completion_details = payload.get("completion_tokens_details") or {} + creation_details = prompt_details.get("cache_creation_token_details") + wrapper_details = ( + CacheCreationTokenDetails( + ephemeral_5m_input_tokens=creation_details.get("ephemeral_5m_input_tokens"), + ephemeral_1h_input_tokens=creation_details.get("ephemeral_1h_input_tokens"), + ) + if isinstance(creation_details, dict) + else None + ) + return Usage( + prompt_tokens=payload.get("prompt_tokens"), + completion_tokens=payload.get("completion_tokens"), + total_tokens=payload.get("total_tokens"), + prompt_tokens_details=PromptTokensDetailsWrapper( + cached_tokens=prompt_details.get("cached_tokens"), + cache_creation_tokens=prompt_details.get("cache_creation_tokens"), + cache_creation_token_details=wrapper_details, + text_tokens=prompt_details.get("text_tokens"), + ), + cache_creation_input_tokens=payload.get("cache_creation_input_tokens"), + cache_read_input_tokens=payload.get("cache_read_input_tokens"), + completion_tokens_details=CompletionTokensDetailsWrapper( + reasoning_tokens=completion_details.get("reasoning_tokens"), + text_tokens=completion_details.get("text_tokens"), + ), + server_tool_use=None, + inference_geo=None, + speed=None, + ) + + +def to_model_response_stream(body: Body, stream_id: str): + """Adapt one v2 chunk body onto ModelResponseStream; the stream id is + minted once per stream by the caller (v1 reuses one id for every chunk). + system_fingerprint/citations are set explicitly because v1's wrapper + always sets them and extra-field serialization only dumps set fields.""" + from litellm.types.utils import ModelResponseStream + + chunk = ModelResponseStream( + id=stream_id, system_fingerprint=None, **cast(Dict[str, Any], body) + ) + choices = body.get("choices") + first = choices[0] if isinstance(choices, list) and choices else {} + if isinstance(first, dict) and first.get("finish_reason") is None: + # v1 sets citations only on content chunks, never the finish chunk. + setattr(chunk, "citations", None) + return chunk + + +class HttpxJsonPort: + """HttpPort over a fresh httpx.AsyncClient per request. + + A fresh client avoids binding litellm's cached async clients to the + short-lived event loop the sync wrapper creates. Pooled-client reuse on + the async path is a follow-up optimization; the flag-gated M3 traffic + does not need it. + """ + + async def post_json(self, endpoint, body): # noqa: ANN001, ANN201 + import httpx + + from litellm.translation.engine.http import HttpResponse + + async with httpx.AsyncClient(timeout=endpoint.timeout_seconds) as client: + raw = await client.post( + endpoint.url, headers=dict(endpoint.headers), json=dict(body) + ) + try: + payload = raw.json() + except ValueError: + payload = None + return HttpResponse( + status_code=raw.status_code, + body=payload, + text=raw.text, + headers=dict(raw.headers), + ) + + +def try_completion_v2( + *, + model: str, + messages: list, + optional_param_args: dict, + non_default_params: dict, + api_key: Optional[str], + api_base: Optional[str], + timeout: Optional[float], + stream: Optional[bool], + acompletion: Optional[bool], + logging_obj, + model_response: ModelResponse, + request_drop_params: Optional[bool], +): + """The completion() fork for translation v2 (anthropic, non-streaming). + + Returns None to stay on v1: flag off, streaming (the v2 stream seam is a + follow-up), modify_params behaviors, or any request shape outside v2's + proven surface (the fail-closed translate). Once the request is sent, + failures raise the provider error contract exactly like v1; there is no + silent re-send. + """ + from litellm.translation import route + from litellm.translation.engine.pipeline import prepare_chat_request + + if "anthropic" not in enabled_providers(): + return None + if stream is True or litellm.modify_params is True: + return None + decision = route( + schema="openai_chat", + provider="anthropic", + enabled_providers=frozenset({"anthropic"}), # checked above + body_touching=False, + ) + if decision.tag != "v2": + return None + raw_body = _raw_openai_body( + model, messages, optional_param_args, non_default_params + ) + deps = build_translation_deps(request_drop_params=request_drop_params) + prepared_result = prepare_chat_request(raw_body, "anthropic", deps) + if prepared_result.is_error(): + litellm.verbose_logger.debug( + "translation v2 fallback to v1: %s", prepared_result.error.summary + ) + return None + prepared = prepared_result.ok + coroutine = _send_v2( + prepared=prepared, + deps=deps, + model=model, + messages=messages, + api_key=api_key, + api_base=api_base, + timeout=timeout, + logging_obj=logging_obj, + model_response=model_response, + ) + if acompletion is True: + return coroutine + import asyncio + + return asyncio.run(coroutine) + + +_BODY_FIELDS = ( + "temperature", + "top_p", + "max_tokens", + "max_completion_tokens", + "stop", + "stream", + "tools", + "tool_choice", + "parallel_tool_calls", + "response_format", + "user", + "reasoning_effort", + "thinking", +) + + +def _raw_openai_body( + model: str, messages: list, optional_param_args: dict, non_default_params: dict +) -> dict: + """Rebuild the caller's OpenAI-shape body from completion()'s pre-mapping + locals. Provider-specific extras (non_default_params, e.g. top_k) ride + along verbatim: anything v2 does not account for becomes a typed + unsupported error and falls back to v1 -- the fail-closed allowlist.""" + named = { + key: optional_param_args.get(key) + for key in _BODY_FIELDS + if optional_param_args.get(key) is not None + } + return {"model": model, "messages": messages, **named, **non_default_params} + + +async def _send_v2( + *, + prepared, + deps: TranslationDeps, + model: str, + messages: list, + api_key: Optional[str], + api_base: Optional[str], + timeout: Optional[float], + logging_obj, + model_response: ModelResponse, +) -> ModelResponse: + from litellm.llms.anthropic.common_utils import ( + AnthropicError, + process_anthropic_headers, + ) + from litellm.llms.anthropic.chat.transformation import AnthropicConfig + + from litellm.translation.engine.http import Endpoint + from litellm.translation.engine.pipeline import send_prepared, wire_body + + config = AnthropicConfig() + headers = config.validate_environment( + headers={}, + model=model, + messages=messages, + optional_params=dict(prepared.body), + litellm_params={}, + api_key=api_key, + api_base=api_base, + ) + endpoint = Endpoint( + url=api_base or "https://api.anthropic.com/v1/messages", + headers=headers, + timeout_seconds=float(timeout) if timeout else 600.0, + ) + logging_obj.pre_call( + input=messages, + api_key=api_key, + additional_args={ + "complete_input_dict": wire_body(prepared), + "api_base": endpoint.url, + "headers": dict(headers), + }, + ) + result = await send_prepared(prepared, "anthropic", deps, HttpxJsonPort(), endpoint) + if result.is_error(): + error = result.error + if error.tag == "provider_http": + logging_obj.post_call( + input=messages, + api_key=api_key, + original_response=error.provider_http.text, + ) + raise AnthropicError( + status_code=error.provider_http.status_code, + message=error.provider_http.text, + ) + raise AnthropicError(status_code=500, message=error.summary) + body = result.ok + logging_obj.post_call( + input=messages, + api_key=api_key, + original_response=json.dumps(body, default=str), + ) + response = to_model_response(body, model_response) + response._hidden_params["additional_headers"] = process_anthropic_headers({}) + return response diff --git a/tests/test_litellm/translation/DIFFERENTIAL_REPORT.md b/tests/test_litellm/translation/DIFFERENTIAL_REPORT.md new file mode 100644 index 0000000000..3030180b4b --- /dev/null +++ b/tests/test_litellm/translation/DIFFERENTIAL_REPORT.md @@ -0,0 +1,77 @@ +# Translation v2 differential report (anthropic) + +v1 and v2 run over the same corpus; every row must be IDENTICAL for +the anthropic flag to turn on. Regenerate with: +`python -m tests.test_litellm.translation.generate_differential_report` + +- commit: 23724c4392 + +## Request bodies (v1 map_openai_params + transform_request vs v2) + +- IDENTICAL: assistant_text_and_tool_call +- IDENTICAL: cache_control_everywhere +- IDENTICAL: cache_control_on_string_message +- IDENTICAL: cache_control_on_tool_result +- IDENTICAL: current_model_default_max_tokens +- IDENTICAL: duplicate_tool_call_ids_dedupe +- IDENTICAL: empty_user_content_placeholder +- IDENTICAL: final_assistant_text_rstripped +- IDENTICAL: https_image_url +- IDENTICAL: image_data_uri +- IDENTICAL: image_format_override +- IDENTICAL: max_completion_tokens +- IDENTICAL: multiturn_stop_stream +- IDENTICAL: no_max_tokens_legacy_model +- IDENTICAL: parallel_tool_calls_false +- IDENTICAL: parallel_tool_results_merge +- IDENTICAL: parallel_with_string_none_choice +- IDENTICAL: parallel_with_tool_choice +- IDENTICAL: pydantic_style_schema_filtered +- IDENTICAL: reasoning_effort_high_no_max_tokens +- IDENTICAL: reasoning_effort_low +- IDENTICAL: reasoning_effort_none +- IDENTICAL: response_format_json_object_current +- IDENTICAL: response_format_json_object_legacy_noop +- IDENTICAL: response_format_json_schema_current +- IDENTICAL: response_format_json_schema_legacy_tool +- IDENTICAL: response_format_with_thinking_no_forced_choice +- IDENTICAL: stop_as_string +- IDENTICAL: stop_whitespace_kept_without_drop_params +- IDENTICAL: system_and_sampling +- IDENTICAL: system_as_array +- IDENTICAL: temperature_int_stays_int +- IDENTICAL: text +- IDENTICAL: thinking_explicit +- IDENTICAL: thinking_history_blocks +- IDENTICAL: thinking_no_max_tokens_bumps +- IDENTICAL: tool_call_roundtrip +- IDENTICAL: tool_choice_dict_forms +- IDENTICAL: tool_choice_none +- IDENTICAL: tool_choice_required +- IDENTICAL: tool_choice_specific +- IDENTICAL: tool_name_sanitization_with_history +- IDENTICAL: tool_result_parts_not_placeholdered +- IDENTICAL: tool_schema_missing_parameters +- IDENTICAL: tool_schema_type_coerced +- IDENTICAL: tool_use_id_sanitized +- IDENTICAL: tool_without_description +- IDENTICAL: tools_auto +- IDENTICAL: top_k +- IDENTICAL: user_email_skipped +- IDENTICAL: user_metadata +- IDENTICAL: whitespace_text_part_placeholder + +## Responses (v1 transform_response vs v2) + +- IDENTICAL: json_tool +- IDENTICAL: text +- IDENTICAL: thinking +- IDENTICAL: tools + +## Streams (v1 CustomStreamWrapper replay vs v2 engine/stream) + +- IDENTICAL: text +- IDENTICAL: thinking +- IDENTICAL: tools + +Result: 0 divergent rows. Shapes outside the corpus fall back to v1 (fail-closed), so this table is the complete flag-on surface. diff --git a/tests/test_litellm/translation/conftest.py b/tests/test_litellm/translation/conftest.py index 524add7cc5..683fe79b0f 100644 --- a/tests/test_litellm/translation/conftest.py +++ b/tests/test_litellm/translation/conftest.py @@ -13,7 +13,7 @@ import pytest os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True") from litellm.llms.anthropic.common_utils import AnthropicModelInfo # noqa: E402 -from litellm.utils import get_max_tokens # noqa: E402 +from litellm.utils import get_max_tokens, token_counter # noqa: E402 from litellm.translation import TranslationDeps # noqa: E402 @@ -34,6 +34,9 @@ def build_real_deps( max_tokens_for_model=_max_tokens_for_model, supports_capability=AnthropicModelInfo._supports_model_capability, capability_flag=AnthropicModelInfo._get_model_capability, + count_response_tokens=lambda text: token_counter( + text=text, count_response_tokens=True + ), drop_params=drop_params, drop_params_global=drop_params_global, modify_params=modify_params, diff --git a/tests/test_litellm/translation/generate_differential_report.py b/tests/test_litellm/translation/generate_differential_report.py new file mode 100644 index 0000000000..72d5d69991 --- /dev/null +++ b/tests/test_litellm/translation/generate_differential_report.py @@ -0,0 +1,83 @@ +"""Regenerate DIFFERENTIAL_REPORT.md: the v1-vs-v2 parity merge artifact. + +Run: python -m tests.test_litellm.translation.generate_differential_report +""" + +import json +import pathlib +import subprocess +import sys +import time +import uuid + +_HERE = pathlib.Path(__file__).parent + + +def main() -> None: + from . import test_differential_anthropic_request as req + from . import test_differential_anthropic_response as resp + from . import test_differential_anthropic_stream as stream + + counter = iter(range(1, 1_000_000)) + uuid.uuid4 = lambda: uuid.UUID(int=next(counter)) # type: ignore[assignment] + time.time = lambda: resp.FROZEN_TIME # type: ignore[assignment] + + lines = [ + "# Translation v2 differential report (anthropic)", + "", + "v1 and v2 run over the same corpus; every row must be IDENTICAL for", + "the anthropic flag to turn on. Regenerate with:", + "`python -m tests.test_litellm.translation.generate_differential_report`", + "", + f"- commit: {_git_sha()}", + "", + "## Request bodies (v1 map_openai_params + transform_request vs v2)", + "", + ] + failures = 0 + for name in sorted(req.CORPUS): + same = req._norm(req._v2_body(req.CORPUS[name])) == req._norm( + req._v1_body(req.CORPUS[name]) + ) + failures += 0 if same else 1 + lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}") + lines += ["", "## Responses (v1 transform_response vs v2)", ""] + for name in sorted(resp._REQUESTS): + same = resp._norm(resp._v2_model_response(name)) == resp._norm( + resp._v1_model_response(name) + ) + failures += 0 if same else 1 + lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}") + lines += ["", "## Streams (v1 CustomStreamWrapper replay vs v2 engine/stream)", ""] + for name in sorted(stream.STREAMS): + same = stream._norm(stream._v2_chunks(stream.STREAMS[name])) == stream._norm( + stream._v1_chunks(stream.STREAMS[name]) + ) + failures += 0 if same else 1 + lines.append(f"- {'IDENTICAL' if same else 'DIVERGENT'}: {name}") + lines += [ + "", + f"Result: {failures} divergent rows." + " Shapes outside the corpus fall back to v1 (fail-closed), so this" + " table is the complete flag-on surface.", + "", + ] + (_HERE / "DIFFERENTIAL_REPORT.md").write_text("\n".join(lines)) + print("\n".join(lines)) + sys.exit(1 if failures else 0) + + +def _git_sha() -> str: + try: + return subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + except Exception: + return "unknown" + + +if __name__ == "__main__": + main() diff --git a/tests/test_litellm/translation/test_differential_anthropic_response.py b/tests/test_litellm/translation/test_differential_anthropic_response.py new file mode 100644 index 0000000000..0ecf3206f7 --- /dev/null +++ b/tests/test_litellm/translation/test_differential_anthropic_response.py @@ -0,0 +1,238 @@ +"""Differential parity for the response path: v1 transform_response vs v2. + +Each case is a full cycle: the OpenAI request goes through both request +translators (so per-request state like the tool-name reverse map flows the +same way), then a recorded anthropic response JSON goes through v1's +``transform_response`` and v2's ``parse_response`` -> ``serialize_response`` +-> ``ModelResponse(**body)``, and the two ``ModelResponse`` dumps must be +identical. uuid/time are frozen because both sides mint ambient ids. +""" + +import copy +import json + +import httpx +import pytest + +import litellm +from litellm.llms.anthropic.chat.transformation import AnthropicConfig +from litellm.types.utils import ModelResponse + +from litellm.translation_seam import build_translation_deps, to_model_response + +from litellm.translation.inbound.openai_chat import parse_request +from litellm.translation.inbound.openai_chat.response import serialize_response +from litellm.translation.providers.anthropic.response import parse_response + +MODEL = "claude-sonnet-4-5" + +_REQUESTS = { + "text": { + "model": MODEL, + "max_tokens": 64, + "messages": [{"role": "user", "content": "hi"}], + }, + "tools": { + "model": MODEL, + "max_tokens": 64, + "tools": [ + { + "type": "function", + "function": { + "name": "mcp.server/get_weather", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + "messages": [{"role": "user", "content": "weather in Paris"}], + }, + "thinking": { + "model": "claude-3-7-sonnet-20250219", + "max_tokens": 2048, + "thinking": {"type": "enabled", "budget_tokens": 1024}, + "messages": [{"role": "user", "content": "think"}], + }, + "json_tool": { + "model": "claude-3-5-haiku-20241022", + "max_tokens": 256, + "messages": [{"role": "user", "content": "capital of France?"}], + "response_format": { + "type": "json_schema", + "json_schema": { + "name": "answer", + "schema": { + "type": "object", + "properties": {"capital": {"type": "string"}}, + }, + }, + }, + }, +} + +_RESPONSES = { + "text": { + "id": "msg_01XYZ", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5", + "content": [{"type": "text", "text": "Hello there."}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 12, + "output_tokens": 6, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + "cache_creation": { + "ephemeral_5m_input_tokens": 0, + "ephemeral_1h_input_tokens": 0, + }, + "service_tier": "standard", + }, + }, + "tools": { + "id": "msg_01ABC", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5", + "content": [ + {"type": "text", "text": "Checking."}, + { + "type": "tool_use", + "id": "toolu_01", + "name": "mcp_server_get_weather", + "input": {"city": "Paris"}, + }, + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": { + "input_tokens": 30, + "output_tokens": 20, + "cache_creation_input_tokens": 4, + "cache_read_input_tokens": 8, + "cache_creation": { + "ephemeral_5m_input_tokens": 4, + "ephemeral_1h_input_tokens": 0, + }, + }, + }, + "thinking": { + "id": "msg_01THINK", + "type": "message", + "role": "assistant", + "model": "claude-3-7-sonnet-20250219", + "content": [ + { + "type": "thinking", + "thinking": "The capital of France is Paris.", + "signature": "sig==", + }, + {"type": "text", "text": "Paris."}, + ], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 40, "output_tokens": 30}, + }, + "json_tool": { + "id": "msg_01JSON", + "type": "message", + "role": "assistant", + "model": "claude-3-5-haiku-20241022", + "content": [ + { + "type": "tool_use", + "id": "toolu_02", + "name": "json_tool_call", + "input": {"capital": "Paris"}, + } + ], + "stop_reason": "tool_use", + "stop_sequence": None, + "usage": {"input_tokens": 25, "output_tokens": 12}, + }, +} + +FROZEN_TIME = 1718064000.0 + + +@pytest.fixture(autouse=True) +def _deterministic_ambient(monkeypatch): + import time + import uuid + + counter = iter(range(1, 10_000)) + monkeypatch.setattr( + uuid, + "uuid4", + lambda: uuid.UUID(int=next(counter)), + ) + monkeypatch.setattr(time, "time", lambda: FROZEN_TIME) + + +def _v1_model_response(name: str) -> dict: + request = copy.deepcopy(_REQUESTS[name]) + config = AnthropicConfig() + model = request["model"] + params = { + key: value for key, value in request.items() if key not in ("model", "messages") + } + optional = config.map_openai_params( + copy.deepcopy(params), {}, model, drop_params=False + ) + litellm_params: dict = {} + config.transform_request( + model, copy.deepcopy(request["messages"]), optional, litellm_params, {} + ) + json_mode = optional.pop("json_mode", False) + raw = httpx.Response( + status_code=200, + json=_RESPONSES[name], + request=httpx.Request("POST", "https://api.anthropic.com/v1/messages"), + ) + logging = litellm.litellm_core_utils.litellm_logging.Logging( + model=model, + messages=copy.deepcopy(request["messages"]), + stream=False, + call_type="completion", + start_time=None, + litellm_call_id="char-test", + function_id="char-test", + ) + result = config.transform_response( + model=model, + raw_response=raw, + model_response=ModelResponse(), + logging_obj=logging, + request_data={}, + messages=copy.deepcopy(request["messages"]), + optional_params=optional, + litellm_params=litellm_params, + encoding=None, + json_mode=json_mode, + ) + return result.model_dump() + + +def _v2_model_response(name: str) -> dict: + request = copy.deepcopy(_REQUESTS[name]) + deps = build_translation_deps() + parsed = parse_request(request) + assert parsed.is_ok(), parsed.error.summary + response = parse_response(copy.deepcopy(_RESPONSES[name]), parsed.ok) + assert response.is_ok(), response.error.summary + body = serialize_response(response.ok, deps) + return to_model_response(body).model_dump() + + +def _norm(payload: dict) -> str: + # The chatcmpl id is ambient (each side mints its own from the frozen + # uuid counter in a different order); everything else must match. + return json.dumps({**payload, "id": "chatcmpl-X"}, sort_keys=True, default=str) + + +@pytest.mark.parametrize("name", sorted(_REQUESTS)) +def test_v2_response_matches_v1(name: str) -> None: + v1 = _v1_model_response(name) + v2 = _v2_model_response(name) + assert _norm(v2) == _norm(v1) diff --git a/tests/test_litellm/translation/test_differential_anthropic_stream.py b/tests/test_litellm/translation/test_differential_anthropic_stream.py new file mode 100644 index 0000000000..3f5b408db0 --- /dev/null +++ b/tests/test_litellm/translation/test_differential_anthropic_stream.py @@ -0,0 +1,245 @@ +"""Differential parity for streaming: v1's full CustomStreamWrapper replay vs +v2's engine/stream fold over the same recorded SSE lines. + +The replay method mirrors the characterization corpus +(tests/translation_characterization/_seams.py): raw SSE lines through the +real ``ModelResponseIterator`` inside ``CustomStreamWrapper`` on the v1 side; +``engine.stream.fold_lines`` plus the ``ModelResponseStream`` seam adapter on +the v2 side. uuid/time are frozen; stream ids are ambient and normalized. +""" + +import json + +import pytest + +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper + +from litellm.translation.engine.stream import fold_lines +from litellm.translation.inbound.openai_chat import parse_request +from litellm.translation.providers.anthropic.stream import parse_sse_line, reverse_names +from litellm.translation_seam import to_model_response_stream + +MODEL = "claude-sonnet-4-5" + +FROZEN_TIME = 1718064000.0 + + +def _sse(events: list) -> list: + lines = [] + for event in events: + lines.append(f"event: {event['type']}") + lines.append("data: " + json.dumps(event)) + lines.append("") + return lines + + +_MESSAGE_START = { + "type": "message_start", + "message": { + "id": "msg_stream_01", + "type": "message", + "role": "assistant", + "model": MODEL, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 12, "output_tokens": 1}, + }, +} + +STREAMS = { + "text": _sse( + [ + _MESSAGE_START, + {"type": "ping"}, + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "Paris is the"}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": " capital."}, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 9}, + }, + {"type": "message_stop"}, + ] + ), + "tools": _sse( + [ + _MESSAGE_START, + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "Checking."}, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "tool_use", + "id": "toolu_s1", + "name": "mcp_server_get_weather", + "input": {}, + }, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "input_json_delta", "partial_json": '{"ci'}, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "input_json_delta", "partial_json": 'ty": "Paris"}'}, + }, + {"type": "content_block_stop", "index": 1}, + { + "type": "message_delta", + "delta": {"stop_reason": "tool_use", "stop_sequence": None}, + "usage": {"output_tokens": 30}, + }, + {"type": "message_stop"}, + ] + ), + "thinking": _sse( + [ + _MESSAGE_START, + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "thinking_delta", "thinking": "France. Capital."}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": "sig=="}, + }, + {"type": "content_block_stop", "index": 0}, + { + "type": "content_block_start", + "index": 1, + "content_block": {"type": "text", "text": ""}, + }, + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "text_delta", "text": "Paris."}, + }, + {"type": "content_block_stop", "index": 1}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 20}, + }, + {"type": "message_stop"}, + ] + ), +} + +_REQUEST = { + "model": MODEL, + "max_tokens": 64, + "tools": [ + { + "type": "function", + "function": { + "name": "mcp.server/get_weather", + "parameters": {"type": "object", "properties": {}}, + }, + } + ], + "messages": [{"role": "user", "content": "weather in Paris"}], +} + + +@pytest.fixture(autouse=True) +def _deterministic_ambient(monkeypatch): + import time + import uuid + + counter = iter(range(1, 10_000)) + monkeypatch.setattr(uuid, "uuid4", lambda: uuid.UUID(int=next(counter))) + monkeypatch.setattr(time, "time", lambda: FROZEN_TIME) + + +def _v1_chunks(lines: list) -> list: + from litellm.llms.anthropic.chat.handler import ModelResponseIterator + from litellm.llms.anthropic.chat.transformation import AnthropicConfig + + # Production passes the per-request tool-name reverse map into the + # iterator (built by transform_request); replicate that wiring. + config = AnthropicConfig() + params = {k: v for k, v in _REQUEST.items() if k not in ("model", "messages")} + optional = config.map_openai_params(dict(params), {}, MODEL, drop_params=False) + litellm_params: dict = {} + config.transform_request(MODEL, [dict(m) for m in _REQUEST["messages"]], optional, litellm_params, {}) + reverse_map = litellm_params.get("_anthropic_tool_name_map") or {} + + iterator = ModelResponseIterator( + streaming_response=iter(lines), + sync_stream=True, + tool_name_reverse_map=reverse_map, + ) + wrapper = CustomStreamWrapper( + completion_stream=iterator, + model=MODEL, + custom_llm_provider="anthropic", + logging_obj=Logging( + model=MODEL, + messages=[{"role": "user", "content": "stream"}], + stream=True, + call_type="completion", + start_time=None, + litellm_call_id="diff-stream", + function_id="diff-stream", + ), + ) + return [chunk.model_dump() for chunk in wrapper] + + +def _v2_chunks(lines: list) -> list: + parsed = parse_request(dict(_REQUEST)) + assert parsed.is_ok(), parsed.error.summary + reverse = reverse_names(parsed.ok) + folded = fold_lines(lines, lambda line: parse_sse_line(line, reverse)) + assert folded.is_ok(), folded.error.summary + return [ + to_model_response_stream(chunk, "chatcmpl-X").model_dump() + for chunk in folded.ok + ] + + +def _norm(chunks: list) -> str: + return json.dumps( + [{**chunk, "id": "chatcmpl-X"} for chunk in chunks], sort_keys=True, default=str + ) + + +@pytest.mark.parametrize("name", sorted(STREAMS)) +def test_v2_stream_matches_v1(name: str) -> None: + lines = STREAMS[name] + assert _norm(_v2_chunks(lines)) == _norm(_v1_chunks(lines)) diff --git a/tests/test_litellm/translation/test_seam.py b/tests/test_litellm/translation/test_seam.py new file mode 100644 index 0000000000..f438e95d14 --- /dev/null +++ b/tests/test_litellm/translation/test_seam.py @@ -0,0 +1,139 @@ +"""The completion() fork: flag-gated v2 execution with fail-closed fallback. + +Uses respx to intercept the anthropic endpoint, so these run the REAL +pipeline (translate -> http port -> response translate -> ModelResponse) +end to end without network. +""" + +import json + +import pytest +import respx +from httpx import Response + +import litellm + +_URL = "https://api.anthropic.com/v1/messages" + +_PROVIDER_RESPONSE = { + "id": "msg_seam_01", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5", + "content": [{"type": "text", "text": "Hello from v2."}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 10, "output_tokens": 5}, +} + + +@pytest.fixture(autouse=True) +def _flag(monkeypatch): + monkeypatch.setattr(litellm, "translation_v2_providers", ["anthropic"]) + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-test") + + +@respx.mock +def test_flag_on_serves_through_v2() -> None: + route = respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE)) + response = litellm.completion( + model="anthropic/claude-sonnet-4-5", + max_tokens=32, + messages=[{"role": "user", "content": "hi"}], + ) + assert route.called + sent = json.loads(route.calls.last.request.content) + assert sent["model"] == "claude-sonnet-4-5" + assert "json_mode" not in sent + assert response.choices[0].message.content == "Hello from v2." + assert response.usage.prompt_tokens == 10 + assert response.usage.completion_tokens == 5 + + +@respx.mock +def test_unsupported_shape_falls_back_to_v1() -> None: + route = respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE)) + response = litellm.completion( + model="anthropic/claude-sonnet-4-5", + max_tokens=32, + messages=[ + {"role": "user", "content": "hi"}, + # assistant prefix is a v1 feature outside v2's surface: the + # request must be served (by v1), never rejected or dropped. + {"role": "assistant", "content": "Hel", "prefix": True}, + ], + ) + assert route.called + assert response.choices[0].message.content == "Hello from v2." + + +@respx.mock +def test_provider_error_raises_contract_not_fallback() -> None: + route = respx.post(_URL).mock( + return_value=Response(429, json={"error": {"message": "rate limited"}}) + ) + with pytest.raises(litellm.exceptions.RateLimitError): + litellm.completion( + model="anthropic/claude-sonnet-4-5", + max_tokens=32, + messages=[{"role": "user", "content": "hi"}], + num_retries=0, + ) + assert route.call_count == 1 # sent once: no silent re-send through v1 + + +@pytest.mark.asyncio +@respx.mock +async def test_acompletion_path_returns_awaitable() -> None: + respx.post(_URL).mock(return_value=Response(200, json=_PROVIDER_RESPONSE)) + response = await litellm.acompletion( + model="anthropic/claude-sonnet-4-5", + max_tokens=32, + messages=[{"role": "user", "content": "hi"}], + ) + assert response.choices[0].message.content == "Hello from v2." + + +def test_flag_off_returns_none_from_seam() -> None: + from litellm.translation_seam import try_completion_v2 + + litellm.translation_v2_providers = [] + assert ( + try_completion_v2( + model="claude-sonnet-4-5", + messages=[{"role": "user", "content": "hi"}], + optional_param_args={}, + non_default_params={}, + api_key=None, + api_base=None, + timeout=None, + stream=None, + acompletion=None, + logging_obj=None, + model_response=None, + request_drop_params=None, + ) + is None + ) + + +def test_streaming_stays_on_v1() -> None: + from litellm.translation_seam import try_completion_v2 + + assert ( + try_completion_v2( + model="claude-sonnet-4-5", + messages=[{"role": "user", "content": "hi"}], + optional_param_args={}, + non_default_params={}, + api_key=None, + api_base=None, + timeout=None, + stream=True, + acompletion=None, + logging_obj=None, + model_response=None, + request_drop_params=None, + ) + is None + )