How to Integrate LangGraph for healthcare with Redis for multi-agent systems

If you’re building healthcare agents, you need two things working together: stateful orchestration and durable shared memory. LangGraph gives you the control flow for multi-agent healthcare workflows, and Redis gives you fast, low-latency state storage for coordination, session context, and handoffs between agents.

This combination is useful when one agent triages symptoms, another checks policy or care pathways, and a third drafts a clinician-facing summary. Redis keeps the shared state consistent across steps, while LangGraph keeps the workflow deterministic enough for regulated environments.

Prerequisites

• •Python 3.10+
• •A running Redis instance
  • •Local: redis-server
  • •Docker: docker run -p 6379:6379 redis:7
• •Installed packages:
  • •langgraph
  • •redis
  • •langchain-openai or your preferred model provider
• •Access to your healthcare-oriented LLM setup
• •A basic understanding of:
  • •LangGraph StateGraph
  • •Redis key/value storage
  • •Python async/sync execution

Install dependencies:

pip install langgraph redis langchain-openai

Integration Steps

1) Define the shared healthcare state

Keep the graph state small and explicit. For healthcare workflows, store only what each agent needs: patient context, triage result, routing decision, and final summary.

from typing import TypedDict, Annotated
from operator import add

class HealthcareState(TypedDict):
    patient_id: str
    symptoms: str
    triage_level: str
    care_pathway: str
    summary: str
    notes: Annotated[list[str], add]

This pattern works well because LangGraph can pass structured state between nodes without relying on hidden globals. In production, that makes debugging and audit trails much easier.

2) Set up Redis as the shared persistence layer

Use Redis for cross-agent memory and workflow checkpoints. The simplest pattern is to store a per-patient session object keyed by patient ID.

import json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def save_patient_context(patient_id: str, payload: dict) -> None:
    key = f"healthcare:{patient_id}:context"
    r.set(key, json.dumps(payload))

def load_patient_context(patient_id: str) -> dict:
    key = f"healthcare:{patient_id}:context"
    raw = r.get(key)
    return json.loads(raw) if raw else {}

For multi-agent systems, this is the handoff point. One agent writes its result to Redis; another agent reads it later without needing to stay in memory on the same process.

3) Build LangGraph nodes that read from and write to Redis

Here’s a practical three-node flow:

• •triage agent
• •routing agent
• •summarization agent

Each node updates both the LangGraph state and the Redis record.

from langgraph.graph import StateGraph, START, END

def triage_agent(state: HealthcareState) -> HealthcareState:
    symptoms = state["symptoms"].lower()

    if "chest pain" in symptoms or "shortness of breath" in symptoms:
        triage_level = "urgent"
    elif "fever" in symptoms or "cough" in symptoms:
        triage_level = "routine"
    else:
        triage_level = "review"

    save_patient_context(state["patient_id"], {
        "symptoms": state["symptoms"],
        "triage_level": triage_level,
    })

    return {
        **state,
        "triage_level": triage_level,
        "notes": [f"Triage assigned: {triage_level}"],
    }

def routing_agent(state: HealthcareState) -> HealthcareState:
    context = load_patient_context(state["patient_id"])
    triage_level = context.get("triage_level", state["triage_level"])

    if triage_level == "urgent":
        care_pathway = "escalate_to_clinician"
    elif triage_level == "routine":
        care_pathway = "schedule_telehealth"
    else:
        care_pathway = "request_more_info"

    save_patient_context(state["patient_id"], {
        **context,
        "care_pathway": care_pathway,
    })

    return {
        **state,
        "care_pathway": care_pathway,
        "notes": state["notes"] + [f"Routing decided: {care_pathway}"],
    }

def summary_agent(state: HealthcareState) -> HealthcareState:
    summary = (
        f"Patient {state['patient_id']} with symptoms '{state['symptoms']}' "
        f"was classified as {state['triage_level']} and routed to {state['care_pathway']}."
    )

    save_patient_context(state["patient_id"], {
        **load_patient_context(state["patient_id"]),
        "summary": summary,
    })

    return {
        **state,
        "summary": summary,
        "notes": state["notes"] + ["Summary generated"],
    }

This is where LangGraph shines. You get a clear node-by-node workflow instead of a pile of ad hoc function calls.

4) Wire the graph together with LangGraph APIs

Use StateGraph, add your nodes with add_node, define edges with add_edge, then compile it.

workflow = StateGraph(HealthcareState)

workflow.add_node("triage", triage_agent)
workflow.add_node("route", routing_agent)
workflow.add_node("summarize", summary_agent)

workflow.add_edge(START, "triage")
workflow.add_edge("triage", "route")
workflow.add_edge("route", "summarize")
workflow.add_edge("summarize", END)

app = workflow.compile()

If you want durable execution across requests, pair this with a checkpointer. In Redis-backed deployments, that means each step can be resumed after interruption instead of starting over.

5) Execute the graph and persist the final output

Run the graph with a real patient payload. The output is available both in-memory and in Redis for downstream agents or services.

initial_state = {
    "patient_id": "p-10021",
    "symptoms": "Patient reports chest pain and shortness of breath",
    "triage_level": "",
    "care_pathway": "",
    "summary": "",
    "notes": [],
}

result = app.invoke(initial_state)

print(result["summary"])
print(result["notes"])

stored = load_patient_context("p-10021")
print(stored)

At this point you have a working multi-agent pipeline with shared persistence. One service can trigger the graph, another can inspect Redis for the latest clinical routing decision.

Testing the Integration

Run a quick verification against Redis and the compiled graph:

test_state = {
    "patient_id": "p-20001",
    "symptoms": "Fever and cough for three days",
    "triage_level": "",
    "care_pathway": "",
    "summary": "",
    "notes": [],
}

result = app.invoke(test_state)
cached = load_patient_context("p-20001")

print(result["triage_level"])
print(result["care_pathway"])
print(cached["summary"])

Expected output:

routine
schedule_telehealth
Patient p-20001 with symptoms 'Fever and cough for three days' was classified as routine and routed to schedule_telehealth.

If that prints correctly, your LangGraph workflow is executing and Redis is persisting shared agent state.

Real-World Use Cases

• •

  Symptom intake + escalation

  • •One agent collects symptoms.
  • •Another classifies urgency.
  • •A third routes to telehealth or clinician review.
  • •Redis stores the patient session so multiple services can continue the same case.
• •

  Prior authorization support

  • •An intake agent extracts procedure details.
  • •A policy agent checks coverage rules.
  • •A documentation agent drafts submission notes.
  • •Redis keeps claims context available across retries and asynchronous processing.
• •

  Care coordination workflows

  • •One agent identifies next steps after discharge.
  • •Another schedules follow-up tasks.
  • •Another monitors unresolved items.
  • •LangGraph controls sequencing while Redis tracks task status across workers.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.