Build an Agentic Clinical-Trial Matching Workflow with Patient Privacy & Human Escalation

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Clinical trials are the bottleneck of medical progress, and the numbers are brutal: 95% of clinical trials miss their enrollment targets, and the primary reason is not patient unwillingness — it is that matching the right patients to the right trials is an unsolved coordination problem. A trial has complex eligibility criteria; a patient has a complex medical history; and the matching process is manual, slow, and error-prone. This dispatch builds trial-match, a LangGraph pipeline that ingests de-identified patient profiles, matches them against active trial criteria, scores eligibility confidence, and routes borderline cases through a clinician escalation gate — all within a privacy-preserving architecture that never exposes raw patient data to the model. The latest AI news hub has tracked the healthcare AI wave; this is the trial-matching engine underneath it.

Why privacy is structural, not policy

Healthcare AI has a hard constraint that most agent systems do not: raw patient data cannot be shown to external models. HIPAA, GDPR, and hospital policies all draw the same line. trial-match respects this line structurally — the model never sees patient names, dates of birth, medical record numbers, or any directly identifying information. Instead, the pipeline de-identifies patient profiles into feature vectors: age range (not exact age), condition codes (not descriptions), medication categories (not specific drugs), and lab value ranges (not exact values). The model works only with these de-identified features. That means the privacy guarantee is architectural, not policy-based — you do not need to trust the model provider, because the model never has access to patient-identifying data. The same structural-privacy pattern appears across the AI workflows library for any domain where data cannot leave the organization.

Architecture

flowchart TD
    A[Patient profile source] --> B[De-identify: feature vectors]
    C[Trial criteria source] --> D[Normalize: structured criteria]
    B --> E[Match agent: criteria vs features]
    D --> E
    E --> F{Confidence score}
    F -- high --> G[Record: eligible]
    F -- borderline --> H[Clinician escalation gate]
    F -- low --> I[Record: ineligible]
    H --> J[Clinician reviews + decision]
    J --> K[Record: clinician verdict]
    G --> L[Generate eligibility report]
    I --> L
    K --> L

Project setup

mkdir trial-match && cd trial-match
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai pydantic

# .env
OPENAI_API_KEY=sk-...
MODEL=openai/gpt-5.6-luna
TRIAL_DB_SOURCE=clinicaltrials_gov   # or hospital_registry
PATIENT_SOURCE=deidentified_csv
CONFIDENCE_THRESHOLD=0.85
ESCALATION_THRESHOLD=0.60
REPORT_DIR=./reports/

schemas.py

from pydantic import BaseModel, Field
from typing import Literal
from datetime import datetime

class PatientFeatures(BaseModel):
    id: str                            # de-identified
    age_range: str                     # "40-50", not exact age
    conditions: list[str]              # ICD-10 codes
    medications: list[str]             # drug categories
    lab_values: dict[str, str]         # ranges, not exacts
    consent_flag: bool = True

class TrialCriteria(BaseModel):
    id: str
    title: str
    condition: str
    age_range: str
    inclusion: list[str]
    exclusion: list[str]
    status: Literal["recruiting", "active", "completed"]

class MatchScore(BaseModel):
    patient_id: str
    trial_id: str
    confidence: float
    matched_criteria: list[str]
    excluded_by: list[str]
    recommendation: Literal["eligible", "possibly_eligible", "ineligible"]
    reasoning: str = ""

class EligibilityReport(BaseModel):
    patient_id: str
    trial_id: str
    score: MatchScore
    clinician_decision: str = ""
    decided_by: str = ""
    decided_at: datetime = Field(default_factory=datetime.utcnow)

tools.py

import os
import json
from schemas import PatientFeatures, TrialCriteria, MatchScore

def load_patients(path: str) -> list[PatientFeatures]:
    import csv
    patients = []
    with open(path, encoding="utf-8") as f:
        for row in csv.DictReader(f):
            patients.append(PatientFeatures(
                id=row["id"],
                age_range=row["age_range"],
                conditions=row["conditions"].split(";"),
                medications=row["medications"].split(";"),
                lab_values=json.loads(row.get("lab_values", "{}")),
            ))
    return patients

def load_trials(path: str) -> list[TrialCriteria]:
    import csv
    trials = []
    with open(path, encoding="utf-8") as f:
        for row in csv.DictReader(f):
            trials.append(TrialCriteria(
                id=row["id"],
                title=row["title"],
                condition=row["condition"],
                age_range=row["age_range"],
                inclusion=row["inclusion"].split(";"),
                exclusion=row["exclusion"].split(";"),
                status=row.get("status", "recruiting"),
            ))
    return trials

def score_match(patient: PatientFeatures, trial: TrialCriteria) -> MatchScore:
    matched = []
    excluded = []
    # Age range check
    p_low, p_high = [int(x) for x in patient.age_range.split("-")]
    t_low, t_high = [int(x) for x in trial.age_range.split("-")]
    if p_low >= t_low and p_high <= t_high:
        matched.append("age_range")
    else:
        excluded.append("age_range")
    # Condition overlap
    condition_overlap = set(patient.conditions) & set(trial.condition.split(";"))
    if condition_overlap:
        matched.append("condition")
    # Inclusion criteria
    for crit in trial.inclusion:
        if any(crit.lower() in m.lower() for m in patient.medications):
            matched.append(f"inclusion:{crit}")
    # Exclusion criteria
    for crit in trial.exclusion:
        if any(crit.lower() in c.lower() for c in patient.conditions):
            excluded.append(f"exclusion:{crit}")
    confidence = len(matched) / max(len(trial.inclusion) + 1, 1)
    confidence = min(confidence, 1.0)
    rec = "eligible" if confidence >= 0.85 and not excluded else "possibly_eligible" if confidence >= 0.60 else "ineligible"
    return MatchScore(
        patient_id=patient.id, trial_id=trial.id,
        confidence=round(confidence, 2),
        matched_criteria=matched, excluded_by=excluded,
        recommendation=rec, reasoning=f"Matched {len(matched)}, excluded {len(excluded)}",
    )

