عملکرد AvalAI
در این صفحه، عملکرد AvalAI را از چند زاویه بررسی میکنیم: میزان استفاده از cache پرامپت، تأخیر در مناطق مختلف، سرعت تولید توکن و شیوه درست اندازهگیری در محیط عملیاتی. تازهترین آزمایشها بالاتر آمدهاند. این اعداد نتیجه چند اجرای مشخصاند و نباید آنها را تضمین همیشگی سرویس در نظر گرفت.
گزارشهای ۱۴۰۵-۰۵-۲۳
نتیجههای cache در هر دو endpoint متنی OpenAI
آخرین نسخه بنچمارک، درخواست واقعگرایانه تحلیل سند حقوقی را از طریق AvalAI و provider رسمی هر مدل روی v1/chat/completions و v1/responses میفرستد. دور اول cache هر سرویس را آماده میکند، runner بهمدت 15 ثانیه صبر میکند و دورهای 2 تا 5 رفتار cache در حالت warm را میسنجند. یک درخواست فقط زمانی cache hit محسوب میشود که API تعداد توکنهای cacheشده را گزارش کند—عمدتا با نسبت usage.prompt_tokens_details.cached_tokens / usage.prompt_tokens—نه بر اساس ترتیب درخواست، زمان پاسخ یا فرض ثابت بودن prefix.
این نتیجهها مشاهدات یک اجرای مشخص با پنج دور paired و ترتیبی برای هر API و سه مدل deepseek-v4-flash، glm-5.2 و kimi-k3 هستند. دادهها رفتار مشاهدهشده در همین اجرا را نشان میدهند و تضمین سطح سرویس نیستند.
deepseek-v4-flash: مقایسه AvalAI و DeepSeek
هر دو سرویس در هر دو API تمام 5 درخواست را با موفقیت پاسخ دادند. AvalAI در همه دورها 99.9% cache hit گزارش کرد (1,400 توکن از 1,401 توکن)، درحالیکه DeepSeek به 95.1% رسید (1,408 توکن از 1,480 توکن). بنابراین اختلاف warm در هر دو API به سود AvalAI +4.8 واحد درصد بود.
| سرویس | API | cache hit در حالت warm | میانگین latency | موفقیت |
|---|---|---|---|---|
| AvalAI | v1/chat/completions | 99.9% | 1.704s | 5/5 |
| DeepSeek | v1/chat/completions | 95.1% | 6.201s | 5/5 |
| AvalAI | v1/responses | 99.9% | 18.282s | 5/5 |
| DeepSeek | v1/responses | 95.1% | 6.444s | 5/5 |
glm-5.2: مقایسه AvalAI و Z.AI
در v1/chat/completions هر دو سرویس تمام 5 درخواست را پاسخ دادند و به 99.9% cache hit در حالت warm رسیدند. AvalAI در دورهای 2 تا 5 تعداد 1,409 توکن از 1,410 توکن و Z.AI تعداد 1,408 توکن از 1,410 توکن را cacheشده گزارش کرد. در v1/responses، AvalAI هر 5 درخواست را با موفقیت پاسخ داد و پس از دور آمادهسازی به 99.9% رسید. Z.AI برای هر پنج درخواست خطای 404 Not Found برگرداند، زیرا base URL عمومی آزمایششده این endpoint را ارائه نمیکند؛ این خطاها نتیجه دسترسپذیری endpoint هستند، نه درخواستهای موفق با cache hit صفر درصد.
| سرویس | API | cache hit در حالت warm | میانگین latency | موفقیت |
|---|---|---|---|---|
| AvalAI | v1/chat/completions | 99.9% | 20.394s | 5/5 |
| Z.AI | v1/chat/completions | 99.9% | 4.971s | 5/5 |
| AvalAI | v1/responses | 99.9% | 2.830s | 5/5 |
| Z.AI | v1/responses | در دسترس نیست | — | 0/5 |
kimi-k3: مقایسه AvalAI و Moonshot AI
هر دو سرویس تمام 5 درخواست Chat Completions را پاسخ دادند و در تمام دورها 85.8% cache hit گزارش کردند (1,280 توکن از 1,492 توکن). AvalAI در Responses نیز هر 5 درخواست را با 85.8% cache hit پاسخ داد. API عمومی آزمایششده Moonshot درخواستهای v1/responses را نپذیرفت: چهار درخواست با خطای دسترسی و یک درخواست با خطای اتصال روبهرو شد. AvalAI همچنان درخواست Responses این مدل را میپذیرد و آن را در لایه مسیریابی داخلی به جریان سازگار بالادستی تبدیل میکند.
| سرویس | API | cache hit در حالت warm | میانگین latency | موفقیت |
|---|---|---|---|---|
| AvalAI | v1/chat/completions | 85.8% | 4.187s | 5/5 |
| Moonshot AI | v1/chat/completions | 85.8% | 7.819s | 5/5 |
| AvalAI | v1/responses | 85.8% | 4.401s | 5/5 |
| Moonshot AI | v1/responses | در دسترس نیست | — | 0/5 |
روش اجرا و بازتولید
runner درخواستها را بهصورت ترتیبی—ابتدا AvalAI و سپس provider رسمی—ارسال میکند و برای دورهای Responses یک زنجیره مستقل previous_response_id برای هر سرویس میسازد. artifact مربوط به JSON، شیء کامل usage گزارششده توسط provider را برای بررسی مستقل نگه میدارد؛ اما این مستندات عمدا هیچ مسیر محلی یا credential را منتشر نمیکنند.
bash
python tests/benchmarks/test_cache_hit_ratio.py \
--model deepseek-v4-flash --rounds 5 --prefix-tokens 2000 \
--official-provider deepseek
python tests/benchmarks/test_cache_hit_ratio.py \
--model glm-5.2 --rounds 5 --prefix-tokens 2000 \
--official-provider zai
python tests/benchmarks/test_cache_hit_ratio.py \
--model kimi-k3 --rounds 5 --prefix-tokens 2000 \
--official-provider moonshot.aiسورس کامل بنچمارک
نسخه کامل و اصلاحشده اسکریپت از هر دو API، زنجیره مستقل Responses برای هر سرویس، شکلهای مختلف usage در APIهای سازگار با OpenAI و Anthropic، تشخیص خودکار provider و aliasها، محافظ stream در برابر توقف، artifactهای JSON و نمودارهای جداگانه هر endpoint پشتیبانی میکند. کلیدهای API از flagها یا environment variableهای استاندارد خوانده میشوند و هرگز داخل artifact نوشته نمیشوند.
نمایش سورس کامل بنچمارک
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دامنه و محدودیتها
این سه آزمایش پنجدورهای در تاریخ ۱۴۰۵-۰۵-۲۳ مشاهده شدهاند. خطای Responses در provider رسمی یعنی endpoint موردنظر در API عمومی آزمایششده در دسترس نبوده است؛ این خطا نباید بهعنوان cache hit صفر درصد برای یک درخواست موفق تفسیر شود. latency تحتتأثیر شرایط مدل و شبکه است و یک درخواست کند میتواند میانگین پنجدورهای را بهشکل محسوسی تغییر دهد. برنامه باید در صورت cache miss نیز درست کار کند و نباید affinity را جایگزین state مکالمه بداند. برای جزئیات پیادهسازی، راهنمای cache پرامپت را ببینید.
راهنمای بهینهسازی عملکرد
بهتر کردن cache-hit ratio
- instructions ثابت، متن policy، schema ابزارها، تصاویر و مثالهای قابل reuse را ابتدا و context پویا و timestampها را انتها قرار دهید.
- prefix را از نظر byte کاملا یکسان نگه دارید و
usage.prompt_tokens_details.cached_tokensیا فیلد native ارائهدهنده را بررسی کنید. - درخواستهای warm را جدا از priming مقایسه کنید و مدل، route، نسخه prompt، request ID و timestampها را ثبت کنید.
کاهش latency و افزایش throughput
- با کوچکترین مدلی شروع کنید که evalهای شما را پاس میکند و مدلهای reasoning بزرگتر را برای تصمیمهای سخت نگه دارید.
- خروجی را با
max_output_tokensبرای/v1/responsesیاmax_completion_tokensبرای/v1/chat/completionsمحدود کنید. - خروجی کاربرمحور را stream کنید، فقط کارهای مستقل را موازی اجرا کنید و connectionهای HTTP را reuse کنید.
- فراخوانیهای غیرضروری مدل را حذف و مراحل نزدیک را در صورت امکان در یک structured response ترکیب کنید.
اندازهگیری رفتار production
TTFT/TTFB، latency کل، مدل، endpoint، توکنهای input/output/cacheشده، request ID، status و تعداد retry/fallback را log کنید. بهجای اتکا به یک میانگین، p50، p90 و p95 را گزارش کنید. این صفحه را کنار بهینهسازی latency و بهینهسازی هزینه بخوانید.
بنچمارکهای latency منطقهای
مطالعههای latency زیر از دامنه اصلی AvalAI یعنی api.avalai.ir استفاده کردند و برای مقایسه یکسان زیرساخت اصلی، Guardrail غیرفعال بود. Guardrail برای بیشتر اپلیکیشنها توصیه میشود، اما در این تستهای تاریخی حدود 200 تا 300 میلیثانیه latency اضافه میکرد. دامنه جایگزین اتصال یعنی api.avalapis.ir ذاتا latency بیشتری دارد و در این تستها استفاده نشد.
AvalAI connectionهای پایدار و connection poolهای provider را نگه میدارد و بهینهسازی پردازش درخواست، شبکه و فراخوانی تکراری سربار پلتفرم را کم میکند. موقعیت جغرافیایی، بار مدل، طول خروجی، تنظیمات امنیتی و شرایط شبکه همچنان بر هر اندازهگیری اثر دارند.
جدیدترین latency منطقهای - ۲۰-۰۷-۱۴۰۴
عملکرد مرکز داده اروپا (EU) - ۲۰-۰۷-۱۴۰۴
این تست از یک ماشین مجازی در یک مرکز داده Azure در اتحادیه اروپا انجام شد و عملکرد AvalAI را با فراخوانی مستقیم API OpenAI از همان مکان مقایسه میکند.
محیط تست:
- مدل:
gpt-4o-mini - ارائه دهنده ابری: Microsoft Azure
- سختافزار: 4GB RAM، 2 vCPUs
- مکان: اروپا
نتایج عملکرد اروپا
| معیار | AvalAI (gpt-4o-mini) | OpenAI (gpt-4o-mini) |
|---|---|---|
| میانگین TTFB (ثانیه) | 0.435 | 0.717 |
| میانه TTFB (ثانیه) | 0.393 | 0.685 |
| صدک ۹۵ TTFB (ثانیه) | 0.605 | 1.032 |
| میانگین توکن در ثانیه | 24.5 | 13.8 |
| نرخ موفقیت | 100.00% | 100.00% |
تحلیل
نتایج خودگویاست: AvalAI اکنون ۳۹٪ سریعتر از دسترسی مستقیم به OpenAI از مرکز داده اروپایی ما است. میانه TTFB ما ۰.۳۹۳ ثانیه در مقایسه با ۰.۶۸۵ ثانیه OpenAI، به همراه توان عملیاتی توکن ۷۷٪ بالاتر (۲۴.۵ در مقابل ۱۳.۸ توکن/ثانیه)، نشان میدهد که یک پلتفرم API یکپارچه با معماری خوب میتواند از دسترسی مستقیم به ارائهدهنده بهتر عمل کند.
این مزیت عملکردی از اتصالات پایدار ما به زیرساخت OpenAI و مدیریت بهینهشده درخواستهای ما ناشی میشود. درحالیکه کاربران فردی باید اتصالات را برقرار کرده و سربار را با هر درخواست مدیریت کنند، سیستم با ترافیک بالای ما اتصالات گرم و مسیرهای بهینهشده به تمام ارائهدهندگان را حفظ میکند. بهینهسازیهای اخیر زیرساخت ما هر سربار داخلی را بیشتر کاهش داده است که منجر به عملکرد برتری شده که در این بنچمارکها نشان داده شده است.
عملکرد مرکز داده خاورمیانه (ME) - ۲۰-۰۷-۱۴۰۴
این تست از یک ماشین مجازی در یک مرکز داده Arvancloud در خاورمیانه انجام شد و عملکرد AvalAI را با فراخوانی مستقیم API OpenAI از همان مکان مقایسه میکند.
محیط تست:
- مدل:
gpt-4o-mini - ارائه دهنده ابری: Arvancloud
- سختافزار: 4GB RAM، 2 vCPUs
- مکان: خاورمیانه
نتایج عملکرد خاورمیانه
| معیار | AvalAI (gpt-4o-mini) | OpenAI (gpt-4o-mini) |
|---|---|---|
| میانگین TTFB (ثانیه) | 0.703 | 1.246 |
| میانه TTFB (ثانیه) | 0.668 | 1.048 |
| صدک ۹۵ TTFB (ثانیه) | 0.947 | 2.309 |
| میانگین توکن در ثانیه | 14.9 | 8.4 |
| نرخ موفقیت | 100.00% | 100.00% |
تحلیل
مزیت عملکردی حتی برای کاربران خاورمیانه نیز برجستهتر است: AvalAI ۴۴٪ زمان پاسخ سریعتر با میانه TTFB ۰.۶۶۸ ثانیه در مقایسه با ۱.۰۴۸ ثانیه OpenAI ارائه میدهد. توان عملیاتی توکن ۷۷٪ بالاتر است (۱۴.۹ در مقابل ۸.۴ توکن/ثانیه) که تجربه بسیار بهتری برای استقرارهای منطقهای فراهم میکند.
موقعیت استراتژیک زیرساخت ما و مسیرهای مسیریابی بهینهشده، به همراه اتصالات پایدار به ارائهدهندگان و بهینهسازیهای اخیر، عملکرد استثنایی را برای کاربران در این منطقه ارائه میدهد. فاصله عملکردی در مقایسه با دسترسی مستقیم به OpenAI از زمان بنچمارکهای خردادماه بهطور قابلتوجهی افزایش یافته است که تاثیر تلاشهای بهینهسازی ما را نشان میدهد.
عملکرد تاریخی - خرداد ۱۴۰۴
برای شفافیت و نمایش بهبود مستمر ما، نتایج بنچمارک قبلی خود را از خرداد ۱۴۰۴ حفظ میکنیم. این نتایج تاریخی نقطه شروع قبل از ابتکار بهینهسازی اخیر ما را نشان میدهند.
عملکرد مرکز داده اروپا (EU) - تاریخ ۲۲-۰۳-۱۴۰۴
محیط تست:
- مدل:
gpt-4o-mini - ارائه دهنده ابری: Microsoft Azure
- مکان: اروپا
نتایج عملکرد اروپا (تاریخی)
| معیار | AvalAI (gpt-4o-mini) | OpenAI (gpt-4o-mini) |
|---|---|---|
| میانگین TTFB (ثانیه) | 0.728 | 0.531 |
| میانه TTFB (ثانیه) | 0.683 | 0.510 |
| صدک ۹۵ TTFB (ثانیه) | 1.056 | 0.740 |
| میانگین توکن در ثانیه | 15.9 | 18.9 |
| نرخ موفقیت | 100.00% | 100.00% |
تحلیل تاریخی
در خرداد ۱۴۰۴، AvalAI سربار تاخیر جزئی در حدود ۲۰۰ میلیثانیه نسبت به فراخوانی مستقیم OpenAI از مراکز داده Azure نشان داد. این امر با توجه به میزبانی اصلی OpenAI بر روی زیرساخت Azure قابل انتظار بود. خدمات ارزشافزودهای که ما ارائه میدادیم—مسیریابی API یکپارچه، لایههای امنیتی قوی و پشتیبانی از چند ارائهدهنده—با این سربار جزئی همراه بود.
عملکرد مرکز داده خاورمیانه (ME) - تاریخ ۲۲-۰۳-۱۴۰۴
محیط تست:
- مدل:
gpt-4o-mini - ارائه دهنده ابری: Arvancloud
- مکان: خاورمیانه
نتایج عملکرد خاورمیانه (تاریخی)
| معیار | AvalAI (gpt-4o-mini) | OpenAI (gpt-4o-mini) |
|---|---|---|
| میانگین TTFB (ثانیه) | 0.993 | 1.095 |
| میانه TTFB (ثانیه) | 0.929 | 0.951 |
| صدک ۹۵ TTFB (ثانیه) | 1.479 | 1.386 |
| میانگین توکن در ثانیه | 11.4 | 9.6 |
| نرخ موفقیت | 100.00% | 100.00% |
تحلیل تاریخی
حتی در خرداد ۱۴۰۴، AvalAI عملکرد رقابتی را برای کاربران خاورمیانه با تاخیر کمتر و توان عملیاتی بالاتر نسبت به دسترسی مستقیم به OpenAI ارائه میداد. نتایج مهرماه ما بهبودهای قابلتوجه بیشتری را از تلاشهای بهینهسازی ما نشان میدهد.
بازتولید نتایج latency منطقهای
ما به شفافیت کامل اعتقاد داریم. شما میتوانید از اسکریپت پایتون زیر برای اجرای این تستهای عملکردی خودتان استفاده کنید و نتایج ما را تایید کنید.
توجه: درحالیکه این بنچمارکها با استفاده از مدل gpt-4o-mini انجام شدهاند، بهبودهای عملکردی برای تمام مدلهای موجود در پلتفرم ما اعمال میشود. شما میتوانید هر مدلی از انتخاب خود را با استفاده از این اسکریپت تست کنید تا ببینید AvalAI برای مورد استفاده خاص شما چگونه عمل میکند.
لطفا اطمینان حاصل کنید که کتابخانههای لازم (requests، numpy، matplotlib، seaborn، tabulate، tqdm) را نصب کردهاید.
python
import os
import requests
import time
import numpy as np
import matplotlib.pyplot as plt
import json
from tabulate import tabulate
from tqdm import tqdm
from datetime import datetime
import seaborn as sns
def test_api_performance(
api_name, api_url, api_key, model, num_requests=10, prompt="Say hi"
):
"""Test API performance and collect comprehensive metrics"""
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
data = {"model": model, "messages": [{"role": "user", "content": prompt}]}
ttfb_times = []
total_times = []
token_counts = []
tokens_per_second = []
errors = 0
print(f"Testing {api_name} API with {num_requests} requests...")
for _ in tqdm(range(num_requests)):
try:
start_time = time.time()
response = requests.post(
api_url, headers=headers, json=data, timeout=(10, 30)
)
response_time = time.time()
# Process the response
response_json = response.json()
end_time = time.time()
# Calculate metrics
ttfb = response_time - start_time
total_time = end_time - start_time
# Try to get token count if available
try:
usage = response_json.get("usage", {})
total_tokens = usage.get("total_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
token_counts.append(total_tokens)
# Calculate tokens per second (using completion tokens)
if completion_tokens > 0 and total_time > 0:
tokens_per_second.append(completion_tokens / total_time)
else:
tokens_per_second.append(0)
except Exception as e:
token_counts.append(0)
tokens_per_second.append(0)
ttfb_times.append(ttfb)
total_times.append(total_time)
# Add a small delay to avoid rate limiting
time.sleep(0.5)
except Exception as e:
print(f"Error on request: {e}")
errors += 1
return {
"name": api_name,
"url": api_url,
"model": model,
"average_ttfb": np.mean(ttfb_times) if ttfb_times else None,
"median_ttfb": np.median(ttfb_times) if ttfb_times else None,
"p95_ttfb": np.percentile(ttfb_times, 95) if ttfb_times else None,
"average_total": np.mean(total_times) if total_times else None,
"median_total": np.median(total_times) if total_times else None,
"p95_total": np.percentile(total_times, 95) if total_times else None,
"ttfb_times": ttfb_times,
"total_times": total_times,
"token_counts": token_counts,
"avg_tokens": (
np.mean(token_counts) if token_counts and any(token_counts) else None
),
"tokens_per_second": tokens_per_second,
"avg_tokens_per_second": (
np.mean([t for t in tokens_per_second if t > 0])
if tokens_per_second and any(tokens_per_second)
else None
),
"median_tokens_per_second": (
np.median([t for t in tokens_per_second if t > 0])
if tokens_per_second and any(tokens_per_second)
else None
),
"success_rate": (
len(ttfb_times) / (len(ttfb_times) + errors)
if (len(ttfb_times) + errors) > 0
else 0
),
"error_count": errors,
}
def print_comparison_table(results_avalai, results_openai):
"""Print comparison table between two APIs"""
headers = [
"Metric",
f"AvalAI ({results_avalai['model']})",
f"OpenAI ({results_openai['model']})",
]
data = [
[
"Average TTFB (s)",
f"{results_avalai['average_ttfb']:.3f}",
f"{results_openai['average_ttfb']:.3f}",
],
[
"Median TTFB (s)",
f"{results_avalai['median_ttfb']:.3f}",
f"{results_openai['median_ttfb']:.3f}",
],
[
"95th Percentile TTFB (s)",
f"{results_avalai['p95_ttfb']:.3f}",
f"{results_openai['p95_ttfb']:.3f}",
],
[
"Average Total Time (s)",
f"{results_avalai['average_total']:.3f}",
f"{results_openai['average_total']:.3f}",
],
[
"Median Total Time (s)",
f"{results_avalai['median_total']:.3f}",
f"{results_openai['median_total']:.3f}",
],
[
"95th Percentile Total (s)",
f"{results_avalai['p95_total']:.3f}",
f"{results_openai['p95_total']:.3f}",
],
[
"Success Rate",
f"{results_avalai['success_rate']:.2%}",
f"{results_openai['success_rate']:.2%}",
],
]
# Add token metrics if available
if (
results_avalai["avg_tokens"] is not None
and results_openai["avg_tokens"] is not None
):
data.append(
[
"Avg Tokens per Response",
f"{results_avalai['avg_tokens']:.1f}",
f"{results_openai['avg_tokens']:.1f}",
]
)
# Add tokens per second metrics if available
if (
results_avalai["avg_tokens_per_second"] is not None
and results_openai["avg_tokens_per_second"] is not None
):
data.append(
[
"Avg Tokens per Second",
f"{results_avalai['avg_tokens_per_second']:.1f}",
f"{results_openai['avg_tokens_per_second']:.1f}",
]
)
data.append(
[
"Median Tokens per Second",
f"{results_avalai['median_tokens_per_second']:.1f}",
f"{results_openai['median_tokens_per_second']:.1f}",
]
)
print("\nAPI Performance Comparison:")
print(tabulate(data, headers=headers, tablefmt="grid"))
def plot_comparison(results_avalai, results_openai, output_file=None):
"""Create improved visualization plots for API comparison"""
# Set the style
sns.set(style="whitegrid")
# Create figure with subplots - adding a third subplot for tokens per second
fig, axes = plt.subplots(3, 1, figsize=(12, 15))
# Define metrics to plot
metrics = [
("ttfb_times", "Time to First Byte (s)"),
("total_times", "Total Request Time (s)"),
("tokens_per_second", "Tokens per Second"),
]
# Define colors for each API
colors = {"AvalAI": "#3498db", "OpenAI": "#2ecc71"}
for i, (metric, title) in enumerate(metrics):
# Create violin plots with individual points
ax = axes[i]
# Prepare data for plotting
data_to_plot = []
labels = []
for result, label in [(results_avalai, "AvalAI"), (results_openai, "OpenAI")]:
# Filter out zeros for tokens per second
if metric == "tokens_per_second":
data_to_plot.append([t for t in result[metric] if t > 0])
else:
data_to_plot.append(result[metric])
labels.append(f"{label}\n({result['model']})")
# Create violin plot
parts = ax.violinplot(data_to_plot, showmeans=True, showmedians=True)
# Customize violin plots
for pc, color_key in zip(parts["bodies"], colors.keys()):
pc.set_facecolor(colors[color_key])
pc.set_alpha(0.7)
# Add boxplot inside violin
bp = ax.boxplot(
data_to_plot,
positions=range(1, len(data_to_plot) + 1),
widths=0.15,
patch_artist=True,
showfliers=False,
)
# Customize boxplots
for box, color_key in zip(bp["boxes"], colors.keys()):
box.set(color="black", linewidth=1.5)
box.set(facecolor="white")
# Add scatter points with jitter
for j, data in enumerate(
[
(
results_avalai[metric]
if metric != "tokens_per_second"
else [t for t in results_avalai[metric] if t > 0]
),
(
results_openai[metric]
if metric != "tokens_per_second"
else [t for t in results_openai[metric] if t > 0]
),
]
):
# Add jitter to x position
x = np.random.normal(j + 1, 0.05, size=len(data))
ax.scatter(
x,
data,
alpha=0.4,
s=20,
color=list(colors.values())[j],
edgecolor="white",
linewidth=0.5,
)
# Set labels and title
ax.set_title(title, fontsize=14, fontweight="bold")
if metric == "tokens_per_second":
ax.set_ylabel("Tokens/second", fontsize=12)
else:
ax.set_ylabel("Time (seconds)", fontsize=12)
ax.set_xticks(range(1, len(labels) + 1))
ax.set_xticklabels(labels, fontsize=12)
# Add horizontal grid lines
ax.yaxis.grid(True, linestyle="--", alpha=0.7)
# Add stats as text
for j, (result, label) in enumerate(
[(results_avalai, "AvalAI"), (results_openai, "OpenAI")]
):
if metric == "tokens_per_second":
if result["avg_tokens_per_second"] is not None:
stats = (
f"Mean: {result['avg_tokens_per_second']:.1f}\n"
f"Median: {result['median_tokens_per_second']:.1f}"
)
max_val = (
max([t for t in result[metric] if t > 0])
if any(t > 0 for t in result[metric])
else 0
)
ax.annotate(
stats,
xy=(j + 1, max_val * 1.05),
ha="center",
va="bottom",
fontsize=10,
bbox=dict(boxstyle="round,pad=0.5", fc="white", alpha=0.7),
)
else:
stats = (
f"Mean: {result[f'average_{metric.split("_")[0]}']:.3f}s\n"
f"Median: {result[f'median_{metric.split("_")[0]}']:.3f}s\n"
f"95th: {result[f'p95_{metric.split("_")[0]}']:.3f}s"
)
ax.annotate(
stats,
xy=(j + 1, result[f'p95_{metric.split("_")[0]}'] * 1.05),
ha="center",
va="bottom",
fontsize=10,
bbox=dict(boxstyle="round,pad=0.5", fc="white", alpha=0.7),
)
# Add title and timestamp
plt.suptitle(
f"API Performance Comparison: AvalAI vs OpenAI", fontsize=16, fontweight="bold"
)
plt.figtext(
0.5,
0.01,
f'Generated on {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}',
ha="center",
fontsize=10,
)
plt.tight_layout(rect=[0, 0.03, 1, 0.97])
if output_file:
plt.savefig(output_file, dpi=300, bbox_inches="tight")
print(f"Plot saved to {output_file}")
else:
plt.show()
def save_results(results_avalai, results_openai, filename):
"""Save results to JSON file"""
# Convert numpy arrays to lists for JSON serialization
results_avalai_copy = results_avalai.copy()
results_openai_copy = results_openai.copy()
for key in ["ttfb_times", "total_times", "token_counts"]:
if key in results_avalai_copy:
results_avalai_copy[key] = [float(x) for x in results_avalai_copy[key]]
if key in results_openai_copy:
results_openai_copy[key] = [float(x) for x in results_openai_copy[key]]
data = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"results": {"avalai": results_avalai_copy, "openai": results_openai_copy},
}
with open(filename, "w") as f:
json.dump(data, f, indent=2)
print(f"Results saved to {filename}")
def main():
# API configuration
model_name = "gpt-4o-mini"
url_avalai = "https://api.avalai.ir/v1/chat/completions"
api_key_avalai = os.getenv("AVALAI_API_KEY") # Replace with actual key
url_openai = "https://api.openai.com/v1/chat/completions"
api_key_openai = os.getenv("OPENAI_API_KEY") # Replace with actual key
# Number of requests to make for each API
num_requests = 60
# Test prompt
prompt = "Say hi"
# Run the tests
results_avalai = test_api_performance(
"AvalAI", url_avalai, api_key_avalai, model_name, num_requests, prompt
)
results_openai = test_api_performance(
"OpenAI", url_openai, api_key_openai, model_name, num_requests, prompt
)
# Print comparison table
print_comparison_table(results_avalai, results_openai)
# Generate visualization
plot_comparison(results_avalai, results_openai, "api_performance_comparison.png")
# Save results
save_results(results_avalai, results_openai, "api_performance_results.json")
if __name__ == "__main__":
main()