feat (start): initial setup

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.venv

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.9
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user ./app /app
+CMD ["uvicorn", "server_sentiment_analysis:app", "--host", "0.0.0.0", "--port", "7860"]

app/clinical_embedding.py

@@ -0,0 +1,67 @@
+import numpy as np
+from transformers import pipeline
+from typing import List
+
+class ClinicalBERT:
+    """
+    A wrapper class for Bio_ClinicalBERT model to generate sentence embeddings.
+    """
+    
+    def __init__(self, model_name: str = "emilyalsentzer/Bio_ClinicalBERT", device: int = -1):
+        """
+        Initialize the ClinicalBERT model using pipeline.
+        
+        Args:
+            model_name: The Hugging Face model identifier
+            device: Device to run the model on (-1 for CPU, 0 for first GPU, etc.)
+        """
+        self.model_name = model_name
+        
+        # Create feature extraction pipeline
+        print(f"Loading {model_name}...")
+        self.pipe = pipeline(
+            "feature-extraction",
+            model=model_name,
+            device=device
+        )
+        print(f"Model loaded successfully on device {device}")
+    
+    def get_embeddings(self, sentences: List[str], pooling: str = 'mean') -> np.ndarray:
+        """
+        Generate embeddings for a list of sentences.
+        
+        Args:
+            sentences: List of input sentences
+            pooling: Pooling strategy ('mean', 'cls', or 'max')
+        
+        Returns:
+            numpy array of shape (num_sentences, embedding_dim)
+        """
+        if not sentences:
+            return np.array([])
+        
+        # Get embeddings from pipeline
+        # The pipeline returns a list with shape (1, num_tokens, embedding_dim) per sentence
+        outputs = self.pipe(sentences)
+        
+        # Apply pooling strategy to each sentence
+        embeddings = []
+        for sentence_output in outputs:
+            # Convert to numpy array and squeeze the first dimension
+            # Shape: (1, num_tokens, embedding_dim) -> (num_tokens, embedding_dim)
+            tokens_array = np.array(sentence_output).squeeze(0)
+            
+            if pooling == 'cls':
+                # Use [CLS] token (first token)
+                embedding = tokens_array[0]
+            elif pooling == 'max':
+                # Max pooling across tokens (dim 0)
+                embedding = np.max(tokens_array, axis=0)
+            else:  # mean pooling (default)
+                # Average across all tokens (dim 0)
+                embedding = np.mean(tokens_array, axis=0)
+            
+            embeddings.append(embedding)
+        
+        # Stack embeddings into a 2D array: (num_sentences, embedding_dim)
+        return np.vstack(embeddings)

app/server_clinical_embedding.py

@@ -0,0 +1,100 @@
+from typing import List
+from fastapi import FastAPI, Query
+from pydantic import BaseModel
+import uvicorn
+
+from clinical_embedding import ClinicalBERT
+
+# Pydantic models for request/response
+class EmbeddingRequest(BaseModel):
+    sentences: List[str]
+    pooling: str = 'mean'
+
+
+class EmbeddingResponse(BaseModel):
+    embeddings: List[List[float]]
+    shape: List[int]
+    pooling: str
+
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="Clinical BERT Embeddings API",
+    description="API for generating embeddings using Bio_ClinicalBERT model",
+    version="1.0.0"
+)
+
+# Initialize model (global instance)
+clinical_bert = None
+
+
+@app.on_event("startup")
+async def startup_event():
+    """Load model on startup"""
+    global clinical_bert
+    clinical_bert = ClinicalBERT(device=-1)  # Use device=0 for GPU
+
+
+@app.get("/")
+async def root():
+    """Root endpoint with API information"""
+    return {
+        "message": "Clinical BERT Embeddings API",
+        "endpoints": {
+            "/embeddings": "GET - Generate embeddings from sentences",
+            "/docs": "GET - Interactive API documentation"
+        }
+    }
+
+
+@app.get("/embeddings", response_model=EmbeddingResponse)
+async def get_embeddings(
+    sentences: List[str] = Query(..., description="List of sentences to embed"),
+    pooling: str = Query('mean', description="Pooling strategy: mean, cls, or max")
+):
+    """
+    Generate embeddings for a list of sentences.
+    
+    Args:
+        sentences: List of input sentences
+        pooling: Pooling strategy ('mean', 'cls', or 'max')
+    
+    Returns:
+        EmbeddingResponse with embeddings and metadata
+    """
+    # Validate pooling method
+    if pooling not in ['mean', 'cls', 'max']:
+        return {
+            "error": "Invalid pooling method. Choose from: mean, cls, max"
+        }
+    
+    # Generate embeddings
+    embeddings = clinical_bert.get_embeddings(sentences, pooling=pooling)
+    
+    # Convert to list for JSON serialization
+    embeddings_list = embeddings.tolist()
+    
+    return EmbeddingResponse(
+        embeddings=embeddings_list,
+        shape=list(embeddings.shape),
+        pooling=pooling
+    )
+
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {
+        "status": "healthy",
+        "model_loaded": clinical_bert is not None
+    }
+
+
+if __name__ == "__main__":
+    # Run the server
+    uvicorn.run(
+        "main:app",
+        host="0.0.0.0",
+        port=8000,
+        reload=False
+    )

app/test_clinical_embedding.py

@@ -0,0 +1,22 @@
+from clinical_embedding import ClinicalBERT
+
+# Initialize the model (use device=0 for GPU, device=-1 for CPU)
+clinical_bert = ClinicalBERT(device=-1)
+
+# Example sentences
+sentences = [
+    "Heart Attack",
+    "Myocardial Infarction"
+]
+
+# Get embeddings with mean pooling
+embeddings = clinical_bert.get_embeddings(sentences, pooling='mean')
+print(f"Embeddings shape: {embeddings.shape}")
+print(f"First embedding (truncated): {embeddings[0][:5]}...")
+
+# Try different pooling strategies
+embeddings_cls = clinical_bert.get_embeddings(sentences, pooling='cls')
+print(f"\nCLS pooling shape: {embeddings_cls.shape}")
+
+embeddings_max = clinical_bert.get_embeddings(sentences, pooling='max')
+print(f"Max pooling shape: {embeddings_max.shape}")

requirements.txt

@@ -0,0 +1,12 @@
+# Web Framework
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+pydantic==2.5.0
+
+# Machine Learning
+transformers==4.35.2
+torch==2.1.1
+numpy==1.24.3
+
+# Optional: for GPU support, also install:
+# torch==2.1.1+cu118 -f https://download.pytorch.org/whl/torch_stable.html