Detect Adverse Drug Events in Medical Text with GLiClass
Use zero-shot classification to identify adverse drug events, assess severity, and categorize reaction types in medical text -- all running locally with no API keys required.
Overview
Adverse Drug Events (ADEs) are undesirable experiences associated with medication use, including adverse drug reactions, side effects, drug interactions, and medication errors. Early detection of ADEs in clinical notes and medical literature is critical for patient safety and regulatory compliance.
This cookbook demonstrates how to build an ADE detection and monitoring pipeline using GLiClass, Knowledgator's open-source zero-shot classification framework. GLiClass runs entirely on your local machine, making it suitable for environments where patient data cannot leave the premises.
What you will build:
- A basic ADE detector that flags medical text containing adverse events
- A severity classifier for pharmacovigilance triage
- An ADE type classifier for structured reporting
- A batch screening tool for literature surveillance
- A complete
ADEMonitorpipeline that ties everything together
Installation
pip install gliclass torch transformers
Quick Start
Detect an adverse drug event in under 15 lines:
from gliclass import GLiClassModel, ZeroShotClassificationPipeline
from transformers import AutoTokenizer
model = GLiClassModel.from_pretrained("knowledgator/gliclass-base-v3.0")
tokenizer = AutoTokenizer.from_pretrained("knowledgator/gliclass-base-v3.0")
pipeline = ZeroShotClassificationPipeline(
model, tokenizer, classification_type="multi-label", device="cpu"
)
text = (
"The patient developed severe hepatotoxicity after three weeks of treatment. "
"Liver enzymes were elevated to five times the upper limit of normal, "
"requiring immediate discontinuation of the drug."
)
labels = ["contains adverse drug event", "no adverse drug event", "discusses potential drug risk"]
results = pipeline(text, labels, threshold=0.5)[0]
for r in results:
print(f" {r['label']}: {r['score']:.2f}")
# Expected output:
# contains adverse drug event: 0.94
Define ADE Classification Labels
Effective ADE detection depends on well-chosen label sets. Below are four label sets covering different classification tasks.
Basic ADE Detection
ADE_LABELS = [
"contains adverse drug event",
"no adverse drug event mentioned",
"discusses potential drug risk",
]
Detailed ADE Classification
DETAILED_ADE_LABELS = [
"confirmed adverse drug reaction",
"suspected adverse drug reaction",
"drug interaction reported",
"medication error",
"normal therapeutic effect",
"no drug-related event",
]
Severity Classification
SEVERITY_LABELS = [
"life-threatening adverse event",
"serious adverse event requiring hospitalization",
"moderate adverse event requiring medical attention",
"mild adverse event",
"no adverse event described",
]
ADE Type Classification
ADE_TYPE_LABELS = [
"allergic reaction or hypersensitivity",
"gastrointestinal adverse effect",
"cardiovascular adverse effect",
"neurological adverse effect",
"hepatotoxicity or liver injury",
"nephrotoxicity or kidney injury",
"hematological adverse effect",
"dermatological adverse effect",
"drug-drug interaction",
"overdose or toxicity",
]
Basic ADE Detection
A simple function that returns every label whose confidence exceeds the threshold:
from gliclass import GLiClassModel, ZeroShotClassificationPipeline
from transformers import AutoTokenizer
model = GLiClassModel.from_pretrained("knowledgator/gliclass-base-v3.0")
tokenizer = AutoTokenizer.from_pretrained("knowledgator/gliclass-base-v3.0")
pipeline = ZeroShotClassificationPipeline(
model, tokenizer, classification_type="multi-label", device="cpu"
)
ADE_LABELS = [
"contains adverse drug event",
"no adverse drug event mentioned",
"discusses potential drug risk",
]
def detect_ade(text: str, threshold: float = 0.5) -> list[dict]:
"""Return labels that exceed the confidence threshold."""
return pipeline(text, ADE_LABELS, threshold=threshold)[0]
# Example
sample = (
"The patient developed severe hepatotoxicity after three weeks "
"of treatment with the medication. Liver enzymes were elevated "
"to five times the upper limit of normal, requiring immediate "
"discontinuation of the drug."
)
for hit in detect_ade(sample):
print(f" {hit['label']}: {hit['score']:.2f}")
# Expected output:
# contains adverse drug event: 0.94
Classify ADE Severity
Once an ADE is detected, classify how serious it is. A lower default threshold captures weaker signals that may still warrant review:
SEVERITY_LABELS = [
"life-threatening adverse event",
"serious adverse event requiring hospitalization",
"moderate adverse event requiring medical attention",
"mild adverse event",
"no adverse event described",
]
def classify_severity(text: str, threshold: float = 0.3) -> list[dict]:
"""Classify the severity of an adverse drug event."""
return pipeline(text, SEVERITY_LABELS, threshold=threshold)[0]
def get_highest_severity(results: list[dict]) -> tuple[str, float]:
"""Return the label with the highest score."""
if not results:
return ("unknown", 0.0)
best = max(results, key=lambda r: r["score"])
return (best["label"], best["score"])
# Example
text = (
"Patient experienced anaphylactic shock within minutes of "
"drug administration, requiring emergency epinephrine and ICU admission."
)
severity_results = classify_severity(text)
label, score = get_highest_severity(severity_results)
print(f"Severity: {label} (confidence: {score:.2f})")
# Expected output:
# Severity: life-threatening adverse event (confidence: 0.91)
Identify ADE Types
Classify the organ system or mechanism involved for structured pharmacovigilance reporting:
ADE_TYPE_LABELS = [
"allergic reaction or hypersensitivity",
"gastrointestinal adverse effect",
"cardiovascular adverse effect",
"neurological adverse effect",
"hepatotoxicity or liver injury",
"nephrotoxicity or kidney injury",
"hematological adverse effect",
"dermatological adverse effect",
"drug-drug interaction",
"overdose or toxicity",
]
def classify_ade_type(text: str, threshold: float = 0.4) -> list[dict]:
"""Classify the type of adverse drug event."""
return pipeline(text, ADE_TYPE_LABELS, threshold=threshold)[0]
# Example
text = (
"Following initiation of anticoagulant therapy, the patient "
"developed petechiae and easy bruising. Laboratory results showed "
"thrombocytopenia with platelet count of 45,000/uL."
)
for hit in classify_ade_type(text):
print(f" {hit['label']}: {hit['score']:.2f}")
# Expected output:
# hematological adverse effect: 0.89
Batch Processing for Literature Screening
Screen a collection of documents (e.g., PubMed abstracts or clinical notes) and flag those that contain adverse events:
from typing import Any
def screen_documents(
documents: list[dict[str, str]],
threshold: float = 0.5,
) -> list[dict[str, Any]]:
"""
Screen multiple documents for adverse drug events.
Args:
documents: List of dicts, each with 'id' and 'text' keys.
threshold: Minimum confidence for ADE detection.
Returns:
List of result dicts with ADE classifications.
"""
results: list[dict[str, Any]] = []
for doc in documents:
hits = pipeline(
doc["text"],
["contains adverse drug event", "no adverse drug event mentioned"],
threshold=threshold,
)[0]
hit_labels = {h["label"]: h["score"] for h in hits}
ade_detected = "contains adverse drug event" in hit_labels
result: dict[str, Any] = {
"id": doc["id"],
"ade_detected": ade_detected,
"scores": hit_labels,
}
if ade_detected:
sev = classify_severity(doc["text"])
result["severity"] = get_highest_severity(sev)
results.append(result)
return results
# Example
abstracts = [
{"id": "PMID_12345", "text": "A 45-year-old patient developed rhabdomyolysis after statin therapy..."},
{"id": "PMID_12346", "text": "The drug demonstrated efficacy with no significant adverse effects..."},
{"id": "PMID_12347", "text": "Three patients reported mild nausea that resolved without intervention..."},
]
screening_results = screen_documents(abstracts)
ade_docs = [r for r in screening_results if r["ade_detected"]]
print(f"Found {len(ade_docs)} document(s) with potential ADEs out of {len(abstracts)}")
Full ADE Monitoring Pipeline
The ADEMonitor class ties detection, severity classification, and type classification into a single reusable pipeline. It stores results as typed ADEReport dataclass instances and supports JSON export.
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Optional
import json
from gliclass import GLiClassModel, ZeroShotClassificationPipeline
from transformers import AutoTokenizer
# ---------------------------------------------------------------------------
# Label sets
# ---------------------------------------------------------------------------
DETECTION_LABELS = [
"contains adverse drug event or side effect",
"no adverse drug event mentioned",
]
SEVERITY_LABELS = [
"life-threatening or fatal",
"serious requiring hospitalization",
"moderate requiring treatment",
"mild or minor",
]
ADE_TYPE_LABELS = [
"allergic reaction or hypersensitivity",
"gastrointestinal adverse effect",
"cardiovascular adverse effect",
"neurological adverse effect",
"hepatotoxicity or liver injury",
"nephrotoxicity or kidney injury",
"hematological adverse effect",
"dermatological adverse effect",
"drug-drug interaction",
"overdose or toxicity",
]
# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass
class ADEReport:
"""Structured ADE report for pharmacovigilance."""
document_id: str
text: str
ade_detected: bool
confidence: float
severity: Optional[str] = None
severity_confidence: Optional[float] = None
ade_types: list[str] = field(default_factory=list)
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
def to_dict(self) -> dict:
data = asdict(self)
data["text_preview"] = (self.text[:200] + "...") if len(self.text) > 200 else self.text
del data["text"]
return data
# ---------------------------------------------------------------------------
# Pipeline
# ---------------------------------------------------------------------------
class ADEMonitor:
"""
Local adverse-drug-event monitoring pipeline.
Loads the GLiClass model once and reuses it for detection,
severity classification, and ADE type classification.
"""
def __init__(
self,
model_name: str = "knowledgator/gliclass-base-v3.0",
device: str = "cpu",
ade_threshold: float = 0.6,
):
model = GLiClassModel.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)
self.pipeline = ZeroShotClassificationPipeline(
model, tokenizer, classification_type="multi-label", device=device
)
self.ade_threshold = ade_threshold
self.reports: list[ADEReport] = []
# -- internal helpers ---------------------------------------------------
def _classify(
self, text: str, labels: list[str], threshold: float
) -> list[dict]:
return self.pipeline(text, labels, threshold=threshold)[0]
# -- public API ---------------------------------------------------------
def analyze_document(self, doc_id: str, text: str) -> ADEReport:
"""Run the full detection -> severity -> type pipeline on one document."""
# 1. Detect ADE presence
detection = self._classify(text, DETECTION_LABELS, threshold=0.3)
det_map = {r["label"]: r["score"] for r in detection}
ade_score = det_map.get("contains adverse drug event or side effect", 0.0)
ade_detected = ade_score >= self.ade_threshold
severity: Optional[str] = None
severity_conf: Optional[float] = None
ade_types: list[str] = []
if ade_detected:
# 2. Classify severity
sev_results = self._classify(text, SEVERITY_LABELS, threshold=0.3)
if sev_results:
best = max(sev_results, key=lambda r: r["score"])
severity = best["label"]
severity_conf = best["score"]
# 3. Identify ADE types
type_results = self._classify(text, ADE_TYPE_LABELS, threshold=0.4)
ade_types = [r["label"] for r in type_results]
report = ADEReport(
document_id=doc_id,
text=text,
ade_detected=ade_detected,
confidence=ade_score,
severity=severity,
severity_confidence=severity_conf,
ade_types=ade_types,
)
self.reports.append(report)
return report
def get_high_priority_alerts(self) -> list[ADEReport]:
"""Return reports that need immediate pharmacovigilance review."""
alerts: list[ADEReport] = []
for report in self.reports:
if not report.ade_detected:
continue
if report.severity in ("life-threatening or fatal", "serious requiring hospitalization"):
alerts.append(report)
elif report.confidence >= 0.85:
alerts.append(report)
return alerts
def export_reports(self, filepath: str) -> None:
"""Write all reports to a JSON file."""
data = [r.to_dict() for r in self.reports]
with open(filepath, "w") as f:
json.dump(data, f, indent=2)
# ---------------------------------------------------------------------------
# Usage
# ---------------------------------------------------------------------------
monitor = ADEMonitor(device="cpu", ade_threshold=0.6)
documents = [
("DOC001", "Patient developed Stevens-Johnson syndrome after taking the antibiotic..."),
("DOC002", "The medication was well-tolerated with no adverse effects reported..."),
("DOC003", "Mild headache reported by 3 patients, resolving within 24 hours..."),
]
for doc_id, text in documents:
report = monitor.analyze_document(doc_id, text)
if report.ade_detected:
print(f"ADE in {doc_id}: {report.severity} (confidence {report.confidence:.2f})")
alerts = monitor.get_high_priority_alerts()
print(f"\n{len(alerts)} high-priority alert(s) requiring review")
monitor.export_reports("ade_reports.json")
Best Practices
Tune thresholds for your use case
Use lower thresholds (0.3 -- 0.5) during screening to maximize recall. Raise the threshold (0.7+) when generating reports that feed into regulatory submissions.
SCREENING_THRESHOLD = 0.4 # catch more potential signals
REPORTING_THRESHOLD = 0.75 # higher confidence for formal reports
Create domain-specific labels
Generic labels work well for broad surveillance. For a specific therapeutic area, tailor your labels:
# Oncology
ONCOLOGY_ADE_LABELS = [
"chemotherapy-induced nausea and vomiting",
"myelosuppression or bone marrow toxicity",
"cardiotoxicity",
"nephrotoxicity",
"neurotoxicity or neuropathy",
"immunotherapy-related adverse event",
"infusion reaction",
]
# Cardiology
CARDIOLOGY_ADE_LABELS = [
"QT prolongation or arrhythmia",
"hypotension",
"bleeding event",
"heart failure exacerbation",
"bradycardia",
"drug-induced myocarditis",
]
Handle negation and speculation
Medical text frequently contains negated or speculative statements. Add a context-classification step to avoid false positives:
CONTEXT_LABELS = [
"confirmed adverse event occurred",
"adverse event ruled out or negated",
"potential or suspected adverse event",
"historical adverse event mentioned",
]
def classify_ade_context(text: str) -> list[dict]:
"""Distinguish confirmed vs. negated or speculative ADEs."""
return pipeline(text, CONTEXT_LABELS, threshold=0.4)[0]
# Example
text = "No evidence of hepatotoxicity was observed during treatment."
for r in classify_ade_context(text):
print(f" {r['label']}: {r['score']:.2f}")
# Expected output:
# adverse event ruled out or negated: 0.87
Implement a human review workflow
Automated classification should always feed into a human review process. A simple triage function:
def triage_for_review(report: ADEReport) -> str:
"""Assign a review priority level to an ADE report."""
if not report.ade_detected:
return "NO_REVIEW_NEEDED"
if report.severity == "life-threatening or fatal":
return "IMMEDIATE_REVIEW"
if report.severity == "serious requiring hospitalization":
return "URGENT_REVIEW"
if report.confidence >= 0.8:
return "STANDARD_REVIEW"
return "LOW_PRIORITY_REVIEW"
Limitations and Considerations
-
Not a replacement for expert review. Automated ADE detection should augment pharmacovigilance professionals, not replace them.
-
Context matters. The same sentence may or may not describe an ADE depending on surrounding context. Always review flagged documents.
-
Regulatory compliance. Ensure your pipeline meets FDA, EMA, or other applicable regulatory requirements before using it in production pharmacovigilance workflows.
-
Data privacy. When processing clinical notes, comply with HIPAA, GDPR, or other applicable privacy regulations. Running GLiClass locally helps keep data on-premises.
-
Model updates. Periodically evaluate classification performance as medical terminology and drug safety knowledge evolve. Consider fine-tuning on your own labeled ADE data for improved accuracy.
Next Steps
- Combine with Extract Biomedical Entities to identify specific drugs and conditions mentioned alongside ADEs.
- Use Link Medical Entities to Hospital Databases to connect detected ADEs to your drug master data.
- Schedule batch screening of new PubMed abstracts for continuous literature surveillance.
- Fine-tune
gliclass-base-v3.0on your own labeled ADE dataset for higher accuracy in your therapeutic domain.