scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-10 08:28:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

fix(sap): apply similar changes to embed module

Browse source
This commit is contained in:
Yamac Ay 2026-10-27 16:13:18 +01:00
parent 2925c07cea
commit 95b9ccc2c9
No known key found for this signature in database
GPG key ID: D113B438819CA628
6 changed files with 313 additions and 108 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
litellm/llms/sap/chat/transformation.py
View file

@ -2,18 +2,14 @@
Translate from OpenAI's `/v1/chat/completions` to SAP Generative AI Hub's Orchestration Service`v2/completion`
"""
import threading
import types
import os
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Any, ClassVar, Final, Union
from typing import TYPE_CHECKING, Any, Final, Union
import httpx
from pydantic import BaseModel
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm._logging import verbose_logger
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponse
@ -27,7 +23,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
from ..credentials import AICoreConfiguration, AICoreDeploymentList, get_token_creator
from ..credentials import get_token_creator
from ..deployment import fetch_deployment_url, resolve_deployment_url
from .handler import (
AsyncSAPStreamIterator,
GenAIHubOrchestrationError,
@ -45,8 +42,6 @@ from .models import (
SAPUserMessage,
)
# Env var that an operator can set to pin a specific orchestration deployment URL.
# An empty string is treated as unset and falls through to auto-discovery.
_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR = "AICORE_ORCHESTRATION_DEPLOYMENT_URL"
# Keys routed outside SAP orchestration `model.params` (prompt, stream, fallbacks, etc.)
@ -156,7 +151,6 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
self.token_creator = None
self._base_url = None
self._resource_group = None
self._cached_deployment_url: str | None = None # None = not yet resolved
def run_env_setup(self, service_key: str | None = None) -> None:
try:
@ -188,77 +182,23 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
self.run_env_setup()
return self._resource_group
# Process-wide cache: the provider map builds a fresh config instance per
# call, so an instance-scoped cached_property re-runs the two discovery
# GETs on every LLM call. Keyed on (base_url, resource_group) - the inputs
# that determine the URL - so a credential change never serves a stale URL.
_deployment_url_cache: ClassVar[dict[tuple[str, str], str]] = {} # mutable-ok: process cache, guarded by the lock
_deployment_url_lock: ClassVar[threading.Lock] = threading.Lock()
@property
def deployment_url(self) -> str:
"""Resolve the orchestration deployment URL, with caching.
Resolution order:
1. ``_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR`` env var (operator-level pin).
An empty string is treated as unset and falls through to discovery.
2. Auto-discovery via ``/lm/deployments`` (one network call, cached).
A per-request override via ``optional_params["deployment_url"]`` is
handled one level up in ``get_complete_url`` before this property
is ever called.
"""
cache_key: Final = (self.base_url, self.resource_group)
cached_url: Final = GenAIHubOrchestrationConfig._deployment_url_cache.get(cache_key)
if cached_url is not None:
return cached_url
with GenAIHubOrchestrationConfig._deployment_url_lock:
rechecked_url: Final = GenAIHubOrchestrationConfig._deployment_url_cache.get(cache_key)
if rechecked_url is not None:
return rechecked_url
fetched_url: Final = os.environ.get(_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR) or self._fetch_deployment_url()
GenAIHubOrchestrationConfig._deployment_url_cache[cache_key] = fetched_url
return fetched_url
return resolve_deployment_url(
self.base_url,
self.resource_group,
_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR,
self._fetch_deployment_url,
)
def _fetch_deployment_url(self) -> str:
client: Final = litellm.module_level_client
# with httpx.Client(timeout=30) as client:
deployments: Final[AICoreDeploymentList] = (
client.get(
join_api_base(self.base_url, "lm/deployments"),
headers=self.headers,
params={ # mutable-ok: one-shot query-params dict passed directly to httpx, not stored
"scenarioId": "orchestration",
"executableIds": ["orchestration"],
"status": "RUNNING",
},
).json()
return fetch_deployment_url(
self.base_url,
self.headers,
"No orchestration deployment found in SAP AI Core.",
"orchestration",
)
candidates: Final[list[tuple[str, str, str]]] = sorted(
(
(resource["deploymentUrl"], resource["createdAt"], resource.get("configurationName", ""))
for resource in deployments.get("resources", [])
if resource.get("deploymentUrl")
),
key=lambda c: c[1],
reverse=True,
)
if not candidates:
raise GenAIHubOrchestrationError(
status_code=404,
message="No orchestration deployment found in SAP AI Core.",
)
if len(candidates) > 1:
verbose_logger.warning(
"SAP: %d orchestration deployments found; using newest (name=%r, url=%r). Others ignored: %s.",
len(candidates),
candidates[0][2],
candidates[0][0],
tuple(v[2] or v[0] for v in candidates[1:]),
)
return candidates[0][0]
@classmethod
def get_config(cls):
return super().get_config()

81
litellm/llms/sap/deployment.py Normal file
View file

@ -0,0 +1,81 @@
"""Shared deployment URL resolution for SAP GenAI Hub providers.
Both chat (orchestration) and embedding configs discover the same
orchestration deployment, so a single process-wide cache keyed on
(base_url, resource_group) is shared across all SAP provider types.
"""
import os
import threading
from collections.abc import Callable, Mapping
from typing import Final
import litellm
from litellm._logging import verbose_logger
from .chat.handler import GenAIHubOrchestrationError
from .credentials import AICoreDeploymentList
from litellm.litellm_core_utils.url_utils import join_api_base
_deployment_url_cache: dict[tuple[str, str], str] = {} # mutable-ok: process-wide cache, guarded by _deployment_url_lock
_deployment_url_lock: Final = threading.Lock()
def fetch_deployment_url(
base_url: str,
headers: Mapping[str, str],
not_found_message: str,
kind: str,
) -> str:
deployments: Final[AICoreDeploymentList] = (
litellm.module_level_client.get(
join_api_base(base_url, "lm/deployments"),
headers=headers,
params={ # mutable-ok: one-shot query-params dict passed directly to httpx, not stored
"scenarioId": "orchestration",
"executableIds": ["orchestration"],
"status": "RUNNING",
},
).json()
)
candidates: Final[list[tuple[str, str, str]]] = sorted(
(
(resource["deploymentUrl"], resource["createdAt"], resource.get("configurationName", ""))
for resource in deployments.get("resources", [])
if resource.get("deploymentUrl")
),
key=lambda c: c[1],
reverse=True,
)
if not candidates:
raise GenAIHubOrchestrationError(status_code=404, message=not_found_message)
if len(candidates) > 1:
verbose_logger.warning(
"SAP: %d %s deployments found; using newest (name=%r, url=%r). Others ignored: %s.",
len(candidates),
kind,
candidates[0][2],
candidates[0][0],
tuple(v[2] or v[0] for v in candidates[1:]),
)
return candidates[0][0]
def resolve_deployment_url(
base_url: str,
resource_group: str,
env_var: str,
fetcher: Callable[[], str],
) -> str:
cache_key: Final[tuple[str, str]] = (base_url, resource_group)
cached_url = _deployment_url_cache.get(cache_key)
if cached_url is not None:
return cached_url
with _deployment_url_lock:
rechecked_url = _deployment_url_cache.get(cache_key)
if rechecked_url is not None:
return rechecked_url
fetched_url: Final[str] = os.environ.get(env_var) or fetcher()
_deployment_url_cache[cache_key] = fetched_url
return fetched_url

50
litellm/llms/sap/embed/transformation.py
View file

@ -2,8 +2,7 @@
Translates from OpenAI's `/v1/embeddings` to IBM's `/text/embeddings` route.
"""
import threading
from typing import ClassVar, Final, Literal
from typing import Final, Literal
import httpx
from pydantic import Field
@ -18,8 +17,11 @@ from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import AllEmbeddingInputValues
from litellm.types.utils import EmbeddingResponse
_AICORE_EMBEDDING_DEPLOYMENT_URL_ENV_VAR = "AICORE_EMBEDDING_DEPLOYMENT_URL"
from ..chat.handler import GenAIHubOrchestrationError
from ..credentials import AICoreConfiguration, AICoreDeploymentList, get_token_creator
from ..credentials import get_token_creator
from ..deployment import fetch_deployment_url, resolve_deployment_url
class Usage(LiteLLMBaseModel):
@ -98,40 +100,22 @@ class GenAIHubEmbeddingConfig(BaseEmbeddingConfig):
}
return headers
# Process-wide cache, same reasoning as the chat config: per-call config
# instances would otherwise re-run the two discovery GETs on every call.
_deployment_url_cache: ClassVar[dict[tuple[str, str], str]] = {} # mutable-ok: process cache, guarded by the lock
_deployment_url_lock: ClassVar[threading.Lock] = threading.Lock()
@property
def deployment_url(self) -> str:
cache_key: Final = (self.base_url, self.resource_group)
cached_url: Final = GenAIHubEmbeddingConfig._deployment_url_cache.get(cache_key)
if cached_url is not None:
return cached_url
with GenAIHubEmbeddingConfig._deployment_url_lock:
rechecked_url: Final = GenAIHubEmbeddingConfig._deployment_url_cache.get(cache_key)
if rechecked_url is not None:
return rechecked_url
fetched_url: Final = self._fetch_deployment_url()
GenAIHubEmbeddingConfig._deployment_url_cache[cache_key] = fetched_url
return fetched_url
return resolve_deployment_url(
self.base_url,
self.resource_group,
_AICORE_EMBEDDING_DEPLOYMENT_URL_ENV_VAR,
self._fetch_deployment_url,
)
def _fetch_deployment_url(self) -> str:
with httpx.Client(timeout=30) as client:
valid_deployments: Final = []
deployments: Final[AICoreDeploymentList] = client.get(
join_api_base(self.base_url, "lm/deployments"), headers=self.headers
).json()
for deployment in deployments.get("resources", []):
if deployment["scenarioId"] == "orchestration":
config_details: AICoreConfiguration = client.get(
join_api_base(self.base_url, f"lm/configurations/{deployment['configurationId']}"),
headers=self.headers,
).json()
if config_details["executableId"] == "orchestration":
valid_deployments.append((deployment["deploymentUrl"], deployment["createdAt"]))
return sorted(valid_deployments, key=lambda x: x[1], reverse=True)[0][0]
return fetch_deployment_url(
self.base_url,
self.headers,
"No embedding deployment found in SAP AI Core.",
"embedding",
)
def get_error_class(self, error_message, status_code, headers):
return GenAIHubOrchestrationError(status_code, error_message)

