Upload 20 files

.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+*$py.class
+.env
+venv/
+uploads/
+*.egg-info/
+dist/
+build/
+.pytest_cache/
+*.db
+*.db-journal

Dockerfile

@@ -0,0 +1,21 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Create necessary directories
+RUN mkdir -p uploads
+
+# Note: For production, use volumes to persist:
+# - /app/uploads (uploaded documents)
+# - /app/conversations.db (SQLite database)
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,10 +1,40 @@
----
-title: Ai Knowledge Assistant
-emoji: 🐢
-colorFrom: blue
-colorTo: yellow
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Backend - AI RAG Chatbot API
+
+## Setup
+
+```bash
+cd backend
+python -m venv venv
+venv\Scripts\activate   # Windows
+# source venv/bin/activate  # Mac/Linux
+
+pip install -r requirements.txt
+```
+
+## Configuration
+
+Copy `.env.example` to `.env` and fill in your API keys:
+
+```bash
+cp .env.example .env
+```
+
+## Run
+
+```bash
+uvicorn app.main:app --reload --port 8000
+```
+
+## API Endpoints
+
+| Method | Endpoint                        | Description                  |
+| ------ | ------------------------------- | ---------------------------- |
+| GET    | `/health`                       | Health check                 |
+| POST   | `/api/ask`                      | Ask question (non-streaming) |
+| POST   | `/api/ask/stream`               | Ask question (SSE streaming) |
+| POST   | `/api/ingest`                   | Upload & ingest document     |
+| POST   | `/api/ingest/batch`             | Batch upload documents       |
+| GET    | `/api/conversations`            | List conversations           |
+| GET    | `/api/conversations/{id}`       | Get conversation             |
+| DELETE | `/api/conversations/{id}`       | Delete conversation          |
+| DELETE | `/api/conversations`            | Clear all conversations      |

app/config.py

@@ -0,0 +1,35 @@
+from pydantic_settings import BaseSettings
+from typing import List
+import os
+
+
+class Settings(BaseSettings):
+    # Google Gemini
+    GOOGLE_API_KEY: str = ""
+
+    # Pinecone
+    PINECONE_API_KEY: str = ""
+    PINECONE_INDEX_NAME: str = "health-tech-kb"
+
+    # App settings
+    APP_ENV: str = "development"
+    CORS_ORIGINS: str = "http://localhost:3000"
+    UPLOAD_DIR: str = "./uploads"
+    CHUNK_SIZE: int = 1000
+    CHUNK_OVERLAP: int = 200
+    EMBEDDING_MODEL: str = "sentence-transformers/all-MiniLM-L6-v2"
+    LLM_MODEL: str = "gemini-2.5-flash"
+
+    @property
+    def cors_origins_list(self) -> List[str]:
+        return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()
+
+# Ensure upload directory exists
+os.makedirs(settings.UPLOAD_DIR, exist_ok=True)

app/database.py

@@ -0,0 +1,40 @@
+"""Database configuration and session management."""
+
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
+from sqlalchemy.orm import declarative_base
+import os
+
+# Database URL - SQLite for simplicity
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./conversations.db")
+
+# Create async engine
+engine = create_async_engine(
+    DATABASE_URL,
+    echo=False,  # Set to True for SQL debugging
+    future=True,
+)
+
+# Session factory
+async_session_maker = async_sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+)
+
+# Base class for models
+Base = declarative_base()
+
+
+async def get_db() -> AsyncSession:
+    """Get database session."""
+    async with async_session_maker() as session:
+        try:
+            yield session
+        finally:
+            await session.close()
+
+
+async def init_db():
+    """Initialize database tables."""
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)

app/db_models.py

@@ -0,0 +1,41 @@
+"""SQLAlchemy database models for conversations."""
+
+from sqlalchemy import Column, String, Integer, Text, ForeignKey, JSON
+from sqlalchemy.orm import relationship
+from app.database import Base
+import uuid
+
+
+class DBConversation(Base):
+    """Database model for conversations."""
+    
+    __tablename__ = "conversations"
+    
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    title = Column(String, default="New Chat")
+    created_at = Column(String, nullable=False)
+    updated_at = Column(String, nullable=False)
+    
+    # Relationship to messages
+    messages = relationship(
+        "DBMessage",
+        back_populates="conversation",
+        cascade="all, delete-orphan",
+        order_by="DBMessage.id"
+    )
+
+
+class DBMessage(Base):
+    """Database model for conversation messages."""
+    
+    __tablename__ = "messages"
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    conversation_id = Column(String, ForeignKey("conversations.id"), nullable=False)
+    role = Column(String, nullable=False)  # 'user' or 'assistant'
+    content = Column(Text, nullable=False)
+    timestamp = Column(String, nullable=False)
+    sources = Column(JSON, default=list)  # Stored as JSON array
+    
+    # Relationship to conversation
+    conversation = relationship("DBConversation", back_populates="messages")

