How to Integrate LangGraph for banking with Redis for startups

LangGraph for banking gives you the orchestration layer for multi-step financial workflows: KYC checks, transaction review, policy lookup, fraud triage, and human handoff. Redis gives you the fast shared state you need to keep those agent runs coherent across retries, sessions, and concurrent users.

Together, they unlock a practical startup pattern: build banking agents that remember context, recover from failures, and move quickly without stuffing everything into your LLM prompt.

Prerequisites

• •Python 3.10+
• •A running Redis instance
  • •Local: redis-server
  • •Managed: Upstash Redis, AWS ElastiCache, Azure Cache for Redis
• •LangGraph installed
• •Redis Python client installed
• •Your LLM provider configured via environment variables
• •Banking-specific tools or APIs ready
  • •Example: account lookup API, transaction risk API, KYC verification API

Install the packages:

pip install langgraph redis langchain-openai python-dotenv

Set your environment variables:

export OPENAI_API_KEY="your-key"
export REDIS_URL="redis://localhost:6379/0"

Integration Steps

1. •Create a Redis connection and define shared state

Use Redis as the persistence layer for agent memory and run metadata. In practice, this means storing conversation state, checkpoints, or workflow status keyed by customer or session ID.

import os
import redis
from typing import TypedDict

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
r = redis.Redis.from_url(REDIS_URL, decode_responses=True)

class BankingState(TypedDict):
    customer_id: str
    message: str
    risk_score: int
    decision: str

# quick sanity check
r.set("banking:healthcheck", "ok")
print(r.get("banking:healthcheck"))

2. •Build a LangGraph banking workflow

Define a simple graph with nodes for risk scoring and decisioning. This is the core LangGraph pattern: each node transforms state, and edges control the workflow.

from langgraph.graph import StateGraph, END

def assess_risk(state: BankingState) -> BankingState:
    message = state["message"].lower()
    risk_score = 90 if "urgent transfer" in message or "new beneficiary" in message else 20
    return {**state, "risk_score": risk_score}

def decide(state: BankingState) -> BankingState:
    decision = "manual_review" if state["risk_score"] >= 70 else "approve"
    return {**state, "decision": decision}

workflow = StateGraph(BankingState)
workflow.add_node("assess_risk", assess_risk)
workflow.add_node("decide", decide)

workflow.set_entry_point("assess_risk")
workflow.add_edge("assess_risk", "decide")
workflow.add_edge("decide", END)

app = workflow.compile()

3. •Persist workflow state in Redis

LangGraph supports checkpointing patterns through checkpointers. For startup systems, the practical move is to store run snapshots in Redis so you can resume conversations and inspect decisions later.

If you want lightweight persistence without overengineering it on day one, write the final state into Redis after each run.

import json

def run_banking_agent(customer_id: str, message: str):
    initial_state = {
        "customer_id": customer_id,
        "message": message,
        "risk_score": 0,
        "decision": ""
    }

    result = app.invoke(initial_state)

    redis_key = f"banking:{customer_id}:latest_run"
    r.set(redis_key, json.dumps(result))

    return result

output = run_banking_agent(
    customer_id="cust_123",
    message="I need an urgent transfer to a new beneficiary"
)

print(output)

4. •Add Redis-backed session memory

For real deployments, don’t rely on one-off invocations. Store per-customer history in Redis so the agent can pick up context across multiple turns.

def append_message(customer_id: str, user_message: str):
    key = f"banking:{customer_id}:messages"
    r.rpush(key, user_message)
    return r.lrange(key, 0, -1)

history = append_message("cust_123", "Check my transfer status")
history = append_message("cust_123", "Also add a new beneficiary")

print(history)

You can then feed recent messages back into LangGraph before invoking the workflow.

def build_context(customer_id: str):
    key = f"banking:{customer_id}:messages"
    messages = r.lrange(key, -5, -1)
    return "\n".join(messages)

context = build_context("cust_123")

result = app.invoke({
    "customer_id": "cust_123",
    "message": context,
    "risk_score": 0,
    "decision": ""
})
print(result)

5. •Track status and retry safely

Redis is useful for idempotency keys and execution locks. In banking flows, that matters because duplicate transfers or repeated KYC checks are expensive mistakes.

import time

def process_once(customer_id: str, request_id: str, message: str):
    lock_key = f"lock:{request_id}"
    if not r.set(lock_key, "1", nx=True, ex=60):
        return {"status": "duplicate_request"}

    result = app.invoke({
        "customer_id": customer_id,
        "message": message,
        "risk_score": 0,
        "decision": ""
    })

    r.set(f"banking:{request_id}:result", json.dumps(result))
    return {"status": "processed", "result": result}

print(process_once("cust_123", "req_001", "Transfer to a new beneficiary"))

Testing the Integration

Run a basic end-to-end check with Redis storage plus LangGraph execution.

test_result = run_banking_agent(
    customer_id="cust_999",
    message="Please approve my standard bill payment"
)

cached_result = r.get("banking:cust_999:latest_run")
print("LangGraph result:", test_result)
print("Redis cached result:", cached_result)

Expected output:

LangGraph result: {'customer_id': 'cust_999', 'message': 'Please approve my standard bill payment', 'risk_score': 20, 'decision': 'approve'}
Redis cached result: {"customer_id": "cust_999", "message": "Please approve my standard bill payment", "risk_score": 20, "decision": "approve"}

Real-World Use Cases

• •

  Fraud triage agents

  • •Use LangGraph to route suspicious transactions through scoring, enrichment, and manual review.
  • •Use Redis to store short-lived case state and prevent duplicate processing.
• •

  KYC support assistants

  • •Let LangGraph orchestrate document collection, validation checks, and escalation.
  • •Use Redis to keep session context across multiple user interactions.
• •

  Banking ops copilots for startups

  • •Build internal tools that help support teams answer account questions faster.
  • •Use Redis as shared memory for active cases and LangGraph for deterministic workflow control.

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.