چگونه یک API هوش مصنوعی قابل‌اعتماد برای محیط عملیاتی بسازیم؟ راهنمای Timeout، Retry، Rate Limit، Fallback و Monitoring

مقدمه

اتصال یک نرم‌افزار به مدل هوش مصنوعی در محیط آزمایشی معمولاً ساده است. یک درخواست به API ارسال می‌کنیم، چند ثانیه منتظر می‌مانیم و پاسخ مدل را نمایش می‌دهیم:

response = client.chat.completions.create(
    model="MODEL_ID",
    messages=[
        {
            "role": "user",
            "content": "یک متن کوتاه درباره هوش مصنوعی بنویس."
        }
    ]
)

این کد برای اولین آزمایش مناسب است، اما یک سیستم واقعی باید با وضعیت‌های بسیار پیچیده‌تری روبه‌رو شود:

• مدل دیر پاسخ می‌دهد.
• اتصال کاربر هنگام Streaming قطع می‌شود.
• ارائه‌دهنده خطای موقت برمی‌گرداند.
• سقف RPM یا TPM پر می‌شود.
• موجودی حساب کافی نیست.
• درخواست دو بار ارسال می‌شود.
• مدل اصلی از دسترس خارج می‌شود.
• مدل جایگزین قابلیت موردنیاز را ندارد.
• خروجی مدل JSON نامعتبر است.
• یک درخواست بسیار بزرگ هزینه غیرمنتظره ایجاد می‌کند.
• کاربر Prompt مخرب وارد می‌کند.
• ابزار Agent عملیاتی حساس را اجرا می‌کند.
• یک Retry ساده، عملیات را دو بار انجام می‌دهد.
• هیچ‌کس نمی‌داند خطا در کدام بخش مسیر رخ داده است.

در محیط Production نباید فرض کنیم شبکه، مدل، کاربر، ارائه‌دهنده یا حتی کد خودمان همیشه درست عمل می‌کنند. یک سرویس هوش مصنوعی قابل‌اعتماد باید خطا را پیش‌بینی، محدود، ثبت و مدیریت کند.

این مقاله معماری و الگوهای فنی لازم برای ساخت یک API هوش مصنوعی پایدار را بررسی می‌کند و نمونه‌کدهای عملی آن را با API سازگار با OpenAI درواره ارائه می‌دهد.

فهرست مطالب

• تفاوت Prototype و Production
• معماری یک API هوش مصنوعی قابل‌اعتماد
• طبقه‌بندی خطاهای API
• مدیریت Timeout
• Retry و Exponential Backoff
• Jitter چیست؟
• Rate Limit، RPM و TPM
• کنترل درخواست‌های هم‌زمان
• Fallback میان مدل‌ها
• Circuit Breaker
• Idempotency
• Streaming در محیط Production
• Queue و پردازش غیرهم‌زمان
• اعتبارسنجی ورودی و خروجی
• کنترل Token و هزینه
• Logging، Metrics و Tracing
• امنیت API Key
• امنیت Prompt و Tool Calling
• Caching
• پیاده‌سازی با Python
• پیاده‌سازی با TypeScript
• تست و ارزیابی پایداری
• چک‌لیست Production
• پرسش‌های متداول
• جمع‌بندی

تفاوت Prototype و Production

در Prototype هدف این است که ثابت کنیم ایده کار می‌کند. در Production باید ثابت کنیم سیستم در شرایط واقعی نیز قابل‌استفاده، قابل‌پیش‌بینی و قابل‌کنترل است.

یک سیستم Production فقط «پاسخ تولید نمی‌کند». باید بتواند در شرایط شکست نیز رفتار کنترل‌شده‌ای داشته باشد.

معماری یک API هوش مصنوعی قابل‌اعتماد

یک مسیر استاندارد برای درخواست هوش مصنوعی می‌تواند شامل لایه‌های زیر باشد:

Client
→ Authentication
→ Input Validation
→ User Rate Limit
→ Wallet and Quota Check
→ Token and Cost Estimation
→ Prompt and Context Builder
→ Model Router
→ Provider API
→ Output Validation
→ Usage Recording
→ Response

در کنار مسیر اصلی، چند قابلیت مشترک نیز لازم است:

• Timeout
• Retry
• Circuit Breaker
• Idempotency
• Logging
• Metrics
• Tracing
• Cache
• Queue
• Fallback
• Audit Log

مسئولیت هر لایه

Authentication

هویت درخواست‌کننده و اعتبار API Key را بررسی می‌کند.

Authorization

مشخص می‌کند کاربر اجازه استفاده از کدام مدل، قابلیت یا سازمان را دارد.

Rate Limiter

تعداد درخواست‌ها، Tokenها و درخواست‌های هم‌زمان را محدود می‌کند.

Billing Guard

موجودی، سقف هزینه و محدودیت مصرف را پیش از فراخوانی مدل بررسی می‌کند.

Context Builder

System Prompt، تاریخچه، اسناد، حافظه و ابزارهای مرتبط را می‌سازد.

Model Router

مدل و ارائه‌دهنده مناسب را براساس قابلیت، هزینه، سلامت و سیاست انتخاب می‌کند.

Output Validator

پاسخ مدل را از نظر ساختار، ایمنی و قواعد کسب‌وکار بررسی می‌کند.

Usage Recorder

مصرف Token، هزینه، مدل، Latency و وضعیت نهایی را ثبت می‌کند.

طبقه‌بندی خطاهای API

همه خطاها نباید یکسان مدیریت شوند. اولین قدم، تشخیص نوع خطاست.

خطاهای درخواست

معمولاً شامل کدهای 400، 401، 403 و 404 هستند:

• بدنه درخواست نامعتبر
• API Key اشتباه
• مدل ناموجود
• دسترسی ناکافی
• پارامتر پشتیبانی‌نشده
• Context بزرگ‌تر از ظرفیت مدل
• Schema نامعتبر

این خطاها معمولاً با Retry ساده حل نمی‌شوند.

خطاهای محدودیت مصرف

کد 429 می‌تواند دلایل متفاوتی داشته باشد:

• تعداد درخواست بیش‌ازحد
• Token بیش‌ازحد در یک بازه
• درخواست‌های هم‌زمان زیاد
• سقف اعتبار یا بودجه
• محدودیت حساب یا پروژه

بنابراین فقط دیدن عدد 429 کافی نیست؛ باید متن خطا و Headerهای پاسخ نیز بررسی شوند. مستندات OpenAI نیز میان Rate Limit و تمام‌شدن سهمیه یا اعتبار تمایز قائل می‌شوند. راهنمای کدهای خطای OpenAI

خطاهای موقت سرور

کدهای 500، 502، 503 و 504 معمولاً می‌توانند موقت باشند:

• خطای داخلی ارائه‌دهنده
• Gateway نامعتبر
• سرویس موقتاً در دسترس نیست
• Timeout بالادستی

Retry محدود یا Fallback در این وضعیت منطقی است.

خطاهای شبکه

• DNS
• قطع اتصال
• TLS
• Connection Reset
• Socket Timeout
• قطع ارتباط هنگام Streaming

این خطاها ممکن است موقت باشند، اما باید مشخص شود درخواست اصلی در بالادست اجرا شده یا نه.

خطاهای معنایی

ممکن است HTTP Status موفق باشد، اما خروجی قابل‌استفاده نباشد:

• پاسخ خالی
• JSON نامعتبر
• پاسخ خارج از Schema
• Refusal
• پاسخ ناقص
• Hallucination
• پایان خروجی به‌دلیل محدودیت Token
• Tool Call با آرگومان نامعتبر

این خطاها در لایه Validation شناسایی می‌شوند، نه صرفاً از روی HTTP Status.

مدیریت Timeout

بدون Timeout، یک درخواست ممکن است بسیار بیشتر از زمان قابل‌قبول کاربر باز بماند و منابع سرور را اشغال کند.

Timeout مناسب باید در چند سطح تعریف شود.

