Gokulvarthan-Ramesh/aionboarding
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: AI patient onboarding backend and documentation

Browse Source
This commit is contained in:
Gokulvarthan-Ramesh
2026-08-27 12:15:09 +05:30
parent b9440772d1
commit 59c1372fe3
49 changed files with 4012 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ai-patient-onboarding/app/services/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Services package initialization."""
ai-patient-onboarding/app/services/ai_onboarding_service.py
+21
View File
@@ -0,0 +1,21 @@
from app.ai.orchestrator import AIOrchestrator
class AIOnboardingService:
def __init__(self):
self.orchestrator = AIOrchestrator()
async def process_message(
self,
session,
state,
conversation_history,
user_message,
):
return await self.orchestrator.process(
session=session,
state=state,
conversation_history=conversation_history,
user_message=user_message,
)
ai-patient-onboarding/app/services/completion_service.py
+52
View File
@@ -0,0 +1,52 @@
import json
class CompletionService:
def is_complete(
self,
session,
state,
) -> bool:
data = json.loads(
state.state_json
)
reason = data.get("reason_for_visit")
symptoms = data.get("symptoms")
history = data.get("medical_history")
medications = data.get("medications")
allergies = data.get("allergies")
# Check structured status dictionary format
if isinstance(reason, dict):
if reason.get("status") == "not_collected" or not reason.get("value"):
return False
elif not reason:
return False
if isinstance(symptoms, dict):
if symptoms.get("status") == "not_collected" or len(symptoms.get("items", [])) == 0:
return False
elif not symptoms:
return False
if isinstance(history, dict):
if history.get("status") == "not_collected":
return False
elif history is None:
return False
if isinstance(medications, dict):
if medications.get("status") == "not_collected":
return False
elif medications is None:
return False
if isinstance(allergies, dict):
if allergies.get("status") == "not_collected":
return False
elif allergies is None:
return False
return True
ai-patient-onboarding/app/services/conversation_engine.py
+75
View File
@@ -0,0 +1,75 @@
class ConversationEngine:
def generate_response(
self,
session,
state,
user_message,
):
stage = session.stage
if stage == "greeting":
return {
"message": (
"What brings you to the hospital today?"
),
"next_stage": "reason_for_visit",
}
if stage == "reason_for_visit":
return {
"message": (
"Can you tell me more about what "
"you are experiencing?"
),
"next_stage": "symptom_collection",
}
if stage == "symptom_collection":
return {
"message": (
"How long have you been experiencing "
"this problem?"
),
"next_stage": "symptom_details",
}
if stage == "symptom_details":
return {
"message": (
"Are you currently taking any "
"medications?"
),
"next_stage": "medications",
}
if stage == "medications":
return {
"message": (
"Do you have any known allergies?"
),
"next_stage": "allergies",
}
if stage == "allergies":
return {
"message": (
"Thank you. I have collected the "
"initial information."
),
"next_stage": "summary",
}
return {
"message": (
"Thank you. Let's continue."
),
"next_stage": stage,
}
ai-patient-onboarding/app/services/medical_speech_normalizer.py
+65
View File
@@ -0,0 +1,65 @@
import re
# Common Indian phonetic variations & clinical abbreviations
PHONETIC_REPLACEMENTS = {
# Medical terms & symptoms
r"\b(suger|shugar|sugur)\b": "sugar (diabetes)",
r"\b(presure|presur|blood preser)\b": "blood pressure",
r"\b(bp)\b": "BP",
r"\b(chest painu|chestpain|chst pain)\b": "chest pain",
r"\b(hart attack|heart attacku)\b": "heart attack",
r"\b(stomech pain|stomach painu)\b": "stomach pain",
r"\b(head ache|hedache|headacheu)\b": "headache",
r"\b(breath kashtam|breathing problem|moochu thinaral)\b": "breathing difficulty",
r"\b(injekshan|injaction|injecshun)\b": "injection",
r"\b(opration|oparation)\b": "operation",
r"\b(tablat|tablit|mathirai|goli)\b": "tablets",
r"\b(alargy|alergi)\b": "allergy",
r"\b(fevar|favar)\b": "fever",
r"\b(coughu|coff)\b": "cough",
r"\b(vomit|vomitingu|vomitin)\b": "vomiting",
# Common Indian Medicines (Phonetic normalization)
r"\b(parasetamal|paracitamol|paracetamol 650|dolo 650|dolo)\b": "Paracetamol (Dolo 650)",
r"\b(metformin|methformin|glycomet)\b": "Metformin",
r"\b(telmisartan|telma|telmikind)\b": "Telmisartan",
r"\b(amlodipine|amlong|amlo)\b": "Amlodipine",
r"\b(pantoprazole|pantocid|pan d|pan 40)\b": "Pantoprazole",
r"\b(omeprazole|omez)\b": "Omeprazole",
r"\b(atorvastatin|atorva|atorlip)\b": "Atorvastatin",
r"\b(cetirizine|cetzine|citrizin)\b": "Cetirizine",
r"\b(azithromycin|azithral|azithro)\b": "Azithromycin",
r"\b(amoxicillin|mox|augmentin)\b": "Amoxicillin",
r"\b(insulin injection|insuline)\b": "Insulin",
}
def normalize_medical_speech(transcript: str, language_code: str = "en-IN") -> str:
"""
Cleans, sanitizes, and normalizes spoken transcripts from STT models
to ensure high clinical fidelity while retaining the patient's original meaning.
"""
if not transcript or not transcript.strip():
return ""
cleaned = transcript.strip()
# Remove repeated filler words or audio stutter artifacts
cleaned = re.sub(r"\b(um+|uh+|ah+|erm+|er+)\b", "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
# If it's English or Romanized Indic code-mixed speech, apply phonetic dictionary
lang_lower = language_code.lower()
if lang_lower.startswith("en") or any(c.isascii() for c in cleaned):
for pattern, replacement in PHONETIC_REPLACEMENTS.items():
cleaned = re.sub(pattern, replacement, cleaned, flags=re.IGNORECASE)
# Standardize spaces and punctuation
cleaned = re.sub(r"\s+([.,!?])", r"\1", cleaned)
cleaned = re.sub(r"\s+", " ", cleaned).strip()
# Capitalize first letter if needed
if cleaned and len(cleaned) > 1:
cleaned = cleaned[0].upper() + cleaned[1:]
return cleaned
ai-patient-onboarding/app/services/onboarding_service.py
+103
View File
@@ -0,0 +1,103 @@
import json
from sqlalchemy.orm import Session
from app.database.models import (
OnboardingMessage,
OnboardingSession,
OnboardingState,
)
class OnboardingService:
def get_session(
self,
db: Session,
session_id: str,
):
return (
db.query(OnboardingSession)
.filter(
OnboardingSession.session_id == session_id
)
.first()
)
def get_state(
self,
db: Session,
session_id: str,
):
state = (
db.query(OnboardingState)
.filter(
OnboardingState.session_id == session_id
)
.first()
)
if not state:
default_state = {
"reason_for_visit": {
"status": "not_collected",
"value": None
},
"symptoms": {
"status": "not_collected",
"items": []
},
"medical_history": {
"status": "not_collected",
"items": []
},
"medications": {
"status": "not_collected",
"items": []
},
"allergies": {
"status": "not_collected",
"items": []
}
}
state = OnboardingState(
session_id=session_id,
state_json=json.dumps(default_state),
)
db.add(state)
db.commit()
db.refresh(state)
return state
def save_message(
self,
db: Session,
session_id: str,
role: str,
message: str,
stage: str,
):
new_message = OnboardingMessage(
session_id=session_id,
role=role,
message=message,
stage=stage,
)
db.add(new_message)
db.commit()
return new_message
def update_state(
self,
db: Session,
state: OnboardingState,
data: dict,
):
state.state_json = json.dumps(data)
db.commit()
db.refresh(state)
return state
ai-patient-onboarding/app/services/patient_submission_service.py
+13
View File
@@ -0,0 +1,13 @@
class PatientSubmissionService:
async def submit(
self,
result: dict,
) -> dict:
# Destination patient / hospital EHR / HIMS API integration point
# In production, this can perform an async HTTP POST to your Hospital Backend
return {
"success": True,
"message": "Patient data successfully submitted to hospital system",
"patient_id": result.get("patient_id"),
}
ai-patient-onboarding/app/services/result_extraction_service.py
+139
View File
@@ -0,0 +1,139 @@
import json
from app.ai.openai_client import OpenAIClient
from app.config import OPENAI_API_KEY
from app.schemas.onboarding import PatientOnboardingResult
class ResultExtractionService:
def __init__(self):
self.openai = OpenAIClient()
async def extract_result(
self,
session,
state,
conversation_history,
) -> dict:
current_state = json.loads(
state.state_json
)
prompt = f"""
You are a hospital patient onboarding data extraction assistant.
Your task is to convert the completed onboarding information
into structured patient intake data.
You are NOT a doctor.
Do not diagnose the patient.
Do not add information that was not provided.
PATIENT:
{json.dumps({
"id": session.patient_id,
"name": session.patient_name,
"age": session.age,
"gender": session.gender,
}, indent=2)}
CATEGORY:
{json.dumps({
"type": session.category_type,
"code": session.category_code,
"name": session.category_name,
}, indent=2)}
CURRENT STATE:
{json.dumps(current_state, indent=2)}
CONVERSATION:
{json.dumps(conversation_history, indent=2)}
Return ONLY JSON conforming to this exact structure:
{{
"patient_id": "{session.patient_id}",
"category": {{
"type": "{session.category_type}",
"code": "{session.category_code}",
"name": "{session.category_name}"
}},
"chief_complaint": "string or null",
"symptoms": [
{{
"name": "string",
"duration": "string or null",
"severity": "string or null",
"description": "string or null"
}}
],
"medical_history": [],
"medications": [],
"allergies": [],
"additional_information": "string or null",
"language": "{session.language}",
"onboarding_completed": true
}}
"""
if OPENAI_API_KEY and OPENAI_API_KEY.strip() not in ["your_openai_api_key", "your_openai_key", ""]:
try:
raw_response = await self.openai.generate_response(
system_prompt=prompt,
user_message="Generate the final structured patient onboarding result.",
)
cleaned = raw_response.strip()
if cleaned.startswith("```json"):
cleaned = cleaned[7:]
if cleaned.startswith("```"):
cleaned = cleaned[3:]
if cleaned.endswith("```"):
cleaned = cleaned[:-3]
cleaned = cleaned.strip()
parsed = json.loads(cleaned)
validated = PatientOnboardingResult.model_validate(parsed)
return validated.model_dump()
except Exception as e:
print(f"[Result Extraction Error]: {e}, falling back to structured state")
# Fallback simulator from state
chief_complaint = current_state.get("reason_for_visit")
if isinstance(chief_complaint, dict):
chief_complaint = chief_complaint.get("value")
symptoms_raw = current_state.get("symptoms")
symptoms_list = []
if isinstance(symptoms_raw, dict):
symptoms_list = symptoms_raw.get("items", [])
elif isinstance(symptoms_raw, list):
symptoms_list = symptoms_raw
meds_raw = current_state.get("medications")
meds_list = meds_raw.get("items", []) if isinstance(meds_raw, dict) else (meds_raw or [])
allergies_raw = current_state.get("allergies")
allergies_list = allergies_raw.get("items", []) if isinstance(allergies_raw, dict) else (allergies_raw or [])
history_raw = current_state.get("medical_history")
history_list = history_raw.get("items", []) if isinstance(history_raw, dict) else (history_raw or [])
fallback_result = {
"patient_id": session.patient_id,
"category": {
"type": session.category_type,
"code": session.category_code,
"name": session.category_name,
},
"chief_complaint": chief_complaint or "Consultation Intake",
"symptoms": symptoms_list or [{"name": chief_complaint or "Reported discomfort", "duration": "recent", "severity": "mild", "description": None}],
"medical_history": history_list,
"medications": meds_list,
"allergies": allergies_list,
"additional_information": None,
"language": session.language,
"onboarding_completed": True,
}
validated = PatientOnboardingResult.model_validate(fallback_result)
return validated.model_dump()
ai-patient-onboarding/app/services/validation_service.py
+32
View File
@@ -0,0 +1,32 @@
from app.ai.claude_client import ClaudeClient
from app.specialties.registry import get_specialty_config
class ValidationService:
def __init__(self):
self.claude = ClaudeClient()
async def validate_state_extraction(
self,
session,
conversation_history: list,
extracted_state: dict,
) -> dict:
specialty_config = get_specialty_config(session.category_code)
patient_context = {
"id": session.patient_id,
"name": session.patient_name,
"age": session.age,
"gender": session.gender,
"category": session.category_name,
}
validation = await self.claude.validate_extraction(
patient_context=patient_context,
specialty_config=specialty_config,
conversation_history=conversation_history,
extracted_data=extracted_state,
)
return validation
ai-patient-onboarding/app/services/voice_service.py
+270
View File
@@ -0,0 +1,270 @@
import base64
import io
import os
import wave
import struct
from pathlib import Path
from uuid import uuid4
import httpx2
from openai import AsyncOpenAI
from app.config import OPENAI_API_KEY, SARVAM_API_KEY, GEMINI_API_KEY, GEMINI_MODEL
from app.services.medical_speech_normalizer import normalize_medical_speech
BASE_DIR = Path(__file__).resolve().parent.parent.parent
AUDIO_DIR = BASE_DIR / "data" / "audio"
AUDIO_DIR.mkdir(parents=True, exist_ok=True)
class VoiceService:
def __init__(self):
self.openai_key = OPENAI_API_KEY
self.sarvam_key = SARVAM_API_KEY
self.gemini_key = GEMINI_API_KEY
self.gemini_model = GEMINI_MODEL
def get_initial_greeting(self, patient_name: str | None, language: str) -> str:
name = patient_name or "அவர்களே"
lang_lower = language.lower()
if lang_lower.startswith("ta"):
return f"வணக்கம் {name}, CureSelect மருத்துவமனைக்கு நல்வரவு. இன்று தங்களுக்கு ஏற்பட்டுள்ள உடல்நல பாதிப்பு பற்றி கூற முடியுமா?"
elif lang_lower.startswith("hi"):
return f"नमस्ते {name} जी, CureSelect अस्पताल में आपका स्वागत है। आज आप किस स्वास्थ्य समस्या के लिए परामर्श लेना चाहते हैं?"
elif lang_lower.startswith("te"):
return f"నమస్కారం {name} గారు, CureSelect హాస్పిటల్‌కు స్వాగతం. ఈరోజు మీకున్న ఆరోగ్య సమస్య గురించి చెప్పగలరా?"
elif lang_lower.startswith("kn"):
return f"ನಮಸ್ಕಾರ {name} ಅವರೇ, CureSelect ಆಸ್ಪತ್ರೆಗೆ ಸುಸ್ವಾಗತ. ಇಂದು ನಿಮ್ಮ ಆರೋಗ್ಯದ ತೊಂದರೆ ಬಗ್ಗೆ ತಿಳಿಸಬಹುದೇ?"
else:
return f"Hello {name}, welcome to CureSelect Healthcare. Could you please share what health symptoms you are experiencing today?"
async def speech_to_text(
self,
audio_bytes: bytes,
filename: str = "audio.webm",
language_code: str = "en-IN",
specialty_code: str = "general",
patient_context: dict | None = None,
) -> str:
"""
High-precision multi-model Indian speech transcription pipeline:
1. Sarvam AI Saaras v2 (Specialized Indic Speech Engine)
2. Google Gemini Multimodal Audio (Multilingual Code-Switching & Acoustic Context)
3. OpenAI Whisper-1 (Medical Specialty Context Boosted)
4. Clinical & Phonetic Post-Processing Normalizer
"""
lang_prefix = language_code.lower()[:2]
raw_transcript = ""
# 1. Sarvam AI Saaras v2 for Native Indian Languages
if (
self.sarvam_key
and self.sarvam_key.strip()
and self.sarvam_key not in ["your_sarvam_api_key", "your_sarvam_key", ""]
and lang_prefix in ["ta", "hi", "te", "kn", "bn", "mr", "gu", "ml", "pa", "od", "en"]
):
try:
async with httpx2.AsyncClient(timeout=8.0) as client:
files = {
"file": (filename, audio_bytes, "audio/webm")
}
data = {
"model": "saaras:v2",
"language_code": language_code,
"with_diacritics": "true",
}
headers = {
"api-subscription-key": self.sarvam_key
}
res = await client.post(
"https://api.sarvam.ai/speech-to-text",
files=files,
data=data,
headers=headers,
)
if res.status_code == 200:
res_json = res.json()
transcript = res_json.get("transcript", "").strip()
if transcript:
raw_transcript = transcript
except Exception as e:
print(f"[Sarvam STT Error]: {e}")
# 2. Google Gemini Multimodal Audio (Excellent with Tanglish/Hinglish & Medical terms)
if (
not raw_transcript
and self.gemini_key
and self.gemini_key.strip()
and self.gemini_key not in ["your_gemini_key", ""]
):
try:
import google.generativeai as genai
genai.configure(api_key=self.gemini_key)
model = genai.GenerativeModel(self.gemini_model or "gemini-1.5-flash")
mime_type = "audio/webm"
if filename.endswith(".wav"):
mime_type = "audio/wav"
elif filename.endswith(".mp3"):
mime_type = "audio/mp3"
audio_part = {
"mime_type": mime_type,
"data": audio_bytes,
}
audio_prompt = f"""
You are an expert medical transcription engine for Indian hospitals.
Accurately transcribe the spoken audio into text.
Language context: {language_code}. Specialty: {specialty_code}.
Handle code-switching (e.g. Tamil/Hindi mixed with English medical words like BP, sugar, chest pain, tablets, fever, vomiting).
Return ONLY the exact verbatim transcription without explanation, punctuation commentary, or markdown quotes.
"""
response = await model.generate_content_async([audio_part, audio_prompt])
text = response.text.strip().replace('"', '').replace("```", "")
if text and len(text) > 1 and "unable to transcribe" not in text.lower():
raw_transcript = text
except Exception as e:
print(f"[Gemini Audio STT Error]: {e}")
# 3. OpenAI Whisper-1 with Medical Specialty Context Prompting
if (
not raw_transcript
and self.openai_key
and self.openai_key.strip() not in ["your_openai_api_key", "your_openai_key", ""]
):
client = AsyncOpenAI(api_key=self.openai_key, timeout=8.0)
try:
audio_file = io.BytesIO(audio_bytes)
audio_file.name = filename if filename.endswith((".wav", ".mp3", ".webm", ".m4a")) else "audio.webm"
specialty_vocab = {
"cardiology": "Cardiology, chest pain, palpitations, ECG, hypertension, BP, heart attack, shortness of breath, Metformin, Telmisartan, Amlodipine, Atorvastatin, Sorbitrate",
"orthopedics": "Orthopedics, joint pain, knee pain, fracture, swelling, arthritis, Paracetamol, Tramadol, Calcium, Vitamin D, physiotherapy",
"general": "General medicine, fever, cough, cold, headache, diabetes, high blood pressure, vomiting, loose motion, Paracetamol 650, Dolo 650, Azithromycin, Pantoprazole",
"pediatrics": "Pediatrics, baby fever, vaccination, cough, cold, vomiting, appetite loss, Paracetamol drops",
"dermatology": "Dermatology, skin rash, itching, redness, fungal infection, Cetirizine, ointment, allergy",
}
spec_terms = specialty_vocab.get(specialty_code, specialty_vocab["general"])
lang_prompts = {
"ta": f"மருத்துவ ஆலோசனை, நெஞ்சு வலி, சர்க்கரை வியாதி, பிபி, மாத்திரை, காய்ச்சல், இருமல், வலி, {spec_terms}",
"hi": f"चिकित्सा, सीने में दर्द, बीपी, शुगर, बुखार, दवाई, सिरदर्द, खांसी, {spec_terms}",
"te": f"వైద్యం, ఛాతీ నొప్పి, షుగర్, బీపీ, మందులు, జ్వరం, {spec_terms}",
"kn": f"ವೈದ್ಯಕೀಯ, ಎದೆ ನೋವು, ಬಿಪಿ, ಸಕ್ಕರೆ ಕಾಯಿಲೆ, ಮಾತ್ರೆಗಳು, ಜ್ವರ, {spec_terms}",
"en": f"Medical symptoms, chest pain, diabetes, blood pressure, fever, cough, {spec_terms}",
}
med_prompt = lang_prompts.get(lang_prefix, lang_prompts["en"])
transcript = await client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language=lang_prefix if lang_prefix in ["ta", "hi", "te", "kn", "en"] else "en",
prompt=med_prompt,
temperature=0.0,
)
if transcript.text and len(transcript.text.strip()) > 1:
raw_transcript = transcript.text.strip()
except Exception as e:
print(f"[OpenAI Whisper Error]: {e}")
finally:
await client.close()
# 4. Apply Medical Speech Normalizer to eliminate phonetic mishearings
normalized = normalize_medical_speech(raw_transcript, language_code=language_code)
return normalized
async def text_to_speech(
self,
text: str,
language_code: str = "en-IN",
) -> tuple[str, bytes]:
lang_prefix = language_code.lower()[:2]
audio_id = f"aud_{uuid4().hex[:12]}"
# Clean punctuation and brackets before synthesis
cleaned_text = (
text.replace("(", "")
.replace(")", "")
.replace("*", "")
.replace("#", "")
.replace("- ", "")
.strip()
)
# 1. Use Sarvam AI TTS for Indian Languages (Native Indian Accent)
if (
self.sarvam_key
and self.sarvam_key.strip()
and self.sarvam_key not in ["your_sarvam_api_key", "your_sarvam_key", ""]
and lang_prefix in ["ta", "hi", "te", "kn", "bn", "mr", "gu", "ml", "pa", "od", "en"]
):
try:
async with httpx2.AsyncClient(timeout=6.0) as client:
headers = {
"api-subscription-key": self.sarvam_key,
"Content-Type": "application/json",
}
payload = {
"inputs": [cleaned_text],
"target_language_code": language_code,
"speaker": "meera",
"model": "bulbul:v1",
}
res = await client.post(
"https://api.sarvam.ai/text-to-speech",
json=payload,
headers=headers,
)
if res.status_code == 200:
res_json = res.json()
audios = res_json.get("audios", [])
if audios:
audio_data = base64.b64decode(audios[0])
file_path = AUDIO_DIR / f"{audio_id}.wav"
file_path.write_bytes(audio_data)
return audio_id, audio_data
except Exception as e:
print(f"[Sarvam TTS Error]: {e}")
# 2. Use OpenAI TTS
if self.openai_key and self.openai_key.strip() not in ["your_openai_api_key", "your_openai_key", ""]:
client = AsyncOpenAI(api_key=self.openai_key, timeout=6.0)
try:
response = await client.audio.speech.create(
model="tts-1",
voice="shimmer",
input=cleaned_text,
speed=1.0,
)
audio_data = response.content
file_path = AUDIO_DIR / f"{audio_id}.mp3"
file_path.write_bytes(audio_data)
return audio_id, audio_data
except Exception as e:
print(f"[OpenAI TTS Error]: {e}")
finally:
await client.close()
# 3. Fallback
audio_data = self._generate_fallback_wav()
file_path = AUDIO_DIR / f"{audio_id}.wav"
file_path.write_bytes(audio_data)
return audio_id, audio_data
def _generate_fallback_wav(self) -> bytes:
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(16000)
data = [int(1000 * (i % 32 < 16)) for i in range(4000)]
wf.writeframes(struct.pack(f"<{len(data)}h", *data))
return buf.getvalue()
def get_audio_path(self, audio_id: str) -> Path | None:
for ext in [".wav", ".mp3", ".webm", ".m4a"]:
p = AUDIO_DIR / f"{audio_id}{ext}"
if p.exists():
return p
return None