API reference¶
Generated from the apogee-ai-serving source with mkdocstrings. Every symbol below is exported from apogee_ai_serving, so it is part of the supported public surface.
Application · DTOs¶
Bases: BaseModel
class-attribute
instance-attribute
¶
Fraction of requests that should be near-duplicates of earlier prompts.
Application · Use cases¶
Python
BenchCacheHitUseCase(registry: EngineRegistry, cache: ISemanticCache)
Synthetic load — measure cache hit rate against the configured cache.
Generates a stream of prompts where a fraction are paraphrases of
earlier queries. Used by the serve:bench CLI to verify the
ST-056-8 ≥30% target on the echo engine.
Source code in apogee_ai_serving/application/use_cases/bench_cache_hit_use_case.py
async
¶
Python
execute(dto: BenchDTO, *, model: str = 'echo') -> dict[str, float]
Source code in apogee_ai_serving/application/use_cases/bench_cache_hit_use_case.py
Python
async def execute(self, dto: BenchDTO, *, model: str = "echo") -> dict[str, float]:
if dto.requests <= 0:
raise ValueError("requests must be > 0")
recent_prompts: list[str] = []
for i in range(dto.requests):
if recent_prompts and secrets.SystemRandom().random() < dto.paraphrase_ratio:
# Paraphrase: take a recent prompt and add a small permutation
base = secrets.SystemRandom().choice(recent_prompts)
prompt = base
else:
prompt = f"{dto.base_prompt} #{i}"
recent_prompts.append(prompt)
if len(recent_prompts) > 32:
recent_prompts.pop(0)
await self._complete.execute(
ServingCompletionRequest(prompt=prompt, model=model)
)
return await self._cache.stats()
Python
CompleteUseCase(registry: EngineRegistry, *, cache: ISemanticCache | None = None, router: IModelRouter | None = None, rate_limiter: IRateLimiter | None = None, fallback: IFallbackChain | None = None)
Pipeline: rate-limit → cache → route → engine → fallback.
Source code in apogee_ai_serving/application/use_cases/complete_use_case.py
Python
def __init__(
self,
registry: EngineRegistry,
*,
cache: ISemanticCache | None = None,
router: IModelRouter | None = None,
rate_limiter: IRateLimiter | None = None,
fallback: IFallbackChain | None = None,
) -> None:
self._registry = registry
self._cache = cache
self._router = router
self._rate_limiter = rate_limiter
self._fallback = fallback
async
¶
Python
execute(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/application/use_cases/complete_use_case.py
Python
async def execute(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
# 1. Rate limit
if self._rate_limiter is not None:
scope = request.tenant_id or request.user_id or "_global"
allowed = await self._rate_limiter.acquire(scope_key=scope)
if not allowed:
raise RateLimitExceededException(scope, 0)
# 2. Route
if self._router is not None:
decision = self._router.route(request)
engine_name = decision.engine
model = decision.model
else:
engine_name = request.model or self._registry.list()[0]
model = request.model or engine_name
request_for_engine = replace(request, model=model) if request.model != model else request
# 3. Cache lookup
cache_key: CacheKey | None = None
if request.cache_enabled and self._cache is not None:
cache_key = CacheKey.of(
prompt=request.prompt, model=model, tenant_id=request.tenant_id
)
lookup = await self._cache.get(cache_key, prompt=request.prompt)
if lookup.hit and lookup.payload:
return _from_cache_payload(lookup.payload, model=model, engine=engine_name)
# 4. Execute (with fallback if configured)
if self._fallback is not None:
response = await self._fallback.execute(
request_for_engine, primary_engine=engine_name
)
else:
engine = self._registry.find(engine_name)
if engine is None:
raise EngineNotAvailableException(engine_name, "not registered")
response = await engine.complete(request_for_engine)
# 5. Store in cache
if cache_key is not None and self._cache is not None:
await self._cache.put(cache_key, request.prompt, response)
return response
Python
ListEnginesUseCase(registry: EngineRegistry)
Source code in apogee_ai_serving/application/use_cases/list_engines_use_case.py
async
¶
Python
execute() -> list[tuple[str, list[ModelDescriptor]]]
Source code in apogee_ai_serving/application/use_cases/list_engines_use_case.py
Python
async def execute(self) -> list[tuple[str, list[ModelDescriptor]]]:
out: list[tuple[str, list[ModelDescriptor]]] = []
for name in self._registry.list():
engine = self._registry.get(name)
try:
models = await engine.list_models()
except Exception: # noqa: BLE001 - skip unhealthy engines
models = []
out.append((name, models))
return out
Python
RouteUseCase(router: IModelRouter)
Source code in apogee_ai_serving/application/use_cases/route_use_case.py
async
¶
Python
execute(request: ServingCompletionRequest) -> RoutingDecision
Domain¶
dataclass
¶
Composite key for cache lookups.
classmethod
¶
Python
of(prompt: str, *, model: str, tenant_id: str | None = None) -> CacheKey
Source code in apogee_ai_serving/domain/value_objects/cache_key.py
Python
@classmethod
def of(
cls,
prompt: str,
*,
model: str,
tenant_id: str | None = None,
) -> CacheKey:
h = hashlib.sha256()
h.update(model.encode())
h.update(b"\x00")
h.update((tenant_id or "").encode())
h.update(b"\x00")
h.update(prompt.strip().lower().encode())
return cls(digest=h.hexdigest(), model=model, tenant_id=tenant_id)
dataclass
¶
Python
CacheLookup(hit: bool, score: float = 0.0, payload: dict | None = None, matched_prompt: str | None = None, age_seconds: float = 0.0)
Result of a semantic cache hit/miss.
dataclass
¶
Python
FallbackPolicy(engines: tuple[str, ...] = tuple(), max_attempts: int | None = None, retry_on_rate_limit: bool = True, retry_on_engine_error: bool = True)
dataclass
¶
Python
ModelDescriptor(name: str, engine: EngineKind, tier: ModelTier = BALANCED, context_window: int = 8192, description: str | None = None, tags: tuple[str, ...] = tuple())
Metadata returned by ICompletionEngine.list_models.
class-attribute
instance-attribute
¶
Bases: str, Enum
Cost / capability tiers for routing.
Conventionally:
- CHEAP: Haiku-class, GPT-4o-mini, Gemini Flash, local small
- BALANCED: Sonnet-class, GPT-4o, Gemini Pro
- EXPENSIVE: Opus, o1/o3 reasoning, GPT-4 Turbo
dataclass
¶
Python
QuotaSnapshot(scope: str, requests_remaining: int, requests_per_minute: int, tokens_remaining: int | None = None, reset_in_seconds: float = 60.0)
Remaining capacity reported by an IRateLimiter.
dataclass
¶
Python
RateLimit(scope: QuotaScope = TENANT, requests_per_minute: int = 60, tokens_per_minute: int | None = None, burst: int = 1)
Token-bucket configuration for one scope.
class-attribute
instance-attribute
¶
Multiplier for the bucket size relative to per-minute rate.
dataclass
¶
Python
RouterRule(tier: ModelTier, max_chars: int | None = None, keywords: tuple[str, ...] = tuple(), priority: int = 0)
Picks a tier when the prompt matches some heuristic.
class-attribute
instance-attribute
¶
Use this tier when len(prompt) <= max_chars.
class-attribute
instance-attribute
¶
Force this tier if any keyword is found in the prompt (case-insensitive).
class-attribute
instance-attribute
¶
Higher wins ties when several rules match.
dataclass
¶
Python
RoutingDecision(engine: str, model: str, tier: ModelTier, strategy: RoutingStrategy, reason: str = '')
dataclass
¶
dataclass
¶
Python
ServingCompletionRequest(prompt: str, model: str | None = None, max_tokens: int = 512, temperature: float = 0.7, top_p: float = 1.0, stop: tuple[str, ...] = tuple(), tenant_id: str | None = None, user_id: str | None = None, metadata: dict[str, str] = dict(), cache_enabled: bool = True)
dataclass
¶
Python
ServingCompletionResponse(text: str, model: str, engine: str, prompt_tokens: int = 0, completion_tokens: int = 0, total_tokens: int = 0, latency_ms: float = 0.0, cached: bool = False, finish_reason: str | None = None, fallback_chain: tuple[str, ...] = tuple(), metadata: dict[str, str] = dict())
Domain · Enums¶
Bases: str, Enum
Domain · Exceptions¶
Bases: ServingError
Source code in apogee_ai_serving/domain/exceptions/serving_exceptions.py
Bases: ServingError
Source code in apogee_ai_serving/domain/exceptions/serving_exceptions.py
Bases: ServingError
Source code in apogee_ai_serving/domain/exceptions/serving_exceptions.py
Bases: Exception
Base for apogee-ai-serving errors.
Domain · Protocols (ports)¶
Bases: Protocol
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
async
¶
Python
list_models() -> list[ModelDescriptor]
async
¶
async
¶
Bases: Protocol
async
¶
Python
execute(request: ServingCompletionRequest, *, primary_engine: str) -> ServingCompletionResponse
Bases: Protocol
Python
route(request: ServingCompletionRequest) -> RoutingDecision
Bases: Protocol
Lightweight embedder used by ISemanticCache for similarity lookup.
Bases: Protocol
Bases: Protocol
async
¶
Python
get(key: CacheKey, *, prompt: str) -> CacheLookup
async
¶
Python
put(key: CacheKey, prompt: str, response: ServingCompletionResponse) -> None
async
¶
async
¶
Infrastructure¶
Python
BridgedPromptEmbedder(encoder: Callable[[str], list[float] | Awaitable[list[float]]], *, dimension: int)
Wrap any callable returning list[float] (e.g. apogee-ai-providers).
Source code in apogee_ai_serving/infrastructure/cache/embedders.py
async
¶
Deterministic engine: echoes the prompt back. CI-safe, no network.
Source code in apogee_ai_serving/infrastructure/engines/echo_engine.py
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/engines/echo_engine.py
Python
async def complete(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
start = time.perf_counter()
prompt_tokens = max(1, len(request.prompt) // 4)
completion_tokens = min(request.max_tokens, prompt_tokens)
text = f"echo: {request.prompt[: request.max_tokens]}"
latency = (time.perf_counter() - start) * 1000.0
return ServingCompletionResponse(
text=text,
model=request.model or self._model,
engine=self.name,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
latency_ms=latency,
finish_reason="stop",
)
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
Source code in apogee_ai_serving/infrastructure/engines/echo_engine.py
Python
async def stream(
self, request: ServingCompletionRequest
) -> AsyncIterator[ServingChunk]:
result = await self.complete(request)
async def gen() -> AsyncIterator[ServingChunk]:
words = result.text.split()
for i, word in enumerate(words):
payload = (" " if i else "") + word
yield ServingChunk(
delta=payload,
sequence=i,
is_final=i == len(words) - 1,
finish_reason=result.finish_reason if i == len(words) - 1 else None,
)
await asyncio.sleep(0)
return gen()
async
¶
Python
list_models() -> list[ModelDescriptor]
Source code in apogee_ai_serving/infrastructure/engines/echo_engine.py
async
¶
async
¶
Python
EngineRegistry(engines: Mapping[str, ICompletionEngine] | None = None)
Source code in apogee_ai_serving/infrastructure/registry/engine_registry.py
Python
register(engine: ICompletionEngine) -> None
Python
get(name: str) -> ICompletionEngine
Python
find(name: str) -> ICompletionEngine | None
Deterministic embedding based on token hashing — no API calls.
Source code in apogee_ai_serving/infrastructure/cache/embedders.py
async
¶
Source code in apogee_ai_serving/infrastructure/cache/embedders.py
Python
async def embed(self, text: str) -> tuple[float, ...]:
vec = [0.0] * self.dimension
tokens = [t for t in re.findall(r"[a-zA-ZÀ-ÿ0-9]+", text.lower()) if t]
if not tokens:
tokens = [text or " "]
for token in tokens:
digest = hashlib.sha256(token.encode("utf-8")).digest()
idx = int.from_bytes(digest[:4], "big") % self.dimension
sign = 1.0 if digest[4] & 1 else -1.0
vec[idx] += sign
norm = math.sqrt(sum(v * v for v in vec)) or 1.0
return tuple(v / norm for v in vec)
Python
InMemoryRateLimiter(*, default_limit: RateLimit | None = None, raise_on_exceeded: bool = False)
Token bucket per scope. Refills linearly to requests_per_minute.
Source code in apogee_ai_serving/infrastructure/quota/in_memory_rate_limiter.py
async
¶
Source code in apogee_ai_serving/infrastructure/quota/in_memory_rate_limiter.py
Python
async def acquire(
self,
*,
scope_key: str,
cost_tokens: int = 0, # noqa: ARG002 - reserved for token-budget mode
) -> bool:
async with self._lock:
bucket = self._bucket(scope_key, self._default)
self._refill(bucket)
if bucket.tokens < 1.0:
if self._raise:
raise RateLimitExceededException(scope_key, bucket.request_capacity)
return False
bucket.tokens -= 1.0
bucket.requests_remaining = max(0, bucket.requests_remaining - 1)
return True
async
¶
Python
remaining(*, scope_key: str) -> QuotaSnapshot
Source code in apogee_ai_serving/infrastructure/quota/in_memory_rate_limiter.py
Python
async def remaining(self, *, scope_key: str) -> QuotaSnapshot:
async with self._lock:
bucket = self._bucket(scope_key, self._default)
self._refill(bucket)
reset = max(0.0, (bucket.capacity - bucket.tokens) / max(bucket.rate_per_second, 1e-9))
return QuotaSnapshot(
scope=scope_key,
requests_remaining=int(bucket.tokens),
requests_per_minute=bucket.request_capacity,
tokens_remaining=None,
reset_in_seconds=reset,
)
async
¶
Python
InMemorySemanticCache(*, embedder: IPromptEmbedder | None = None, threshold: float = 0.85, max_entries: int = 1000)
Stores recent prompt embeddings + responses; serves nearest neighbor.
threshold is the minimum cosine similarity to count as a hit.
Source code in apogee_ai_serving/infrastructure/cache/in_memory_semantic_cache.py
Python
def __init__(
self,
*,
embedder: IPromptEmbedder | None = None,
threshold: float = 0.85,
max_entries: int = 1000,
) -> None:
if not 0.0 <= threshold <= 1.0:
raise ValueError("threshold must be in [0,1]")
if max_entries <= 0:
raise ValueError("max_entries must be > 0")
self._embedder = embedder or HashingPromptEmbedder()
self._threshold = threshold
self._max = max_entries
self._entries: deque[
tuple[CacheKey, str, tuple[float, ...], ServingCompletionResponse, float]
] = deque(maxlen=max_entries)
self._hits = 0
self._misses = 0
async
¶
Python
get(key: CacheKey, *, prompt: str) -> CacheLookup
Source code in apogee_ai_serving/infrastructure/cache/in_memory_semantic_cache.py
Python
async def get(self, key: CacheKey, *, prompt: str) -> CacheLookup:
# Exact-match first (fast path; same digest)
for stored_key, stored_prompt, _vec, response, ts in self._entries:
if stored_key.digest == key.digest:
self._hits += 1
age = max(0.0, time.time() - ts)
return CacheLookup(
hit=True,
score=1.0,
payload=asdict(response),
matched_prompt=stored_prompt,
age_seconds=age,
)
# Semantic similarity scan (linear — fine up to a few thousand entries)
if not self._entries:
self._misses += 1
return CacheLookup(hit=False, score=0.0, payload=None)
query_vec = await self._embedder.embed(prompt)
best_score = -1.0
best: tuple[CacheKey, str, ServingCompletionResponse, float] | None = None
for stored_key, stored_prompt, vec, response, ts in self._entries:
if stored_key.model != key.model:
continue
if stored_key.tenant_id != key.tenant_id:
continue
score = cosine(query_vec, vec)
if score > best_score:
best_score = score
best = (stored_key, stored_prompt, response, ts)
if best is None or best_score < self._threshold:
self._misses += 1
return CacheLookup(hit=False, score=max(0.0, best_score), payload=None)
self._hits += 1
stored_key, stored_prompt, response, ts = best
return CacheLookup(
hit=True,
score=best_score,
payload=asdict(response),
matched_prompt=stored_prompt,
age_seconds=max(0.0, time.time() - ts),
)
async
¶
Python
put(key: CacheKey, prompt: str, response: ServingCompletionResponse) -> None
async
¶
Source code in apogee_ai_serving/infrastructure/cache/in_memory_semantic_cache.py
async
¶
Python
JsonSemanticCache(root: str | Path, *, embedder: IPromptEmbedder | None = None, threshold: float = 0.85)
File-backed cache: <root>/<model>/<digest>.json.
Designed for single-process workloads. For concurrent processes use Redis or a dedicated KV store.
Source code in apogee_ai_serving/infrastructure/cache/json_semantic_cache.py
Python
def __init__(
self,
root: str | Path,
*,
embedder: IPromptEmbedder | None = None,
threshold: float = 0.85,
) -> None:
if not 0.0 <= threshold <= 1.0:
raise ValueError("threshold must be in [0,1]")
self._root = Path(root)
self._embedder = embedder or HashingPromptEmbedder()
self._threshold = threshold
self._hits = 0
self._misses = 0
async
¶
Python
get(key: CacheKey, *, prompt: str) -> CacheLookup
Source code in apogee_ai_serving/infrastructure/cache/json_semantic_cache.py
Python
async def get(self, key: CacheKey, *, prompt: str) -> CacheLookup:
# Exact match
path = self._model_dir(key.model) / f"{key.digest}.json"
if await asyncio.to_thread(path.is_file):
self._hits += 1
payload = await asyncio.to_thread(self._read, path)
return CacheLookup(
hit=True,
score=1.0,
payload=payload["response"],
matched_prompt=payload.get("prompt"),
age_seconds=max(0.0, time.time() - float(payload.get("ts", 0))),
)
# Semantic search
candidates = await asyncio.to_thread(self._scan, key.model, key.tenant_id)
if not candidates:
self._misses += 1
return CacheLookup(hit=False)
query_vec = await self._embedder.embed(prompt)
best_score = -1.0
best: dict | None = None
for entry in candidates:
score = cosine(query_vec, tuple(entry["embedding"]))
if score > best_score:
best_score = score
best = entry
if best is None or best_score < self._threshold:
self._misses += 1
return CacheLookup(hit=False, score=max(0.0, best_score))
self._hits += 1
return CacheLookup(
hit=True,
score=best_score,
payload=best["response"],
matched_prompt=best.get("prompt"),
age_seconds=max(0.0, time.time() - float(best.get("ts", 0))),
)
async
¶
Python
put(key: CacheKey, prompt: str, response: ServingCompletionResponse) -> None
async
¶
Source code in apogee_ai_serving/infrastructure/cache/json_semantic_cache.py
async
¶
Python
LMStudioEngine(*, base_url: str = 'http://localhost:1234', model_name: str = 'lmstudio-local', timeout: float = 60.0)
Bases: HttpCompletionsBase
LM Studio local server (OpenAI-compatible REST).
Source code in apogee_ai_serving/infrastructure/engines/lmstudio_engine.py
async
¶
Python
list_models() -> list[ModelDescriptor]
Source code in apogee_ai_serving/infrastructure/engines/lmstudio_engine.py
Adapter for llama-cpp-python in-process inference.
Lazy import: install via pip install 'apogee-ai-serving[llamacpp]'.
Source code in apogee_ai_serving/infrastructure/engines/llamacpp_engine.py
Python
def __init__(
self,
*,
model_path: str,
n_ctx: int = 4096,
n_threads: int | None = None,
) -> None:
try:
import llama_cpp # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"LlamaCppEngine requires `llama-cpp-python`. "
"Install with: pip install 'apogee-ai-serving[llamacpp]'"
) from exc
from llama_cpp import Llama # type: ignore
try:
self._llm = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_threads=n_threads,
verbose=False,
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
self._model_path = model_path
self._n_ctx = n_ctx
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/engines/llamacpp_engine.py
Python
async def complete(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
def _run() -> ServingCompletionResponse:
start = time.perf_counter()
try:
output = self._llm(
prompt=request.prompt,
max_tokens=request.max_tokens,
temperature=request.temperature,
top_p=request.top_p,
stop=list(request.stop) if request.stop else None,
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
choice = (output.get("choices") or [{}])[0]
usage = output.get("usage") or {}
return ServingCompletionResponse(
text=str(choice.get("text") or ""),
model=request.model or self._model_path,
engine=self.name,
prompt_tokens=int(usage.get("prompt_tokens", 0)),
completion_tokens=int(usage.get("completion_tokens", 0)),
total_tokens=int(usage.get("total_tokens", 0)),
latency_ms=latency,
finish_reason=choice.get("finish_reason"),
)
return await asyncio.to_thread(_run)
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
Source code in apogee_ai_serving/infrastructure/engines/llamacpp_engine.py
async
¶
Python
list_models() -> list[ModelDescriptor]
Source code in apogee_ai_serving/infrastructure/engines/llamacpp_engine.py
async
¶
async
¶
Adapter for mlx_lm (Apple Silicon native).
Lazy import: install via pip install 'apogee-ai-serving[mlx]'.
Source code in apogee_ai_serving/infrastructure/engines/mlx_engine.py
Python
def __init__(self, *, model_path: str = "mlx-community/Llama-3.2-3B-Instruct-4bit") -> None:
try:
import mlx_lm # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"MlxEngine requires `mlx-lm`. "
"Install with: pip install 'apogee-ai-serving[mlx]'"
) from exc
from mlx_lm import load # type: ignore
try:
self._model, self._tokenizer = load(model_path)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
self._model_path = model_path
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/engines/mlx_engine.py
Python
async def complete(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
def _run() -> ServingCompletionResponse:
from mlx_lm import generate # type: ignore
start = time.perf_counter()
try:
text = generate(
self._model,
self._tokenizer,
prompt=request.prompt,
max_tokens=request.max_tokens,
verbose=False,
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
return ServingCompletionResponse(
text=str(text),
model=request.model or self._model_path,
engine=self.name,
latency_ms=latency,
finish_reason="stop",
)
return await asyncio.to_thread(_run)
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
Source code in apogee_ai_serving/infrastructure/engines/mlx_engine.py
async
¶
Python
list_models() -> list[ModelDescriptor]
async
¶
async
¶
Adapter for Ollama local server (ollama serve).
Lazy import: install via pip install 'apogee-ai-serving[ollama]'.
Source code in apogee_ai_serving/infrastructure/engines/ollama_engine.py
Python
def __init__(
self,
*,
host: str = "http://localhost:11434",
model_name: str = "llama3.2",
) -> None:
try:
import ollama # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"OllamaEngine requires `ollama`. "
"Install with: pip install 'apogee-ai-serving[ollama]'"
) from exc
self._host = host
self._model = model_name
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/engines/ollama_engine.py
Python
async def complete(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
try:
from ollama import AsyncClient # type: ignore
except ImportError as exc: # pragma: no cover
raise EngineNotAvailableException(self.name, str(exc)) from exc
client = AsyncClient(host=self._host)
start = time.perf_counter()
try:
response = await client.generate(
model=request.model or self._model,
prompt=request.prompt,
options={
"num_predict": request.max_tokens,
"temperature": request.temperature,
"top_p": request.top_p,
"stop": list(request.stop) if request.stop else None,
},
stream=False,
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
prompt_tokens = int(response.get("prompt_eval_count", 0) or 0)
completion_tokens = int(response.get("eval_count", 0) or 0)
return ServingCompletionResponse(
text=str(response.get("response") or ""),
model=request.model or self._model,
engine=self.name,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
latency_ms=latency,
finish_reason="stop" if response.get("done") else None,
)
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
Source code in apogee_ai_serving/infrastructure/engines/ollama_engine.py
Python
async def stream(
self, request: ServingCompletionRequest
) -> AsyncIterator[ServingChunk]:
try:
from ollama import AsyncClient # type: ignore
except ImportError as exc: # pragma: no cover
raise EngineNotAvailableException(self.name, str(exc)) from exc
client = AsyncClient(host=self._host)
async def gen() -> AsyncIterator[ServingChunk]:
seq = 0
try:
async for chunk in await client.generate(
model=request.model or self._model,
prompt=request.prompt,
options={"num_predict": request.max_tokens},
stream=True,
):
yield ServingChunk(
delta=str(chunk.get("response") or ""),
sequence=seq,
is_final=bool(chunk.get("done")),
)
seq += 1
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
return gen()
async
¶
Python
list_models() -> list[ModelDescriptor]
Source code in apogee_ai_serving/infrastructure/engines/ollama_engine.py
Python
async def list_models(self) -> list[ModelDescriptor]:
try:
from ollama import AsyncClient # type: ignore
except ImportError as exc: # pragma: no cover
raise EngineNotAvailableException(self.name, str(exc)) from exc
client = AsyncClient(host=self._host)
try:
response = await client.list()
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
items = response.get("models", []) if isinstance(response, dict) else []
return [
ModelDescriptor(
name=str(item.get("name") or item.get("model") or self._model),
engine=EngineKind.OLLAMA,
tier=ModelTier.CHEAP,
context_window=8192,
)
for item in items
]
async
¶
Source code in apogee_ai_serving/infrastructure/engines/ollama_engine.py
async
¶
Python
OrderedFallbackChain(registry: EngineRegistry, policy: FallbackPolicy)
Walks FallbackPolicy.engines in order until one succeeds.
Source code in apogee_ai_serving/infrastructure/fallback/ordered_fallback_chain.py
async
¶
Python
execute(request: ServingCompletionRequest, *, primary_engine: str) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/fallback/ordered_fallback_chain.py
Python
async def execute(
self,
request: ServingCompletionRequest,
*,
primary_engine: str,
) -> ServingCompletionResponse:
# Compose ordered chain: primary first, then policy.engines (dedup)
chain: list[str] = [primary_engine]
for name in self._policy.engines:
if name != primary_engine and name not in chain:
chain.append(name)
if self._policy.max_attempts is not None:
chain = chain[: self._policy.max_attempts]
attempts: list[str] = []
last_error: str = ""
for engine_name in chain:
attempts.append(engine_name)
engine = self._registry.find(engine_name)
if engine is None:
last_error = f"engine {engine_name!r} not registered"
continue
try:
response = await engine.complete(request)
return replace(response, fallback_chain=tuple(attempts))
except RateLimitExceededException as exc:
last_error = str(exc)
if not self._policy.retry_on_rate_limit:
break
except EngineNotAvailableException as exc:
last_error = str(exc)
if not self._policy.retry_on_engine_error:
break
except Exception as exc: # noqa: BLE001
last_error = repr(exc)
if not self._policy.retry_on_engine_error:
break
raise FallbackExhaustedException(tuple(attempts), last_error)
Python
RedisRateLimiter(*, redis_url: str = 'redis://localhost:6379/0', default_limit: RateLimit | None = None, raise_on_exceeded: bool = False, namespace: str = 'apogee:ratelimit')
Atomic token-bucket via Lua. Survives multi-process workers.
Lazy import: install via pip install 'apogee-ai-serving[redis]'.
Source code in apogee_ai_serving/infrastructure/quota/redis_rate_limiter.py
Python
def __init__(
self,
*,
redis_url: str = "redis://localhost:6379/0",
default_limit: RateLimit | None = None,
raise_on_exceeded: bool = False,
namespace: str = "apogee:ratelimit",
) -> None:
try:
import redis # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"RedisRateLimiter requires `redis`. "
"Install with: pip install 'apogee-ai-serving[redis]'"
) from exc
from redis.asyncio import Redis # type: ignore
self._client = Redis.from_url(redis_url, decode_responses=True)
self._default = default_limit or RateLimit()
self._raise = raise_on_exceeded
self._namespace = namespace
self._script = self._client.register_script(_LUA_SCRIPT)
async
¶
Source code in apogee_ai_serving/infrastructure/quota/redis_rate_limiter.py
Python
async def acquire(
self,
*,
scope_key: str,
cost_tokens: int = 0, # noqa: ARG002
) -> bool:
import time
try:
allowed, _tokens = await self._script(
keys=[self._key(scope_key)],
args=[
self._default.bucket_size,
self._default.requests_per_minute / 60.0,
time.time(),
],
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
if not int(allowed):
if self._raise:
raise RateLimitExceededException(
scope_key, self._default.requests_per_minute
)
return False
return True
async
¶
Python
remaining(*, scope_key: str) -> QuotaSnapshot
Source code in apogee_ai_serving/infrastructure/quota/redis_rate_limiter.py
Python
async def remaining(self, *, scope_key: str) -> QuotaSnapshot:
try:
tokens = await self._client.hget(self._key(scope_key), "tokens")
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
try:
tokens_value = float(tokens) if tokens is not None else float(self._default.bucket_size)
except (TypeError, ValueError):
tokens_value = float(self._default.bucket_size)
return QuotaSnapshot(
scope=scope_key,
requests_remaining=int(tokens_value),
requests_per_minute=self._default.requests_per_minute,
tokens_remaining=None,
reset_in_seconds=60.0,
)
async
¶
Python
RedisSemanticCache(*, redis_url: str = 'redis://localhost:6379/0', embedder: IPromptEmbedder | None = None, threshold: float = 0.85, ttl_seconds: int | None = 3600, namespace: str = 'apogee:serving')
Redis-backed cache.
Stores each entry under apogee:serving:{model}:{digest} with the
serialized payload (prompt + embedding + response + ts). Semantic
search scans a SCAN-cursor over the model namespace; for very large
deployments use RediSearch with HNSW.
Lazy import: install via pip install 'apogee-ai-serving[redis]'.
Source code in apogee_ai_serving/infrastructure/cache/redis_semantic_cache.py
Python
def __init__(
self,
*,
redis_url: str = "redis://localhost:6379/0",
embedder: IPromptEmbedder | None = None,
threshold: float = 0.85,
ttl_seconds: int | None = 3600,
namespace: str = "apogee:serving",
) -> None:
try:
import redis # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"RedisSemanticCache requires `redis`. "
"Install with: pip install 'apogee-ai-serving[redis]'"
) from exc
if not 0.0 <= threshold <= 1.0:
raise ValueError("threshold must be in [0,1]")
from redis.asyncio import Redis # type: ignore
self._client = Redis.from_url(redis_url, decode_responses=True)
self._embedder = embedder or HashingPromptEmbedder()
self._threshold = threshold
self._ttl = ttl_seconds
self._namespace = namespace
self._hits = 0
self._misses = 0
async
¶
Python
get(key: CacheKey, *, prompt: str) -> CacheLookup
Source code in apogee_ai_serving/infrastructure/cache/redis_semantic_cache.py
Python
async def get(self, key: CacheKey, *, prompt: str) -> CacheLookup:
try:
exact = await self._client.get(self._key(key))
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
if exact is not None:
self._hits += 1
data = json.loads(exact)
return CacheLookup(
hit=True,
score=1.0,
payload=data["response"],
matched_prompt=data.get("prompt"),
age_seconds=max(0.0, time.time() - float(data.get("ts", 0))),
)
# Semantic scan
try:
candidates = await self._scan_candidates(key)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
if not candidates:
self._misses += 1
return CacheLookup(hit=False)
query_vec = await self._embedder.embed(prompt)
best_score = -1.0
best: dict | None = None
for entry in candidates:
score = cosine(query_vec, tuple(entry["embedding"]))
if score > best_score:
best_score = score
best = entry
if best is None or best_score < self._threshold:
self._misses += 1
return CacheLookup(hit=False, score=max(0.0, best_score))
self._hits += 1
return CacheLookup(
hit=True,
score=best_score,
payload=best["response"],
matched_prompt=best.get("prompt"),
age_seconds=max(0.0, time.time() - float(best.get("ts", 0))),
)
async
¶
Python
put(key: CacheKey, prompt: str, response: ServingCompletionResponse) -> None
Source code in apogee_ai_serving/infrastructure/cache/redis_semantic_cache.py
Python
async def put(
self,
key: CacheKey,
prompt: str,
response: ServingCompletionResponse,
) -> None:
vec = await self._embedder.embed(prompt)
payload = {
"prompt": prompt,
"embedding": list(vec),
"response": asdict(response),
"ts": time.time(),
}
try:
await self._client.set(
self._key(key),
json.dumps(payload, default=str),
ex=self._ttl,
)
except Exception as exc: # noqa: BLE001
raise EngineNotAvailableException(self.name, str(exc)) from exc
async
¶
Source code in apogee_ai_serving/infrastructure/cache/redis_semantic_cache.py
Python
async def stats(self) -> dict[str, float]:
total = self._hits + self._misses
try:
keys = await self._client.dbsize()
except Exception: # noqa: BLE001
keys = 0
return {
"hits": float(self._hits),
"misses": float(self._misses),
"hit_rate": (self._hits / total) if total else 0.0,
"entries": float(keys),
}
async
¶
Source code in apogee_ai_serving/infrastructure/cache/redis_semantic_cache.py
Python
async def clear(self) -> None:
try:
cursor = 0
while True:
cursor, keys = await self._client.scan(
cursor=cursor, match=f"{self._namespace}:*"
)
if keys:
await self._client.delete(*keys)
if cursor == 0:
break
except Exception: # noqa: BLE001 - best effort
return
self._hits = 0
self._misses = 0
Python
TgiEngine(*, base_url: str = 'http://localhost:8080', model_name: str = 'tgi-default', timeout: float = 60.0)
HuggingFace Text Generation Inference (text-generation-inference).
Source code in apogee_ai_serving/infrastructure/engines/tgi_engine.py
async
¶
Python
complete(request: ServingCompletionRequest) -> ServingCompletionResponse
Source code in apogee_ai_serving/infrastructure/engines/tgi_engine.py
Python
async def complete(
self, request: ServingCompletionRequest
) -> ServingCompletionResponse:
payload = {
"inputs": request.prompt,
"parameters": {
"max_new_tokens": request.max_tokens,
"temperature": request.temperature,
"top_p": request.top_p,
"stop": list(request.stop) if request.stop else None,
"return_full_text": False,
},
}
start = time.perf_counter()
try:
async with httpx.AsyncClient(timeout=self._timeout) as client:
response = await client.post(f"{self._base_url}/generate", json=payload)
response.raise_for_status()
data = response.json()
except httpx.HTTPError as exc:
raise EngineNotAvailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
text = data.get("generated_text", "") if isinstance(data, dict) else ""
details = data.get("details") if isinstance(data, dict) else None
completion_tokens = (
int(details.get("generated_tokens", 0)) if isinstance(details, dict) else 0
)
return ServingCompletionResponse(
text=text,
model=request.model or self._model,
engine=self.name,
completion_tokens=completion_tokens,
total_tokens=completion_tokens,
latency_ms=latency,
finish_reason=
details.get("finish_reason") if isinstance(details, dict) else None,
)
async
¶
Python
stream(request: ServingCompletionRequest) -> AsyncIterator[ServingChunk]
Source code in apogee_ai_serving/infrastructure/engines/tgi_engine.py
async
¶
Python
list_models() -> list[ModelDescriptor]
async
¶
async
¶
Python
TierAwareRouter(*, tier_to_model: Mapping[ModelTier, tuple[str, str]], rules: tuple[RouterRule, ...] = _DEFAULT_RULES)
Picks the cheapest model that fits the prompt complexity.
Tier mapping is supplied as tier_to_model: ModelTier →
(engine_name, model_name). Routing strategy is EXPLICIT when
the request already names a model.
Source code in apogee_ai_serving/infrastructure/router/tier_aware_router.py
Python
route(request: ServingCompletionRequest) -> RoutingDecision
Source code in apogee_ai_serving/infrastructure/router/tier_aware_router.py
Python
def route(self, request: ServingCompletionRequest) -> RoutingDecision:
if request.model:
engine = self._engine_for_explicit_model(request.model)
return RoutingDecision(
engine=engine,
model=request.model,
tier=ModelTier.BALANCED,
strategy=RoutingStrategy.EXPLICIT,
reason="caller specified model",
)
tier = self._pick_tier(request.prompt)
if tier not in self._tier_to_model:
raise NoEngineMatchedException(request.prompt[:64])
engine, model = self._tier_to_model[tier]
return RoutingDecision(
engine=engine,
model=model,
tier=tier,
strategy=RoutingStrategy.TIER_AWARE,
reason=f"prompt mapped to tier {tier.value}",
)
Python
VllmEngine(*, base_url: str = 'http://localhost:8000', model_name: str = 'meta-llama/Llama-3.1-8B-Instruct', timeout: float = 60.0)
Bases: HttpCompletionsBase
vLLM via its OpenAI-compatible HTTP server (vllm serve).
Source code in apogee_ai_serving/infrastructure/engines/vllm_engine.py
async
¶
Python
list_models() -> list[ModelDescriptor]