feat(sap): synchronize many features such as cache_control, reasoning_effort, thinking

This commit is contained in:
Yamac Ay 2026-07-17 19:25:28 +02:00
parent 7c8364c991
commit f1e7d04b54
No known key found for this signature in database
GPG key ID: D113B438819CA628
3 changed files with 647 additions and 19 deletions

View file

@ -2,7 +2,7 @@ import warnings
from enum import Enum
from typing import Literal, Union
from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic import BaseModel, Field, field_validator, model_serializer, model_validator
def validate_different_content(v: str | dict | list) -> str:
@ -24,9 +24,25 @@ def validate_different_content(v: str | dict | list) -> str:
raise ValueError("Content must be a string")
class CacheControl(BaseModel):
type: Literal["ephemeral"]
class TextContent(BaseModel):
type_: Literal["text"] = Field(default="text", alias="type")
text: str
cache_control: CacheControl | None = None
def model_dump(self, **kwargs) -> dict: # mutable-ok: pydantic override; wire serialization output
kwargs["exclude_none"] = True
return super().model_dump(**kwargs)
@model_serializer(mode="wrap")
def _serialize(self, handler, info) -> dict: # mutable-ok: pydantic serializer contract requires bare dict return
result = handler(self)
if result.get("cache_control") is None:
result.pop("cache_control", None)
return result
class ImageURLContent(BaseModel):
@ -88,9 +104,7 @@ class SAPMessage(BaseModel):
"""
role: Literal["system", "developer"] = "system"
content: str
_content_validator = field_validator("content", mode="before")(validate_different_content)
content: list[TextContent] | str # mutable-ok: pydantic field; list[TextContent] carries cache_control natively
class SAPUserMessage(BaseModel):
@ -184,13 +198,15 @@ class Template(BaseModel):
template: list[ChatMessage]
defaults: dict[str, str] | None = None
response_format: ResponseFormat | ResponseFormatJSONSchema | None = None
tools: list[ChatCompletionTool] | None = None
tools: list[dict] | None = None # mutable-ok: already-validated dicts passed to wire; list preserves insertion order, dict preserves extension fields like cache_control
class LLMModelDetails(BaseModel):
name: str
version: str = "latest"
params: dict | None = None
timeout: int | None = None
max_retries: int | None = Field(default=None, ge=0, le=5)
class PromptTemplatingModuleConfig(BaseModel):

View file

@ -4,6 +4,7 @@ Translate from OpenAI's `/v1/chat/completions` to SAP Generative AI Hub's Orches
from collections.abc import AsyncIterator, Iterator
from functools import cached_property
import re
from typing import (
TYPE_CHECKING,
Any,
@ -46,19 +47,92 @@ from .models import (
_SAP_MODEL_PARAMS_EXCLUDED_KEYS: frozenset[str] = frozenset(
{
"tools",
"tool_choice",
"stream_options",
"fallback_sap_modules",
"placeholder_values",
"model_version",
"timeout",
"max_retries",
}
)
# ---------------------------------------------------------------------------
# SAP capability registry
# ---------------------------------------------------------------------------
# Models that accept reasoning_effort / thinking parameters on SAP GenAI Hub.
_REASONING_MODELS: re.Pattern[str] = re.compile(
r"^(?:anthropic--claude-(?:4(?:\.[5-9])?|3-7)|o\d|gpt-5(?:[.\-]|$)|cohere--\S*reasoning\S*)"
)
def validate_dict(data: dict, model) -> dict:
# Models that support Anthropic-style cache_control on message content parts.
_CACHE_CONTROL_MODELS: re.Pattern[str] = re.compile(r"^anthropic--")
def validate_dict(data: dict, model) -> dict: # mutable-ok: pydantic validation boundary; both input and output are untyped wire dicts
return model(**data).model_dump(by_alias=True, exclude_unset=True)
def _validate_tool(
tool: dict, # mutable-ok: untyped tool dict from litellm boundary
) -> dict: # mutable-ok: wire serialization helper; dict is the required output shape for JSON encoding
"""Validate a tool definition against ChatCompletionTool and preserve cache_control.
cache_control is an Anthropic prompt-caching extension that sits on the tool
object itself (not inside `function`). ChatCompletionTool does not declare it
as a field because its model_dump forces exclude_unset=False for FunctionTool
defaults -- adding cache_control there would emit `cache_control: null` for every
tool that omits it, which the API rejects. We therefore validate the known schema
and re-attach the extension field explicitly.
"""
result = validate_dict(tool, ChatCompletionTool)
if "cache_control" in tool and tool["cache_control"] is not None:
result["cache_control"] = tool["cache_control"]
return result
def _fold_message_cache_control(message: dict) -> dict: # mutable-ok: message dicts are untyped at the litellm boundary
"""Fold a message-level cache_control onto the content block.
litellm's cache_control_injection_points hook places cache_control on the
message dict itself when content is a plain string. SAPMessage has no such
field, so it would be silently dropped. Convert to a single-element content
list so the marker reaches the wire payload.
"""
cc = message.get("cache_control")
if cc is None:
return message
content = message.get("content")
if isinstance(content, str):
return { # mutable-ok: ephemeral wire dict; built once and immediately returned to caller
**{k: v for k, v in message.items() if k != "cache_control"}, # mutable-ok: dict comprehension filters one key; returned immediately
"content": [ # mutable-ok: list literal builds the wire content block in one shot
{"type": "text", "text": content, "cache_control": cc}, # mutable-ok: inner dict literal is the wire content block
],
}
# content already a list -- marker is redundant; drop it to avoid duplication
return {k: v for k, v in message.items() if k != "cache_control"} # mutable-ok: dict comprehension filters one key; returned immediately
def _build_model_details(
model_name: str,
model_version: str,
params: dict, # mutable-ok: model params dict forwarded directly to wire payload
timeout: int | None,
max_retries: int | None,
) -> dict: # mutable-ok: wire serialization helper; dict is the required output shape for JSON encoding
"""Build the model dict for the orchestration request, adding optional fields only when set."""
model_details: dict = { # mutable-ok: ephemeral wire dict built in one shot before JSON encoding
"name": model_name,
"params": params,
"version": model_version,
}
if timeout is not None:
model_details["timeout"] = timeout
if max_retries is not None:
model_details["max_retries"] = max_retries
return model_details
def _messages_to_sap_template(messages: list[dict[str, str]]) -> list: # type: ignore[type-arg]
template = []
for message in messages:
@ -69,13 +143,13 @@ def _messages_to_sap_template(messages: list[dict[str, str]]) -> list: # type:
elif message["role"] == "tool":
template.append(validate_dict(message, SAPToolChatMessage))
else:
template.append(validate_dict(message, SAPMessage))
template.append(validate_dict(_fold_message_cache_control(message), SAPMessage))
return template
def _tools_response_format_and_stream(optional_params: dict, model_params: dict) -> tuple[dict, dict, dict]:
tools_ = optional_params.pop("tools", [])
tools_ = [validate_dict(tool, ChatCompletionTool) for tool in tools_]
tools_ = [_validate_tool(tool) for tool in tools_]
tools: dict = {"tools": tools_} if tools_ else {}
response_format = model_params.pop("response_format", {})
@ -209,13 +283,15 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
"temperature",
"top_p",
"tools",
"tool_choice",
"function_call",
"functions",
"extra_headers",
"parallel_tool_calls",
"response_format",
"timeout",
"max_retries",
"model_version",
"user",
]
# Remove response_format for providers that don't support it on SAP GenAI Hub
if (
@ -225,8 +301,8 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
or model == "gpt-4"
):
params.remove("response_format")
if model.startswith("gemini") or model.startswith("amazon"):
params.remove("tool_choice")
if self._sap_supports_reasoning(model):
params.extend(["reasoning_effort", "thinking"])
return params
def validate_environment(
@ -268,9 +344,11 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
params.pop("strict")
model_version = params.pop("model_version", "latest")
timeout = params.pop("timeout", None)
max_retries = params.pop("max_retries", None)
tools_ = params.pop("tools", [])
tools_ = [validate_dict(tool, ChatCompletionTool) for tool in tools_]
tools_ = [_validate_tool(tool) for tool in tools_]
tools = {"tools": tools_} if tools_ else {}
response_format = params.pop("response_format", {})
@ -301,11 +379,7 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
**tools,
**response_format,
},
"model": {
"name": model_name,
"params": params,
"version": model_version,
},
"model": _build_model_details(model_name, model_version, params, timeout, max_retries),
},
**optional_modules,
}
@ -350,7 +424,8 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
fallback_model = modules_dict.pop("model", None)
if fallback_model is None:
raise ValueError("Each entry in `fallback_sap_modules` must include a 'model' key.")
fallback_model = fallback_model.removeprefix("sap/")
if fallback_model.startswith("sap/"):
fallback_model = fallback_model[4:]
fallback_template = modules_dict.pop("messages", [])
modules.append(
@ -408,6 +483,32 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
return response
@staticmethod
def _sap_supports_reasoning(model: str) -> bool:
"""Return True if *model* accepts reasoning_effort / thinking on SAP GenAI Hub."""
return bool(_REASONING_MODELS.match(model))
@staticmethod
def _sap_supports_cache_control(model: str) -> bool:
"""Return True if *model* supports Anthropic-style cache_control content parts."""
return bool(_CACHE_CONTROL_MODELS.match(model))
@staticmethod
def _normalize_gemini_reasoning(final_result: dict) -> None:
"""Coerce Gemini's list-shaped reasoning_content to a plain string in-place.
Gemini models on SAP GenAI Hub return reasoning_content as a list of
{"thought": str, "signature": str} objects. ModelResponse expects a
plain str, so model_validate crashes without this normalisation step.
"""
for choice in final_result.get("choices") or []:
msg = (choice.get("message") or {}) if isinstance(choice, dict) else {}
rc = msg.get("reasoning_content")
if isinstance(rc, list):
msg["reasoning_content"] = (
"\n\n".join(item.get("thought", "") for item in rc if isinstance(item, dict)) or None
)
def _strip_markdown_json(self, response: ModelResponse) -> ModelResponse:
"""Strip markdown code block wrapper from JSON content if present.

