This document provides comprehensive coverage of the FastAPI backend and Streamlit frontend, showing how they create a complete medical RAG system interface.
The system provides two complementary interfaces:
- FastAPI: Professional REST API for programmatic access and integration
- Streamlit: Interactive web UI for human exploration and testing
- Shared Core: Both interfaces use the same RAG pipeline for consistency
[User/Client] → [FastAPI/Streamlit] → [RAG Core] → [Vector DB + LLM] → [Response]
↓ ↓ ↓
[Feedback] → [Background Tasks] → [PostgreSQL + S3 Logging]
# src/api/main_api.py
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Dict, List, Optional
from fastapi import BackgroundTasks, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup procedures
print("Medical RAG API starting up...")
yield
# Shutdown procedures
print("Medical RAG API shutting down...")
app = FastAPI(
title="Medical RAG Assistant API",
description="Professional medical question answering using Retrieval-Augmented Generation",
version="1.0.0",
docs_url="/docs", # Interactive API documentation
redoc_url="/redoc", # Alternative API documentation
lifespan=lifespan, # Startup/shutdown event handling
)# src/api/main_api.py
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure for production domains
allow_credentials=True, # Support authentication cookies
allow_methods=["*"], # Support all HTTP methods
allow_headers=["*"], # Support custom headers
)# src/api/main_api.py
from pydantic import BaseModel, Field
class QuestionRequest(BaseModel):
question: str = Field(..., min_length=1, description="Medical question to answer")
model: str = Field(default="gpt-4o-mini", description="OpenAI model to use")
class QuestionResponse(BaseModel):
conversation_id: str
question: str
answer: str
response_time: float
relevance: str
total_cost: float
search_results_count: int
class FeedbackRequest(BaseModel):
conversation_id: str = Field(..., description="Conversation ID")
feedback: int = Field(..., ge=-1, le=1, description="Feedback: 1 (positive), -1 (negative)")# src/api/main_api.py
@app.post("/question", response_model=QuestionResponse)
async def handle_question(request: QuestionRequest, background_tasks: BackgroundTasks):
"""
Process medical question and return RAG-generated answer with full metrics
"""
start_time = datetime.utcnow()
conversation_id = str(uuid.uuid4())
try:
# Execute RAG pipeline
answer_data = rag(request.question, model=request.model)
# Prepare comprehensive response
response = QuestionResponse(
conversation_id=conversation_id,
question=request.question,
answer=answer_data["answer"],
response_time=answer_data["response_time"],
relevance=answer_data["relevance"],
total_cost=answer_data["total_cost"],
search_results_count=answer_data["search_results_count"],
)
# Asynchronous persistence (non-blocking)
background_tasks.add_task(
save_conversation_async, conversation_id, request.question, answer_data
)
# Optional S3 audit logging
if os.getenv("ENABLE_S3_LOGGING", "false").lower() == "true":
background_tasks.add_task(
log_api_call_to_s3, "question", conversation_id,
request.question, response.answer, start_time,
datetime.utcnow(), "success"
)
return response
except Exception as e:
# Error logging and graceful failure
if os.getenv("ENABLE_S3_LOGGING", "false").lower() == "true":
background_tasks.add_task(
log_api_call_to_s3, "question", conversation_id,
request.question, None, start_time, datetime.utcnow(),
"error", str(e)
)
raise HTTPException(status_code=500, detail=f"Error processing question: {str(e)}")# src/api/main_api.py
@app.post("/search")
async def search_only(request: Dict[str, str]):
"""
Test search functionality without OpenAI (for development/testing)
"""
try:
from rag import search
question = request.get("question", "")
results = search(question, top_k=3)
return {
"question": question,
"results_count": len(results),
"results": results
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Search error: {str(e)}")# src/api/main_api.py
async def save_conversation_async(conversation_id: str, question: str, answer_data: Dict):
"""Save conversation to PostgreSQL database asynchronously"""
try:
db.save_conversation(
conversation_id=conversation_id,
question=question,
answer_data=answer_data
)
except Exception as e:
print(f"Error saving conversation {conversation_id}: {e}")
async def log_api_call_to_s3(endpoint: str, conversation_id: str, request_data: str,
response_data: Optional[str], start_time: datetime,
end_time: datetime, status: str, error: Optional[str] = None):
"""Comprehensive API call logging to S3 for audit and analytics"""
try:
if is_s3_available():
log_data = {
"endpoint": endpoint,
"conversation_id": conversation_id,
"request_data": request_data,
"response_data": response_data,
"start_time": start_time.isoformat(),
"end_time": end_time.isoformat(),
"duration_ms": (end_time - start_time).total_seconds() * 1000,
"status": status,
"error": error,
"timestamp": datetime.utcnow().isoformat(),
}
upload_logs_to_s3([log_data], f"api_{endpoint}")
except Exception as e:
print(f"Error logging API call to S3: {e}")# src/api/main_api.py
@app.post("/feedback")
async def handle_feedback(request: FeedbackRequest):
background_tasks.add_task(save_feedback_async, request.conversation_id, request.feedback)@app.get("/metrics")
async def get_metrics():
metrics = await get_system_metrics()
return metrics# src/api/web_interface.py
import streamlit as st
import uuid
from datetime import datetime
from src.database import db
from src.core.rag import rag
from dotenv import load_dotenv
load_dotenv()
# Configure Streamlit page
st.set_page_config(
page_title="Medical RAG Assistant",
page_icon="🏥",
layout="wide"
)
st.title("🏥 Medical RAG Assistant")
st.markdown("*Powered by Qdrant Vector Database + Hybrid Search*")
st.markdown("---")# src/api/web_interface.py
if "db_initialized" not in st.session_state:
with st.spinner("Initializing medical knowledge database..."):
try:
from ingest import ingest_data
ingest_data("../data/medical_qa_metadata_sample.csv")
st.session_state["db_initialized"] = True
st.success("✅ Medical database ready!")
except Exception as e:
st.error(f"❌ Database initialization failed: {e}")
st.session_state["db_initialized"] = False# src/api/web_interface.py
if st.session_state.get("db_initialized", False):
# Enhanced input with examples
question = st.text_input(
"Ask a medical question:",
placeholder="e.g., What antibiotic is safe for pregnant women with UTI?",
help="Enter any medical question. The system will search through thousands of medical Q&A pairs."
)
# Model selection in sidebar
with st.sidebar:
st.header("⚙️ Configuration")
model_choice = st.selectbox(
"Select AI Model:",
["gpt-4o-mini", "gpt-4o"],
help="GPT-4o-mini is faster and more cost-effective"
)
st.metric("Database Status", "✅ Ready")
st.metric("Documents Indexed", "6,000+")# src/api/web_interface.py
if st.button("🔍 Search & Analyze", type="primary"):
if question:
conversation_id = str(uuid.uuid4())
with st.spinner("Searching medical knowledge base..."):
try:
# Execute RAG pipeline
answer_data = rag(question, model=model_choice)
# Display comprehensive results
st.markdown("### 🤖 AI Medical Assistant Response")
st.markdown(f"**Answer:** {answer_data['answer']}")
# Metrics dashboard
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric("Relevance", answer_data["relevance"])
with col2:
st.metric("Response Time", f"{answer_data['response_time']:.2f}s")
with col3:
st.metric("Total Cost", f"${answer_data['total_cost']:.4f}")
with col4:
st.metric("Sources Found", answer_data["search_results_count"])
# Save successful conversation
db.save_conversation(
conversation_id=conversation_id,
question=question,
answer_data=answer_data,
)
st.session_state["conversation_id"] = conversation_id
except Exception as e:
st.error(f"❌ Error processing question: {e}")
# Fallback to search-only mode
st.markdown("### 🔍 Search Results (Fallback Mode)")
try:
from ingest import hybrid_query_rrf
results = hybrid_query_rrf(question)
for i, result in enumerate(results[:3], 1):
with st.expander(
f"Result {i}: {result.get('medical_department', 'N/A')} - "
f"Score: {result.get('fusion_score', 0):.4f}"
):
st.write(f"**Question:** {result.get('question', '')[:200]}...")
st.write(f"**Answer:** {result.get('answer', '')[:300]}...")
st.write(f"**Department:** {result.get('medical_department', 'N/A')}")
st.write(f"**Severity:** {result.get('severity', 'N/A')}")
except Exception as e2:
st.error(f"❌ Search also failed: {e2}")
else:
st.warning("⚠️ Please enter a medical question.")# src/api/web_interface.py
if "conversation_id" in st.session_state:
st.markdown("### 📝 Feedback")
st.write("Was this answer helpful?")
col1, col2 = st.columns(2)
with col1:
if st.button("👍 Helpful", key="positive"):
db.save_feedback(
conversation_id=st.session_state["conversation_id"],
feedback=1
)
st.success("Thank you for your positive feedback!")
del st.session_state["conversation_id"]
with col2:
if st.button("👎 Not Helpful", key="negative"):
db.save_feedback(
conversation_id=st.session_state["conversation_id"],
feedback=-1
)
st.success("Thank you for your feedback! We'll use it to improve.")
del st.session_state["conversation_id"]# src/api/web_interface.py
# Data export functionality
st.markdown("### 📊 Data Management")
if st.button("📤 Export to CSV and Upload to S3"):
try:
# Get conversation data from database
conversation_data = get_conversation_data()
# Generate timestamped filename
CSV_FILE_NAME = f'medical_assistant/conversations_feedback_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
# Upload to S3 with comprehensive error handling
upload_csv_to_s3(conversation_data, S3_BUCKET_NAME, CSV_FILE_NAME)
st.success(
f'✅ CSV file uploaded to S3 bucket "{S3_BUCKET_NAME}" '
f'in "medical_assistant/" as "{CSV_FILE_NAME}"'
)
except Exception as e:
st.error(f"❌ Export failed: {e}")# src/api/web_interface.py
import csv
import boto3
from io import StringIO
def upload_csv_to_s3(data, bucket_name, file_name):
"""Upload conversation data to S3 as CSV with comprehensive error handling"""
try:
# Create CSV in memory
csv_buffer = StringIO()
csv_writer = csv.writer(csv_buffer)
# Write headers
csv_writer.writerow(["ID", "Question", "Answer", "Feedback", "Timestamp", "Cost", "Relevance"])
# Write conversation data
for row in data:
csv_writer.writerow([
str(row.get("id", "")),
str(row.get("question", "")),
str(row.get("answer", "")),
str(row.get("feedback", "")),
str(row.get("timestamp", "")),
str(row.get("cost", "")),
str(row.get("relevance", "")),
])
# Upload to S3 with metadata
s3_client = boto3.client("s3")
s3_client.put_object(
Bucket=bucket_name,
Key=file_name,
Body=csv_buffer.getvalue(),
Metadata={
'source': 'medical-rag-assistant',
'export_timestamp': datetime.utcnow().isoformat(),
'record_count': str(len(data))
}
)
logging.info(f"Successfully uploaded {len(data)} records to S3: {bucket_name}/{file_name}")
except Exception as e:
logging.error(f"S3 upload failed: {e}")
raise| Component | Primary Responsibilities | Secondary Functions |
|---|---|---|
| FastAPI Backend | REST API endpoints, async processing, error handling | S3 logging, metrics collection, health checks |
| Streamlit Frontend | Interactive UI, real-time feedback, data visualization | CSV export, configuration management, fallback modes |
| RAG Core Pipeline | Question processing, hybrid search, LLM generation | Cost calculation, relevance evaluation, response assembly |
| Vector Database | Document retrieval, hybrid search, RRF fusion | Collection management, performance optimization |
| PostgreSQL | Conversation persistence, feedback storage, metrics | Data export, historical analysis, reporting |
| S3 Services | Audit logging, backup storage, compliance | Data export, long-term archival, analytics |
[User Request] → [FastAPI Endpoint]
↓
[Immediate Response] ← [RAG Pipeline Execution]
↓
[Background Tasks] → [Database Storage] + [S3 Logging]
# src/api/main_api.py
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""Comprehensive system health check"""
return HealthResponse(
status="healthy",
service="medical-rag-api",
timestamp=datetime.utcnow().isoformat(),
components={
"qdrant": check_qdrant_health(),
"postgresql": check_postgres_health(),
"s3": check_s3_health(),
"openai": check_openai_health()
}
)
@app.get("/metrics")
async def get_metrics():
"""System performance and usage metrics"""
try:
metrics = await get_system_metrics()
return {
"response_times": metrics.get("avg_response_time"),
"query_volume": metrics.get("daily_queries"),
"success_rate": metrics.get("success_percentage"),
"cost_efficiency": metrics.get("avg_cost_per_query"),
"user_satisfaction": metrics.get("positive_feedback_ratio")
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Metrics unavailable: {str(e)}")# Custom exception handling for medical context
class MedicalRAGException(Exception):
def __init__(self, message: str, error_code: str, conversation_id: str = None):
self.message = message
self.error_code = error_code
self.conversation_id = conversation_id
super().__init__(self.message)
@app.exception_handler(MedicalRAGException)
async def medical_rag_exception_handler(request: Request, exc: MedicalRAGException):
return JSONResponse(
status_code=422,
content={
"error": exc.error_code,
"message": exc.message,
"conversation_id": exc.conversation_id,
"timestamp": datetime.utcnow().isoformat()
}
)# Production-ready API key validation
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException
security = HTTPBearer()
async def validate_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Validate API key for production endpoints"""
api_key = credentials.credentials
if not verify_api_key(api_key):
raise HTTPException(
status_code=401,
detail="Invalid API key",
headers={"WWW-Authenticate": "Bearer"},
)
return api_key
# Protected endpoint example
@app.post("/question", dependencies=[Depends(validate_api_key)])
async def protected_question_endpoint(...):
# Implementation
pass# Rate limiting for production use
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post("/question")
@limiter.limit("10/minute") # Limit medical queries to prevent abuse
async def rate_limited_question(request: Request, ...):
# Implementation
pass- Query Caching: Cache frequent medical questions and their responses
- Embedding Caching: Store computed embeddings for repeated queries
- Result Caching: Cache search results for similar questions
- Session Management: Maintain user context across multiple queries
- Background Tasks: Non-blocking database operations and logging
- Connection Pooling: Efficient database connection management
- Batch Processing: Group similar operations for efficiency
- Resource Management: Proper cleanup of LLM client connections
- Real-time Streaming: Server-sent events for streaming LLM responses
- Advanced Authentication: OAuth2 integration with medical institutions
- Query Analytics: Real-time usage dashboards and insights
- Mobile Optimization: Responsive design for mobile medical professionals
- Multi-tenant Support: Separate medical organization data isolation
- Integration APIs: FHIR compatibility for EMR system integration
- Audit Compliance: HIPAA-compliant logging and data handling
- Custom Model Support: Organization-specific fine-tuned models
- Voice Interface: Speech-to-text medical query processing
- Image Integration: Medical image analysis and interpretation
- Clinical Decision Support: Integration with clinical guidelines
- Multilingual Support: Support for non-English medical literature
- User Behavior Analytics: Track query patterns and success rates
- Performance Monitoring: Real-time system performance dashboards
- Cost Optimization: Automated model selection based on query complexity
- Quality Assurance: Automated medical accuracy validation