| import os | |
| from typing import List, Dict, Tuple | |
| import numpy as np | |
| from openai import OpenAI | |
| from 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")) | |
| _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]]: | |
| 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 | |
| scores = [(fid, cosine(vec, fvec)) for fid, fvec in FAQ_VECS] | |
| scores.sort(key=lambda x: x[1], reverse=True) | |
| if not scores or scores[0][1] < RAG_CONFIDENCE_THRESHOLD: | |
| return FALLBACK_MESSAGE, [] | |
| 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}) | |
| 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 | |