How to Integrate LangGraph for healthcare with Redis for production AI

Combining LangGraph for healthcare with Redis gives you a practical production pattern: stateful clinical workflows with fast, durable memory. Use it when your agent needs to track patient intake, triage steps, prior tool calls, and short-lived session state without rebuilding everything on every request.

Redis handles the low-latency persistence layer. LangGraph for healthcare handles the orchestration layer, so your agent can move through clinical states deterministically and recover cleanly after retries or restarts.

Prerequisites

• •Python 3.10+
• •A Redis instance running locally or in your environment
• •langgraph installed
• •redis Python client installed
• •Access to your LangGraph healthcare workflow codebase
• •A configured model provider for any LLM nodes you use
• •Environment variables set for Redis connection details

Install the packages:

pip install langgraph redis pydantic

If you are using a managed Redis service, make sure you have:

• •Hostname
• •Port
• •Password
• •TLS settings if required

Integration Steps

1. •

   Set up Redis as the shared persistence layer

   Start by creating a Redis client that your graph can use for checkpoints, session state, or cached clinical context.

import os
import redis

redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", "6379")),
    password=os.getenv("REDIS_PASSWORD"),
    decode_responses=True,
)

print(redis_client.ping())

For production, keep this client in a singleton or dependency container. Do not create a new connection on every request.

2. •

   Define your LangGraph healthcare state model

   In healthcare workflows, keep state explicit. Use a typed schema so every node knows what it can read and write.

from typing import TypedDict, Optional, List

class PatientState(TypedDict):
    patient_id: str
    symptoms: List[str]
    triage_level: Optional[str]
    notes: Optional[str]
    redis_session_key: Optional[str]

This gives you deterministic state transitions and makes it easier to persist partial progress into Redis.

3. •

   Build the graph and persist intermediate state to Redis

   Here is a simple LangGraph workflow with two nodes: intake and triage. The intake node writes session data to Redis; the triage node reads from graph state and stores the result back in Redis.

from langgraph.graph import StateGraph, END

def intake_node(state: PatientState):
    session_key = f"patient:{state['patient_id']}:session"
    redis_client.hset(session_key, mapping={
        "symptoms": ",".join(state["symptoms"]),
        "notes": state.get("notes", ""),
        "status": "intake_complete",
    })
    return {
        **state,
        "redis_session_key": session_key,
    }

def triage_node(state: PatientState):
    symptoms = state["symptoms"]
    triage_level = "high" if any(s in ["chest pain", "shortness of breath"] for s in symptoms) else "routine"

    if state.get("redis_session_key"):
        redis_client.hset(state["redis_session_key"], mapping={
            "triage_level": triage_level,
            "status": "triage_complete",
        })

    return {
        **state,
        "triage_level": triage_level,
    }

workflow = StateGraph(PatientState)
workflow.add_node("intake", intake_node)
workflow.add_node("triage", triage_node)
workflow.set_entry_point("intake")
workflow.add_edge("intake", "triage")
workflow.add_edge("triage", END)

app = workflow.compile()

This is the core integration pattern: LangGraph manages execution flow, while Redis stores durable session artifacts that survive process restarts.

4. •

   Run the graph with a real patient session

   Pass in a patient payload and let the graph execute. After execution, verify that both graph output and Redis storage are aligned.

initial_state = {
    "patient_id": "12345",
    "symptoms": ["fever", "cough"],
    "triage_level": None,
    "notes": "Patient reports symptoms for 2 days",
    "redis_session_key": None,
}

result = app.invoke(initial_state)

print(result)
print(redis_client.hgetall("patient:12345:session"))

If you want retry-safe execution across requests, store a run ID in Redis and reload it before continuing the graph.

5. •

   Add checkpoint-style recovery for production

   For production AI systems, you want resumability. A common pattern is to store execution metadata in Redis so you can resume from the last known good step after failure.

import json

def save_checkpoint(run_id: str, state: dict):
    redis_client.set(f"checkpoint:{run_id}", json.dumps(state), ex=3600)

def load_checkpoint(run_id: str):
    raw = redis_client.get(f"checkpoint:{run_id}")
    return json.loads(raw) if raw else None

run_id = "run_abc_001"
save_checkpoint(run_id, initial_state)
restored_state = load_checkpoint(run_id)

print(restored_state["patient_id"])

In real systems, pair this with LangGraph’s durable execution features so node outputs can be replayed safely after interruption.

Testing the Integration

Run this end-to-end test to confirm that LangGraph executes correctly and Redis receives the expected data.

test_state = {
    "patient_id": "99999",
    "symptoms": ["headache"],
    "triage_level": None,
    "notes": "Mild symptoms",
    "redis_session_key": None,
}

output = app.invoke(test_state)
stored = redis_client.hgetall("patient:99999:session")

assert output["triage_level"] == "routine"
assert stored["status"] == "triage_complete"

print("Integration OK")
print(output["triage_level"])
print(stored)

Expected output:

Integration OK
routine
{'symptoms': 'headache', 'notes': 'Mild symptoms', 'status': 'triage_complete', 'triage_level': 'routine'}

Real-World Use Cases

• •

  Patient intake orchestration

  • •Collect symptoms, validate required fields, route to triage logic, and persist each step in Redis for auditability.
• •

  Clinical support agents with session memory

  • •Keep recent conversation context and workflow checkpoints in Redis so follow-up questions do not restart from zero.
• •

  Retry-safe care coordination workflows

  • •Resume interrupted scheduling, lab follow-up, or referral flows from the last saved checkpoint instead of reprocessing everything.

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.