import io import logging import os import re import threading import time from contextlib import asynccontextmanager from typing import List, Optional, Tuple import faiss import numpy as np import pandas as pd import requests from datasets import load_dataset from fastapi import FastAPI, File, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from rank_bm25 import BM25Okapi from sentence_transformers import SentenceTransformer logger = logging.getLogger("hr_genie_rag") logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") # ─────────────────────────── Config ─────────────────────────── MODEL_NAME = os.getenv("OPENROUTER_MODEL", "deepseek/deepseek-chat-v3-0324") OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY", "").strip() EMBEDDING_MODEL_NAME = os.getenv("EMBEDDING_MODEL", "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2") KNOWLEDGE_URL = os.getenv("KNOWLEDGE_URL", "https://huggingface.co/datasets/sharshar1/arabic-hr-rag-dataset/resolve/main/knowledge_chunks.jsonl") FAQ_URL = os.getenv("FAQ_URL", "https://huggingface.co/datasets/sharshar1/arabic-hr-rag-dataset/resolve/main/faq_pairs.jsonl") # ─────────────────────────── Lifespan ─────────────────────────── @asynccontextmanager async def lifespan(application: FastAPI): """Modern lifespan handler — replaces deprecated @app.on_event('startup').""" build_indexes() yield app = FastAPI(title="HR Genie RAG API", version="2.0.0", lifespan=lifespan) # ─────────────────────────── CORS ─────────────────────────── app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) MIN_VECTOR_SCORE: float = float(os.getenv("MIN_VECTOR_SCORE", "0.10")) VECTOR_WEIGHT: float = float(os.getenv("VECTOR_WEIGHT", "0.65")) BM25_WEIGHT: float = float(os.getenv("BM25_WEIGHT", "0.35")) # ─────────────────────────── Schemas ─────────────────────────── class ChatRequest(BaseModel): message: str = Field(..., min_length=1, max_length=10000) session_id: Optional[str] = None user_id: Optional[int] = None class SourceItem(BaseModel): title: str source_file: str page: Optional[int] = None score: float class ChatResponse(BaseModel): answer: str mode: str sources: List[SourceItem] latency_ms: int class UploadResponse(BaseModel): filename: str chunks_added: int summary: str # ─────────────────────────── Global state ─────────────────────────── search_df: Optional[pd.DataFrame] = None embedding_model: Optional[SentenceTransformer] = None faiss_index = None bm25 = None _index_lock = threading.Lock() # protects search_df, faiss_index, bm25 # ─────────────────────────── Arabic helpers ─────────────────────────── def normalize_ar(text: str) -> str: text = str(text) text = re.sub(r"[\u064B-\u0652\u0670\u0640]", "", text) text = re.sub(r"[أإآٱ]", "ا", text) text = text.replace("ى", "ي").replace("ة", "ه") text = re.sub(r"\s+", " ", text).strip() return text def simple_ar_tokenize(text: str) -> List[str]: text = normalize_ar(text) text = re.sub(r"[^\w\s]", " ", text) return text.split() def detect_response_style(query: str) -> str: text = str(query).strip() arabic_chars = len(re.findall(r"[\u0600-\u06FF]", text)) latin_chars = len(re.findall(r"[A-Za-z]", text)) if latin_chars > arabic_chars and latin_chars > 0: return "english" if arabic_chars > 0: egyptian_markers = ["ازاي", "عاوز", "عايز", "ليه", "عامل", "ازيك", "دلوقتي", "ممكن"] if any(m in text.lower() for m in egyptian_markers): return "egyptian_ar" return "fusha_ar" return "english" def style_instruction(style: str) -> str: if style == "english": return "Answer in clear natural English." if style == "egyptian_ar": return "أجب باللهجة المصرية بشكل طبيعي وواضح، وبأسلوب شات بوت ودود." return "أجب بالعربية الفصحى بشكل واضح ومهذب." def source_label_for_style(style: str) -> str: return "Source" if style == "english" else "المصدر" def sanitize_model_output(text: str) -> str: text = str(text or "") text = re.sub(r"[\u4e00-\u9fff]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return re.sub(r"[ \t]{2,}", " ", text).strip() # ─────────────────────────── Source formatting ─────────────────────────── def format_source_display(source_file: str, page: Optional[int] = None, record_type: str = "") -> str: """ Convert raw source_file value into a human-readable label. Page numbers are ONLY shown for uploaded PDFs (record_type='pdf_upload'). """ if not source_file or source_file in ("", "nan", "None"): return "قاعدة المعرفة" name = str(source_file).strip() fallback_map = { "knowledge_chunks": "قاعدة المعرفة HR", "faq_pairs": "الأسئلة الشائعة HR", "resignation_termination": "سياسة الاستقالة والإنهاء", "leave_policy": "سياسة الإجازات", "salary_policy": "سياسة الرواتب", "hr_policy": "السياسة العامة HR", "attendance_policy": "سياسة الحضور والانصراف", "benefits_policy": "سياسة المزايا والمكافآت", "recruitment_policy": "سياسة التوظيف", "performance_policy": "سياسة تقييم الأداء", } base = os.path.basename(name) name_no_ext, _ = os.path.splitext(base) if name_no_ext in fallback_map: friendly = fallback_map[name_no_ext] else: friendly = name_no_ext.replace("_", " ").replace("-", " ").title() if record_type == "pdf_upload" and page and page > 0: return f"{friendly} (صفحة {page})" return friendly def openrouter_chat( messages: List[dict], temperature: float = 0.2, max_tokens: int = 300, _max_retries: int = 3, ) -> str: if not OPENROUTER_API_KEY: raise RuntimeError("OPENROUTER_API_KEY is missing.") headers = { "Authorization": f"Bearer {OPENROUTER_API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL_NAME, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } last_exc: Optional[Exception] = None for attempt in range(1, _max_retries + 1): try: response = requests.post( "https://openrouter.ai/api/v1/chat/completions", headers=headers, json=payload, timeout=60, ) response.raise_for_status() return sanitize_model_output(response.json()["choices"][0]["message"]["content"]) except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as exc: last_exc = exc logger.warning(f"OpenRouter attempt {attempt}/{_max_retries} failed (transient): {exc}") except requests.exceptions.HTTPError as exc: # Retry on 429 (rate limit) and 5xx server errors if exc.response is not None and exc.response.status_code in (429, 500, 502, 503, 504): last_exc = exc logger.warning(f"OpenRouter attempt {attempt}/{_max_retries} failed (HTTP {exc.response.status_code}): {exc}") else: raise # non-retryable HTTP error (4xx) if attempt < _max_retries: backoff = 2 ** (attempt - 1) # 1s, 2s logger.info(f"Retrying in {backoff}s...") time.sleep(backoff) raise RuntimeError(f"OpenRouter API failed after {_max_retries} retries: {last_exc}") # ─────────────────────────── Data loading ─────────────────────────── def _safe_col(df: pd.DataFrame, *candidates: str, default: str = "") -> pd.Series: """Return first existing column from candidates, or a constant default series.""" for col in candidates: if col in df.columns: return df[col].fillna(default) return pd.Series([default] * len(df), index=df.index) def load_data() -> pd.DataFrame: knowledge_ds = load_dataset("json", data_files=KNOWLEDGE_URL)["train"] faq_ds = load_dataset("json", data_files=FAQ_URL)["train"] knowledge_df = pd.DataFrame(knowledge_ds) faq_df = pd.DataFrame(faq_ds) # ── Knowledge chunks ────────────────────────────────────────── k = knowledge_df.copy() k["search_text"] = ( "الفئة: " + _safe_col(k, "category_ar", "category") + "\nالعنوان: " + _safe_col(k, "topic", "title", "subject") + "\nالمحتوى: " + _safe_col(k, "content", "text", "body") ) k["answer_text"] = _safe_col(k, "content", "text", "body") k["display_title"] = _safe_col(k, "topic", "title", "subject") k["source_file"] = _safe_col(k, "source_file", "source", "file") k["page_number"] = _safe_col(k, "page", "page_number", "page_num").replace("", None) k["record_type"] = "knowledge" k["category_ar"] = _safe_col(k, "category_ar", "category") # ── FAQ pairs ───────────────────────────────────────────────── f = faq_df.copy() f["search_text"] = ( "الفئة: " + _safe_col(f, "category_ar", "category") + "\nالسؤال: " + _safe_col(f, "question", "q") + "\nالإجابة: " + _safe_col(f, "answer", "a", "response") ) f["answer_text"] = _safe_col(f, "answer", "a", "response") f["display_title"] = _safe_col(f, "question", "q") f["source_file"] = _safe_col(f, "source_file", "source", "file") f["page_number"] = _safe_col(f, "page", "page_number", "page_num").replace("", None) f["record_type"] = "faq" f["category_ar"] = _safe_col(f, "category_ar", "category") common_cols = [ "category_ar", "source_file", "page_number", "record_type", "display_title", "answer_text", "search_text", ] return pd.concat([k[common_cols], f[common_cols]], ignore_index=True) # ─────────────────────────── Index building ─────────────────────────── def build_indexes() -> None: global search_df, embedding_model, faiss_index, bm25 search_df = load_data() print(f"[startup] Loaded {len(search_df)} records.") embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME) search_texts = search_df["search_text"].apply(normalize_ar).tolist() vectors = embedding_model.encode( search_texts, batch_size=32, convert_to_numpy=True, normalize_embeddings=True, ).astype("float32") faiss_index = faiss.IndexFlatIP(vectors.shape[1]) faiss_index.add(vectors) bm25 = BM25Okapi([simple_ar_tokenize(t) for t in search_df["search_text"].tolist()]) print("[startup] Indexes ready.") def _add_rows_to_index(new_df: pd.DataFrame) -> None: """Append new rows to the live indexes without full rebuild (thread-safe).""" global search_df, faiss_index, bm25 texts = new_df["search_text"].apply(normalize_ar).tolist() vectors = embedding_model.encode( texts, batch_size=32, convert_to_numpy=True, normalize_embeddings=True, ).astype("float32") with _index_lock: search_df = pd.concat([search_df, new_df], ignore_index=True) faiss_index.add(vectors) bm25 = BM25Okapi([simple_ar_tokenize(t) for t in search_df["search_text"].tolist()]) # ─────────────────────────── Retrieval ─────────────────────────── def retrieve(query: str, top_k: int = 3) -> List[dict]: if search_df is None or embedding_model is None or faiss_index is None or bm25 is None: raise RuntimeError("Indexes are not initialized.") query_norm = normalize_ar(query) # Query expansion: strip meta-task prefixes so retrieval focuses on topic # e.g. "لخص سياسة الإجازات" → "سياسة الإجازات" for better embedding match expansion_prefixes = [ r"^(لخص|اشرح|اذكر|وضح|اعطني|أعطني|قولي|اخبرني|أخبرني|ممكن\s+تقولي|ممكن\s+تشرحلي)" r"\s+(لي|لنا|ل)?\s*", r"^(summarize|explain|describe|tell me about|what is|what are)\s+", ] query_for_retrieval = query_norm for pattern in expansion_prefixes: query_for_retrieval = re.sub(pattern, "", query_for_retrieval, flags=re.IGNORECASE | re.UNICODE).strip() if not query_for_retrieval: query_for_retrieval = query_norm query_vec = embedding_model.encode( [query_for_retrieval], convert_to_numpy=True, normalize_embeddings=True, ).astype("float32") with _index_lock: local_df = search_df.copy() local_faiss = faiss_index local_bm25 = bm25 vector_scores, vector_indices = local_faiss.search(query_vec, 8) bm25_scores = local_bm25.get_scores(simple_ar_tokenize(query_for_retrieval)) bm25_top_idx = np.argsort(bm25_scores)[::-1][:8] merged: dict = {} for rank, (score, idx) in enumerate(zip(vector_scores[0], vector_indices[0]), start=1): merged[int(idx)] = {"vector_rank": rank, "vector_score": float(score), "bm25_rank": None} for rank, idx in enumerate(bm25_top_idx, start=1): idx = int(idx) merged.setdefault(idx, {"vector_rank": None, "vector_score": 0.0, "bm25_rank": None}) merged[idx]["bm25_rank"] = rank items = [] for idx, s in merged.items(): # استخدام RRF القياسي لتقليل حدة الفروق بين المراكز v = 1.0 / (10 + s["vector_rank"]) if s["vector_rank"] else 0.0 b = 1.0 / (10 + s["bm25_rank"]) if s["bm25_rank"] else 0.0 hybrid = VECTOR_WEIGHT * v + BM25_WEIGHT * b row = local_df.iloc[idx] # Resolve page number safely raw_page = row.get("page_number", None) try: page_num = int(float(raw_page)) if raw_page not in (None, "", "nan", "None") else None except (ValueError, TypeError): page_num = None items.append({ "display_title": str(row.get("display_title", "") or ""), "answer_text": str(row.get("answer_text", "") or ""), "source_file": str(row.get("source_file", "") or ""), "page_number": page_num, "record_type": str(row.get("record_type", "") or ""), "category_ar": str(row.get("category_ar", "") or ""), "score": float(hybrid), "raw_vector_score": float(s.get("vector_score", 0.0)), # cosine similarity }) return sorted(items, key=lambda x: x["score"], reverse=True)[:top_k] # ─────────────────────────── Prompt building ─────────────────────────── def build_prompt(query: str, chunks: List[dict]) -> Tuple[str, int]: style = detect_response_style(query) if not chunks: context_text = "لا توجد مصادر متاحة (No sources available)." else: context_parts = [] for i, c in enumerate(chunks, start=1): src_display = format_source_display(c["source_file"], c.get("page_number"), c.get("record_type", "")) context_parts.append( f"[المصدر {i}]\n" f"النوع: {c['record_type']}\n" f"الفئة: {c['category_ar']}\n" f"العنوان: {c['display_title']}\n" f"المحتوى: {c['answer_text']}\n" f"المرجع: {src_display}" ) context_text = "\n\n".join(context_parts) system_instruction = ( "أنت مساعد موارد بشرية (HR Genie) ذكي، خبير، وودود للغاية.\n\n" f"{style_instruction(style)}\n\n" "تعليمات هامة جداً:\n" "1. إذا كانت رسالة المستخدم مجرد تحية (مثل: مرحبا، ازيك، شكرا)، تفاعل معه بشكل طبيعي وودود وبدون تكلف.\n" "2. بالنسبة لأسئلة الموارد البشرية:\n" " - اعتمد بشكل أساسي على 'المصادر المسترجعة' المرفقة أدناه.\n" " - إذا لم تجد الإجابة الدقيقة في المصادر، استخدم ذكاءك ومعرفتك كخبير موارد بشرية لتقديم إجابة مفيدة.\n" "3. ⚠️ قاعدة إضافة المصادر (مهم جداً):\n" " - لا تكتب كلمة 'المصدر' أو 'المرجع' أو أي ملاحظة تشير إلى أنك استخدمت مصدراً أو لم تجد مصدراً. النظام سيضيف المصدر تلقائياً.\n" " - لا تكتب ملاحظات مثل '(ملاحظة: الإجابة مستندة إلى...)'. قدم الإجابة مباشرة.\n" " - إذا كانت إجابتك مجرد تحية أو إجابة من معرفتك العامة (لأن المعلومة غير موجودة بوضوح في المصادر)، **يجب** أن تبدأ ردك بكلمة [بدون_مصدر] لكي لا يضيف النظام مصادر خاطئة.\n" ) return ( f"{system_instruction}\n\n" f"المصادر المسترجعة:\n{context_text}\n\n" f"رسالة المستخدم:\n{query}\n\n" "الرد النهائي:" ), 1500 # ─────────────────────────── PDF processing ─────────────────────────── def extract_pdf_text(file_bytes: bytes) -> tuple[List[dict], str]: """ Extract text from PDF pages and attempt to extract title. Returns (pages, title). Requires PyMuPDF (fitz) installed. """ try: import fitz # PyMuPDF except ImportError: raise RuntimeError("PyMuPDF غير مثبت. أضف 'pymupdf' لـ requirements.txt") doc = fitz.open(stream=file_bytes, filetype="pdf") doc_title = doc.metadata.get("title", "").strip() if doc.metadata else "" pages = [] for i, page in enumerate(doc, start=1): text = page.get_text("text").strip() if text: pages.append({"page": i, "text": text}) doc.close() return pages, doc_title def chunk_text(text: str, chunk_size: int = 400, overlap: int = 80) -> List[str]: """Split text into overlapping chunks by word count.""" words = text.split() chunks = [] start = 0 while start < len(words): end = min(start + chunk_size, len(words)) chunks.append(" ".join(words[start:end])) if end == len(words): break start += chunk_size - overlap return chunks def summarize_text(text: str, filename: str, style: str = "fusha_ar") -> str: """Ask the LLM to summarize extracted PDF content.""" prompt = ( f"{style_instruction(style)}\n\n" f"لديك محتوى مستخرج من مستند '{filename}'.\n" "اكتب ملخصاً موجزاً في 3-5 جمل يوضح أهم النقاط.\n\n" f"المحتوى:\n{text[:3000]}\n\n" "الملخص:" ) return openrouter_chat([{"role": "user", "content": prompt}], temperature=0.2, max_tokens=1200) # ─────────────────────────── Core logic ─────────────────────────── def ask_assistant(query: str) -> dict: style = detect_response_style(query) retrieval_top_k = 5 chunks = retrieve(query, top_k=retrieval_top_k) source_label = source_label_for_style(style) max_vector_score = max((c.get("raw_vector_score", 0.0) for c in chunks), default=0.0) if max_vector_score < MIN_VECTOR_SCORE and (not chunks or chunks[0].get("score", 0.0) < 0.02): chunks_to_use = [] else: chunks_to_use = chunks prompt, max_tok = build_prompt(query, chunks_to_use) answer = openrouter_chat([{"role": "user", "content": prompt}], temperature=0.25, max_tokens=max_tok) answer = sanitize_model_output(answer) hide_sources = False if "[بدون_مصدر]" in answer: hide_sources = True answer = answer.replace("[بدون_مصدر]", "").strip() source_line_pattern = re.compile( r"\n?[\-\u2013\u2014\s]*(" r"المصدر|Source|المرجع|المصادر|الملف|اسم الملف|المستند" r")[^\n]*", re.IGNORECASE, ) answer = source_line_pattern.sub("", answer).strip() if chunks_to_use and not hide_sources: unique_sources = [] for c in chunks_to_use: src = format_source_display(c["source_file"], None, c.get("record_type", "")) if src and src not in unique_sources: unique_sources.append(src) unique_sources = unique_sources[:3] if unique_sources: answer = f"{answer}\n\n{source_label}: {'، '.join(unique_sources)}" sources = [] if not hide_sources: sources = [ { "title": c["display_title"], "source_file": format_source_display(c["source_file"], c.get("page_number"), c.get("record_type", "")), "page": c.get("page_number"), "score": round(c["score"], 4), } for c in chunks_to_use ] return {"answer": answer, "mode": "chat", "sources": sources} # ─────────────────────────── FastAPI routes ─────────────────────────── @app.get("/") def root() -> dict: return { "name": "HR Genie RAG API", "version": "2.0.0", "status": "running", "endpoints": { "health": "/health", "chat": "/chat", "upload_pdf": "/upload-pdf", "docs": "/docs", }, } @app.get("/health") def health() -> dict: record_count = len(search_df) if search_df is not None else 0 return { "status": "ok", "model": MODEL_NAME, "has_openrouter_key": bool(OPENROUTER_API_KEY), "records_indexed": record_count, } @app.post("/chat", response_model=ChatResponse) def chat(payload: ChatRequest) -> ChatResponse: start = time.time() logger.info(f"[chat] user_id={payload.user_id} session={payload.session_id} msg_len={len(payload.message)}") try: result = ask_assistant(payload.message.strip()) except Exception as exc: logger.error(f"[chat] error: {exc}") raise HTTPException(status_code=500, detail=str(exc)) from exc latency_ms = int((time.time() - start) * 1000) sources = [SourceItem(**src) for src in result.get("sources", [])] logger.info(f"[chat] responded in {latency_ms}ms, sources={len(sources)}") return ChatResponse( answer=result.get("answer", ""), mode=result.get("mode", "unknown"), sources=sources, latency_ms=latency_ms, ) @app.post("/upload-pdf", response_model=UploadResponse) async def upload_pdf(file: UploadFile = File(...)) -> UploadResponse: """ Upload a PDF file → extract text → chunk → add to live knowledge base → return summary. The uploaded content is available immediately for /chat queries. """ logger.info(f"[upload-pdf] filename={file.filename}") if not file.filename or not file.filename.lower().endswith(".pdf"): raise HTTPException(status_code=400, detail="فقط ملفات PDF مدعومة.") if search_df is None: raise HTTPException(status_code=503, detail="الـ indexes لم تُهيأ بعد.") file_bytes = await file.read() if len(file_bytes) > 50 * 1024 * 1024: # 50 MB limit raise HTTPException(status_code=413, detail="حجم الملف يتجاوز 50 MB.") try: pages, doc_title = extract_pdf_text(file_bytes) source_name = doc_title if doc_title else file.filename except RuntimeError as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc if not pages: raise HTTPException(status_code=422, detail="لم يتم استخراج أي نص من الملف. تأكد أنه PDF نصي وليس صورة.") # Build chunks dataframe rows = [] for p in pages: for chunk in chunk_text(p["text"]): rows.append({ "category_ar": "مستند مرفوع", "source_file": source_name, "page_number": p["page"], "record_type": "pdf_upload", "display_title": f"{source_name} — صفحة {p['page']}", "answer_text": chunk, "search_text": ( f"المستند: {source_name}\n" f"الصفحة: {p['page']}\n" f"المحتوى: {chunk}" ), }) if not rows: raise HTTPException(status_code=422, detail="لم ينتج أي chunk من الملف.") new_df = pd.DataFrame(rows) _add_rows_to_index(new_df) # Summarize first ~3000 chars of the document full_text = " ".join(p["text"] for p in pages) summary = summarize_text(full_text, file.filename) return UploadResponse( filename=file.filename, chunks_added=len(rows), summary=summary, )