Connection Timeout

حداکثر زمان برای برقراری اتصال با API بالادستی.

Read Timeout

حداکثر زمان انتظار برای دریافت داده پس از برقراری اتصال.

Write Timeout

حداکثر زمان ارسال بدنه درخواست، مخصوصاً برای ورودی‌های بزرگ یا فایل.

Pool Timeout

حداکثر زمان انتظار برای دریافت اتصال آزاد از Connection Pool.

End-to-End Deadline

حداکثر زمان کل عملیات از دید کاربر، شامل:

• بررسی دسترسی
• بازیابی RAG
• فراخوانی مدل
• Retry
• اعتبارسنجی
• ثبت مصرف

تنظیم Timeout سمت Client یکی از اصول مهم پایداری است؛ پس از Timeout، برنامه می‌تواند تصمیم بگیرد Retry کند، Circuit را باز کند یا پاسخ کنترل‌شده برگرداند. راهنمای AWS برای Client Timeout

نمونه تنظیم Timeout در Python

import os
import httpx
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DARVAREH_API_KEY"],
    base_url="https://api.darvareh.ir/v1",
    timeout=httpx.Timeout(
        timeout=45.0,
        connect=5.0,
        read=40.0,
        write=10.0,
        pool=5.0,
    ),
)

نمونه تنظیم Timeout در JavaScript

const controller = new AbortController();

const timeoutId = setTimeout(() => {
  controller.abort();
}, 45_000);

try {
  const response = await fetch(
    "https://api.darvareh.ir/v1/chat/completions",
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.DARVAREH_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify(requestBody),
      signal: controller.signal,
    },
  );

  // Process response
} finally {
  clearTimeout(timeoutId);
}

Timeout را چقدر تنظیم کنیم؟

یک مقدار ثابت برای همه درخواست‌ها مناسب نیست.

این اعداد فقط نقطه شروع‌اند. مقدار واقعی باید براساس داده‌های Latency، مدل و تجربه کاربری تعیین شود.

Retry چیست؟

Retry یعنی درخواست شکست‌خورده را دوباره اجرا کنیم. این کار فقط زمانی مفید است که خطا موقت باشد.

خطاهای مناسب Retry

• قطع موقت شبکه
• Connection Timeout
• 408 Request Timeout
• 429 Rate Limit با ظرفیت قابل‌بازیابی
• 500 Internal Server Error
• 502 Bad Gateway
• 503 Service Unavailable
• 504 Gateway Timeout

خطاهای نامناسب Retry

• 400 Bad Request
• API Key نامعتبر
• دسترسی غیرمجاز
• مدل ناموجود
• Context بیش‌ازحد بزرگ
• Schema نامعتبر
• موجودی ناکافی
• محتوای ردشده
• خطایی که با همان ورودی تکرار می‌شود

Retry بدون طبقه‌بندی خطا، فقط هزینه و فشار سیستم را افزایش می‌دهد.

Exponential Backoff و Jitter

ارسال فوری چند Retry پشت‌سرهم می‌تواند وضعیت را بدتر کند. اگر هزار Client هم‌زمان خطا دریافت کنند و همگی دقیقاً یک ثانیه بعد Retry کنند، موج جدیدی از بار ایجاد می‌شود.

Exponential Backoff فاصله Retryها را به‌تدریج افزایش می‌دهد:

Retry 1: حدود ۱ ثانیه
Retry 2: حدود ۲ ثانیه
Retry 3: حدود ۴ ثانیه
Retry 4: حدود ۸ ثانیه

Jitter مقدار تصادفی کوچکی به تأخیر اضافه می‌کند تا Clientها هم‌زمان Retry نکنند:

import random


def backoff_delay(
    attempt: int,
    base: float = 1.0,
    maximum: float = 30.0,
) -> float:
    exponential = min(base * (2 ** attempt), maximum)
    return random.uniform(0, exponential)

AWS نیز Retry با Backoff را برای خطاهای موقت مانند Throttling، اختلال شبکه و در‌دسترس‌نبودن موقت سرویس پیشنهاد می‌کند و بر اهمیت Idempotency و محدودبودن Retry تأکید دارد. الگوی Retry با Backoff در AWS

Retry Budget

Retry باید محدودیت داشته باشد:

• حداکثر تعداد تلاش
• حداکثر زمان کل
• حداکثر هزینه
• حداکثر Token
• Deadline کاربر

برای مثال:

حداکثر تلاش: ۳
Deadline کل: ۴۵ ثانیه
حداکثر تأخیر هر Retry: ۸ ثانیه
حداکثر یک Fallback

اگر هر لایه سیستم سه بار Retry کند، تعداد درخواست‌ها می‌تواند به‌صورت تصاعدی افزایش یابد. بهتر است مالک Retry مشخص باشد و Retry در چند لایه روی هم قرار نگیرد.

Retry داخلی SDK

نسخه‌های SDK ممکن است برخی خطاهای موقت را خودکار Retry کنند. برای مثال، SDK رسمی Python در OpenAI به‌صورت پیش‌فرض بعضی خطاهای اتصال، 408، 409، 429 و خطاهای 5xx را Retry می‌کند. مرجع SDK پایتون OpenAI

اگر روی SDK یک لایه Retry دیگر می‌سازید، تعداد Retryهای واقعی را محاسبه کنید تا ناخواسته درخواست را چندین بار اجرا نکنید.

Rate Limit، RPM و TPM

Rate Limit برای جلوگیری از مصرف غیرمنصفانه، فشار بیش‌ازحد و هزینه کنترل‌نشده استفاده می‌شود.

مستندات OpenAI نیز Rate Limit را محدودیت تعداد دفعات دسترسی کاربر یا Client به سرویس در یک بازه زمانی تعریف می‌کنند. راهنمای Rate Limit

RPM

Requests Per Minute یا تعداد درخواست در دقیقه.

اگر RPM برابر ۶۰ باشد، نمی‌توان فرض کرد ارسال ۶۰ درخواست هم‌زمان در ابتدای دقیقه همیشه مجاز یا مناسب است. بهتر است درخواست‌ها در بازه زمانی توزیع شوند.

TPM

Tokens Per Minute یا مجموع Token ورودی و خروجی مجاز در دقیقه.

دو درخواست بزرگ ممکن است TPM را پر کنند، حتی اگر RPM پایین باشد.

RPD و TPD

برخی سرویس‌ها محدودیت روزانه نیز دارند:

• Requests Per Day
• Tokens Per Day

Concurrent Requests

تعداد درخواست‌هایی که هم‌زمان در حال اجرا هستند.

این محدودیت برای مدل‌های کند، Streaming و عملیات طولانی بسیار مهم است.

Rate Limit در چند سطح

یک سرویس مناسب باید محدودیت را در چند سطح اعمال کند:

• به‌ازای API Key
• به‌ازای کاربر
• به‌ازای سازمان
• به‌ازای IP
• به‌ازای مدل
• به‌ازای Endpoint
• به‌ازای پلن
• محدودیت سراسری سیستم

الگوریتم Token Bucket

در Token Bucket، ظرفی با ظرفیت مشخص داریم که با نرخ ثابت پر می‌شود. هر درخواست تعدادی Token مصرف می‌کند.

این روش Burst محدود را می‌پذیرد، اما مصرف طولانی‌مدت را کنترل می‌کند.

Pseudo-code:

def allow_request(bucket, cost):
    refill(bucket)

    if bucket.tokens < cost:
        return False

    bucket.tokens -= cost
    return True

در API هوش مصنوعی می‌توان Cost را براساس موارد زیر تعیین کرد:

• یک واحد برای هر درخواست
• Token تخمینی ورودی
• max_tokens
• قیمت مدل
• وزن بیشتر برای مدل‌های گران

پاسخ مناسب Rate Limit

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 8

