الوحدة 9 — التخزين المؤقّت والتحكّم في الكلفة

الوحدات السابقة ركّزت على الجودة. هذه الوحدة تُطرح السؤال الآخر الذي يقرّر بقاء المشروع أو موته في الإنتاج: كم يكلّف هذا كلّه؟، وكيف نجعله يبقى محتملًا مع نموّ الاستفسارات.

تفكيك كلفة سؤال واحد

سؤال في نظام RAG يستهلك في العموم:

1. تضمين السؤال: نحو 20 – 50 رمزًا في المتوسّط، ~0.000001 دولار عبر نموذج مغلق.
2. بحث الفهرس: عدّة مللي‌ثوانٍ على CPU، كلفة الاستضافة الثابتة تُوزَّع.
3. إعادة الترتيب: 20 مقطعًا × ~500 رمز = 10 آلاف رمز عبر النموذج (أو GPU محلّي).
4. توليد الإجابة: تعليمة نظام (200 رمزًا) + 5 مقاطع (2500 رمزًا) + سؤال (30 رمزًا) + جواب (300 رمزًا) ≈ 3000 رمز إدخال + 300 رمز إخراج.
5. تسجيل ومراقبة: ثابت وقليل.

جدول تقريبي بأسعار مغلقة شائعة في 2026 (تتغيّر باستمرار):

البند	الرموز	السعر (دولار / مليون رمز)	الكلفة (سنت)
تضمين السؤال	40	0.02	0.00
إعادة الترتيب (خدمة)	10000	2.00	2.00
إدخال التوليد	3000	5.00	1.50
إخراج التوليد	300	15.00	0.45
المجموع			~4 سنتات

على 1000 سؤال يوميًّا: 40 دولارًا في اليوم، 1200 دولار في الشهر. رقم حقيقي جدًّا، ويتضاعف إن لم تُخفَّض. التخزين المؤقّت والتحسينات الآتية يقدرون على تقليصه إلى الثلث دون خسارة في الجودة.

تخزين تضمينات السؤال

التضمين عملية حتمية: نفس النصّ يعطي نفس الشعاع دائمًا. ومن ثمّ الأسئلة المتكرّرة (الأسئلة الشائعة) لا داعي لإعادة تضمينها.

import hashlib
import redis
import json
import numpy as np

r = redis.Redis(host="localhost", port=6379)

def cached_embed(text, model):
    key = f"emb:{hashlib.sha256(text.encode()).hexdigest()}"
    cached = r.get(key)
    if cached:
        return np.array(json.loads(cached))
    vec = model.encode(text, normalize_embeddings=True)
    r.setex(key, 86400 * 30, json.dumps(vec.tolist()))  # 30 يومًا
    return vec

الوفر ملموس فقط إن كانت الأسئلة تتكرّر. عادةً على نظام مؤسّسي: 20 – 40 % من الاستفسارات اليومية إعادات لأسئلة سُئلت في الأسبوع الأخير.

تخزين الإجابات الكاملة

الأثر الأكبر يأتي من تخزين الإجابة نفسها. إن كان السؤال «مدّة إشعار الاستقالة؟» يصل عشر مرّات في اليوم، لا داعٍ لعشر سلاسل RAG كاملة.

def answer_or_cache(question, ttl=3600):
    key = f"ans:{hashlib.sha256(question.strip().lower().encode()).hexdigest()}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    result = full_rag_pipeline(question)  # الاسترجاع + التوليد + التحقّق
    r.setex(key, ttl, json.dumps(result))
    return result

نقطة حاسمة: زمن الحياة (ttl). إجابة عن إجراء إداري قد تبقى صالحة أسبوعًا، لكنّها تصير خطأً بعد تعديل الإجراء. الأمن: إبطال الذاكرة عند تحديث المستندات المرتبطة.

def invalidate_cache_for_source(source_path):
    """يمسح كلّ الإجابات التي استُشهد فيها بهذا المصدر."""
    for key in r.scan_iter("ans:*"):
        data = json.loads(r.get(key))
        if any(c["source"] == source_path for c in data["citations"]):
            r.delete(key)

التخزين الدلاليّ

الاستراتيجيتان السابقتان مبنيّتان على تطابق حرفي. لكن «مدّة إشعار الاستقالة؟» و«كم أدوم قبل ترك العمل؟» و«ما فترة الإخطار بالاستقالة؟» أسئلة متكافئة يجب أن تشترك في الجواب المخزَّن.

التخزين الدلاليّ (semantic caching) يخزّن الأسئلة بأشعّتها. عند سؤال جديد، إن وجد سؤال محفوظ بمسافة أصغر من عتبة (مثلًا 0.95 في جيب التمام)، يُعاد نفس الجواب.

def semantic_cache_lookup(question, threshold=0.95):
    q_vec = cached_embed(question, embedding_model)
    hits = cache_store.query(query_embeddings=[q_vec.tolist()], n_results=1)
    if hits["distances"][0] and (1 - hits["distances"][0][0]) > threshold:
        return json.loads(hits["metadatas"][0][0]["answer"])
    return None

اختيار العتبة حسّاس. عتبة منخفضة جدًّا (0.85) تُعيد إجابات لأسئلة قريبة لكن مختلفة الفعل، فتخسر الدقّة. عتبة مرتفعة جدًّا (0.99) لا تكاد تُطلَق. قِسْها على مجموعة الأسئلة الموصوفة، فلا قيمة نظرية.

تخزين التعليمة على مستوى النموذج

