الوحدة 10 — مشروع: مساعد مستندات كامل من الاستخراج إلى الواجهة

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

بنية المشروع

مجلَّد بسيط قابل للتوسّع:

assistant-procedures/
├── data/
│   ├── raw/               # المستندات الأصلية
│   └── processed/         # نصّ مستخرَج مع ما وراء البيانات
├── index/                 # قاعدة البيانات المتّجهية
├── src/
│   ├── ingest.py          # استخراج + تقطيع + تضمين + تخزين
│   ├── retriever.py       # هجين + إعادة ترتيب
│   ├── prompt.py          # بناء التعليمة
│   ├── pipeline.py        # الاستدعاء الكامل
│   ├── auth.py            # فلاتر الأذونات
│   └── cache.py           # التخزين المؤقّت
├── api/
│   └── main.py            # FastAPI
├── ui/
│   └── app.py             # Streamlit
├── eval/
│   ├── questions.yml      # مجموعة الأسئلة الموصوفة
│   └── run.py             # تشغيل التقييم
└── requirements.txt

فصل src/ عن api/ وui/ يُتيح إعادة استعمال الخطّ نفسه في نصوص برمجية دفعية (تحديث الفهرس ليلًا) أو من CLI.

سلسلة الاستيعاب الكاملة

# src/ingest.py
from pathlib import Path
import hashlib
from unstructured.partition.auto import partition
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import chromadb

EMBEDDING_MODEL = SentenceTransformer("intfloat/multilingual-e5-large")
CLIENT = chromadb.PersistentClient(path="./index")
COLLECTION = CLIENT.get_or_create_collection(
    name="procedures", metadata={"hnsw:space": "cosine"},
)

def ingest_document(path: Path, access_level: str = "public"):
    doc_hash = hashlib.sha256(path.read_bytes()).hexdigest()
    if is_up_to_date(path, doc_hash):
        return
    delete_chunks_for_source(str(path))

    elements = partition(filename=str(path))
    text = "\n\n".join(el.text for el in elements if el.text)

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=400, chunk_overlap=60,
        separators=["\n\n", "\n", "。", ".", "؟", "!", "،", " "],
    )
    chunks = splitter.split_text(text)

    embeddings = EMBEDDING_MODEL.encode(
        [f"passage: {c}" for c in chunks], normalize_embeddings=True,
    )

    COLLECTION.add(
        ids=[f"{path.stem}-{i:04d}" for i in range(len(chunks))],
        documents=chunks,
        embeddings=embeddings.tolist(),
        metadatas=[
            {
                "source": str(path),
                "access_level": access_level,
                "hash": doc_hash,
            } for _ in chunks
        ],
    )

def ingest_folder(folder: Path):
    for path in folder.rglob("*"):
        if path.suffix.lower() in [".pdf", ".docx", ".html", ".pptx"]:
            level = detect_access_level(path)  # حسب المسار مثلًا
            ingest_document(path, access_level=level)

سلسلة الاستفسار

# src/pipeline.py
from src.retriever import hybrid_retrieve, rerank
from src.prompt import build_prompt
from src.cache import get_cached_answer, cache_answer
from src.auth import user_access_filter

def answer(question: str, user):
    cached = get_cached_answer(question, user)
    if cached:
        return cached

    filters = user_access_filter(user)
    candidates = hybrid_retrieve(question, filters=filters, k=20)
    top = rerank(question, candidates, k=5)

    messages = build_prompt(question, top)
    response = LLM.invoke(messages)

    result = {
        "answer": response.content,
        "citations": [
            {"source": p.metadata["source"], "page": p.metadata.get("page")}
            for p in top
        ],
    }
    cache_answer(question, user, result, ttl=3600)
    return result

فرض الأذونات

النقطة الأكثر تجاهلًا في المشاريع الأولى، وأخطرها. مساعد يجيب موظّفًا عاديًّا بمعطيات ماليّة سرّية عطب أمني، لا مجرّد إزعاج.

# src/auth.py
ACCESS_HIERARCHY = {
    "public": ["public"],
    "employee": ["public", "employee"],
    "manager": ["public", "employee", "manager"],
    "hr": ["public", "employee", "hr"],
    "finance": ["public", "employee", "finance"],
    "admin": ["public", "employee", "manager", "hr", "finance", "admin"],
}