{
  "error": {
    "code": "rate_limit_exceeded",
    "message": "تعداد درخواست‌های مجاز در این بازه تکمیل شده است.",
    "retry_after_seconds": 8
  }
}

Client باید در صورت وجود، Retry-After و Headerهای Rate Limit را رعایت کند.

کنترل درخواست‌های هم‌زمان

ممکن است کاربر از نظر RPM هنوز ظرفیت داشته باشد، اما ۵۰ درخواست را هم‌زمان ارسال کند. این وضعیت می‌تواند:

• Connection Pool را پر کند؛
• Latency را افزایش دهد؛
• TPM را ناگهان مصرف کند؛
• ارائه‌دهنده را Throttle کند؛
• حافظه سرور را اشغال کند.

Semaphore در Python

import asyncio

semaphore = asyncio.Semaphore(5)


async def limited_request(callable_request):
    async with semaphore:
        return await callable_request()

محدودیت هم‌زمانی به‌ازای کاربر

بهتر است Semaphore فقط سراسری نباشد. کاربر یا سازمان نیز باید سقف جداگانه داشته باشد:

کاربر عادی: ۲ درخواست هم‌زمان
پلن حرفه‌ای: ۱۰ درخواست هم‌زمان
پردازش Batch: ورود به Queue

Backpressure

اگر ظرفیت سیستم پر است، درخواست جدید باید:

• رد شود؛
• وارد Queue شود؛
• به مدل سبک‌تر هدایت شود؛
• یا پاسخ Degraded دریافت کند.

پذیرفتن نامحدود درخواست و امیدواربودن به اینکه بعداً پردازش شوند، معمولاً باعث شکست گسترده‌تر می‌شود.

Fallback میان مدل‌ها و ارائه‌دهندگان

Fallback یعنی اگر مسیر اصلی در دسترس نبود، درخواست به مسیر جایگزین هدایت شود.

اما Fallback فقط تغییر نام مدل نیست. مدل جایگزین باید با نیاز درخواست سازگار باشد.

Capability Matching

پیش از Fallback بررسی کنید مدل جایگزین از قابلیت‌های لازم پشتیبانی می‌کند:

• Text Generation
• Vision
• Tool Calling
• Structured Outputs
• JSON Mode
• Streaming
• Context Window کافی
• زبان فارسی
• طول خروجی لازم
• سیاست‌های داده
• سطح کیفیت موردنیاز

اگر درخواست دارای تصویر است، Fallback به مدل Text-only شکست خواهد خورد. اگر برنامه به JSON Schema متکی است، مدل جایگزین باید آن را پشتیبانی کند یا مسیر Validation متفاوتی داشته باشد.

Fallback بر چه اساسی انجام شود؟

• خطای موقت ارائه‌دهنده
• Circuit باز
• Latency بیش‌ازحد
• محدودیت Rate
• مدل موقتاً غیرفعال
• Region ناموجود
• خطای ظرفیت

چه زمانی Fallback نکنیم؟

• ورودی نامعتبر است.
• کاربر اجازه مدل جایگزین را ندارد.
• مدل جایگزین گران‌تر از سقف بودجه است.
• سیاست داده متفاوت است.
• قابلیت موردنیاز را ندارد.
• پاسخ اصلی قبلاً به کاربر Stream شده است.
• عملیات ممکن است در مسیر اصلی اجرا شده باشد.

ماتریس Fallback

جلوگیری از Fallback Loop

هر درخواست باید لیست مسیرهای امتحان‌شده و تعداد تلاش‌ها را نگه دارد:

{
  "attempted_routes": [
    "provider_a/model_x",
    "provider_b/model_y"
  ],
  "max_provider_attempts": 2
}

Circuit Breaker

اگر ارائه‌دهنده‌ای دائماً خطا می‌دهد، ارسال مداوم درخواست به آن باعث افزایش Latency، هزینه و فشار بیشتر می‌شود.

Circuit Breaker سه وضعیت اصلی دارد.

Closed

درخواست‌ها عادی ارسال می‌شوند.

Open

نرخ خطا از آستانه عبور کرده و درخواست جدید به ارائه‌دهنده ارسال نمی‌شود.

Half-Open

پس از مدت مشخص، تعداد محدودی درخواست آزمایشی ارسال می‌شود. اگر موفق بودند Circuit بسته می‌شود؛ اگر شکست خوردند دوباره باز می‌شود.

نمونه تنظیمات

{
  "failure_threshold": 5,
  "window_seconds": 60,
  "open_duration_seconds": 30,
  "half_open_requests": 2
}

چه خطاهایی Circuit را باز کنند؟

• Timeoutهای متوالی
• خطاهای 5xx
• Connection Failure
• Latency خارج از SLO
• پاسخ‌های خراب گسترده

خطاهای 400 ناشی از درخواست کاربر نباید به‌عنوان خرابی ارائه‌دهنده شمرده شوند.

Circuit Breaker و Fallback

درخواست
→ بررسی Circuit مدل اصلی
→ اگر Closed: ارسال
→ اگر Open: مسیر جایگزین
→ اگر همه مسیرها Open: پاسخ Degraded یا Queue

Idempotency چیست؟

Idempotency یعنی اجرای چندباره یک عملیات با شناسه یکسان، اثر آن را بیش از یک بار اعمال نکند.

در تولید متن ساده، دو پاسخ ممکن است فقط هزینه اضافی ایجاد کنند. اما در Agentها و Tool Calling، Retry می‌تواند عواقب جدی داشته باشد:

• ثبت دو تیکت
• ارسال دو ایمیل
• کسر دوباره موجودی
• ایجاد دو سفارش
• اجرای دوباره پرداخت
• ذخیره رکورد تکراری

AWS نیز Idempotency را تضمینی می‌داند که یک عملیات تغییردهنده، حتی با Retry، بیش از یک بار منابع را تغییر ندهد. راهنمای Idempotency در AWS

Idempotency Key

Client برای عملیات یک کلید یکتا ارسال می‌کند:

Idempotency-Key: order-8421-generate-invoice-v1

سمت سرور:

1. کلید را جست‌وجو می‌کند.
2. اگر نتیجه قبلی موجود است، همان را برمی‌گرداند.
3. اگر درخواست در حال اجراست، وضعیت مناسب می‌دهد.
4. اگر جدید است، عملیات را اجرا و نتیجه را ذخیره می‌کند.

جدول پیشنهادی

CREATE TABLE idempotency_keys (
    key TEXT PRIMARY KEY,
    user_id UUID NOT NULL,
    request_hash TEXT NOT NULL,
    status TEXT NOT NULL,
    response_json JSONB,
    created_at TIMESTAMPTZ NOT NULL,
    expires_at TIMESTAMPTZ NOT NULL
);

Request Hash

اگر یک Idempotency Key با بدنه متفاوت دوباره استفاده شد، سرور باید خطا بدهد؛ نه اینکه نتیجه درخواست قبلی را برای ورودی جدید برگرداند.

Streaming در محیط Production

Streaming باعث می‌شود کاربر پیش از تکمیل کل پاسخ، بخش‌های اولیه را دریافت کند. در رابط‌های سازگار با OpenAI معمولاً از Server-Sent Events استفاده می‌شود. راهنمای Streaming API

مزایای Streaming

• کاهش زمان ادراک‌شده پاسخ
• تجربه بهتر Chat
• نمایش تدریجی پاسخ‌های طولانی
• امکان توقف زودهنگام از سمت کاربر

چالش‌های Streaming

• قطع اتصال Client
• Timeout در Proxy
• Bufferشدن پاسخ توسط CDN
• مشکل ثبت مصرف نهایی
• دریافت پاسخ ناقص
• سخت‌شدن Fallback پس از ارسال اولین Chunk
• دشواری Validation کامل
• Moderation خروجی در حال تولید
• Tool Callهای چندبخشی
• JSON ناقص در Chunkهای میانی

نکته مهم درباره Fallback

