File size: 3,789 Bytes
1702b26 41088d6 1702b26 41088d6 1702b26 376d7c4 41088d6 a347f56 41088d6 376d7c4 a347f56 376d7c4 41088d6 a347f56 41088d6 a347f56 41088d6 1702b26 a347f56 1702b26 41088d6 1702b26 41088d6 1702b26 41088d6 1702b26 a347f56 1702b26 41088d6 1702b26 |
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 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
import os
import zipfile
import tempfile
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_groq import ChatGroq
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
app = FastAPI()
# === Globals ===
llm = None
embeddings = None
vectorstore = None
retriever = None
chain = None
class QueryRequest(BaseModel):
question: str
def _unpack_faiss(src_path: str) -> str:
"""
If src_path is a ZIP, unzip it into a temp dir and return the folder
containing the .faiss files; if it’s already a folder, return it.
"""
if zipfile.is_zipfile(src_path):
tmp = tempfile.TemporaryDirectory()
with zipfile.ZipFile(src_path, "r") as zf:
zf.extractall(tmp.name)
for root, _, files in os.walk(tmp.name):
if any(f.endswith(".faiss") for f in files):
return root
raise RuntimeError(f"No .faiss index found inside ZIP: {src_path}")
elif os.path.isdir(src_path):
return src_path
else:
raise RuntimeError(f"Path is neither a valid ZIP nor a directory: {src_path}")
def load_and_merge_faiss(path1: str, path2: str, embeddings: HuggingFaceEmbeddings) -> FAISS:
"""
Load two FAISS indexes (either zip files or folders), merge them,
and return the combined FAISS vectorstore.
"""
dir1 = _unpack_faiss(path1)
dir2 = _unpack_faiss(path2)
vs1 = FAISS.load_local(dir1, embeddings, allow_dangerous_deserialization=True)
vs2 = FAISS.load_local(dir2, embeddings, allow_dangerous_deserialization=True)
vs1.merge_from(vs2)
return vs1
@app.on_event("startup")
def load_components():
global llm, embeddings, vectorstore, retriever, chain
# --- 1) Init LLM & Embeddings ---
llm = ChatGroq(
model="meta-llama/llama-4-scout-17b-16e-instruct",
temperature=0,
max_tokens=1024,
api_key=os.getenv("API_KEY"),
)
embeddings = HuggingFaceEmbeddings(
model_name="intfloat/multilingual-e5-large",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
# --- 2) Load & merge two FAISS indexes ---
src1 = os.getenv("FAISS_INDEX_PATH_1", "faiss_index.zip")
src2 = os.getenv("FAISS_INDEX_PATH_2", "faiss_index_extra.zip")
vectorstore = load_and_merge_faiss(src1, src2, embeddings)
# --- 3) Build retriever & QA chain ---
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
prompt = PromptTemplate(
template="""
You are an expert assistant on Islamic knowledge.
Use **only** the information in the “Retrieved context” to answer the user’s question.
Do **not** add any outside information, personal opinions, or conjecture—if the answer is not contained in the context, reply with “لا أعلم”.
Be concise, accurate, and directly address the user’s question.
Retrieved context:
{context}
User’s question:
{question}
Your response:
""",
input_variables=["context", "question"],
)
chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=False,
chain_type_kwargs={"prompt": prompt},
)
print("✅ Loaded & merged both FAISS indexes, QA chain ready.")
@app.get("/")
def root():
return {"message": "Arabic Hadith Finder API is up and running!"}
@app.post("/query")
def query(request: QueryRequest):
try:
result = chain.invoke({"query": request.question})
return {"answer": result["result"]}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|