How to Integrate LangGraph for wealth management with Redis for multi-agent systems

Integrating LangGraph for wealth management with Redis gives you a clean way to run stateful, multi-agent workflows without keeping everything in process memory. In practice, that means you can split portfolio analysis, risk checks, compliance review, and client communication into separate agents while Redis handles shared state, checkpoints, and coordination.

For wealth management systems, this matters because the workflow is rarely linear. A recommendation might need to be paused for a compliance review, resumed later, and audited across multiple agents without losing context.

Prerequisites

• •Python 3.10+
• •A running Redis instance
  • •Local: redis-server
  • •Or managed Redis like AWS ElastiCache or Azure Cache for Redis
• •LangGraph installed
• •Redis Python client installed
• •An OpenAI-compatible model provider or another chat model supported by LangGraph
• •Basic familiarity with:
  • •langgraph.graph.StateGraph
  • •langgraph.checkpoint.redis.RedisSaver
  • •redis.Redis

Install the packages:

pip install langgraph redis langchain-openai

Integration Steps

1) Start with a shared workflow state

In wealth management, your agents usually need the same client context: risk profile, portfolio holdings, constraints, and recommendations. Define that state once and pass it through the graph.

from typing import TypedDict, Annotated
from operator import add

class WealthState(TypedDict):
    client_id: str
    risk_profile: str
    holdings: list[str]
    notes: Annotated[list[str], add]
    recommendation: str

This keeps the workflow explicit. Each agent node updates only the fields it owns.

2) Connect LangGraph to Redis for checkpointing

Redis is where you persist graph state between steps and across agent runs. LangGraph’s Redis checkpoint saver lets you resume workflows without rebuilding context from scratch.

import redis
from langgraph.checkpoint.redis import RedisSaver

redis_client = redis.Redis(host="localhost", port=6379, db=0)

checkpointer = RedisSaver(
    redis_client=redis_client,
    ttl=60 * 60 * 24,  # keep checkpoints for 24 hours
)

If you’re running this in production, use TLS and authentication on the Redis connection. For managed Redis, pass ssl=True, username/password, or a connection URL depending on your environment.

3) Build your LangGraph agent nodes

A common pattern is to split wealth management into specialized nodes:

• •one for portfolio analysis
• •one for suitability/risk validation
• •one for final recommendation synthesis

Below is a minimal graph using StateGraph.

from langgraph.graph import StateGraph, START, END

def analyze_portfolio(state: WealthState):
    holdings = ", ".join(state["holdings"])
    return {
        "notes": [f"Analyzed holdings: {holdings}"],
        "recommendation": "Portfolio appears concentrated in tech equities."
    }

def compliance_review(state: WealthState):
    if state["risk_profile"] == "conservative":
        return {"notes": ["Compliance flag: recommendation may exceed risk tolerance"]}
    return {"notes": ["Compliance check passed"]}

def finalize(state: WealthState):
    return {
        "notes": ["Final recommendation prepared"],
        "recommendation": state["recommendation"]
    }

builder = StateGraph(WealthState)
builder.add_node("analyze_portfolio", analyze_portfolio)
builder.add_node("compliance_review", compliance_review)
builder.add_node("finalize", finalize)

builder.add_edge(START, "analyze_portfolio")
builder.add_edge("analyze_portfolio", "compliance_review")
builder.add_edge("compliance_review", "finalize")
builder.add_edge("finalize", END)

graph = builder.compile(checkpointer=checkpointer)

The important part here is compile(checkpointer=checkpointer). That’s what binds the graph execution to Redis-backed persistence.

4) Run multi-agent sessions with thread IDs

For multi-agent systems, each client or case should have its own thread ID. That lets multiple runs share the same persisted conversation or workflow state in Redis.

config = {
    "configurable": {
        "thread_id": "client-10027"
    }
}

initial_state = {
    "client_id": "client-10027",
    "risk_profile": "moderate",
    "holdings": ["AAPL", "MSFT", "VTI"],
    "notes": [],
    "recommendation": ""
}

result = graph.invoke(initial_state, config=config)

print(result["notes"])
print(result["recommendation"])

This pattern works well when one agent starts the workflow and another agent resumes it later. The thread ID becomes your durable handle across agents and services.

5) Add direct Redis coordination for shared signals

LangGraph handles workflow state. Plain Redis is still useful for cross-agent signals like locks, queues, counters, or transient flags.

For example, you can mark a case as reviewed before another agent proceeds:

import json

case_key = f"wealth_case:{initial_state['client_id']}"

redis_client.hset(case_key, mapping={
    "status": "reviewed",
    "risk_profile": initial_state["risk_profile"],
})

stored = redis_client.hgetall(case_key)
decoded = {k.decode(): v.decode() for k, v in stored.items()}

print(json.dumps(decoded, indent=2))

Use this when you need lightweight coordination outside the graph itself. Don’t overload LangGraph state with every operational concern.

Testing the Integration

Run a full invocation and confirm both LangGraph execution and Redis persistence are working.

test_config = {
    "configurable": {
        "thread_id": "test-client-001"
    }
}

test_input = {
    "client_id": "test-client-001",
    "risk_profile": "moderate",
    "holdings": ["SPY", "QQQ"],
    "notes": [],
    "recommendation": ""
}

output = graph.invoke(test_input, config=test_config)

print("Recommendation:", output["recommendation"])
print("Notes:", output["notes"])

# Verify checkpoint data exists in Redis
keys = redis_client.keys("*test-client-001*")
print("Redis keys:", [k.decode() for k in keys])

Expected output:

Recommendation: Portfolio appears concentrated in tech equities.
Notes: ['Analyzed holdings: SPY, QQQ', 'Compliance check passed', 'Final recommendation prepared']
Redis keys: ['...test-client-001...']

If you see keys in Redis and the graph returns structured output, the integration is working.

Real-World Use Cases

• •

  Client portfolio review workflows

  • •One agent analyzes asset allocation.
  • •Another checks suitability against risk policy.
  • •A third prepares advisor-facing commentary.
• •

  Human-in-the-loop approval pipelines

  • •Persist review state in Redis.
  • •Pause the LangGraph workflow until an advisor approves.
  • •Resume from the last checkpoint without recomputing prior steps.
• •

  Multi-agent advisory orchestration

  • •Use separate agents for tax considerations, ESG screening, and liquidity analysis.
  • •Coordinate them through LangGraph while Redis stores shared session state and operational flags.

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.