main.py

# 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="المعرف غير موجود")