app/main.py

@@ -0,0 +1,59 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app.config import settings
+from app.routes import ask, ingest, conversations, health
+
+app = FastAPI(
+    title="Health-Tech AI RAG Chatbot",
+    description="AI-powered chatbot with Retrieval-Augmented Generation",
+    version="1.0.0",
+)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.cors_origins_list,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Register routes
+app.include_router(health.router, tags=["Health"])
+app.include_router(ask.router, prefix="/api", tags=["Chat"])
+app.include_router(ingest.router, prefix="/api", tags=["Ingestion"])
+app.include_router(conversations.router, prefix="/api", tags=["Conversations"])
+
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize services on startup."""
+    from app.services.vector_store import vector_store_service
+    from app.database import init_db
+    import os
+    
+    # Ensure uploads directory exists
+    os.makedirs(settings.UPLOAD_DIR, exist_ok=True)
+    upload_path = os.path.abspath(settings.UPLOAD_DIR)
+    print(f"📁 Upload directory: {upload_path}")
+    
+    # Initialize database
+    await init_db()
+    db_path = os.path.abspath("conversations.db")
+    print(f"✅ Database initialized: {db_path}")
+    
+    # Initialize vector store
+    await vector_store_service.initialize()
+    print("✅ Vector store initialized")
+    
+    # Count existing documents
+    if os.path.exists(settings.UPLOAD_DIR):
+        doc_count = len([f for f in os.listdir(settings.UPLOAD_DIR) if os.path.isfile(os.path.join(settings.UPLOAD_DIR, f))])
+        print(f"📄 Existing documents: {doc_count}")
+    
+    print(f"🚀 Backend ready on port 8000 (env: {settings.APP_ENV})")
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

app/models.py

@@ -0,0 +1,78 @@
+from pydantic import BaseModel, Field
+from typing import Optional, List
+from datetime import datetime
+import uuid
+
+
+class MessageRequest(BaseModel):
+    question: str = Field(..., min_length=1, max_length=2000, description="User question")
+    conversation_id: Optional[str] = Field(None, description="Conversation ID for context")
+
+
+class SourceDocument(BaseModel):
+    content: str
+    source: str
+    page: Optional[int] = None
+    chunk_id: Optional[str] = None
+
+
+class MessageResponse(BaseModel):
+    answer: str
+    conversation_id: str
+    sources: List[SourceDocument] = []
+    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
+
+
+class IngestRequest(BaseModel):
+    metadata: Optional[dict] = None
+
+
+class IngestResponse(BaseModel):
+    message: str
+    documents_processed: int
+    chunks_created: int
+    filename: str
+
+
+class ConversationMessage(BaseModel):
+    role: str  # "user" or "assistant"
+    content: str
+    sources: List[SourceDocument] = []
+    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
+
+
+class Conversation(BaseModel):
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    title: str = "New Conversation"
+    messages: List[ConversationMessage] = []
+    created_at: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
+    updated_at: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
+
+
+class ConversationListItem(BaseModel):
+    id: str
+    title: str
+    message_count: int
+    created_at: str
+    updated_at: str
+
+
+class HealthResponse(BaseModel):
+    status: str = "healthy"
+    version: str = "1.0.0"
+    environment: str = "development"
+
+
+class VectorStatsResponse(BaseModel):
+    total_vectors: int = 0
+    total_documents: int = 0
+    index_name: str = ""
+    dimension: int = 384
+    error: Optional[str] = None
+
+
+class DocumentInfo(BaseModel):
+    name: str
+    size: int  # bytes
+    type: str  # PDF, Markdown, etc.
+    uploaded_at: str

app/routes/ask.py

@@ -0,0 +1,39 @@
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
+from app.models import MessageRequest, MessageResponse
+from app.services.rag_chain import rag_service
+
+router = APIRouter()
+
+
+@router.post("/ask", response_model=MessageResponse)
+async def ask_question(request: MessageRequest):
+    """Ask a question to the RAG chatbot (non-streaming)."""
+    try:
+        result = await rag_service.ask(
+            question=request.question,
+            conversation_id=request.conversation_id,
+        )
+        return MessageResponse(**result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post("/ask/stream")
+async def ask_question_stream(request: MessageRequest):
+    """Ask a question to the RAG chatbot (streaming via SSE)."""
+    try:
+        return StreamingResponse(
+            rag_service.ask_stream(
+                question=request.question,
+                conversation_id=request.conversation_id,
+            ),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no",
+            },
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

app/routes/conversations.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, HTTPException
+from typing import List
+from app.models import Conversation, ConversationListItem
+from app.services.conversation_store import conversation_store
+
+router = APIRouter()
+
+
+@router.get("/conversations", response_model=List[ConversationListItem])
+async def list_conversations():
+    """List all conversations."""
+    return await conversation_store.list_conversations()
+
+
+@router.get("/conversations/{conversation_id}", response_model=Conversation)
+async def get_conversation(conversation_id: str):
+    """Get a specific conversation with messages."""
+    conversation = await conversation_store.get_conversation(conversation_id)
+    if not conversation:
+        raise HTTPException(status_code=404, detail="Conversation not found")
+    return conversation
+
+
+@router.delete("/conversations/{conversation_id}")
+async def delete_conversation(conversation_id: str):
+    """Delete a specific conversation."""
+    success = await conversation_store.delete_conversation(conversation_id)
+    if not success:
+        raise HTTPException(status_code=404, detail="Conversation not found")
+    return {"message": "Conversation deleted"}
+
+
+@router.delete("/conversations")
+async def clear_all_conversations():
+    """Clear all conversations."""
+    await conversation_store.clear_all()
+    return {"message": "All conversations cleared"}

app/routes/health.py

@@ -0,0 +1,127 @@
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import FileResponse
+from app.config import settings
+from app.models import HealthResponse, VectorStatsResponse, DocumentInfo
+from app.services.vector_store import vector_store_service
+from typing import List
+import os
+from datetime import datetime
+
+router = APIRouter()
+
+
+@router.get("/health", response_model=HealthResponse)
+async def health_check():
+    """Health check endpoint."""
+    return HealthResponse(
+        status="healthy",
+        version="1.0.0",
+        environment=settings.APP_ENV,
+    )
+
+
+@router.get("/api/stats/vectors", response_model=VectorStatsResponse)
+async def get_vector_stats():
+    """Get vector store statistics."""
+    try:
+        # Initialize if needed
+        await vector_store_service.initialize()
+        
+        # Get stats from Pinecone
+        index = vector_store_service.pc.Index(settings.PINECONE_INDEX_NAME)
+        stats = index.describe_index_stats()
+        
+        # Count documents in uploads folder
+        upload_dir = settings.UPLOAD_DIR
+        documents = 0
+        if os.path.exists(upload_dir):
+            documents = len([f for f in os.listdir(upload_dir) 
+                           if os.path.isfile(os.path.join(upload_dir, f))])
+        
+        return VectorStatsResponse(
+            total_vectors=stats.total_vector_count,
+            total_documents=documents,
+            index_name=settings.PINECONE_INDEX_NAME,
+            dimension=stats.dimension,
+        )
+    except Exception as e:
+        return VectorStatsResponse(
+            total_vectors=0,
+            total_documents=0,
+            index_name=settings.PINECONE_INDEX_NAME,
+            dimension=384,
+            error=str(e),
+        )
+
+
+@router.get("/api/documents", response_model=List[DocumentInfo])
+async def list_documents():
+    """List all uploaded documents with metadata."""
+    upload_dir = settings.UPLOAD_DIR
+    documents = []
+    
+    if not os.path.exists(upload_dir):
+        return documents
+    
+    for filename in os.listdir(upload_dir):
+        filepath = os.path.join(upload_dir, filename)
+        if os.path.isfile(filepath):
+            stat = os.stat(filepath)
+            ext = os.path.splitext(filename)[1].lower()
+            
+            # Get file type label
+            type_map = {
+                '.pdf': 'PDF',
+                '.md': 'Markdown',
+                '.txt': 'Text',
+                '.csv': 'CSV',
+            }
+            
+            documents.append(DocumentInfo(
+                name=filename,
+                size=stat.st_size,
+                type=type_map.get(ext, ext.upper()),
+                uploaded_at=datetime.fromtimestamp(stat.st_mtime).isoformat(),
+            ))
+    
+    # Sort by upload date descending
+    documents.sort(key=lambda x: x.uploaded_at, reverse=True)
+    return documents
+
+
+@router.get("/api/documents/{filename}")
+async def download_document(filename: str):
+    """Download a specific document."""
+    # Security: prevent directory traversal
+    if ".." in filename or "/" in filename or "\\" in filename:
+        raise HTTPException(status_code=400, detail="Invalid filename")
+    
+    filepath = os.path.join(settings.UPLOAD_DIR, filename)
+    
+    if not os.path.exists(filepath):
+        raise HTTPException(status_code=404, detail="Document not found")
+    
+    return FileResponse(
+        path=filepath,
+        filename=filename,
+        media_type="application/octet-stream"
+    )
+
+
+@router.delete("/api/documents/{filename}")
+async def delete_document(filename: str):
+    """Delete a specific document."""
+    # Security: prevent directory traversal
+    if ".." in filename or "/" in filename or "\\" in filename:
+        raise HTTPException(status_code=400, detail="Invalid filename")
+    
+    filepath = os.path.join(settings.UPLOAD_DIR, filename)
+    
+    if not os.path.exists(filepath):
+        raise HTTPException(status_code=404, detail="Document not found")
+    
+    try:
+        os.remove(filepath)
+        return {"message": f"Document '{filename}' deleted successfully"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to delete document: {str(e)}")

app/routes/ingest.py

@@ -0,0 +1,99 @@
+import os
+import aiofiles
+from fastapi import APIRouter, HTTPException, UploadFile, File, Form
+from typing import Optional
+from app.models import IngestResponse
+from app.services.document_processor import document_processor
+from app.services.vector_store import vector_store_service
+from app.config import settings
+
+router = APIRouter()
+
+ALLOWED_EXTENSIONS = {".pdf", ".csv", ".md", ".txt"}
+
+
+@router.post("/ingest", response_model=IngestResponse)
+async def ingest_document(
+    file: UploadFile = File(...),
+    metadata: Optional[str] = Form(None),
+):
+    """Upload and ingest a document into the vector store."""
+    # Validate file extension
+    ext = os.path.splitext(file.filename)[1].lower()
+    if ext not in ALLOWED_EXTENSIONS:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Unsupported file type: {ext}. Allowed: {', '.join(ALLOWED_EXTENSIONS)}",
+        )
+
+    # Save uploaded file
+    file_path = os.path.join(settings.UPLOAD_DIR, file.filename)
+    try:
+        async with aiofiles.open(file_path, "wb") as f:
+            content = await file.read()
+            await f.write(content)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to save file: {str(e)}")
+
+    # Parse metadata
+    meta = None
+    if metadata:
+        import json
+
+        try:
+            meta = json.loads(metadata)
+        except json.JSONDecodeError:
+            meta = {"description": metadata}
+
+    # Process and ingest
+    try:
+        chunks = await document_processor.process_file(file_path, meta)
+        num_added = await vector_store_service.add_documents(chunks)
+
+        return IngestResponse(
+            message=f"Successfully ingested '{file.filename}'",
+            documents_processed=1,
+            chunks_created=num_added,
+            filename=file.filename,
+        )
+    except Exception as e:
+        raise HTTPException(
+            status_code=500, detail=f"Failed to process document: {str(e)}"
+        )
+
+
+@router.post("/ingest/batch")
+async def ingest_batch(files: list[UploadFile] = File(...)):
+    """Upload and ingest multiple documents."""
+    results = []
+
+    for file in files:
+        ext = os.path.splitext(file.filename)[1].lower()
+        if ext not in ALLOWED_EXTENSIONS:
+            results.append(
+                {"filename": file.filename, "status": "skipped", "reason": f"Unsupported type: {ext}"}
+            )
+            continue
+
+        file_path = os.path.join(settings.UPLOAD_DIR, file.filename)
+        try:
+            async with aiofiles.open(file_path, "wb") as f:
+                content = await file.read()
+                await f.write(content)
+
+            chunks = await document_processor.process_file(file_path)
+            num_added = await vector_store_service.add_documents(chunks)
+
+            results.append(
+                {
+                    "filename": file.filename,
+                    "status": "success",
+                    "chunks_created": num_added,
+                }
+            )
+        except Exception as e:
+            results.append(
+                {"filename": file.filename, "status": "error", "reason": str(e)}
+            )
+
+    return {"results": results, "total_files": len(files)}

app/services/conversation_store.py

@@ -0,0 +1,203 @@
+"""Database-backed conversation store."""
+
+from typing import List, Optional
+from datetime import datetime
+from sqlalchemy import select, desc
+from sqlalchemy.orm import selectinload
+from app.models import Conversation, ConversationMessage, ConversationListItem
+from app.database import async_session_maker
+from app.db_models import DBConversation, DBMessage
+
+
+class ConversationStore:
+    """Database-backed conversation storage."""
+
+    async def get_or_create(self, conversation_id: Optional[str] = None) -> Conversation:
+        """Get existing conversation or create a new one."""
+        async with async_session_maker() as session:
+            if conversation_id:
+                # Try to get existing conversation
+                result = await session.execute(
+                    select(DBConversation)
+                    .options(selectinload(DBConversation.messages))
+                    .where(DBConversation.id == conversation_id)
+                )
+                db_conv = result.scalar_one_or_none()
+                
+                if db_conv:
+                    # Convert DB model to Pydantic model
+                    return Conversation(
+                        id=db_conv.id,
+                        title=db_conv.title,
+                        messages=[
+                            ConversationMessage(
+                                role=msg.role,
+                                content=msg.content,
+                                timestamp=msg.timestamp,
+                                sources=msg.sources or [],
+                            )
+                            for msg in db_conv.messages
+                        ],
+                        created_at=db_conv.created_at,
+                        updated_at=db_conv.updated_at,
+                    )
+            
+            # Create new conversation
+            now = datetime.utcnow().isoformat() + "Z"
+            db_conv = DBConversation(
+                title="New Chat",
+                created_at=now,
+                updated_at=now,
+            )
+            session.add(db_conv)
+            await session.commit()
+            await session.refresh(db_conv)
+            
+            return Conversation(
+                id=db_conv.id,
+                title=db_conv.title,
+                messages=[],
+                created_at=db_conv.created_at,
+                updated_at=db_conv.updated_at,
+            )
+
+    async def add_message(
+        self,
+        conversation_id: str,
+        role: str,
+        content: str,
+        sources: list = None,
+    ) -> ConversationMessage:
+        """Add a message to a conversation."""
+        async with async_session_maker() as session:
+            # Get conversation
+            result = await session.execute(
+                select(DBConversation).where(DBConversation.id == conversation_id)
+            )
+            db_conv = result.scalar_one_or_none()
+            
+            if not db_conv:
+                raise ValueError(f"Conversation {conversation_id} not found")
+            
+            # Create message
+            now = datetime.utcnow().isoformat() + "Z"
+            db_message = DBMessage(
+                conversation_id=conversation_id,
+                role=role,
+                content=content,
+                timestamp=now,
+                sources=sources or [],
+            )
+            session.add(db_message)
+            
+            # Update conversation
+            db_conv.updated_at = now
+            
+            # Update title from first user message
+            if role == "user" and db_conv.title == "New Chat":
+                db_conv.title = content[:80] + ("..." if len(content) > 80 else "")
+            
+            await session.commit()
+            
+            return ConversationMessage(
+                role=role,
+                content=content,
+                timestamp=now,
+                sources=sources or [],
+            )
+
+    async def get_conversation(self, conversation_id: str) -> Optional[Conversation]:
+        """Get a conversation by ID."""
+        async with async_session_maker() as session:
+            result = await session.execute(
+                select(DBConversation)
+                .options(selectinload(DBConversation.messages))
+                .where(DBConversation.id == conversation_id)
+            )
+            db_conv = result.scalar_one_or_none()
+            
+            if not db_conv:
+                return None
+            
+            return Conversation(
+                id=db_conv.id,
+                title=db_conv.title,
+                messages=[
+                    ConversationMessage(
+                        role=msg.role,
+                        content=msg.content,
+                        timestamp=msg.timestamp,
+                        sources=msg.sources or [],
+                    )
+                    for msg in db_conv.messages
+                ],
+                created_at=db_conv.created_at,
+                updated_at=db_conv.updated_at,
+            )
+
+    async def get_history(self, conversation_id: str, limit: int = 10) -> List[dict]:
+        """Get recent message history for a conversation as list of dicts."""
+        async with async_session_maker() as session:
+            result = await session.execute(
+                select(DBMessage)
+                .where(DBMessage.conversation_id == conversation_id)
+                .order_by(desc(DBMessage.id))
+                .limit(limit)
+            )
+            messages = result.scalars().all()
+            
+            # Reverse to get chronological order
+            return [
+                {"role": msg.role, "content": msg.content}
+                for msg in reversed(messages)
+            ]
+
+    async def list_conversations(self) -> List[ConversationListItem]:
+        """List all conversations."""
+        async with async_session_maker() as session:
+            result = await session.execute(
+                select(DBConversation)
+                .options(selectinload(DBConversation.messages))
+                .order_by(desc(DBConversation.updated_at))
+            )
+            conversations = result.scalars().all()
+            
+            return [
+                ConversationListItem(
+                    id=conv.id,
+                    title=conv.title,
+                    message_count=len(conv.messages),
+                    created_at=conv.created_at,
+                    updated_at=conv.updated_at,
+                )
+                for conv in conversations
+            ]
+
+    async def delete_conversation(self, conversation_id: str) -> bool:
+        """Delete a conversation."""
+        async with async_session_maker() as session:
+            result = await session.execute(
+                select(DBConversation).where(DBConversation.id == conversation_id)
+            )
+            db_conv = result.scalar_one_or_none()
+            
+            if db_conv:
+                await session.delete(db_conv)
+                await session.commit()
+                return True
+            return False
+
+    async def clear_all(self):
+        """Clear all conversations."""
+        async with async_session_maker() as session:
+            await session.execute(select(DBConversation))
+            result = await session.execute(select(DBConversation))
+            conversations = result.scalars().all()
+            
+            for conv in conversations:
+                await session.delete(conv)
+            
+            await session.commit()
+
+
+conversation_store = ConversationStore()

app/services/document_processor.py

@@ -0,0 +1,92 @@
+"""Document processing service for PDF, CSV, and Markdown files."""
+
+import os
+from typing import List, Optional
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain_community.document_loaders import (
+    PyPDFLoader,
+    CSVLoader,
+    TextLoader,
+)
+from langchain.schema import Document
+from app.config import settings
+
+
+class DocumentProcessor:
+    """Handles loading and chunking of various document types."""
+
+    def __init__(self):
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=settings.CHUNK_SIZE,
+            chunk_overlap=settings.CHUNK_OVERLAP,
+            length_function=len,
+            separators=["\n\n", "\n", ". ", " ", ""],
+        )
+
+        self.supported_extensions = {
+            ".pdf": self._load_pdf,
+            ".csv": self._load_csv,
+            ".md": self._load_markdown,
+            ".txt": self._load_text,
+        }
+
+    def get_supported_extensions(self) -> List[str]:
+        return list(self.supported_extensions.keys())
+
+    async def process_file(
+        self, file_path: str, metadata: Optional[dict] = None
+    ) -> List[Document]:
+        """Process a file and return chunked documents with metadata."""
+        ext = os.path.splitext(file_path)[1].lower()
+
+        if ext not in self.supported_extensions:
+            raise ValueError(
+                f"Unsupported file type: {ext}. "
+                f"Supported: {', '.join(self.supported_extensions.keys())}"
+            )
+
+        # Load documents
+        loader_fn = self.supported_extensions[ext]
+        documents = loader_fn(file_path)
+
+        # Add custom metadata
+        filename = os.path.basename(file_path)
+        for doc in documents:
+            doc.metadata["source"] = filename
+            doc.metadata["file_type"] = ext
+            if metadata:
+                doc.metadata.update(metadata)
+
+        # Split into chunks
+        chunks = self.text_splitter.split_documents(documents)
+
+        # Add chunk IDs
+        for i, chunk in enumerate(chunks):
+            chunk.metadata["chunk_id"] = f"{filename}_chunk_{i}"
+            chunk.metadata["chunk_index"] = i
+            chunk.metadata["total_chunks"] = len(chunks)
+
+        return chunks
+
+    def _load_pdf(self, file_path: str) -> List[Document]:
+        """Load PDF file."""
+        loader = PyPDFLoader(file_path)
+        return loader.load()
+
+    def _load_csv(self, file_path: str) -> List[Document]:
+        """Load CSV file."""
+        loader = CSVLoader(file_path, encoding="utf-8")
+        return loader.load()
+
+    def _load_markdown(self, file_path: str) -> List[Document]:
+        """Load Markdown file."""
+        loader = TextLoader(file_path, encoding="utf-8")
+        return loader.load()
+
+    def _load_text(self, file_path: str) -> List[Document]:
+        """Load plain text file."""
+        loader = TextLoader(file_path, encoding="utf-8")
+        return loader.load()
+
+
+document_processor = DocumentProcessor()

app/services/rag_chain.py

@@ -0,0 +1,204 @@
+"""RAG chain service using LangChain with Gemini."""
+
+import json
+from typing import AsyncGenerator, List, Optional
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain.schema import Document, HumanMessage, AIMessage
+from app.config import settings
+from app.services.vector_store import vector_store_service
+from app.services.conversation_store import conversation_store
+from app.models import SourceDocument
+
+import asyncio
+
+
+SYSTEM_PROMPT = """You are an intelligent AI assistant for a health-tech knowledge base. \
+Your role is to answer questions accurately based on the provided context from our documents.
+
+INSTRUCTIONS:
+- Answer questions based ONLY on the provided context below.
+- When listing items (like "pillars", "steps", "types"), make sure to include ALL items found in the context.
+- If the context doesn't contain relevant information, say "I don't have enough information in the knowledge base to answer that question."
+- Be thorough and complete - don't skip information that's present in the context.
+- When referencing information, mention the source document when possible.
+- Format your answers with proper markdown for readability.
+- If asked about topics outside the knowledge base, politely redirect to relevant topics.
+
+CONTEXT FROM KNOWLEDGE BASE:
+{context}
+"""
+
+
+class RAGService:
+    """Retrieval-Augmented Generation service using Gemini."""
+
+    def __init__(self):
+        self.llm: Optional[ChatGoogleGenerativeAI] = None
+
+    def _get_llm(self, streaming: bool = False) -> ChatGoogleGenerativeAI:
+        """Get or create Gemini LLM instance."""
+        return ChatGoogleGenerativeAI(
+            model=settings.LLM_MODEL,
+            google_api_key=settings.GOOGLE_API_KEY,
+            temperature=0.3,
+            streaming=streaming,
+        )
+
+    def _build_prompt(self) -> ChatPromptTemplate:
+        """Build the RAG prompt template."""
+        return ChatPromptTemplate.from_messages(
+            [
+                ("system", SYSTEM_PROMPT),
+                MessagesPlaceholder(variable_name="chat_history"),
+                ("human", "{question}"),
+            ]
+        )
+
+    def _format_context(self, documents: List[Document]) -> str:
+        """Format retrieved documents into context string."""
+        if not documents:
+            return "No relevant documents found."
+
+        context_parts = []
+        for i, doc in enumerate(documents, 1):
+            source = doc.metadata.get("source", "Unknown")
+            page = doc.metadata.get("page", "")
+            page_str = f" (Page {page})" if page else ""
+            context_parts.append(
+                f"[Source {i}: {source}{page_str}]\n{doc.page_content}"
+            )
+
+        return "\n\n---\n\n".join(context_parts)
+
+    def _format_sources(self, documents: List[Document]) -> List[SourceDocument]:
+        """Convert retrieved documents to SourceDocument models."""
+        sources = []
+        seen = set()
+        for doc in documents:
+            source_key = (doc.metadata.get("source", ""), doc.metadata.get("page"))
+            if source_key not in seen:
+                seen.add(source_key)
+                sources.append(
+                    SourceDocument(
+                        content=doc.page_content[:300],
+                        source=doc.metadata.get("source", "Unknown"),
+                        page=doc.metadata.get("page"),
+                        chunk_id=doc.metadata.get("chunk_id"),
+                    )
+                )
+        return sources
+
+    def _build_chat_history(self, history: List[dict]) -> list:
+        """Convert conversation history to LangChain message format."""
+        messages = []
+        for msg in history:
+            if msg["role"] == "user":
+                messages.append(HumanMessage(content=msg["content"]))
+            elif msg["role"] == "assistant":
+                messages.append(AIMessage(content=msg["content"]))
+        return messages
+
+    async def ask(
+        self,
+        question: str,
+        conversation_id: Optional[str] = None,
+    ) -> dict:
+        """Ask a question with RAG (non-streaming)."""
+        # Get or create conversation
+        conversation = await conversation_store.get_or_create(conversation_id)
+        conv_id = conversation.id
+
+        # Store user message
+        await conversation_store.add_message(conv_id, "user", question)
+
+        # Retrieve relevant documents
+        documents = await vector_store_service.similarity_search(question, k=6)
+
+        # Build context and history
+        context = self._format_context(documents)
+        chat_history = self._build_chat_history(
+            await conversation_store.get_history(conv_id, limit=8)
+        )
+
+        # Build and invoke chain
+        llm = self._get_llm(streaming=False)
+        prompt = self._build_prompt()
+        chain = prompt | llm
+
+        response = await chain.ainvoke(
+            {
+                "context": context,
+                "chat_history": chat_history[:-1],  # Exclude current question
+                "question": question,
+            }
+        )
+
+        answer = response.content
+        sources = self._format_sources(documents)
+
+        # Store assistant response
+        await conversation_store.add_message(conv_id, "assistant", answer, sources)
+
+        return {
+            "answer": answer,
+            "conversation_id": conv_id,
+            "sources": sources,
+        }
+
+    async def ask_stream(
+        self,
+        question: str,
+        conversation_id: Optional[str] = None,
+    ) -> AsyncGenerator[str, None]:
+        """Ask a question with RAG (streaming via SSE)."""
+        # Get or create conversation
+        conversation = await conversation_store.get_or_create(conversation_id)
+        conv_id = conversation.id
+
+        # Store user message
+        await conversation_store.add_message(conv_id, "user", question)
+
+        # Retrieve relevant documents
+        documents = await vector_store_service.similarity_search(question, k=6)
+
+        # Build context and history
+        context = self._format_context(documents)
+        chat_history = self._build_chat_history(
+            await conversation_store.get_history(conv_id, limit=8)
+        )
+        sources = self._format_sources(documents)
+
+        # Send initial event with conversation ID and sources
+        yield f"data: {json.dumps({'type': 'metadata', 'conversation_id': conv_id, 'sources': [s.model_dump() for s in sources]})}\n\n"
+
+        # Stream the response using Gemini
+        llm = self._get_llm(streaming=True)
+        prompt = self._build_prompt()
+        chain = prompt | llm
+
+        full_response = []
+
+        try:
+            async for chunk in chain.astream(
+                {
+                    "context": context,
+                    "chat_history": chat_history[:-1],
+                    "question": question,
+                }
+            ):
+                if hasattr(chunk, 'content') and chunk.content:
+                    full_response.append(chunk.content)
+                    yield f"data: {json.dumps({'type': 'token', 'content': chunk.content})}\n\n"
+        except Exception as e:
+            yield f"data: {json.dumps({'type': 'error', 'content': str(e)})}\n\n"
+
+        # Store complete response
+        complete_answer = "".join(full_response)
+        await conversation_store.add_message(conv_id, "assistant", complete_answer, sources)
+
+        # Send done event
+        yield f"data: {json.dumps({'type': 'done'})}\n\n"
+
+
+rag_service = RAGService()

app/services/vector_store.py

@@ -0,0 +1,93 @@
+"""Vector store service using Pinecone."""
+
+from typing import List, Optional
+from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_pinecone import PineconeVectorStore
+from langchain.schema import Document
+from pinecone import Pinecone, ServerlessSpec
+from app.config import settings
+
+
+class VectorStoreService:
+    """Manages Pinecone vector store operations."""
+
+    def __init__(self):
+        self.embeddings: Optional[HuggingFaceEmbeddings] = None
+        self.vector_store: Optional[PineconeVectorStore] = None
+        self.pc: Optional[Pinecone] = None
+        self._initialized = False
+
+    async def initialize(self):
+        """Initialize Pinecone and embeddings."""
+        if self._initialized:
+            return
+
+        # Initialize HuggingFace embeddings
+        self.embeddings = HuggingFaceEmbeddings(
+            model_name=settings.EMBEDDING_MODEL,
+            model_kwargs={'device': 'cpu'},
+            encode_kwargs={'normalize_embeddings': True}
+        )
+
+        # Initialize Pinecone
+        self.pc = Pinecone(api_key=settings.PINECONE_API_KEY)
+
+        # Create index if it doesn't exist
+        existing_indexes = [idx.name for idx in self.pc.list_indexes()]
+        if settings.PINECONE_INDEX_NAME not in existing_indexes:
+            self.pc.create_index(
+                name=settings.PINECONE_INDEX_NAME,
+                dimension=384,  # all-MiniLM-L6-v2 dimension
+                metric="cosine",
+                spec=ServerlessSpec(cloud="aws", region="us-east-1"),
+            )
+
+        # Initialize vector store
+        self.vector_store = PineconeVectorStore(
+            index_name=settings.PINECONE_INDEX_NAME,
+            embedding=self.embeddings,
+            pinecone_api_key=settings.PINECONE_API_KEY,
+        )
+
+        self._initialized = True
+
+    async def add_documents(self, documents: List[Document]) -> int:
+        """Add documents to the vector store."""
+        if not self._initialized:
+            await self.initialize()
+
+        self.vector_store.add_documents(documents)
+        return len(documents)
+
+    async def similarity_search(
+        self, query: str, k: int = 4
+    ) -> List[Document]:
+        """Search for similar documents."""
+        if not self._initialized:
+            await self.initialize()
+
+        results = self.vector_store.similarity_search(query, k=k)
+        return results
+
+    async def similarity_search_with_score(
+        self, query: str, k: int = 4
+    ) -> List[tuple]:
+        """Search for similar documents with relevance scores."""
+        if not self._initialized:
+            await self.initialize()
+
+        results = self.vector_store.similarity_search_with_score(query, k=k)
+        return results
+
+    def get_retriever(self, k: int = 4):
+        """Get a retriever for use in chains."""
+        if not self._initialized:
+            raise RuntimeError("Vector store not initialized")
+
+        return self.vector_store.as_retriever(
+            search_type="similarity",
+            search_kwargs={"k": k},
+        )
+
+
+vector_store_service = VectorStoreService()

requirements.txt

@@ -0,0 +1,21 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+python-dotenv==1.0.1
+langchain==0.3.14
+langchain-core==0.3.14
+langchain-google-genai==2.0.8
+langchain-community==0.3.14
+langchain-pinecone==0.2.0
+langchain-huggingface==0.1.0
+sentence-transformers==3.3.1
+pinecone-client==5.0.1
+google-generativeai==0.8.3
+pypdf==5.1.0
+python-multipart==0.0.20
+pydantic==2.10.4
+pydantic-settings==2.7.1
+unstructured==0.16.12
+aiofiles==24.1.0
+uuid6==2024.7.10
+sqlalchemy==2.0.25
+aiosqlite==0.19.0