الابتكار المهمّ حديثًا: تخزين البادئة الثابتة للتعليمة على مستوى النموذج نفسه (prompt caching لدى Anthropic وOpenAI). تعليمة نظام من 200 رمز تُرسَل ألف مرّة يوميًّا لا تُحاسَب في كلّ استدعاء إلّا مرّة واحدة، والباقي بسعر أقلّ بعشرة أضعاف.

# مثال Anthropic
message = client.messages.create(
    model="claude-3-5-sonnet-latest",
    system=[
        {
            "type": "text",
            "text": LONG_SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},
        },
    ],
    messages=[{"role": "user", "content": user_message}],
)

هذا التوفير كبير جدًّا في تكاليف الإدخال، وينبغي تفعيله بمجرّد أن تكون تعليمة النظام ثابتة عبر الاستفسارات.

إعادة الفهرسة التدريجية

الفهرسة الأولى مكلفة (300 مستند، عدّة ملايين رمز)، لكنّها تُنجَز مرّة. الأخطر: الفهرسة الدورية حين تُعدَّل المستندات. الحماقة الشائعة: إعادة كلّ شيء أسبوعيًّا.

الحلّ: بصمة المحتوى. لكلّ مستند نحسب hash (SHA-256 مثلًا) لمحتواه، ونحفظه.

import hashlib
from pathlib import Path

def content_hash(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def sync_index(source_paths):
    for path in source_paths:
        new_hash = content_hash(path)
        existing = get_doc_metadata(path)
        if existing and existing["hash"] == new_hash:
            continue  # لم يتغيّر، تخطَّ
        # المستند جديد أو مُعدَّل
        if existing:
            delete_chunks_for_source(path)
        chunks = extract_and_chunk(path)
        embed_and_store(chunks, path, new_hash)

على 300 مستند إن تغيّرت 3 يوميًّا، تكلفة الفهرسة اليومية 1 % من كلفة الفهرسة الكاملة. توفير ذو مقدار.

اختيار النموذج حسب السؤال

كلّ الأسئلة لا تستحقّ النموذج نفسه. سؤال مباشر بسيط (استخراج تاريخ من مقطع) نموذج صغير رخيص يحلّه. سؤال يستلزم استنتاجًا معقّدًا يستحقّ النموذج الرئيسي.

التوجيه (routing) قرار مبكر يُلحق كلّ سؤال بمستوى النموذج المناسب:

CLASSIFIER_PROMPT = """صنّف السؤال في فئة واحدة:
- بسيط: استخراج مباشر من نصّ (تاريخ، رقم، اسم)
- متوسّط: يستدعي فهمًا لجملتين إلى ثلاث
- معقّد: يستدعي تركيبًا عبر عدّة مقاطع

السؤال: {q}
أجب بكلمة واحدة."""

def route_and_answer(question):
    level = cheap_llm.invoke(CLASSIFIER_PROMPT.format(q=question)).content.strip()
    model = {"بسيط": cheap_llm, "متوسّط": mid_llm, "معقّد": premium_llm}[level]
    return rag_with_model(question, model)

في نظام مؤسّسي حيث 70 % من الأسئلة بسيطة، هذا التوجيه يخفض الكلفة إلى النصف تقريبًا.

المراقبة والتسجيل

بلا مراقبة، الاقتصاد أعمى. المطلوب على الأقلّ:

• عدد الاستفسارات يوميًّا.
• الكلفة الوسطى لكلّ سؤال، والتوزيع (المتوسّط والذيل).
• نسبة إصابات الذاكرة المؤقّتة على كلّ مستوى.
• الأخطاء والانتظارات المفرطة.

الأدوات: LangSmith (خاصّ بـLangChain، شامل)، Langfuse (مفتوح، ذاتيّ الاستضافة)، أو تسجيل يدويّ بسيط في PostgreSQL. الأخير كافٍ للبداية، والأدوات المتخصّصة تُفيد حين يتجاوز النظام آلاف الاستفسارات اليومية.

تخزين مؤقّت من غير إبطال قنبلة موقوتة

إجابة مخزَّنة عن إجراء أُلغِيَ منذ ثلاثة أشهر، تُعاد اليوم لموظّف يعمل بها. الضرر أكبر من عدم التخزين. كلّ عمليّة تحديث للفهرس يجب أن تُتبَع بإبطال منتقى للذاكرة المؤقّتة، إمّا بحسب المصدر (الأدقّ) أو بمسح كامل (الأضمن).

نقطة انطلاق واقعية للتحسين

لا تُطبِّق كلّ التحسينات دفعة واحدة. الترتيب المُقترَح: (1) prompt caching على تعليمة النظام — تنفيذ بسيط ومكسب كبير. (2) تخزين إجابات الأسئلة الشائعة بـTTL قصير. (3) بصمة المستندات لإعادة فهرسة تدريجية. (4) توجيه الأسئلة على مستويات نماذج مختلفة. القياس بعد كلّ خطوة يمنعك من تعقيد شيء لم يعطِ فائدة.

في الخلاصة

• سؤال في نظام RAG يتفكّك إلى بنود قابلة للتفكيك (تضمين، إعادة ترتيب، إدخال، إخراج)؛ ولا يمكن تحسين ما لا يقاس.
• التخزين المؤقّت على ثلاث طبقات: تضمين السؤال، الإجابة الكاملة (حرفيًّا ثم دلاليًّا)، والبادئة الثابتة عبر prompt caching.
• إعادة الفهرسة التدريجية ببصمات المحتوى تخفض كلفة الصيانة اليومية بأكثر من 90 %.
• التوجيه يوزّع الأسئلة على نماذج بمستويات مختلفة؛ 70 % من الأسئلة تستحقّ النموذج الرخيص فقط.

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