""" Minimal NVIDIA OpenAI-Compatible Gateway (نسخة محسّنة) Purpose: - OpenAI-compatible proxy for NVIDIA API. - Dynamic /v1/models (بدون فلترة — نفس السلوك الأصلي). - Chat completions. - Transparent SSE streaming. - Native tools pass-through. - Clear upstream errors. - Silent automatic retry on transient upstream errors (429 / 500 / 502 / 503 / 504 / 529) with exponential backoff. - One file, no Agent framework, no SSE parsing. التغييرات عن النسخة الأصلية: 1. إصلاح خطأ حقيقي: `asyncio` كان مُستخدَماً (asyncio.CancelledError) دون استيراده — كان يسبب NameError عند انقطاع اتصال العميل أثناء البث. 2. إعادة محاولة تلقائية صامتة عند أخطاء مؤقتة من NVIDIA (429 Rate Limit / 500 / 502 / 503 / 504 / 529 Overloaded) — الحالة التي ظهرت في اللوق عندك (529) لم تكن تُعاد محاولتها إطلاقاً سابقاً. 3. شكل الاستجابة النهائية (نجاح أو فشل) يبقى مطابقاً 100% للأصلي — Passthrough خام لبايتات NVIDIA كما هي، بدون أي تغليف أو رسائل مخصصة، حتى لا ينكسر أي عميل يعتمد على شكل استجابة NVIDIA الأصلي. 4. لا فلترة لأي نموذج في /v1/models — القائمة تُعرض كما ترجعها NVIDIA تماماً. """ import os import json import uuid import time import asyncio import logging from collections import deque from contextlib import asynccontextmanager from typing import Any, AsyncIterator import httpx import uvicorn from fastapi import FastAPI, HTTPException, Request, Security from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, Response, StreamingResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer # ============================================================ # CONFIG # ============================================================ NVIDIA_API_KEY = os.getenv("PROVIDER_API_KEY", "").strip() # دعم مفاتيح متعددة: PROVIDER_API_KEYS="key1,key2,key3" # يبقى PROVIDER_API_KEY (مفرد) يعمل كما هو للتوافق العكسي. _raw_multi_keys = os.getenv("PROVIDER_API_KEYS", "").strip() if _raw_multi_keys: NVIDIA_API_KEYS = [ k.strip() for k in _raw_multi_keys.split(",") if k.strip() ] elif NVIDIA_API_KEY: NVIDIA_API_KEYS = [NVIDIA_API_KEY] else: NVIDIA_API_KEYS = [] # حصة كل مفتاح على حدة بالطلب/دقيقة (اجعلها أقل قليلاً من حد NVIDIA الفعلي # كهامش أمان — الحد المعروف هو ~40، لذا الافتراضي هنا 38). RATE_LIMIT_RPM_PER_KEY = int(os.getenv("RATE_LIMIT_RPM_PER_KEY", "38")) # أقصى مدة انتظار محلي (queue) قبل إرجاع خطأ للعميل، بدل الانتظار للأبد RATE_LIMIT_MAX_WAIT_SECONDS = float(os.getenv("RATE_LIMIT_MAX_WAIT_SECONDS", "60")) # مدة تبريد مفتاح تلقّى 429 منه تحديداً، قبل إعادة استخدامه KEY_COOLDOWN_SECONDS = float(os.getenv("KEY_COOLDOWN_SECONDS", "15")) NVIDIA_BASE_URL = os.getenv( "PROVIDER_BASE_URL", "https://integrate.api.nvidia.com", ).rstrip("/") DEFAULT_MODEL = os.getenv( "DEFAULT_MODEL", "", ).strip() HOST = os.getenv("HOST", "0.0.0.0") PORT = int(os.getenv("PORT", "7860")) LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper() CONNECT_TIMEOUT = float( os.getenv("CONNECT_TIMEOUT", "20") ) READ_TIMEOUT = float( os.getenv("READ_TIMEOUT", "300") ) WRITE_TIMEOUT = float( os.getenv("WRITE_TIMEOUT", "60") ) POOL_TIMEOUT = float( os.getenv("POOL_TIMEOUT", "30") ) MAX_CONNECTIONS = int( os.getenv("MAX_CONNECTIONS", "100") ) MAX_KEEPALIVE_CONNECTIONS = int( os.getenv("MAX_KEEPALIVE_CONNECTIONS", "20") ) # عدد مرات إعادة المحاولة عند خطأ مؤقت من NVIDIA (بدون احتساب المحاولة الأولى) MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) # الحد الأقصى للانتظار بين المحاولات (ثواني) MAX_BACKOFF_SECONDS = float(os.getenv("MAX_BACKOFF_SECONDS", "10")) # أكواد الحالة المؤقتة التي تستحق إعادة المحاولة تلقائياً # 429 = تجاوز حد الطلبات | 500/502/503/504 = أخطاء سيرفر/بوابة # 529 = ازدحام مؤقت (شائع جداً مع النماذج الكبيرة على NVIDIA) RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504, 529} # أنماط نصية تدل على خطأ حتمي (deterministic) وليس عابراً — # إعادة المحاولة على هذه الأخطاء عديمة الفائدة لأن النتيجة ستتكرر دائماً # (مشاكل توافق قالب النموذج مع صيغة الطلب، وليست ازدحاماً أو عطلاً مؤقتاً). NON_RETRYABLE_ERROR_PATTERNS = ( "prompt template", "invalid operation", "cannot convert value into pairs", "badrequesterror", "extra data", "unexpected token", "jsondecodeerror", ) def is_deterministic_error(body_text: str) -> bool: """يتحقق إذا كان نص الخطأ يدل على مشكلة توافق ثابتة (لا فائدة من إعادة المحاولة).""" lowered = body_text.lower() return any(pattern in lowered for pattern in NON_RETRYABLE_ERROR_PATTERNS) # ============================================================ # RATE LIMITING + MULTI-KEY POOL # ============================================================ # # كل مفتاح NVIDIA له حصة RPM منفصلة تماماً عن الآخر. الـ KeyPool يوزّع # الطلبات بالتناوب (round-robin) على المفاتيح المتاحة، وينتظر محلياً # (queue) بدل إرسال الطلب مباشرة وتلقي 429 لو كل المفاتيح مشغولة. class RateLimiter: """محدد معدل (Sliding Window) لمفتاح واحد.""" def __init__(self, max_calls: int, window_seconds: float = 60.0): self.max_calls = max_calls self.window_seconds = window_seconds self.timestamps: deque[float] = deque() self.lock = asyncio.Lock() async def try_acquire(self) -> bool: """يحاول الحصول على فتحة فوراً، بدون انتظار. True إن نجح.""" async with self.lock: now = time.monotonic() while self.timestamps and now - self.timestamps[0] >= self.window_seconds: self.timestamps.popleft() if len(self.timestamps) < self.max_calls: self.timestamps.append(now) return True return False class KeyPool: """ يدير مفتاحاً واحداً أو أكثر من مفاتيح NVIDIA. - مفتاح واحد فقط: يتصرف كـ rate limiter بسيط (queue محلي بدل انهيار عند 429). - عدة مفاتيح: يوزّع الطلبات بالتناوب، ويتجاوز مؤقتاً أي مفتاح تلقّى 429. """ def __init__( self, keys: list[str], rpm_per_key: int, cooldown_seconds: float = 15.0, ): if not keys: raise ValueError("at least one API key is required") self.keys = keys self.limiters = [RateLimiter(rpm_per_key) for _ in keys] self.cooldown_seconds = cooldown_seconds self._next_index = 0 self._index_lock = asyncio.Lock() self._cooldown_until: dict[int, float] = {} async def acquire( self, max_wait_seconds: float = 60.0, ) -> tuple[str | None, int]: """ يرجع (key, index) بعد ضمان توفر حصة ضمن حد RPM لأحد المفاتيح. يجرب كل المفاتيح بالتناوب، ويتجاوز أي مفتاح في فترة تبريد. يرجع (None, -1) إن انتظرنا أكثر من max_wait_seconds بلا جدوى. """ start = time.monotonic() n = len(self.keys) while True: async with self._index_lock: start_index = self._next_index self._next_index = (self._next_index + 1) % n now = time.monotonic() for offset in range(n): idx = (start_index + offset) % n if now < self._cooldown_until.get(idx, 0): continue if await self.limiters[idx].try_acquire(): return self.keys[idx], idx if time.monotonic() - start > max_wait_seconds: return None, -1 await asyncio.sleep(0.25) def mark_cooldown(self, idx: int) -> None: """يضع مفتاحاً في فترة تبريد بعد تلقيه 429 تحديداً (وليس خطأ عام).""" if 0 <= idx < len(self.keys): self._cooldown_until[idx] = ( time.monotonic() + self.cooldown_seconds ) def load_gateway_keys() -> set[str]: raw = ( os.getenv("GATEWAY_API_KEYS") or os.getenv("API_KEY") or "" ) return { key.strip() for key in raw.split(",") if key.strip() } GATEWAY_API_KEYS = load_gateway_keys() # ============================================================ # LOGGING # ============================================================ logging.basicConfig( level=LOG_LEVEL, format=( "%(asctime)s " "%(levelname)s " "%(name)s " "%(message)s" ), ) logger = logging.getLogger("nvidia-gateway") # ============================================================ # SECURITY # ============================================================ security = HTTPBearer( auto_error=False ) def verify_gateway_key( credentials: HTTPAuthorizationCredentials | None = Security(security), ) -> str: if not GATEWAY_API_KEYS: raise HTTPException( status_code=500, detail={ "error": { "message": ( "GATEWAY_API_KEYS or API_KEY " "is not configured." ), "type": "server_error", "code": "gateway_key_not_configured", } }, ) if credentials is None: raise HTTPException( status_code=401, detail={ "error": { "message": "Missing Authorization header.", "type": "authentication_error", "code": "missing_api_key", } }, ) if credentials.credentials not in GATEWAY_API_KEYS: raise HTTPException( status_code=401, detail={ "error": { "message": "Invalid API key.", "type": "authentication_error", "code": "invalid_api_key", } }, ) return credentials.credentials # ============================================================ # HELPERS # ============================================================ def nvidia_headers( *, key: str, stream: bool = False, ) -> dict[str, str]: headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json", "Accept": ( "text/event-stream" if stream else "application/json" ), } return headers def json_response_from_upstream( response: httpx.Response, ) -> Response: content_type = response.headers.get( "content-type", "application/json", ) return Response( content=response.content, status_code=response.status_code, media_type=content_type.split(";")[0], ) def gateway_error( message: str, *, status_code: int = 502, error_type: str = "upstream_error", code: str = "upstream_failure", ) -> JSONResponse: return JSONResponse( status_code=status_code, content={ "error": { "message": message, "type": error_type, "param": None, "code": code, } }, ) def backoff_seconds(attempt: int) -> float: """تأخير تصاعدي بين المحاولات، بحد أقصى MAX_BACKOFF_SECONDS.""" return min(2 ** attempt, MAX_BACKOFF_SECONDS) # ============================================================ # LIFESPAN # ============================================================ @asynccontextmanager async def lifespan(app: FastAPI): if not NVIDIA_API_KEYS: logger.error( "PROVIDER_API_KEY / PROVIDER_API_KEYS is not configured." ) app.state.key_pool = None else: app.state.key_pool = KeyPool( keys=NVIDIA_API_KEYS, rpm_per_key=RATE_LIMIT_RPM_PER_KEY, cooldown_seconds=KEY_COOLDOWN_SECONDS, ) if not GATEWAY_API_KEYS: logger.error( "GATEWAY_API_KEYS/API_KEY is not configured." ) timeout = httpx.Timeout( connect=CONNECT_TIMEOUT, read=READ_TIMEOUT, write=WRITE_TIMEOUT, pool=POOL_TIMEOUT, ) limits = httpx.Limits( max_connections=MAX_CONNECTIONS, max_keepalive_connections=MAX_KEEPALIVE_CONNECTIONS, ) app.state.client = httpx.AsyncClient( timeout=timeout, limits=limits, follow_redirects=False, ) logger.info( "Gateway started upstream=%s max_retries=%s keys=%d rpm_per_key=%s", NVIDIA_BASE_URL, MAX_RETRIES, len(NVIDIA_API_KEYS), RATE_LIMIT_RPM_PER_KEY, ) try: yield finally: await app.state.client.aclose() logger.info( "Gateway stopped." ) # ============================================================ # APP # ============================================================ app = FastAPI( title="Minimal NVIDIA OpenAI Gateway", version="1.1.0", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # ============================================================ # ROOT / HEALTH # ============================================================ @app.get("/") async def root() -> dict[str, Any]: return { "status": "online", "engine": "Minimal-NVIDIA-Gateway", "version": "1.1.0", "upstream": NVIDIA_BASE_URL, "configured_keys": len(NVIDIA_API_KEYS), "rpm_per_key": RATE_LIMIT_RPM_PER_KEY, } @app.get("/health") async def health() -> dict[str, str]: return { "status": "ok", } # ============================================================ # MODELS # ============================================================ @app.get("/models") @app.get("/v1/models") async def list_models( request: Request, _: str = Security(verify_gateway_key), ): key_pool: KeyPool | None = request.app.state.key_pool if key_pool is None: return gateway_error( "PROVIDER_API_KEY / PROVIDER_API_KEYS is not configured.", status_code=500, error_type="server_error", code="provider_key_not_configured", ) key, key_idx = await key_pool.acquire(max_wait_seconds=RATE_LIMIT_MAX_WAIT_SECONDS) if key is None: return gateway_error( "All NVIDIA API keys are currently rate-limited " "(local queue timeout). Try again shortly.", status_code=429, error_type="rate_limit_error", code="local_queue_timeout", ) client: httpx.AsyncClient = request.app.state.client try: response = await client.get( f"{NVIDIA_BASE_URL}/v1/models", headers=nvidia_headers(key=key), ) except httpx.TimeoutException as exc: return gateway_error( f"NVIDIA models request timed out: {exc}", code="upstream_timeout", ) except httpx.RequestError as exc: return gateway_error( f"Cannot reach NVIDIA API: {exc}", code="upstream_unavailable", ) if response.status_code == 429: key_pool.mark_cooldown(key_idx) return json_response_from_upstream(response) # ============================================================ # CHAT COMPLETIONS # ============================================================ @app.post("/chat/completions") @app.post("/v1/chat/completions") async def chat_completions( request: Request, _: str = Security(verify_gateway_key), ): key_pool: KeyPool | None = request.app.state.key_pool if key_pool is None: return gateway_error( "PROVIDER_API_KEY / PROVIDER_API_KEYS is not configured.", status_code=500, error_type="server_error", code="provider_key_not_configured", ) # -------------------------------------------------------- # READ JSON BODY # -------------------------------------------------------- try: payload = await request.json() except Exception: return gateway_error( "Request body must be valid JSON.", status_code=400, error_type="invalid_request_error", code="invalid_json", ) if not isinstance(payload, dict): return gateway_error( "Request body must be a JSON object.", status_code=400, error_type="invalid_request_error", code="invalid_request_body", ) # -------------------------------------------------------- # MINIMUM VALIDATION # -------------------------------------------------------- model = payload.get("model") if not model: if DEFAULT_MODEL: payload["model"] = DEFAULT_MODEL else: return gateway_error( "model is required.", status_code=400, error_type="invalid_request_error", code="missing_model", ) messages = payload.get("messages") if not isinstance(messages, list): return gateway_error( "messages must be an array.", status_code=400, error_type="invalid_request_error", code="invalid_messages", ) stream = payload.get("stream") is True client: httpx.AsyncClient = request.app.state.client # ======================================================== # STREAMING # # Important: # Do not parse SSE. # Do not use aiter_lines(). # Do not rebuild chunks. # # Forward NVIDIA response bytes transparently. # # إعادة المحاولة هنا تشمل فقط محاولة إنشاء الاتصال والحصول على # أول استجابة (headers/status) من NVIDIA. بمجرد بدء البث الفعلي # للبيانات لا تُعاد أي محاولة (لتفادي تكرار محتوى للعميل). # ======================================================== if stream: upstream_response = None for attempt in range(MAX_RETRIES + 1): key, key_idx = await key_pool.acquire( max_wait_seconds=RATE_LIMIT_MAX_WAIT_SECONDS ) if key is None: return gateway_error( "All NVIDIA API keys are currently rate-limited " "(local queue timeout). Try again shortly.", status_code=429, error_type="rate_limit_error", code="local_queue_timeout", ) upstream_request = client.build_request( "POST", f"{NVIDIA_BASE_URL}/v1/chat/completions", headers=nvidia_headers(key=key, stream=True), json=payload, ) try: upstream_response = await client.send( upstream_request, stream=True, ) except httpx.TimeoutException as exc: if attempt < MAX_RETRIES: logger.warning( "Stream connect timeout, retrying " "attempt=%d/%d model=%r", attempt + 1, MAX_RETRIES, payload.get("model"), ) await asyncio.sleep(backoff_seconds(attempt)) continue return gateway_error( f"NVIDIA connection timed out: {exc}", code="upstream_timeout", ) except httpx.RequestError as exc: if attempt < MAX_RETRIES: logger.warning( "Stream connect error, retrying " "attempt=%d/%d model=%r", attempt + 1, MAX_RETRIES, payload.get("model"), ) await asyncio.sleep(backoff_seconds(attempt)) continue return gateway_error( f"Cannot reach NVIDIA API: {exc}", code="upstream_unavailable", ) if upstream_response.status_code == 429: key_pool.mark_cooldown(key_idx) # نجح الاتصال بكود غير قابل لإعادة المحاولة (نجاح، أو خطأ نهائي # مثل 404/400/401) — سجّل إن كان خطأ، ثم تابع للأسفل كما هو if upstream_response.status_code not in RETRYABLE_STATUS_CODES: if upstream_response.status_code >= 400: logger.error( "NVIDIA HTTP error status=%s model=%r " "(non-retryable, e.g. 404/400/401)", upstream_response.status_code, payload.get("model"), ) break # كود حالة مؤقت محتمل (429/500/502/503/504/529) — # نقرأ الجسم لنقرر: عابر فنعيد المحاولة، أم حتمي فنرجعه فوراً. try: error_body = await upstream_response.aread() finally: await upstream_response.aclose() error_text = error_body.decode("utf-8", errors="replace") logger.error( "NVIDIA HTTP error status=%s model=%r attempt=%d body=%s", upstream_response.status_code, payload.get("model"), attempt + 1, error_text[:2000], ) # خطأ حتمي (توافق قالب/صيغة) — لا فائدة من إعادة المحاولة if is_deterministic_error(error_text): logger.warning( "NVIDIA deterministic error (not retrying) status=%s " "model=%r — likely a template/payload compatibility " "issue, not transient.", upstream_response.status_code, payload.get("model"), ) return Response( content=error_body, status_code=upstream_response.status_code, media_type=( upstream_response.headers .get("content-type", "application/json") .split(";")[0] ), ) # خطأ عابر — أعد المحاولة إن بقيت محاولات if attempt < MAX_RETRIES: logger.warning( "♻️ إعادة محاولة البث %d/%d للنموذج %s (status=%s)", attempt + 1, MAX_RETRIES, payload.get("model"), upstream_response.status_code, ) await asyncio.sleep(backoff_seconds(attempt)) continue # استنفدنا المحاولات — أرجع آخر خطأ كما هو return Response( content=error_body, status_code=upstream_response.status_code, media_type=( upstream_response.headers .get("content-type", "application/json") .split(";")[0] ), ) # ---------------------------------------------------- # RAW BYTE STREAM (فقط عند نجاح الاتصال بكود < 400) # ---------------------------------------------------- async def proxy_stream() -> AsyncIterator[bytes]: try: async for chunk in upstream_response.aiter_raw(): if await request.is_disconnected(): logger.info( "Client disconnected " "model=%r", payload.get("model"), ) break if chunk: yield chunk except asyncio.CancelledError: raise except httpx.TimeoutException: logger.exception( "NVIDIA stream timed out " "model=%r", payload.get("model"), ) except httpx.RequestError: logger.exception( "NVIDIA stream network failure " "model=%r", payload.get("model"), ) except Exception: logger.exception( "Unexpected stream failure " "model=%r", payload.get("model"), ) finally: await upstream_response.aclose() response_headers = { "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "X-Request-ID": ( "req_" + uuid.uuid4().hex[:24] ), } upstream_content_type = ( upstream_response.headers.get( "content-type", "text/event-stream", ) ) return StreamingResponse( proxy_stream(), status_code=upstream_response.status_code, media_type=upstream_content_type.split(";")[0], headers=response_headers, ) # ======================================================== # NON-STREAMING — مع إعادة محاولة صامتة عند أخطاء مؤقتة # ======================================================== response = None for attempt in range(MAX_RETRIES + 1): key, key_idx = await key_pool.acquire( max_wait_seconds=RATE_LIMIT_MAX_WAIT_SECONDS ) if key is None: return gateway_error( "All NVIDIA API keys are currently rate-limited " "(local queue timeout). Try again shortly.", status_code=429, error_type="rate_limit_error", code="local_queue_timeout", ) try: response = await client.post( f"{NVIDIA_BASE_URL}/v1/chat/completions", headers=nvidia_headers(key=key), json=payload, ) except httpx.TimeoutException as exc: if attempt < MAX_RETRIES: logger.warning( "Request timeout, retrying " "attempt=%d/%d model=%r", attempt + 1, MAX_RETRIES, payload.get("model"), ) await asyncio.sleep(backoff_seconds(attempt)) continue return gateway_error( f"NVIDIA request timed out: {exc}", code="upstream_timeout", ) except httpx.RequestError as exc: if attempt < MAX_RETRIES: logger.warning( "Request error, retrying " "attempt=%d/%d model=%r", attempt + 1, MAX_RETRIES, payload.get("model"), ) await asyncio.sleep(backoff_seconds(attempt)) continue return gateway_error( f"Cannot reach NVIDIA API: {exc}", code="upstream_unavailable", ) if response.status_code == 429: key_pool.mark_cooldown(key_idx) if response.status_code not in RETRYABLE_STATUS_CODES: break # تحقق: هل هذا خطأ حتمي (توافق قالب/صيغة) وليس عابراً؟ # إن كان كذلك، لا فائدة من إعادة المحاولة — النتيجة ستتكرر دائماً. if is_deterministic_error(response.text): logger.warning( "NVIDIA deterministic error (not retrying) status=%s " "model=%r — likely a template/payload compatibility issue, " "not a transient failure. body=%s", response.status_code, payload.get("model"), response.text[:500], ) break if attempt < MAX_RETRIES: logger.warning( "NVIDIA transient error status=%s model=%r " "attempt=%d/%d — retrying. body=%s", response.status_code, payload.get("model"), attempt + 1, MAX_RETRIES, response.text[:500], ) await asyncio.sleep(backoff_seconds(attempt)) continue # استنفدنا المحاولات break if response.status_code >= 400: logger.error( "NVIDIA HTTP error " "status=%s model=%r body=%s", response.status_code, payload.get("model"), response.text[:8000], ) return json_response_from_upstream(response) # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": uvicorn.run( app, host=HOST, port=PORT, log_level=LOG_LEVEL.lower(), )