How to Integrate LangGraph for retail banking with Redis for multi-agent systems

Combining LangGraph for retail banking with Redis gives you a practical way to run multi-agent banking workflows with shared state, routing, and short-lived memory. In retail banking, that usually means one agent can handle customer intent, another can fetch account context, and a third can enforce policy or escalation rules without losing the thread.

Redis fills the gap between graph execution and distributed coordination. You get fast state storage, message passing, session persistence, and a clean path to scale agent workers horizontally.

Prerequisites

• •Python 3.10+
• •A LangGraph-compatible banking workflow project already set up
• •Redis 7.x running locally or via managed service
• •langgraph, redis, and langchain-core installed
• •Access to your banking tools or mock services:
  • •account lookup API
  • •transaction history API
  • •fraud/risk policy service
• •Environment variables configured:
  • •REDIS_URL
  • •any bank API keys or internal service credentials

Install the core packages:

pip install langgraph redis langchain-core

Integration Steps

1. •Set up Redis as the shared state backend for agent coordination.

Use Redis for session state, task queues, or event storage depending on your architecture. For a simple multi-agent setup, start with a Redis client and namespace keys by customer session.

import os
import redis

redis_client = redis.Redis.from_url(
    os.environ["REDIS_URL"],
    decode_responses=True,
)

session_id = "retail-banking:customer-123"
redis_client.hset(
    f"{session_id}:state",
    mapping={
        "intent": "balance_inquiry",
        "status": "started",
        "agent": "router",
    },
)

print(redis_client.hgetall(f"{session_id}:state"))

2. •Define your LangGraph workflow with nodes for routing and banking actions.

LangGraph gives you a graph-based orchestration layer. For retail banking, use separate nodes for classification, account lookup, policy checks, and response assembly.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class BankingState(TypedDict):
    customer_id: str
    intent: str
    account_balance: str
    risk_flag: bool
    response: str

def route_intent(state: BankingState) -> BankingState:
    # Replace with your classifier or LLM call
    state["intent"] = "balance_inquiry"
    return state

def fetch_balance(state: BankingState) -> BankingState:
    # Replace with real bank API call
    state["account_balance"] = "$4,250.18"
    return state

def finalize_response(state: BankingState) -> BankingState:
    state["response"] = f"Your current balance is {state['account_balance']}."
    return state

graph = StateGraph(BankingState)
graph.add_node("route_intent", route_intent)
graph.add_node("fetch_balance", fetch_balance)
graph.add_node("finalize_response", finalize_response)

graph.add_edge(START, "route_intent")
graph.add_edge("route_intent", "fetch_balance")
graph.add_edge("fetch_balance", "finalize_response")
graph.add_edge("finalize_response", END)

app = graph.compile()

3. •Persist graph execution metadata in Redis between agent turns.

In multi-agent systems, you do not want every turn to depend on in-memory Python objects. Store intermediate outputs in Redis so another worker can continue the flow later.

import json

input_state = {
    "customer_id": "cust_001",
    "intent": "",
    "account_balance": "",
    "risk_flag": False,
    "response": "",
}

result = app.invoke(input_state)

redis_client.set(
    f"{session_id}:last_result",
    json.dumps(result),
)

redis_client.hset(
    f"{session_id}:state",
    mapping={
        "intent": result["intent"],
        "response": result["response"],
        "status": "completed",
    },
)

print(result)

4. •Add Redis-backed coordination for multiple agents.

A common pattern is one agent writes work items into Redis while specialized workers consume them. You can use Redis lists for simple queues or streams if you need stronger replay semantics.

import json

def enqueue_task(task_type: str, payload: dict):
    redis_client.lpush(
        "banking-agent-tasks",
        json.dumps({
            "task_type": task_type,
            "payload": payload,
        }),
    )

enqueue_task(
    "fraud_check",
    {
        "customer_id": "cust_001",
        "amount": 1200,
        "channel": "mobile_app",
    },
)

task = redis_client.rpop("banking-agent-tasks")
if task:
    message = json.loads(task)
    print(message["task_type"], message["payload"])

5. •Wire LangGraph node outputs into Redis events for downstream agents.

This is where the integration becomes useful in production. One node can emit a routing decision into Redis; another agent process can subscribe by polling or consuming from streams and continue the workflow.

def policy_check(state: BankingState) -> BankingState:
    # Example risk rule; replace with your compliance service.
    state["risk_flag"] = False if state["intent"] == "balance_inquiry" else True
    redis_client.hset(
        f"{session_id}:events",
        mapping={
            "last_event": f"policy_checked:{state['risk_flag']}",
        },
    )
    return state

graph2 = StateGraph(BankingState)
graph2.add_node("route_intent", route_intent)
graph2.add_node("policy_check", policy_check)
graph2.add_node("finalize_response", finalize_response)

graph2.add_edge(START, "route_intent")
graph2.add_edge("route_intent", "policy_check")
graph2.add_edge("policy_check", "finalize_response")
graph2.add_edge("finalize_response", END)

app2 = graph2.compile()

Testing the Integration

Run an end-to-end check that executes the graph and confirms Redis has the stored output.

import json

test_state = {
    "customer_id": "cust_001",
    "intent": "",
    "account_balance": "",
    "risk_flag": False,
    "response": "",
}

output = app.invoke(test_state)

redis_client.set(f"{session_id}:verification", json.dumps(output))
stored = json.loads(redis_client.get(f"{session_id}:verification"))

print("Graph output:", output)
print("Redis stored:", stored)
print("Balance:", stored["account_balance"])

Expected output:

Graph output: {'customer_id': 'cust_001', 'intent': 'balance_inquiry', 'account_balance': '$4,250.18', 'risk_flag': False, 'response': 'Your current balance is $4,250.18.'}
Redis stored: {'customer_id': 'cust_001', 'intent': 'balance_inquiry', 'account_balance': '$4,250.18', 'risk_flag': False, 'response': 'Your current balance is $4,250.18.'}
Balance: $4,250.18

Real-World Use Cases

• •

  Customer support orchestration:

  • •one agent classifies the request
  • •another fetches account data
  • •another handles compliance-sensitive responses using shared Redis state
• •

  Fraud triage pipelines:

  • •LangGraph routes suspicious transactions to a risk agent
  • •Redis stores event history and coordination flags across workers
• •

  Branch assistant workflows:

  • •multiple agents handle appointment booking, product eligibility checks, and follow-up reminders with Redis-backed session continuity

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.