rag_pipeline.py

import time
import numpy as np
from transformers import AutoTokenizer, AutoModelForCausalLM

class RAGPipeline:
    def __init__(self, logger):
        self.logger = logger
        self.logger("[RAG] جاري تحميل النموذج والمحول...")
        self.tokenizer = AutoTokenizer.from_pretrained("aubmindlab/aragpt2-mega", trust_remote_code=True)
        self.generator = AutoModelForCausalLM.from_pretrained("aubmindlab/aragpt2-mega", trust_remote_code=True)
        self.chunk_embeddings = []
        self.index = []
        self.logger("[RAG] تم التحميل بنجاح.")

    def build_index(self, chunks):
        start = time.time()
        self.logger(f"[RAG] بناء الفهرس لـ {len(chunks)} مقاطع...")
        self.chunk_embeddings = []
        self.index = []

        for i, chunk in enumerate(chunks):
            embedding = self._dummy_embedding(chunk)
            self.chunk_embeddings.append(embedding)
            self.index.append(chunk)
            if (i+1) % 10 == 0 or (i+1) == len(chunks):
                self.logger(f"[RAG] تم معالجة {i+1}/{len(chunks)} مقاطع.")

        self.chunk_embeddings = np.array(self.chunk_embeddings)
        dim = self.chunk_embeddings.shape[1] if len(self.chunk_embeddings) > 0 else 0
        elapsed = time.time() - start
        self.logger(f"[RAG] تم بناء الفهرس بأبعاد {dim} في {elapsed:.2f} ثانية.")
        return "تم بناء الفهرس بنجاح."

    def _dummy_embedding(self, text):
        return np.random.rand(768)

    def generate_answer(self, question, top_k=3):
        start = time.time()
        self.logger(f"[RAG] توليد إجابة للسؤال: {question}")

        if len(self.index) == 0:
            self.logger("[RAG] تحذير: الفهرس فارغ، الرجاء بناء الفهرس أولاً.")
            return "لم يتم بناء الفهرس بعد.", []

        passages = self.index[:top_k]
        prompt = question + "\n\nالمراجع:\n" + "\n".join(passages)

        inputs = self.tokenizer(prompt, return_tensors="pt")
        output = self.generator.generate(inputs.input_ids, max_new_tokens=150, do_sample=True)
        response = self.tokenizer.decode(output[0], skip_special_tokens=True)

        elapsed = time.time() - start
        self.logger(f"[RAG] تم توليد الإجابة في {elapsed:.2f} ثانية.")
        return response, passages