Update rag.py

rag.py

@@ -1,73 +1,73 @@
-import os
-from typing import List, Dict, Tuple
-import numpy as np
-from openai import OpenAI
-from services.faq_store import FAQ_ENTRIES, FAQ_VECS
-
-RAG_CONFIDENCE_THRESHOLD = 0.6
-MAX_FAQ_MATCHES = 3
-_EMBED_MODEL = "text-embedding-3-small"
-_CHAT_MODEL = "gpt-4o-mini"
-
-SYSTEM_PROMPT = (
-    "You are a helpful assistant for ScanAssured, a medical document OCR and NER app. "
-    "Answer only based on the provided FAQ context. "
-    "You do NOT have access to any user scan results or personal medical data. "
-    "For personal medical advice, always direct users to a qualified healthcare professional. "
-    "Keep answers concise and clear."
-)
-FALLBACK_MESSAGE = (
-    "I'm not certain about that. Please consult a qualified healthcare professional "
-    "for personal medical advice, or refer to the app documentation for usage questions."
-)
-
-client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
-
-# In-memory embedding cache for repeated queries
-_query_cache: dict[str, np.ndarray] = {}
-
-
-def cosine(a: np.ndarray, b: np.ndarray) -> float:
-    return float(a.dot(b) / (np.linalg.norm(a) * np.linalg.norm(b)))
-
-
-async def get_answer(question: str, history: List[Dict]) -> Tuple[str, List[Dict]]:
-    # Embed query (with in-memory cache)
-    if question in _query_cache:
-        vec = _query_cache[question]
-    else:
-        resp = client.embeddings.create(model=_EMBED_MODEL, input=question)
-        vec = np.array(resp.data[0].embedding, dtype=np.float32)
-        _query_cache[question] = vec
-
-    # Cosine similarity against all FAQ vectors
-    scores = [(fid, cosine(vec, fvec)) for fid, fvec in FAQ_VECS]
-    scores.sort(key=lambda x: x[1], reverse=True)
-
-    # Fallback if no FAQ meets the confidence threshold
-    if not scores or scores[0][1] < RAG_CONFIDENCE_THRESHOLD:
-        return FALLBACK_MESSAGE, []
-
-    # Gather top matches
-    matches = []
-    for fid, score in scores[:MAX_FAQ_MATCHES]:
-        faq = FAQ_ENTRIES[fid]
-        matches.append({"id": fid, "answer": faq["answer"], "source": faq["source"], "score": score})
-
-    # Build message list for GPT
-    messages: List[Dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
-    for msg in history:
-        messages.append({"role": msg["role"], "content": msg["content"]})
-    for faq in matches:
-        messages.append({"role": "system", "content": faq["answer"]})
-    messages.append({"role": "user", "content": question})
-
-    chat_resp = client.chat.completions.create(
-        model=_CHAT_MODEL,
-        messages=messages,
-        stream=False,
-    )
-    answer = chat_resp.choices[0].message.content
-
-    citations = [{"id": faq["id"], "source": faq["source"]} for faq in matches]
-    return answer, citations
+import os
+from typing import List, Dict, Tuple
+import numpy as np
+from openai import OpenAI
+from services.faq_store import FAQ_ENTRIES, FAQ_VECS
+
+RAG_CONFIDENCE_THRESHOLD = 0.6
+MAX_FAQ_MATCHES = 3
+_EMBED_MODEL = "text-embedding-3-small"
+_CHAT_MODEL = "gpt-4o-mini"
+
+SYSTEM_PROMPT = (
+    "You are a helpful assistant for ScanAssured, a medical document OCR and NER app. "
+    "Answer only based on the provided FAQ context. "
+    "You do NOT have access to any user scan results or personal medical data. "
+    "For personal medical advice, always direct users to a qualified healthcare professional. "
+    "Keep answers concise and clear."
+)
+FALLBACK_MESSAGE = (
+    "I'm not certain about that. Please consult a qualified healthcare professional "
+    "for personal medical advice, or refer to the app documentation for usage questions."
+)
+
+client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+# In-memory embedding cache for repeated queries
+_query_cache: dict[str, np.ndarray] = {}
+
+
+def cosine(a: np.ndarray, b: np.ndarray) -> float:
+    return float(a.dot(b) / (np.linalg.norm(a) * np.linalg.norm(b)))
+
+
+async def get_answer(question: str, history: List[Dict]) -> Tuple[str, List[Dict]]:
+    # Embed query (with in-memory cache)
+    if question in _query_cache:
+        vec = _query_cache[question]
+    else:
+        resp = client.embeddings.create(model=_EMBED_MODEL, input=question)
+        vec = np.array(resp.data[0].embedding, dtype=np.float32)
+        _query_cache[question] = vec
+
+    # Cosine similarity against all FAQ vectors
+    scores = [(fid, cosine(vec, fvec)) for fid, fvec in FAQ_VECS]
+    scores.sort(key=lambda x: x[1], reverse=True)
+
+    # Fallback if no FAQ meets the confidence threshold
+    if not scores or scores[0][1] < RAG_CONFIDENCE_THRESHOLD:
+        return FALLBACK_MESSAGE, []
+
+    # Gather top matches
+    matches = []
+    for fid, score in scores[:MAX_FAQ_MATCHES]:
+        faq = FAQ_ENTRIES[fid]
+        matches.append({"id": fid, "answer": faq["answer"], "source": faq["source"], "score": score})
+
+    # Build message list for GPT
+    messages: List[Dict] = [{"role": "system", "content": SYSTEM_PROMPT}]
+    for msg in history:
+        messages.append({"role": msg["role"], "content": msg["content"]})
+    for faq in matches:
+        messages.append({"role": "system", "content": faq["answer"]})
+    messages.append({"role": "user", "content": question})
+
+    chat_resp = client.chat.completions.create(
+        model=_CHAT_MODEL,
+        messages=messages,
+        stream=False,
+    )
+    answer = chat_resp.choices[0].message.content
+
+    citations = [{"id": faq["id"], "source": faq["source"]} for faq in matches]
+    return answer, citations