LLM API Engineering Basics

LLM API غير حتمي.
   - قد يفشل (network, rate limit, timeout).
   - قد يبطئ (cold start, model size).
   - قد يعطي نتائج متفاوتة (temperature، model variant).

تطبيق LLM الإنتاجي يحتاج:
   - retry ذكي.
   - streaming للـ UX.
   - timeout صارم.
   - logging منظّم.
   - caching حيث ممكن.
   - secrets آمنة.
   - اختبارات لـ prompts.
   - مراقبة.

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
model = os.environ["OPENAI_MODEL"]


def stream_response(prompt: str) -> None:
    """يطبع النصّ تدريجيًا مع التعرّف على نوع الحدث."""
    with client.responses.stream(
        model=model,
        input=prompt,
    ) as stream:
        for event in stream:
            # Responses API يصدر أحداثًا مُسمّاة بأنواع.
            # للتدفّق النصّي: response.output_text.delta
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "response.completed":
                # اكتمل التوليد. يمكنك قراءة response النهائية لاحقًا.
                pass
            elif event.type == "response.error":
                # خطأ في التدفّق. سجّل وتعامل معه.
                print(f"\n[error] {event.error}", flush=True)
    print()

UX:  أول chunk يظهر خلال ~0.5-2 ثانية.
      مقارنة مع ~5-30 ثانية بدون streaming.

الإدراك: المستخدم يشعر بسرعة حتى لو كان الإجمالي نفسه.

✅ افحص event.type (قيمة نصّية) لا hasattr(event, "delta").
   حقول delta قد تظهر في أكثر من نوع حدث.

✅ عالج على الأقل:
   - response.output_text.delta  → chunk نصّي.
   - response.completed          → اكتمال.
   - response.error              → خطأ.

⚠️  قد توجد أنواع أحداث أخرى (response.created, response.in_progress, ...)
   لا تفترض أن delta موجود فقط في output_text.delta.

import os
import time
import random
from openai import OpenAI
from openai import (
    APIConnectionError,
    APITimeoutError,
    RateLimitError,
)

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
model = os.environ["OPENAI_MODEL"]


def ask_with_retry(prompt: str, max_attempts: int = 4) -> str | None:
    """استدعاء مع retry و backoff."""
    delay = 1.0   # ثانية

    for attempt in range(1, max_attempts + 1):
        try:
            response = client.responses.create(
                model=model,
                input=prompt,
                timeout=30.0,
            )
            return response.output_text

        except (APIConnectionError, APITimeoutError, RateLimitError) as e:
            if attempt == max_attempts:
                # استنفد المحاولات
                print(f"[retry] فشلت كل المحاولات: {e}")
                return None

            # exponential backoff + jitter
            jitter = random.uniform(0, 0.5)
            sleep_for = delay + jitter
            print(f"[retry] المحاولة {attempt} فشلت، إعادة بعد {sleep_for:.1f}s")
            time.sleep(sleep_for)
            delay *= 2     # 1 → 2 → 4 → 8

✅ Exponential backoff (1s, 2s, 4s, 8s...).
✅ Jitter (عشوائية صغيرة) لتفادي thundering herd.
✅ حدّ أقصى للمحاولات (3-5).
✅ retry فقط على أخطاء عابرة (RateLimit, Connection, Timeout).
❌ retry على BadRequest أو AuthenticationError (لن تنجح).

# على مستوى العميل (يؤثر على كل الطلبات)
client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    timeout=60.0,
)

# على مستوى الطلب الواحد (يتجاوز الـ default)
response = client.responses.create(
    model=model,
    input=prompt,
    timeout=30.0,
)

- طلب LLM قد يعلّق دقائق بدون timeout.
- بدون timeout، التطبيق كله قد يتجمّد.
- timeout صارم = التطبيق يخفف الفشل بدلًا من الانتظار.

import logging
import json

logger = logging.getLogger("llm_app")


def log_request(prompt: str, response: str, usage: dict, duration_ms: int) -> None:
    """سجّل كل طلب بشكل منظّم (بدون الأسرار أو مدخلات حسّاسة)."""
    logger.info(
        "llm_request",
        extra={
            "model": os.environ.get("OPENAI_MODEL"),
            "prompt_chars": len(prompt),          # الطول فقط، لا النصّ
            "response_chars": len(response),
            "input_tokens": usage.get("input_tokens"),
            "output_tokens": usage.get("output_tokens"),
            "duration_ms": duration_ms,
        },
    )

❌ المفتاح الكامل.
❌ الـ prompt إذا كان يحوي PII.
❌ الـ response إذا كان يحوي بيانات حسّاسة (بدون فلترة).
❌ الـ headers الكاملة.

✅ الطول (chars).
✅ usage tokens.
✅ duration.
✅ model name.
✅ status code.
✅ error type (بدون تفاصيل sensitive).

