smartiag / services /rag_service.py
hamdim's picture
Upload 25 files
02d44c3 verified
from typing import List, Dict, Optional
class RAGService:
def __init__(self):
# In-memory storage for POC: { patient_id: [context_strings] }
self.context_store: Dict[str, List[str]] = {}
def add_document(self, patient_id: str, content: str, doc_type: str = "general"):
"""Adds a piece of information to the patient's context."""
if patient_id not in self.context_store:
self.context_store[patient_id] = []
entry = f"--- DOCUMENT ({doc_type}) ---\n{content}\n"
self.context_store[patient_id].append(entry)
print(f"RAG: Added {doc_type} for patient {patient_id}")
def get_context(self, patient_id: str) -> str:
"""Retrieves all context for a patient as a formatted string."""
docs = self.context_store.get(patient_id, [])
if not docs:
return ""
return "\n".join(docs)
# Singleton instance
rag_service = RAGService()