mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
fix(anthropic): keep refusal stop_reason and stop_details on the logged response (#46104)
* fix(anthropic): keep refusal stop_reason and stop_details on the logged response * fix(anthropic): keep the refusal stop-fields boundary tolerant of a non-dict upstream body * test(anthropic): use Anthropic's documented refusal example as the unit-test fixture * fix(streaming): keep provider fields on async stream cache hits and received_model_id on synthesized finish chunks * fix(caching): replay cached completion_tokens_details on stream cache hits and skip non-object cached provider fields * test(integration): cover Anthropic-family refusal stop_details on the wire, the cache and under chaos * fix(caching): skip a cached completion_tokens_details that fails validation on stream replay * fix(anthropic): keep refusal stop fields on a JSON-mode non-stream response Also adds the round 3 audit cells for JSON-mode refusals, the proxy response serialization, and wrong-typed cached usage details on both stream replays * test(anthropic): expect content_filter on streamed JSON tool refusals after main's stream change --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
f10134728d
commit
47e4ed88c2
18 changed files with 2789 additions and 13 deletions
|
|
@ -6,7 +6,7 @@ import traceback
|
|||
from collections.abc import Mapping, Sequence
|
||||
from typing import Final, Literal, cast
|
||||
|
||||
from pydantic import ConfigDict, TypeAdapter, with_config
|
||||
from pydantic import ConfigDict, TypeAdapter, ValidationError, with_config
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
import litellm
|
||||
|
|
@ -222,6 +222,9 @@ async def convert_to_streaming_response_async(
|
|||
role=choice["message"]["role"],
|
||||
function_call=choice["message"].get("function_call", None),
|
||||
tool_calls=choice["message"].get("tool_calls", None),
|
||||
provider_specific_fields=_cached_provider_specific_fields(
|
||||
_CACHED_MESSAGE_PROVIDER_FIELDS.validate_python(choice["message"])
|
||||
),
|
||||
)
|
||||
finish_reason = choice.get("finish_reason", None)
|
||||
|
||||
|
|
@ -243,6 +246,9 @@ async def convert_to_streaming_response_async(
|
|||
completion_tokens=response_object["usage"].get("completion_tokens", 0),
|
||||
prompt_tokens=response_object["usage"].get("prompt_tokens", 0),
|
||||
total_tokens=response_object["usage"].get("total_tokens", 0),
|
||||
completion_tokens_details=_cached_completion_tokens_details(
|
||||
_CACHED_USAGE_DETAILS.validate_python(response_object["usage"])
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
|
@ -334,6 +340,9 @@ def convert_to_streaming_response(
|
|||
model_response_object.usage.completion_tokens = response_object["usage"].get("completion_tokens", 0)
|
||||
model_response_object.usage.prompt_tokens = response_object["usage"].get("prompt_tokens", 0)
|
||||
model_response_object.usage.total_tokens = response_object["usage"].get("total_tokens", 0)
|
||||
model_response_object.usage.completion_tokens_details = _cached_completion_tokens_details(
|
||||
_CACHED_USAGE_DETAILS.validate_python(response_object["usage"])
|
||||
)
|
||||
|
||||
if "id" in response_object:
|
||||
model_response_object.id = response_object["id"]
|
||||
|
|
@ -390,6 +399,35 @@ class _ParallelToolUseArgs(TypedDict):
|
|||
_PARALLEL_TOOL_USE_ARGS: Final = TypeAdapter(_ParallelToolUseArgs)
|
||||
|
||||
|
||||
class _CachedMessageProviderFields(TypedDict, total=False):
|
||||
provider_specific_fields: ReadOnly[object]
|
||||
|
||||
|
||||
class _CachedUsageDetails(TypedDict, total=False):
|
||||
completion_tokens_details: ReadOnly[object]
|
||||
|
||||
|
||||
_CACHED_MESSAGE_PROVIDER_FIELDS: Final = TypeAdapter(_CachedMessageProviderFields)
|
||||
_CACHED_USAGE_DETAILS: Final = TypeAdapter(_CachedUsageDetails)
|
||||
_FIELD_MAP: Final = TypeAdapter(dict[str, object])
|
||||
_COMPLETION_TOKENS_DETAILS: Final = TypeAdapter(CompletionTokensDetailsWrapper)
|
||||
|
||||
|
||||
def _cached_provider_specific_fields(message: _CachedMessageProviderFields) -> Mapping[str, object] | None:
|
||||
fields: Final = message.get("provider_specific_fields")
|
||||
return _FIELD_MAP.validate_python(fields) if isinstance(fields, dict) else None
|
||||
|
||||
|
||||
def _cached_completion_tokens_details(usage: _CachedUsageDetails) -> CompletionTokensDetailsWrapper | None:
|
||||
details: Final = usage.get("completion_tokens_details")
|
||||
if not isinstance(details, dict):
|
||||
return None
|
||||
try:
|
||||
return _COMPLETION_TOKENS_DETAILS.validate_python(details)
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def handle_invalid_parallel_tool_calls(
|
||||
tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall],
|
||||
) -> list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None:
|
||||
|
|
|
|||
|
|
@ -280,6 +280,7 @@ class CustomStreamWrapper:
|
|||
|
||||
self._response_headers = _response_headers
|
||||
self.response_id: str | None = None
|
||||
self.received_model_id: str | None = None
|
||||
self.logging_loop = None
|
||||
self.rules = Rules()
|
||||
self.stream_options = stream_options or getattr(logging_obj, "stream_options", None)
|
||||
|
|
@ -838,6 +839,7 @@ class CustomStreamWrapper:
|
|||
self.response_id = id
|
||||
|
||||
if id and isinstance(id, str) and id.strip():
|
||||
self.received_model_id = id
|
||||
model_response.hidden_params["received_model_id"] = id
|
||||
|
||||
if self.response_id is not None and isinstance(self.response_id, str):
|
||||
|
|
@ -1756,7 +1758,9 @@ class CustomStreamWrapper:
|
|||
self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)
|
||||
|
||||
def finish_reason_handler(self):
|
||||
model_response: Final = self.model_response_creator()
|
||||
model_response: Final = self.model_response_creator(
|
||||
hidden_params=None if self.received_model_id is None else {"received_model_id": self.received_model_id}
|
||||
)
|
||||
_finish_reason: Final = self.received_finish_reason or self.intermittent_finish_reason
|
||||
if _finish_reason is not None:
|
||||
model_response.choices[0].finish_reason = _finish_reason
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@ from litellm.types.utils import (
|
|||
)
|
||||
|
||||
from ...base import BaseLLM
|
||||
from ..common_utils import AnthropicError, process_anthropic_headers
|
||||
from ..common_utils import AnthropicError, process_anthropic_headers, refusal_provider_specific_fields
|
||||
from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig, JsonModeRequest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -1102,7 +1102,8 @@ class ModelResponseIterator:
|
|||
# These are automatically handled by Anthropic API, we just pass them through
|
||||
pass
|
||||
elif type_chunk == "message_delta":
|
||||
finish_reason, usage, container = self._handle_message_delta(chunk)
|
||||
finish_reason, usage, container, refusal_fields = self._handle_message_delta(chunk)
|
||||
provider_specific_fields.update(refusal_fields)
|
||||
text = self._release_answer_close() + self._drain_split_answer_blocks()
|
||||
if container:
|
||||
provider_specific_fields["container"] = container
|
||||
|
|
@ -1297,7 +1298,9 @@ class ModelResponseIterator:
|
|||
return ""
|
||||
return _split_answer_text(blocks, self.response_format_schema)
|
||||
|
||||
def _handle_message_delta(self, chunk: dict) -> tuple[str, Usage | None, dict[str, Any] | None]:
|
||||
def _handle_message_delta(
|
||||
self, chunk: dict
|
||||
) -> tuple[str, Usage | None, dict[str, Any] | None, Mapping[str, object]]:
|
||||
"""
|
||||
Handle message_delta event for finish_reason, usage, and container.
|
||||
|
||||
|
|
@ -1319,7 +1322,7 @@ class ModelResponseIterator:
|
|||
self._handle_usage(anthropic_usage_chunk=message_delta["usage"]) if "usage" in message_delta else None
|
||||
)
|
||||
container: Final = message_delta["delta"].get("container")
|
||||
return finish_reason, usage, container
|
||||
return finish_reason, usage, container, refusal_provider_specific_fields(message_delta["delta"])
|
||||
|
||||
def _handle_accumulated_json_chunk(self, data_str: str, is_final: bool = False) -> ModelResponseStream | None:
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -104,10 +104,12 @@ from litellm.utils import (
|
|||
)
|
||||
|
||||
from ..common_utils import (
|
||||
REFUSAL_STOP_FIELDS,
|
||||
AnthropicError,
|
||||
AnthropicModelInfo,
|
||||
eager_input_streaming_flag,
|
||||
process_anthropic_headers,
|
||||
refusal_provider_specific_fields,
|
||||
requires_native_compaction_beta,
|
||||
strip_advisor_blocks_from_messages,
|
||||
)
|
||||
|
|
@ -2723,7 +2725,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
|
|||
if compaction_blocks is not None:
|
||||
provider_specific_fields["compaction_blocks"] = compaction_blocks
|
||||
|
||||
return provider_specific_fields
|
||||
refusal_fields: Final = refusal_provider_specific_fields(
|
||||
REFUSAL_STOP_FIELDS.validate_python(completion_response)
|
||||
)
|
||||
return {**provider_specific_fields, **refusal_fields}
|
||||
|
||||
def transform_parsed_response(
|
||||
self,
|
||||
|
|
@ -2808,7 +2813,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
|
|||
_message.provider_specific_fields = provider_specific_fields
|
||||
|
||||
if json_mode_message is not None:
|
||||
refusal_fields: Final = refusal_provider_specific_fields(
|
||||
REFUSAL_STOP_FIELDS.validate_python(completion_response)
|
||||
)
|
||||
completion_response["stop_reason"] = "stop"
|
||||
if refusal_fields:
|
||||
json_mode_message.provider_specific_fields = dict(refusal_fields)
|
||||
_message = json_mode_message
|
||||
|
||||
model_response.choices[0].message = _message
|
||||
|
|
|
|||
|
|
@ -8,11 +8,12 @@ import re
|
|||
from collections.abc import Mapping, MutableMapping, Sequence
|
||||
from datetime import datetime, timezone
|
||||
from types import MappingProxyType
|
||||
from typing import Any, ClassVar, Final, Literal, TypeVar
|
||||
from typing import Annotated, Any, ClassVar, Final, Literal, TypeVar
|
||||
from urllib.parse import quote, urlparse
|
||||
|
||||
import httpx
|
||||
from pydantic import ConfigDict, Field, StrictBool, TypeAdapter, ValidationError
|
||||
from pydantic import BeforeValidator, ConfigDict, Field, StrictBool, TypeAdapter, ValidationError
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
import litellm
|
||||
from litellm.constants import (
|
||||
|
|
@ -2150,3 +2151,35 @@ def create_anthropic_model_list_response(
|
|||
"first_id": data[0]["id"] if data else None,
|
||||
"last_id": data[-1]["id"] if data else None,
|
||||
}
|
||||
|
||||
|
||||
class RefusalStopFields(TypedDict, total=False):
|
||||
stop_reason: ReadOnly[object]
|
||||
stop_details: ReadOnly[object]
|
||||
|
||||
|
||||
def _is_dict(value: object) -> bool:
|
||||
return isinstance(value, dict)
|
||||
|
||||
|
||||
def _no_stop_fields() -> Mapping[str, object]:
|
||||
return {}
|
||||
|
||||
|
||||
def _dict_or_empty(value: object) -> object:
|
||||
return value if _is_dict(value) else _no_stop_fields()
|
||||
|
||||
|
||||
REFUSAL_STOP_FIELDS: Final[TypeAdapter[RefusalStopFields]] = TypeAdapter(
|
||||
Annotated[RefusalStopFields, BeforeValidator(_dict_or_empty)]
|
||||
)
|
||||
|
||||
|
||||
def refusal_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]:
|
||||
stop_reason: Final = message.get("stop_reason")
|
||||
stop_details: Final = message.get("stop_details")
|
||||
if stop_details is not None:
|
||||
return {"stop_reason": stop_reason, "stop_details": stop_details}
|
||||
if stop_reason == "refusal":
|
||||
return {"stop_reason": stop_reason}
|
||||
return {}
|
||||
|
|
|
|||
533
tests/integration/_support/anthropic_refusal.py
Normal file
533
tests/integration/_support/anthropic_refusal.py
Normal file
|
|
@ -0,0 +1,533 @@
|
|||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from itertools import chain
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final, TypeAlias
|
||||
|
||||
import yaml
|
||||
from integration._support.anthropic_thinking import (
|
||||
JSON_LIST,
|
||||
JSON_OBJECT,
|
||||
Event,
|
||||
marker_of,
|
||||
stream_reply,
|
||||
streams,
|
||||
)
|
||||
from integration._support.client import eventually
|
||||
from integration._support.wire import Reply, Request, Wire
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
from redis import Redis
|
||||
|
||||
DOCS_REFUSAL_URL: Final = "https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback#refusal-response"
|
||||
DOCS_STREAMING_REFUSAL_URL: Final = (
|
||||
"https://platform.claude.com/docs/en/test-and-evaluate/strengthen-guardrails/handle-streaming-refusals"
|
||||
)
|
||||
|
||||
# Copied verbatim from DOCS_REFUSAL_URL and DOCS_STREAMING_REFUSAL_URL on 2026-10-12.
|
||||
DOCS_REFUSAL: Final[Event] = {
|
||||
"id": "msg_01XFUDYJgAACzvnptvVoYEL",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-fable-5",
|
||||
"content": [],
|
||||
"stop_reason": "refusal",
|
||||
"stop_details": {
|
||||
"type": "refusal",
|
||||
"category": "cyber",
|
||||
"explanation": "This request was declined because it could enable cyber harm.",
|
||||
},
|
||||
"usage": {"input_tokens": 412, "output_tokens": 0},
|
||||
}
|
||||
DOCS_PARTIAL_REFUSAL: Final[Event] = {
|
||||
"role": "assistant",
|
||||
"content": [{"type": "text", "text": "Hello.."}],
|
||||
"stop_reason": "refusal",
|
||||
"stop_details": {
|
||||
"type": "refusal",
|
||||
"category": "cyber",
|
||||
"explanation": "This request was declined because it could enable cyber harm.",
|
||||
},
|
||||
}
|
||||
|
||||
# Captured from api.anthropic.com (claude-sonnet-5-5, prompt "Reply with the single word: ready") on 2026-10-13.
|
||||
CAPTURED_TURN: Final[Event] = {
|
||||
"model": "claude-sonnet-5-5",
|
||||
"id": "msg_011CfyXFGze1cDkk8o7gDRRS",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "text", "text": "ready"}],
|
||||
"container": None,
|
||||
"stop_reason": "end_turn",
|
||||
"stop_sequence": None,
|
||||
"stop_details": None,
|
||||
"usage": {
|
||||
"input_tokens": 18,
|
||||
"cache_creation_input_tokens": 0,
|
||||
"cache_read_input_tokens": 0,
|
||||
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0},
|
||||
"output_tokens": 4,
|
||||
"output_tokens_details": {"thinking_tokens": 0},
|
||||
"service_tier": "standard",
|
||||
"inference_geo": "global",
|
||||
},
|
||||
"diagnostics": None,
|
||||
}
|
||||
CAPTURED_EVENTS: Final[tuple[Event, ...]] = (
|
||||
{
|
||||
"type": "message_start",
|
||||
"message": {
|
||||
"model": "claude-sonnet-5-5",
|
||||
"id": "msg_011CfyXFLwXUxsp7nNAm3gcu",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [],
|
||||
"container": None,
|
||||
"stop_reason": None,
|
||||
"stop_sequence": None,
|
||||
"stop_details": None,
|
||||
"usage": {
|
||||
"input_tokens": 18,
|
||||
"cache_creation_input_tokens": 0,
|
||||
"cache_read_input_tokens": 0,
|
||||
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0},
|
||||
"output_tokens": 4,
|
||||
"service_tier": "standard",
|
||||
"inference_geo": "global",
|
||||
},
|
||||
"diagnostics": None,
|
||||
},
|
||||
},
|
||||
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
|
||||
{"type": "ping"},
|
||||
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "ready"}},
|
||||
{"type": "content_block_stop", "index": 0},
|
||||
{
|
||||
"type": "message_delta",
|
||||
"delta": {"stop_reason": "end_turn", "stop_sequence": None, "stop_details": None, "container": None},
|
||||
"usage": {
|
||||
"input_tokens": 18,
|
||||
"cache_creation_input_tokens": 0,
|
||||
"cache_read_input_tokens": 0,
|
||||
"output_tokens": 4,
|
||||
"output_tokens_details": {"thinking_tokens": 0},
|
||||
},
|
||||
},
|
||||
{"type": "message_stop"},
|
||||
)
|
||||
|
||||
DOCS_STOP_DETAILS: Final = DOCS_REFUSAL["stop_details"]
|
||||
REFUSAL_FIELDS: Final[Event] = {"stop_reason": "refusal", "stop_details": DOCS_STOP_DETAILS}
|
||||
STOP_FIELDS: Final = ("stop_reason", "stop_details")
|
||||
LONG_STOP_DETAILS: Final = "d" * 5120
|
||||
MALFORMED_STOP_DETAILS: Final[Mapping[str, JsonValue]] = MappingProxyType(
|
||||
{"int": 7, "list": ["refusal"], "empty": "", "long": LONG_STOP_DETAILS}
|
||||
)
|
||||
_FIXTURE: Final = re.compile(r"refusal-fixture-([a-z]+)")
|
||||
|
||||
Location: TypeAlias = tuple[str | int, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Fixture:
|
||||
status: int
|
||||
body: bytes
|
||||
events: tuple[Event, ...] = ()
|
||||
|
||||
|
||||
def _text_block(index: int, block: JsonValue) -> tuple[Event, ...]:
|
||||
text: Final = JSON_OBJECT.validate_python(block)["text"]
|
||||
return (
|
||||
{"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}},
|
||||
{"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": text}},
|
||||
{"type": "content_block_stop", "index": index},
|
||||
)
|
||||
|
||||
|
||||
def refusal_events(message: Mapping[str, JsonValue]) -> tuple[Event, ...]:
|
||||
usage: Final = JSON_OBJECT.validate_python(message["usage"])
|
||||
start: Final[Event] = {
|
||||
"type": "message_start",
|
||||
"message": {
|
||||
"id": message["id"],
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": message["model"],
|
||||
"content": [],
|
||||
"stop_reason": None,
|
||||
"stop_sequence": None,
|
||||
"stop_details": None,
|
||||
"usage": usage,
|
||||
},
|
||||
}
|
||||
blocks: Final = chain.from_iterable(
|
||||
_text_block(index, block) for index, block in enumerate(JSON_LIST.validate_python(message["content"]))
|
||||
)
|
||||
delta: Final[Event] = {
|
||||
"type": "message_delta",
|
||||
"delta": {
|
||||
"stop_reason": message["stop_reason"],
|
||||
"stop_sequence": None,
|
||||
"stop_details": message["stop_details"],
|
||||
},
|
||||
"usage": {name: usage[name] for name in ("output_tokens", "output_tokens_details") if name in usage},
|
||||
}
|
||||
return (start, *blocks, delta, {"type": "message_stop"})
|
||||
|
||||
|
||||
def _without_stop_details(value: Mapping[str, JsonValue]) -> Event:
|
||||
return {name: item for name, item in value.items() if name != "stop_details"}
|
||||
|
||||
|
||||
def _captured_event(event: Event, stop_details: JsonValue, *, keep: bool) -> Event:
|
||||
if event["type"] == "message_start":
|
||||
message: Final = JSON_OBJECT.validate_python(event["message"])
|
||||
return {**event, "message": message if keep else _without_stop_details(message)}
|
||||
if event["type"] != "message_delta":
|
||||
return event
|
||||
delta: Final = JSON_OBJECT.validate_python(event["delta"])
|
||||
replaced: Final = {**delta, "stop_details": stop_details} if keep else _without_stop_details(delta)
|
||||
return {**event, "delta": replaced}
|
||||
|
||||
|
||||
def _captured(stop_details: JsonValue, *, keep: bool = True) -> Fixture:
|
||||
body: Final = {**CAPTURED_TURN, "stop_details": stop_details} if keep else _without_stop_details(CAPTURED_TURN)
|
||||
return Fixture(
|
||||
200,
|
||||
json.dumps(body).encode(),
|
||||
tuple(_captured_event(event, stop_details, keep=keep) for event in CAPTURED_EVENTS),
|
||||
)
|
||||
|
||||
|
||||
def _duplicated_stop_details() -> bytes:
|
||||
pair: Final = f'"stop_details": {json.dumps(DOCS_STOP_DETAILS)}'
|
||||
return json.dumps(DOCS_REFUSAL).replace(pair, f"{pair}, {pair}").encode()
|
||||
|
||||
|
||||
def _anthropic_error(kind: str, message: str) -> bytes:
|
||||
return json.dumps({"type": "error", "error": {"type": kind, "message": message}}).encode()
|
||||
|
||||
|
||||
_PARTIAL: Final[Event] = {**DOCS_REFUSAL, **DOCS_PARTIAL_REFUSAL}
|
||||
_REASONED: Final[Event] = {
|
||||
**_PARTIAL,
|
||||
"usage": {"input_tokens": 412, "output_tokens": 10, "output_tokens_details": {"thinking_tokens": 4}},
|
||||
}
|
||||
|
||||
JSON_TOOL_NAME: Final = "json_tool_call"
|
||||
JSON_ANSWER_SCHEMA: Final[Event] = {
|
||||
"type": "object",
|
||||
"properties": {"answer": {"type": "string"}},
|
||||
"required": ["answer"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
JSON_ANSWER_FORMAT: Final[Event] = {
|
||||
"type": "json_schema",
|
||||
"json_schema": {"name": "answer", "schema": JSON_ANSWER_SCHEMA, "strict": True},
|
||||
}
|
||||
DOCS_PARTIAL_TEXT: Final = "Hello.."
|
||||
_DECLARED_INPUT: Final[Event] = {"answer": DOCS_PARTIAL_TEXT}
|
||||
_ENVELOPED_INPUT: Final[Event] = {"parameters": _DECLARED_INPUT}
|
||||
DECLARED_PARTIAL_JSON: Final = '{"answer": "Hello..'
|
||||
ENVELOPED_PARTIAL_JSON: Final = '{"parameters": {"answer": "Hello..'
|
||||
_JSON_TOOL_USE_ID: Final = "toolu_01RefusalFixtureJsonTool"
|
||||
|
||||
|
||||
def _json_tool_refusal(tool_input: Event, partial_json: str) -> Fixture:
|
||||
message: Final[Event] = {
|
||||
**DOCS_REFUSAL,
|
||||
"content": [{"type": "tool_use", "id": _JSON_TOOL_USE_ID, "name": JSON_TOOL_NAME, "input": tool_input}],
|
||||
}
|
||||
start, delta, stop = refusal_events(DOCS_REFUSAL)
|
||||
tool_block: Final[tuple[Event, ...]] = (
|
||||
{
|
||||
"type": "content_block_start",
|
||||
"index": 0,
|
||||
"content_block": {"type": "tool_use", "id": _JSON_TOOL_USE_ID, "name": JSON_TOOL_NAME, "input": {}},
|
||||
},
|
||||
{
|
||||
"type": "content_block_delta",
|
||||
"index": 0,
|
||||
"delta": {"type": "input_json_delta", "partial_json": partial_json},
|
||||
},
|
||||
{"type": "content_block_stop", "index": 0},
|
||||
)
|
||||
return Fixture(200, json.dumps(message).encode(), (start, *tool_block, delta, stop))
|
||||
|
||||
|
||||
FIXTURES: Final[Mapping[str, Fixture]] = MappingProxyType(
|
||||
{
|
||||
"refusal": Fixture(200, json.dumps(DOCS_REFUSAL).encode(), refusal_events(DOCS_REFUSAL)),
|
||||
"partial": Fixture(200, json.dumps(_PARTIAL).encode(), refusal_events(_PARTIAL)),
|
||||
"reasoned": Fixture(200, json.dumps(_REASONED).encode(), refusal_events(_REASONED)),
|
||||
"jsontool": _json_tool_refusal(_DECLARED_INPUT, DECLARED_PARTIAL_JSON),
|
||||
"jsonheld": _json_tool_refusal(_ENVELOPED_INPUT, ENVELOPED_PARTIAL_JSON),
|
||||
"duplicate": Fixture(200, _duplicated_stop_details()),
|
||||
"null": _captured(None),
|
||||
"missing": _captured(None, keep=False),
|
||||
**{name: _captured(value) for name, value in MALFORMED_STOP_DETAILS.items()},
|
||||
"nondict": Fixture(200, b"[]"),
|
||||
"unauthorized": Fixture(401, _anthropic_error("authentication_error", "invalid x-api-key")),
|
||||
"invalid": Fixture(400, _anthropic_error("invalid_request_error", "max_tokens: Field required")),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def refusal_prompt(fixture: str, marker: str) -> str:
|
||||
return f"refusal-fixture-{fixture} marker-{marker}"
|
||||
|
||||
|
||||
CITED_FIELDS: Final[Event] = {"citations": ["https://example.com/source"]}
|
||||
_PLAIN_DELTA: Final[Event] = {"role": "assistant", "content": "ready"}
|
||||
_CITED_DELTA: Final[Event] = {**_PLAIN_DELTA, "provider_specific_fields": CITED_FIELDS}
|
||||
_OPENAI_USAGE: Final[Event] = {
|
||||
"prompt_tokens": 5,
|
||||
"completion_tokens": 7,
|
||||
"total_tokens": 12,
|
||||
"completion_tokens_details": {"reasoning_tokens": 4, "text_tokens": 3},
|
||||
}
|
||||
UNVALIDATED_USAGE_DETAILS: Final[Mapping[str, JsonValue]] = MappingProxyType(
|
||||
{
|
||||
"text": {"reasoning_tokens": "abc"},
|
||||
"fraction": {"reasoning_tokens": 1.5},
|
||||
"object": {"reasoning_tokens": {"a": 1}},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class OpenAIFixture:
|
||||
first_delta: Event
|
||||
usage: tuple[Event, ...] = ()
|
||||
|
||||
|
||||
_OPENAI_FIXTURES: Final[Mapping[str, OpenAIFixture]] = MappingProxyType(
|
||||
{
|
||||
"plain": OpenAIFixture(_PLAIN_DELTA),
|
||||
"cited": OpenAIFixture(_CITED_DELTA),
|
||||
"plainusage": OpenAIFixture(_PLAIN_DELTA, (_OPENAI_USAGE,)),
|
||||
"citedusage": OpenAIFixture(_CITED_DELTA, (_OPENAI_USAGE,)),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def openai_id(marker: str) -> str:
|
||||
return f"chatcmpl-{marker}"
|
||||
|
||||
|
||||
def _openai_frame(marker: str, body: Event) -> bytes:
|
||||
chunk: Final[Event] = {
|
||||
"id": openai_id(marker),
|
||||
"object": "chat.completion.chunk",
|
||||
"created": 1760000000,
|
||||
"model": "gpt-4o-mini",
|
||||
**body,
|
||||
}
|
||||
return b"data: " + json.dumps(chunk).encode() + b"\n\n"
|
||||
|
||||
|
||||
def _openai_choice(delta: Event, finish_reason: str | None) -> Event:
|
||||
return {"choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}]}
|
||||
|
||||
|
||||
def _openai_stream(name: str, request: Request) -> Reply:
|
||||
assert streams(request), request.body
|
||||
marker: Final = marker_of(request)
|
||||
fixture: Final = _OPENAI_FIXTURES[name]
|
||||
return Reply(
|
||||
content_type="text/event-stream",
|
||||
chunks=(
|
||||
_openai_frame(marker, _openai_choice(fixture.first_delta, None)),
|
||||
_openai_frame(marker, _openai_choice({"content": " now"}, "stop")),
|
||||
*(_openai_frame(marker, {"choices": [], "usage": usage}) for usage in fixture.usage),
|
||||
b"data: [DONE]\n\n",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
found: Final[Sequence[str]] = _FIXTURE.findall(request.body.decode())
|
||||
assert found, request.body
|
||||
if request.target.endswith("/chat/completions"):
|
||||
return _openai_stream(found[-1], request)
|
||||
fixture: Final = FIXTURES[found[-1]]
|
||||
if fixture.status != 200 or not fixture.events or not streams(request):
|
||||
return Reply(status=fixture.status, body=fixture.body)
|
||||
return stream_reply(request, fixture.events)
|
||||
|
||||
|
||||
def _stop_fields(fields: JsonValue) -> Event:
|
||||
if not isinstance(fields, dict):
|
||||
return {}
|
||||
return {name: fields[name] for name in STOP_FIELDS if name in fields}
|
||||
|
||||
|
||||
def relayed(value: JsonValue, location: Location = ()) -> tuple[tuple[Location, Event], ...]:
|
||||
if isinstance(value, list):
|
||||
return tuple(chain.from_iterable(relayed(item, (*location, index)) for index, item in enumerate(value)))
|
||||
if not isinstance(value, dict):
|
||||
return ()
|
||||
found: Final = _stop_fields(value.get("provider_specific_fields"))
|
||||
here: Final = (((*location, "provider_specific_fields"), found),) if found else ()
|
||||
return (*here, *chain.from_iterable(relayed(item, (*location, name)) for name, item in value.items()))
|
||||
|
||||
|
||||
_CACHE_KEY: Final = re.compile(r"[0-9a-f]{64}")
|
||||
_STORED: Final = TypeAdapter(bytes | None)
|
||||
STALE_ID_SUFFIX: Final = "-stale"
|
||||
|
||||
|
||||
def _stored(store: Redis, key: str) -> bytes:
|
||||
if not _CACHE_KEY.fullmatch(key) or store.type(key) != b"string":
|
||||
return b""
|
||||
return _STORED.validate_python(store.get(key)) or b""
|
||||
|
||||
|
||||
def _entry_key(store: Redis, needle: str) -> str | None:
|
||||
keys: Final = (raw.decode() for raw in store.scan_iter(count=1000))
|
||||
return next((key for key in keys if needle.encode() in _stored(store, key)), None)
|
||||
|
||||
|
||||
def _with_message_fields(response: Event, fields: JsonValue) -> Event:
|
||||
choice: Final = JSON_OBJECT.validate_python(JSON_LIST.validate_python(response["choices"])[0])
|
||||
message: Final = JSON_OBJECT.validate_python(choice["message"])
|
||||
return {**response, "choices": [{**choice, "message": {**message, "provider_specific_fields": fields}}]}
|
||||
|
||||
|
||||
def _with_usage_details(response: Event, details: JsonValue) -> Event:
|
||||
usage: Final = JSON_OBJECT.validate_python(response["usage"])
|
||||
return {**response, "usage": {**usage, "completion_tokens_details": details}}
|
||||
|
||||
|
||||
def _rewrite_cached_response(needle: str, rewrite: Callable[[Event], Event]) -> str:
|
||||
store: Final = Redis(host=os.environ["REDIS_HOST"], port=int(os.environ["REDIS_PORT"]))
|
||||
key: Final = eventually(lambda: _entry_key(store, needle), lambda found: found is not None)
|
||||
assert key is not None, needle
|
||||
entry: Final = JSON_OBJECT.validate_json(_stored(store, key))
|
||||
stored: Final = entry["response"]
|
||||
encoded: Final = isinstance(stored, str)
|
||||
response: Final = (
|
||||
JSON_OBJECT.validate_json(stored) if isinstance(stored, str) else JSON_OBJECT.validate_python(stored)
|
||||
)
|
||||
rewritten: Final = {**rewrite(response), "id": f"{response['id']}{STALE_ID_SUFFIX}"}
|
||||
store.set(key, json.dumps({**entry, "response": json.dumps(rewritten) if encoded else rewritten}), ex=60)
|
||||
return str(rewritten["id"])
|
||||
|
||||
|
||||
def rewrite_cached_message_fields(needle: str, fields: JsonValue) -> str:
|
||||
return _rewrite_cached_response(needle, lambda response: _with_message_fields(response, fields))
|
||||
|
||||
|
||||
def rewrite_cached_usage_details(needle: str, details: JsonValue) -> str:
|
||||
return _rewrite_cached_response(needle, lambda response: _with_usage_details(response, details))
|
||||
|
||||
|
||||
def key_hash(key: str) -> str:
|
||||
return hashlib.sha256(key.encode()).hexdigest()
|
||||
|
||||
|
||||
def _key_hash_of(event: Mapping[str, JsonValue]) -> JsonValue:
|
||||
return JSON_OBJECT.validate_python(event.get("metadata") or {}).get("user_api_key_hash")
|
||||
|
||||
|
||||
class Ledger:
|
||||
def __init__(self, wire: Wire) -> None:
|
||||
self.wire: Final = wire
|
||||
self.batches: tuple[Request, ...] = ()
|
||||
|
||||
def events(self) -> tuple[Event, ...]:
|
||||
self.batches = (*self.batches, *self.wire.drain())
|
||||
decoded: Final = chain.from_iterable(JSON_LIST.validate_json(batch.body) for batch in self.batches)
|
||||
return tuple(JSON_OBJECT.validate_python(event) for event in decoded)
|
||||
|
||||
def for_key(self, key: str) -> tuple[Event, ...]:
|
||||
hashed: Final = key_hash(key)
|
||||
return tuple(event for event in self.events() if _key_hash_of(event) == hashed)
|
||||
|
||||
def await_key(self, key: str, count: int = 1, seconds: float = 30) -> tuple[Event, ...]:
|
||||
return eventually(lambda: self.for_key(key), lambda found: len(found) >= count, seconds=seconds)
|
||||
|
||||
|
||||
_HIDDEN_PARAMS_PROBE: Final = """
|
||||
import httpx
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
|
||||
|
||||
class HiddenParamsProbe(CustomLogger):
|
||||
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
|
||||
payload = kwargs.get("standard_logging_object") or {{}}
|
||||
hidden = getattr(response_obj, "_hidden_params", None) or {{}}
|
||||
event = {{
|
||||
"metadata": {{"user_api_key_hash": (payload.get("metadata") or {{}}).get("user_api_key_hash")}},
|
||||
"response_id": getattr(response_obj, "id", None),
|
||||
"received_model_id": hidden.get("received_model_id"),
|
||||
}}
|
||||
async with httpx.AsyncClient(trust_env=False) as client:
|
||||
await client.post({url!r}, json=[event])
|
||||
|
||||
|
||||
instance = HiddenParamsProbe()
|
||||
"""
|
||||
|
||||
|
||||
def hidden_params_probe(directory: Path, url: str) -> str:
|
||||
(directory / "hidden_params_probe.py").write_text(_HIDDEN_PARAMS_PROBE.format(url=url))
|
||||
return "hidden_params_probe.instance"
|
||||
|
||||
|
||||
def sink_config(
|
||||
directory: Path,
|
||||
sinks: Mapping[str, str],
|
||||
*,
|
||||
retries: int = 0,
|
||||
extra_callbacks: Sequence[str] = (),
|
||||
store_prompts: bool = False,
|
||||
) -> Path:
|
||||
shared: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()))
|
||||
settings: Final = {
|
||||
**JSON_OBJECT.validate_python(shared["litellm_settings"]),
|
||||
"callbacks": [*sinks, *extra_callbacks],
|
||||
"DEFAULT_FLUSH_INTERVAL_SECONDS": 1,
|
||||
}
|
||||
callback_settings: Final = {
|
||||
name: {
|
||||
"callback_type": "generic_api",
|
||||
"endpoint": url,
|
||||
"headers": {"x-refusal-sink": name},
|
||||
"max_retries": retries,
|
||||
"retry_delay": 0.5,
|
||||
}
|
||||
for name, url in sinks.items()
|
||||
}
|
||||
general_settings: Final = {
|
||||
**JSON_OBJECT.validate_python(shared["general_settings"]),
|
||||
"store_prompts_in_spend_logs": store_prompts,
|
||||
}
|
||||
path: Final = directory / "anthropic-refusal-sinks.yaml"
|
||||
path.write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
**shared,
|
||||
"general_settings": general_settings,
|
||||
"litellm_settings": settings,
|
||||
"callback_settings": callback_settings,
|
||||
}
|
||||
)
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
def frames(text: str) -> tuple[Event, ...]:
|
||||
return tuple(
|
||||
JSON_OBJECT.validate_json(line.removeprefix("data: "))
|
||||
for line in text.splitlines()
|
||||
if line.startswith("data: {")
|
||||
)
|
||||
|
||||
|
||||
def frame_types(events: Sequence[Mapping[str, JsonValue]]) -> tuple[JsonValue, ...]:
|
||||
return tuple(event.get("type") for event in events)
|
||||
|
|
@ -336,6 +336,29 @@ def flip_entry_to_empty(store: Redis, key: str, field: str) -> str:
|
|||
return stale_id
|
||||
|
||||
|
||||
def rewrite_usage_details(store: Redis, key: str, details: JsonValue) -> str:
|
||||
"""Rewrite the stored usage's ``completion_tokens_details`` to ``details`` and stamp the id ``-stale``, keeping the
|
||||
encoding, and return the stale id."""
|
||||
raw: Final = store.get(key)
|
||||
assert raw is not None, key
|
||||
entry, is_json = _decode_entry(raw)
|
||||
response: Final = entry["response"]
|
||||
stored: Final = object_value(json.loads(response) if isinstance(response, str) else response)
|
||||
stale_id: Final = f"{string_value(stored['id'])}{STALE_ID_SUFFIX}"
|
||||
usage: Final = object_value(stored["usage"])
|
||||
rewritten_response: Final[dict[str, JsonValue]] = {
|
||||
**stored,
|
||||
"usage": {**usage, "completion_tokens_details": details},
|
||||
"id": stale_id,
|
||||
}
|
||||
rewritten: Final[dict[str, JsonValue]] = {
|
||||
**entry,
|
||||
"response": json.dumps(rewritten_response) if isinstance(response, str) else rewritten_response,
|
||||
}
|
||||
store.set(key, json.dumps(rewritten) if is_json else str(rewritten), ex=STALE_ENTRY_TTL_SECONDS)
|
||||
return stale_id
|
||||
|
||||
|
||||
def chat_body(model: str, text: str, **extra: JsonValue) -> dict[str, JsonValue]:
|
||||
return {"model": model, "messages": [{"role": "user", "content": text}], **extra}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,337 @@
|
|||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from collections import Counter
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from contextlib import ExitStack
|
||||
from dataclasses import dataclass
|
||||
from itertools import chain
|
||||
from pathlib import Path
|
||||
from queue import SimpleQueue
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Literal, TypeAlias
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
import httpx
|
||||
import psutil
|
||||
import pytest
|
||||
import yaml
|
||||
from integration._support.anthropic_refusal import (
|
||||
REFUSAL_FIELDS,
|
||||
Ledger,
|
||||
Location,
|
||||
refusal_prompt,
|
||||
relayed,
|
||||
respond,
|
||||
sink_config,
|
||||
)
|
||||
from integration._support.anthropic_thinking import JSON_OBJECT, MODEL, Event, chunks_of
|
||||
from integration._support.client import Gateway, eventually, held
|
||||
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
|
||||
from integration._support.timing import assert_not_delayed_by
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
|
||||
pytestmark: Final = [
|
||||
pytest.mark.timeout(2 * graceful_stop_seconds() + 120),
|
||||
pytest.mark.xdist_group("anthropic_refusal_stop_details_chaos"),
|
||||
]
|
||||
|
||||
_MODEL_NAME: Final = "anthropic-refusal-chaos"
|
||||
_SLOW_SINK_SECONDS: Final = 4
|
||||
_SINK: Final = "refusal_chaos_sink"
|
||||
_STARTED_WORKER: Final = re.compile(r"Started server process \[(\d+)\]")
|
||||
_MARKER: Final = re.compile(r"marker-([0-9a-f]{32})")
|
||||
_CHAT_FIELDS: Final[Location] = ("choices", 0, "message", "provider_specific_fields")
|
||||
_DELTA_FIELDS: Final[Location] = ("choices", 0, "delta", "provider_specific_fields")
|
||||
_PATHS: Final[Mapping[str, str]] = MappingProxyType(
|
||||
{"chat": "/v1/chat/completions", "messages": "/v1/messages", "responses": "/v1/responses"}
|
||||
)
|
||||
|
||||
Surface: TypeAlias = Literal["chat", "messages", "responses"]
|
||||
_RecordProperty: TypeAlias = Callable[[str, object], None]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Call:
|
||||
surface: Surface
|
||||
stream: bool
|
||||
marker: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Served:
|
||||
call: _Call
|
||||
status: int
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Held:
|
||||
release: threading.Event
|
||||
markers: SimpleQueue[str]
|
||||
|
||||
def respond(self, request: Request) -> Reply:
|
||||
self.markers.put(_markers_in(request.body.decode())[0])
|
||||
assert self.release.wait(timeout=90), "The burst was never released"
|
||||
return respond(request)
|
||||
|
||||
|
||||
def _markers_in(text: str) -> tuple[str, ...]:
|
||||
return tuple(dict.fromkeys(_MARKER.findall(text)))
|
||||
|
||||
|
||||
def _calls(count: int) -> tuple[_Call, ...]:
|
||||
surfaces: Final[tuple[Surface, ...]] = ("chat", "messages", "responses")
|
||||
return tuple(
|
||||
_Call(surfaces[index % 3], stream=(index // 3) % 2 == 0, marker=uuid.uuid4().hex) for index in range(count)
|
||||
)
|
||||
|
||||
|
||||
def _body(call: _Call) -> Event:
|
||||
text: Final = refusal_prompt("refusal", call.marker)
|
||||
shared: Final[Event] = {"model": _MODEL_NAME, "stream": call.stream, "cache": {"no-cache": True}}
|
||||
match call.surface:
|
||||
case "chat":
|
||||
return {**shared, "messages": [{"role": "user", "content": text}], "max_tokens": 64}
|
||||
case "messages":
|
||||
return {**shared, "messages": [{"role": "user", "content": text}], "max_tokens": 64}
|
||||
case "responses":
|
||||
return {**shared, "input": text, "max_output_tokens": 64}
|
||||
|
||||
|
||||
async def _send(client: httpx.AsyncClient, key: str, call: _Call) -> _Served:
|
||||
try:
|
||||
response: Final = await client.post(
|
||||
_PATHS[call.surface], json=_body(call), headers={"Authorization": f"Bearer {key}"}
|
||||
)
|
||||
except httpx.TransportError as error:
|
||||
return _Served(call, 0, repr(error))
|
||||
return _Served(call, response.status_code, response.text)
|
||||
|
||||
|
||||
async def _burst(gateway: Gateway, calls: Sequence[_Call]) -> tuple[_Served, ...]:
|
||||
async with httpx.AsyncClient(base_url=str(gateway.client.base_url), timeout=120, trust_env=False) as client:
|
||||
return tuple(await asyncio.gather(*(_send(client, gateway.key, call) for call in calls)))
|
||||
|
||||
|
||||
def _chaos_config(directory: Path, upstream: Wire, sink: str, *, retries: int) -> Path:
|
||||
base: Final = sink_config(directory, {_SINK: sink}, retries=retries)
|
||||
config: Final = JSON_OBJECT.validate_python(yaml.safe_load(base.read_text()))
|
||||
model_list: Final = [
|
||||
{
|
||||
"model_name": _MODEL_NAME,
|
||||
"litellm_params": {"model": f"anthropic/{MODEL}", "api_base": upstream.url, "api_key": "scripted-key"},
|
||||
}
|
||||
]
|
||||
path: Final = directory / "anthropic-refusal-chaos.yaml"
|
||||
path.write_text(yaml.safe_dump({**config, "model_list": model_list}))
|
||||
return path
|
||||
|
||||
|
||||
def _ledger_events(ledgers: Sequence[Ledger]) -> tuple[Event, ...]:
|
||||
return tuple(chain.from_iterable(ledger.events() for ledger in ledgers))
|
||||
|
||||
|
||||
def _marker_counts(ledgers: Sequence[Ledger], markers: frozenset[str]) -> Mapping[str, int]:
|
||||
found: Final = chain.from_iterable(_markers_in(json.dumps(event)) for event in _ledger_events(ledgers))
|
||||
return MappingProxyType(Counter(marker for marker in found if marker in markers))
|
||||
|
||||
|
||||
def _events_for(ledgers: Sequence[Ledger], marker: str) -> tuple[Event, ...]:
|
||||
return tuple(event for event in _ledger_events(ledgers) if marker in json.dumps(event))
|
||||
|
||||
|
||||
def _assert_answered(item: _Served) -> None:
|
||||
assert item.status == 200, (item.call, item.text)
|
||||
if item.call.surface != "chat":
|
||||
return
|
||||
if item.call.stream:
|
||||
found: Final = tuple(chain.from_iterable(relayed(chunk) for chunk in chunks_of(item.text)))
|
||||
assert found == ((_DELTA_FIELDS, REFUSAL_FIELDS),), item.text
|
||||
return
|
||||
assert relayed(JSON_OBJECT.validate_json(item.text)) == ((_CHAT_FIELDS, REFUSAL_FIELDS),), item.text
|
||||
|
||||
|
||||
def _assert_logged_chat_relays(ledgers: Sequence[Ledger], calls: Sequence[_Call]) -> None:
|
||||
for call in calls:
|
||||
if call.surface != "chat":
|
||||
continue
|
||||
(event,) = _events_for(ledgers, call.marker)
|
||||
assert relayed(event["response"]) == ((_CHAT_FIELDS, REFUSAL_FIELDS),), event
|
||||
|
||||
|
||||
def _worker_pids(owned: OwnedProxy) -> tuple[int, ...]:
|
||||
return eventually(
|
||||
lambda: tuple(int(pid) for pid in _STARTED_WORKER.findall(owned.log.read_text())),
|
||||
lambda pids: len(pids) == 2,
|
||||
seconds=30,
|
||||
)
|
||||
|
||||
|
||||
def _open_upstream_connections(pid: int, upstream: str) -> int:
|
||||
port: Final = urlsplit(upstream).port
|
||||
return sum(
|
||||
1
|
||||
for connection in psutil.Process(pid).net_connections(kind="tcp")
|
||||
if connection.status == psutil.CONN_ESTABLISHED and connection.raddr and connection.raddr.port == port
|
||||
)
|
||||
|
||||
|
||||
def _port(wire: Wire) -> int:
|
||||
port: Final = urlsplit(wire.url).port
|
||||
assert port is not None, wire.url
|
||||
return port
|
||||
|
||||
|
||||
def _accept(_request: Request) -> Reply:
|
||||
return Reply()
|
||||
|
||||
|
||||
async def _timed_chat(gateway: Gateway) -> float:
|
||||
started: Final = time.monotonic()
|
||||
(served,) = await _burst(gateway, (_Call("chat", stream=False, marker=uuid.uuid4().hex),))
|
||||
_assert_answered(served)
|
||||
return time.monotonic() - started
|
||||
|
||||
|
||||
async def test_sink_outage_mid_burst_delivers_every_refusal_event_once(gateway: Gateway, tmp_path: Path) -> None:
|
||||
calls: Final = _calls(30)
|
||||
markers: Final = frozenset(call.marker for call in calls)
|
||||
with wire_server(respond) as upstream, ExitStack() as first_sink:
|
||||
sink: Final = first_sink.enter_context(wire_server(_accept))
|
||||
first: Final = Ledger(sink)
|
||||
config: Final = _chaos_config(tmp_path, upstream, sink.url, retries=6)
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
|
||||
_worker_pids(owned)
|
||||
served_first: Final = await _burst(owned.gateway, calls[:10])
|
||||
await asyncio.to_thread(
|
||||
eventually, lambda: sum(_marker_counts((first,), markers).values()), lambda total: total >= 1, 30
|
||||
)
|
||||
first.events()
|
||||
first_sink.close()
|
||||
served_during: Final = await _burst(owned.gateway, calls[10:20])
|
||||
with wire_server(_accept, port=_port(sink)) as restarted:
|
||||
second: Final = Ledger(restarted)
|
||||
served_after: Final = await _burst(owned.gateway, calls[20:])
|
||||
for item in (*served_first, *served_during, *served_after):
|
||||
_assert_answered(item)
|
||||
counts: Final = await asyncio.to_thread(
|
||||
held,
|
||||
lambda: _marker_counts((first, second), markers),
|
||||
lambda found: set(found) == markers and set(found.values()) == {1},
|
||||
holding=3,
|
||||
seconds=60,
|
||||
)
|
||||
assert sum(counts.values()) == 30, counts
|
||||
_assert_logged_chat_relays((first, second), calls)
|
||||
assert len(upstream.drain()) == 30
|
||||
|
||||
|
||||
async def test_slow_sink_neither_deadlocks_nor_uploads_a_refusal_event_twice(gateway: Gateway, tmp_path: Path) -> None:
|
||||
calls: Final = _calls(30)
|
||||
markers: Final = frozenset(call.marker for call in calls)
|
||||
|
||||
def slow(_request: Request) -> Reply:
|
||||
time.sleep(_SLOW_SINK_SECONDS)
|
||||
return Reply()
|
||||
|
||||
with wire_server(respond) as upstream, wire_server(slow) as sink:
|
||||
ledger: Final = Ledger(sink)
|
||||
config: Final = _chaos_config(tmp_path, upstream, sink.url, retries=0)
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
|
||||
_worker_pids(owned)
|
||||
control: Final = await _timed_chat(owned.gateway)
|
||||
served: Final = await _burst(owned.gateway, calls)
|
||||
for item in served:
|
||||
_assert_answered(item)
|
||||
follow_up: Final = await _timed_chat(owned.gateway)
|
||||
assert_not_delayed_by((follow_up,), control=(control,), injected_delay=_SLOW_SINK_SECONDS)
|
||||
counts: Final = await asyncio.to_thread(
|
||||
held,
|
||||
lambda: _marker_counts((ledger,), markers),
|
||||
lambda found: set(found) == markers and set(found.values()) == {1},
|
||||
holding=5,
|
||||
seconds=90,
|
||||
)
|
||||
assert sum(counts.values()) == 30, counts
|
||||
_assert_logged_chat_relays((ledger,), calls)
|
||||
|
||||
|
||||
async def test_worker_sigkill_mid_burst_leaves_the_sibling_relaying_and_logging_once(
|
||||
gateway: Gateway, tmp_path: Path
|
||||
) -> None:
|
||||
calls: Final = _calls(20)
|
||||
hold: Final = _Held(threading.Event(), SimpleQueue())
|
||||
with wire_server(hold.respond) as upstream, wire_server(_accept) as sink:
|
||||
ledger: Final = Ledger(sink)
|
||||
config: Final = _chaos_config(tmp_path, upstream, sink.url, retries=0)
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as owned:
|
||||
workers: Final = _worker_pids(owned)
|
||||
burst: Final = asyncio.create_task(_burst(owned.gateway, calls))
|
||||
await asyncio.to_thread(eventually, hold.markers.qsize, lambda size: size == 20, 60)
|
||||
held_by: Final = MappingProxyType({pid: _open_upstream_connections(pid, upstream.url) for pid in workers})
|
||||
assert sum(held_by.values()) == 20, held_by
|
||||
victim_pid, survivor_pid = sorted(workers, key=held_by.__getitem__)
|
||||
with owned.replacing_worker(victim_pid) as kill:
|
||||
kill.freeze()
|
||||
await asyncio.to_thread(kill)
|
||||
hold.release.set()
|
||||
served: Final = await burst
|
||||
assert held_by[survivor_pid] >= 10, held_by
|
||||
completed: Final = tuple(item for item in served if item.status == 200)
|
||||
assert len(completed) == held_by[survivor_pid], (held_by, [item.status for item in served])
|
||||
for item in completed:
|
||||
_assert_answered(item)
|
||||
follow_up: Final = _Call("chat", stream=True, marker=uuid.uuid4().hex)
|
||||
(answered,) = await _burst(owned.gateway, (follow_up,))
|
||||
_assert_answered(answered)
|
||||
survivors: Final = frozenset((*(item.call.marker for item in completed), follow_up.marker))
|
||||
counts: Final = await asyncio.to_thread(
|
||||
held,
|
||||
lambda: _marker_counts((ledger,), frozenset(call.marker for call in (*calls, follow_up))),
|
||||
lambda found: set(found) == survivors and set(found.values()) == {1},
|
||||
holding=3,
|
||||
seconds=60,
|
||||
)
|
||||
assert sum(counts.values()) == len(survivors), counts
|
||||
_assert_logged_chat_relays((ledger,), tuple(item.call for item in completed))
|
||||
|
||||
|
||||
async def test_proxy_restart_mid_burst_never_logs_a_refusal_twice(
|
||||
gateway: Gateway, tmp_path: Path, record_property: _RecordProperty
|
||||
) -> None:
|
||||
calls: Final = _calls(20)
|
||||
markers: Final = frozenset(call.marker for call in calls)
|
||||
hold: Final = _Held(threading.Event(), SimpleQueue())
|
||||
with wire_server(hold.respond) as upstream, wire_server(_accept) as sink:
|
||||
ledger: Final = Ledger(sink)
|
||||
config: Final = _chaos_config(tmp_path, upstream, sink.url, retries=0)
|
||||
with owned_proxy_process(gateway, tmp_path, {}, config=config, workers=2) as first:
|
||||
_worker_pids(first)
|
||||
burst: Final = asyncio.create_task(_burst(first.gateway, calls))
|
||||
await asyncio.to_thread(eventually, hold.markers.qsize, lambda size: size == 20, 60)
|
||||
first.begin_shutdown()
|
||||
hold.release.set()
|
||||
served: Final = await burst
|
||||
for item in served:
|
||||
_assert_answered(item)
|
||||
second_directory: Final = tmp_path / "second"
|
||||
second_directory.mkdir()
|
||||
with owned_proxy_process(gateway, second_directory, {}, config=config, workers=2) as second:
|
||||
follow_up: Final = _Call("chat", stream=False, marker=uuid.uuid4().hex)
|
||||
(answered,) = await _burst(second.gateway, (follow_up,))
|
||||
_assert_answered(answered)
|
||||
await asyncio.to_thread(
|
||||
eventually,
|
||||
lambda: _marker_counts((ledger,), frozenset({follow_up.marker})),
|
||||
lambda found: found[follow_up.marker] == 1,
|
||||
30,
|
||||
)
|
||||
counts: Final = _marker_counts((ledger,), markers)
|
||||
assert set(counts.values()) <= {1}, counts
|
||||
_assert_logged_chat_relays((ledger,), tuple(call for call in calls if call.marker in counts))
|
||||
record_property("served", len(served))
|
||||
record_property("landed", sum(counts.values()))
|
||||
File diff suppressed because it is too large
Load diff
|
|
@ -9,17 +9,23 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
from itertools import chain
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from pydantic import JsonValue, TypeAdapter
|
||||
|
||||
import litellm
|
||||
from litellm.caching.caching import Cache
|
||||
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
|
||||
from litellm.types.caching import LiteLLMCacheType
|
||||
from litellm.types.utils import ModelResponse, TextCompletionResponse
|
||||
from tests.integration._support.client import JSON_OBJECT
|
||||
from tests.integration.caching.response_cache_case import (
|
||||
CHAT,
|
||||
CHAT_CHUNK,
|
||||
CHAT_MODEL,
|
||||
CHAT_STREAM,
|
||||
PROVIDER_KEY,
|
||||
TEXT,
|
||||
TEXT_MODEL,
|
||||
|
|
@ -31,8 +37,10 @@ from tests.integration.caching.response_cache_case import (
|
|||
prompt,
|
||||
redis_store,
|
||||
rescript,
|
||||
rewrite_usage_details,
|
||||
scripted_in_process,
|
||||
)
|
||||
from tests.integration.cost_calculation.cost_tracking_case import SseResponse
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -123,3 +131,77 @@ def test_sync_text_completion_with_empty_choices_reaches_upstream_again() -> Non
|
|||
hit: Final = _text_completion(handle.api_base(), text)
|
||||
assert hit.id == refilled.id, hit.model_dump_json()
|
||||
assert upstream.calls(handle.scenario_id) == 2
|
||||
|
||||
|
||||
_REASONING_USAGE_FRAME: Final = (
|
||||
CHAT_CHUNK + '[],"usage":{"prompt_tokens":5,"completion_tokens":7,"total_tokens":12,'
|
||||
'"completion_tokens_details":{"reasoning_tokens":4}}}'
|
||||
)
|
||||
_REASONING_STREAM: Final = SseResponse(
|
||||
content_type="text/event-stream",
|
||||
frames=(*CHAT_STREAM.frames[:-1], _REASONING_USAGE_FRAME, CHAT_STREAM.frames[-1]),
|
||||
)
|
||||
_JSON_LIST: Final = TypeAdapter(list[JsonValue])
|
||||
_UNVALIDATED_USAGE_DETAILS: Final[dict[str, JsonValue]] = {
|
||||
"text": {"reasoning_tokens": "abc"},
|
||||
"fraction": {"reasoning_tokens": 1.5},
|
||||
"object": {"reasoning_tokens": {"a": 1}},
|
||||
}
|
||||
|
||||
|
||||
def _stream(api_base: str, text: str) -> tuple[dict[str, JsonValue], ...]:
|
||||
stream: Final = litellm.completion(
|
||||
model=CHAT_MODEL,
|
||||
messages=[{"role": "user", "content": text}],
|
||||
api_base=api_base,
|
||||
api_key=PROVIDER_KEY,
|
||||
stream=True,
|
||||
stream_options={"include_usage": True},
|
||||
)
|
||||
assert isinstance(stream, CustomStreamWrapper), type(stream)
|
||||
return tuple(JSON_OBJECT.validate_python(chunk.model_dump(mode="json")) for chunk in stream)
|
||||
|
||||
|
||||
def _replayed(chunks: tuple[dict[str, JsonValue], ...]) -> tuple[frozenset[JsonValue], str, tuple[JsonValue, ...]]:
|
||||
choices: Final = tuple(
|
||||
JSON_OBJECT.validate_python(choice)
|
||||
for choice in chain.from_iterable(_JSON_LIST.validate_python(chunk.get("choices") or []) for chunk in chunks)
|
||||
)
|
||||
text: Final = "".join(str(JSON_OBJECT.validate_python(choice["delta"]).get("content") or "") for choice in choices)
|
||||
usages: Final = tuple(JSON_OBJECT.validate_python(chunk["usage"]) for chunk in chunks if chunk.get("usage"))
|
||||
reasoning: Final = tuple(
|
||||
JSON_OBJECT.validate_python(usage.get("completion_tokens_details") or {}).get("reasoning_tokens")
|
||||
for usage in usages
|
||||
)
|
||||
return frozenset(chunk["id"] for chunk in chunks), text, reasoning
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("sdk_cache")
|
||||
def test_sync_stream_cache_hit_replays_the_usage_details_of_the_miss() -> None:
|
||||
upstream: Final = Upstream()
|
||||
text: Final = prompt()
|
||||
with scripted_in_process(_REASONING_STREAM) as handle, redis_store() as store:
|
||||
first: Final = _replayed(_stream(handle.api_base(), text))
|
||||
(first_id,) = first[0]
|
||||
assert first[1:] == ("streamed response", (4,)), first
|
||||
await_entry(store, str(first_id))
|
||||
hit: Final = _replayed(_stream(handle.api_base(), text))
|
||||
assert hit == first, hit
|
||||
assert upstream.calls(handle.scenario_id) == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("sdk_cache")
|
||||
@pytest.mark.parametrize("variant", tuple(_UNVALIDATED_USAGE_DETAILS))
|
||||
def test_sync_stream_cache_hit_skips_cached_usage_details_that_fail_validation(variant: str) -> None:
|
||||
upstream: Final = Upstream()
|
||||
text: Final = prompt()
|
||||
with scripted_in_process(_REASONING_STREAM) as handle, redis_store() as store:
|
||||
first: Final = _replayed(_stream(handle.api_base(), text))
|
||||
(first_id,) = first[0]
|
||||
assert first[1:] == ("streamed response", (4,)), first
|
||||
stale_id: Final = rewrite_usage_details(
|
||||
store, await_entry(store, str(first_id)), _UNVALIDATED_USAGE_DETAILS[variant]
|
||||
)
|
||||
hit: Final = _replayed(_stream(handle.api_base(), text))
|
||||
assert hit == (frozenset({stale_id}), "streamed response", (0,)), hit
|
||||
assert upstream.calls(handle.scenario_id) == 1
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ content frame, restoring per-token cadence on cache hits.
|
|||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
|
||||
_split_assembled_content_for_replay,
|
||||
convert_to_streaming_response,
|
||||
|
|
@ -87,6 +88,71 @@ async def test_async_yields_multiple_content_chunks_with_lossless_join():
|
|||
assert reassembled == text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("content", ["", "Hello world! How are you?"], ids=["single_frame", "sliced"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_and_sync_replays_carry_the_cached_provider_specific_fields(content):
|
||||
provider_specific_fields = {"stop_reason": "refusal", "stop_details": {"category": "cyber"}}
|
||||
|
||||
def payload():
|
||||
cached = _async_payload(content=content)
|
||||
cached["choices"][0]["message"]["provider_specific_fields"] = provider_specific_fields
|
||||
return cached
|
||||
|
||||
async_chunks = await _collect_async(payload())
|
||||
sync_chunks = list(convert_to_streaming_response(payload()))
|
||||
|
||||
assert [c.choices[0].delta.provider_specific_fields for c in async_chunks] == [
|
||||
c.choices[0].delta.provider_specific_fields for c in sync_chunks
|
||||
]
|
||||
assert async_chunks[0].choices[0].delta.provider_specific_fields == provider_specific_fields
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider_fields", [{"thought_signatures": ["sig"]}, None], ids=["provider_fields", "text"])
|
||||
@pytest.mark.parametrize("replay", ["async", "sync"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_replayed_stream_keeps_the_cached_completion_tokens_details(replay, provider_fields):
|
||||
cached = _async_payload()
|
||||
cached["choices"][0]["message"]["provider_specific_fields"] = provider_fields
|
||||
cached["usage"]["completion_tokens_details"] = {"reasoning_tokens": 4, "text_tokens": 3}
|
||||
|
||||
chunks = await _collect_async(cached) if replay == "async" else list(convert_to_streaming_response(cached))
|
||||
rebuilt = litellm.stream_chunk_builder(chunks=chunks)
|
||||
|
||||
assert rebuilt.choices[0].message.get("provider_specific_fields") == provider_fields
|
||||
assert rebuilt.usage.completion_tokens_details.reasoning_tokens == 4
|
||||
assert rebuilt.usage.completion_tokens_details.text_tokens == 3
|
||||
|
||||
|
||||
@pytest.mark.parametrize("cached_fields", [7, [1, 2], "", "x" * 5000], ids=["int", "list", "empty", "long"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_async_replay_ignores_cached_provider_fields_that_are_not_an_object(cached_fields):
|
||||
cached = _async_payload()
|
||||
cached["choices"][0]["message"]["provider_specific_fields"] = cached_fields
|
||||
|
||||
chunks = await _collect_async(cached)
|
||||
|
||||
assert "".join(c.choices[0].delta.content or "" for c in chunks) == "Hello world! How are you?"
|
||||
assert all(c.choices[0].delta.provider_specific_fields is None for c in chunks)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"details",
|
||||
[{"reasoning_tokens": "abc"}, {"reasoning_tokens": 1.5}, {"reasoning_tokens": {"a": 1}}],
|
||||
ids=["text", "fraction", "object"],
|
||||
)
|
||||
@pytest.mark.parametrize("replay", ["async", "sync"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_replayed_stream_skips_cached_completion_tokens_details_that_do_not_validate(replay, details):
|
||||
cached = _async_payload()
|
||||
cached["usage"]["completion_tokens_details"] = details
|
||||
|
||||
chunks = await _collect_async(cached) if replay == "async" else list(convert_to_streaming_response(cached))
|
||||
rebuilt = litellm.stream_chunk_builder(chunks=chunks)
|
||||
|
||||
assert rebuilt.choices[0].message.content == "Hello world! How are you?"
|
||||
assert rebuilt.usage.completion_tokens == 7
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_finish_reason_only_on_last_chunk():
|
||||
chunks = await _collect_async(_async_payload())
|
||||
|
|
@ -204,9 +270,7 @@ def test_sync_multi_chunk_and_lossless_join():
|
|||
|
||||
|
||||
def test_sync_delta_and_choice_metadata_only_on_first_chunk():
|
||||
thinking_blocks = [
|
||||
{"type": "thinking", "thinking": "cached thinking", "signature": "sig"}
|
||||
]
|
||||
thinking_blocks = [{"type": "thinking", "thinking": "cached thinking", "signature": "sig"}]
|
||||
payload = {
|
||||
"id": "chatcmpl-test-sync",
|
||||
"object": "chat.completion",
|
||||
|
|
|
|||
|
|
@ -732,6 +732,42 @@ def test_set_response_id_propagation_valid_to_invalid(
|
|||
assert initialized_custom_stream_wrapper.response_id == "first-valid-id"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"last_delta",
|
||||
[
|
||||
Delta(content="Hello"),
|
||||
Delta(provider_specific_fields={"stop_reason": "refusal"}),
|
||||
],
|
||||
ids=["content", "provider_specific_fields"],
|
||||
)
|
||||
def test_logged_response_keeps_received_model_id_when_the_finish_chunk_is_synthesized(
|
||||
logging_obj: Logging, last_delta: Delta
|
||||
):
|
||||
upstream_id: Final = "chatcmpl-upstream-id"
|
||||
wrapper: Final = CustomStreamWrapper(
|
||||
completion_stream=ModelResponseListIterator(
|
||||
model_responses=[
|
||||
ModelResponseStream(
|
||||
id=upstream_id,
|
||||
model="gpt-4o-mini",
|
||||
choices=[StreamingChoices(index=0, delta=last_delta, finish_reason="content_filter")],
|
||||
)
|
||||
]
|
||||
),
|
||||
model="gpt-4o-mini",
|
||||
logging_obj=logging_obj,
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
emitted: Final = list(wrapper)
|
||||
logged: Final = litellm.stream_chunk_builder(chunks=wrapper.chunks)
|
||||
|
||||
assert len(emitted) == 2
|
||||
assert [chunk._hidden_params.get("received_model_id") for chunk in emitted] == [upstream_id, upstream_id]
|
||||
assert logged is not None
|
||||
assert logged._hidden_params.get("received_model_id") == upstream_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_completion_start_time(logging_obj: Logging):
|
||||
"""Test that the start time is set correctly"""
|
||||
|
|
|
|||
19
tests/unit/llms/anthropic/anthropic_docs_refusal.py
Normal file
19
tests/unit/llms/anthropic/anthropic_docs_refusal.py
Normal file
|
|
@ -0,0 +1,19 @@
|
|||
from typing import Final
|
||||
|
||||
ANTHROPIC_DOCS_REFUSAL_URL: Final = "https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback#refusal-response"
|
||||
|
||||
# Copied verbatim from ANTHROPIC_DOCS_REFUSAL_URL on 2026-10-12. Streams carry the same stop_details on message_delta.
|
||||
ANTHROPIC_DOCS_REFUSAL_MESSAGE: Final = {
|
||||
"id": "msg_01XFUDYJgAACzvnptvVoYEL",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-fable-5",
|
||||
"content": [],
|
||||
"stop_reason": "refusal",
|
||||
"stop_details": {
|
||||
"type": "refusal",
|
||||
"category": "cyber",
|
||||
"explanation": "This request was declined because it could enable cyber harm.",
|
||||
},
|
||||
"usage": {"input_tokens": 412, "output_tokens": 0},
|
||||
}
|
||||
|
|
@ -22,6 +22,7 @@ from litellm.types.llms.openai import (
|
|||
)
|
||||
from litellm.types.responses.main import OutputCodeInterpreterCall
|
||||
from litellm.types.utils import ChatCompletionDeltaToolCall, ModelResponseStream
|
||||
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -3366,3 +3367,50 @@ def test_anthropic_streaming_invalid_key_maps_to_authentication_error() -> None:
|
|||
assert exc_info.value.status_code == 401
|
||||
assert exc_info.value.llm_provider == "anthropic"
|
||||
assert route.calls.last.request.headers["x-api-key"] == "bad-anthropic-key"
|
||||
|
||||
|
||||
def _message_stream_events(stop_reason: str, stop_details: dict | None) -> tuple[dict, ...]:
|
||||
return (
|
||||
{
|
||||
"type": "message_start",
|
||||
"message": {
|
||||
"id": ANTHROPIC_DOCS_REFUSAL_MESSAGE["id"],
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": ANTHROPIC_DOCS_REFUSAL_MESSAGE["model"],
|
||||
"content": [],
|
||||
"usage": {"input_tokens": 412, "output_tokens": 0},
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "message_delta",
|
||||
"delta": {"stop_reason": stop_reason, "stop_sequence": None, "stop_details": stop_details},
|
||||
"usage": {"output_tokens": 0},
|
||||
},
|
||||
{"type": "message_stop"},
|
||||
)
|
||||
|
||||
|
||||
def _assemble_stream(events: tuple[dict, ...]) -> litellm.ModelResponse:
|
||||
iterator: Final = ModelResponseIterator(streaming_response=MagicMock(), sync_stream=True, json_mode=False)
|
||||
chunks: Final = [iterator.chunk_parser(event) for event in events]
|
||||
return litellm.stream_chunk_builder(chunks=chunks, messages=[{"role": "user", "content": "hi"}])
|
||||
|
||||
|
||||
def test_streamed_refusal_stop_details_reach_the_assembled_message():
|
||||
stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
assembled: Final = _assemble_stream(_message_stream_events("refusal", stop_details))
|
||||
|
||||
assert assembled.choices[0].finish_reason == "content_filter"
|
||||
provider_fields: Final = assembled.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert provider_fields["stop_details"] == stop_details
|
||||
|
||||
|
||||
def test_streamed_end_turn_carries_no_stop_details():
|
||||
assembled: Final = _assemble_stream(_message_stream_events("end_turn", None))
|
||||
|
||||
assert assembled.choices[0].finish_reason == "stop"
|
||||
provider_fields: Final = assembled.choices[0].message.provider_specific_fields or {}
|
||||
assert "stop_details" not in provider_fields
|
||||
assert "stop_reason" not in provider_fields
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ from litellm.llms.vertex_ai.vertex_ai_partner_models.anthropic.transformation im
|
|||
from litellm.types.llms.anthropic import ANTHROPIC_BETA_HEADER_VALUES
|
||||
from litellm.types.llms.openai import ChatCompletionToolCallFunctionChunk
|
||||
from litellm.types.utils import ChatCompletionToolCallChunk, ModelResponse, ServerToolUse, Usage
|
||||
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
|
||||
|
||||
def test_response_format_transformation_unit_test():
|
||||
|
|
@ -9082,3 +9083,93 @@ def test_completion_surfaces_response_citations_in_provider_specific_fields(resp
|
|||
assert response.choices[0].message.provider_specific_fields["citations"] == [
|
||||
[{**citation, "supported_text": "the grass is green"}]
|
||||
]
|
||||
|
||||
|
||||
def _refusal_message_body(stop_reason: str, stop_details: dict | None) -> dict:
|
||||
return {**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "stop_reason": stop_reason, "stop_details": stop_details}
|
||||
|
||||
|
||||
def _transform_message_body(body: dict) -> litellm.ModelResponse:
|
||||
return AnthropicConfig().transform_parsed_response(
|
||||
completion_response=body,
|
||||
raw_response=httpx.Response(status_code=200, headers={}),
|
||||
model_response=litellm.ModelResponse(),
|
||||
json_mode=False,
|
||||
prefix_prompt=None,
|
||||
)
|
||||
|
||||
|
||||
def test_refusal_stop_details_kept_in_provider_specific_fields():
|
||||
stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
result: Final = _transform_message_body(ANTHROPIC_DOCS_REFUSAL_MESSAGE)
|
||||
|
||||
assert result.choices[0].finish_reason == "content_filter"
|
||||
provider_fields: Final = result.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert provider_fields["stop_details"] == stop_details
|
||||
assert result._hidden_params["provider_specific_fields"]["stop_details"] == stop_details
|
||||
|
||||
|
||||
def test_refusal_without_stop_details_keeps_the_native_stop_reason():
|
||||
result: Final = _transform_message_body(_refusal_message_body("refusal", None))
|
||||
|
||||
provider_fields: Final = result.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert "stop_details" not in provider_fields
|
||||
|
||||
|
||||
def test_end_turn_response_carries_no_stop_details():
|
||||
result: Final = _transform_message_body(_refusal_message_body("end_turn", None))
|
||||
|
||||
assert result.choices[0].finish_reason == "stop"
|
||||
provider_fields: Final = result.choices[0].message.provider_specific_fields
|
||||
assert "stop_details" not in provider_fields
|
||||
assert "stop_reason" not in provider_fields
|
||||
|
||||
|
||||
def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict:
|
||||
return {
|
||||
**_refusal_message_body(stop_reason, stop_details),
|
||||
"content": [
|
||||
{"type": "tool_use", "id": "toolu_json", "name": RESPONSE_FORMAT_TOOL_NAME, "input": {"answer": "Hello"}}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _transform_json_mode_body(body: dict) -> litellm.ModelResponse:
|
||||
return AnthropicConfig().transform_parsed_response(
|
||||
completion_response=body,
|
||||
raw_response=httpx.Response(status_code=200, headers={}),
|
||||
model_response=litellm.ModelResponse(),
|
||||
json_mode=True,
|
||||
prefix_prompt=None,
|
||||
)
|
||||
|
||||
|
||||
def test_json_mode_refusal_keeps_stop_details_on_the_message():
|
||||
stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
result: Final = _transform_json_mode_body(_json_mode_message_body("refusal", stop_details))
|
||||
|
||||
message: Final = result.choices[0].message
|
||||
assert json.loads(message.content) == {"answer": "Hello"}
|
||||
assert message.provider_specific_fields == {"stop_reason": "refusal", "stop_details": stop_details}
|
||||
|
||||
|
||||
def test_json_mode_tool_use_turn_adds_no_stop_fields_to_the_message():
|
||||
result: Final = _transform_json_mode_body(_json_mode_message_body("tool_use", None))
|
||||
|
||||
message: Final = result.choices[0].message
|
||||
assert json.loads(message.content) == {"answer": "Hello"}
|
||||
assert not message.provider_specific_fields
|
||||
|
||||
|
||||
def test_non_dict_upstream_body_still_transforms_without_refusal_fields() -> None:
|
||||
result = AnthropicConfig().transform_parsed_response(
|
||||
completion_response=MagicMock(),
|
||||
raw_response=httpx.Response(200, headers={}),
|
||||
model_response=litellm.ModelResponse(),
|
||||
json_mode=False,
|
||||
prefix_prompt=None,
|
||||
)
|
||||
provider_specific_fields = result.choices[0].message.provider_specific_fields or {}
|
||||
assert "stop_details" not in provider_specific_fields
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import json
|
|||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from typing import Final
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
|
@ -17,6 +17,7 @@ from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transfor
|
|||
)
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
|
||||
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
|
||||
ONE_PIXEL_PNG = base64.b64decode(
|
||||
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
|
||||
|
|
@ -1689,3 +1690,23 @@ class TestBedrockAnthropicCombinedRegressions:
|
|||
user_msg = result["messages"][0]
|
||||
has_cache_control = any(isinstance(c, dict) and "cache_control" in c for c in user_msg["content"])
|
||||
assert has_cache_control
|
||||
|
||||
|
||||
def test_refusal_stop_details_kept_through_bedrock_invoke_transform():
|
||||
body: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
result: Final = AmazonAnthropicClaudeConfig().transform_response(
|
||||
model="anthropic.claude-sonnet-5-5-v1:0",
|
||||
raw_response=httpx.Response(status_code=200, headers={}, json=body),
|
||||
model_response=litellm.ModelResponse(),
|
||||
logging_obj=MagicMock(),
|
||||
request_data={},
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
encoding=None,
|
||||
)
|
||||
|
||||
assert result.choices[0].finish_reason == "content_filter"
|
||||
provider_fields: Final = result.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert provider_fields["stop_details"] == ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import copy
|
|||
import json
|
||||
from collections.abc import Mapping
|
||||
from typing import Final
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
|
@ -19,6 +20,7 @@ from litellm.llms.vertex_ai.vertex_ai_partner_models.anthropic.transformation im
|
|||
)
|
||||
from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
|
||||
from litellm.types.llms.vertex_ai import VertexPartnerProvider
|
||||
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
|
@ -1125,3 +1127,23 @@ def test_chat_flagged_model_replays_a_byte_identical_prefix_around_a_mid_convers
|
|||
_assert_prefix_stable(requests)
|
||||
assert [m["role"] for m in requests[1]["messages"]] == ["user", "assistant", "user", "system"]
|
||||
assert [m["role"] for m in requests[2]["messages"]] == ["user", "assistant", "user", "system", "assistant", "user"]
|
||||
|
||||
|
||||
def test_refusal_stop_details_kept_through_vertex_anthropic_transform():
|
||||
body: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
result: Final = VertexAIAnthropicConfig().transform_response(
|
||||
model="claude-sonnet-5-5",
|
||||
raw_response=httpx.Response(status_code=200, headers={}, json=body),
|
||||
model_response=litellm.ModelResponse(),
|
||||
logging_obj=MagicMock(),
|
||||
request_data={},
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
encoding=None,
|
||||
)
|
||||
|
||||
assert result.choices[0].finish_reason == "content_filter"
|
||||
provider_fields: Final = result.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert provider_fields["stop_details"] == ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ from litellm.llms.anthropic.pass_through.stream_assembly import extract_sse_data
|
|||
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
|
||||
AnthropicPassthroughLoggingHandler,
|
||||
)
|
||||
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
|
||||
|
||||
|
||||
async def _drain_tasks():
|
||||
|
|
@ -2951,3 +2952,58 @@ def test_get_user_from_metadata(end_user_id):
|
|||
)
|
||||
|
||||
assert response == "test"
|
||||
|
||||
|
||||
def _anthropic_sse_stream(stop_reason: str, stop_details: dict | None) -> list[str]:
|
||||
events: Final = (
|
||||
(
|
||||
"message_start",
|
||||
{
|
||||
"type": "message_start",
|
||||
"message": {
|
||||
"id": ANTHROPIC_DOCS_REFUSAL_MESSAGE["id"],
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": ANTHROPIC_DOCS_REFUSAL_MESSAGE["model"],
|
||||
"content": [],
|
||||
"usage": {"input_tokens": 412, "output_tokens": 0},
|
||||
},
|
||||
},
|
||||
),
|
||||
(
|
||||
"message_delta",
|
||||
{
|
||||
"type": "message_delta",
|
||||
"delta": {"stop_reason": stop_reason, "stop_sequence": None, "stop_details": stop_details},
|
||||
"usage": {"output_tokens": 0},
|
||||
},
|
||||
),
|
||||
("message_stop", {"type": "message_stop"}),
|
||||
)
|
||||
return [f"event: {name}\ndata: {json.dumps(data)}\n\n" for name, data in events]
|
||||
|
||||
|
||||
def test_streamed_messages_refusal_stop_details_on_the_logged_response(mock_logging_obj):
|
||||
stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
|
||||
logged: Final = AnthropicPassthroughLoggingHandler.build_complete_streaming_response(
|
||||
all_chunks=_anthropic_sse_stream("refusal", stop_details),
|
||||
litellm_logging_obj=mock_logging_obj,
|
||||
model="claude-sonnet-5-5",
|
||||
)
|
||||
|
||||
assert logged.choices[0].finish_reason == "content_filter"
|
||||
provider_fields: Final = logged.choices[0].message.provider_specific_fields
|
||||
assert provider_fields["stop_reason"] == "refusal"
|
||||
assert provider_fields["stop_details"] == stop_details
|
||||
|
||||
|
||||
def test_streamed_messages_end_turn_logs_no_stop_details(mock_logging_obj):
|
||||
logged: Final = AnthropicPassthroughLoggingHandler.build_complete_streaming_response(
|
||||
all_chunks=_anthropic_sse_stream("end_turn", None),
|
||||
litellm_logging_obj=mock_logging_obj,
|
||||
model="claude-sonnet-5-5",
|
||||
)
|
||||
|
||||
assert logged.choices[0].finish_reason == "stop"
|
||||
provider_fields: Final = logged.choices[0].message.provider_specific_fields or {}
|
||||
assert "stop_details" not in provider_fields
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue