mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
fix(router): support mid-stream fallback for anthropic_messages route type (#38153)
anthropic_messages goes through _ageneric_api_call_with_fallbacks rather than _acompletion, so its returned streaming iterator was never wrapped by the chat-completions fallback handler. A retriable SSE event: error frame (overloaded_error, internal_server_error) from a native Anthropic/Bedrock passthrough passed through to the client unchanged, and a MidStreamFallbackError raised by the completion-bridge path's CustomStreamWrapper propagated unhandled. Add _aanthropic_messages_streaming_iterator, mirroring _acompletion_streaming_iterator: it detects a retriable SSE error event via the new parse_anthropic_error_event helper, raises MidStreamFallbackError once real generated content (a content_block_delta frame) has not yet reached the caller, and re-enters the Router's fallback chain. A MidStreamFallbackError raised directly by the source iterator (the completion-bridge path) is gated the same way via its own is_pre_first_chunk flag. The raised MidStreamFallbackError carries a status-coded original_exception built from the parsed error type, so status_code/cooldown logic sees the real 429/500/503/etc. instead of a hardcoded 503. Lifecycle/bookkeeping frames (message_start, content_block_start, ping, ...) never disqualify a fallback attempt by themselves, since Anthropic routinely sends message_start before an overload error - but they are buffered rather than forwarded immediately, since forwarding one and then appending a fallback attempt's own message_start would produce two overlapping message lifecycles on one SSE stream. Buffered frames flush, in order, once real content arrives or the stream ends without error. Once real content has streamed, or the error is a non-retriable 4xx, the chunk (or exception) is forwarded as-is rather than starting a second lifecycle. Content and error coalesced into a single physical read are handled the same way: once the client has genuinely received the content (bundled in that same forwarded chunk), no fallback is attempted. A `ping` keepalive is dropped outright before any real content arrives (it recurs indefinitely on a slow-starting connection and carries nothing worth buffering), and the pre-content lifecycle buffer is capped at MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS, forcing an early commit to the primary stream so a hostile or pathological upstream can't grow it without bound. is_anthropic_ping_chunk only matches a chunk whose every event: line is event: ping, so a ping coalesced with real content or a retriable error into one physical transport chunk is never dropped. The fallback request kwargs also deep-copy nested litellm_metadata/metadata (matching the Responses API path) so the primary attempt's deployment-specific fields never leak into the fallback request, and the fallback deployment's own provider headers are merged onto the wrapper's _hidden_params so they still reach the client/logging pipeline. A fallback that resolves to a non-streaming response (e.g. an agentic tool-use interception loop) is synthesized into a real Anthropic SSE event sequence via the new anthropic_messages_response_as_sse_events helper, instead of yielding a raw dict into the byte stream - including a trailing signature_delta for a thinking block, and a message_start whose stop_reason/stop_sequence/output_tokens stay null/zero the way a real stream's does instead of leaking the completed response's final state. Resolves #24004
This commit is contained in:
Yassin Kortam
GitHub
parent
c96245d7bf
commit
c70b911122
6 changed files with 1911 additions and 25 deletions
|
|
@ -38,6 +38,21 @@ DROP_DISABLED_THINKING_WARNING: Final = (
|
|||
"thinking blocks, and those thinking tokens are billed as output tokens."
|
||||
)
|
||||
|
||||
# Anthropic error `type` (both the JSON error body and SSE `event: error`
|
||||
# payloads use this field) mapped to the HTTP status code it corresponds to.
|
||||
ANTHROPIC_ERROR_STATUS_CODE_MAP: Final = MappingProxyType(
|
||||
{
|
||||
"invalid_request_error": 400,
|
||||
"authentication_error": 401,
|
||||
"permission_error": 403,
|
||||
"not_found_error": 404,
|
||||
"rate_limit_error": 429,
|
||||
"api_error": 500,
|
||||
"overloaded_error": 503,
|
||||
"timeout_error": 504,
|
||||
}
|
||||
)
|
||||
|
||||
_BEDROCK_VERSION_SUFFIX_RE: Final = re.compile(r"-v\d+(?::\d+)?$")
|
||||
_INFERENCE_PROFILE_MINOR_RE: Final = re.compile(r":\d+$")
|
||||
_DATED_RELEASE_SUFFIX_RE: Final = re.compile(r"-\d{8}$")
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import asyncio
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from collections.abc import AsyncIterator, Mapping
|
||||
from datetime import datetime
|
||||
from typing import Any, Final, Protocol, runtime_checkable
|
||||
|
||||
|
|
@ -11,9 +11,11 @@ from typing_extensions import TypedDict
|
|||
from litellm.litellm_core_utils.core_helpers import process_response_headers
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP
|
||||
from litellm.proxy.pass_through_endpoints.success_handler import (
|
||||
PassThroughEndpointLogging,
|
||||
)
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
|
||||
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
|
||||
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
|
||||
|
||||
|
|
@ -33,26 +35,239 @@ def _is_message_stop_chunk(chunk: object) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def _is_provider_error_chunk(chunk: object) -> bool:
|
||||
def is_anthropic_ping_chunk(chunk: object) -> bool:
|
||||
"""
|
||||
Whether a chunk is a pure ``ping`` keepalive frame. It carries no content
|
||||
and can recur indefinitely on a slow-starting or idle connection, so a
|
||||
mid-stream fallback wrapper drops it outright while still deciding
|
||||
whether to commit to the primary stream, rather than buffering it.
|
||||
|
||||
A physical transport chunk that coalesces a ping with any other SSE
|
||||
event (``message_start``, ``content_block_delta``, ``event: error``, ...)
|
||||
is NOT a pure ping - dropping it whole would discard those events - so
|
||||
only a chunk whose every ``event:`` line is ``event: ping`` qualifies.
|
||||
"""
|
||||
if isinstance(chunk, dict):
|
||||
return chunk.get("type") == "error"
|
||||
return chunk.get("type") == "ping"
|
||||
if isinstance(chunk, (bytes, bytearray)):
|
||||
return any(line == b"event: error" for line in chunk.splitlines())
|
||||
event_lines: Final = tuple(line for line in chunk.splitlines() if line.startswith(b"event:"))
|
||||
return bool(event_lines) and all(line == b"event: ping" for line in event_lines)
|
||||
return False
|
||||
|
||||
|
||||
def is_anthropic_content_delta_chunk(chunk: object) -> bool:
|
||||
"""
|
||||
Whether a chunk carries actual assistant-generated output (a
|
||||
``content_block_delta`` frame), as opposed to a lifecycle/bookkeeping
|
||||
frame (``message_start``, ``content_block_start``/``stop``,
|
||||
``message_delta``, ``message_stop``, ``ping``) that carries nothing
|
||||
worth preserving before an invisible mid-stream fallback retry.
|
||||
"""
|
||||
if isinstance(chunk, dict):
|
||||
return chunk.get("type") == "content_block_delta"
|
||||
if isinstance(chunk, (bytes, bytearray)):
|
||||
return any(line == b"event: content_block_delta" for line in chunk.splitlines())
|
||||
return False
|
||||
|
||||
|
||||
def _decoded_sse_data_line(line: bytes) -> object | None:
|
||||
if not line.startswith(b"data:"):
|
||||
return None
|
||||
try:
|
||||
return json.loads(line[len(b"data:") :].strip())
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None:
|
||||
if isinstance(chunk, dict):
|
||||
return chunk if chunk.get("type") == "error" else None
|
||||
if isinstance(chunk, (bytes, bytearray)):
|
||||
decoded_lines: Final = (_decoded_sse_data_line(line) for line in chunk.splitlines())
|
||||
return next(
|
||||
(
|
||||
candidate
|
||||
for candidate in decoded_lines
|
||||
if isinstance(candidate, dict) and candidate.get("type") == "error"
|
||||
),
|
||||
None,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _anthropic_error_body(chunk: object) -> Mapping[str, object] | None:
|
||||
"""Return the ``error`` object of an Anthropic SSE ``event: error`` chunk, or None."""
|
||||
payload: Final = _anthropic_error_event_payload(chunk)
|
||||
error_body: Final = payload.get("error") if payload is not None else None
|
||||
return error_body if isinstance(error_body, dict) else None
|
||||
|
||||
|
||||
def _is_provider_error_chunk(chunk: object) -> bool:
|
||||
return _anthropic_error_body(chunk) is not None
|
||||
|
||||
|
||||
def parse_anthropic_error_event(chunk: object) -> tuple[str, str, int] | None:
|
||||
"""
|
||||
Extract ``(error_type, message, http_status_code)`` from an Anthropic SSE
|
||||
``event: error`` chunk (raw bytes or an already-decoded dict), or None if
|
||||
``chunk`` is not an error event.
|
||||
|
||||
The status code is looked up via ANTHROPIC_ERROR_STATUS_CODE_MAP,
|
||||
defaulting to 500 for an error ``type`` Anthropic hasn't documented yet.
|
||||
"""
|
||||
error_body: Final = _anthropic_error_body(chunk)
|
||||
if error_body is None:
|
||||
return None
|
||||
error_type: Final = error_body.get("type")
|
||||
if not isinstance(error_type, str):
|
||||
return None
|
||||
message: Final = error_body.get("message")
|
||||
return (
|
||||
error_type,
|
||||
message if isinstance(message, str) else error_type,
|
||||
ANTHROPIC_ERROR_STATUS_CODE_MAP.get(error_type, 500),
|
||||
)
|
||||
|
||||
|
||||
def _is_terminal_stream_chunk(chunk: object) -> bool:
|
||||
return _is_message_stop_chunk(chunk) or _is_provider_error_chunk(chunk)
|
||||
|
||||
|
||||
def _sse_event(event_type: str, payload: Mapping[str, object]) -> bytes:
|
||||
return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()
|
||||
|
||||
|
||||
def _incomplete_stream_error_sse_event() -> bytes:
|
||||
payload: Final = json.dumps(
|
||||
{
|
||||
"type": "error",
|
||||
"error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE},
|
||||
}
|
||||
return _sse_event( # mutable-ok: one-shot JSON payload, never mutated after construction
|
||||
"error",
|
||||
{"type": "error", "error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}},
|
||||
)
|
||||
|
||||
|
||||
def _anthropic_content_block_start_and_deltas(
|
||||
block: Mapping[str, object],
|
||||
) -> tuple[Mapping[str, object], tuple[Mapping[str, object], ...]]:
|
||||
"""
|
||||
``(content_block_start.content_block, content_block_delta.delta events)``
|
||||
for one Anthropic response content block. A thinking block emits both a
|
||||
thinking_delta and a trailing signature_delta - a real Anthropic stream
|
||||
does the same, and dropping the signature makes any replay of that
|
||||
assistant message (a follow-up turn, a tool-use continuation) fail
|
||||
Anthropic's thinking-signature verification. redacted_thinking has no
|
||||
delta at all - it is sent complete in content_block_start.
|
||||
"""
|
||||
match block.get("type"):
|
||||
case "tool_use":
|
||||
return (
|
||||
{ # mutable-ok: one-shot payload
|
||||
"id": block.get("id"),
|
||||
"name": block.get("name"),
|
||||
"input": {}, # mutable-ok: one-shot payload
|
||||
"type": "tool_use",
|
||||
},
|
||||
(
|
||||
{ # mutable-ok: one-shot payload
|
||||
"partial_json": json.dumps(block.get("input") or {}), # mutable-ok: one-shot payload
|
||||
"type": "input_json_delta",
|
||||
},
|
||||
),
|
||||
)
|
||||
case "thinking":
|
||||
signature: Final = block.get("signature")
|
||||
signature_deltas: Final = (
|
||||
({"signature": signature, "type": "signature_delta"},) # mutable-ok: one-shot payload
|
||||
if isinstance(signature, str) and signature
|
||||
else ()
|
||||
)
|
||||
return (
|
||||
{"thinking": "", "signature": "", "type": "thinking"}, # mutable-ok: one-shot payload
|
||||
(
|
||||
{"thinking": block.get("thinking") or "", "type": "thinking_delta"}, # mutable-ok: one-shot payload
|
||||
*signature_deltas,
|
||||
),
|
||||
)
|
||||
case "redacted_thinking":
|
||||
return ({"type": "redacted_thinking", "data": block.get("data")}, ()) # mutable-ok: one-shot JSON payload
|
||||
case _:
|
||||
return (
|
||||
{"type": "text", "text": ""}, # mutable-ok: one-shot JSON payload
|
||||
({"type": "text_delta", "text": block.get("text") or ""},), # mutable-ok: one-shot JSON payload
|
||||
)
|
||||
|
||||
|
||||
def anthropic_messages_response_as_sse_events(response: AnthropicMessagesResponse) -> tuple[bytes, ...]:
|
||||
"""
|
||||
Render a complete (non-streaming) AnthropicMessagesResponse as the SSE
|
||||
event sequence a real streaming request would have produced.
|
||||
|
||||
A mid-stream fallback can resolve to a non-streaming response even
|
||||
though the client asked to stream (e.g. an agentic tool-use loop that
|
||||
intercepts and returns a complete message) - yielding that dict directly
|
||||
into a `/v1/messages` SSE byte stream would produce a malformed
|
||||
response, so it's synthesized into the message_start/content_block_*/
|
||||
message_delta/message_stop lifecycle a real stream would have sent.
|
||||
"""
|
||||
content_blocks: Final = response.get("content") or ()
|
||||
content_events: Final = (
|
||||
event for index, block in enumerate(content_blocks) for event in _anthropic_content_block_events(index, block)
|
||||
)
|
||||
# A real message_start always carries a null stop_reason/stop_sequence and
|
||||
# a zero output_tokens - those are only known once generation finishes, so
|
||||
# copying the completed response's final values here would let a client
|
||||
# treat the message as already finished, or double-count output tokens.
|
||||
message_start_usage: Final = { # mutable-ok: one-shot JSON payload
|
||||
**(response.get("usage") or {}),
|
||||
"output_tokens": 0,
|
||||
}
|
||||
message_start_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction
|
||||
"type": "message_start",
|
||||
"message": { # mutable-ok: one-shot JSON payload
|
||||
**response,
|
||||
"content": [], # mutable-ok: one-shot JSON payload
|
||||
"stop_reason": None,
|
||||
"stop_sequence": None,
|
||||
"usage": message_start_usage,
|
||||
},
|
||||
}
|
||||
message_delta_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction
|
||||
"type": "message_delta",
|
||||
"delta": { # mutable-ok: one-shot JSON payload
|
||||
"stop_reason": response.get("stop_reason"),
|
||||
"stop_sequence": response.get("stop_sequence"),
|
||||
},
|
||||
"usage": response.get("usage") or {}, # mutable-ok: one-shot JSON payload
|
||||
}
|
||||
return (
|
||||
_sse_event("message_start", message_start_payload),
|
||||
*content_events,
|
||||
_sse_event("message_delta", message_delta_payload),
|
||||
_sse_event("message_stop", {"type": "message_stop"}), # mutable-ok: one-shot JSON payload
|
||||
)
|
||||
|
||||
|
||||
def _anthropic_content_block_events(index: int, block: Mapping[str, object]) -> tuple[bytes, ...]:
|
||||
start_block, deltas = _anthropic_content_block_start_and_deltas(block)
|
||||
start_payload: Final = { # mutable-ok: one-shot payload
|
||||
"type": "content_block_start",
|
||||
"index": index,
|
||||
"content_block": start_block,
|
||||
}
|
||||
stop_payload: Final = { # mutable-ok: one-shot payload
|
||||
"type": "content_block_stop",
|
||||
"index": index,
|
||||
}
|
||||
delta_events: Final = tuple(
|
||||
_sse_event(
|
||||
"content_block_delta",
|
||||
{"type": "content_block_delta", "index": index, "delta": delta}, # mutable-ok: one-shot payload
|
||||
)
|
||||
for delta in deltas
|
||||
)
|
||||
return (
|
||||
_sse_event("content_block_start", start_payload),
|
||||
*delta_events,
|
||||
_sse_event("content_block_stop", stop_payload),
|
||||
)
|
||||
return f"event: error\ndata: {payload}\n\n".encode()
|
||||
|
||||
|
||||
class AnthropicMessagesStreamHiddenParams(TypedDict):
|
||||
|
|
|
|||
|
|
@ -22,19 +22,7 @@ from litellm.types.llms.openai import (
|
|||
from litellm.types.utils import CallTypes, LlmProviders, ModelResponse
|
||||
|
||||
from ..chat.transformation import AnthropicConfig
|
||||
from ..common_utils import AnthropicModelInfo
|
||||
|
||||
# Map Anthropic error types to HTTP status codes
|
||||
ANTHROPIC_ERROR_STATUS_CODE_MAP: Final = {
|
||||
"invalid_request_error": 400,
|
||||
"authentication_error": 401,
|
||||
"permission_error": 403,
|
||||
"not_found_error": 404,
|
||||
"rate_limit_error": 429,
|
||||
"api_error": 500,
|
||||
"overloaded_error": 503,
|
||||
"timeout_error": 504,
|
||||
}
|
||||
from ..common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP, AnthropicModelInfo
|
||||
|
||||
|
||||
class AnthropicFilesHandler:
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@
|
|||
# Thank you ! We ❤️ you! - Krrish & Ishaan
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import copy
|
||||
import enum
|
||||
import hashlib
|
||||
|
|
@ -20,7 +21,7 @@ import time
|
|||
import traceback
|
||||
import weakref
|
||||
from collections import defaultdict
|
||||
from collections.abc import AsyncGenerator, Callable, Generator, Mapping, Sequence
|
||||
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Generator, Mapping, Sequence
|
||||
from functools import lru_cache
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypeAlias, TypeVar, Union, cast
|
||||
|
|
@ -248,6 +249,7 @@ from .router_utils.pattern_match_deployments import PatternMatchRouter
|
|||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span as _Span
|
||||
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
from litellm.responses.streaming_iterator import (
|
||||
BaseResponsesAPIStreamingIterator,
|
||||
)
|
||||
|
|
@ -264,6 +266,9 @@ if TYPE_CHECKING:
|
|||
from litellm.router_strategy.quality_router.quality_router import (
|
||||
QualityRouter,
|
||||
)
|
||||
from litellm.types.llms.anthropic_messages.anthropic_response import (
|
||||
AnthropicMessagesResponse,
|
||||
)
|
||||
from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject
|
||||
from litellm.types.llms.openai import (
|
||||
ResponseAPIUsage,
|
||||
|
|
@ -361,6 +366,101 @@ def _stream_chunks_have_generated_content(chunks: Sequence[ModelResponseStream])
|
|||
return False
|
||||
|
||||
|
||||
# Router._aanthropic_messages_streaming_iterator buffers lifecycle chunks
|
||||
# until real content commits the primary stream; a hostile or slow-starting
|
||||
# upstream that never emits content or an error could otherwise grow that
|
||||
# buffer without bound, so hitting this cap forces an early commit instead.
|
||||
MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS: Final = 200
|
||||
|
||||
|
||||
def _anthropic_stream_should_drop_pre_content_ping(chunk: object, has_generated_content: bool) -> bool:
|
||||
"""A `ping` keepalive seen before any real content is dropped outright - it recurs indefinitely on a
|
||||
slow-starting connection and carries nothing worth buffering toward a possible fallback."""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import is_anthropic_ping_chunk
|
||||
|
||||
if has_generated_content:
|
||||
return False
|
||||
return is_anthropic_ping_chunk(chunk)
|
||||
|
||||
|
||||
def _is_retriable_anthropic_status(status_code: int) -> bool:
|
||||
return status_code == 429 or status_code >= 500
|
||||
|
||||
|
||||
def _anthropic_stream_should_decline_fallback(has_generated_content: bool, error: "MidStreamFallbackError") -> bool:
|
||||
"""
|
||||
A MidStreamFallbackError raised directly by the source iterator (the
|
||||
completion-bridge path's CustomStreamWrapper, e.g. on a transport drop)
|
||||
carries its own pre_first_chunk bookkeeping - gated the same way a
|
||||
detected SSE error event is, so a fallback is never appended after real
|
||||
content already reached the client on either path.
|
||||
"""
|
||||
return has_generated_content or not error.is_pre_first_chunk
|
||||
|
||||
|
||||
def _anthropic_stream_commits_now(chunk: object, has_generated_content: bool, buffered_chunk_count: int) -> bool:
|
||||
"""
|
||||
Whether `chunk` should make Router._aanthropic_messages_streaming_iterator
|
||||
commit to the primary Anthropic stream (real content arrived, or the
|
||||
pre-content buffer cap was hit) rather than keep buffering lifecycle
|
||||
frames toward a possible fallback.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
|
||||
is_anthropic_content_delta_chunk,
|
||||
)
|
||||
|
||||
if has_generated_content:
|
||||
return False
|
||||
return is_anthropic_content_delta_chunk(chunk) or buffered_chunk_count >= MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS
|
||||
|
||||
|
||||
class FallbackAwareAnthropicMessagesStream:
|
||||
"""
|
||||
Bare async generators can't carry the `_hidden_params` attribute the
|
||||
proxy reads response headers off of (see
|
||||
router_utils.add_retry_fallback_headers.get_hidden_params_dict), so this
|
||||
thin wrapper carries it through from the source iterator - mirrors
|
||||
AnthropicMessagesStreamingResponse. Used by
|
||||
Router._aanthropic_messages_streaming_iterator.
|
||||
"""
|
||||
|
||||
def __init__(self, async_generator: AsyncGenerator[bytes, None], source_iterator: object) -> None:
|
||||
self._async_generator = async_generator
|
||||
self._hidden_params = dict( # mutable-ok: mutated in place by merge_fallback_hidden_params
|
||||
getattr(source_iterator, "_hidden_params", None) or {}
|
||||
)
|
||||
|
||||
def __aiter__(self) -> "FallbackAwareAnthropicMessagesStream":
|
||||
return self
|
||||
|
||||
async def __anext__(self) -> bytes:
|
||||
return await self._async_generator.__anext__()
|
||||
|
||||
async def aclose(self) -> None:
|
||||
await self._async_generator.aclose()
|
||||
|
||||
def merge_fallback_hidden_params(
|
||||
self,
|
||||
fallback_hidden_params: Mapping[str, object],
|
||||
fallback_headers: Mapping[str, object],
|
||||
) -> None:
|
||||
"""
|
||||
Raw bytes can't carry their own _hidden_params the way a
|
||||
ModelResponseStream/ResponsesAPI event can, so a mid-stream
|
||||
fallback's provider headers (e.g. Bedrock's x-amzn-requestid) are
|
||||
merged onto the wrapper itself instead - mirrors
|
||||
Router._apply_fallback_hidden_params_to_item's merge shape.
|
||||
"""
|
||||
existing_headers: Final = cast( # cast-ok: additional_headers is always a dict[str, object] when present
|
||||
"dict[str, object]", self._hidden_params.get("additional_headers") or {}
|
||||
)
|
||||
self._hidden_params = { # mutable-ok: matches _hidden_params' existing dict[str, object] shape
|
||||
**self._hidden_params,
|
||||
**fallback_hidden_params,
|
||||
"additional_headers": {**existing_headers, **fallback_headers}, # mutable-ok: same shape
|
||||
}
|
||||
|
||||
|
||||
class RoutingArgs(enum.Enum):
|
||||
ttl = 60 # 1min (RPM/TPM expire key)
|
||||
|
||||
|
|
@ -4806,6 +4906,264 @@ class Router:
|
|||
)
|
||||
return response
|
||||
|
||||
async def _aanthropic_messages_streaming_iterator(
|
||||
self,
|
||||
response: AsyncIterator[bytes],
|
||||
initial_kwargs: dict[str, Any], # mutable-ok: mutated in-place before re-entering the fallback chain
|
||||
) -> AsyncIterator[bytes]:
|
||||
"""
|
||||
Wrap an anthropic_messages (/v1/messages) streaming response so a
|
||||
mid-stream provider error triggers the Router's fallback chain
|
||||
(parity with _acompletion_streaming_iterator for the
|
||||
chat-completions path). See #24004.
|
||||
|
||||
anthropic_messages goes through _ageneric_api_call_with_fallbacks
|
||||
rather than _acompletion, so the returned byte iterator is never
|
||||
wrapped by the chat-completions fallback handler. Two failure
|
||||
shapes land here:
|
||||
- the completion-bridge path (deployments with no native
|
||||
/v1/messages endpoint, via
|
||||
LiteLLMMessagesToCompletionTransformationHandler) already
|
||||
raises MidStreamFallbackError out of its underlying
|
||||
CustomStreamWrapper; this wrapper only needs to catch it.
|
||||
- a native Anthropic/Bedrock passthrough never raises anything
|
||||
for a provider SSE `event: error` frame (e.g. `overloaded_error`,
|
||||
`internal_server_error`) - it is forwarded to the client as-is -
|
||||
so this wrapper detects it via parse_anthropic_error_event and
|
||||
raises MidStreamFallbackError itself.
|
||||
|
||||
Only an error before any real content (a content_block_delta frame)
|
||||
has reached the caller triggers a fallback attempt, mirroring the
|
||||
restriction _acompletion_streaming_iterator applies: once generated
|
||||
output has already reached the caller, retrying would start a
|
||||
second, overlapping Anthropic message lifecycle on the same SSE
|
||||
stream, so the error is left to propagate instead of being retried
|
||||
invisibly. A non-retriable client error (4xx other than 429) is
|
||||
never worth a fallback attempt either, so it is also left to
|
||||
propagate.
|
||||
|
||||
Lifecycle/bookkeeping frames (message_start, content_block_start,
|
||||
ping, ...) do not by themselves disqualify a fallback attempt -
|
||||
Anthropic routinely sends message_start before an overload error -
|
||||
but they are BUFFERED rather than forwarded immediately, since
|
||||
forwarding one and then appending a fallback attempt's own
|
||||
message_start would produce two overlapping message lifecycles on
|
||||
one SSE stream. Buffered frames are flushed, in order, the moment
|
||||
real content arrives (the primary attempt has committed by then
|
||||
anyway) or once the stream ends without ever producing content or
|
||||
an error.
|
||||
"""
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
|
||||
aclose_if_supported,
|
||||
parse_anthropic_error_event,
|
||||
)
|
||||
|
||||
source_iterator: Final = response
|
||||
|
||||
async def stream_with_fallbacks() -> AsyncGenerator[bytes, None]:
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
|
||||
# Lifecycle/bookkeeping frames (message_start, content_block_start,
|
||||
# ping, ...) are held back rather than forwarded immediately:
|
||||
# Anthropic routinely sends message_start before an overload
|
||||
# error, and once a byte reaches the client a fallback attempt
|
||||
# can only append its OWN message_start, producing two
|
||||
# overlapping message lifecycles on one SSE stream. Buffered
|
||||
# frames are flushed the moment real content (content_block_delta)
|
||||
# arrives - at that point the primary attempt has committed and a
|
||||
# clean retry is no longer possible anyway - or once the primary
|
||||
# stream ends without ever producing content. A `ping` keepalive
|
||||
# is dropped outright rather than buffered, since it can recur
|
||||
# indefinitely on a slow-starting connection and carries nothing
|
||||
# worth preserving; hitting MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS
|
||||
# forces the same early commit as real content arriving, so a
|
||||
# hostile or pathological upstream can't grow the buffer forever.
|
||||
has_generated_content = False # rebind-ok: set once real content is seen, or the buffer cap is hit
|
||||
buffered_lifecycle_chunks: tuple[bytes, ...] = () # rebind-ok: flushed once committed or on decline
|
||||
model: Final = cast(str, initial_kwargs.get("model")) # cast-ok: kwargs always carries the model group
|
||||
try:
|
||||
async for chunk in source_iterator:
|
||||
if _anthropic_stream_should_drop_pre_content_ping(chunk, has_generated_content):
|
||||
continue
|
||||
if _anthropic_stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)):
|
||||
has_generated_content = True # rebind-ok: real content seen, or the buffer cap was hit
|
||||
error_event = parse_anthropic_error_event(chunk)
|
||||
retriable_pending_error = ( # rebind-ok: freshly computed each iteration, never carried over
|
||||
not has_generated_content
|
||||
and error_event is not None
|
||||
and _is_retriable_anthropic_status(error_event[2])
|
||||
)
|
||||
if not has_generated_content and not retriable_pending_error and error_event is None:
|
||||
buffered_lifecycle_chunks = (*buffered_lifecycle_chunks, chunk)
|
||||
continue
|
||||
if retriable_pending_error:
|
||||
assert error_event is not None # guard-ok: retriable_pending_error implies this
|
||||
_error_type, message, status_code = error_event
|
||||
raise MidStreamFallbackError(
|
||||
message=message,
|
||||
model=model,
|
||||
llm_provider="anthropic",
|
||||
original_exception=litellm.exceptions.APIError(
|
||||
status_code=status_code,
|
||||
message=message,
|
||||
llm_provider="anthropic",
|
||||
model=model,
|
||||
),
|
||||
is_pre_first_chunk=True,
|
||||
)
|
||||
for buffered_chunk in buffered_lifecycle_chunks:
|
||||
yield buffered_chunk
|
||||
buffered_lifecycle_chunks = ()
|
||||
yield chunk
|
||||
for buffered_chunk in buffered_lifecycle_chunks:
|
||||
yield buffered_chunk
|
||||
except MidStreamFallbackError as e:
|
||||
if _anthropic_stream_should_decline_fallback(has_generated_content, e):
|
||||
for buffered_chunk in buffered_lifecycle_chunks:
|
||||
yield buffered_chunk
|
||||
if e.original_exception is not None:
|
||||
raise e.original_exception from e
|
||||
raise
|
||||
async for item in self._aanthropic_messages_fallback_attempt(e, initial_kwargs, wrapper):
|
||||
yield item
|
||||
finally:
|
||||
with anyio.CancelScope(shield=True), contextlib.suppress(BaseException):
|
||||
await aclose_if_supported(source_iterator)
|
||||
|
||||
# Referenced by stream_with_fallbacks via closure - assigned here, before
|
||||
# the generator body ever runs, so the reference resolves fine despite
|
||||
# being defined textually after the function that captures it.
|
||||
wrapper: Final = FallbackAwareAnthropicMessagesStream(stream_with_fallbacks(), source_iterator)
|
||||
return wrapper
|
||||
|
||||
async def _aanthropic_messages_fallback_attempt(
|
||||
self,
|
||||
e: "MidStreamFallbackError",
|
||||
initial_kwargs: dict[str, Any], # mutable-ok: mutated in-place before re-entering the fallback chain
|
||||
wrapper: "FallbackAwareAnthropicMessagesStream",
|
||||
) -> AsyncGenerator[bytes, None]:
|
||||
"""
|
||||
Re-enters the Router's fallback chain for a mid-stream
|
||||
anthropic_messages error and yields whatever the fallback attempt
|
||||
produces. Split out of _aanthropic_messages_streaming_iterator to
|
||||
keep each function's cyclomatic complexity within the repo's C901
|
||||
budget.
|
||||
"""
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
|
||||
aclose_if_supported,
|
||||
anthropic_messages_response_as_sse_events,
|
||||
)
|
||||
|
||||
fallback_response = None # rebind-ok: pre-init so finally can close it if a fallback was actually attempted
|
||||
try:
|
||||
model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: model group
|
||||
fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the common_utils list|None param
|
||||
"fallbacks", self.fallbacks
|
||||
)
|
||||
context_window_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below
|
||||
"context_window_fallbacks", self.context_window_fallbacks
|
||||
)
|
||||
content_policy_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below
|
||||
"content_policy_fallbacks", self.content_policy_fallbacks
|
||||
)
|
||||
initial_kwargs["original_function"] = self._ageneric_api_call_with_fallbacks_helper
|
||||
self._update_kwargs_before_fallbacks(
|
||||
model=model_group,
|
||||
kwargs=initial_kwargs,
|
||||
metadata_variable_name="litellm_metadata",
|
||||
)
|
||||
fallback_response = await self.async_function_with_fallbacks_common_utils( # rebind-ok: set on success
|
||||
e=e,
|
||||
disable_fallbacks=False,
|
||||
fallbacks=fallbacks,
|
||||
context_window_fallbacks=context_window_fallbacks,
|
||||
content_policy_fallbacks=content_policy_fallbacks,
|
||||
model_group=model_group,
|
||||
args=(),
|
||||
kwargs=initial_kwargs,
|
||||
include_fallback_errors=initial_kwargs.get("include_fallback_errors", False) is True,
|
||||
)
|
||||
fallback_hidden_params, fallback_headers = Router._prepare_fallback_hidden_params(fallback_response)
|
||||
wrapper.merge_fallback_hidden_params(fallback_hidden_params, fallback_headers)
|
||||
if hasattr(fallback_response, "__aiter__"):
|
||||
async for fallback_item in fallback_response:
|
||||
yield fallback_item
|
||||
else:
|
||||
# A fallback can resolve to a complete AnthropicMessagesResponse
|
||||
# dict even for a streaming request (e.g. an agentic tool-use
|
||||
# interception loop) - yielding it as-is would put a raw dict
|
||||
# into a byte stream, so it's synthesized into the SSE
|
||||
# lifecycle a real stream would have sent instead.
|
||||
for event in anthropic_messages_response_as_sse_events(
|
||||
cast("AnthropicMessagesResponse", fallback_response) # cast-ok: non-streaming shape by elimination
|
||||
):
|
||||
yield event
|
||||
except Exception as fallback_error:
|
||||
verbose_router_logger.error("Anthropic messages streaming fallback also failed: %s", fallback_error)
|
||||
if isinstance(fallback_error, MidStreamFallbackError) and fallback_error.original_exception is not None:
|
||||
raise fallback_error.original_exception from fallback_error
|
||||
raise
|
||||
finally:
|
||||
if fallback_response is not None:
|
||||
with anyio.CancelScope(shield=True), contextlib.suppress(BaseException):
|
||||
await aclose_if_supported(fallback_response)
|
||||
|
||||
async def _aanthropic_messages_with_streaming_fallbacks(
|
||||
self,
|
||||
original_function: Callable,
|
||||
**kwargs: object, # kwargs-ok: forwarded verbatim to original_function, shape varies per call site
|
||||
) -> Union["AnthropicMessagesResponse", AsyncIterator[bytes]]:
|
||||
"""
|
||||
_ageneric_api_call_with_fallbacks for anthropic_messages, with the
|
||||
addition of mid-stream fallback handling (see
|
||||
_aanthropic_messages_streaming_iterator). Parity with
|
||||
_aresponses_with_streaming_fallbacks for the Responses API.
|
||||
"""
|
||||
from litellm.litellm_core_utils.core_helpers import safe_deep_copy
|
||||
|
||||
# Snapshot the request kwargs before the primary attempt mutates them
|
||||
# in place: _update_kwargs_with_deployment writes deployment-specific
|
||||
# fields (deployment, model_info, api_base, tags, ...) into the
|
||||
# SAME litellm_metadata/metadata dicts a shallow .copy() would still
|
||||
# share, leaking primary-deployment metadata into the mid-stream
|
||||
# fallback request. safe_deep_copy avoids deep-copying the full
|
||||
# kwargs (which can hold non-deepcopyable logging handles/clients).
|
||||
fallback_kwargs: Final[dict[str, object]] = kwargs.copy() # mutable-ok: mutated below before re-entry
|
||||
if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
|
||||
fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
|
||||
if isinstance(fallback_kwargs.get("metadata"), dict):
|
||||
fallback_kwargs["metadata"] = safe_deep_copy(fallback_kwargs["metadata"])
|
||||
fallback_kwargs["original_generic_function"] = original_function
|
||||
|
||||
response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
|
||||
|
||||
if kwargs.get("stream") and hasattr(response, "__aiter__"):
|
||||
return await self._aanthropic_messages_streaming_iterator(
|
||||
response=cast("AsyncIterator[bytes]", response), # cast-ok: stream=True always returns a byte iterator
|
||||
initial_kwargs=fallback_kwargs,
|
||||
)
|
||||
return response
|
||||
|
||||
async def _dispatch_generic_call_type(
|
||||
self,
|
||||
call_type: str,
|
||||
original_function: Callable,
|
||||
**kwargs: object, # kwargs-ok: forwarded verbatim to the per-call-type helper, shape varies per call site
|
||||
):
|
||||
"""
|
||||
factory_function's shared dispatch for call types with no
|
||||
call-specific handling, except anthropic_messages: kept out of
|
||||
factory_function's own async_wrapper (already at the repo's C901
|
||||
complexity ceiling) so routing its mid-stream fallback handling
|
||||
(#24004) doesn't add another branch there.
|
||||
"""
|
||||
if call_type == "anthropic_messages":
|
||||
return await self._aanthropic_messages_with_streaming_fallbacks(
|
||||
original_function=original_function, **kwargs
|
||||
)
|
||||
return await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
|
||||
|
||||
def _generic_api_call_with_fallbacks(self, model: str, original_function: Callable, **kwargs):
|
||||
"""
|
||||
Make a generic LLM API call through the router, this allows you to use retries/fallbacks with litellm router
|
||||
|
|
@ -5992,7 +6350,8 @@ class Router:
|
|||
"aget_skill",
|
||||
"adelete_skill",
|
||||
):
|
||||
return await self._ageneric_api_call_with_fallbacks(
|
||||
return await self._dispatch_generic_call_type(
|
||||
call_type=call_type,
|
||||
original_function=original_function,
|
||||
**kwargs,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -11,6 +11,10 @@ from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterato
|
|||
BaseAnthropicMessagesStreamingIterator,
|
||||
_incomplete_stream_error_sse_event,
|
||||
_is_message_stop_chunk,
|
||||
_is_provider_error_chunk,
|
||||
anthropic_messages_response_as_sse_events,
|
||||
is_anthropic_content_delta_chunk,
|
||||
parse_anthropic_error_event,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -157,6 +161,96 @@ def test_is_message_stop_chunk_ignores_substring_in_payload():
|
|||
assert _is_message_stop_chunk(delta_frame_with_substring) is False
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_from_dict_chunk():
|
||||
"""Regression for #24004: dict-shaped error chunks parse to
|
||||
(type, message, status) so the Router can decide whether to fall back."""
|
||||
chunk = {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}
|
||||
assert parse_anthropic_error_event(chunk) == ("overloaded_error", "Overloaded", 503)
|
||||
assert _is_provider_error_chunk(chunk) is True
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_from_sse_bytes():
|
||||
"""Regression for #24004: a raw `event: error` SSE frame (what a native
|
||||
Anthropic/Bedrock passthrough forwards verbatim today) must parse
|
||||
identically to the dict shape so the Router can raise a fallback."""
|
||||
sse_chunk = (
|
||||
b"event: error\n"
|
||||
b'data: {"type": "error", "error": {"type": "internal_server_error", "message": "boom"}}\n\n'
|
||||
)
|
||||
assert parse_anthropic_error_event(sse_chunk) == ("internal_server_error", "boom", 500)
|
||||
assert _is_provider_error_chunk(sse_chunk) is True
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_defaults_status_for_unknown_type():
|
||||
chunk = {"type": "error", "error": {"type": "some_future_error_type", "message": "?"}}
|
||||
assert parse_anthropic_error_event(chunk) == ("some_future_error_type", "?", 500)
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_missing_message_falls_back_to_type():
|
||||
chunk = {"type": "error", "error": {"type": "overloaded_error"}}
|
||||
assert parse_anthropic_error_event(chunk) == ("overloaded_error", "overloaded_error", 503)
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_non_string_error_type_returns_none():
|
||||
"""A malformed error body whose `type` field isn't a string (e.g. an
|
||||
upstream bug sends null or a number) must not be treated as an error
|
||||
event rather than crashing or forwarding a garbage error_type."""
|
||||
chunk = {"type": "error", "error": {"type": None, "message": "boom"}}
|
||||
assert parse_anthropic_error_event(chunk) is None
|
||||
|
||||
|
||||
def test_decoded_sse_data_line_swallows_invalid_json():
|
||||
"""A `data:` line that isn't valid JSON (a malformed/truncated frame)
|
||||
must not be treated as an error event or raise, just be ignored."""
|
||||
malformed_frame = b"event: error\ndata: {not valid json\n\n"
|
||||
assert parse_anthropic_error_event(malformed_frame) is None
|
||||
assert _is_provider_error_chunk(malformed_frame) is False
|
||||
|
||||
|
||||
class TestIsAnthropicContentDeltaChunk:
|
||||
def test_dict_content_block_delta(self):
|
||||
assert is_anthropic_content_delta_chunk({"type": "content_block_delta"}) is True
|
||||
|
||||
def test_dict_other_type(self):
|
||||
assert is_anthropic_content_delta_chunk({"type": "message_start"}) is False
|
||||
|
||||
def test_bytes_content_block_delta(self):
|
||||
assert is_anthropic_content_delta_chunk(b"event: content_block_delta\ndata: {}\n\n") is True
|
||||
|
||||
def test_bytes_other_event(self):
|
||||
assert is_anthropic_content_delta_chunk(b"event: message_start\ndata: {}\n\n") is False
|
||||
|
||||
def test_neither_dict_nor_bytes(self):
|
||||
assert is_anthropic_content_delta_chunk("content_block_delta") is False
|
||||
assert is_anthropic_content_delta_chunk(None) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"chunk",
|
||||
[
|
||||
{"type": "content_block_delta", "delta": {"type": "text_delta", "text": "hi"}},
|
||||
b'event: content_block_delta\ndata: {"type": "content_block_delta"}\n\n',
|
||||
b"raw-bytes",
|
||||
"error",
|
||||
None,
|
||||
],
|
||||
)
|
||||
def test_parse_anthropic_error_event_non_error_chunks_return_none(chunk):
|
||||
assert parse_anthropic_error_event(chunk) is None
|
||||
assert _is_provider_error_chunk(chunk) is False
|
||||
|
||||
|
||||
def test_parse_anthropic_error_event_ignores_substring_in_payload():
|
||||
"""A content_block_delta whose partial_json happens to contain the
|
||||
literal string `"type": "error"` must not be misread as an error event."""
|
||||
delta_frame_with_substring = (
|
||||
b"event: content_block_delta\n"
|
||||
b'data: {"type": "content_block_delta", "delta": '
|
||||
b'{"type": "input_json_delta", "partial_json": "\\"type\\": \\"error\\""}}\n\n'
|
||||
)
|
||||
assert parse_anthropic_error_event(delta_frame_with_substring) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_sse_wrapper_emits_error_when_bytes_stream_only_mentions_message_stop_in_payload():
|
||||
"""
|
||||
|
|
@ -307,3 +401,117 @@ def test_incomplete_stream_error_sse_event_is_valid_anthropic_error():
|
|||
"error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE},
|
||||
}
|
||||
assert event.endswith("\n\n")
|
||||
|
||||
|
||||
def _decode_sse_events(events: tuple[bytes, ...]) -> list[tuple[str, dict]]:
|
||||
decoded = []
|
||||
for event in events:
|
||||
assert isinstance(event, bytes)
|
||||
lines = event.decode().split("\n")
|
||||
assert lines[0].startswith("event: ")
|
||||
decoded.append((lines[0].removeprefix("event: "), json.loads(lines[1].removeprefix("data: "))))
|
||||
return decoded
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_text_block():
|
||||
response = {
|
||||
"id": "msg_1",
|
||||
"model": "claude-haiku",
|
||||
"role": "assistant",
|
||||
"type": "message",
|
||||
"stop_reason": "end_turn",
|
||||
"stop_sequence": None,
|
||||
"content": [{"type": "text", "text": "hello"}],
|
||||
"usage": {"input_tokens": 3, "output_tokens": 2},
|
||||
}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
types = [event_type for event_type, _ in decoded]
|
||||
assert types == [
|
||||
"message_start",
|
||||
"content_block_start",
|
||||
"content_block_delta",
|
||||
"content_block_stop",
|
||||
"message_delta",
|
||||
"message_stop",
|
||||
]
|
||||
# message_start must not carry generated content itself, matching a real
|
||||
# streaming response - it arrives via the content_block_delta that follows.
|
||||
assert decoded[0][1]["message"]["content"] == []
|
||||
assert decoded[0][1]["message"]["id"] == "msg_1"
|
||||
# Bugbot regression: message_start must not carry the completed response's
|
||||
# final stop_reason/stop_sequence/output_tokens - a real stream keeps those
|
||||
# null/zero until message_delta, so a client could otherwise treat the
|
||||
# message as already finished, or double-count output tokens.
|
||||
assert decoded[0][1]["message"]["stop_reason"] is None
|
||||
assert decoded[0][1]["message"]["stop_sequence"] is None
|
||||
assert decoded[0][1]["message"]["usage"] == {"input_tokens": 3, "output_tokens": 0}
|
||||
assert decoded[1][1]["content_block"] == {"type": "text", "text": ""}
|
||||
assert decoded[2][1]["delta"] == {"type": "text_delta", "text": "hello"}
|
||||
assert decoded[4][1]["delta"]["stop_reason"] == "end_turn"
|
||||
assert decoded[4][1]["usage"] == {"input_tokens": 3, "output_tokens": 2}
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_tool_use_block():
|
||||
response = {
|
||||
"id": "msg_2",
|
||||
"content": [{"type": "tool_use", "id": "toolu_1", "name": "get_weather", "input": {"city": "NYC"}}],
|
||||
"stop_reason": "tool_use",
|
||||
}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
content_block_start = dict(decoded)["content_block_start"]
|
||||
assert content_block_start["content_block"] == {
|
||||
"type": "tool_use",
|
||||
"id": "toolu_1",
|
||||
"name": "get_weather",
|
||||
"input": {},
|
||||
}
|
||||
content_block_delta = dict(decoded)["content_block_delta"]
|
||||
assert json.loads(content_block_delta["delta"]["partial_json"]) == {"city": "NYC"}
|
||||
assert content_block_delta["delta"]["type"] == "input_json_delta"
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_thinking_block_emits_signature_delta():
|
||||
"""Bugbot regression: a thinking block's real `signature` must reach the
|
||||
client via a trailing signature_delta, not be silently dropped - Anthropic
|
||||
rejects a replayed assistant message (a follow-up turn, a tool-use
|
||||
continuation) whose thinking block lacks its original signature."""
|
||||
response = {
|
||||
"id": "msg_5",
|
||||
"content": [{"type": "thinking", "thinking": "let me think", "signature": "sig-abc123"}],
|
||||
"stop_reason": "end_turn",
|
||||
}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
deltas = [payload["delta"] for event_type, payload in decoded if event_type == "content_block_delta"]
|
||||
assert deltas == [
|
||||
{"type": "thinking_delta", "thinking": "let me think"},
|
||||
{"type": "signature_delta", "signature": "sig-abc123"},
|
||||
]
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_thinking_block_without_signature_omits_delta():
|
||||
response = {
|
||||
"id": "msg_6",
|
||||
"content": [{"type": "thinking", "thinking": "let me think", "signature": None}],
|
||||
"stop_reason": "end_turn",
|
||||
}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
deltas = [payload["delta"] for event_type, payload in decoded if event_type == "content_block_delta"]
|
||||
assert deltas == [{"type": "thinking_delta", "thinking": "let me think"}]
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_multiple_blocks_are_indexed():
|
||||
response = {
|
||||
"id": "msg_3",
|
||||
"content": [{"type": "text", "text": "a"}, {"type": "text", "text": "b"}],
|
||||
}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
starts = [payload for event_type, payload in decoded if event_type == "content_block_start"]
|
||||
assert [s["index"] for s in starts] == [0, 1]
|
||||
deltas = [payload for event_type, payload in decoded if event_type == "content_block_delta"]
|
||||
assert [d["delta"]["text"] for d in deltas] == ["a", "b"]
|
||||
|
||||
|
||||
def test_anthropic_messages_response_as_sse_events_no_content_blocks():
|
||||
response = {"id": "msg_4", "content": [], "stop_reason": "end_turn"}
|
||||
decoded = _decode_sse_events(anthropic_messages_response_as_sse_events(response))
|
||||
assert [event_type for event_type, _ in decoded] == ["message_start", "message_delta", "message_stop"]
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
Loading…
Add table
Reference in a new issue