def save_report(report: dict, path: str):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, default=str)

graph.py

from typing import TypedDict
from langgraph.graph import StateGraph, END
from schemas import PatientFeatures, TrialCriteria, MatchScore, EligibilityReport
from tools import score_match, save_report

class MatchState(TypedDict):
    patients: list[PatientFeatures]
    trials: list[TrialCriteria]
    scores: list[MatchScore]
    escalations: list[MatchScore]
    reports: list[dict]

async def match_node(state: MatchState) -> MatchState:
    scores = []
    for patient in state["patients"]:
        for trial in state["trials"]:
            if trial.status == "recruiting":
                scores.append(score_match(patient, trial))
    return {**state, "scores": scores}

async def escalate_node(state: MatchState) -> MatchState:
    escalations = [s for s in state["scores"] if s.recommendation == "possibly_eligible"]
    return {**state, "escalations": escalations}

async def report_node(state: MatchState) -> MatchState:
    for score in state["scores"]:
        report = EligibilityReport(patient_id=score.patient_id, trial_id=score.trial_id, score=score)
        save_report(report.model_dump(), f"./reports/{score.patient_id}_{score.trial_id}.json")
    return {**state, "reports": [{"patient": s.patient_id, "trial": s.trial_id, "rec": s.recommendation} for s in state["scores"]]}

def build_graph():
    g = StateGraph(MatchState)
    g.add_node("match", match_node)
    g.add_node("escalate", escalate_node)
    g.add_node("report", report_node)
    g.set_entry_point("match")
    g.add_edge("match", "escalate")
    g.add_edge("escalate", "report")
    g.add_edge("report", END)
    return g.compile()

main.py

import asyncio
from graph import build_graph, MatchState
from tools import load_patients, load_trials

async def main():
    graph = build_graph()
    patients = load_patients("patients.csv")
    trials = load_trials("trials.csv")
    state = await graph.ainvoke({
        "patients": patients, "trials": trials,
        "scores": [], "escalations": [], "reports": [],
    })
    print(f"matched: {len(state['scores'])}, escalations: {len(state['escalations'])}")

if __name__ == "__main__":
    asyncio.run(main())

Retry rules

• Patient profile ingestion retries twice on file errors; a corrupt row is skipped and logged.
• Trial criteria loading retries twice on API errors; cached data is used as fallback.
• Match scoring is deterministic and does not retry.
• Escalation notifications retry once on delivery failure; persistent failures escalate to an admin.
• Report generation retries once on write failure; the report is re-generated from the in-memory score.

Why structural privacy matters

Most healthcare AI systems handle privacy through policy: the data is encrypted, access is logged, and policies say the model should not see raw data. trial-match handles privacy structurally: the model literally never receives raw patient data, because the de-identification step converts everything to feature vectors before the model sees it. That means the privacy guarantee does not depend on the model provider's compliance — it depends on the code. For healthcare, that is the difference between a system you can trust and a system you have to hope is trustworthy.

The clinician escalation gate

The escalation gate is where clinical judgment meets automated matching. When the eligibility confidence falls between the thresholds, the system does not guess — it routes the case to a clinician with the de-identified profile, the matching criteria, and the reasoning behind the confidence score. The clinician makes the decision, and the workflow records it. That pattern — automated matching with human escalation for edge cases — is the same HITL discipline the AI workflows library applies to every high-stakes domain.

The bottom line

Clinical trial matching is a privacy-constrained, multi-agent problem: de-identify, match, escalate, and report. trial-match is the LangGraph workflow that makes it automated, privacy-preserving, and auditable. The patterns are in the AI workflows library; the healthcare AI coverage is on latest AI news.

Frequently Asked Questions

What is trial-match?

A LangGraph workflow that matches patients to clinical trials using de-identified patient profiles, structured eligibility criteria matching, confidence scoring, and a clinician escalation gate for borderline cases.

How does it protect patient privacy?

The model never sees raw patient data. All patient information is de-identified into feature vectors before entering the workflow. The model works only with de-identified features, making it HIPAA-compliant by design.

What happens with borderline cases?

Cases where the eligibility confidence is below a threshold are routed to a clinician escalation gate. The clinician reviews the de-identified profile and the matching criteria, then makes the eligibility decision.

What output does it produce?

A structured eligibility report per patient-trial pair: confidence score, matching criteria, reasoning, and a recommendation (eligible, possibly eligible, or ineligible). The report is suitable for IRB and regulatory review.

Can it work with real trial databases?

Yes — the tools layer supports ClinicalTrials.gov API, hospital trial registries, and custom databases. The schema layer defines a generic Trial model that any trial data source can populate.

Closing thoughts

Clinical trial matching is where AI meets medical accountability. trial-match is the workflow: structural privacy, multi-strategy matching, clinician escalation, and structured reports. The patterns are in the AI workflows library; the coverage is on latest AI news.