import hashlib
import json
from functools import lru_cache


def stable_hash(*args, **kwargs) -> str:
    """هاش ثابت لـ args."""
    payload = json.dumps([args, kwargs], sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(payload.encode()).hexdigest()


@lru_cache(maxsize=256)
def cached_ask(prompt: str, model: str) -> str | None:
    """تخزين مؤقت لنتائج متطابقة."""
    return ask_with_retry(prompt)

✅ استعلامات ثابتة ("ما هي عاصمة اليابان؟").
✅ قوالب prompts ذات مدخلات متكرّرة.
✅ prompts لا تعتمد على عوامل متغيّرة (نموذج، إعدادات، وقت).

❌ مدخلات مستخدم فريدة (chat messages).
❌ prompts حسّاسة للزمن (أسعار، أخبار).
❌ كل ما تكلّفته التخزين أعلى من تكلفة الطلب.

# ❌ خطأ: hardcoded
api_key = "sk-proj-xxxxxx"   # ستُسرق فورًا عند push.

# ✅ صحيح: من متغيّر بيئة
api_key = os.environ["OPENAI_API_KEY"]

# ✅ أفضل: من secret manager (AWS Secrets Manager، GCP Secret Manager، Vault)
import boto3
def get_openai_key():
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="openai/api-key")
    return response["SecretString"]

- لا تكتب key في الكود.
- لا تطبع key في logs.
- لا ترسل key في URL أو query string.
- لا تحفظ key في ملفات logs أو error tracking.
- استخدم secret manager في الإنتاج.
- دوّر الـ keys دوريًا.

"""
tests/test_prompts.py (مثال توضيحي)

اختبارات prompts تتحقّق من:
- صيغة المخرج (JSON صالح).
- حقول مطلوبة موجودة.
- عدم انتهاك قواعد (لا معلومات حسّاسة).
"""
import pytest


@pytest.fixture
def classifier():
    from my_app.prompts import classify_ticket  # استيراد وظيفتك
    return classify_ticket


def test_classifies_bug_correctly(classifier):
    result = classifier("التطبيق يتعطّل عند الفتح")
    assert result is not None
    assert result.category in {"bug", "billing", "account", "other"}


def test_handles_ambiguous(classifier):
    """حالة غامضة: يجب أن لا يفشل."""
    result = classifier("مرحبًا")
    assert result is not None


def test_no_hallucinated_fields(classifier):
    """تحقّق من عدم إضافة حقول زائدة."""
    result = classifier("...")
    allowed = {"category", "urgency", "needs_human", "reasoning"}
    assert set(result.model_dump().keys()) <= allowed

- شغّل prompt tests على كل تغيير.
- وثّق النتائج المرجعية (golden outputs).
- قِس drift: هل تغيّرت النتائج بعد تحديث النموذج؟
- ارفض merge إذا فشلت اختبارات prompts حسّاسة.

ما يجب مراقبته:

- Latency p50 / p95 / p99.
- Error rate (4xx, 5xx, timeout).
- Token usage (لكل طلب وتراكمي).
- التكلفة (تراكمية ولكل مستخدم).
- جودة المخرج (عينات يدوية دورية).
- توزيع temperature، top_p (كشف misuse).

import time

def monitored_ask(prompt: str) -> str | None:
    start = time.perf_counter()
    try:
        response = client.responses.create(
            model=model,
            input=prompt,
        )
        elapsed_ms = (time.perf_counter() - start) * 1000

        # سجّل metrics (مثلاً إلى Prometheus أو statsd)
        metrics.timing("llm.request.duration_ms", elapsed_ms)
        metrics.increment("llm.request.success")
        metrics.increment("llm.tokens.input", response.usage.input_tokens)
        metrics.increment("llm.tokens.output", response.usage.output_tokens)

        return response.output_text
    except Exception as e:
        metrics.increment("llm.request.error")
        raise

التكلفة التقريبية لكل طلب:
  cost ≈ input_tokens × input_rate + output_tokens × output_rate

  - input_rate, output_rate يحدّدهما مزوّد الـ API.
  - تختلف بين النماذج والـ tiers والـ regions.
  - تتغيّر مع الوقت.

- لا تنشر أسعارًا ثابتة في الكود.
- احسب من صفحة التسعير الفعلية للمزوّد وقت التحليل.
- أضف margin وتتبّع cost per user.
- في الإنتاج: budget alerts.

تطبيق LLM إنتاجي:

   User → Backend → [Auth] → [Rate Limit] → [Cache Check]
                                          ↓ (miss)
                                     [LLM API call]
                                          ↓
                                  [Retry + Backoff]
                                          ↓
                                  [Response + Tokens]
                                          ↓
                            [Log + Metrics] → [Return to User]

   Parallel:
   - Secrets Manager → API Key
   - Monitoring → Dashboards
   - CI/CD → Prompt Tests