پیش از ارسال اولین Chunk می‌توان مسیر را تغییر داد. اما بعد از ارسال بخشی از پاسخ، Fallback به مدل دیگر ممکن است لحن، ساختار و محتوای پاسخ را مخلوط کند.

پس از شروع Streaming معمولاً بهتر است:

• اتصال را با خطای کنترل‌شده پایان دهید؛
• وضعیت پاسخ را incomplete ثبت کنید؛
• امکان «تلاش مجدد» فراهم کنید؛
• از ترکیب دو مدل در یک پاسخ خودداری کنید.

Heartbeat

برای عملیات طولانی می‌توان Comment یا رویداد Heartbeat ارسال کرد تا Proxy اتصال را Idle تشخیص ندهد:

: heartbeat

Buffering

Headerهای مناسب بسته به زیرساخت:

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
X-Accel-Buffering: no

ذخیره پاسخ

دو راه متداول:

• Chunkها را هنگام Streaming ذخیره کنید.
• پاسخ نهایی را پس از پایان Stream ثبت کنید.

اگر Audit اهمیت دارد، وضعیت‌های زیر را نیز ذخیره کنید:

started
streaming
completed
client_disconnected
provider_error
timed_out
incomplete

Queue و پردازش غیرهم‌زمان

همه عملیات AI نباید در یک درخواست HTTP هم‌زمان انجام شوند.

Queue برای موارد زیر مناسب است:

• تولید تصویر
• تولید ویدئو
• پردازش چند فایل
• استخراج اطلاعات هزاران سند
• Batch Evaluation
• Embedding مجموعه بزرگ
• گزارش‌های طولانی
• Workflow چندمرحله‌ای
• عملیاتی که چند دقیقه طول می‌کشند

معماری Async Job

Client
→ POST /jobs
→ Validation
→ Create Job
→ Queue
→ Worker
→ AI Provider
→ Save Result
→ Webhook or Polling

پاسخ اولیه:

HTTP/1.1 202 Accepted

{
  "job_id": "job_2048",
  "status": "queued",
  "status_url": "/jobs/job_2048"
}

وضعیت‌های Job

queued
running
retrying
completed
failed
cancelled
expired

Queue نیز باید محدود باشد

صف نامحدود فقط خرابی را پنهان می‌کند. باید موارد زیر کنترل شوند:

• حداکثر طول Queue
• اولویت Job
• Deadline
• تعداد Retry
• Dead Letter Queue
• محدودیت کاربر
• لغو Job
• هزینه تخمینی
• Worker Concurrency

اعتبارسنجی ورودی

ورودی باید قبل از فراخوانی مدل بررسی شود.

موارد مهم

• نوع پیام‌ها
• طول متن
• تعداد پیام‌ها
• اندازه فایل
• نوع MIME
• تعداد تصاویر
• URLهای مجاز
• max_tokens
• مدل انتخابی
• پارامترهای Sampling
• Schema ابزارها
• مجوز استفاده از قابلیت
• وجود API Key
• موجودی و Quota

محدودیت صریح

MAX_MESSAGES = 50
MAX_MESSAGE_CHARS = 50_000
MAX_TOTAL_CHARS = 200_000


def validate_messages(messages):
    if not messages:
        raise ValueError("messages cannot be empty")

    if len(messages) > MAX_MESSAGES:
        raise ValueError("too many messages")

    total = 0

    for message in messages:
        content = message.get("content", "")

        if not isinstance(content, str):
            raise ValueError("invalid message content")

        if len(content) > MAX_MESSAGE_CHARS:
            raise ValueError("message is too long")

        total += len(content)

    if total > MAX_TOTAL_CHARS:
        raise ValueError("conversation is too large")

محدودیت کاراکتر جای Token Counter دقیق را نمی‌گیرد، اما یک لایه محافظ اولیه است.

اعتبارسنجی خروجی مدل

HTTP 200 به معنای قابل‌استفاده‌بودن پاسخ نیست.

موارد زیر را بررسی کنید:

• وجود choices
• وجود محتوا
• finish_reason
• Refusal
• Tool Call معتبر
• JSON قابل Parse
• تطابق با Schema
• طول پاسخ
• قواعد کسب‌وکار
• محتوای خطرناک
• ارجاع به منابع در صورت نیاز

Structured Outputs

برای خروجی‌های ماشینی از JSON Schema استفاده کنید:

{
  "response_format": {
    "type": "json_schema",
    "json_schema": {
      "name": "ticket_classification",
      "strict": true,
      "schema": {
        "type": "object",
        "properties": {
          "category": {
            "type": "string",
            "enum": ["billing", "technical", "account"]
          },
          "priority": {
            "type": "string",
            "enum": ["low", "medium", "high"]
          }
        },
        "required": ["category", "priority"],
        "additionalProperties": false
      }
    }
  }
}

پشتیبانی از Structured Outputs به مدل و ارائه‌دهنده بستگی دارد. خروجی را همچنان با Pydantic، Zod یا Validator مشابه بررسی کنید.

قواعد کسب‌وکار

حتی خروجی Schema-valid ممکن است غیرمنطقی باشد:

if result["refund_amount"] > original_payment_amount:
    raise ValueError("invalid refund amount")

کنترل Token و هزینه

هزینه باید قبل، هنگام و بعد از درخواست کنترل شود.

پیش از درخواست

• Token ورودی را تخمین بزنید.
• max_tokens را محدود کنید.
• قیمت مدل را بررسی کنید.
• موجودی کاربر را کنترل کنید.
• سقف هزینه درخواست را اعمال کنید.
• Context غیرضروری را حذف کنید.

هنگام درخواست

در Streaming یا Jobهای طولانی، در صورت پشتیبانی می‌توان مصرف یا وضعیت را مانیتور کرد و امکان لغو داشت.

پس از درخواست

• Token واقعی را ثبت کنید.
• هزینه واقعی را محاسبه کنید.
• اختلاف تخمین و واقعیت را بررسی کنید.
• کیف پول یا Quota را به‌صورت Atomic به‌روزرسانی کنید.
• رویداد مصرف را ثبت کنید.

Budget Guard

def enforce_budget(
    estimated_input_tokens,
    max_output_tokens,
    input_price,
    output_price,
    max_cost,
):
    estimated_cost = (
        estimated_input_tokens * input_price
        + max_output_tokens * output_price
    )

    if estimated_cost > max_cost:
        raise ValueError("estimated request cost exceeds limit")

قیمت‌ها باید با واحد یکسان و براساس قیمت هر Token یا هر میلیون Token محاسبه شوند.

محدودیت max_tokens

دادن مقدار بسیار بالا به تمام درخواست‌ها باعث می‌شود:

• احتمال هزینه بیشتر شود؛
• Rate Limit زودتر پر شود؛
• پاسخ‌های بی‌جهت طولانی تولید شوند.

مقدار خروجی را براساس وظیفه تعیین کنید:

Classification: 50–200 Token
Extraction: متناسب با Schema
Chat کوتاه: 500–1500 Token
گزارش: براساس نیاز واقعی

Logging، Metrics و Tracing

اگر سیستم قابل مشاهده نباشد، تشخیص علت خطا دشوار خواهد بود.

OpenTelemetry سه سیگنال اصلی را مطرح می‌کند:

• Traces: مسیر درخواست در اجزای سیستم
• Metrics: اندازه‌گیری‌های زمان اجرا
• Logs: ثبت رویدادها

مفاهیم Observability در OpenTelemetry

Logging ساختاریافته

به‌جای متن آزاد:

API failed again!

از Log ساختاریافته استفاده کنید:

{
  "level": "error",
  "event": "provider_request_failed",
  "request_id": "req_2048",
  "trace_id": "trace_8421",
  "user_id": "user_52",
  "model": "MODEL_ID",
  "provider": "provider_a",
  "attempt": 2,
  "status_code": 503,
  "error_code": "provider_unavailable",
  "latency_ms": 12400,
  "input_tokens_estimated": 8200,
  "timestamp": "2026-07-11T12:00:00Z"
}