View file

@ -639,3 +639,514 @@ class TestSAPTransformationIntegration:
config["config"]["modules"][1]["translation"]["input"]["type"]
== "sap_document_translation"
)
class TestGeminiReasoningNormalization:
"""Unit tests for _normalize_gemini_reasoning.
Verifies that list-shaped reasoning_content from Gemini is coerced to str
before model_validate is called, without touching other shapes.
"""
def _make_final_result(self, reasoning_content):
"""Helper: build a minimal final_result dict with given reasoning_content."""
result = {
"id": "test-id",
"object": "chat.completion",
"model": "gemini-2.0-flash",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello",
},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
}
if reasoning_content is not None:
result["choices"][0]["message"]["reasoning_content"] = reasoning_content
return result
def test_list_shaped_is_joined_to_string(self):
"""Gemini list of thought dicts is joined into a newline-separated string."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
final_result = self._make_final_result(
[
{"thought": "First thought.", "signature": "sig1"},
{"thought": "Second thought.", "signature": "sig2"},
]
)
GenAIHubOrchestrationConfig._normalize_gemini_reasoning(final_result)
assert (
final_result["choices"][0]["message"]["reasoning_content"]
== "First thought.\n\nSecond thought."
)
def test_string_reasoning_content_is_untouched(self):
"""A reasoning_content that is already a str is left as-is."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
final_result = self._make_final_result("already a string")
GenAIHubOrchestrationConfig._normalize_gemini_reasoning(final_result)
assert final_result["choices"][0]["message"]["reasoning_content"] == "already a string"
def test_missing_reasoning_content_is_untouched(self):
"""A message without reasoning_content is left unchanged."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
final_result = self._make_final_result(None)
# reasoning_content key is absent — _make_final_result(None) does not add it
assert "reasoning_content" not in final_result["choices"][0]["message"]
GenAIHubOrchestrationConfig._normalize_gemini_reasoning(final_result)
assert "reasoning_content" not in final_result["choices"][0]["message"]
def test_empty_list_becomes_none(self):
"""An empty list collapses to None so model_validate sees no reasoning."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
final_result = self._make_final_result([])
GenAIHubOrchestrationConfig._normalize_gemini_reasoning(final_result)
assert final_result["choices"][0]["message"]["reasoning_content"] is None
def test_model_validate_succeeds_after_normalization(self):
"""model_validate no longer raises after normalisation."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
from litellm.types.utils import ModelResponse
final_result = self._make_final_result(
[{"thought": "Thinking hard.", "signature": "abc"}]
)
GenAIHubOrchestrationConfig._normalize_gemini_reasoning(final_result)
response = ModelResponse.model_validate(final_result)
assert response.choices[0].message.reasoning_content == "Thinking hard."
class TestReasoningCapability:
"""Unit tests for reasoning_effort / thinking parameter routing.
Verifies that capable models expose the params and that they land in
model.params, while non-capable models have them silently dropped.
"""
def _transform(self, model: str, **kwargs) -> dict:
"""Run transform_request and return the parsed body."""
from unittest.mock import MagicMock
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
logging_obj = MagicMock()
return cfg.transform_request(
model=model,
messages=[{"role": "user", "content": "Hi"}],
optional_params=dict(kwargs),
litellm_params={},
headers={},
)
def _model_params(self, body: dict) -> dict:
return body["config"]["modules"]["prompt_templating"]["model"]["params"]
# --- get_supported_openai_params ---
def test_reasoning_params_exposed_for_claude_3_7(self):
"""reasoning_effort and thinking appear for claude-3-7 models."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
params = cfg.get_supported_openai_params("anthropic--claude-3-7-sonnet")
assert "reasoning_effort" in params
assert "thinking" in params
def test_reasoning_params_exposed_for_claude_4(self):
"""reasoning_effort and thinking appear for claude-4 models."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
params = cfg.get_supported_openai_params("anthropic--claude-4-opus")
assert "reasoning_effort" in params
assert "thinking" in params
def test_reasoning_params_absent_for_gpt4o(self):
"""reasoning_effort and thinking are not exposed for gpt-4o."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
params = cfg.get_supported_openai_params("gpt-4o")
assert "reasoning_effort" not in params
assert "thinking" not in params
# --- transform_request model.params ---
def test_reasoning_effort_lands_in_model_params_for_o3(self):
"""reasoning_effort is forwarded into model.params for o-series models."""
body = self._transform("o3", reasoning_effort="high")
assert self._model_params(body).get("reasoning_effort") == "high"
def test_thinking_lands_in_model_params_for_claude_3_7(self):
"""thinking dict is forwarded into model.params for claude-3-7."""
thinking = {"type": "enabled", "budget_tokens": 8000}
body = self._transform("anthropic--claude-3-7-sonnet", thinking=thinking)
assert self._model_params(body).get("thinking") == thinking
class TestCacheControl:
"""Unit tests for cache_control preservation on message content parts."""
def _transform(self, model: str, messages: list) -> dict:
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
return cfg.transform_request(
model=model,
messages=messages,
optional_params={},
litellm_params={},
headers={},
)
def _template(self, body: dict) -> list:
return body["config"]["modules"]["prompt_templating"]["prompt"]["template"]
def test_cache_control_preserved_on_text_content(self):
"""cache_control on a TextContent part survives validation and appears in payload."""
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Hello",
"cache_control": {"type": "ephemeral"},
}
],
}
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
template = self._template(body)
content = template[0]["content"]
assert isinstance(content, list)
assert content[0].get("cache_control") == {"type": "ephemeral"}
def test_plain_text_content_unaffected(self):
"""Text content without cache_control still serialises cleanly."""
messages = [{"role": "user", "content": "Hello"}]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
template = self._template(body)
assert template[0]["content"] == "Hello"
def test_cache_control_preserved_on_multiple_parts(self):
"""cache_control is preserved on each part independently."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Part A", "cache_control": {"type": "ephemeral"}},
{"type": "text", "text": "Part B"},
],
}
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
content = self._template(body)[0]["content"]
assert content[0].get("cache_control") == {"type": "ephemeral"}
assert "cache_control" not in content[1]
def test_cache_control_preserved_on_system_message(self):
"""cache_control on a system message content part reaches the payload."""
messages = [
{
"role": "system",
"content": [
{
"type": "text",
"text": "You are a helpful assistant.",
"cache_control": {"type": "ephemeral"},
}
],
},
{"role": "user", "content": "Hello"},
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
template = self._template(body)
system_content = template[0]["content"]
assert isinstance(system_content, list)
assert system_content[0].get("cache_control") == {"type": "ephemeral"}
def test_system_message_without_cache_control_stays_string(self):
"""A plain system message is still serialised as a string, not a list."""
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello"},
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
template = self._template(body)
assert isinstance(template[0]["content"], str)
def test_cache_control_preserved_on_tool_definition(self):
"""cache_control on a tool definition reaches the payload."""
tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather",
"parameters": {"type": "object", "properties": {}},
},
"cache_control": {"type": "ephemeral"},
}
messages = [{"role": "user", "content": "Hi"}]
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
body = cfg.transform_request(
model="anthropic--claude-3-5-sonnet",
messages=messages,
optional_params={"tools": [tool]},
litellm_params={},
headers={},
)
tools = body["config"]["modules"]["prompt_templating"]["prompt"]["tools"]
assert tools[0].get("cache_control") == {"type": "ephemeral"}
def test_tool_without_cache_control_omits_field(self):
"""A tool without cache_control does not emit cache_control: null."""
tool = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather",
"parameters": {"type": "object", "properties": {}},
},
}
messages = [{"role": "user", "content": "Hi"}]
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
body = cfg.transform_request(
model="anthropic--claude-3-5-sonnet",
messages=messages,
optional_params={"tools": [tool]},
litellm_params={},
headers={},
)
tools = body["config"]["modules"]["prompt_templating"]["prompt"]["tools"]
assert "cache_control" not in tools[0]
def test_message_level_cache_control_folded_onto_string_content(self):
"""cache_control on the message dict (string content) is folded into a content block.
litellm's cache_control_injection_points hook produces this shape for
plain-string content. The marker must not be silently dropped.
"""
messages = [
{
"role": "system",
"content": "You are a helpful assistant.",
"cache_control": {"type": "ephemeral"},
},
{"role": "user", "content": "Hello"},
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
template = self._template(body)
system_content = template[0]["content"]
assert isinstance(system_content, list), "string content should have been promoted to a list"
assert system_content[0]["text"] == "You are a helpful assistant."
assert system_content[0].get("cache_control") == {"type": "ephemeral"}
def test_null_cache_control_on_content_block_is_omitted(self):
"""cache_control: null on a content block must not appear in the payload.
Sending null reaches AI Core and causes a 400 ('None is not of type object').
"""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Hello", "cache_control": None},
],
}
]
body = self._transform("anthropic--claude-3-5-sonnet", messages)
content = self._template(body)[0]["content"]
assert isinstance(content, list)
assert "cache_control" not in content[0]
class TestModelVersionAdvertisement:
"""model_version is advertised in get_supported_openai_params and lands in
model.version (not model.params) in the serialised request body.
"""
def _cfg(self):
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
return cfg
def _transform(self, model: str, **kwargs) -> dict:
cfg = self._cfg()
return cfg.transform_request(
model=model,
messages=[{"role": "user", "content": "Hi"}],
optional_params=dict(kwargs),
litellm_params={},
headers={},
)
def test_model_version_advertised_for_all_models(self):
for model in ("gpt-4o", "anthropic--claude-4-sonnet", "gemini-1.5-pro"):
params = self._cfg().get_supported_openai_params(model)
assert "model_version" in params, f"model_version missing for {model}"
def test_model_version_lands_in_model_version_field(self):
body = self._transform("gpt-4o", model_version="1.2.3")
pt = body["config"]["modules"]["prompt_templating"]
assert pt["model"]["version"] == "1.2.3"
def test_model_version_absent_from_model_params(self):
body = self._transform("gpt-4o", model_version="1.2.3")
pt = body["config"]["modules"]["prompt_templating"]
assert "model_version" not in pt["model"]["params"]
def test_model_version_defaults_to_latest(self):
body = self._transform("gpt-4o")
pt = body["config"]["modules"]["prompt_templating"]
assert pt["model"]["version"] == "latest"
class TestToolChoiceDropped:
"""SAP orchestration v2 rejects tool_choice in the request body (HTTP 400).
It is not advertised in get_supported_openai_params so callers receive
UnsupportedParamsError immediately. The defensive pop in _build_prompt_module
ensures it never reaches the wire even if injected via fallback_sap_modules.
tools themselves are still forwarded normally.
"""
_TOOL = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Return weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
}
def _transform(self, model: str, **kwargs) -> dict:
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
return cfg.transform_request(
model=model,
messages=[{"role": "user", "content": "What is the weather?"}],
optional_params=dict(kwargs),
litellm_params={},
headers={},
)
def _prompt(self, body: dict) -> dict:
return body["config"]["modules"]["prompt_templating"]["prompt"]
def test_tool_choice_not_advertised(self):
"""tool_choice must not appear in get_supported_openai_params for any model."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
for model in ("gpt-4o", "anthropic--claude-4-sonnet", "gemini-1.5-pro", "amazon--titan"):
assert "tool_choice" not in cfg.get_supported_openai_params(model), (
f"tool_choice must not be advertised for {model}"
)
def test_tools_still_advertised(self):
"""tools must still be advertised — only tool_choice is unsupported."""
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
for model in ("gpt-4o", "anthropic--claude-4-sonnet"):
assert "tools" in cfg.get_supported_openai_params(model)
def test_tools_still_forwarded(self):
"""Dropping tool_choice must not also suppress the tools list."""
body = self._transform("gpt-4o", tools=[self._TOOL])
assert "tools" in self._prompt(body)
assert self._prompt(body)["tools"][0]["function"]["name"] == "get_weather"
class TestTimeoutAndMaxRetries:
"""timeout and max_retries land in model-level sibling fields,
not inside model.params.
"""
def _transform(self, model: str, **kwargs) -> dict:
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
return cfg.transform_request(
model=model,
messages=[{"role": "user", "content": "Hi"}],
optional_params=dict(kwargs),
litellm_params={},
headers={},
)
def _model(self, body: dict) -> dict:
return body["config"]["modules"]["prompt_templating"]["model"]
def test_timeout_and_max_retries_advertised(self):
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
cfg = GenAIHubOrchestrationConfig()
for model in ("gpt-4o", "anthropic--claude-4-sonnet"):
params = cfg.get_supported_openai_params(model)
assert "timeout" in params, f"timeout missing for {model}"
assert "max_retries" in params, f"max_retries missing for {model}"
def test_timeout_lands_at_model_level(self):
body = self._transform("gpt-4o", timeout=120)
model = self._model(body)
assert model.get("timeout") == 120
assert "timeout" not in model.get("params", {})
def test_max_retries_lands_at_model_level(self):
body = self._transform("gpt-4o", max_retries=3)
model = self._model(body)
assert model.get("max_retries") == 3
assert "max_retries" not in model.get("params", {})
def test_timeout_and_max_retries_together(self):
body = self._transform("gpt-4o", timeout=60, max_retries=2)
model = self._model(body)
assert model.get("timeout") == 60
assert model.get("max_retries") == 2
assert "timeout" not in model.get("params", {})
assert "max_retries" not in model.get("params", {})
def test_absent_when_not_passed(self):
"""Neither key appears in the serialised body when not supplied."""
body = self._transform("gpt-4o", temperature=0.7)
model = self._model(body)
assert "timeout" not in model
assert "max_retries" not in model
class TestUserForwarding:
"""user param is advertised and lands in model.params (correct per SDK v2)."""
def _cfg(self):
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
return GenAIHubOrchestrationConfig()
def _transform(self, **kwargs) -> dict:
cfg = self._cfg()
return cfg.transform_request(
model="gpt-4o",
messages=[{"role": "user", "content": "Hi"}],
optional_params=dict(kwargs),
litellm_params={},
headers={},
)
def test_user_advertised(self):
params = self._cfg().get_supported_openai_params("gpt-4o")
assert "user" in params
def test_user_lands_in_model_params(self):
body = self._transform(user="uid-abc123")
pt = body["config"]["modules"]["prompt_templating"]
assert pt["model"]["params"].get("user") == "uid-abc123"