Spaces:
Sleeping
Sleeping
File size: 2,444 Bytes
4028152 4f78a11 a3e5847 4028152 a3e5847 fcec14a 4f78a11 fcec14a 4f78a11 a3e5847 fcec14a 4f78a11 a3e5847 fcec14a 4f78a11 a3e5847 fcec14a 4f78a11 a3e5847 fcec14a a3e5847 fcec14a a3e5847 4f78a11 a3e5847 4f78a11 a3e5847 fcec14a 4f78a11 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
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
|