import os
import re
import json
import tempfile
from pathlib import Path
from typing import List
import fitz
import pytesseract
from pdf2image import convert_from_path
from groq import Groq
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, HTMLResponse
from pydantic import BaseModel
app = FastAPI(title="ContraGenAI - A Protocol Contradiction Detector")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ── Static files + SPA ───────────────────────────────────────────────────────
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
import pathlib
_static_dir = pathlib.Path(__file__).parent / "static"
if _static_dir.exists():
app.mount("/assets", StaticFiles(directory=str(_static_dir / "assets")), name="assets")
@app.get("/")
async def serve_index():
return FileResponse(str(_static_dir / "index.html"))
# ── Config ────────────────────────────────────────────────────────────────────
GROQ_MODEL = "llama-3.3-70b-versatile"
GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "")
# ── Prompts ───────────────────────────────────────────────────────────────────
EXTRACTION_SYSTEM = """Extract protocol parameters from this research paper.
IMPORTANT:
Do not extract numbers randomly.
Only extract values that are clearly associated with an experimental step or procedure. Always preserve context.
Return ONLY valid JSON:
{
"title": "paper title",
"authors": "first author et al.",
"year": "year",
"protocol_type": "e.g. Western blot / ELISA / PCR",
"parameters": {
"reagents": [{"name": "...", "concentration": "...", "vendor": "..."}],
"cell_lines": [{"name": "...", "culture_conditions": "...", "serum": "..."}],
"temperatures": [{"step": "...", "value": "...", "unit": "C"}],
"timings": [{"step": "...", "duration": "..."}],
"equipment": [{"name": "...", "settings": "..."}],
"buffers": [{"name": "...", "composition": "...", "pH": "..."}],
"antibodies": [{"target": "...", "dilution": "...", "vendor": "..."}],
"other": [{"parameter": "...", "value": "..."}]
}
}"""
COMPARISON_SYSTEM = """You are an expert in bioengineering reproducibility and experimental methodology.
Compare protocols from multiple research papers and identify ALL methodological contradictions. Always preserve context.
SEVERITY ASSIGNMENT MUST FOLLOW THE RULES BELOW.
Do NOT decide severity based on subjective judgment.
Always determine severity using the predefined parameter hierarchy.
If a parameter appears in the hierarchy below, assign the corresponding severity level.
If multiple rules apply, assign the highest severity.
If a parameter is not listed, assign LOW severity by default.
PARAMETER SEVERITY HIERARCHY
HIGH SEVERITY — Directly affects experimental outcome or biological behavior
Assign HIGH severity if contradiction involves:
- cell line identity or organism
- incubation temperature
- antibiotic or drug concentration
- enzyme concentration
- transfection conditions
- MOI (multiplicity of infection)
- centrifugation speed
- incubation duration
- buffer composition
- pH
- reagent concentration
- sample storage conditions
- reaction volume
- PCR cycle conditions
These parameters directly impact reproducibility or experimental success.
MEDIUM SEVERITY — Affects efficiency, yield, or data quality
Assign MEDIUM severity if contradiction involves:
- incubation time variations within acceptable ranges
- reagent vendor differences
- equipment model differences
- washing conditions
- dilution ratios
- culture media composition
- cell density
- mixing speed
- incubation environment (shaking/static)
LOW SEVERITY — Minor procedural or documentation differences
Assign LOW severity if contradiction involves:
- equipment brand differences
- container type
- labeling differences
- formatting differences
- reporting style differences
- minor descriptive variations
Return ONLY valid JSON (no markdown fences, no extra text) using this schema:
{
"protocol_type": "detected protocol type",
"contradictions": [
{
"parameter": "parameter name",
"category": "reagents|cell_lines|temperatures|timings|equipment|buffers|antibodies|other",
"severity": "high|medium|low",
"values": {"paper_0": "value from paper 1", "paper_1": "value from paper 2"},
"explanation": "why this is a contradiction and its likely impact on reproducibility"
}
],
"ranked_issues": [
{"rank": 1, "parameter": "...", "severity": "high|medium|low", "brief": "one-sentence impact"}
],
"optimal_protocol": {
"rationale": "overall recommendation rationale",
"parameters": [
{"label": "parameter name", "value": "recommended value", "reason": "why this is optimal"}
]
},
"summary": {"total_contradictions": 0, "high": 0, "medium": 0, "low": 0}
}"""
# ── Core helpers ──────────────────────────────────────────────────────────────
def extract_text_from_pdf(pdf_path: str) -> str:
text = ""
# ---------- Try PyMuPDF extraction ----------
try:
doc = fitz.open(pdf_path)
for page in doc:
text += page.get_text()
except Exception:
pass
# ---------- OCR fallback if no text ----------
if not text.strip():
try:
images = convert_from_path(pdf_path)
for img in images:
text += pytesseract.image_to_string(img)
except Exception:
raise ValueError("PDF text extraction failed.")
if not text.strip():
raise ValueError("PDF text extraction failed.")
if len(text.strip()) < 500:
text = text[:2000]
# ---------- Try to isolate Methods section ----------
methods_match = re.search(
r'(?:^\s*(?:\d+\.?\d*\.?\s+)?'
r'(?:materials?\s+and\s+methods?|experimental\s+procedures?'
r'|methods?\s+and\s+materials?|methods?)\s*$)'
r'(.*?)'
r'(?=^\s*(?:\d+\.?\d*\.?\s+)?'
r'(?:results?|discussion|conclusion|references|acknowledgements?))',
text, re.IGNORECASE | re.DOTALL | re.MULTILINE
)
if methods_match and len(methods_match.group(1).strip()) > 200:
text = methods_match.group(1)
else:
chars = len(text)
text = text[chars // 10: chars // 10 + 15000]
words = text.split()
if len(words) > 1500:
text = " ".join(words[:1500])
text = text.replace("\n", " ")
return text
def parse_json_response(raw: str) -> dict:
cleaned = re.sub(r"```(?:json)?", "", raw).strip()
cleaned = re.sub(r"```", "", cleaned).strip()
brace_start = cleaned.find("{")
if brace_start > 0:
cleaned = cleaned[brace_start:]
try:
return json.loads(cleaned)
except json.JSONDecodeError:
pass
try:
open_braces = cleaned.count("{") - cleaned.count("}")
open_brackets = cleaned.count("[") - cleaned.count("]")
repaired = cleaned.rstrip(",\n ")
repaired += "]" * max(open_brackets, 0)
repaired += "}" * max(open_braces, 0)
return json.loads(repaired)
except json.JSONDecodeError:
pass
match = re.search(r"\{.*\}", cleaned, re.DOTALL)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
raise ValueError("No valid JSON found in response.\n" + raw[:500])
def call_groq_with_backoff(client: Groq, messages: list) -> str:
word_limits = [2500, 2000, 1500, 1000]
attempt = 0
while attempt < len(word_limits):
try:
response = client.chat.completions.create(
model=GROQ_MODEL,
messages=messages,
temperature=0.1,
max_tokens=2000,
)
return response.choices[0].message.content
except Exception as e:
err = str(e)
if "413" in err or "too large" in err.lower() or "rate_limit_exceeded" in err.lower():
attempt += 1
if attempt >= len(word_limits):
if "429" in err:
retry_after = None
try:
# Try to read headers if available
if hasattr(e, "response") and e.response:
headers = getattr(e.response, "headers", {})
retry_after = headers.get("retry-after")
except Exception:
pass
if retry_after:
raise ValueError(
f"Rate limit exceeded. Retry after {retry_after} seconds."
)
else:
raise ValueError(
"Rate limit exceeded. Retry later."
)
raise ValueError("Request too large even after maximum reductions.")
new_limit = word_limits[attempt]
for msg in messages:
if msg["role"] == "user":
words = msg["content"].split()
msg["content"] = " ".join(words[:new_limit])
else:
raise
def extract_protocol(client: Groq, pdf_path: str, filename: str) -> dict:
paper_text = extract_text_from_pdf(pdf_path)
messages = [
{"role": "system", "content": EXTRACTION_SYSTEM},
{"role": "user", "content":
"Extract the complete experimental protocol from this research paper. "
"Be thorough - capture all concentrations, timings, temperatures, cell lines, "
"reagents, antibodies, and equipment settings.\n\nPAPER TEXT:\n" + paper_text
}
]
raw = call_groq_with_backoff(client, messages)
try:
result = parse_json_response(raw)
except Exception as e:
raise ValueError(f"Invalid JSON returned by LLM: {str(e)}")
result["_filename"] = filename
return result
def compare_protocols(client: Groq, extracted: list, paper_names: list) -> dict:
sections = []
for i, (e, name) in enumerate(zip(extracted, paper_names)):
sections.append(
f"=== Paper {i+1}: {e.get('title', name)} ===\n"
f"{json.dumps(e.get('parameters', {}), indent=2)}"
)
prompt = (
f"Compare these {len(extracted)} research paper protocols "
f"and identify ALL methodological contradictions:\n\n"
+ "\n\n".join(sections)
)
messages = [
{"role": "system", "content": COMPARISON_SYSTEM},
{"role": "user", "content": prompt},
]
raw = call_groq_with_backoff(client, messages)
return parse_json_response(raw)
def generate_html_report(extracted: list, comparison: dict, paper_names: list) -> str:
summary = comparison.get("summary", {})
contras = comparison.get("contradictions", [])
ranked = comparison.get("ranked_issues", [])
optimal = comparison.get("optimal_protocol", {})
sev_color = {"high": "#ff4d6d", "medium": "#ff9f1c", "low": "#2ec4b6"}
sev_bg = {"high": "#fff0f3", "medium": "#fff8ee", "low": "#f0fafa"}
def badge(sev):
col = sev_color.get(sev, "#888")
return (f'{sev}')
extraction_rows = ""
for i, e in enumerate(extracted):
cats = e.get("parameters", {})
total = sum(len(v) for v in cats.values() if isinstance(v, list))
extraction_rows += f"""
Paper {i+1}
{e.get('_filename','')} |
{e.get('title','—')} |
{e.get('authors','—')} |
{e.get('year','—')} |
{e.get('protocol_type','—')} |
{total} |
"""
contra_rows = ""
for c in contras:
sev = c.get("severity", "low")
col = sev_color.get(sev, "#888")
vals = c.get("values", {})
paper_cols = "".join(
f'{vals.get(f"paper_{i}","-")} | '
for i in range(len(paper_names))
)
contra_rows += f"""
| {c.get('parameter','')} |
{c.get('category','')} |
{badge(sev)} |
{paper_cols}
{c.get('explanation','')} |
"""
ranked_html = ""
for item in ranked:
sev = item.get("severity", "low")
col = sev_color.get(sev, "#888")
bg = sev_bg.get(sev, "#fafafa")
ranked_html += f"""
#{item.get('rank','?')}
{item.get('parameter','')} {badge(sev)}
{item.get('brief','')}
"""
optimal_html = ""
for p in optimal.get("parameters", []):
optimal_html += f"""
{p.get('label','')}
{p.get('value','')}
{p.get('reason','')}
"""
paper_th = "".join(
f'Paper {i+1}
{n} | '
for i, n in enumerate(paper_names)
)
return f"""
ContraGenAI - Protocol Contradiction Report
Summary
{summary.get('total_contradictions',0)}
Total
{summary.get('high',0)}
High
{summary.get('medium',0)}
Medium
{summary.get('low',0)}
Low
Extracted Parameters
| Paper | Title | Authors | Year | Protocol Type | Parameters |
|---|
{extraction_rows}
Ranked by Severity
{ranked_html}
Side-by-side Comparison
| Parameter | Category | Severity | {paper_th}Impact on Reproducibility |
|---|
{contra_rows}
Recommended Optimal Protocol
{optimal.get('rationale','')}
{optimal_html}
"""
# ── API Routes ────────────────────────────────────────────────────────────────
@app.get("/health")
def health():
return {"status": "ok", "model": GROQ_MODEL}
@app.post("/analyze")
async def analyze(files: List[UploadFile] = File(...)):
if len(files) < 2:
raise HTTPException(status_code=400, detail="Please upload at least 2 PDF files.")
api_key = GROQ_API_KEY
if not api_key:
raise HTTPException(status_code=500, detail="GROQ_API_KEY not set in environment.")
client = Groq(api_key=api_key)
extracted = []
paper_names = []
tmp_paths = []
try:
# Save uploads to temp files
for f in files:
if not f.filename.lower().endswith(".pdf"):
raise HTTPException(status_code=400, detail=f"{f.filename} is not a PDF.")
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".pdf")
tmp.write(await f.read())
tmp.close()
tmp_paths.append(tmp.name)
paper_names.append(f.filename)
# Extract each paper
for path, name in zip(tmp_paths, paper_names):
try:
result = extract_protocol(client, path, name)
extracted.append(result)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Failed to process {name}: {str(e)}"
)
# Compare
comparison = compare_protocols(client, extracted, paper_names)
# Build HTML report
html_report = generate_html_report(extracted, comparison, paper_names)
return JSONResponse({
"extracted": extracted,
"comparison": comparison,
"html_report": html_report,
"paper_names": paper_names,
})
finally:
for p in tmp_paths:
try:
os.unlink(p)
except Exception:
pass