How to Integrate LangGraph for retail banking with Redis for startups

Combining LangGraph for retail banking with Redis gives you a practical way to run stateful banking agents without turning every request into a cold start. LangGraph handles the multi-step decision flow for customer servicing, fraud checks, and policy routing, while Redis gives you fast shared memory for session state, rate limits, and short-lived context across workers.

For startups, that means you can build agents that remember account-level context, survive restarts, and coordinate across multiple API calls without bolting on a heavy orchestration layer.

Prerequisites

• •Python 3.10+
• •A Redis instance running locally or in the cloud
• •langgraph installed
• •redis Python client installed
• •Access to your banking LLM/tooling stack
• •Environment variables set for Redis connection details

Install the dependencies:

pip install langgraph redis python-dotenv

Set your Redis URL:

export REDIS_URL="redis://localhost:6379/0"

Integration Steps

1. •Create a Redis client and verify connectivity

Start by connecting to Redis directly. In production, this is where you store conversation checkpoints, customer session metadata, and temporary banking workflow state.

import os
import redis

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")

r = redis.from_url(REDIS_URL, decode_responses=True)

# Basic connectivity check
print(r.ping())

If this prints True, your app can talk to Redis. For startup systems, keep this client in a shared module so every worker uses the same connection settings.

2. •Define a LangGraph state model for a retail banking flow

LangGraph works best when your workflow state is explicit. For retail banking, keep fields like customer intent, account id, risk flags, and final response in the graph state.

from typing import TypedDict, Annotated
from operator import add

class BankingState(TypedDict):
    messages: Annotated[list[str], add]
    customer_id: str
    intent: str
    risk_flag: bool
    response: str

This structure gives you deterministic state transitions. You can extend it later with KYC status, product eligibility, or dispute case IDs.

3. •Build the graph nodes and persist session data in Redis

Use LangGraph’s StateGraph to define nodes for intent routing and response generation. Then write the latest state into Redis so the next request can resume from where it left off.

import json
from langgraph.graph import StateGraph, END

def classify_intent(state: BankingState):
    text = " ".join(state["messages"]).lower()
    if "card" in text or "fraud" in text:
        intent = "fraud_support"
        risk_flag = True
    else:
        intent = "general_banking"
        risk_flag = False

    return {
        "intent": intent,
        "risk_flag": risk_flag,
    }

def generate_response(state: BankingState):
    if state["intent"] == "fraud_support":
        response = f"Customer {state['customer_id']}: route to fraud ops and freeze card if needed."
    else:
        response = f"Customer {state['customer_id']}: answer general banking request."

    return {"response": response}

workflow = StateGraph(BankingState)
workflow.add_node("classify_intent", classify_intent)
workflow.add_node("generate_response", generate_response)
workflow.set_entry_point("classify_intent")
workflow.add_edge("classify_intent", "generate_response")
workflow.add_edge("generate_response", END)

app = workflow.compile()

session_key = "banking-session:12345"
initial_state = {
    "messages": ["My debit card was charged twice"],
    "customer_id": "12345",
    "intent": "",
    "risk_flag": False,
    "response": ""
}

result = app.invoke(initial_state)

# Persist result to Redis for replay/resume/debugging
r.set(session_key, json.dumps(result), ex=3600)

print(result)

This pattern is enough for most startup-grade retail banking assistants. If you need stronger durability later, swap simple set() storage for LangGraph checkpointing backed by Redis-compatible persistence.

4. •Resume agent context from Redis on the next request

A banking assistant needs continuity. Pull prior state from Redis before invoking the graph again so follow-up questions stay anchored to the same customer session.

import json

def load_session(session_key: str):
    raw = r.get(session_key)
    if not raw:
        return None
    return json.loads(raw)

previous_state = load_session("banking-session:12345")

if previous_state:
    follow_up_state = {
        "messages": previous_state["messages"] + ["What happened to my refund?"],
        "customer_id": previous_state["customer_id"],
        "intent": previous_state["intent"],
        "risk_flag": previous_state["risk_flag"],
        "response": ""
    }
else:
    follow_up_state = initial_state

next_result = app.invoke(follow_up_state)
print(next_result["response"])

This is the basic shape of multi-turn support in retail banking. The graph stays stateless at runtime, while Redis keeps session continuity across requests and pods.

5. •Add TTL-based cleanup for sensitive banking context

Do not keep customer session data forever. Use TTLs so transient conversational data expires automatically after your compliance window.

ttl_seconds = 1800  # 30 minutes

r.set(
    f"banking-session:{initial_state['customer_id']}",
    json.dumps(result),
    ex=ttl_seconds,
)

exists_now = r.exists(f"banking-session:{initial_state['customer_id']}")
print({"stored": bool(exists_now), "ttl": r.ttl(f"banking-session:{initial_state['customer_id']}")})

For regulated workflows, keep only what you need. Long-term records should live in your system of record, not in ephemeral agent memory.

Testing the Integration

Run a quick end-to-end check that confirms both the graph execution and Redis persistence work.

test_input = {
    "messages": ["I think my card was used without permission"],
    "customer_id": "90001",
    "intent": "",
    "risk_flag": False,
    "response": ""
}

output = app.invoke(test_input)
session_key = f"banking-session:{test_input['customer_id']}"
r.set(session_key, json.dumps(output), ex=600)

loaded = json.loads(r.get(session_key))
print("Intent:", loaded["intent"])
print("Risk flag:", loaded["risk_flag"])
print("Response:", loaded["response"])

Expected output:

Intent: fraud_support
Risk flag: True
Response: Customer 90001: route to fraud ops and freeze card if needed.

If that matches, your LangGraph flow is producing structured decisions and Redis is storing retrievable session state correctly.

Real-World Use Cases

• •Fraud triage assistants that classify suspicious activity, store case context in Redis, and route customers through escalation paths.
• •Retail support copilots that remember account-specific context across multiple turns without reloading everything from scratch.
• •Eligibility workflows for loans or savings products where each step depends on prior answers cached briefly in Redis.

If you want this pattern to hold up in production, keep LangGraph responsible for orchestration and decision logic, and use Redis strictly as fast ephemeral state. That separation keeps your agent system simpler to debug when finance teams start asking why a customer got routed down a specific path.

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.