چه چیزهایی را Log نکنیم؟

• API Key
• Authorization Header
• رمز عبور
• اطلاعات بانکی
• Token دسترسی
• محتوای حساس کاربران
• کل Prompt بدون ضرورت
• فایل خصوصی
• اطلاعات شخصی غیرضروری

می‌توان برای Debug از Hash، شناسه، طول متن و نسخه Prompt استفاده کرد.

Metrics ضروری

ترافیک

• Request Rate
• Active Requests
• Queue Depth
• Requests per Model

Latency

• Time to First Token
• Total Latency
• Provider Latency
• RAG Latency
• Tool Latency
• p50، p95 و p99

میانگین به‌تنهایی کافی نیست؛ مشکلات کاربران معمولاً در صدک‌های بالاتر دیده می‌شوند.

خطا

• Error Rate
• Timeout Rate
• Rate Limit Rate
• Validation Failure Rate
• Refusal Rate
• Fallback Rate
• Circuit Breaker State

مصرف و هزینه

• Input Tokens
• Output Tokens
• Cached Tokens
• Cost per Request
• Cost per User
• Cost per Successful Task
• Budget Rejections

کیفیت

• User Feedback
• Groundedness
• Tool Success Rate
• JSON Validation Rate
• Task Completion Rate
• Hallucination Reports

Distributed Tracing

یک درخواست ممکن است از چند سرویس عبور کند:

API Gateway
→ Auth
→ Billing
→ RAG
→ Router
→ Provider
→ Validator
→ Usage Logger

Trace باید Spanهای جداگانه داشته باشد:

POST /chat/completions
  ├── auth.check
  ├── wallet.precheck
  ├── rag.retrieve
  ├── router.select
  ├── provider.request
  ├── output.validate
  └── usage.record

با trace_id مشترک می‌توان فهمید Latency یا خطا دقیقاً در کدام مرحله رخ داده است.

Request ID

برای هر درخواست یک شناسه یکتا تولید کنید:

X-Request-ID: req_2048

اگر ارائه‌دهنده نیز Request ID برمی‌گرداند، هر دو را ثبت کنید:

{
  "internal_request_id": "req_2048",
  "provider_request_id": "provider_req_7752"
}

این کار برای پشتیبانی و بررسی اختلاف مصرف بسیار مهم است.

Alerting

برای هر خطا Alert نفرستید؛ در سیستم پرترافیک باعث Alert Fatigue می‌شود.

Alert مناسب باید براساس آستانه و بازه باشد:

• نرخ خطای 5xx بیش از ۵٪ در پنج دقیقه
• p95 Latency بیشتر از SLO
• افزایش ناگهانی هزینه
• Queue Depth بالاتر از ظرفیت
• Circuit باز برای مدل اصلی
• نرخ Validation Failure غیرعادی
• اختلاف مصرف ثبت‌شده با Billing
• افزایش خطاهای موجودی یا Rate Limit

SLI، SLO و SLA

SLI

معیاری که اندازه‌گیری می‌شود:

• Availability
• Latency
• Task Success Rate
• Streaming Completion Rate

SLO

هدف داخلی:

۹۹٫۹٪ درخواست‌های معتبر باید بدون خطای زیرساختی پاسخ بگیرند.
۹۵٪ پاسخ‌های Chat باید در کمتر از ۳۰ ثانیه تکمیل شوند.

SLA

تعهد قراردادی به مشتری که ممکن است پیامد مالی یا حقوقی داشته باشد.

در سرویس AI بهتر است علاوه بر Availability، موفقیت واقعی وظیفه نیز ارزیابی شود. ممکن است API همیشه 200 برگرداند، اما خروجی قابل‌استفاده نباشد.

امنیت API Key

کلید API باید فقط روی سرور نگهداری شود. قراردادن آن در Browser، اپ موبایل یا مخزن عمومی خطرناک است.

مستندات API OpenAI نیز تأکید می‌کند کلید نباید در کد Client-side قرار گیرد و باید از متغیر محیطی یا سرویس مدیریت Secret بارگذاری شود. مرجع API OpenAI

اصول نگهداری کلید

• استفاده از Environment Variable یا Secret Manager
• عدم ثبت در Git
• عدم نمایش در Log
• Rotation دوره‌ای
• کلید جدا برای هر محیط
• دسترسی حداقلی
• امکان Revoke فوری
• ثبت آخرین زمان استفاده
• هشدار برای الگوی مصرف غیرعادی

محیط‌های جداگانه

development
staging
production

هر محیط باید API Key، بودجه و محدودیت مستقل داشته باشد.

کلید در Frontend

نامناسب:

const client = new OpenAI({
  apiKey: "sk-secret",
  dangerouslyAllowBrowser: true,
});

مناسب:

Browser
→ Backend شما
→ API درواره

Backend باید کاربر را احراز هویت و محدودیت‌های مصرف را اعمال کند.

امنیت Prompt و Tool Calling

Prompt Injection زمانی خطرناک‌تر می‌شود که مدل به ابزارها دسترسی داشته باشد.

مدل مرجع مجوز نیست

حتی اگر مدل بگوید کاربر اجازه دارد، Backend باید دسترسی را مستقل بررسی کند:

def execute_delete_tool(user, resource_id):
    if not user.can_delete(resource_id):
        raise PermissionError("forbidden")

    return delete_resource(resource_id)

ابزارهای کم‌خطر و پرخطر

ابزارهای Read-only را از ابزارهای تغییردهنده جدا کنید.

Read:
get_order
search_documents
get_wallet_balance

Write:
cancel_order
create_refund
send_email
delete_document

ابزارهای Write باید کنترل بیشتری داشته باشند:

• تأیید کاربر
• Idempotency Key
• Audit Log
• محدودیت مبلغ یا دامنه
• Validation
• امکان Rollback در صورت امکان

خروجی Tool نیز غیرقابل‌اعتماد است

خروجی API خارجی، وب‌سایت یا فایل ممکن است شامل متن مخرب باشد. آن را به‌عنوان داده وارد Context کنید، نه دستور.

اجرای کد

اگر Agent قابلیت اجرای کد دارد:

• Sandbox
• محدودیت CPU و حافظه
• Timeout
• شبکه محدود
• فایل‌سیستم محدود
• عدم دسترسی به Secret
• محدودیت Process
• حذف محیط پس از اجرا

ضروری است.

Caching

Cache می‌تواند هزینه و Latency را کاهش دهد، اما برای همه درخواست‌ها مناسب نیست.

موارد مناسب Cache

• Embedding متن ثابت
• پاسخ‌های عمومی با ورودی یکسان
• فهرست مدل‌ها
• Metadata
• اسناد RAG
• نتیجه Query کم‌تغییر
• System Prompt یا Context ثابت در مدل‌های پشتیبانی‌شده

موارد نامناسب Cache

• پاسخ شخصی و حساس بدون Scope
• موجودی لحظه‌ای
• وضعیت پرداخت
• داده با تغییر سریع
• پاسخ وابسته به مجوز
• خروجی دارای اطلاعات خصوصی

Cache Key

Cache Key باید عوامل مؤثر را در نظر بگیرد:

hash(
  model
  + prompt_version
  + normalized_input
  + user_scope
  + response_format
  + temperature
)

اگر نسخه Prompt یا مدل تغییر کرد، Cache قدیمی نباید اشتباه استفاده شود.

جلوگیری از نشت اطلاعات

Cache باید حداقل براساس Tenant یا User Scope تفکیک شود. پاسخ سازمان A نباید برای سازمان B بازگردانده شود.

Graceful Degradation

وقتی قابلیت اصلی در دسترس نیست، سیستم می‌تواند نسخه محدودتری از سرویس را ارائه دهد:

• استفاده از مدل جایگزین ارزان‌تر
• غیرفعال‌کردن RAG موقت و اعلام محدودیت
• نمایش پاسخ Cache‌شده با زمان به‌روزرسانی
• انتقال Job به Queue
• غیرفعال‌کردن قابلیت غیرضروری
• بازگرداندن پاسخ کوتاه‌تر
• درخواست تلاش مجدد از کاربر

Graceful Degradation نباید پنهانی کیفیت یا سیاست داده را تغییر دهد. اگر پاسخ محدود شده، بهتر است به کاربر اطلاع داده شود.

پیاده‌سازی مقاوم با Python و API درواره

نصب کتابخانه‌ها

pip install -U openai httpx pydantic

Client و مدل خروجی

import os
import time
import random
import uuid
from typing import Literal

import httpx
from openai import (
    OpenAI,
    APIConnectionError,
    APIStatusError,
    APITimeoutError,
    RateLimitError,
)
from pydantic import BaseModel, ConfigDict, ValidationError


class TicketResult(BaseModel):
    model_config = ConfigDict(extra="forbid")

    category: Literal["billing", "technical", "account", "other"]
    priority: Literal["low", "medium", "high"]
    summary: str


client = OpenAI(
    api_key=os.environ["DARVAREH_API_KEY"],
    base_url="https://api.darvareh.ir/v1",
    timeout=httpx.Timeout(
        timeout=40.0,
        connect=5.0,
        read=35.0,
        write=10.0,
        pool=5.0,
    ),
    max_retries=0,
)

در این مثال Retry داخلی SDK غیرفعال شده تا Retry در یک محل مدیریت شود.

تشخیص خطای قابل Retry

RETRYABLE_STATUS_CODES = {
    408,
    429,
    500,
    502,
    503,
    504,
}


def is_retryable_status(status_code: int) -> bool:
    return status_code in RETRYABLE_STATUS_CODES

اجرای درخواست با Backoff

def calculate_delay(
    attempt: int,
    retry_after: float | None = None,
) -> float:
    if retry_after is not None:
        return min(retry_after, 30.0)

    maximum = min(2 ** attempt, 10)
    return random.uniform(0, maximum)


def classify_ticket(
    text: str,
    model: str,
    max_attempts: int = 3,
) -> TicketResult:
    request_id = f"req_{uuid.uuid4().hex}"

    for attempt in range(max_attempts):
        started_at = time.monotonic()

        try:
            response = client.chat.completions.create(
                model=model,
                messages=[
                    {
                        "role": "system",
                        "content": (
                            "درخواست پشتیبانی را دقیق دسته‌بندی کن. "
                            "اطلاعاتی که در متن نیست حدس نزن."
                        ),
                    },
                    {
                        "role": "user",
                        "content": text,
                    },
                ],
                temperature=0,
                max_tokens=300,
                response_format={
                    "type": "json_schema",
                    "json_schema": {
                        "name": "ticket_result",
                        "strict": True,
                        "schema": TicketResult.model_json_schema(),
                    },
                },
            )

            content = response.choices[0].message.content

            if not content:
                raise ValueError("empty model response")

            result = TicketResult.model_validate_json(content)

            latency_ms = int(
                (time.monotonic() - started_at) * 1000
            )

            print(
                {
                    "event": "ai_request_succeeded",
                    "request_id": request_id,
                    "model": model,
                    "attempt": attempt + 1,
                    "latency_ms": latency_ms,
                }
            )

            return result

        except (APITimeoutError, APIConnectionError) as error:
            retryable = True
            error_name = type(error).__name__

        except RateLimitError as error:
            retryable = True
            error_name = "RateLimitError"

        except APIStatusError as error:
            retryable = is_retryable_status(
                error.status_code
            )
            error_name = f"HTTP_{error.status_code}"

        except ValidationError as error:
            retryable = False
            error_name = "ValidationError"

        except ValueError as error:
            retryable = False
            error_name = "InvalidResponse"

        print(
            {
                "event": "ai_request_failed",
                "request_id": request_id,
                "model": model,
                "attempt": attempt + 1,
                "error": error_name,
                "retryable": retryable,
            }
        )

        if not retryable or attempt == max_attempts - 1:
            raise

        time.sleep(calculate_delay(attempt + 1))

    raise RuntimeError("unreachable")

محدودیت این نمونه

این کد برای آموزش است. در Production باید موارد زیر نیز اضافه شوند:

• Parsing دقیق Retry-After
• Deadline کل
• Circuit Breaker اشتراکی
• Metrics
• Trace
• Sanitization Log
• Fallback سازگار
• Billing Guard
• Rate Limiter توزیع‌شده
• Idempotency
• تست قابلیت json_schema برای مدل انتخابی

Fallback در Python

from dataclasses import dataclass


@dataclass
class ModelRoute:
    model: str
    supports_json_schema: bool
    max_context_tokens: int
    max_cost_per_request: float


def select_routes(
    routes: list[ModelRoute],
    requires_json_schema: bool,
    estimated_context_tokens: int,
    max_cost: float,
) -> list[ModelRoute]:
    compatible = []

    for route in routes:
        if (
            requires_json_schema
            and not route.supports_json_schema
        ):
            continue

        if (
            estimated_context_tokens
            > route.max_context_tokens
        ):
            continue

        if route.max_cost_per_request > max_cost:
            continue

        compatible.append(route)

    return compatible

Fallback باید از فهرست مدل‌های سازگار انجام شود، نه از هر مدل در دسترس.

پیاده‌سازی با TypeScript و API درواره

نصب SDK و Zod

npm install openai zod

Client و Schema

import OpenAI from "openai";
import { z } from "zod";

const client = new OpenAI({
  apiKey: process.env.DARVAREH_API_KEY,
  baseURL: "https://api.darvareh.ir/v1",
  timeout: 40_000,
  maxRetries: 0,
});

const TicketSchema = z
  .object({
    category: z.enum([
      "billing",
      "technical",
      "account",
      "other",
    ]),
    priority: z.enum(["low", "medium", "high"]),
    summary: z.string(),
  })
  .strict();

type TicketResult = z.infer<typeof TicketSchema>;

Retry با Backoff

const RETRYABLE_STATUS_CODES = new Set([
  408,
  429,
  500,
  502,
  503,
  504,
]);

function delay(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function backoffDelay(attempt: number): number {
  const maximum = Math.min(2 ** attempt * 1000, 10_000);
  return Math.floor(Math.random() * maximum);
}

function isRetryable(error: unknown): boolean {
  if (error instanceof OpenAI.APIConnectionError) {
    return true;
  }

  if (error instanceof OpenAI.APIError) {
    return RETRYABLE_STATUS_CODES.has(error.status ?? 0);
  }

  return false;
}

اجرای مقاوم

async function classifyTicket(
  text: string,
  model: string,
  maxAttempts = 3,
): Promise<TicketResult> {
  const requestId = crypto.randomUUID();

  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    const startedAt = performance.now();

    try {
      const response = await client.chat.completions.create({
        model,
        messages: [
          {
            role: "system",
            content:
              "درخواست پشتیبانی را دقیق دسته‌بندی کن و حدس نزن.",
          },
          {
            role: "user",
            content: text,
          },
        ],
        temperature: 0,
        max_tokens: 300,
        response_format: {
          type: "json_schema",
          json_schema: {
            name: "ticket_result",
            strict: true,
            schema: z.toJSONSchema(TicketSchema),
          },
        },
      });

      const content = response.choices[0]?.message?.content;

      if (!content) {
        throw new Error("Empty model response");
      }

      const untrustedData: unknown = JSON.parse(content);
      const result = TicketSchema.parse(untrustedData);

      console.log({
        event: "ai_request_succeeded",
        requestId,
        model,
        attempt,
        latencyMs: Math.round(
          performance.now() - startedAt,
        ),
      });

      return result;
    } catch (error) {
      const retryable = isRetryable(error);

      console.error({
        event: "ai_request_failed",
        requestId,
        model,
        attempt,
        retryable,
        errorType:
          error instanceof Error
            ? error.constructor.name
            : "UnknownError",
      });

      if (!retryable || attempt === maxAttempts) {
        throw error;
      }

      await delay(backoffDelay(attempt));
    }
  }

  throw new Error("Request failed");
}

