بناء نظام RAG من الصفر

import anthropic
import chromadb
from sentence_transformers import SentenceTransformer
from pathlib import Path

print("=" * 60)
print("RAG System: نظام Q&A على المستندات")
print("=" * 60)

# ─────────────────────────────────────────
# الإعداد
# ─────────────────────────────────────────
claude_client = anthropic.Anthropic()
embed_model   = SentenceTransformer("all-MiniLM-L6-v2")
chroma_client = chromadb.Client()

# ─────────────────────────────────────────
# 1. Indexing — بناء قاعدة المعرفة
# ─────────────────────────────────────────
def chunk_text(text: str, chunk_size: int = 300, overlap: int = 50) -> list[str]:
    """تقسيم النص إلى chunks متداخلة"""
    chunks, start = [], 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunk = text[start:end].strip()
        if len(chunk) > 50:
            chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

# مستندات تجريبية (في الواقع: PDF, Word, Web pages)
documents = {
    "ai_basics.txt": """
الذكاء الاصطناعي هو مجال في علوم الحاسوب يهدف إلى محاكاة الذكاء البشري.
يشمل تعلم الآلة والتعلم العميق ومعالجة اللغة الطبيعية.
Claude هو نموذج لغوي من Anthropic يتميز بالأمان والدقة.
GPT-4 من OpenAI هو أحد أقوى النماذج في السوق حالياً.
يمكن للذكاء الاصطناعي الحديث كتابة الكود وتحليل البيانات وترجمة النصوص.
    """,
    "programming.txt": """
Python هي لغة برمجة عالية المستوى تتميز بسهولة القراءة.
تُستخدم Python على نطاق واسع في تعلم الآلة والذكاء الاصطناعي.
مكتبة NumPy تتيح العمليات الرياضية على المصفوفات بكفاءة عالية.
Pandas توفر أدوات قوية لتحليل البيانات الجدولية.
Scikit-learn هي المكتبة الأكثر استخداماً لتعلم الآلة الكلاسيكي.
    """,
    "rag_info.txt": """
RAG اختصار Retrieval-Augmented Generation أي التوليد المُعزَّز بالاسترداد.
تقنية RAG تحلّ مشكلة Hallucination في نماذج اللغة الكبيرة.
تعمل RAG بمرحلتين: الأولى Indexing والثانية Retrieval مع Generation.
في مرحلة Indexing يتم تحويل المستندات إلى embeddings وتخزينها في vector database.
في مرحلة الاستخدام يتم تحويل السؤال لـ embedding والبحث عن أقرب المستندات.
    """
}

# بناء الـ collection
collection = chroma_client.create_collection("knowledge_base")
all_chunks, all_ids, all_metas = [], [], []

for doc_name, doc_text in documents.items():
    chunks = chunk_text(doc_text)
    for i, chunk in enumerate(chunks):
        all_chunks.append(chunk)
        all_ids.append(f"{doc_name}_{i}")
        all_metas.append({"source": doc_name, "chunk_idx": i})

# توليد embeddings وإضافة للـ collection
embeddings = embed_model.encode(all_chunks).tolist()
collection.add(
    ids=all_ids,
    documents=all_chunks,
    embeddings=embeddings,
    metadatas=all_metas
)

print(f"✅ تم إنشاء قاعدة المعرفة: {len(all_chunks)} chunk من {len(documents)} مستند")

# ─────────────────────────────────────────
# 2. RAG Pipeline — الاستعلام والإجابة
# ─────────────────────────────────────────
def rag_query(question: str, top_k: int = 3, verbose: bool = True) -> str:
    """نظام RAG كامل: بحث + توليد"""

    # البحث عن أقرب chunks
    q_emb = embed_model.encode([question]).tolist()
    results = collection.query(
        query_embeddings=q_emb,
        n_results=top_k
    )

    # تجميع السياق
    chunks = results["documents"][0]
    sources = [m["source"] for m in results["metadatas"][0]]
    context = "\n\n---\n\n".join(chunks)

    if verbose:
        print(f"\n{'='*50}")
        print(f"السؤال: {question}")
        print(f"\nالـ Chunks المُسترجَعة ({top_k}):")
        for chunk, src in zip(chunks, sources):
            print(f"  [{src}] {chunk[:80].strip()}...")

    # بناء الـ Prompt
    prompt = f"""أجب على السؤال التالي بناءً على السياق المقدَّم فقط.
إذا لم يكن السياق يحتوي على المعلومات الكافية، قل ذلك بوضوح.

السياق:
{context}

السؤال: {question}

أجب بإيجاز ودقة باللغة العربية."""

    # استدعاء Claude
    message = claude_client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}]
    )

    answer = message.content[0].text

    if verbose:
        print(f"\nالإجابة:")
        print(answer)

    return answer

# ─────────────────────────────────────────
# 3. تجربة الـ RAG
# ─────────────────────────────────────────
questions = [
    "ما هو Claude وما مميزاته؟",
    "ما الفرق بين NumPy وPandas؟",
    "كيف تعمل تقنية RAG؟",
    "ما هي عاصمة فرنسا؟",  # سؤال خارج نطاق المعرفة
]

for q in questions:
    rag_query(q, top_k=3, verbose=True)