How AI agents route patient messages to the right team

Healthcare contact centers drown in unstructured patient messages: prescription refills, billing disputes, symptom descriptions. AI agents route patient messages to the right team by classifying intent and urgency, then handing off to deterministic workflows. This guide walks through building that pipeline end to end with runnable Python and an OpenAI-compatible LLM endpoint.

Step 1: Map teams and enforce a strict output contract

Before writing any model code, pin down the destination teams. In most clinics those are clinical (symptoms, medication questions), billing, scheduling, pharmacy, and an explicit urgent bucket for anything life-threatening or requiring same-day clinician review.

AI agents route patient messages reliably only when the LLM output is machine-verifiable. Define a JSON schema and reject anything that doesn’t validate. A strict enum prevents the model from inventing queue names that your downstream code can’t handle.

{
  "type": "object",
  "properties": {
    "team": {
      "type": "string",
      "enum": ["clinical", "billing", "scheduling", "pharmacy", "urgent"]
    },
    "urgency": { "type": "integer", "minimum": 1, "maximum": 5 },
    "confidence": { "type": "number", "minimum": 0, "maximum": 1 },
    "reason": { "type": "string" }
  },
  "required": ["team", "urgency", "confidence", "reason"]
}

HIPAA doesn’t forbid LLM triage, but you must ensure no unprotected PHI leaves your boundary. The schema above returns only labels and a short reason string—never the patient’s narrative. Keep the raw text in your own store, correlated by a hashed message ID.

Step 2: Normalize ingestion from any channel

Patient messages arrive from portal web forms, SMS, and FHIR-based EHR exports. Normalize them into a single shape before classification. This decouples triage logic from transport concerns and makes testing straightforward.

def normalize(raw: dict) -> dict:
    return {
        "patient_id": raw.get("patient_id") or raw.get("fhir_id"),
        "text": raw["message"].strip()[:4000],
        "channel": raw.get("channel", "web"),
        "received_at": raw.get("ts")
    }

Truncate to a sane limit. Most triage decisions survive on the first 1–2k characters; the rest is boilerplate. Log only a salted hash of patient_id in production metrics. Store the original message separately if you need a full audit trail for compliance.

Step 3: Run LLM classification with a JSON schema

Call an LLM with the schema from Step 1. Using an OpenAI-compatible gateway such as n4n.ai gives you automatic fallback when a provider is rate-limited and per-token metering without custom code, which matters when uptime of triage directly affects patient safety.

from openai import OpenAI
import os, json

client = OpenAI(
    base_url="https://api.n4n.ai/v1",
    api_key=os.environ["N4N_API_KEY"]
)

SYSTEM = "You are a triage classifier for an outpatient clinic. Output strict JSON."
SCHEMA = { ... }  # from Step 1

def classify(msg: dict) -> dict:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": json.dumps(msg)}
        ],
        response_format={"type": "json_schema", "json_schema": {"name": "triage", "schema": SCHEMA}}
    )
    data = json.loads(resp.choices[0].message.content)
    return data

Small models are enough for triage; reserve larger ones for summarization or complex clinical reasoning. Include two or three few-shot examples in the system prompt showing intent boundaries: “I think my copay was wrong” maps to billing; “I ran out of lisinopril” maps to pharmacy. The model handles edge cases better when you demonstrate the edges.

Validate the response with jsonschema before trusting it. If validation fails, treat it as a routing miss and send to human review.

Step 4: Dispatch to team-specific handlers

Routing is deterministic. Once you have a validated team label, push the message to the correct queue. Do not let the LLM call side-effecting tools directly in production—keep the agent stateless and let your code own the handoff.

def publish(queue: str, msg: dict, idempotency_key: str):
    # SQS/Kafka/Redis publish with dedup key
    print(f"PUBLISH {queue}: {idempotency_key}")

handlers = {
    "clinical": lambda m, k: publish("queue.clinical", m, k),
    "billing": lambda m, k: publish("queue.billing", m, k),
    "scheduling": lambda m, k: publish("queue.scheduling", m, k),
    "pharmacy": lambda m, k: publish("queue.pharmacy", m, k),
}

def route(msg: dict, result: dict, idem_key: str):
    if result["urgency"] >= 4 or result["team"] == "urgent":
        publish("queue.human_triage", msg, idem_key)
        return
    handlers[result["team"]](msg, idem_key)

Idempotency keys prevent duplicate sends when a provider retry delivers the same message twice. AI agents route patient messages, but your infrastructure decides what “route” means and guarantees exactly-once delivery to the team queue.

Step 5: Add confidence thresholds and human fallback

Models are confident even when wrong. Set a floor. If confidence is below 0.7, or the message mentions chest pain, suicidal ideation, or uncontrolled bleeding, bypass the auto-queue and page a human.

def route_safe(msg: dict, result: dict, idem_key: str):
    high_risk_terms = ["chest pain", "suicidal", "uncontrolled bleeding"]
    if any(term in msg["text"].lower() for term in high_risk_terms):
        publish("queue.human_triage", msg, idem_key)
        return
    if result["confidence"] < 0.7:
        publish("queue.human_triage", msg, idem_key)
        return
    route(msg, result, idem_key)

When AI agents route patient messages at scale, the cost of a misroute is a delayed clinical response. A human-in-the-loop fallback is not optional; it’s the only defensible design. Define an SLA: urgent queue items must be acknowledged by a clinician within 5 minutes during business hours, 15 off-hours.

Step 6: Verify the pipeline

You cannot ship triage blind. Build a labeled set of 200–500 historical messages with known correct teams. Run the classifier offline and compute precision/recall per team.

def test_classifier_on_labels(labeled):
    correct = 0
    for msg, expected_team in labeled:
        res = classify(normalize(msg))
        if res["team"] == expected_team:
            correct += 1
    return correct / len(labeled)

Add a CI job that fails if overall accuracy drops below 0.9 on that set. In production, log every reason string and sample 5% of routed messages for manual review. If the pharmacy queue suddenly receives “I have a fever” messages, your schema validation passed but the model drifted—catch it via log alerts.

Verifying success

Success means: (1) every message ends up in exactly one queue, (2) urgent items hit human triage within seconds, (3) classifier accuracy on the labeled set stays above threshold for 30 days. Stand up a dead-letter queue for validation failures and alert if its depth exceeds zero. That’s your signal the LLM returned malformed JSON or a team enum drifted.

Build the agent stateless, keep routing explicit, and treat the model as a callable that converts text to a signed struct. Do that, and AI agents route patient messages without becoming a liability.