طراحی Error Response برای API خودتان

خطای داخلی ارائه‌دهنده را مستقیماً به کاربر نمایش ندهید. یک قرارداد خطای ثابت بسازید:

{
  "error": {
    "code": "upstream_temporarily_unavailable",
    "message": "سرویس هوش مصنوعی موقتاً در دسترس نیست.",
    "request_id": "req_2048",
    "retryable": true,
    "retry_after_seconds": 10
  }
}

کدهای داخلی پیشنهادی

invalid_request
authentication_failed
permission_denied
model_not_found
model_capability_mismatch
context_too_large
rate_limit_exceeded
insufficient_balance
budget_exceeded
provider_timeout
provider_unavailable
output_validation_failed
content_refused
stream_interrupted
internal_error

این کدها به Client اجازه می‌دهند بدون وابستگی به متن فارسی یا انگلیسی، رفتار مناسب انتخاب کند.

تست پایداری

تست Happy Path کافی نیست.

تست Timeout

پاسخ ارائه‌دهنده را عمداً کند کنید و بررسی کنید:

• درخواست در زمان تعیین‌شده متوقف می‌شود.
• منابع آزاد می‌شوند.
• خطای مناسب برمی‌گردد.
• Retry از Deadline عبور نمی‌کند.

تست Rate Limit

Burst ایجاد کنید و بررسی کنید:

• Rate Limiter درست کار می‌کند.
• 429 و Retry-After برمی‌گردند.
• یک کاربر ظرفیت دیگران را مصرف نمی‌کند.
• درخواست ردشده به ارائه‌دهنده ارسال نمی‌شود.

تست Provider Failure

خطاهای زیر را شبیه‌سازی کنید:

500
502
503
504
Connection Reset
Malformed JSON
Empty Response

تست Fallback

بررسی کنید:

• فقط برای خطاهای مجاز اجرا می‌شود.
• مدل جایگزین قابلیت لازم دارد.
• حلقه بی‌نهایت ایجاد نمی‌شود.
• مصرف دو مسیر ثبت می‌شود.
• هزینه از سقف عبور نمی‌کند.

تست Idempotency

یک درخواست تغییردهنده را چند بار با یک کلید ارسال کنید و مطمئن شوید اثر فقط یک بار اعمال می‌شود.

تست قطع Streaming

Client را هنگام دریافت پاسخ قطع کنید و بررسی کنید:

• پردازش قابل لغو است یا وضعیت آن ثبت می‌شود.
• Connection نشت نمی‌کند.
• مصرف گم نمی‌شود.
• پاسخ ناقص به‌عنوان کامل ثبت نمی‌شود.

تست خروجی نامعتبر

مدل یا Mock Provider پاسخ‌های زیر را برگرداند:

Not JSON

{}

{
  "priority": "super-urgent"
}

Validator باید همه را رد کند.

Load Test

فقط Requests per Second را اندازه نگیرید. برای AI باید موارد زیر نیز بررسی شوند:

• Token per Second
• Concurrent Streams
• Time to First Token
• Queue Depth
• Connection Pool
• Memory
• Provider Throttling
• Cost during Test

Chaos Testing

در محیط کنترل‌شده:

• ارائه‌دهنده را قطع کنید.
• Latency اضافه کنید.
• بخشی از پاسخ‌ها را خراب کنید.
• Redis یا Queue را از دسترس خارج کنید.
• Circuit Breaker را فعال کنید.
• Workerها را متوقف کنید.

هدف این است که مطمئن شویم شکست یک جزء به شکست کامل سیستم منجر نمی‌شود.

استقرار مرحله‌ای

تغییر مدل، Prompt یا Router می‌تواند رفتار Production را تغییر دهد.

Shadow Traffic

درخواست واقعی به مدل جدید نیز ارسال می‌شود، اما پاسخ آن به کاربر نمایش داده نمی‌شود. نتایج برای ارزیابی مقایسه می‌شوند.

این روش هزینه اضافی دارد و باید ملاحظات داده و رضایت را رعایت کند.

Canary Release

درصد کمی از ترافیک به نسخه جدید ارسال می‌شود:

نسخه فعلی: ۹۵٪
نسخه جدید: ۵٪

اگر Metrics مناسب بودند، سهم نسخه جدید افزایش می‌یابد.

Rollback

امکان بازگشت سریع برای موارد زیر داشته باشید:

• Prompt Version
• Model Route
• Provider
• Context Builder
• Schema
• Safety Policy
• SDK Version

چک‌لیست Production

پایداری

• برای تمام درخواست‌های خارجی Timeout تعریف شده است.
• Deadline کل درخواست مشخص است.
• Retry فقط برای خطاهای موقت انجام می‌شود.
• Retry دارای Exponential Backoff و Jitter است.
• تعداد Retry محدود است.
• Retry داخلی SDK در محاسبات لحاظ شده است.
• Circuit Breaker برای مسیرهای بالادستی وجود دارد.
• Fallback براساس Capability انجام می‌شود.
• Fallback Loop غیرممکن است.
• برای عملیات تغییردهنده Idempotency وجود دارد.
• درخواست‌های طولانی وارد Queue می‌شوند.
• Dead Letter Queue تعریف شده است.
• محدودیت Concurrency اعمال می‌شود.
• Backpressure وجود دارد.

ورودی و خروجی

• ورودی با Schema اعتبارسنجی می‌شود.
• اندازه پیام و فایل محدود است.
• Token ورودی تخمین زده می‌شود.
• Context Window مدل بررسی می‌شود.
• max_tokens محدود است.
• خروجی مدل Parse و Validate می‌شود.
• finish_reason بررسی می‌شود.
• Refusal و پاسخ ناقص مدیریت می‌شوند.
• قواعد کسب‌وکار مستقل از مدل اجرا می‌شوند.

هزینه و Billing

• موجودی پیش از درخواست بررسی می‌شود.
• سقف هزینه هر درخواست وجود دارد.
• مصرف واقعی پس از درخواست ثبت می‌شود.
• به‌روزرسانی موجودی Atomic است.
• درخواست Retryشده دوبار صورتحساب داخلی نمی‌شود.
• مصرف Fallback ثبت می‌شود.
• هشدار افزایش غیرعادی هزینه وجود دارد.
• محدودیت کاربر و سازمان جداست.

امنیت

• API Key در Frontend قرار ندارد.
• Secretها در Log ثبت نمی‌شوند.
• کلیدها قابل Rotation و Revoke هستند.
• محیط‌های توسعه و Production کلید جدا دارند.
• ابزارها حداقل دسترسی را دارند.
• عملیات حساس تأیید مستقل دارند.
• Prompt Injection در نظر گرفته شده است.
• خروجی Tool غیرقابل‌اعتماد فرض می‌شود.
• اجرای کد در Sandbox انجام می‌شود.
• اطلاعات حساس قبل از ورود به مدل محدود می‌شوند.
• Cache میان Tenantها جداست.

Observability

• برای هر درخواست Request ID وجود دارد.
• Provider Request ID ثبت می‌شود.
• Logها ساختاریافته‌اند.
• Prompt Version و Model Route ثبت می‌شوند.
• Metrics ترافیک، خطا، Latency و هزینه وجود دارند.
• p50، p95 و p99 اندازه‌گیری می‌شوند.
• Time to First Token ثبت می‌شود.
• Tracing مسیر کامل درخواست را پوشش می‌دهد.
• Alertها براساس آستانه‌اند.
• Dashboard سلامت مدل‌ها و ارائه‌دهندگان وجود دارد.
• Logها اطلاعات حساس ندارند.

انتشار و کیفیت

