# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import faiss import numpy as np from sentence_transformers import SentenceTransformer from typing import List app = FastAPI(title="مدير ذاكرة المحادثة") # تحميل نموذج تحويل النصوص إلى متجهات (Embedding) باللغة العربية model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") # أبعاد المتجه (حسب النموذج) embedding_dim = 384 # إنشاء فهرس FAISS (بحث بالتشابه الداخلي) مع دعم المعرفات index = faiss.IndexIDMap(faiss.IndexFlatIP(embedding_dim)) # قائمة لتخزين مدخلات الذاكرة (كل مدخل: سؤال وجواب ومعرف) memory = [] next_id = 0 # عتبة التشابه الكوني (cosine similarity) SIMILARITY_THRESHOLD = 0.7 class QAEntry(BaseModel): input: str output: str class Query(BaseModel): input: str # دالة لحساب متجه embedding لنص وإرجاعه بعد التطبيع def get_embedding(text: str) -> np.ndarray: embed = model.encode([text])[0] embed = embed / np.linalg.norm(embed) return embed.astype('float32') @app.post("/add") def add_entry(entry: QAEntry): """ إضافة مدخل جديد للذاكرة (سؤال + جواب). """ global next_id embedding = get_embedding(entry.input) entry_id = next_id next_id += 1 # إضافة المتجه إلى فهرس FAISS مع المعرف index.add_with_ids(np.array([embedding]), np.array([entry_id], dtype='int64')) # حفظ السؤال والجواب في القائمة memory.append({"id": entry_id, "input": entry.input, "output": entry.output}) return {"status": "تم الإضافة", "id": entry_id} @app.post("/query") def query_memory(query: Query): """ الاستعلام عن سؤال جديد ومحاولة إيجاد جواب من الذاكرة. """ if not memory: return {"output": None, "info": "الذاكرة فارغة"} embedding = get_embedding(query.input) # البحث عن أقرب إجابة D, I = index.search(np.array([embedding]), k=1) if I[0][0] == -1: return {"output": None} similarity = float(D[0][0]) matched_id = int(I[0][0]) if similarity >= SIMILARITY_THRESHOLD: for entry in memory: if entry["id"] == matched_id: return {"output": entry["output"], "id": matched_id, "similarity": similarity} return {"output": None, "id": None, "similarity": similarity} @app.get("/memory") def get_memory(): """ استعراض جميع محتويات الذاكرة (الأسئلة المخزنة + الردود). """ return {"memory": memory} @app.delete("/delete/{entry_id}") def delete_entry(entry_id: int): """ حذف مدخل من الذاكرة حسب المعرف وإزالته من فهرس FAISS. """ for idx, entry in enumerate(memory): if entry["id"] == entry_id: del memory[idx] # إزالة المتجه من فهرس FAISS باستخدام remove_ids index.remove_ids(np.array([entry_id], dtype='int64')) return {"status": "تم الحذف", "id": entry_id} raise HTTPException(status_code=404, detail="المعرف غير موجود")