def user_access_filter(user):
    allowed = ACCESS_HIERARCHY.get(user.role, ["public"])
    return {"access_level": {"$in": allowed}}

مبدأ الأمان: افتراضيًّا كلّ شيء سرّيّ، وترفع الأذونات فقط عند الحاجة. عكس هذا («افتراضيًّا كلّ شيء عامّ» ثم تخفيض) يعطي بيانات مسرَّبة يوم تنسى وسم مستند جديد.

واجهة الاستخدام المصغَّرة

Streamlit كافية للنموذج الأوّلي، وتُتيح عرض الإجابة مع الاستشهادات القابلة للنقر.

# ui/app.py
import streamlit as st
from src.pipeline import answer

st.title("مساعد الإجراءات الداخلية")

user_role = st.sidebar.selectbox("الدور", ["employee", "manager", "hr"])
user = type("U", (), {"role": user_role})()

question = st.chat_input("اطرح سؤالك...")
if question:
    with st.spinner("أبحث في المستندات..."):
        result = answer(question, user)
    st.markdown(result["answer"])
    st.divider()
    st.caption("المصادر:")
    for c in result["citations"]:
        st.caption(f"— {c['source']} (ص {c.get('page', '؟')})")

خمس عشرة سطرًا تُتيح تجربة النظام حقيقةً، بيد مستخدمين حقيقيين، قبل بذل جهد على واجهة إنتاجية.

واجهة برمجية للاندماج

للنشر مع تطبيقات موجودة، FastAPI توفّر REST endpoints:

# api/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from src.pipeline import answer as rag_answer

app = FastAPI()

class Query(BaseModel):
    question: str
    user_role: str = "employee"

@app.post("/ask")
def ask(q: Query):
    if not q.question.strip():
        raise HTTPException(400, "سؤال فارغ")
    user = type("U", (), {"role": q.user_role})()
    return rag_answer(q.question, user)

@app.post("/reindex")
def reindex():
    from src.ingest import ingest_folder
    from pathlib import Path
    ingest_folder(Path("./data/raw"))
    return {"status": "ok"}

نقطة عملية: حماية /reindex بمفتاح أو بمصادقة قوية. تشغيلها من طرف مجهول يُشغّل أنابيب الاستيعاب على الخادم ويولّد كلفة عبثية.

تحليل الأعطال المتوقّعة

بعد تشغيل النظام في يد المستخدمين لأسبوعين، الأعطال المتكرّرة عادةً:

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

«يذكر أنّني «مدير مبيعات» وأنا ماليّ.» سبب: لا ذاكرة سياقية للمستخدم. علاج: تمرير معلومات المستخدم في التعليمة صراحةً («الموظّف من قسم المالية»)، أو إضافة فلتر على قسم عمله.

«يجيب عن الإجراء القديم مع أنّ هناك إجراء جديد.» سبب: عدم حذف الإجراء الملغى من الفهرس. علاج: علامة deprecated في ما وراء البيانات وفلتر تلقائيّ.

«يقول لا يوجد جواب مع أنّي أعرف الجواب موجود.» سبب: قد يكون فشل تقطيع أو تضمين. علاج: افحص يدويًّا هل المقطع الصحيح موجود في الفهرس ومسترجَع عند البحث، ثم اصعد للأعلى لتحديد أين انكسرت السلسلة.

«الوقت طويل جدًّا (7 ثوان).» سبب: تسلسل مراحل بدون تخزين مؤقّت. علاج: prompt caching، وتوجيه على النموذج الرخيص للأسئلة البسيطة.

دورة التطوير الطبيعية

بعد الإطلاق الأوّلي، الحلقة الأسبوعية تصير:

1. جمع الأسئلة الحديثة من السجلّ، مع الإجابات وتقييم المستخدمين إن وجد.
2. تحديد الأسئلة التي فشل النظام فيها (اعتراف بعدم المعرفة عند وجود الجواب، إجابة خاطئة، بطء).
3. تصنيف الأعطال حسب المرحلة (استخراج، تقطيع، استرجاع، توليد).
4. إضافة هذه الأسئلة إلى مجموعة الاختبار الموصوفة (الوحدة 8).
5. تعديل واحد في المرّة، مع قياس قبل/بعد.