• Dataset ارزیابی واقعی وجود دارد.
• مدل و Prompt جدید قبل از انتشار Eval می‌شوند.
• Canary یا Shadow Testing انجام می‌شود.
• امکان Rollback سریع وجود دارد.
• پاسخ‌های Degraded برای کاربر شفاف‌اند.
• Runbook مدیریت Incident نوشته شده است.
• مسئول هر Alert مشخص است.
• Postmortem برای خطاهای مهم انجام می‌شود.

پرسش‌های متداول

برای API هوش مصنوعی چه Timeoutی مناسب است؟

یک عدد واحد برای همه کاربردها وجود ندارد. دسته‌بندی کوتاه ممکن است به ۵ تا ۱۵ ثانیه نیاز داشته باشد، درحالی‌که Reasoning یا تولید محتوای طولانی زمان بیشتری می‌برد. Timeout باید براساس داده واقعی Latency و Deadline تجربه کاربر تعیین شود.

کدام خطاها باید Retry شوند؟

خطاهای موقت شبکه، Timeout، 429 قابل‌بازیابی و برخی خطاهای 5xx مناسب Retry هستند. خطاهای ورودی، احراز هویت، موجودی ناکافی و Schema نامعتبر نباید بدون تغییر درخواست Retry شوند.

چند بار Retry کنیم؟

معمولاً دو یا سه تلاش محدود نقطه شروع مناسبی است، اما تعداد واقعی باید با Deadline، هزینه و نوع عملیات تعیین شود. Retry نامحدود خطرناک است.

تفاوت Rate Limit و Concurrency Limit چیست؟

Rate Limit مصرف را در یک بازه زمانی محدود می‌کند؛ مثلاً ۶۰ درخواست در دقیقه. Concurrency Limit تعداد درخواست‌های در حال اجرای هم‌زمان را محدود می‌کند؛ مثلاً حداکثر پنج درخواست هم‌زمان.

Fallback چه تفاوتی با Retry دارد؟

Retry همان مسیر را دوباره امتحان می‌کند. Fallback درخواست را به مدل یا ارائه‌دهنده دیگری می‌فرستد. مسیر جایگزین باید قابلیت‌های موردنیاز درخواست را داشته باشد.

Circuit Breaker چیست؟

Circuit Breaker پس از مشاهده خطاهای متوالی، ارسال درخواست به مسیر خراب را موقتاً متوقف می‌کند. پس از یک دوره انتظار، تعداد کمی درخواست آزمایشی برای بررسی بازیابی سرویس ارسال می‌شوند.

Idempotency چرا برای Agentها مهم است؟

Agent ممکن است ابزارهایی مانند ارسال پیام، ثبت سفارش یا تغییر داده را اجرا کند. اگر درخواست Retry شود، Idempotency از اجرای چندباره یک عملیات جلوگیری می‌کند.

آیا Streaming هزینه را کاهش می‌دهد؟

Streaming عمدتاً زمان ادراک‌شده پاسخ را کاهش می‌دهد، نه الزاماً هزینه را. هزینه همچنان براساس مصرف مدل محاسبه می‌شود، مگر اینکه تولید زودتر لغو شود و ارائه‌دهنده آن را در مصرف لحاظ کند.

آیا بعد از شروع Streaming می‌توان Fallback کرد؟

از نظر فنی ممکن است، اما معمولاً مناسب نیست؛ زیرا بخشی از پاسخ مدل اول به کاربر ارسال شده است. ترکیب ادامه پاسخ از مدل دیگر می‌تواند ناسازگار باشد. بهتر است Stream ناقص پایان یابد و امکان تلاش مجدد فراهم شود.

آیا HTTP 200 به معنای موفقیت درخواست هوش مصنوعی است؟

خیر. پاسخ ممکن است خالی، ناقص، Refusal، خارج از Schema یا از نظر معنایی نادرست باشد. موفقیت واقعی باید پس از Validation و بررسی نتیجه تعیین شود.

چگونه هزینه API را کنترل کنیم؟

با تخمین Token، محدودکردن Context و max_tokens، انتخاب مدل مناسب، سقف هزینه هر درخواست، محدودیت مصرف کاربر، Cache و ثبت دقیق Usage می‌توان هزینه را کنترل کرد.

آیا API Key در اپلیکیشن موبایل یا Frontend امن است؟

خیر. Secret قابل استخراج است. درخواست باید ابتدا به Backend شما ارسال شود و Backend پس از احراز هویت و کنترل محدودیت‌ها، API درواره را فراخوانی کند.

آیا درواره از Fallback و Retry همه مدل‌ها پشتیبانی می‌کند؟

رفتار مدل‌ها و ارائه‌دهندگان یکسان نیست. در سطح اپلیکیشن باید Retry، Timeout و Fallback را براساس قابلیت مدل انتخابی طراحی کنید. Fallback نباید مدل فاقد قابلیت موردنیاز را انتخاب کند.

جمع‌بندی

ساخت یک API هوش مصنوعی قابل‌اعتماد بسیار فراتر از ارسال یک Prompt و دریافت پاسخ است. مدل و ارائه‌دهنده فقط بخشی از یک سیستم توزیع‌شده‌اند و هر بخش ممکن است کند، محدود یا موقتاً غیرقابل‌دسترس شود.

یک معماری Production باید:

• برای هر عملیات Deadline و Timeout داشته باشد؛
• خطاهای موقت و دائمی را از هم جدا کند؛
• Retry محدود با Backoff و Jitter انجام دهد؛
• RPM، TPM و Concurrency را کنترل کند؛
• Fallback را براساس قابلیت مدل انجام دهد؛
• با Circuit Breaker از مسیر خراب محافظت کند؛
• عملیات تغییردهنده را Idempotent سازد؛
• وظایف طولانی را به Queue منتقل کند؛
• ورودی و خروجی را اعتبارسنجی کند؛
• Token و هزینه را قبل و بعد از درخواست کنترل کند؛
• Logs، Metrics و Traces قابل‌استفاده داشته باشد؛
• Secretها و داده کاربران را محافظت کند؛
• و تغییرات مدل و Prompt را مرحله‌ای منتشر کند.

هدف یک سیستم قابل‌اعتماد این نیست که هرگز خطا نکند. هدف این است که خطا قابل‌پیش‌بینی، محدود، قابل‌مشاهده و قابل‌بازیابی باشد.

ساخت سرویس‌های هوش مصنوعی با API درواره

درواره زیرساخت دسترسی به مدل‌های هوش مصنوعی را از طریق یک API سازگار با OpenAI فراهم می‌کند. توسعه‌دهندگان می‌توانند با یک اتصال، مدل‌های مختلف را در نرم‌افزارها، Workflowها، سامانه‌های RAG و Agentهای هوش مصنوعی به کار بگیرند.

آدرس پایه API درواره:

https://api.darvareh.ir/v1

برای استفاده در محیط عملیاتی، مدل را براساس قابلیت، Context Window، هزینه، سرعت و نوع ورودی انتخاب کنید و لایه‌های Timeout، Retry، Validation، کنترل هزینه و Monitoring را در معماری اپلیکیشن خود قرار دهید.

مقالات مرتبط

• Context Engineering چیست؟ آموزش مهندسی زمینه برای مدل‌ها و Agentهای هوش مصنوعی
• Structured Outputs چیست؟ آموزش دریافت خروجی JSON از مدل‌های هوش مصنوعی
• Fallback چیست؟ افزایش پایداری سرویس‌های هوش مصنوعی
• Prompt Caching چیست؟ راهنمای کاهش هزینه و زمان پاسخ API
• Token چیست و چگونه هزینه API هوش مصنوعی محاسبه می‌شود؟
• Tool Calling و Function Calling چیست؟
• API Gateway هوش مصنوعی چیست و چگونه کار می‌کند؟
• چگونه هزینه API هوش مصنوعی را کاهش دهیم؟
• OpenAI-Compatible API چیست؟
• RAG چیست؟ آموزش ساخت سیستم بازیابی و تولید تقویت‌شده