Full-stack computational software development for biomedical and life science applications. Use when building AI-powered research platforms, clinical decision support systems, multi-LLM ensemble architectures, RAG pipelines, single-cell analysis tools, or biomedical web applications. Includes production-ready patterns for PyTorch deep learning, Flask web apps, citation verification systems, and scientific software distribution.
Scanned 2/12/2026
Install via CLI
Skills Directoryopenskills install mdbabumiamssm/LLMs-Universal-Life-Science-and-Clinical-Skills----
name: computational-software-development
description: "Full-stack computational software development for biomedical and life science applications. Use when building AI-powered research platforms, clinical decision support systems, multi-LLM ensemble architectures, RAG pipelines, single-cell analysis tools, or biomedical web applications. Includes production-ready patterns for PyTorch deep learning, Flask web apps, citation verification systems, and scientific software distribution."
license: Proprietary
---
# Computational Software Development for Life Sciences
## Core Technologies & Architecture Patterns
### Technology Stack Overview
```python
# AI/ML Stack
import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer
import anthropic # Claude API
import openai # GPT API
import google.generativeai as genai # Gemini API
# Scientific Computing
import scanpy as sc
import anndata as ad
import scvi
import pandas as pd
import numpy as np
from scipy import stats
# Web Framework
from flask import Flask, render_template, jsonify, request
import asyncio
import aiohttp
# Data Management
import faiss # Vector database
from sentence_transformers import SentenceTransformer
import redis
from diskcache import Cache
```
---
## 1. Multi-LLM Ensemble Architecture
### Weighted Consensus System (BioMedAI Pattern)
```python
from dataclasses import dataclass
from typing import List, Dict, Optional
import asyncio
from abc import ABC, abstractmethod
@dataclass
class LLMResponse:
content: str
model: str
confidence: float
citations: List[Dict]
latency_ms: float
class AIProvider(ABC):
@abstractmethod
async def query(self, prompt: str, context: str) -> LLMResponse:
pass
class ClaudeProvider(AIProvider):
def __init__(self, api_key: str, model: str = "claude-opus-4-5-20251101"):
self.client = anthropic.Anthropic(api_key=api_key)
self.model = model
self.weight = 1.2 # Primary synthesizer weight
async def query(self, prompt: str, context: str) -> LLMResponse:
start = time.time()
response = await asyncio.to_thread(
self.client.messages.create,
model=self.model,
max_tokens=4096,
messages=[{"role": "user", "content": f"{context}\n\n{prompt}"}]
)
return LLMResponse(
content=response.content[0].text,
model=self.model,
confidence=0.0, # Computed post-hoc
citations=[],
latency_ms=(time.time() - start) * 1000
)
class MultiLLMEnsemble:
"""Parallel multi-LLM execution with weighted consensus."""
def __init__(self, providers: List[AIProvider]):
self.providers = providers
self.weights = {p.__class__.__name__: getattr(p, 'weight', 1.0) for p in providers}
async def query_parallel(self, prompt: str, context: str) -> Dict:
"""Execute all LLMs in parallel for <5s response time."""
tasks = [provider.query(prompt, context) for provider in self.providers]
responses = await asyncio.gather(*tasks, return_exceptions=True)
valid_responses = [r for r in responses if isinstance(r, LLMResponse)]
return self._synthesize_consensus(valid_responses)
def _synthesize_consensus(self, responses: List[LLMResponse]) -> Dict:
"""Weighted synthesis of multi-model responses."""
# Extract key claims from each response
all_claims = []
for r in responses:
claims = self._extract_claims(r.content)
weight = self.weights.get(r.model.split('-')[0].title() + 'Provider', 1.0)
all_claims.extend([(c, weight) for c in claims])
# Rank claims by weighted frequency
claim_scores = {}
for claim, weight in all_claims:
claim_scores[claim] = claim_scores.get(claim, 0) + weight
return {
'consensus': sorted(claim_scores.items(), key=lambda x: -x[1]),
'individual_responses': responses,
'agreement_score': self._compute_agreement(responses)
}
```
---
## 2. Hybrid RAG v3.0 Architecture
### Dense + Sparse Retrieval with Reciprocal Rank Fusion
```python
from sentence_transformers import SentenceTransformer
import faiss
from rank_bm25 import BM25Okapi
from typing import List, Tuple
import numpy as np
class HybridRAG:
"""Production-ready RAG with dense and sparse retrieval fusion."""
def __init__(self, embedding_model: str = "intfloat/e5-large-v2"):
self.embedder = SentenceTransformer(embedding_model)
self.dense_index = None
self.sparse_index = None
self.documents = []
def build_index(self, documents: List[Dict]):
"""Build dual dense + sparse indices."""
self.documents = documents
texts = [d['text'] for d in documents]
# Dense index (FAISS)
embeddings = self.embedder.encode(texts, normalize_embeddings=True)
dimension = embeddings.shape[1]
self.dense_index = faiss.IndexFlatIP(dimension) # Inner product for cosine sim
self.dense_index.add(embeddings.astype('float32'))
# Sparse index (BM25)
tokenized = [t.lower().split() for t in texts]
self.sparse_index = BM25Okapi(tokenized)
def retrieve(self, query: str, k: int = 10, alpha: float = 0.5) -> List[Dict]:
"""Reciprocal Rank Fusion of dense + sparse results."""
# Dense retrieval
query_emb = self.embedder.encode([query], normalize_embeddings=True)
dense_scores, dense_ids = self.dense_index.search(query_emb.astype('float32'), k * 2)
# Sparse retrieval
sparse_scores = self.sparse_index.get_scores(query.lower().split())
sparse_ids = np.argsort(sparse_scores)[-k * 2:][::-1]
# Reciprocal Rank Fusion (RRF)
rrf_k = 60 # Standard RRF parameter
doc_scores = {}
for rank, doc_id in enumerate(dense_ids[0]):
doc_scores[doc_id] = doc_scores.get(doc_id, 0) + alpha / (rrf_k + rank + 1)
for rank, doc_id in enumerate(sparse_ids):
doc_scores[doc_id] = doc_scores.get(doc_id, 0) + (1 - alpha) / (rrf_k + rank + 1)
# Return top-k fused results
ranked = sorted(doc_scores.items(), key=lambda x: -x[1])[:k]
return [{'document': self.documents[did], 'score': score} for did, score in ranked]
```
---
## 3. Citation Validation Pipeline
### Real-Time PubMed Citation Verification
```python
import re
import aiohttp
from typing import Optional, Dict, List
class CitationValidator:
"""3-layer citation validation: Format → Existence → Relevance."""
PUBMED_API = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
async def validate_citation(self, citation: str, claim_context: str) -> Dict:
"""Full validation pipeline for a single citation."""
result = {
'original': citation,
'format_valid': False,
'exists': False,
'relevant': False,
'pmid': None,
'title': None,
'abstract': None
}
# Layer 1: Format validation
pmid = self._extract_pmid(citation)
if not pmid:
pmid = await self._search_pubmed(citation)
if not pmid:
return result
result['format_valid'] = True
result['pmid'] = pmid
# Layer 2: Existence check
metadata = await self._fetch_pubmed_metadata(pmid)
if not metadata:
return result
result['exists'] = True
result['title'] = metadata.get('title')
result['abstract'] = metadata.get('abstract')
# Layer 3: Relevance scoring
relevance = self._compute_relevance(claim_context, metadata)
result['relevant'] = relevance > 0.3
result['relevance_score'] = relevance
return result
async def _fetch_pubmed_metadata(self, pmid: str) -> Optional[Dict]:
"""Fetch article metadata from PubMed."""
url = f"{self.PUBMED_API}/efetch.fcgi"
params = {'db': 'pubmed', 'id': pmid, 'retmode': 'xml'}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
if resp.status == 200:
xml = await resp.text()
return self._parse_pubmed_xml(xml)
return None
async def validate_all_citations(self, text: str, claims: List[str]) -> Dict:
"""Validate all citations in a response. Target: 97.8% accuracy."""
citations = self._extract_all_citations(text)
tasks = [self.validate_citation(c, ' '.join(claims)) for c in citations]
results = await asyncio.gather(*tasks)
valid = sum(1 for r in results if r['exists'] and r['relevant'])
return {
'total_citations': len(citations),
'valid_citations': valid,
'accuracy': valid / len(citations) if citations else 1.0,
'details': results
}
```
---
## 4. Clinical Decision Support System
### Deep Learning Ensemble with Conformal Prediction (MPN Clinical Pattern)
```python
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from sklearn.preprocessing import RobustScaler
import numpy as np
class DeepResidualNN(nn.Module):
"""Residual neural network for clinical risk prediction."""
def __init__(self, input_dim: int, hidden_dims: List[int], num_classes: int, dropout: float = 0.3):
super().__init__()
layers = []
prev_dim = input_dim
for hidden_dim in hidden_dims:
layers.extend([
nn.Linear(prev_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout)
])
prev_dim = hidden_dim
self.feature_extractor = nn.Sequential(*layers)
self.classifier = nn.Linear(prev_dim, num_classes)
# Skip connection for residual learning
self.skip = nn.Linear(input_dim, prev_dim) if input_dim != prev_dim else nn.Identity()
def forward(self, x):
features = self.feature_extractor(x)
skip_features = self.skip(x)
combined = features + skip_features # Residual connection
return self.classifier(combined)
class SimpleTransformer(nn.Module):
"""Attention-based model for clinical feature interactions."""
def __init__(self, input_dim: int, num_classes: int, n_heads: int = 4, dim_feedforward: int = 256):
super().__init__()
self.embedding = nn.Linear(1, 64) # Embed each feature
encoder_layer = nn.TransformerEncoderLayer(
d_model=64, nhead=n_heads, dim_feedforward=dim_feedforward, batch_first=True
)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=2)
self.classifier = nn.Linear(64 * input_dim, num_classes)
def forward(self, x):
# x: (batch, features) -> (batch, features, 1) -> (batch, features, 64)
x = x.unsqueeze(-1)
x = self.embedding(x)
x = self.transformer(x)
x = x.flatten(1)
return self.classifier(x)
class ClinicalEnsemble:
"""Weighted ensemble with conformal prediction for uncertainty."""
def __init__(self, models: List[nn.Module], weights: List[float]):
self.models = models
self.weights = weights
self.calibration_scores = None
def predict_with_confidence(self, X: torch.Tensor, alpha: float = 0.1) -> Dict:
"""Prediction with conformal prediction sets."""
# Get weighted ensemble predictions
all_probs = []
for model, weight in zip(self.models, self.weights):
model.eval()
with torch.no_grad():
logits = model(X)
probs = torch.softmax(logits, dim=1)
all_probs.append(probs * weight)
ensemble_probs = sum(all_probs) / sum(self.weights)
predictions = ensemble_probs.argmax(dim=1)
# Conformal prediction sets (90% coverage target)
prediction_sets = self._compute_conformal_sets(ensemble_probs, alpha)
return {
'predictions': predictions,
'probabilities': ensemble_probs,
'prediction_sets': prediction_sets,
'confidence': ensemble_probs.max(dim=1).values
}
def _compute_conformal_sets(self, probs: torch.Tensor, alpha: float) -> List[List[int]]:
"""Adaptive prediction sets for each sample."""
sets = []
for prob in probs:
sorted_probs, sorted_idx = torch.sort(prob, descending=True)
cumsum = torch.cumsum(sorted_probs, dim=0)
cutoff = (cumsum >= 1 - alpha).nonzero()[0].item() if (cumsum >= 1 - alpha).any() else len(prob) - 1
sets.append(sorted_idx[:cutoff + 1].tolist())
return sets
```
---
## 5. Flask Web Application Pattern
### Production-Ready Biomedical Web Interface
```python
from flask import Flask, render_template, jsonify, request
from flask_cors import CORS
import asyncio
from concurrent.futures import ThreadPoolExecutor
from loguru import logger
app = Flask(__name__)
CORS(app)
executor = ThreadPoolExecutor(max_workers=10)
# Initialize components
orchestrator = None
@app.before_first_request
def initialize():
global orchestrator
orchestrator = IntelligentOrchestrator()
logger.info("BioMedAI Orchestrator initialized")
@app.route('/api/query', methods=['POST'])
def query_endpoint():
"""Main query endpoint with full RAG + verification pipeline."""
data = request.json
query = data.get('query', '')
options = data.get('options', {})
# Run async operations in executor
loop = asyncio.new_event_loop()
result = loop.run_until_complete(
orchestrator.process_query(
query=query,
use_ensemble=options.get('ensemble', True),
verify_citations=options.get('verify', True),
safety_check=options.get('safety', True)
)
)
loop.close()
return jsonify({
'answer': result.answer,
'citations': result.citations,
'confidence': result.confidence,
'verification': result.verification_report,
'latency_ms': result.latency_ms
})
@app.route('/api/predict', methods=['POST'])
def predict_endpoint():
"""Clinical risk prediction endpoint."""
data = request.json
features = np.array(data['features']).reshape(1, -1)
result = clinical_model.predict_with_confidence(
torch.tensor(features, dtype=torch.float32)
)
return jsonify({
'prediction': RISK_LABELS[result['predictions'][0].item()],
'confidence': result['confidence'][0].item(),
'prediction_set': [RISK_LABELS[i] for i in result['prediction_sets'][0]],
'probabilities': result['probabilities'][0].tolist()
})
@app.route('/health')
def health_check():
return jsonify({'status': 'healthy', 'version': '3.0.1'})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=False)
```
---
## 6. SHAP Explainability Integration
```python
import shap
import matplotlib.pyplot as plt
class ModelExplainer:
"""SHAP-based feature importance for clinical models."""
def __init__(self, model, feature_names: List[str]):
self.model = model
self.feature_names = feature_names
self.explainer = None
def fit(self, X_background: np.ndarray):
"""Initialize explainer with background data."""
def model_predict(X):
with torch.no_grad():
return torch.softmax(self.model(torch.tensor(X, dtype=torch.float32)), dim=1).numpy()
self.explainer = shap.KernelExplainer(model_predict, X_background[:100])
def explain_prediction(self, X: np.ndarray, class_idx: int = None) -> Dict:
"""Generate SHAP explanation for a prediction."""
shap_values = self.explainer.shap_values(X)
if class_idx is not None:
values = shap_values[class_idx][0]
else:
values = np.abs(shap_values).mean(axis=0)[0]
importance = list(zip(self.feature_names, values))
importance.sort(key=lambda x: abs(x[1]), reverse=True)
return {
'top_features': importance[:10],
'shap_values': values,
'base_value': self.explainer.expected_value
}
def plot_summary(self, X: np.ndarray, save_path: str = None):
"""Generate publication-quality SHAP summary plot."""
shap_values = self.explainer.shap_values(X)
plt.figure(figsize=(10, 8))
shap.summary_plot(
shap_values, X, feature_names=self.feature_names,
show=False, max_display=20
)
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=300, bbox_inches='tight')
plt.close()
```
---
## 7. Single-Cell Analysis Tool Patterns
### Cellidentifier-Style Analysis Pipeline
```python
import scanpy as sc
import anndata as ad
from typing import Dict, List
class CellIdentifier:
"""Automated cell type identification from single-cell data."""
MARKER_DB = {
'HSC': ['CD34', 'KIT', 'THY1'],
'Megakaryocyte': ['ITGA2B', 'PF4', 'GP1BA', 'VWF'],
'Erythroid': ['HBB', 'HBA1', 'GYPA', 'KLF1'],
'T_cell': ['CD3D', 'CD3E', 'CD4', 'CD8A'],
'B_cell': ['CD19', 'MS4A1', 'CD79A'],
'Monocyte': ['CD14', 'LYZ', 'S100A8', 'S100A9'],
'NK': ['NKG7', 'GNLY', 'NCAM1']
}
def __init__(self, adata: ad.AnnData):
self.adata = adata
def score_cell_types(self) -> pd.DataFrame:
"""Score each cell for all known cell types."""
for cell_type, markers in self.MARKER_DB.items():
available_markers = [m for m in markers if m in self.adata.var_names]
if available_markers:
sc.tl.score_genes(self.adata, available_markers, score_name=f'{cell_type}_score')
score_cols = [c for c in self.adata.obs.columns if c.endswith('_score')]
return self.adata.obs[score_cols]
def annotate_cells(self, threshold: float = 0.5) -> np.ndarray:
"""Assign cell types based on highest score."""
scores = self.score_cell_types()
cell_types = scores.idxmax(axis=1).str.replace('_score', '')
max_scores = scores.max(axis=1)
# Mark low-confidence as 'Unknown'
cell_types[max_scores < threshold] = 'Unknown'
self.adata.obs['cell_type'] = cell_types
return cell_types.values
```
---
## 8. Package Distribution Pattern
### PyPI-Ready Package Structure
```python
# pyproject.toml content
"""
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "biomedai"
version = "3.0.1"
description = "Multi-LLM ensemble for biomedical research"
readme = "README.md"
license = {text = "MIT"}
authors = [
{name = "MD BABU MIA", email = "md.babu.mia@mssm.edu"}
]
keywords = ["bioinformatics", "LLM", "RAG", "biomedical", "AI"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Science/Research",
"Topic :: Scientific/Engineering :: Bio-Informatics",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.10",
]
dependencies = [
"anthropic>=0.18.0",
"openai>=1.12.0",
"torch>=2.0.0",
"transformers>=4.36.0",
"sentence-transformers>=2.2.0",
"faiss-cpu>=1.7.4",
"flask>=3.0.0",
"scanpy>=1.9.0",
"pandas>=2.0.0",
"numpy>=1.24.0",
]
[project.optional-dependencies]
dev = ["pytest", "black", "ruff", "mypy"]
docs = ["sphinx", "sphinx-rtd-theme"]
[project.scripts]
biomedai = "biomedai.cli:main"
"""
```
---
## Project Portfolio
### BioMedAI (LifeScienceBrowser)
- Multi-LLM ensemble (Claude + GPT + Gemini)
- 97.8% citation accuracy, 92% hallucination reduction
- Hybrid RAG v3.0 with FAISS + BM25
- Flask web interface with real-time verification
### MPN Clinical Software V3.0
- 111-variable deep learning ensemble
- 80% accuracy, 100% recall on high-risk patients
- Conformal prediction with 90% coverage
- SHAP explainability for clinical transparency
### Single-Cell Tools (GitHub)
- cellidentifierdx: Automated cell type identification
- babu-scvi-tools: Deep probabilistic single-cell analysis
- tahoe-x1: Gigascale single-cell foundation model
- mosaic: Tapestri platform visualization
### Biomedical AI Tools
- paper-qa-futurehouse: RAG for scientific documents
- Biomedical-Genomics-Personalized-Reasoning-LLM-Agent
- DNAdamageclassifierDx: DNA damage classification
- ncbi-geo-pubmed-search: NCBI database search package
---
## Key Metrics & Standards
| Metric | Target | Achieved |
|--------|--------|----------|
| Citation Accuracy | >95% | 97.8% |
| Hallucination Rate | <5% | 2.2% |
| Clinical Recall (High Risk) | 100% | 100% |
| Response Latency | <10s | 8-10s |
| Conformal Coverage | 90% | 90% |
| Code Coverage | >80% | 85% |
---
See `references/architecture_patterns.md` for detailed system design documentation.
See `references/deployment_guide.md` for production deployment procedures.
No comments yet. Be the first to comment!