بلا هذه الحلقة، المشروع يتصلّب في حالته الأولى وتتراكم فيه الأعطال.

إغراء الجمع بين كلّ الأدوات المُتاحة

كلّ من LangChain وLlamaIndex وHaystack وغيرها تعرض عشرات المكوّنات. الإغراء طبيعي: أضِفْ محلِّل استعلامات، وموجِّهًا، وذاكرة، ووكلاء. النتيجة نظام معقّد لا يفهمه أحد ولا يُصلَح عند فشله. ابدأ بأبسط سلسلة تعمل، وأضِفْ فقط عندما يُظهر التقييم حاجة.

الإطلاق التجريبي بمجموعة صغيرة أوّلًا

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

في الخلاصة

• المشروع كامل ينتظم في مجلّدات مستقلّة لكلّ مسؤولية: استيعاب، استرجاع، توليد، أذونات، تخزين مؤقّت، واجهة، تقييم.
• فرض الأذونات بمبدأ «افتراضيًّا كلّ شيء سرّيّ» يُنقذ من تسرّبات البيانات الأشيع.
• الواجهات بسيطة أوّلًا: Streamlit للنموذج الأوّلي، FastAPI للاندماج؛ التعقيد يُضاف على الطلب.
• الحلقة الأسبوعية لجمع الأعطال ومعالجتها هي التي تُنقذ المشروع من التصلّب؛ تعديل واحد في المرّة، مع قياس دائم.

تخطيط الإنتاج: من النموذج إلى الخدمة

المشروع الذي عرضناه صالح كنموذج أوّليّ لعشرات المستخدمين. للانتقال إلى مئات أو آلاف، تبرز حاجات جديدة. قاعدة البيانات المتّجهية: Chroma المضمَّنة في العملية تخدم النموذج الأوّلي، لكن مؤسّسة بألف مستخدم متزامن تحتاج خدمة منفصلة كـQdrant أو pgvector مع PostgreSQL موجود أصلًا. الانتقال بسيط تقنيًّا لأنّ واجهة الاسترجاع مغلَّفة في وحدة retriever.py.

النموذج المولّد: استدعاء مزوّد خارجيّ لكلّ سؤال يعطي بساطة، لكنّه يضع كلّ التوفّر بيد طرف ثالث. الحلّ في المؤسّسات الحسّاسة: نموذج مفتوح (llama-3-70b أو qwen-2.5) مستضاف على GPU داخليّ، بواجهة متوافقة مع OpenAI (vLLM أو Text Generation Inference). التطبيق نفسه لا يتغيّر: تعديل base_url واحد.

التوسّع الأفقيّ: عدّة نُسَخ من واجهة FastAPI خلف موازن أحمال، مع تخزين مؤقّت مركزيّ في Redis مشترك بين النُّسَخ. الفهرسة تبقى مركزية على عملية واحدة لتفادي التعارض في الكتابة.

الأمن التشغيليّ

نقاط لا يتحمّل الإنتاج تجاهلها:

• إخفاء المفاتيح: مفاتيح API لا تكتب في الكود ولا في مستودع git، بل في متغيّرات بيئة مُدارة بأداة أسرار (Vault، AWS Secrets Manager، أو حتّى ملفّ .env بأذونات مقيّدة على الخادم).
• تسجيل معطّل حسّاس: إن سجّل النظام كلّ الاستفسارات والإجابات، فسجلّه بحدّ ذاته صار مصدرًا حسّاسًا. تشفير السجلّات في السكون واجب، وحذف دوريّ (30 أو 90 يومًا حسب السياسة).
• حدود المعدّل: مفتاح مسروق قد يُنشئ فاتورة بآلاف الدولارات في ساعات. حدّ استفسارات دقيقيّ لكلّ مستخدم يمنع الأذى الأكبر.
• مراقبة الأنماط الشاذّة: مستخدم يستفسر مئة سؤال في دقيقة، أو أسئلة تحاول اختراق التعليمة (prompt injection)، تُرصَد وتُنبَّه.

الوحدة التالية: خلاصة الدورة كلّها في شجرة تشخيصية «الإجابة سيّئة: أين أبحث؟»، وإعلان الاختبار النهائي.