How to Integrate LangGraph for healthcare with Redis for RAG

Combining LangGraph for healthcare with Redis gives you a clean way to build stateful RAG agents that can handle clinical workflows without turning every request into a full re-index or a stateless chat loop. LangGraph manages the multi-step reasoning and branching, while Redis gives you fast retrieval, short-term memory, and vector search for patient-safe context.

Prerequisites

• •Python 3.10+
• •A running Redis Stack instance with:
  • •Redis Search
  • •Redis Vector Similarity
• •Access to your healthcare-focused LangGraph project or graph definition
• •OpenAI or another embedding provider configured
• •langgraph, redis, and your LLM/embedding SDK installed
• •A Redis index name for RAG documents, for example idx:clinical_docs

Install the core packages:

pip install langgraph redis openai pydantic

If you are using Redis as a vector store through LangChain helpers, install that too:

pip install langchain-community langchain-openai

Integration Steps

1. •

   Connect to Redis and create a vector index

   Use Redis as your retrieval layer. In healthcare, keep the documents scoped by tenant, facility, or department so your agent does not retrieve irrelevant records.

import os
from redis import Redis
from redis.commands.search.field import TextField, TagField, VectorField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType

redis_client = Redis.from_url(os.environ["REDIS_URL"])

index_name = "idx:clinical_docs"
prefix = "doc:"

schema = (
    TextField("content"),
    TagField("patient_id"),
    TagField("source"),
    VectorField(
        "embedding",
        "HNSW",
        {
            "TYPE": "FLOAT32",
            "DIM": 1536,
            "DISTANCE_METRIC": "COSINE",
        },
    ),
)

definition = IndexDefinition(prefix=[prefix], index_type=IndexType.HASH)

try:
    redis_client.ft(index_name).create_index(schema, definition=definition)
except Exception:
    pass  # index already exists in most deployments

2. •

   Embed and store healthcare documents in Redis

   Store chunks with metadata that matters in clinical workflows: patient ID, note type, date, and source system.

import json
import numpy as np
from openai import OpenAI

client = OpenAI()

def embed_text(text: str) -> list[float]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return resp.data[0].embedding

def save_document(doc_id: str, content: str, patient_id: str, source: str):
    embedding = np.array(embed_text(content), dtype=np.float32).tobytes()

    redis_client.hset(
        f"doc:{doc_id}",
        mapping={
            "content": content,
            "patient_id": patient_id,
            "source": source,
            "embedding": embedding,
        },
    )

save_document(
    doc_id="001",
    content="Patient reports chest pain on exertion. ECG ordered. Follow-up in cardiology.",
    patient_id="p-1001",
    source="ehr-note",
)

3. •

   Build a retriever function for RAG

   LangGraph does not retrieve on its own. You wire retrieval into the graph as a tool or node. This step queries Redis with vector similarity and returns top-k clinical context.

from redis.commands.search.query import Query

def retrieve_context(query_text: str, k: int = 3) -> list[dict]:
    query_vec = np.array(embed_text(query_text), dtype=np.float32).tobytes()

    q = (
        Query("*=>[KNN $k @embedding $vec AS score]")
        .sort_by("score")
        .return_fields("content", "patient_id", "source", "score")
        .paging(0, k)
        .dialect(2)
    )

    results = redis_client.ft(index_name).search(
        q,
        query_params={"k": k, "vec": query_vec},
    )

    docs = []
    for doc in results.docs:
      docs.append({
          "content": doc.content,
          "patient_id": doc.patient_id,
          "source": doc.source,
          "score": doc.score,
      })
    return docs

4. •

   Wire retrieval into a LangGraph workflow

   Use LangGraph’s StateGraph to create a small healthcare RAG pipeline: accept query, retrieve from Redis, generate answer.

from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from openai import OpenAI

llm_client = OpenAI()

class GraphState(TypedDict):
    question: str
    context: List[dict]
    answer: str

def retrieve_node(state: GraphState) -> GraphState:
    state["context"] = retrieve_context(state["question"], k=3)
    return state

def answer_node(state: GraphState) -> GraphState:
    context_text = "\n".join(
        [f"- {d['content']} (source={d['source']})" for d in state["context"]]
    )

    prompt = f"""
You are a healthcare assistant.
Use only the retrieved context.
If the answer is not in context, say you don't know.

Question: {state['question']}

Context:
{context_text}
"""

    resp = llm_client.responses.create(
        model="gpt-4o-mini",
        input=prompt,
    )
    state["answer"] = resp.output_text
    return state

workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("answer", answer_node)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "answer")
workflow.add_edge("answer", END)

app = workflow.compile()

5. •

   Run the graph and persist short-term conversation state in Redis

   For multi-turn healthcare workflows, store session state in Redis so follow-up questions stay grounded.

session_key = "session:p-1001"

result = app.invoke(
    {
        "question": "What follow-up was recommended for chest pain?",
        "context": [],
        "answer": "",
    }
)

redis_client.set(session_key, result["answer"], ex=3600)

print(result["answer"])

Testing the Integration

Run a simple end-to-end check that inserts one document, retrieves it through Redis, and answers through LangGraph.

test_query = {
    "question": "What was ordered for chest pain?",
    "context": [],
    "answer": "",
}

output = app.invoke(test_query)

print("Answer:", output["answer"])
print("Context:", output["context"])

Expected output:

Answer: The retrieved note indicates an ECG was ordered and cardiology follow-up was recommended.
Context: [{'content': 'Patient reports chest pain on exertion. ECG ordered. Follow-up in cardiology.', 'patient_id': 'p-1001', 'source': 'ehr-note', 'score': '...'}]

Real-World Use Cases

• •Clinical chart assistant
  • •Retrieve relevant notes from Redis and use LangGraph to ask follow-up questions before drafting a response.
• •Prior authorization support
  • •Pull policy snippets, lab history, and encounter notes into one graph so the agent can generate evidence-backed summaries.
• •Care coordination workflow
  • •Route between retrieval, summarization, escalation, and human review nodes based on what Redis returns.

If you want this production-ready for healthcare, add PHI filtering before indexing, tenant-aware namespaces in Redis keys, audit logs for every retrieval call, and explicit human approval nodes in LangGraph before any downstream action.

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.