61
tests/e2e/test_sap_deployment_url_cache_e2e.py Normal file
View file

@ -0,0 +1,61 @@
"""
E2E: verify SAP deployment URL caching eliminates repeated discovery GETs.
Run: source .env && uv run --no-sync python tests/e2e/test_sap_deployment_url_cache_e2e.py
"""
import time
import litellm
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
def main():
# # Start clean
# GenAIHubOrchestrationConfig._deployment_url_cache.clear()
prompt = [{"role": "user", "content": "Say 'hello' and nothing else."}]
# First call: pays the discovery cost (GET /lm/deployments)
t0 = time.perf_counter()
r1 = litellm.completion(model="sap/gpt-4o-mini", messages=prompt, max_tokens=5)
first_ms = (time.perf_counter() - t0) * 1000
# # Cache should now have exactly one entry
# cache = GenAIHubOrchestrationConfig._deployment_url_cache
# assert len(cache) == 1, f"expected 1 cache entry, got {len(cache)}"
# cached_url = next(iter(cache.values()))
# print(f"cached deployment URL: {cached_url}")
# Second call: should skip discovery entirely
t0 = time.perf_counter()
r2 = litellm.completion(model="sap/gpt-4o-mini", messages=prompt, max_tokens=5)
second_ms = (time.perf_counter() - t0) * 1000
# # Cache should still have the same single entry (no new discovery)
# assert len(cache) == 1, f"cache grew unexpectedly to {len(cache)} entries"
# assert next(iter(cache.values())) == cached_url, "cached URL changed between calls"
# Both calls should return valid responses
c1 = r1.choices[0].message.content
c2 = r2.choices[0].message.content
assert c1 and c2, f"empty responses: {c1!r}, {c2!r}"
print(f"first call: {first_ms:.0f}ms (includes discovery + LLM)")
print(f"second call: {second_ms:.0f}ms (LLM only, no discovery)")
print(f"response 1: {c1!r}")
print(f"response 2: {c2!r}")
# The second call should be meaningfully faster because it skips the
# deployment discovery round-trip. We don't pin a hard threshold since
# network jitter exists, but the saving is typically 100-500ms.
savings = first_ms - second_ms
print(f"savings: {savings:.0f}ms")
if savings > 0:
print("PASS: second call was faster (cache hit skipped discovery)")
else:
print("NOTE: second call was not faster, likely network jitter. Cache still worked (entry count = 1).")
if __name__ == "__main__":
main()

