API Reference

Last Updated: 2025-12-02

The VoiceAssist API provides comprehensive REST endpoints for building medical AI assistant applications.

Documentation

• Complete REST API Reference - Full endpoint documentation with examples
• OpenAPI/Swagger UI - Interactive docs at http://localhost:8000/docs
• ReDoc - Alternative docs at http://localhost:8000/redoc

Quick Reference

Base URLs

• Production: https://assist.asimo.io
• Development: http://localhost:8000

Authentication

All authenticated endpoints require a Bearer token:

Authorization: Bearer <access_token>

Core Endpoint Groups

OpenAPI Specification

• Swagger UI: http://localhost:8000/docs (interactive API explorer)
• ReDoc: http://localhost:8000/redoc (alternative documentation)
• OpenAPI JSON: http://localhost:8000/openapi.json (downloadable spec)

For complete documentation with request/response examples, see api-reference/rest-api.md.

Documentation Coverage Status

As of 2025-12-02, API documentation coverage:

Priority endpoints to document:

1. Core voice endpoints (/api/voice/*) - Most user-facing
2. Authentication 2FA (/api/auth/2fa/*) - Security-critical
3. Admin panel endpoints (/api/admin/panel/*) - Ops-critical
4. Integration endpoints (/api/integrations/*) - Feature expansion

Run pnpm validate:api-sync in apps/docs-site/ to regenerate coverage stats.

Documentation API (For AI Agents)

The docs site provides machine-readable JSON endpoints for AI agents:

Static JSON Endpoints

Base URL: https://assistdocs.asimo.io

Note: All endpoints are static JSON. Use search-index.json with client-side Fuse.js for full-text search.

AI-Docs Semantic Search (Qdrant)

For semantic/vector search, documentation is embedded into Qdrant:

Tool Functions (for AI agents):

# Semantic search across documentation
docs_search(query: str, category: str = None, max_results: int = 5)

# Retrieve full section content
docs_get_section(doc_path: str, section: str = None)

Re-indexing Documentation:

python scripts/embed-docs.py            # Incremental update
python scripts/embed-docs.py --force    # Force re-index all

For full details, see:

• Agent API Reference - JSON endpoints
• Agent Onboarding - AI-Docs integration
• Internal Docs System - Embedding workflow

VoiceAssist REST API Reference

Version: 2.0 Base URL: https://assist.asimo.io/api (production) or http://localhost:8000/api (development) Last Updated: 2025-11-27

See Also: Auto-generated API Routes - Complete route listing from OpenAPI spec

Overview

The VoiceAssist API provides a comprehensive set of endpoints for building medical AI assistant applications. All endpoints follow REST conventions and return JSON responses wrapped in a standard envelope format.

Authentication

Most endpoints require JWT authentication. Include the access token in the Authorization header:

Authorization: Bearer <access_token>

Response Envelope

All responses follow this structure:

{
  "success": true,
  "data": { ... },
  "error": null,
  "meta": {
    "request_id": "uuid",
    "timestamp": "ISO-8601"
  }
}

Rate Limiting

Rate limits are applied per IP address. Headers returned:

• X-RateLimit-Limit: Maximum requests allowed
• X-RateLimit-Remaining: Requests remaining
• X-RateLimit-Reset: Unix timestamp when limit resets

Authentication (/api/auth)

Register User

POST /api/auth/register

Create a new user account.

Rate Limit: 5/hour per IP

Request Body:

{
  "email": "user@example.com",
  "password": "securePassword123",
  "full_name": "John Doe"
}

Response (201):

{
  "id": "uuid",
  "email": "user@example.com",
  "full_name": "John Doe",
  "is_active": true,
  "is_admin": false,
  "created_at": "2025-11-27T10:00:00Z"
}

Login

POST /api/auth/login

Authenticate and receive JWT tokens.

Rate Limit: 10/minute per IP

Request Body:

{
  "email": "user@example.com",
  "password": "securePassword123"
}

Response (200):

{
  "access_token": "eyJhbG...",
  "refresh_token": "eyJhbG...",
  "token_type": "bearer",
  "expires_in": 300
}

Refresh Token

POST /api/auth/refresh

Get a new access token using a refresh token.

Request Body:

{
  "refresh_token": "eyJhbG..."
}

Response (200):

{
  "access_token": "eyJhbG...",
  "refresh_token": "eyJhbG...",
  "token_type": "bearer",
  "expires_in": 300
}

Logout

POST /api/auth/logout

Revoke current tokens.

Headers: Authorization: Bearer <access_token>

Response (200):

{
  "message": "Successfully logged out"
}

Get Current User

GET /api/auth/me

Get authenticated user's information.

Headers: Authorization: Bearer <access_token>

Response (200):

{
  "id": "uuid",
  "email": "user@example.com",
  "full_name": "John Doe",
  "is_active": true,
  "is_admin": false,
  "admin_role": "user",
  "created_at": "2025-11-27T10:00:00Z",
  "last_login": "2025-11-27T12:00:00Z"
}

User Management (/api/users)

Get Current User Profile

GET /api/users/me

Headers: Authorization: Bearer <access_token>

Update Current User Profile

PUT /api/users/me

Headers: Authorization: Bearer <access_token>

Request Body:

{
  "full_name": "John Updated",
  "email": "newemail@example.com"
}

Change Password

POST /api/users/me/change-password

Headers: Authorization: Bearer <access_token>

Request Body:

{
  "old_password": "currentPassword",
  "new_password": "newSecurePassword123"
}

Delete Account

DELETE /api/users/me

Permanently delete user account.

Headers: Authorization: Bearer <access_token>

List Users (Admin)

GET /api/users?offset=0&limit=20

Headers: Authorization: Bearer <admin_token>

Query Parameters:

• offset (int): Pagination offset (default: 0)
• limit (int): Page size (default: 20, max: 100)

Get User by ID (Admin)

GET /api/users/{user_id}

Headers: Authorization: Bearer <admin_token>

Update User (Admin)

PATCH /api/users/{user_id}

Headers: Authorization: Bearer <admin_token>

Activate User (Admin)

PUT /api/users/{user_id}/activate

Headers: Authorization: Bearer <admin_token>

Deactivate User (Admin)

PUT /api/users/{user_id}/deactivate

Headers: Authorization: Bearer <admin_token>

Promote to Admin (Admin)

PUT /api/users/{user_id}/promote-admin

Headers: Authorization: Bearer <admin_token>

Revoke Admin (Admin)

PUT /api/users/{user_id}/revoke-admin

Headers: Authorization: Bearer <admin_token>

Conversations (/conversations)

List Conversations

GET /conversations?page=1&pageSize=20

Headers: Authorization: Bearer <access_token>

Query Parameters:

• page (int): Page number (default: 1)
• pageSize (int): Items per page (default: 20)
• archived (bool): Filter archived conversations
• folderId (string): Filter by folder

Response (200):

{
  "items": [
    {
      "id": "uuid",
      "userId": "uuid",
      "title": "Medical Inquiry",
      "archived": false,
      "messageCount": 5,
      "folderId": null,
      "createdAt": "2025-11-27T10:00:00Z",
      "updatedAt": "2025-11-27T12:00:00Z"
    }
  ],
  "total": 42,
  "page": 1,
  "pageSize": 20
}

Create Conversation

POST /conversations

Headers: Authorization: Bearer <access_token>

Request Body:

{
  "title": "New Medical Discussion",
  "folder_id": "optional-folder-uuid"
}

Get Conversation

GET /conversations/{conversation_id}

Headers: Authorization: Bearer <access_token>

Update Conversation

PUT /conversations/{conversation_id}

Headers: Authorization: Bearer <access_token>

Request Body:

{
  "title": "Updated Title",
  "archived": false,
  "folder_id": "new-folder-uuid"
}

Delete Conversation

DELETE /conversations/{conversation_id}

Headers: Authorization: Bearer <access_token>

Create Branch

POST /conversations/{conversation_id}/branches

Fork a conversation from a specific message.

Headers: Authorization: Bearer <access_token>

Request Body:

{
  "parent_message_id": "message-uuid",
  "initial_message": "Optional first message in branch"
}

List Branches

GET /conversations/{conversation_id}/branches

Headers: Authorization: Bearer <access_token>

Get Messages

GET /conversations/{conversation_id}/messages?branch_id=main

Headers: Authorization: Bearer <access_token>

Query Parameters:

• branch_id (string): Branch identifier (default: "main")
• limit (int): Maximum messages to return

Admin Panel (/api/admin/panel)

Dashboard Summary

GET /api/admin/panel/summary

Get dashboard metrics summary.

Headers: Authorization: Bearer <admin_token>

Response (200):

{
  "users": {
    "total": 150,
    "active": 142,
    "admins": 3
  },
  "conversations": {
    "total": 1250,
    "today": 45
  },
  "system": {
    "status": "healthy",
    "uptime": 864000
  }
}

WebSocket Status

GET /api/admin/panel/websocket-status

Get real-time connection status.

Headers: Authorization: Bearer <admin_token>

List Users

GET /api/admin/panel/users?offset=0&limit=20

Paginated user list with statistics.

Headers: Authorization: Bearer <admin_token>

Get User Details

GET /api/admin/panel/users/{user_id}

Headers: Authorization: Bearer <admin_token>

Update User

PUT /api/admin/panel/users/{user_id}

Headers: Authorization: Bearer <admin_token>

Delete User

DELETE /api/admin/panel/users/{user_id}

Headers: Authorization: Bearer <admin_token>

User Role History

GET /api/admin/panel/users/{user_id}/role-history

Headers: Authorization: Bearer <admin_token>

Account Lock Reasons

GET /api/admin/panel/users/{user_id}/lock-reasons

Headers: Authorization: Bearer <admin_token>

System Metrics

GET /api/admin/panel/metrics

Headers: Authorization: Bearer <admin_token>

Audit Logs

GET /api/admin/panel/audit-logs?page=1&limit=50

Headers: Authorization: Bearer <admin_token>

Query Parameters:

• page (int): Page number
• limit (int): Items per page
• action (string): Filter by action type
• user_id (string): Filter by user

Export Audit Logs

GET /api/admin/panel/audit-logs/export?format=csv

Headers: Authorization: Bearer <admin_token>

Knowledge Base Admin (/api/admin/kb)

List Documents

GET /api/admin/kb/documents?offset=0&limit=20

Headers: Authorization: Bearer <admin_token>

Query Parameters:

• offset (int): Pagination offset
• limit (int): Page size
• status (string): Filter by status (uploaded, processing, indexed, failed)

Upload Document

POST /api/admin/kb/documents
Content-Type: multipart/form-data

Upload a document for indexing.

Headers: Authorization: Bearer <admin_token>

Form Data:

• file: PDF or TXT file (max 50MB)
• title (string): Document title
• category (string): Document category

Get Document Details

GET /api/admin/kb/documents/{document_id}

Headers: Authorization: Bearer <admin_token>

Delete Document

DELETE /api/admin/kb/documents/{document_id}

Headers: Authorization: Bearer <admin_token>

Reindex Document

POST /api/admin/kb/documents/{document_id}/reindex

Headers: Authorization: Bearer <admin_token>

Cache Management (/api/admin/cache)

Cache Statistics

GET /api/admin/cache/stats

Headers: Authorization: Bearer <admin_token>

Response (200):

{
  "redis": {
    "connected": true,
    "used_memory": "15MB",
    "keys": 1250
  },
  "l1_cache": {
    "size": 500,
    "hits": 12500,
    "misses": 250
  }
}

Clear Cache

POST /api/admin/cache/clear

Clear all caches.

Headers: Authorization: Bearer <admin_token>

Invalidate Cache Pattern

POST /api/admin/cache/invalidate

Invalidate specific cache keys.

Headers: Authorization: Bearer <admin_token>

Request Body:

{
  "pattern": "user:*"
}

Feature Flags (/api/admin/feature-flags)

List Feature Flags

GET /api/admin/feature-flags

Headers: Authorization: Bearer <admin_token>

Response (200):

{
  "flags": [
    {
      "name": "voice_mode",
      "enabled": true,
      "description": "Enable voice input mode",
      "rollout_percentage": 100,
      "updated_at": "2025-11-27T10:00:00Z"
    }
  ]
}

Get Feature Flag

GET /api/admin/feature-flags/{flag_name}

Headers: Authorization: Bearer <admin_token>

Create Feature Flag

POST /api/admin/feature-flags

Headers: Authorization: Bearer <admin_token>

Request Body:

{
  "name": "new_feature",
  "enabled": false,
  "description": "Description of the feature",
  "rollout_percentage": 0
}

Update Feature Flag

PATCH /api/admin/feature-flags/{flag_name}

Headers: Authorization: Bearer <admin_token>

Request Body:

{
  "enabled": true,
  "rollout_percentage": 50
}

Delete Feature Flag

DELETE /api/admin/feature-flags/{flag_name}

Headers: Authorization: Bearer <admin_token>

Toggle Feature Flag

POST /api/admin/feature-flags/{flag_name}/toggle

Quick toggle for feature flags.

Headers: Authorization: Bearer <admin_token>

Health Checks

Liveness Check

GET /health

Basic health check. Returns 200 if service is running.

Rate Limit: 100/minute

Response (200):

{
  "status": "healthy",
  "version": "2.0.0",
  "timestamp": 1732703400.123
}

Readiness Check

GET /ready

Checks all dependencies (PostgreSQL, Redis, Qdrant, Nextcloud).

Rate Limit: 100/minute

Response (200):

{
  "status": "ready",
  "checks": {
    "postgres": true,
    "redis": true,
    "qdrant": true,
    "nextcloud": true
  },
  "timestamp": 1732703400.123
}

Response (503): If any dependency is unavailable.

Metrics

Prometheus Metrics

GET /metrics

Prometheus-formatted metrics for monitoring.

Voice Endpoints (/api/voice)

Start Voice Session

POST /api/voice/session

Initialize a voice interaction session.

Headers: Authorization: Bearer <access_token>

Send Audio

POST /api/voice/audio
Content-Type: multipart/form-data

Send audio for transcription and processing.

Headers: Authorization: Bearer <access_token>

WebSocket Realtime (/ws)

Connect

WS /ws?token=<access_token>

Establish WebSocket connection for real-time chat.

Events

Client → Server:

• message_send: Send a new message
• typing_start: Start typing indicator
• typing_stop: Stop typing indicator

Server → Client:

• connected: Connection established
• chunk: Streaming response chunk
• message_done: Complete message with citations
• error: Error notification

Error Codes

OpenAPI Documentation

Interactive API documentation is available at:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

Last Updated: 2025-11-27

Semantic Search & Knowledge Base Design

Overview

VoiceAssist uses a sophisticated semantic search system to retrieve relevant medical knowledge from textbooks, journals, and clinical guidelines. This document describes the complete ingestion and query pipeline.

Note: For canonical entity definitions (KnowledgeDocument, KBChunk, IndexingJob), see DATA_MODEL.md. This document describes their usage in the search pipeline.

Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│                     INGESTION PIPELINE                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  [PDF Upload] → [Text Extraction] → [Chunking] → [Embedding]   │
│                        ↓                  ↓            ↓         │
│                   [OCR/Parse]      [Metadata]    [Vectors]      │
│                        ↓                  ↓            ↓         │
│                 [Preprocessing]     [Enrichment] [Indexing]     │
│                                                        ↓         │
│                                              [Qdrant Vector DB]  │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                      QUERY PIPELINE                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  [User Query] → [Intent Detection] → [Query Enhancement]        │
│                                              ↓                   │
│                                    [Vector Search (Hybrid)]      │
│                                    - Dense vectors (Qdrant)      │
│                                    - Sparse (BM25) fallback      │
│                                              ↓                   │
│                                    [Post-processing]             │
│                                    - Reranking                   │
│                                    - Deduplication               │
│                                    - Filtering                   │
│                                              ↓                   │
│                                    [Top-K Results]               │
│                                              ↓                   │
│                                    [RAG: LLM + Citations]        │
└─────────────────────────────────────────────────────────────────┘

Ingestion Pipeline

1. Document Upload & Storage

Process:

1. User uploads PDF/DOCX via admin panel
2. File saved to /data/documents/{doc_id}/{filename}
3. Document record created in PostgreSQL with status uploaded
4. Background task triggered for processing

Python Implementation:

# app/services/medical/document_processor.py
from pathlib import Path
import hashlib
from typing import Optional
from sqlalchemy.orm import Session
from app.models.document import Document
from app.core.config import settings

class DocumentUploader:
    def __init__(self, db: Session):
        self.db = db
        self.storage_path = Path(settings.DOCUMENT_STORAGE_PATH)

    async def upload_document(
        self,
        file_data: bytes,
        filename: str,
        source_type: str,
        specialty: str,
        metadata: Optional[dict] = None
    ) -> Document:
        """
        Upload and store a document, returning Document model.
        """
        # Generate document ID from hash
        file_hash = hashlib.sha256(file_data).hexdigest()
        doc_id = file_hash[:16]

        # Check if already exists
        existing = self.db.query(Document).filter(
            Document.file_hash == file_hash
        ).first()

        if existing:
            return existing

        # Create storage directory
        doc_dir = self.storage_path / doc_id
        doc_dir.mkdir(parents=True, exist_ok=True)

        # Save file
        file_path = doc_dir / filename
        with open(file_path, 'wb') as f:
            f.write(file_data)

        # Create database record
        document = Document(
            id=doc_id,
            filename=filename,
            file_path=str(file_path),
            file_hash=file_hash,
            file_size=len(file_data),
            source_type=source_type,
            specialty=specialty,
            status='uploaded',
            metadata=metadata or {}
        )

        self.db.add(document)
        self.db.commit()
        self.db.refresh(document)

        return document

2. Text Extraction

Supported Formats:

• PDF (native text extraction + OCR fallback)
• DOCX (python-docx)
• HTML (BeautifulSoup)
• Plain text

Libraries:

• PyPDF2: Fast PDF text extraction
• pdfplumber: Better table/structure handling
• Tesseract OCR: For scanned documents
• python-docx: DOCX extraction

Python Implementation:

# app/services/medical/text_extractor.py
import io
from typing import List, Dict
import PyPDF2
import pdfplumber
from pdf2image import convert_from_path
import pytesseract
from PIL import Image

class TextExtractor:
    """
    Extract text from various document formats.
    """

    def extract_from_pdf(self, file_path: str) -> List[Dict[str, str]]:
        """
        Extract text from PDF, returning list of pages.
        Falls back to OCR if native extraction fails.
        """
        pages = []

        try:
            # Try native text extraction first
            with pdfplumber.open(file_path) as pdf:
                for page_num, page in enumerate(pdf.pages, start=1):
                    text = page.extract_text()

                    # If page has little text, try OCR
                    if len(text.strip()) < 100:
                        text = self._ocr_page(file_path, page_num)

                    pages.append({
                        'page': page_num,
                        'text': text,
                        'width': page.width,
                        'height': page.height
                    })
        except Exception as e:
            print(f"Native extraction failed, falling back to OCR: {e}")
            pages = self._ocr_entire_pdf(file_path)

        return pages

    def _ocr_page(self, file_path: str, page_num: int) -> str:
        """
        OCR a single page using Tesseract.
        """
        images = convert_from_path(
            file_path,
            first_page=page_num,
            last_page=page_num,
            dpi=300
        )

        if images:
            text = pytesseract.image_to_string(images[0])
            return text

        return ""

    def _ocr_entire_pdf(self, file_path: str) -> List[Dict[str, str]]:
        """
        OCR entire PDF.
        """
        images = convert_from_path(file_path, dpi=300)
        pages = []

        for page_num, image in enumerate(images, start=1):
            text = pytesseract.image_to_string(image)
            pages.append({
                'page': page_num,
                'text': text
            })

        return pages

3. Text Preprocessing & Cleaning

Steps:

1. Remove headers/footers (page numbers, running headers)
2. Fix encoding issues
3. Normalize whitespace
4. Remove references section (if at end)
5. Preserve medical formatting (units, dosages)

Python Implementation:

# app/services/medical/text_preprocessor.py
import re
from typing import List, Dict

class TextPreprocessor:
    """
    Clean and normalize extracted text.
    """

    def __init__(self):
        # Common medical textbook footer patterns
        self.footer_patterns = [
            r'^\d+\s*$',  # Page numbers
            r'^Chapter \d+.*$',
            r'^Copyright \d{4}.*$'
        ]

    def preprocess_pages(self, pages: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """
        Preprocess extracted pages.
        """
        cleaned_pages = []

        for page in pages:
            text = page['text']

            # Remove footers/headers
            text = self._remove_footers(text)

            # Fix common encoding issues
            text = self._fix_encoding(text)

            # Normalize whitespace
            text = self._normalize_whitespace(text)

            # Preserve medical formatting
            text = self._preserve_medical_units(text)

            cleaned_pages.append({
                **page,
                'text': text,
                'original_text': page['text']  # Keep original
            })

        return cleaned_pages

    def _remove_footers(self, text: str) -> str:
        """Remove common footer patterns."""
        lines = text.split('\n')
        cleaned_lines = []

        for line in lines:
            is_footer = False
            for pattern in self.footer_patterns:
                if re.match(pattern, line.strip()):
                    is_footer = True
                    break

            if not is_footer:
                cleaned_lines.append(line)

        return '\n'.join(cleaned_lines)

    def _fix_encoding(self, text: str) -> str:
        """Fix common encoding issues."""
        replacements = {
            '�': '',  # Common replacement character
            '’': "'",
            '“': '"',
            'â€': '"',
            'â€"': '—',
        }

        for old, new in replacements.items():
            text = text.replace(old, new)

        return text

    def _normalize_whitespace(self, text: str) -> str:
        """Normalize whitespace while preserving structure."""
        # Replace multiple spaces with single space
        text = re.sub(r' +', ' ', text)

        # Remove spaces before punctuation
        text = re.sub(r' ([,.;:!?])', r'\1', text)

        # Normalize newlines (max 2 consecutive)
        text = re.sub(r'\n{3,}', '\n\n', text)

        return text.strip()

    def _preserve_medical_units(self, text: str) -> str:
        """Ensure medical units and dosages are properly formatted."""
        # Ensure space before units
        text = re.sub(r'(\d+)(mg|mcg|g|kg|mL|L)', r'\1 \2', text)

        # Preserve blood pressure format
        text = re.sub(r'(\d+) / (\d+)', r'\1/\2', text)

        return text

4. Semantic Chunking

Strategy: Semantic chunking with overlap

Parameters:

• Chunk size: 500 tokens (~400 words)
• Overlap: 100 tokens (20%)
• Max chunk size: 750 tokens
• Min chunk size: 200 tokens

Chunking Methods:

1. Sentence-based: Split on sentence boundaries
2. Heading-aware: Keep sections together when possible
3. Table/figure extraction: Handle structured content separately

Python Implementation:

# app/services/medical/chunker.py
from typing import List, Dict
import tiktoken
import re

class SemanticChunker:
    """
    Create semantic chunks from text with intelligent splitting.
    """

    def __init__(self, chunk_size: int = 500, overlap: int = 100):
        self.chunk_size = chunk_size
        self.overlap = overlap
        self.encoder = tiktoken.get_encoding("cl100k_base")

        # Medical section heading patterns
        self.heading_patterns = [
            r'^#+\s+',  # Markdown headings
            r'^[A-Z][A-Z\s]+$',  # ALL CAPS headings
            r'^\d+\.\d+',  # Numbered sections
        ]

    def chunk_document(
        self,
        pages: List[Dict[str, str]],
        metadata: Dict
    ) -> List[Dict]:
        """
        Chunk document into semantic segments.
        """
        chunks = []
        current_chunk = ""
        current_tokens = 0
        chunk_id = 0

        for page in pages:
            sentences = self._split_sentences(page['text'])

            for sentence in sentences:
                sentence_tokens = len(self.encoder.encode(sentence))

                # Check if adding sentence exceeds chunk size
                if current_tokens + sentence_tokens > self.chunk_size:
                    # Save current chunk
                    if current_chunk:
                        chunks.append(self._create_chunk(
                            chunk_id,
                            current_chunk,
                            page['page'],
                            metadata
                        ))
                        chunk_id += 1

                    # Start new chunk with overlap
                    current_chunk = self._get_overlap_text(current_chunk)
                    current_tokens = len(self.encoder.encode(current_chunk))

                # Add sentence to current chunk
                current_chunk += " " + sentence
                current_tokens += sentence_tokens

        # Add final chunk
        if current_chunk:
            chunks.append(self._create_chunk(
                chunk_id,
                current_chunk,
                pages[-1]['page'],
                metadata
            ))

        return chunks

    def _split_sentences(self, text: str) -> List[str]:
        """
        Split text into sentences, handling medical abbreviations.
        """
        # Protect common medical abbreviations from sentence splitting
        protected_abbrevs = [
            'Dr.', 'Mr.', 'Mrs.', 'Ms.',
            'e.g.', 'i.e.', 'et al.', 'vs.',
            'Fig.', 'Ref.', 'Vol.', 'No.'
        ]

        text_protected = text
        for abbrev in protected_abbrevs:
            text_protected = text_protected.replace(abbrev, abbrev.replace('.', '<DOT>'))

        # Split on sentence boundaries
        sentences = re.split(r'[.!?]+\s+', text_protected)

        # Restore abbreviations
        sentences = [s.replace('<DOT>', '.') for s in sentences]

        return [s.strip() for s in sentences if s.strip()]

    def _get_overlap_text(self, text: str) -> str:
        """
        Get last `overlap` tokens from text for next chunk.
        """
        tokens = self.encoder.encode(text)

        if len(tokens) <= self.overlap:
            return text

        overlap_tokens = tokens[-self.overlap:]
        return self.encoder.decode(overlap_tokens)

    def _create_chunk(
        self,
        chunk_id: int,
        text: str,
        page_num: int,
        metadata: Dict
    ) -> Dict:
        """
        Create chunk dictionary with metadata.
        """
        return {
            'chunk_id': chunk_id,
            'text': text.strip(),
            'page': page_num,
            'tokens': len(self.encoder.encode(text)),
            'metadata': {
                **metadata,
                'page': page_num
            }
        }

5. Embedding Generation

Embedding Model: OpenAI text-embedding-3-large

Specifications:

• Dimensions: 3072 (can be reduced to 1024/512 for efficiency)
• Max input: 8191 tokens
• Cost: $0.13 per 1M tokens
• Performance: MTEB score 64.6

Alternative Models:

• Local: sentence-transformers/all-MiniLM-L6-v2 (384 dim)
• Local Medical: microsoft/BiomedNLP-PubMedBERT-base (768 dim)

Python Implementation:

# app/services/medical/embeddings.py
from typing import List, Dict
import openai
from tenacity import retry, stop_after_attempt, wait_exponential
import numpy as np

class EmbeddingGenerator:
    """
    Generate embeddings for text chunks.
    """

    def __init__(self, model: str = "text-embedding-3-large", dimensions: int = 3072):
        self.model = model
        self.dimensions = dimensions
        self.openai_client = openai.OpenAI()

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def embed_chunks(self, chunks: List[Dict]) -> List[Dict]:
        """
        Generate embeddings for list of chunks.
        Uses batch processing for efficiency.
        """
        # Extract texts
        texts = [chunk['text'] for chunk in chunks]

        # Generate embeddings in batches
        batch_size = 100
        all_embeddings = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]

            response = self.openai_client.embeddings.create(
                model=self.model,
                input=batch,
                dimensions=self.dimensions
            )

            batch_embeddings = [item.embedding for item in response.data]
            all_embeddings.extend(batch_embeddings)

        # Add embeddings to chunks
        for chunk, embedding in zip(chunks, all_embeddings):
            chunk['embedding'] = embedding
            chunk['embedding_model'] = self.model
            chunk['embedding_dimensions'] = self.dimensions

        return chunks

    async def embed_query(self, query: str) -> List[float]:
        """
        Generate embedding for a single query.
        """
        response = self.openai_client.embeddings.create(
            model=self.model,
            input=query,
            dimensions=self.dimensions
        )

        return response.data[0].embedding

Idempotency & Deduplication

Document Keys

Each KnowledgeDocument has a stable doc_key that serves as the idempotency key:

• Format: {source_type}-{identifier} (e.g., textbook-harrisons-21e-ch252, guideline-cdc-heart-failure-2023)
• Purpose: Prevent duplicate ingestion of the same document
• Uniqueness: Enforced at database level with unique constraint

Examples:

• textbook-harrisons-21e-ch252 - Harrison's 21st edition, Chapter 252
• guideline-cdc-heart-failure-2023 - CDC heart failure guideline (2023 version)
• journal-nejm-2023-12345 - NEJM article with DOI suffix
• note-user123-clinical-note-456 - User-uploaded clinical note

Upsert Behavior

When a document is re-ingested (same doc_key):

1. Check existing document by doc_key
2. If exists:
   • Compare content_hash (SHA-256 of document content)
   • If hash matches: Skip ingestion, return existing KnowledgeDocument.id
   • If hash differs: Create new version, mark old chunks as superseded
3. If not exists: Create new document

Database Schema:

CREATE TABLE knowledge_documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    doc_key VARCHAR(255) UNIQUE NOT NULL,  -- Idempotency key
    content_hash VARCHAR(64) NOT NULL,      -- SHA-256 for change detection
    version INTEGER DEFAULT 1,              -- Increment on update
    superseded_by UUID REFERENCES knowledge_documents(id),  -- Points to newer version
    created_at TIMESTAMP DEFAULT NOW(),
    ...
);

CREATE INDEX idx_doc_key ON knowledge_documents(doc_key);
CREATE INDEX idx_superseded ON knowledge_documents(superseded_by) WHERE superseded_by IS NOT NULL;

Chunk Deduplication

Chunks from the same document share document_id:

CREATE TABLE kb_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID REFERENCES knowledge_documents(id) ON DELETE CASCADE,
    chunk_index INTEGER NOT NULL,      -- Position in document (0, 1, 2, ...)
    content TEXT NOT NULL,
    embedding VECTOR(768),              -- Or dimension of your embedding model
    superseded BOOLEAN DEFAULT FALSE,   -- Mark old chunks when document updated
    ...
);

CREATE INDEX idx_document_chunks ON kb_chunks(document_id, chunk_index);
CREATE INDEX idx_superseded_chunks ON kb_chunks(document_id) WHERE superseded = false;

When document is updated:

1. Set superseded = true on old chunks
2. Create new chunks with superseded = false
3. Old chunks remain for audit but excluded from search

API Example

from hashlib import sha256

async def ingest_document(
    file_path: str,
    doc_key: str,
    source_type: str,
    metadata: dict,
) -> KnowledgeDocument:
    """
    Ingest document with idempotency.

    Returns existing document if content unchanged,
    or new version if content updated.
    """

    # Read file content
    content = read_file(file_path)
    content_hash = sha256(content.encode()).hexdigest()

    # Check for existing document
    existing = await db.query(
        "SELECT * FROM knowledge_documents WHERE doc_key = $1",
        doc_key,
    )

    if existing:
        if existing.content_hash == content_hash:
            # Content unchanged, return existing
            logger.info(f"Document {doc_key} unchanged, skipping ingestion")
            return existing

        # Content changed, create new version
        logger.info(f"Document {doc_key} updated, creating new version")
        new_version = existing.version + 1

        # Mark old chunks as superseded
        await db.execute(
            "UPDATE kb_chunks SET superseded = true WHERE document_id = $1",
            existing.id,
        )
    else:
        new_version = 1

    # Create new document (or version)
    new_doc = await db.insert(
        "knowledge_documents",
        doc_key=doc_key,
        content_hash=content_hash,
        version=new_version,
        source_type=source_type,
        metadata=metadata,
    )

    # Process chunks (extract, embed, store)
    chunks = await process_and_embed(content, new_doc.id)

    return new_doc

6. Vector Database Indexing

Vector DB: Qdrant

Collection Schema:

# app/services/medical/vector_db.py
from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance,
    VectorParams,
    PointStruct,
    Filter,
    FieldCondition,
    MatchValue
)
from typing import List, Dict
from uuid import uuid4

class VectorDatabase:
    """
    Qdrant vector database interface.
    """

    def __init__(self, url: str = "http://localhost:6333"):
        self.client = QdrantClient(url=url)
        self.collection_name = "medical_knowledge"

    def create_collection(self, dimensions: int = 3072):
        """
        Create collection with schema.
        """
        self.client.recreate_collection(
            collection_name=self.collection_name,
            vectors_config=VectorParams(
                size=dimensions,
                distance=Distance.COSINE
            )
        )

        # Create payload indexes for filtering
        self.client.create_payload_index(
            collection_name=self.collection_name,
            field_name="source_type",
            field_schema="keyword"
        )

        self.client.create_payload_index(
            collection_name=self.collection_name,
            field_name="specialty",
            field_schema="keyword"
        )

        self.client.create_payload_index(
            collection_name=self.collection_name,
            field_name="document_id",
            field_schema="keyword"
        )

    async def index_chunks(self, chunks: List[Dict]):
        """
        Index chunks into vector database.
        """
        points = []

        for chunk in chunks:
            point_id = str(uuid4())

            point = PointStruct(
                id=point_id,
                vector=chunk['embedding'],
                payload={
                    # Text content
                    'text': chunk['text'],
                    'chunk_id': chunk['chunk_id'],
                    'page': chunk['page'],
                    'tokens': chunk['tokens'],

                    # Document metadata
                    'document_id': chunk['metadata']['document_id'],
                    'document_title': chunk['metadata']['title'],
                    'source_type': chunk['metadata']['source_type'],
                    'specialty': chunk['metadata']['specialty'],

                    # Source details
                    'authors': chunk['metadata'].get('authors', []),
                    'publication_year': chunk['metadata'].get('publication_year'),
                    'publisher': chunk['metadata'].get('publisher'),
                    'doi': chunk['metadata'].get('doi'),
                    'pmid': chunk['metadata'].get('pmid'),

                    # Indexing metadata
                    'embedding_model': chunk['embedding_model'],
                    'indexed_at': chunk['metadata'].get('indexed_at')
                }
            )

            points.append(point)

        # Upsert in batches
        batch_size = 100
        for i in range(0, len(points), batch_size):
            batch = points[i:i + batch_size]
            self.client.upsert(
                collection_name=self.collection_name,
                points=batch
            )

Payload Schema:

{
  "text": "Atrial fibrillation (AF) is the most common sustained cardiac arrhythmia...",
  "chunk_id": 42,
  "page": 234,
  "tokens": 487,

  "document_id": "a1b2c3d4e5f6",
  "document_title": "Harrison's Principles of Internal Medicine - Chapter 45",
  "source_type": "textbook",
  "specialty": "cardiology",

  "authors": ["Dennis Kasper", "Stephen Hauser"],
  "publication_year": 2022,
  "publisher": "McGraw-Hill",
  "doi": null,
  "pmid": null,

  "embedding_model": "text-embedding-3-large",
  "indexed_at": "2024-11-19T10:34:00Z"
}

Indexing Job State Machine

Each document ingestion creates an IndexingJob that tracks progress through these states:

┌─────────────────────────────────────────────────────────────┐
│                  IndexingJob State Machine                   │
└─────────────────────────────────────────────────────────────┘

         ┌──────────┐
    ────▶│ PENDING  │ (Job created, queued)
         └─────┬────┘
               │
               ▼
         ┌──────────┐
         │ RUNNING  │ (Worker processing)
         └─────┬────┘
               │
         ┌─────┴─────┐
         │           │
         ▼           ▼
    ┌──────────┐  ┌─────────┐
    │COMPLETED │  │ FAILED  │
    └────┬─────┘  └────┬────┘
         │             │
         │             │ (Manual retry)
         │             └──────────┐
         │                        ▼
         │                 ┌──────────┐
         └────────────────▶│SUPERSEDED│ (Newer version ingested)
                           └──────────┘

State Definitions

State Transitions

class IndexingJobState(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SUPERSEDED = "superseded"

class IndexingJob(BaseModel):
    """From DATA_MODEL.md - enhanced with state machine."""
    id: str
    document_id: str
    doc_key: str
    state: IndexingJobState
    created_at: datetime
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    failed_at: Optional[datetime] = None
    error_message: Optional[str] = None
    error_details: Optional[dict] = None
    retry_count: int = 0
    max_retries: int = 3

    # Progress tracking
    total_chunks: Optional[int] = None
    processed_chunks: int = 0

    # Superseded tracking
    superseded_by: Optional[str] = None  # ID of newer job

async def transition_state(
    job_id: str,
    new_state: IndexingJobState,
    error: Optional[Exception] = None,
) -> IndexingJob:
    """Transition job to new state with validation."""

    job = await get_job(job_id)

    # Validate transition
    valid_transitions = {
        IndexingJobState.PENDING: [IndexingJobState.RUNNING, IndexingJobState.FAILED],
        IndexingJobState.RUNNING: [IndexingJobState.COMPLETED, IndexingJobState.FAILED],
        IndexingJobState.COMPLETED: [IndexingJobState.SUPERSEDED],
        IndexingJobState.FAILED: [IndexingJobState.PENDING, IndexingJobState.SUPERSEDED],
        IndexingJobState.SUPERSEDED: [],  # Terminal state
    }

    if new_state not in valid_transitions[job.state]:
        raise ValueError(
            f"Invalid transition: {job.state} → {new_state}"
        )

    # Update job
    job.state = new_state

    if new_state == IndexingJobState.RUNNING:
        job.started_at = datetime.utcnow()
    elif new_state == IndexingJobState.COMPLETED:
        job.completed_at = datetime.utcnow()
    elif new_state == IndexingJobState.FAILED:
        job.failed_at = datetime.utcnow()
        job.error_message = str(error) if error else None
        job.error_details = {"type": type(error).__name__} if error else None

    await db.update("indexing_jobs", job)
    return job

Retry Logic

async def retry_failed_job(job_id: str) -> IndexingJob:
    """Retry a failed indexing job."""

    job = await get_job(job_id)

    if job.state != IndexingJobState.FAILED:
        raise ValueError(f"Can only retry FAILED jobs, got {job.state}")

    if job.retry_count >= job.max_retries:
        raise ValueError(f"Max retries ({job.max_retries}) exceeded")

    job.retry_count += 1
    job.state = IndexingJobState.PENDING
    job.error_message = None
    job.error_details = None

    await db.update("indexing_jobs", job)

    # Re-queue job
    await queue.enqueue(process_indexing_job, job.id)

    return job

Admin API Endpoints

@router.get("/api/admin/kb/jobs", response_model=APIEnvelope)
async def list_indexing_jobs(
    state: Optional[IndexingJobState] = None,
    limit: int = 50,
) -> APIEnvelope:
    """List indexing jobs with optional state filter."""
    jobs = await db.query_jobs(state=state, limit=limit)
    return success_response(data=[job.dict() for job in jobs])

@router.post("/api/admin/kb/jobs/{job_id}/retry", response_model=APIEnvelope)
async def retry_indexing_job(job_id: str) -> APIEnvelope:
    """Retry a failed indexing job."""
    try:
        job = await retry_failed_job(job_id)
        return success_response(data=job.dict())
    except ValueError as e:
        return error_response(
            code="VALIDATION_ERROR",
            message=str(e),
            status_code=422,
        )

Query Pipeline

1. Query Enhancement

Steps:

1. Detect user intent (quick lookup vs deep analysis)
2. Extract medical entities (drugs, conditions, procedures)
3. Expand abbreviations
4. Add specialty context

Python Implementation:

# app/services/medical/query_enhancer.py
from typing import Dict, List
import re

class QueryEnhancer:
    """
    Enhance user queries for better retrieval.
    """

    def __init__(self):
        # Common medical abbreviations
        self.abbreviations = {
            'HTN': 'hypertension',
            'DM': 'diabetes mellitus',
            'CAD': 'coronary artery disease',
            'CHF': 'congestive heart failure',
            'AF': 'atrial fibrillation',
            'MI': 'myocardial infarction',
            'CVA': 'cerebrovascular accident',
            # ... more abbreviations
        }

    def enhance_query(
        self,
        query: str,
        context: Dict = None
    ) -> Dict[str, any]:
        """
        Enhance query with expansions and metadata.
        """
        # Detect intent
        intent = self._detect_intent(query)

        # Expand abbreviations
        expanded_query = self._expand_abbreviations(query)

        # Extract entities
        entities = self._extract_entities(expanded_query)

        # Add context
        if context and context.get('specialty'):
            specialty_filter = context['specialty']
        else:
            specialty_filter = self._infer_specialty(expanded_query, entities)

        return {
            'original_query': query,
            'enhanced_query': expanded_query,
            'intent': intent,
            'entities': entities,
            'specialty_filter': specialty_filter
        }

    def _detect_intent(self, query: str) -> str:
        """
        Detect user intent from query.
        """
        query_lower = query.lower()

        if any(word in query_lower for word in ['dose', 'dosing', 'how much']):
            return 'dosing'
        elif any(word in query_lower for word in ['side effect', 'adverse', 'toxicity']):
            return 'safety'
        elif any(word in query_lower for word in ['manage', 'treatment', 'therapy']):
            return 'management'
        elif any(word in query_lower for word in ['diagnosis', 'workup', 'test']):
            return 'diagnosis'
        else:
            return 'general'

    def _expand_abbreviations(self, query: str) -> str:
        """
        Expand medical abbreviations.
        """
        words = query.split()
        expanded_words = []

        for word in words:
            word_upper = word.strip('.,!?').upper()
            if word_upper in self.abbreviations:
                expanded_words.append(f"{word} ({self.abbreviations[word_upper]})")
            else:
                expanded_words.append(word)

        return ' '.join(expanded_words)

    def _extract_entities(self, query: str) -> Dict[str, List[str]]:
        """
        Extract medical entities from query.
        Simple pattern-based for now, can use NER later.
        """
        # This is simplified - production should use medical NER
        entities = {
            'conditions': [],
            'medications': [],
            'procedures': []
        }

        # Simple pattern matching
        condition_patterns = [
            r'\b(hypertension|diabetes|heart failure|pneumonia)\b'
        ]

        for pattern in condition_patterns:
            matches = re.findall(pattern, query, re.IGNORECASE)
            entities['conditions'].extend(matches)

        return entities

    def _infer_specialty(self, query: str, entities: Dict) -> List[str]:
        """
        Infer medical specialty from query.
        """
        query_lower = query.lower()
        specialties = []

        # Keyword-based specialty detection
        specialty_keywords = {
            'cardiology': ['heart', 'cardiac', 'af', 'atrial', 'chf', 'mi'],
            'endocrinology': ['diabetes', 'thyroid', 'insulin', 'glucose'],
            'infectious_disease': ['infection', 'antibiotic', 'sepsis', 'fever'],
            'nephrology': ['kidney', 'renal', 'dialysis', 'ckd'],
            # ... more specialties
        }

        for specialty, keywords in specialty_keywords.items():
            if any(keyword in query_lower for keyword in keywords):
                specialties.append(specialty)

        return specialties

2. Hybrid Search (Dense + Sparse)

Strategy:

• Dense (Vector): Semantic similarity using embeddings
• Sparse (BM25): Keyword matching for exact terms
• Fusion: Combine scores with learned weights

Python Implementation:

# app/services/medical/rag.py
from typing import List, Dict
from qdrant_client.models import Filter, FieldCondition, MatchAny, SearchRequest
import numpy as np

class RAGService:
    """
    Retrieval-Augmented Generation service.
    """

    def __init__(self, vector_db: VectorDatabase, embedding_gen: EmbeddingGenerator):
        self.vector_db = vector_db
        self.embedding_gen = embedding_gen

    async def search(
        self,
        query: str,
        filters: Dict = None,
        limit: int = 10,
        hybrid: bool = True
    ) -> List[Dict]:
        """
        Hybrid search combining vector and keyword matching.
        """
        # Enhance query
        enhancer = QueryEnhancer()
        enhanced = enhancer.enhance_query(query, filters)

        # Generate query embedding
        query_embedding = await self.embedding_gen.embed_query(
            enhanced['enhanced_query']
        )

        # Build filters
        search_filter = self._build_filter(enhanced, filters)

        # Vector search
        vector_results = self.vector_db.client.search(
            collection_name=self.vector_db.collection_name,
            query_vector=query_embedding,
            query_filter=search_filter,
            limit=limit * 2,  # Get more for reranking
            with_payload=True
        )

        # Convert to standardized format
        results = []
        for hit in vector_results:
            results.append({
                'id': hit.id,
                'score': hit.score,
                'text': hit.payload['text'],
                'document_id': hit.payload['document_id'],
                'document_title': hit.payload['document_title'],
                'page': hit.payload['page'],
                'source_type': hit.payload['source_type'],
                'specialty': hit.payload['specialty'],
                'metadata': {
                    'authors': hit.payload.get('authors', []),
                    'publication_year': hit.payload.get('publication_year'),
                    'doi': hit.payload.get('doi'),
                    'pmid': hit.payload.get('pmid')
                }
            })

        # Rerank results
        results = self._rerank(results, enhanced['original_query'])

        # Return top-k
        return results[:limit]

    def _build_filter(self, enhanced: Dict, filters: Dict = None) -> Filter:
        """
        Build Qdrant filter from query enhancement and user filters.
        """
        conditions = []

        # Add specialty filter if inferred
        if enhanced.get('specialty_filter'):
            conditions.append(
                FieldCondition(
                    key="specialty",
                    match=MatchAny(any=enhanced['specialty_filter'])
                )
            )

        # Add user-provided filters
        if filters:
            if filters.get('source_type'):
                conditions.append(
                    FieldCondition(
                        key="source_type",
                        match=MatchAny(any=filters['source_type'])
                    )
                )

            if filters.get('specialty'):
                conditions.append(
                    FieldCondition(
                        key="specialty",
                        match=MatchAny(any=filters['specialty'])
                    )
                )

        if conditions:
            return Filter(must=conditions)

        return None

    def _rerank(self, results: List[Dict], query: str) -> List[Dict]:
        """
        Rerank results using cross-encoder or heuristics.
        Simple implementation - can use cross-encoder for better results.
        """
        # For now, boost results that contain exact query terms
        query_terms = set(query.lower().split())

        for result in results:
            text_terms = set(result['text'].lower().split())
            overlap = len(query_terms & text_terms)

            # Boost score based on keyword overlap
            boost = 1.0 + (overlap * 0.05)
            result['score'] *= boost

        # Sort by boosted score
        results.sort(key=lambda x: x['score'], reverse=True)

        return results