mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
5e38a08741
* refactor(cache): organize v2 cache as a package * docs: clarify experimental v2 guidance * fix(cache): verify cache-hit accounting and preserve logging metadata * refactor(cache): separate execution facts from host accounting * refactor(rust): build messages routes with named dependencies * wip * fix(cache): preserve facade policy and preflight fallback * refactor(cache): defer shared Python logging changes * test(gateway-inference): allow dead code in shared test helpers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(cache): key prepared requests and honor facade controls * feat(cache): use Python caches from Rust Messages inference * refactor(cache): separate native and Python cache adapters * refactor(cache): enforce shared composition and adapter boundaries * fix(cache): let Python key delegated Rust Messages entries --------- Co-authored-by: Yujong Lee <yujong@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
112 lines
4.1 KiB
Python
112 lines
4.1 KiB
Python
"""Ordered rollout policy for routes, loggers, and secret managers.
|
|
|
|
The first matching rule wins; unmatched contexts stay on Python. Native
|
|
admission separately decides whether the selected implementation can execute.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Final, TypeAlias
|
|
|
|
from litellm.rust_bridge.configuration import Decision, Rollout
|
|
from litellm.rust_bridge.configuration import decision as _decision
|
|
from litellm.types.secret_managers.main import KeyManagementSystem
|
|
|
|
|
|
class Route(str, Enum):
|
|
CHAT_COMPLETIONS = "chat_completions"
|
|
EMBEDDINGS = "embeddings"
|
|
MESSAGES = "messages"
|
|
RESPONSES = "responses"
|
|
TRANSCRIPTION = "transcription"
|
|
OCR = "ocr"
|
|
TOKEN_COUNTER = "token_counter"
|
|
TOKENIZER = "tokenizer"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RouteContext:
|
|
route: Route
|
|
provider: str | None = None
|
|
model: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RouteRule:
|
|
route: Route
|
|
rollout: Rollout
|
|
providers: frozenset[str] | None = None
|
|
models: frozenset[str] | None = None
|
|
|
|
def matches(self, context: Context) -> bool:
|
|
return (
|
|
isinstance(context, RouteContext)
|
|
and context.route is self.route
|
|
and (self.providers is None or context.provider in self.providers)
|
|
and (self.models is None or context.model in self.models)
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SecretManagerContext:
|
|
system: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SecretManagerRule:
|
|
rollout: Rollout
|
|
systems: frozenset[str] | None = None
|
|
|
|
def matches(self, context: Context) -> bool:
|
|
return isinstance(context, SecretManagerContext) and (self.systems is None or context.system in self.systems)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LoggerContext:
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LoggerRule:
|
|
rollout: Rollout
|
|
|
|
def matches(self, context: Context) -> bool:
|
|
return isinstance(context, LoggerContext)
|
|
|
|
|
|
Context: TypeAlias = RouteContext | SecretManagerContext | LoggerContext
|
|
Rule: TypeAlias = RouteRule | SecretManagerRule | LoggerRule
|
|
Rules: TypeAlias = tuple[Rule, ...]
|
|
|
|
RULES: Final[Rules] = (
|
|
LoggerRule(Rollout.RUST_OPT_IN),
|
|
RouteRule(Route.CHAT_COMPLETIONS, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.EMBEDDINGS, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.OCR, Rollout.RUST_REQUIRED),
|
|
RouteRule(Route.MESSAGES, Rollout.RUST_OPT_IN, providers=frozenset({"anthropic"})),
|
|
RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.RESPONSES, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.TOKEN_COUNTER, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.TOKENIZER, Rollout.PYTHON_ONLY),
|
|
RouteRule(Route.TRANSCRIPTION, Rollout.RUST_REQUIRED, providers=frozenset({"bedrock"})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.GOOGLE_KMS.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.AZURE_KEY_VAULT.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.AWS_SECRET_MANAGER.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.GOOGLE_SECRET_MANAGER.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.HASHICORP_VAULT.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.CYBERARK.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.LOCAL.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.AWS_KMS.value})),
|
|
SecretManagerRule(Rollout.PYTHON_ONLY, systems=frozenset({KeyManagementSystem.CUSTOM.value})),
|
|
)
|
|
|
|
|
|
def rollout(context: Context, rules: Rules | None = None) -> Rollout:
|
|
selected_rules: Final = RULES if rules is None else rules
|
|
return next((rule.rollout for rule in selected_rules if rule.matches(context)), Rollout.PYTHON_ONLY)
|
|
|
|
|
|
def decision(context: Context, rules: Rules | None = None) -> Decision:
|
|
return _decision(rollout(context, rules))
|