11
tests/unit/llms/sap/chat/test_sap_transformation.py
View file

@ -4,6 +4,15 @@ from unittest.mock import MagicMock, patch
from pydantic import ValidationError
@pytest.fixture(autouse=True)
def _clear_sap_deployment_caches():
import litellm.llms.sap.deployment as _depl
_depl._deployment_url_cache.clear()
yield
_depl._deployment_url_cache.clear()
def _make_config():
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
@ -962,7 +971,7 @@ class TestDeploymentResolution:
def test_multiple_deployments_emits_warning(self):
cfg = _make_config()
with patch("litellm.module_level_client", _mock_client("older", "newer")): # test-quality-ok: patching the HTTP client at the litellm transport boundary
with patch("litellm.llms.sap.chat.transformation.verbose_logger") as mock_log: # test-quality-ok: patching module-level logger to intercept warning calls
with patch("litellm.llms.sap.deployment.verbose_logger") as mock_log: # test-quality-ok: patching module-level logger to intercept warning calls
url = cfg._fetch_deployment_url()
mock_log.warning.assert_called_once()
assert 2 == mock_log.warning.call_args[0][1]

130
tests/unit/llms/sap/test_sap_deployment_url_cache.py Normal file
View file

@ -0,0 +1,130 @@
"""Process-wide deployment_url caching for SAP GenAI Hub configs (issue #47096).
The provider map builds a fresh config instance on every LLM call, so an
instance-scoped cached_property re-ran the orchestration-discovery GET on
every call. The cache now lives in deployment.py as a single process-wide
dict keyed on (base_url, resource_group), shared by chat and embed.
"""
import pytest
import litellm.llms.sap.deployment as _depl
from litellm.llms.sap.chat.transformation import GenAIHubOrchestrationConfig
from litellm.llms.sap.embed.transformation import GenAIHubEmbeddingConfig
@pytest.fixture(autouse=True)
def clear_deployment_url_cache():
_depl._deployment_url_cache.clear()
yield
_depl._deployment_url_cache.clear()
def _chat_config(base_url: str, resource_group: str) -> GenAIHubOrchestrationConfig:
config = GenAIHubOrchestrationConfig()
config._base_url = base_url
config._resource_group = resource_group
return config
def _embed_config(monkeypatch, base_url: str, resource_group: str) -> GenAIHubEmbeddingConfig:
monkeypatch.setattr(
"litellm.llms.sap.embed.transformation.get_token_creator",
lambda service_key=None: (lambda: "token", base_url, resource_group),
)
return GenAIHubEmbeddingConfig()
def test_chat_deployment_url_fetched_once_across_instances(monkeypatch):
fetches = []
def fake_fetch(self):
fetches.append(self)
return "https://deploy.example/v2"
monkeypatch.setattr(GenAIHubOrchestrationConfig, "_fetch_deployment_url", fake_fetch)
first = _chat_config("https://api.example", "default")
second = _chat_config("https://api.example", "default")
assert first.deployment_url == "https://deploy.example/v2"
assert second.deployment_url == "https://deploy.example/v2"
assert len(fetches) == 1
def test_chat_deployment_url_cache_keyed_on_base_url_and_resource_group(monkeypatch):
fetches = []
def fake_fetch(self):
fetches.append(self)
return f"https://deploy.example/{self.resource_group}"
monkeypatch.setattr(GenAIHubOrchestrationConfig, "_fetch_deployment_url", fake_fetch)
default_rg = _chat_config("https://api.example", "default")
other_rg = _chat_config("https://api.example", "other")
assert default_rg.deployment_url == "https://deploy.example/default"
assert other_rg.deployment_url == "https://deploy.example/other"
assert len(fetches) == 2
def test_embed_deployment_url_fetched_once_across_instances(monkeypatch):
fetches = []
def fake_fetch(self):
fetches.append(self)
return "https://deploy.example/v2"
monkeypatch.setattr(GenAIHubEmbeddingConfig, "_fetch_deployment_url", fake_fetch)
first = _embed_config(monkeypatch, "https://api.example", "default")
second = _embed_config(monkeypatch, "https://api.example", "default")
assert first.deployment_url == "https://deploy.example/v2"
assert second.deployment_url == "https://deploy.example/v2"
assert len(fetches) == 1
def test_embed_deployment_url_cache_keyed_on_base_url_and_resource_group(monkeypatch):
fetches = []
def fake_fetch(self):
fetches.append(self)
return f"https://deploy.example/{self.resource_group}"
monkeypatch.setattr(GenAIHubEmbeddingConfig, "_fetch_deployment_url", fake_fetch)
default_rg = _embed_config(monkeypatch, "https://api.example", "default")
other_rg = _embed_config(monkeypatch, "https://api.example", "other")
assert default_rg.deployment_url == "https://deploy.example/default"
assert other_rg.deployment_url == "https://deploy.example/other"
assert len(fetches) == 2
def test_chat_and_embed_share_cache(monkeypatch):
"""Chat discovery populates the shared cache so embed never fetches."""
chat_fetches = []
embed_fetches = []
def fake_chat_fetch(self):
chat_fetches.append(self)
return "https://deploy.example/shared"
def fake_embed_fetch(self):
embed_fetches.append(self)
return "https://deploy.example/shared"
monkeypatch.setattr(GenAIHubOrchestrationConfig, "_fetch_deployment_url", fake_chat_fetch)
monkeypatch.setattr(GenAIHubEmbeddingConfig, "_fetch_deployment_url", fake_embed_fetch)
chat = _chat_config("https://api.example", "default")
embed = _embed_config(monkeypatch, "https://api.example", "default")
assert chat.deployment_url == "https://deploy.example/shared"
assert embed.deployment_url == "https://deploy.example/shared"
assert len(chat_fetches) == 1
assert len(embed_fetches) == 0, "embed should reuse the chat cache entry, not fetch again"
assert len(_depl._deployment_url_cache) == 1