بناء نظام RAG (استرجاع المعلومات المعزّز) بكلود AI — دليل عملي شامل

pip install anthropic langchain pypdf python-docx beautifulsoup4 requests

from langchain.document_loaders import PyPDFLoader, Docx2txtLoader
from langchain.document_loaders import WebBaseLoader
import os

def load_documents(source_path: str) -> list:
    """تحميل مستندات من مجلد يحتوي على PDF و Word وروابط ويب"""
    documents = []

    for filename in os.listdir(source_path):
        filepath = os.path.join(source_path, filename)

        if filename.endswith('.pdf'):
            loader = PyPDFLoader(filepath)
            documents.extend(loader.load())

        elif filename.endswith('.docx'):
            loader = Docx2txtLoader(filepath)
            documents.extend(loader.load())

    return documents

# تحميل مستندات من الويب
web_urls = [
    "https://yourcompany.com/policies",
    "https://yourcompany.com/procedures"
]
web_loader = WebBaseLoader(web_urls)
web_docs = web_loader.load()
print(f"تم تحميل {len(web_docs)} صفحة ويب")

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # حجم كل مقطع بالحروف
    chunk_overlap=50,      # تداخل 50 حرف بين المقاطع
    separators=["\n\n", "\n", ".", "!", "؟", " "]  # ترتيب الفصل
)

chunks = splitter.split_documents(documents)
print(f"تم إنشاء {len(chunks)} مقطع")

import voyageai

client = voyageai.Client(api_key="your-voyage-api-key")

def create_embeddings(texts: list[str]) -> list[list[float]]:
    """إنشاء تضمينات لقائمة من النصوص"""
    result = client.embed(
        texts,
        model="voyage-3",          # أحدث نموذج لعام 2026
        input_type="document"      # للمستندات المُفهرسة
    )
    return result.embeddings

# للاستعلامات استخدم input_type="query"
def embed_query(query: str) -> list[float]:
    result = client.embed([query], model="voyage-3", input_type="query")
    return result.embeddings[0]

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid

# الاتصال بـ Qdrant (محلياً أو عبر Docker)
client = QdrantClient("localhost", port=6333)

# إنشاء مجموعة للتضمينات
client.create_collection(
    collection_name="company_knowledge",
    vectors_config=VectorParams(
        size=1024,          # حجم متجهات Voyage AI
        distance=Distance.COSINE
    )
)

def index_chunks(chunks: list, embeddings: list[list[float]]):
    """فهرسة المقاطع مع تضميناتها"""
    points = []
    for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
        points.append(PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={
                "text": chunk.page_content,
                "source": chunk.metadata.get("source", ""),
                "page": chunk.metadata.get("page", 0)
            }
        ))

    client.upsert(collection_name="company_knowledge", points=points)
    print(f"تم فهرسة {len(points)} مقطع")

import anthropic

def retrieve_and_answer(query: str, top_k: int = 5) -> str:
    """استرجاع المقاطع وتوليد إجابة بكلود"""

    # 1. تضمين السؤال
    query_embedding = embed_query(query)

    # 2. البحث في قاعدة البيانات
    results = qdrant_client.search(
        collection_name="company_knowledge",
        query_vector=query_embedding,
        limit=top_k,
        score_threshold=0.7  # تصفية المقاطع ضعيفة الصلة
    )

    # 3. بناء السياق
    context_parts = []
    for i, result in enumerate(results, 1):
        source = result.payload.get("source", "مجهول")
        text = result.payload.get("text", "")
        context_parts.append(f"[مصدر {i}: {source}]\n{text}")

    context = "\n\n---\n\n".join(context_parts)

    # 4. استدعاء كلود مع السياق
    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1500,
        system="""أنت مساعد معرفي دقيق. أجب على الأسئلة حصرياً بناءً على السياق المقدّم.
إذا لم يحتوِ السياق على إجابة كافية، قل ذلك صراحةً.
استشهد دائماً بالمصدر في إجابتك.""",
        messages=[{
            "role": "user",
            "content": f"السياق المتاح:\n\n{context}\n\nالسؤال: {query}"
        }]
    )

    return message.content[0].text

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

# بيانات الاختبار (سؤال، إجابة مثالية، إجابة النظام، السياق المُسترجع)
test_dataset = {
    "question": ["ما هي سياسة الإجازات في شركتنا؟"],
    "ground_truth": ["الموظف يحق له 21 يوم إجازة سنوية..."],
    "answer": [retrieve_and_answer("ما هي سياسة الإجازات؟")],
    "contexts": [[ctx.page_content for ctx in retrieved_chunks]]
}

results = evaluate(
    dataset=test_dataset,
    metrics=[faithfulness, answer_relevancy, context_precision]
)
print(results)
# faithfulness: 0.92 — الإجابة مبنية على السياق
# answer_relevancy: 0.88 — الإجابة ذات صلة بالسؤال
# context_precision: 0.85 — المقاطع المُسترجعة دقيقة