مدلهای جدید Fireworks.ai و مسیریابی بهتر cache
تاریخ: ۱۴۰۵-۰۵-۲۲ / (2026-08-13)
خلاصه
مدل Meta با شناسه muse-glimmer-30b و مدل NVIDIA با شناسه nemotron-3.5-lightning اکنون از طریق Fireworks.ai در AvalAI در دسترس هستند. این انتشار همچنین cache hit را بهشکل چشمگیری بهتر میکند: مسیریاب بهصورت best effort و بر اساس هر کاربر و مدل، آخرین زیرساخت سالم و موفق را در یک بازه sticky و rolling برابر 15 دقیقه ترجیح میدهد.
جزئیات
Fireworks.ai
Muse Glimmer 30B
muse-glimmer-30b مدلی از Meta است که Fireworks.ai آن را میزبانی میکند. این مدل از reasoning، فراخوانی تابع، ورودی تصویری و prompt caching پشتیبانی میکند و پنجره context آن ۱۳۱٬۰۷۲ توکن است.
Nemotron 3.5 Lightning
nemotron-3.5-lightning مدلی از NVIDIA با میزبانی Fireworks.ai است. این مدل برای reasoning و workloadهای agentic کارآمد طراحی شده و در route مربوط به AvalAI، فراخوانی تابع، prompt caching و پنجره context برابر با ۲۶۲٬۱۴۴ توکن دارد.
قیمتگذاری
قیمتها بهازای یک میلیون توکن هستند.
| مدل | توسعهدهنده | میزبان | ورودی | ورودی cacheشده | خروجی |
|---|---|---|---|---|---|
muse-glimmer-30b | Meta | Fireworks.ai | $0.35 | $0.04 | $1.50 |
nemotron-3.5-lightning | NVIDIA | Fireworks.ai | $0.05 | $0.01 | $0.20 |
دسترسپذیری endpointها
| مدل | v1/chat/completions | v1/messages | v1/responses |
|---|---|---|---|
muse-glimmer-30b | کامل | کامل | جزئی |
nemotron-3.5-lightning | کامل | کامل | جزئی |
مسیریابی بهتر cache
AvalAI چند provider و مجموعه زیرساخت را تجمیع میکند. پیش از این، درخواستهای تکراری میتوانستند به زیرساختهای متفاوت برسند و در مقایسه با ترافیک مستقیم provider، cache hit کمتری از کش ضمنی ارائهدهنده دریافت کنند.
مسیریاب هوشمند اکنون مسیرهای موفق را در سطح هر کاربر، هر مدل و هر زیرساخت درک و نگهداری میکند. برای همان کاربر و مدل، آخرین زیرساخت سالم و موفق تا 15 دقیقه پس از آخرین درخواست موفق ترجیح داده میشود. هر موفقیت بازه sticky و rolling پانزدهدقیقهای را دوباره تمدید میکند و احتمال رسیدن prefix تکراری به زیرساخت دارای cache را بسیار بیشتر میکند.
Affinity یک رفتار best effort است و تضمین نمیشود. سلامت، ظرفیت، failover، دسترسپذیری یا رفتار provider میتواند درخواست را جابهجا کند. انتشار cache key و affinity همچنان زمان میبرد. تست کنترلشده deepseek-v4-flash پس از 15 ثانیه تقریبا آنی بود، اما deepseek-v4-pro به زمان بیشتری نیاز داشت تا اجرای بعدی hitهای پایدار نشان دهد. بازه sticky و rolling پانزدهدقیقهای پس از شکلگیری route موفق اعمال میشود و deadline عمومی 15 ثانیهای برای فعالشدن نیست.
بنچمارک paired کش DeepSeek
یک بنچمارک streaming در 2026-08-13، AvalAI و endpoint رسمی DeepSeek را با یک prefix دارای byteهای یکسان priming کرد، 15 ثانیه صبر کرد و سپس 10 round ترتیبی با deepseek-v4-flash اجرا کرد. در rounds 2–10 میانگین warm cache hit هر دو endpoint برابر 98.6% و اختلاف برابر +0.0 percentage points بود. اولین hit اندازهگیریشده AvalAI در round 2 رخ داد. DeepSeek بهطور غیرمنتظره در round 1 نیز hit برابر 98.6% گزارش کرد؛ بنابراین roundهای warm، نه یک cold miss فرضی در endpoint رسمی، مقایسه درستتری هستند.
این نتیجه مربوط به یک مدل و یک اجرا است و affinity بدون delay یا hit rate سراسری سرویس را تضمین نمیکند. نتیجههای دقیق، نمودارها، commandها، روش اجرا و محدودیتها را در عملکرد AvalAI ببینید. سورس کامل و بدون خلاصه بنچمارک نیز در ادامه همین صفحه، داخل یک code block قابل کپی آمده است؛ adapter ارائهدهندهها، timeoutها، خروجی JSON و ساخت نمودار همگی در آن قرار دارند. برای اجرا باید AVALAI_API_KEY و کلید provider رسمی انتخابشده را تنظیم کنید؛ این آزمایش به هر دو endpoint درخواست واقعی و احتمالا هزینهدار میفرستد.
نمایش سورس کامل Python بنچمارک
python
"""Prompt-cache hit-ratio comparison by service and OpenAI API endpoint.
This integration benchmark measures **prompt-caching effectiveness** for a
single model by driving the *same* caching experiment against two services and
both OpenAI text-generation endpoints:
* **avalai** — ``https://api.avalai.ir/v1`` (OpenAI-compatible proxy)
* **official** — the model's upstream provider (e.g. ``deepseek.com`` for
``deepseek-v4-flash``, ``openai`` for ``gpt-*``, ``claude.ai`` for
``claude-*``, ...).
* **v1/chat/completions** — repeated realistic sample requests.
* **v1/responses** — one response chain per service; rounds 2..N send the prior
response's id as ``previous_response_id`` to exercise session affinity.
Supported official providers: ``openai``, ``claude.ai`` (anthropic),
``deepseek``, ``z.ai``, ``dashscope`` (Alibaba/Qwen), ``moonshot.ai``,
``gemini``, ``xai``.
How prompt caching is measured
------------------------------
The benchmark repeatedly sends the same realistic, sufficiently-large sample
conversation. It does **not** infer a hit from prompt text, timing, seed reuse,
or request order. A call is a cache hit only when the API response's ``usage``
dict reports cached input tokens.
For every service/endpoint pair, the complete reported ``usage`` dict is saved
and the ratio is calculated strictly as
``usage.prompt_tokens_details.cached_tokens / usage.prompt_tokens`` (or the
provider's documented equivalent). The table, summary, JSON artifact, and
charts under ``tests/benchmarks/results/`` all use those reported values.
Cache-usage field shapes differ per provider (all normalised by the
extractors below):
* OpenAI / DeepSeek / z.ai / DashScope / Moonshot / xAI (OpenAI-compatible) ->
``usage.prompt_tokens_details.cached_tokens`` (DeepSeek additionally exposes
``prompt_cache_hit_tokens`` / ``prompt_cache_miss_tokens``).
* Anthropic (claude.ai) -> top-level ``usage.cache_read_input_tokens`` +
``usage.cache_creation_input_tokens`` (requires an explicit
``cache_control: {"type": "ephemeral"}`` breakpoint on a content block).
* Gemini -> ``usageMetadata.cachedContentTokenCount``.
Usage
-----
python -m tests.benchmarks.test_cache_hit_ratio \\
--model deepseek-v4-flash \\
--rounds 5 --prefix-tokens 4000
Provide keys via flags (``--avalai-api-key``, ``--deepseek-api-key``, ...) or
the standard environment variables (``AVALAI_API_KEY``, ``DEEPSEEK_API_KEY``,
...). Missing keys fail fast *before* any AvalAI request is issued.
"""
from __future__ import annotations
import argparse
import asyncio
import contextlib
import json
import os
import statistics
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, AsyncIterator, Callable, Mapping, Sequence
RESULTS_DIR = Path(__file__).resolve().parent / "results"
# The sample must clear the provider's activation floor. ~2048 tokens is a safe
# default across the supported providers.
DEFAULT_MIN_CACHE_TOKENS = 2048
# ---------------------------------------------------------------------------
# Cache-usage extractors
#
# Each extractor takes a provider ``usage`` object (SDK model or dict) and
# returns a normalised ``(prompt_tokens, cached_tokens, cache_creation_tokens)``
# tuple. They tolerate both attribute access (SDK models) and mapping access
# (raw dicts / LiteLLM envelopes).
# ---------------------------------------------------------------------------
def _get(obj: Any, key: str, default: Any = None) -> Any:
"""Read ``key`` from ``obj`` whether it is a mapping or an object."""
if obj is None:
return default
if isinstance(obj, Mapping):
return obj.get(key, default)
return getattr(obj, key, default)
def _as_int(value: Any) -> int:
try:
if value is None:
return 0
return int(value)
except (TypeError, ValueError):
return 0
def usage_to_dict(usage: Any) -> dict[str, Any]:
"""Return the raw provider ``usage`` object as a plain, JSON-safe dict.
The cache-hit ratio is judged SOLELY from this reported ``usage`` dict —
never from any prefix/seed assumption. This helper normalises the many
shapes ``usage`` can arrive in (OpenAI SDK model, Anthropic SDK model,
LiteLLM envelope, or a raw dict) into a plain dict so the exact numbers
the API returned are recorded verbatim in the JSON artifact and are what
every ratio is computed from.
"""
if usage is None:
return {}
if isinstance(usage, Mapping):
return {str(k): _jsonable(v) for k, v in usage.items()}
# Pydantic v2 SDK models.
for dumper in ("model_dump", "dict"):
fn = getattr(usage, dumper, None)
if callable(fn):
try:
dumped = fn()
if isinstance(dumped, Mapping):
return {str(k): _jsonable(v) for k, v in dumped.items()}
except Exception: # noqa: BLE001 - fall through to attr scrape
pass
# Last resort: scrape public attributes off an arbitrary object.
out: dict[str, Any] = {}
for name in dir(usage):
if name.startswith("_"):
continue
try:
val = getattr(usage, name)
except Exception: # noqa: BLE001
continue
if callable(val):
continue
out[name] = _jsonable(val)
return out
def _jsonable(value: Any) -> Any:
"""Coerce ``value`` into a JSON-serialisable structure."""
if value is None or isinstance(value, (bool, int, float, str)):
return value
if isinstance(value, Mapping):
return {str(k): _jsonable(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_jsonable(v) for v in value]
for dumper in ("model_dump", "dict"):
fn = getattr(value, dumper, None)
if callable(fn):
try:
dumped = fn()
if isinstance(dumped, Mapping):
return {str(k): _jsonable(v) for k, v in dumped.items()}
except Exception: # noqa: BLE001
pass
# Nested SDK sub-model (e.g. prompt_tokens_details) — scrape attributes.
nested: dict[str, Any] = {}
for name in dir(value):
if name.startswith("_"):
continue
try:
sub = getattr(value, name)
except Exception: # noqa: BLE001
continue
if callable(sub):
continue
nested[name] = _jsonable(sub)
return nested or str(value)
def extract_openai_style(usage: Any) -> tuple[int, int, int]:
"""OpenAI-style Chat Completions or Responses API usage.
The ratio is derived ONLY from the provider-reported ``usage`` dict.
Chat Completions uses ``prompt_tokens`` / ``prompt_tokens_details`` while
Responses uses ``input_tokens`` / ``input_tokens_details``. DeepSeek also
reports ``prompt_cache_hit_tokens``, which is retained as a fallback.
"""
prompt_tokens = _as_int(_get(usage, "prompt_tokens"))
details = _get(usage, "prompt_tokens_details")
if prompt_tokens == 0:
prompt_tokens = _as_int(_get(usage, "input_tokens"))
details = _get(usage, "input_tokens_details")
cached = _as_int(_get(details, "cached_tokens"))
if cached == 0:
# DeepSeek-native fields (present on both avalai passthrough and
# api.deepseek.com direct responses).
cached = _as_int(_get(usage, "prompt_cache_hit_tokens"))
cache_creation = _as_int(_get(details, "cache_creation_tokens"))
return prompt_tokens, cached, cache_creation
def extract_anthropic_style(usage: Any) -> tuple[int, int, int]:
"""Extract native Anthropic or AvalAI-normalised Claude usage.
Native Anthropic reports ``input_tokens``, ``cache_read_input_tokens`` and
``cache_creation_input_tokens`` at the top level. AvalAI returns Claude
usage in the OpenAI-compatible ``prompt_tokens`` / ``prompt_tokens_details``
shape, so fall back to that representation when native fields are absent.
"""
input_tokens = _as_int(_get(usage, "input_tokens"))
cache_read = _as_int(_get(usage, "cache_read_input_tokens"))
cache_creation = _as_int(_get(usage, "cache_creation_input_tokens"))
if input_tokens or cache_read or cache_creation:
prompt_tokens = input_tokens + cache_read + cache_creation
return prompt_tokens, cache_read, cache_creation
return extract_openai_style(usage)
def extract_gemini_style(usage: Any) -> tuple[int, int, int]:
"""Gemini ``usageMetadata`` -> ``cachedContentTokenCount``.
When routed through the OpenAI-compatible endpoint the counts land in the
OpenAI shape, so we try that first and fall back to the native camelCase
field names.
"""
prompt_tokens = _as_int(_get(usage, "prompt_tokens"))
if prompt_tokens == 0:
prompt_tokens = _as_int(_get(usage, "promptTokenCount"))
details = _get(usage, "prompt_tokens_details")
cached = _as_int(_get(details, "cached_tokens"))
if cached == 0:
cached = _as_int(_get(usage, "cachedContentTokenCount"))
if cached == 0:
cached = _as_int(_get(usage, "cached_content_token_count"))
return prompt_tokens, cached, 0
# ---------------------------------------------------------------------------
# Provider registry
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ProviderSpec:
"""Static configuration for one official provider."""
key: str # canonical registry key, e.g. "deepseek"
label: str # human label, e.g. "deepseek.com"
base_url: str # official OpenAI-compatible base URL
env_var: str # env var holding the official key
cli_flag: str # argparse flag, e.g. "--deepseek-api-key"
dest: str # argparse dest, e.g. "deepseek_api_key"
transport: str # "openai" | "anthropic"
extractor: Callable[[Any], tuple[int, int, int]]
min_cache_tokens: int = DEFAULT_MIN_CACHE_TOKENS
# Model-name prefixes that map to this provider (lower-cased).
model_prefixes: tuple[str, ...] = ()
def _spec(key: str, label: str, base_url: str, env_var: str, transport: str,
extractor: Callable[[Any], tuple[int, int, int]],
model_prefixes: tuple[str, ...],
min_cache_tokens: int = DEFAULT_MIN_CACHE_TOKENS) -> ProviderSpec:
return ProviderSpec(
key=key,
label=label,
base_url=base_url,
env_var=env_var,
cli_flag=f"--{key.replace('_', '-')}-api-key",
dest=f"{key}_api_key",
transport=transport,
extractor=extractor,
min_cache_tokens=min_cache_tokens,
model_prefixes=model_prefixes,
)
PROVIDER_REGISTRY: dict[str, ProviderSpec] = {
"openai": _spec(
"openai", "openai", "https://api.openai.com/v1",
"OPENAI_API_KEY", "openai", extract_openai_style,
("gpt-", "o1", "o3", "o4", "chatgpt-"),
min_cache_tokens=1024, # OpenAI auto-caches prompts > 1024 tokens.
),
"anthropic": _spec(
"anthropic", "claude.ai", "https://api.anthropic.com",
"ANTHROPIC_API_KEY", "anthropic", extract_anthropic_style,
("claude-", "claude."),
),
"deepseek": _spec(
"deepseek", "deepseek.com", "https://api.deepseek.com",
"DEEPSEEK_API_KEY", "openai", extract_openai_style,
("deepseek-", "deepseek."),
),
"zai": _spec(
"zai", "z.ai", "https://api.z.ai/api/paas/v4",
"ZAI_API_KEY", "openai", extract_openai_style,
("glm-", "glm."),
),
"dashscope": _spec(
"dashscope", "dashscope",
"https://dashscope-intl.aliyuncs.com/compatible-mode/v1",
"DASHSCOPE_API_KEY", "openai", extract_openai_style,
("qwen-", "qwen.", "qwen2", "qwen3", "qwq-", "qwq.",
"wan-", "wan.", "alibaba-", "alibaba."),
),
"moonshot": _spec(
"moonshot", "moonshot.ai", "https://api.moonshot.ai/v1",
"MOONSHOT_API_KEY", "openai", extract_openai_style,
("kimi-", "kimi.", "moonshot-", "moonshot-v"),
),
"gemini": _spec(
"gemini", "gemini",
"https://generativelanguage.googleapis.com/v1beta/openai",
"GEMINI_API_KEY", "openai", extract_gemini_style,
("gemini-", "gemini."),
),
"xai": _spec(
"xai", "xai", "https://api.x.ai/v1",
"XAI_API_KEY", "openai", extract_openai_style,
("grok-", "grok."),
),
}
# Friendly aliases accepted for --official-provider.
PROVIDER_ALIASES: dict[str, str] = {
"claude.ai": "anthropic",
"claude": "anthropic",
"anthropic": "anthropic",
"z.ai": "zai",
"zai": "zai",
"dashscope": "dashscope",
"alibaba": "dashscope",
"aliyun": "dashscope",
"qwen": "dashscope",
"moonshot.ai": "moonshot",
"moonshot": "moonshot",
"kimi": "moonshot",
"deepseek.com": "deepseek",
"deepseek": "deepseek",
"openai": "openai",
"gemini": "gemini",
"google": "gemini",
"xai": "xai",
"x.ai": "xai",
"grok": "xai",
}
AVALAI_DEFAULT_BASE_URL = "https://api.avalai.ir/v1"
AVALAI_ENV_VAR = "AVALAI_API_KEY"
def resolve_provider(model: str, override: str | None) -> ProviderSpec:
"""Resolve the official provider for ``model``.
An explicit ``--official-provider`` override wins; otherwise the provider
is inferred from the model-name prefix.
"""
if override:
canonical = PROVIDER_ALIASES.get(override.strip().lower())
if canonical is None:
raise ValueError(
f"Unknown official provider '{override}'. Supported: "
f"{', '.join(sorted(set(PROVIDER_ALIASES)))}"
)
return PROVIDER_REGISTRY[canonical]
lowered = model.strip().lower()
for spec in PROVIDER_REGISTRY.values():
for prefix in spec.model_prefixes:
if lowered.startswith(prefix):
return spec
raise ValueError(
f"Could not infer an official provider from model '{model}'. "
f"Pass --official-provider explicitly (one of: "
f"{', '.join(sorted(PROVIDER_REGISTRY))})."
)
# ---------------------------------------------------------------------------
# Metrics dataclasses
# ---------------------------------------------------------------------------
@dataclass
class RoundMetrics:
round_id: int
endpoint: str # "avalai" | official label
api: str # "chat.completions" | "responses"
prompt_tokens: int
cached_tokens: int
cache_creation_tokens: int
cache_hit_ratio: float
latency: float
ttft: float
output_tokens: int
# Complete usage object reported by the API. The normalized fields above
# are derived only from this dict; it is persisted for independent audit.
usage: dict[str, Any] = field(default_factory=dict)
error: str | None = None
@dataclass
class EndpointSummary:
endpoint: str
api: str
rounds: int
successful: int
mean_hit_ratio: float
median_hit_ratio: float
p90_hit_ratio: float
first_hit_round: int | None
first_hit_latency: float | None
mean_latency: float
mean_ttft: float
warm_mean_hit_ratio: float # mean over rounds >= 2 (warm cache)
per_round: list[RoundMetrics] = field(default_factory=list)
# ---------------------------------------------------------------------------
# Sample request construction
# ---------------------------------------------------------------------------
# ~4 characters per token sizes the realistic legal-document sample. It is not
# used to determine whether caching happened; only response.usage is.
_CHARS_PER_TOKEN = 4
_LEGAL_TEXT = (
"Here is the full text of a complex legal agreement. The agreement defines "
"the parties, term, payment obligations, confidentiality duties, warranties, "
"liability limits, termination rights, dispute resolution, governing law, "
"data protection requirements, and amendment procedures. "
)
def build_sample_document(target_tokens: int, min_tokens: int) -> str:
"""Build the realistic repeated legal-document sample used by each call."""
target_chars = max(target_tokens, min_tokens) * _CHARS_PER_TOKEN
introduction = (
"You are an AI assistant tasked with analyzing legal documents. "
"Read the following agreement and answer the user's question.\n\n"
)
repetitions = max(1, (target_chars - len(introduction)) // len(_LEGAL_TEXT) + 1)
return introduction + (_LEGAL_TEXT * repetitions)
def build_user_question(_round_id: int) -> str:
"""Return the same realistic question on every call."""
return "What are the key terms and conditions in this agreement?"
# ---------------------------------------------------------------------------
# Endpoint clients
#
# ``EndpointClient`` abstracts the two transports:
# * "openai" -> openai.AsyncOpenAI (avalai + openai-compatible officials)
# * "anthropic" -> anthropic.AsyncAnthropic (claude.ai)
# Each returns a normalised ``(prompt_tokens, cached, cache_creation,
# output_tokens, ttft, latency)`` tuple for one round.
# ---------------------------------------------------------------------------
@dataclass
class EndpointConfig:
name: str # "avalai" or the official provider label
transport: str # "openai" | "anthropic"
base_url: str
api_key: str
model: str
extractor: Callable[[Any], tuple[int, int, int]]
max_tokens: int = 64
# Wall-clock ceiling for a whole round (connect + full stream). A stalled
# proxy/upstream that keeps the socket open but stops emitting bytes must
# not hang the run forever, so this bounds the total request duration.
request_timeout: float = 120.0
# Idle guard: max seconds allowed *between* two consecutive stream chunks.
# Trips well before ``request_timeout`` when a stream stalls mid-flight
# (e.g. AvalAI stops sending the final usage/priced chunk).
stream_idle_timeout: float = 60.0
class StreamStalled(TimeoutError):
"""Raised when a stream exceeds its idle-gap or wall-clock budget.
Subclasses ``TimeoutError`` so callers that only distinguish "timed out"
keep working, while the message pinpoints *which* guard tripped.
"""
async def _iter_with_idle_guard(
stream: AsyncIterator[Any],
*,
idle_timeout: float,
deadline: float,
) -> AsyncIterator[Any]:
"""Yield from ``stream`` while enforcing an idle-gap + wall-clock guard.
A proxy that keeps the socket open but stops emitting bytes (the classic
"avalai stucks" mid-stream stall) would otherwise block ``async for``
forever, since the SDK stream iterators have no read-idle timeout. This
wraps each ``__anext__`` in ``asyncio.wait_for`` bounded by the smaller of
the remaining wall-clock budget and ``idle_timeout``, raising
``StreamStalled`` instead of hanging.
"""
iterator = stream.__aiter__()
while True:
remaining = deadline - time.perf_counter()
if remaining <= 0:
raise StreamStalled(
"stream exceeded wall-clock budget "
"(no completion within the request timeout)"
)
try:
chunk = await asyncio.wait_for(
iterator.__anext__(),
timeout=min(idle_timeout, remaining),
)
except StopAsyncIteration:
return
except (asyncio.TimeoutError, TimeoutError) as exc:
raise StreamStalled(
f"stream stalled: no chunk received within "
f"{idle_timeout:.1f}s idle window"
) from exc
yield chunk
async def _run_openai_round(cfg: EndpointConfig, document: str,
question: str) -> tuple[int, int, int, int,
float, float, dict[str, Any]]:
"""One streaming round via the OpenAI-compatible transport."""
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url=cfg.base_url,
api_key=cfg.api_key,
timeout=cfg.request_timeout,
max_retries=0, # benchmark measures a single attempt per round
)
messages = [
{"role": "system", "content": document},
{"role": "user", "content": question},
]
start = time.perf_counter()
deadline = start + cfg.request_timeout
first_token_time: float | None = None
output_tokens = 0
usage_obj: Any = None
try:
stream = await asyncio.wait_for(
client.chat.completions.create(
model=cfg.model,
messages=messages, # type: ignore[arg-type]
max_tokens=cfg.max_tokens,
temperature=1.0,
stream=True,
stream_options={"include_usage": True},
),
timeout=max(0.0, deadline - time.perf_counter()),
)
async for chunk in _iter_with_idle_guard(
stream,
idle_timeout=cfg.stream_idle_timeout,
deadline=deadline,
):
if first_token_time is None and getattr(chunk, "choices", None):
delta = chunk.choices[0].delta if chunk.choices else None
if delta is not None and getattr(delta, "content", None):
first_token_time = time.perf_counter()
if getattr(chunk, "usage", None) is not None:
usage_obj = chunk.usage
finally:
with contextlib.suppress(Exception):
await asyncio.wait_for(client.close(), timeout=5.0)
latency = time.perf_counter() - start
ttft = (first_token_time - start) if first_token_time else latency
usage = usage_to_dict(usage_obj)
prompt_tokens, cached, cache_creation = cfg.extractor(usage)
output_tokens = _as_int(_get(usage, "completion_tokens"))
if output_tokens == 0:
output_tokens = _as_int(_get(usage, "output_tokens"))
return prompt_tokens, cached, cache_creation, output_tokens, ttft, latency, usage
async def _run_responses_round(
cfg: EndpointConfig,
document: str,
question: str,
previous_response_id: str | None,
) -> tuple[int, int, int, int, float, float, dict[str, Any], str | None]:
"""One streaming ``v1/responses`` round with a service-local response chain."""
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url=cfg.base_url,
api_key=cfg.api_key,
timeout=cfg.request_timeout,
max_retries=0,
)
start = time.perf_counter()
deadline = start + cfg.request_timeout
first_token_time: float | None = None
final_response: Any = None
request_input = [
{"role": "system", "content": document},
{"role": "user", "content": question},
]
try:
stream = await asyncio.wait_for(
client.responses.create(
model=cfg.model,
input=request_input, # type: ignore[arg-type]
previous_response_id=previous_response_id,
max_output_tokens=cfg.max_tokens,
temperature=1.0,
stream=True,
store=True,
),
timeout=max(0.0, deadline - time.perf_counter()),
)
async for event in _iter_with_idle_guard(
stream,
idle_timeout=cfg.stream_idle_timeout,
deadline=deadline,
):
event_type = _get(event, "type", "")
if first_token_time is None and event_type == "response.output_text.delta":
if _get(event, "delta"):
first_token_time = time.perf_counter()
if event_type in {"response.completed", "response.incomplete"}:
final_response = _get(event, "response")
finally:
with contextlib.suppress(Exception):
await asyncio.wait_for(client.close(), timeout=5.0)
latency = time.perf_counter() - start
ttft = (first_token_time - start) if first_token_time else latency
usage = usage_to_dict(_get(final_response, "usage"))
prompt_tokens, cached, cache_creation = cfg.extractor(usage)
output_tokens = _as_int(_get(usage, "output_tokens"))
response_id = _get(final_response, "id")
if not response_id:
raise ValueError("v1/responses stream completed without a response id")
return (
prompt_tokens, cached, cache_creation, output_tokens, ttft, latency,
usage, str(response_id),
)
async def _run_anthropic_round(cfg: EndpointConfig, document: str,
question: str) -> tuple[int, int, int, int,
float, float,
dict[str, Any]]:
"""One streaming round via the Anthropic transport.
Anthropic requires an explicit ``cache_control: {"type": "ephemeral"}``
breakpoint on the large stable block; usage reports cache tokens at the
top level (handled by ``extract_anthropic_style``).
"""
from anthropic import AsyncAnthropic
client = AsyncAnthropic(
base_url=cfg.base_url,
api_key=cfg.api_key,
timeout=cfg.request_timeout,
max_retries=0, # benchmark measures a single attempt per round
)
system_blocks = [
{
"type": "text",
"text": document,
"cache_control": {"type": "ephemeral"},
}
]
messages = [{"role": "user", "content": question}]
start = time.perf_counter()
deadline = start + cfg.request_timeout
first_token_time: float | None = None
usage_obj: Any = None
try:
# Do not send ``temperature`` here: newer Claude models (including
# Claude Sonnet 5) reject the formerly-supported parameter as deprecated.
# Prompt-cache measurements do not depend on sampling configuration.
async with client.messages.stream(
model=cfg.model,
max_tokens=cfg.max_tokens,
system=system_blocks, # type: ignore[arg-type]
messages=messages, # type: ignore[arg-type]
) as stream:
async for event in _iter_with_idle_guard(
stream,
idle_timeout=cfg.stream_idle_timeout,
deadline=deadline,
):
etype = getattr(event, "type", "")
if first_token_time is None and etype == "content_block_delta":
first_token_time = time.perf_counter()
final_message = await asyncio.wait_for(
stream.get_final_message(),
timeout=max(0.0, deadline - time.perf_counter()),
)
usage_obj = getattr(final_message, "usage", None)
finally:
with contextlib.suppress(Exception):
await asyncio.wait_for(client.close(), timeout=5.0)
latency = time.perf_counter() - start
ttft = (first_token_time - start) if first_token_time else latency
usage = usage_to_dict(usage_obj)
prompt_tokens, cached, cache_creation = cfg.extractor(usage)
output_tokens = _as_int(_get(usage, "output_tokens"))
return prompt_tokens, cached, cache_creation, output_tokens, ttft, latency, usage
async def run_round(
cfg: EndpointConfig,
round_id: int,
document: str,
api: str = "chat.completions",
previous_response_id: str | None = None,
) -> tuple[RoundMetrics, str | None]:
"""Execute one API round and derive metrics only from reported usage."""
question = build_user_question(round_id)
next_response_id: str | None = None
try:
if api == "responses":
(prompt_tokens, cached, cache_creation, output_tokens, ttft, latency,
usage, next_response_id) = await _run_responses_round(
cfg, document, question, previous_response_id
)
elif cfg.transport == "anthropic":
(prompt_tokens, cached, cache_creation, output_tokens,
ttft, latency, usage) = await _run_anthropic_round(
cfg, document, question
)
else:
(prompt_tokens, cached, cache_creation, output_tokens,
ttft, latency, usage) = await _run_openai_round(
cfg, document, question
)
except Exception as exc: # noqa: BLE001 - benchmark records failures
return RoundMetrics(
round_id=round_id,
endpoint=cfg.name,
api=api,
prompt_tokens=0,
cached_tokens=0,
cache_creation_tokens=0,
cache_hit_ratio=0.0,
latency=0.0,
ttft=0.0,
output_tokens=0,
usage={},
error=f"{type(exc).__name__}: {exc}",
), previous_response_id
hit_ratio = (cached / prompt_tokens) if prompt_tokens > 0 else 0.0
return RoundMetrics(
round_id=round_id,
endpoint=cfg.name,
api=api,
prompt_tokens=prompt_tokens,
cached_tokens=cached,
cache_creation_tokens=cache_creation,
cache_hit_ratio=hit_ratio,
latency=latency,
ttft=ttft,
output_tokens=output_tokens,
usage=usage,
), next_response_id
async def run_interleaved(avalai_cfg: EndpointConfig,
official_cfg: EndpointConfig,
rounds: int, document: str, concurrency: int,
round_delay: float = 15.0,
api: str = "chat.completions",
) -> tuple[list[RoundMetrics], list[RoundMetrics]]:
"""Run one AvalAI call, then one official call, and repeat.
Each numbered round is a directly comparable request pair using the same
realistic legal-document sample:
1. call **avalai**
2. call the **official** provider
3. repeat steps 1-2 for the next round
Round 1 primes both caches. ``round_delay`` is applied once after that pair
before round 2 reads them. Calls are deliberately sequential; accepting
``concurrency`` here preserves CLI compatibility, but pairing takes
precedence so timing and request order cannot differ between endpoints.
Returns ``(avalai, official)`` metrics.
"""
if concurrency != 1:
print(
" paired endpoint schedule enforces sequential calls; "
f"ignoring concurrency={concurrency}"
)
avalai_metrics: list[RoundMetrics] = []
official_metrics: list[RoundMetrics] = []
avalai_previous_id: str | None = None
official_previous_id: str | None = None
for round_id in range(1, rounds + 1):
phase = "priming" if round_id == 1 else "reading"
print(f" round {round_id}: [avalai/{api}] {phase}...")
avalai_metric, avalai_previous_id = await run_round(
avalai_cfg, round_id, document, api, avalai_previous_id
)
avalai_metrics.append(avalai_metric)
if avalai_metric.error is None:
print(
" usage reports: "
f"prompt_tokens={avalai_metric.prompt_tokens}, "
f"cached_tokens={avalai_metric.cached_tokens}"
)
print(f" round {round_id}: [{official_cfg.name}/{api}] {phase}...")
official_metric, official_previous_id = await run_round(
official_cfg, round_id, document, api, official_previous_id
)
official_metrics.append(official_metric)
if official_metric.error is None:
print(
" usage reports: "
f"prompt_tokens={official_metric.prompt_tokens}, "
f"cached_tokens={official_metric.cached_tokens}"
)
if round_id == 1 and rounds >= 2 and round_delay > 0:
print(
f" waiting {round_delay:.1f}s after the paired priming "
"calls for cache propagation..."
)
await asyncio.sleep(round_delay)
return avalai_metrics, official_metrics
# ---------------------------------------------------------------------------
# Statistics
# ---------------------------------------------------------------------------
def _percentile(values: Sequence[float], pct: float) -> float:
if not values:
return 0.0
ordered = sorted(values)
if len(ordered) == 1:
return ordered[0]
k = (len(ordered) - 1) * pct
lo = int(k)
hi = min(lo + 1, len(ordered) - 1)
frac = k - lo
return ordered[lo] + (ordered[hi] - ordered[lo]) * frac
def summarize(endpoint: str, api: str, rounds: int,
metrics: list[RoundMetrics]) -> EndpointSummary:
successful = [m for m in metrics if m.error is None]
ratios = [m.cache_hit_ratio for m in successful]
warm = [m.cache_hit_ratio for m in successful if m.round_id >= 2]
first_hit_round: int | None = None
first_hit_latency: float | None = None
for m in sorted(successful, key=lambda r: r.round_id):
if m.cached_tokens > 0:
first_hit_round = m.round_id
first_hit_latency = m.latency
break
return EndpointSummary(
endpoint=endpoint,
api=api,
rounds=rounds,
successful=len(successful),
mean_hit_ratio=statistics.mean(ratios) if ratios else 0.0,
median_hit_ratio=statistics.median(ratios) if ratios else 0.0,
p90_hit_ratio=_percentile(ratios, 0.90),
first_hit_round=first_hit_round,
first_hit_latency=first_hit_latency,
mean_latency=(
statistics.mean(m.latency for m in successful)
if successful else 0.0
),
mean_ttft=(
statistics.mean(m.ttft for m in successful)
if successful else 0.0
),
warm_mean_hit_ratio=statistics.mean(warm) if warm else 0.0,
per_round=metrics,
)
# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
def print_comparison(model: str, avalai: EndpointSummary,
official: EndpointSummary) -> None:
print("\n================ Cache-Hit Ratio Comparison ================")
print(f"Model: {model}")
print(f"API: v1/{avalai.api}")
print(f"avalai vs {official.endpoint}\n")
header = (
f"{'Round':>5} | "
f"{'avalai prompt/cached (ratio)':>34} | "
f"{official.endpoint + ' prompt/cached (ratio)':>34} | "
f"{'Δ ratio':>8}"
)
print(header)
print("-" * len(header))
by_round_a = {m.round_id: m for m in avalai.per_round}
by_round_o = {m.round_id: m for m in official.per_round}
all_rounds = sorted(set(by_round_a) | set(by_round_o))
for rid in all_rounds:
a = by_round_a.get(rid)
o = by_round_o.get(rid)
def _cell(m: RoundMetrics | None) -> str:
if m is None:
return "n/a"
if m.error is not None:
return f"ERROR ({m.error[:22]})"
return (
f"{m.prompt_tokens}/{m.cached_tokens} "
f"({m.cache_hit_ratio * 100:5.1f}%)"
)
delta = ""
if a and o and a.error is None and o.error is None:
delta = f"{(a.cache_hit_ratio - o.cache_hit_ratio) * 100:+6.1f}%"
print(
f"{rid:>5} | {_cell(a):>34} | {_cell(o):>34} | {delta:>8}"
)
failures = [
metric
for summary in (avalai, official)
for metric in summary.per_round
if metric.error is not None
]
if failures:
print("\n---------------- Full errors ----------------")
for metric in failures:
print(
f"[{metric.endpoint}/v1/{metric.api} round {metric.round_id}] "
f"{metric.error}"
)
print("\n---------------- Summary ----------------")
for s in (avalai, official):
fh = (
f"round {s.first_hit_round} ({s.first_hit_latency:.3f}s)"
if s.first_hit_round is not None else "never"
)
print(
f"[{s.endpoint}/v1/{s.api}] successful={s.successful}/{s.rounds} | "
f"mean={s.mean_hit_ratio * 100:.1f}% | "
f"median={s.median_hit_ratio * 100:.1f}% | "
f"p90={s.p90_hit_ratio * 100:.1f}% | "
f"warm(≥2)={s.warm_mean_hit_ratio * 100:.1f}% | "
f"first-hit={fh} | "
f"mean-latency={s.mean_latency:.3f}s | "
f"mean-ttft={s.mean_ttft:.3f}s"
)
warm_delta = (
avalai.warm_mean_hit_ratio - official.warm_mean_hit_ratio
) * 100
print(
f"\nWarm-cache hit-ratio delta (avalai − {official.endpoint}): "
f"{warm_delta:+.1f} percentage points"
)
def print_endpoint_comparison(summaries: Sequence[EndpointSummary]) -> None:
"""Compare warm-cache performance across both APIs for each service."""
print("\n================ Endpoint Comparison =======================")
print(
f"{'Service':>18} | {'API':>18} | {'Warm hit':>9} | "
f"{'Mean latency':>12} | {'Mean TTFT':>9} | {'Success':>9}"
)
print("-" * 91)
for summary in summaries:
print(
f"{summary.endpoint:>18} | {'v1/' + summary.api:>18} | "
f"{summary.warm_mean_hit_ratio * 100:>8.1f}% | "
f"{summary.mean_latency:>11.3f}s | {summary.mean_ttft:>8.3f}s | "
f"{summary.successful:>4}/{summary.rounds:<4}"
)
def persist_json(model: str, config: dict[str, Any],
summaries: Mapping[str, EndpointSummary]) -> Path:
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
safe_model = model.replace("/", "_").replace(":", "_")
path = RESULTS_DIR / f"cache_hit_{safe_model}_{stamp}.json"
def _dump(s: EndpointSummary) -> dict[str, Any]:
d = asdict(s)
d["per_round"] = [asdict(m) for m in s.per_round]
return d
payload = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"model": model,
"config": config,
"results": {key: _dump(summary) for key, summary in summaries.items()},
}
path.write_text(json.dumps(payload, indent=2))
print(f"\nSaved JSON: {path}")
return path
def render_charts(model: str, avalai: EndpointSummary,
official: EndpointSummary) -> list[Path]:
"""Grouped-bar (per-round hit ratio) + cumulative-hit-ratio line charts.
matplotlib is imported lazily and guarded — the benchmark still produces
the table + JSON when it is missing.
"""
try:
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
except Exception as exc: # noqa: BLE001
print(
f"\n[charts skipped] matplotlib unavailable ({exc}). "
f"Install with: pip install matplotlib"
)
return []
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
safe_model = model.replace("/", "_").replace(":", "_")
safe_api = avalai.api.replace(".", "_")
saved: list[Path] = []
a_rounds = {m.round_id: m for m in avalai.per_round}
o_rounds = {m.round_id: m for m in official.per_round}
rounds = sorted(set(a_rounds) | set(o_rounds))
def _ratio_series(source: dict[int, RoundMetrics]) -> list[float]:
series: list[float] = []
for r in rounds:
m = source.get(r)
series.append(m.cache_hit_ratio * 100 if m is not None else 0.0)
return series
a_ratio = _ratio_series(a_rounds)
o_ratio = _ratio_series(o_rounds)
# 1) Grouped bar: per-round hit ratio.
fig, ax = plt.subplots(figsize=(10, 5))
width = 0.4
x = list(range(len(rounds)))
ax.bar([i - width / 2 for i in x], a_ratio, width, label="avalai")
ax.bar([i + width / 2 for i in x], o_ratio, width,
label=official.endpoint)
ax.set_xticks(x)
ax.set_xticklabels([str(r) for r in rounds])
ax.set_xlabel("Round")
ax.set_ylabel("Cache-hit ratio (%)")
ax.set_ylim(0, 100)
ax.set_title(f"Per-round cache-hit ratio — {model} — v1/{avalai.api}")
ax.legend()
ax.grid(axis="y", alpha=0.3)
bar_path = RESULTS_DIR / f"cache_hit_bar_{safe_model}_{safe_api}_{stamp}.png"
fig.tight_layout()
fig.savefig(bar_path, dpi=120)
plt.close(fig)
saved.append(bar_path)
# 2) Line: cumulative (running-mean) hit ratio.
def _cumulative(ratios: list[float]) -> list[float]:
out: list[float] = []
acc = 0.0
for i, v in enumerate(ratios, start=1):
acc += v
out.append(acc / i)
return out
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(rounds, _cumulative(a_ratio), marker="o", label="avalai")
ax.plot(rounds, _cumulative(o_ratio), marker="s",
label=official.endpoint)
ax.set_xlabel("Round")
ax.set_ylabel("Cumulative mean hit ratio (%)")
ax.set_ylim(0, 100)
ax.set_title(f"Cumulative cache-hit ratio — {model} — v1/{avalai.api}")
ax.legend()
ax.grid(alpha=0.3)
line_path = RESULTS_DIR / f"cache_hit_line_{safe_model}_{safe_api}_{stamp}.png"
fig.tight_layout()
fig.savefig(line_path, dpi=120)
plt.close(fig)
saved.append(line_path)
for p in saved:
print(f"Saved chart: {p}")
return saved
# ---------------------------------------------------------------------------
# CLI + orchestration
# ---------------------------------------------------------------------------
def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Compare prompt-cache hit ratios between AvalAI and a model's "
"official provider over v1/chat/completions and v1/responses."
)
)
parser.add_argument(
"--model", default="deepseek-v4-flash",
help="Model id shared by both endpoints (default: deepseek-v4-flash).",
)
parser.add_argument(
"--official-provider", default=None,
help=(
"Override provider inference. One of: openai, claude.ai, "
"deepseek, z.ai, dashscope, moonshot.ai, gemini, xai."
),
)
parser.add_argument(
"--avalai-api-key", default=None,
help=f"AvalAI API key (fallback env: {AVALAI_ENV_VAR}).",
)
parser.add_argument(
"--avalai-base-url", default=AVALAI_DEFAULT_BASE_URL,
help=f"AvalAI base URL (default: {AVALAI_DEFAULT_BASE_URL}).",
)
# One --<provider>-api-key flag per official provider.
for spec in PROVIDER_REGISTRY.values():
parser.add_argument(
spec.cli_flag, dest=spec.dest, default=None,
help=f"{spec.label} API key (fallback env: {spec.env_var}).",
)
parser.add_argument("--rounds", type=int, default=5,
help="Total rounds per endpoint (default: 5).")
parser.add_argument(
"--prefix-tokens", type=int, default=4000,
help=(
"Approximate legal-document sample size in tokens "
"(default: 4000; clamped to the provider minimum)."
),
)
parser.add_argument(
"--concurrency", type=int, default=1,
help=(
"Retained for CLI compatibility. Endpoint calls are always paired "
"sequentially (AvalAI then official), so values above 1 are ignored."
),
)
parser.add_argument(
"--max-tokens", type=int, default=64,
help="max_tokens for each completion (default: 64).",
)
parser.add_argument(
"--request-timeout", type=float, default=120.0,
help=(
"Wall-clock ceiling (seconds) for a single round: connect + the "
"entire stream. Guards against a proxy/upstream that keeps the "
"socket open but never completes (default: 120.0)."
),
)
parser.add_argument(
"--stream-idle-timeout", type=float, default=60.0,
help=(
"Max seconds allowed between two consecutive stream chunks before "
"the round is aborted as a stall. Trips before --request-timeout "
"when a stream freezes mid-flight, e.g. the final usage/priced "
"chunk never arrives (default: 60.0)."
),
)
parser.add_argument(
"--round-delay", type=float, default=15.0,
help=(
"Seconds to wait once, right after the cache-priming round of each "
"provider, giving the AvalAI service time to propagate the cache "
"session before the read rounds (default: 15.0; set 0 to disable)."
),
)
parser.add_argument("--no-charts", action="store_true",
help="Skip matplotlib chart generation.")
return parser
def _resolve_key(cli_value: str | None, env_var: str) -> str | None:
if cli_value:
return cli_value
return os.getenv(env_var) or None
def resolve_keys(args: argparse.Namespace,
spec: ProviderSpec) -> tuple[str, str]:
"""Resolve AvalAI + official keys, failing fast with a clear error.
Validation happens BEFORE any AvalAI request is issued.
"""
official_cli = getattr(args, spec.dest, None)
official_key = _resolve_key(official_cli, spec.env_var)
if not official_key:
raise SystemExit(
f"Missing official-provider API key for '{spec.label}'. "
f"Pass {spec.cli_flag} or set ${spec.env_var}."
)
avalai_key = _resolve_key(args.avalai_api_key, AVALAI_ENV_VAR)
if not avalai_key:
raise SystemExit(
f"Missing AvalAI API key. Pass --avalai-api-key or set "
f"${AVALAI_ENV_VAR}."
)
return avalai_key, official_key
async def run_benchmark(args: argparse.Namespace) -> None:
spec = resolve_provider(args.model, args.official_provider)
# Fail fast on missing keys before touching the network.
avalai_key, official_key = resolve_keys(args, spec)
min_tokens = spec.min_cache_tokens
document = build_sample_document(args.prefix_tokens, min_tokens)
experiments = ("chat.completions", "responses")
print(
f"Model={args.model} | official={spec.label} | "
f"rounds={args.rounds} | sample≈{args.prefix_tokens} tokens "
f"(min {min_tokens}) | concurrency={args.concurrency}"
)
print(
" cache-hit source of truth: response usage dict "
"(prompt_tokens_details.cached_tokens / prompt_tokens)"
)
avalai_cfg = EndpointConfig(
name="avalai",
transport="openai", # AvalAI is OpenAI-compatible for all models.
base_url=args.avalai_base_url,
api_key=avalai_key,
model=args.model,
extractor=spec.extractor,
max_tokens=args.max_tokens,
request_timeout=args.request_timeout,
stream_idle_timeout=args.stream_idle_timeout,
)
official_cfg = EndpointConfig(
name=spec.label,
transport=spec.transport,
base_url=spec.base_url,
api_key=official_key,
model=args.model,
extractor=spec.extractor,
max_tokens=args.max_tokens,
request_timeout=args.request_timeout,
stream_idle_timeout=args.stream_idle_timeout,
)
summaries: dict[str, EndpointSummary] = {}
for api in experiments:
print(f"\nRunning v1/{api} experiment...")
avalai_metrics, official_metrics = await run_interleaved(
avalai_cfg,
official_cfg,
args.rounds,
document,
args.concurrency,
round_delay=args.round_delay,
api=api,
)
avalai_summary = summarize("avalai", api, args.rounds, avalai_metrics)
official_summary = summarize(
spec.label, api, args.rounds, official_metrics
)
summaries[f"avalai/{api}"] = avalai_summary
summaries[f"official/{api}"] = official_summary
print_comparison(args.model, avalai_summary, official_summary)
print_endpoint_comparison(list(summaries.values()))
config = {
"model": args.model,
"official_provider": spec.key,
"official_label": spec.label,
"rounds": args.rounds,
"prefix_tokens": args.prefix_tokens,
"min_cache_tokens": min_tokens,
"concurrency": args.concurrency,
"max_tokens": args.max_tokens,
"round_delay": args.round_delay,
"request_timeout": args.request_timeout,
"stream_idle_timeout": args.stream_idle_timeout,
"cache_hit_source": "response_usage",
"cache_hit_formula": (
"usage.prompt_tokens_details.cached_tokens / usage.prompt_tokens"
),
"sample": "repeated_legal_document",
"responses_previous_response_id": True,
"avalai_base_url": args.avalai_base_url,
"official_base_url": spec.base_url,
}
persist_json(args.model, config, summaries)
if not args.no_charts:
for api in experiments:
render_charts(
args.model,
summaries[f"avalai/{api}"],
summaries[f"official/{api}"],
)
async def main(argv: list[str] | None = None) -> None:
parser = build_arg_parser()
args = parser.parse_args(argv)
await run_benchmark(args)
if __name__ == "__main__":
asyncio.run(main())
# In a Jupyter notebook, replace the block above with:
# await main([...]) # pass CLI-style args as a listایجاد صریح کش، breakpoint کش و markerهای مخصوص provider همچنان پشتیبانی نمیشوند. استفاده از کش ضمنی را با فیلدهای usage پاسخ بررسی کنید. اگر پس از آزمایش کنترلشده، hit rate در AvalAI بهطور معناداری کمتر از استفاده مستقیم provider بود، مدل، زمان درخواستها و usage پاسخ را در تیکت پشتیبانی ثبت نمایید.
درخواست API و پاسخ نماینده
درخواست
bash
curl https://api.avalai.ir/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $AVALAI_API_KEY" \
-d '{
"model": "muse-glimmer-30b",
"messages": [
{
"role": "user",
"content": "Explain why stable prompt prefixes improve cache reuse in two sentences."
}
]
}'پاسخ نماینده
json
{
"id": "chatcmpl-fireworks-example",
"created": 1786611600,
"model": "muse-glimmer-30b",
"object": "chat.completion",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {
"role": "assistant",
"content": "Stable prompt prefixes let the provider recognize and reuse previously processed input. Keeping dynamic content at the end preserves the shared prefix across requests."
}
}
],
"usage": {
"completion_tokens": 31,
"prompt_tokens": 24,
"total_tokens": 55,
"prompt_tokens_details": {
"cached_tokens": 0
}
}
}این پاسخ نماینده است؛ شناسهها، متن، تعداد توکن، cache hit و فیلدهای هزینه با درخواست و route تغییر میکنند.
نمونههای استفاده SDK
bash
curl https://api.avalai.ir/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $AVALAI_API_KEY" \
-d '{
"model": "nemotron-3.5-lightning",
"messages": [{"role": "user", "content": "Draft a concise API migration checklist."}]
}'python
from openai import OpenAI
client = OpenAI(
api_key="your-avalai-api-key",
base_url="https://api.avalai.ir/v1",
)
completion = client.chat.completions.create(
model="nemotron-3.5-lightning",
messages=[{"role": "user", "content": "Draft a concise API migration checklist."}],
)
print(completion.choices[0].message.content)javascript
import OpenAI from "openai";
const client = new OpenAI({
apiKey: process.env.AVALAI_API_KEY,
baseURL: "https://api.avalai.ir/v1",
});
const completion = await client.chat.completions.create({
model: "nemotron-3.5-lightning",
messages: [
{ role: "user", content: "Draft a concise API migration checklist." },
],
});
console.log(completion.choices[0].message.content);