Langchain  

Graph-Augmented Reasoning for Cash Management

Cash Management is the nervous system of corporate fintech. Unlike consumer banking, it involves complex entity hierarchies, multi-bank liquidity structures, regulatory constraints (Basel III LCR, NSFR), and real-time payment rails. Traditional RAG fails here because cash positioning is structurally determined—you cannot vector-search your way to understanding a subsidiary’s trapped cash or a cross-border pooling arrangement. This article demonstrates how to combine graph traversal with LLM reasoning in a production-grade Cash Management system. We implement an enterprise Multi-Agent LangGraph architecture where graph queries provide structural ground truth, RAG supplies regulatory/policy context, and LLMs synthesize both into actionable treasury intelligence—all with persistent state and audit-grade memory.

1. How Graph Traversal + LLM/RAG Reasoning Actually Works

Before the code, let’s demystify the integration pattern. There are three distinct modes of combining these technologies, each serving a different purpose in cash management:

Mode 1: Graph-as-Ground-Truth for LLM Reasoning

The LLM never guesses about entity relationships, balances, or hierarchies. Graph traversal returns structured facts; the LLM reasons over those facts.

User Question → Agent decomposes → Cypher query → Structured result → LLM synthesizes answer

Why this matters: Hallucinating a subsidiary relationship or cash balance in treasury management is catastrophic. The graph is the single source of truth; the LLM is the reasoning layer.

Mode 2: RAG-for-Policy + Graph-for-Position

Regulatory constraints live in unstructured documents (central bank circulars, internal treasury policies). Entity positions live in the graph. The agent fuses both before answering.

Question → Parallel execution:
  ├─ Vector search → Relevant policy excerpts
  └─ Graph traversal → Current cash positions & hierarchy
→ LLM combines policy constraints + actual positions → Compliant recommendation

Why this matters: "Can we sweep €50M from our Dutch BV to the Singapore hub?" requires knowing BOTH the current balance (graph) AND the ECB transfer pricing / tax implications (RAG).

Mode 3: Graph-Informed Query Planning

The LLM uses lightweight graph schema introspection to plan which traversals to execute, rather than generating Cypher blindly.

Question → Schema-aware planning agent → Validated Cypher → Execution → Result validation → Synthesis

Why this matters: Cash management graphs have domain-specific patterns (pooling structures, notional vs. physical sweeps, currency buckets). Blind text-to-Cypher fails on these. Schema-grounded planning succeeds.

407

2. Cash Management Graph Ontology

// Corporate Structure
(:CorporateGroup)-[:CONTAINS]->(:LegalEntity)
(:LegalEntity)-[:OWNED_BY {pct: float}]->(:LegalEntity)
(:LegalEntity)-[:HAS_ACCOUNT]->(:BankAccount)
(:LegalEntity)-[:SUBJECT_TO]->(:RegulatoryJurisdiction)

// Banking & Liquidity
(:BankAccount)-[:HELD_AT]->(:Bank)
(:BankAccount)-[:DENOMINATED_IN]->(:Currency)
(:BankAccount)-[:PARTICIPATES_IN]->(:CashPool)
(:CashPool)-[:MANAGED_BY]->(:HeaderAccount)
(:CashPool)-[:POOL_TYPE]->(:PoolType)  // NOTIONAL, PHYSICAL, ZERO_BALANCE

// Transactions & Positions
(:BankAccount)-[:HAS_POSITION]->(:CashPosition)
(:CashPosition)-[:AS_OF]->(:Timestamp)
(:BankAccount)-[:SENT_PAYMENT]->(:PaymentInstruction)
(:PaymentInstruction)-[:VIA_RAIL]->(:PaymentRail)  // SEPA, FEDWIRE, SWIFT_GPI

// Regulatory & Policy
(:RegulatoryJurisdiction)-[:GOVERNED_BY]->(:Regulation)
(:LegalEntity)-[:BOUND_BY]->(:TreasuryPolicy)
(:TreasuryPolicy)-[:SUPERSEDES]->(:TreasuryPolicy)
(:Regulation)-[:CONSTRAINS]->(:CashMovement)

Key design decisions:

  • CashPosition as separate node (not account property): Enables historical position snapshots without temporal bloat on accounts

  • PoolType as node (not string): Allows policy constraints to reference pool types as first-class entities

  • PaymentRail as node: Rail-specific limits, cut-off times, and fees become traversable attributes

  • Ownership percentage on edge: Critical for consolidation logic and minority interest calculations

3. Real-Time Use Case: Cross-Border Liquidity Optimization Under Regulatory Constraints

Scenario: A multinational treasurer asks at 6:00 AM EST:

"What’s our global cash position by currency? Can we fund the APAC deficit from European surplus without violating ECB liquidity coverage rules or triggering Chinese SAFE repatriation limits? Recommend optimal sweep structure."

This requires:

  1. Global position aggregation across 47 entities (graph traversal)

  2. ECB LCR constraint retrieval (RAG)

  3. Chinese SAFE limit lookup (graph + RAG)

  4. Pool structure feasibility analysis (graph traversal)

  5. Synthesized sweep recommendation with compliance validation (LLM reasoning)

All within a single conversational session with memory of prior queries.

4. End-to-End Implementation

Step 1: Typed State with Treasury-Specific Memory

from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class CashManagementState(TypedDict):
    """Persistent state for treasury investigation sessions."""
    messages: Annotated[List[Any], add_messages]
    
    # Session context
    session_id: str
    treasurer_id: str
    corporate_group_id: str
    
    # Graph-derived positions
    global_positions_by_currency: Dict[str, float]
    entity_hierarchy: List[Dict[str, Any]]
    cash_pool_structures: List[Dict[str, Any]]
    payment_rail_constraints: List[Dict[str, Any]]
    trapped_cash_entities: List[str]
    
    # RAG-derived regulatory context
    applicable_regulations: List[Dict[str, str]]
    treasury_policy_excerpts: List[Dict[str, str]]
    regulatory_constraints: List[Dict[str, Any]]
    
    # Synthesized output
    liquidity_gap_analysis: Optional[Dict[str, float]]
    sweep_recommendations: List[Dict[str, Any]]
    compliance_warnings: List[str]
    
    # Workflow control
    phase: str  # "position_aggregation", "constraint_retrieval", 
                # "optimization", "compliance_check", "complete"
    requires_human_approval: bool
    error_log: List[str]

Step 2: Graph Tools for Cash Position Intelligence

from langchain_neo4j import Neo4jGraph
from langchain_core.tools import tool
from decimal import Decimal

graph = Neo4jGraph(
    url="bolt://treasury-graph-prod.databases.neo4j.io:7687",
    username="neo4j",
    password="${NEO4J_PASSWORD}",
    database="cash-management",
    read_only=True
)

@tool
def get_global_cash_position(group_id: str, as_of_date: str = None) -> dict:
    """
    Aggregates cash positions across all entities in corporate group.
    Returns per-currency totals and identifies trapped cash jurisdictions.
    """
    date_filter = f"AND cp.as_of >= datetime('{as_of_date}')" if as_of_date else ""
    query = f"""
    MATCH (cg:CorporateGroup {{group_id: $gid}})-[:CONTAINS]->(le:LegalEntity)
          -[:HAS_ACCOUNT]->(ba:BankAccount)-[:DENOMINATED_IN]->(cur:Currency)
    MATCH (ba)-[:HAS_POSITION]->(cp:CashPosition)
    WHERE cp.is_current = true {date_filter}
    OPTIONAL MATCH (le)-[:SUBJECT_TO]->(jur:RegulatoryJurisdiction)
    OPTIONAL MATCH (ba)-[:PARTICIPATES_IN]->(pool:CashPool)
    
    WITH cur.code AS currency,
         sum(cp.available_balance) AS total_available,
         sum(cp.ledger_balance) AS total_ledger,
         collect(DISTINCT {{
           entity: le.name,
           jurisdiction: jur.code,
           balance: cp.available_balance,
           pool: pool.pool_id,
           trapped: jur.has_capital_controls
         }}) AS entity_breakdown
    
    RETURN currency, total_available, total_ledger, entity_breakdown,
           [eb IN entity_breakdown WHERE eb.trapped = true | eb.entity] AS trapped_entities
    ORDER BY total_available DESC
    """
    results = graph.query(query, {"gid": group_id})
    
    positions = {}
    all_trapped = []
    for r in results:
        positions[r["currency"]] = {
            "available": float(r["total_available"]),
            "ledger": float(r["total_ledger"]),
            "entities": r["entity_breakdown"]
        }
        all_trapped.extend(r["trapped_entities"])
    
    return {
        "positions_by_currency": positions,
        "trapped_cash_entities": list(set(all_trapped)),
        "total_currencies": len(positions)
    }

@tool
def analyze_pool_feasibility(source_entity: str, target_entity: str, amount: float, currency: str) -> dict:
    """
    Evaluates whether a sweep between two entities is structurally feasible.
    Checks pool membership, rail availability, and intermediary requirements.
    """
    query = """
    // Check direct pool connectivity
    MATCH (src:LegalEntity {name: $source})-[:HAS_ACCOUNT]->(sa:BankAccount)
          -[:DENOMINATED_IN]->(cur:Currency {code: $currency})
    MATCH (tgt:LegalEntity {name: $target})-[:HAS_ACCOUNT]->(ta:BankAccount)
          -[:DENOMINATED_IN]->(cur)
    OPTIONAL MATCH (sa)-[:PARTICIPATES_IN]->(sp:CashPool)<-[:PARTICIPATES_IN]-(ta)
    OPTIONAL MATCH (sa)-[:SENT_PAYMENT]->(pi:PaymentInstruction)-[:VIA_RAIL]->(rail:PaymentRail)
    WHERE pi.status = 'ACTIVE' AND rail.supports_currency = $currency
    
    // Check regulatory path
    OPTIONAL MATCH (src)-[:SUBJECT_TO]->(sj:RegulatoryJurisdiction)
    OPTIONAL MATCH (tgt)-[:SUBJECT_TO]->(tj:RegulatoryJurisdiction)
    OPTIONAL MATCH (sj)-[:GOVERNED_BY]->(reg:Regulation)-[:CONSTRAINS]->(cm:CashMovement)
    
    RETURN 
        sp IS NOT NULL AS same_pool,
        sp.pool_type AS pool_type,
        collect(DISTINCT rail.name) AS available_rails,
        collect(DISTINCT {{
          regulation: reg.name,
          constraint_type: cm.type,
          limit_amount: cm.limit_amount,
          limit_currency: cm.limit_currency
        }}) AS regulatory_constraints,
        sj.code AS source_jurisdiction,
        tj.code AS target_jurisdiction,
        sj.has_capital_controls AS source_controls,
        tj.has_capital_controls AS target_controls
    """
    result = graph.query(query, {
        "source": source_entity, 
        "target": target_entity, 
        "currency": currency
    })
    
    if not result:
        return {"feasible": False, "reason": "No matching accounts found"}
    
    r = result[0]
    constraints_violated = [
        c for c in r["regulatory_constraints"]
        if c["limit_amount"] and float(c["limit_amount"]) < amount
    ]
    
    return {
        "feasible": len(constraints_violated) == 0 and len(r["available_rails"]) > 0,
        "same_pool": r["same_pool"],
        "pool_type": r["pool_type"],
        "available_rails": r["available_rails"],
        "regulatory_constraints": r["regulatory_constraints"],
        "constraints_violated": constraints_violated,
        "cross_border": r["source_jurisdiction"] != r["target_jurisdiction"],
        "capital_controls": r["source_controls"] or r["target_controls"]
    }

@tool
def retrieve_treasury_policy_and_regulation(context_keywords: List[str]) -> dict:
    """
    Hybrid RAG: Retrieves both internal treasury policies AND external regulations.
    Uses separate indices with unified relevance ranking.
    """
    policy_query = """
    CALL db.index.fulltext.queryNodes('treasuryPolicySearch', $query) YIELD node, score
    WHERE node.effective_date <= datetime() AND score > 0.6
    RETURN 'POLICY' AS source, node.title, node.policy_id AS ref_id,
           substring(node.content, 0, 800) AS excerpt, score
    ORDER BY score DESC LIMIT 3
    """
    
    reg_query = """
    CALL db.index.fulltext.queryNodes('regulationSearch', $query) YIELD node, score
    WHERE node.effective_date <= datetime() AND score > 0.6
    RETURN 'REGULATION' AS source, node.title, node.reg_code AS ref_id,
           substring(node.text, 0, 800) AS excerpt, score
    ORDER BY score DESC LIMIT 3
    """
    
    combined = []
    combined.extend(graph.query(policy_query, {"query": " ".join(context_keywords)}))
    combined.extend(graph.query(reg_query, {"query": " ".join(context_keywords)}))
    combined.sort(key=lambda x: x["score"], reverse=True)
    
    return {
        "policies": [c for c in combined if c["source"] == "POLICY"],
        "regulations": [c for c in combined if c["source"] == "REGULATION"],
        "total_relevant": len(combined)
    }

Step 3: Multi-Agent LangGraph Orchestration

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Specialized agent bindings
position_llm = llm.bind_tools([get_global_cash_position, analyze_pool_feasibility])
policy_llm = llm.bind_tools([retrieve_treasury_policy_and_regulation])

def position_aggregation_agent(state: CashManagementState) -> CashManagementState:
    """Retrieves and structures global cash positions."""
    prompt = f"""You are a Treasury Position Analyst. For corporate group {state['corporate_group_id']},
    retrieve the global cash position by currency. Identify any trapped cash entities.
    If the user's question references specific entities or sweep amounts, also analyze pool feasibility."""
    response = position_llm.invoke(
        [{"role": "user", "content": prompt}] + state["messages"]
    )
    return {"messages": [response], "phase": "position_aggregation"}

def constraint_retrieval_agent(state: CashManagementState) -> CashManagementState:
    """Fetches regulatory and policy constraints relevant to current positions."""
    # Dynamically build context from graph findings
    keywords = ["liquidity coverage", "cross-border sweep", "capital controls"]
    currencies = list(state.get("global_positions_by_currency", {}).keys())
    keywords.extend([f"{c} regulation" for c in currencies[:3]])
    
    if state.get("trapped_cash_entities"):
        keywords.append("trapped cash repatriation")
    
    prompt = f"""Retrieve treasury policies and regulations relevant to: {keywords}.
    Focus on constraints that would affect inter-entity sweeps and liquidity optimization."""
    response = policy_llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response], "phase": "constraint_retrieval"}

def optimization_agent(state: CashManagementState) -> CashManagementState:
    """Synthesizes positions + constraints into sweep recommendations."""
    prompt = f"""
    You are a Senior Treasury Strategist. Produce a liquidity optimization recommendation:
    
    GLOBAL POSITIONS:
    {json.dumps(state.get('global_positions_by_currency', {}), indent=2)}
    
    TRAPPED CASH ENTITIES: {state.get('trapped_cash_entities', [])}
    
    REGULATORY CONSTRAINTS:
    {chr(10).join(f"- [{r['ref_id']}] {r['title']}: {r['excerpt'][:300]}" 
                  for r in state.get('applicable_regulations', []))}
    
    TREASURY POLICIES:
    {chr(10).join(f"- [{p['ref_id']}] {p['title']}: {p['excerpt'][:300]}" 
                  for p in state.get('treasury_policy_excerpts', []))}
    
    Produce:
    1. Liquidity Gap Analysis (surplus/deficit by currency)
    2. Recommended Sweep Structure (source → target, amount, rail, timing)
    3. Compliance Validation (which constraints satisfied/violated)
    4. Trapped Cash Mitigation Strategy
    5. Risk Warnings & Caveats
    6. Estimated Cost Savings vs. Current Position
    
    Format as structured JSON for downstream consumption.
    """
    response = llm.invoke(prompt)
    
    # Parse structured output
    try:
        import json
        parsed = json.loads(response.content)
        return {
            "messages": [response],
            "liquidity_gap_analysis": parsed.get("gap_analysis"),
            "sweep_recommendations": parsed.get("sweeps", []),
            "compliance_warnings": parsed.get("warnings", []),
            "phase": "complete",
            "requires_human_approval": len(parsed.get("warnings", [])) > 0
        }
    except json.JSONDecodeError:
        return {
            "messages": [response],
            "phase": "complete",
            "error_log": ["Failed to parse optimization output as JSON"],
            "requires_human_approval": True
        }

# Build workflow
workflow = StateGraph(CashManagementState)

workflow.add_node("position_aggregation", position_aggregation_agent)
workflow.add_node("position_tools", ToolNode([
    get_global_cash_position, 
    analyze_pool_feasibility
]))
workflow.add_node("constraint_retrieval", constraint_retrieval_agent)
workflow.add_node("constraint_tools", ToolNode([retrieve_treasury_policy_and_regulation]))
workflow.add_node("optimization", optimization_agent)

workflow.set_entry_point("position_aggregation")
workflow.add_edge("position_aggregation", "position_tools")
workflow.add_edge("position_tools", "constraint_retrieval")
workflow.add_edge("constraint_retrieval", "constraint_tools")
workflow.add_edge("constraint_tools", "optimization")
workflow.add_conditional_edges(
    "optimization",
    lambda s: END if not s.get("requires_human_approval") else "treasurer_review"
)

# Persistent checkpointing for multi-session treasury workflows
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://${PG_CONN}")
app = workflow.compile(checkpointer=checkpointer)

Step 4: Real-Time Streaming with Memory Resume

import json
from datetime import datetime

async def optimize_liquidity(group_id: str, question: str, treasurer_id: str):
    session_id = f"CASH-{datetime.utcnow().strftime('%Y%m%d%H%M')}-{group_id}"
    config = {"configurable": {"thread_id": session_id}}
    
    initial_state: CashManagementState = {
        "messages": [{"role": "user", "content": question}],
        "session_id": session_id,
        "treasurer_id": treasurer_id,
        "corporate_group_id": group_id,
        "phase": "position_aggregation",
        "requires_human_approval": False,
        "error_log": []
    }
    
    async for event in app.astream_events(initial_state, config=config, version="v2"):
        kind = event["event"]
        
        if kind == "on_tool_end":
            yield {
                "type": "position_signal",
                "tool": event["name"],
                "data": event["data"]["output"],
                "ts": datetime.utcnow().isoformat()
            }
        elif kind == "on_chat_model_stream":
            yield {
                "type": "recommendation_token",
                "content": event["data"]["chunk"].content
            }
        elif kind == "on_chain_end" and event["name"] == "optimization":
            output = event["data"]["output"]
            yield {
                "type": "optimization_complete",
                "session_id": session_id,
                "requires_approval": output.get("requires_human_approval"),
                "warnings_count": len(output.get("compliance_warnings", []))
            }

# Resume after treasurer approval
async def approve_sweep(session_id: str, approved: bool, modifications: dict = None):
    config = {"configurable": {"thread_id": session_id}}
    update = {
        "messages": [{
            "role": "user", 
            "content": f"Treasurer decision: {'APPROVED' if approved else 'REJECTED'}."
                       + (f" Modifications: {json.dumps(modifications)}" if modifications else "")
        }],
        "requires_human_approval": False
    }
    await app.aupdate_state(config, update)
    
    async for event in app.astream(None, config=config):
        yield event

5. Performance Engineering for Treasury Scale

ChallengeSolutionImpact
10K+ legal entities, 50K+ accountsMaterialized position views refreshed via CDCPosition queries < 50ms
Regulatory document corpus (100K+ pages)Chunked embeddings with jurisdiction metadata filteringRAG precision +35%
Pool structure complexityPre-computed reachability index per poolFeasibility checks < 20ms
Multi-session state persistenceCheckpoint compression + TTL-based archivalStorage < 5KB/session
Concurrent treasurer sessionsRead replica routing + connection pooling200+ concurrent users

6. Why This Architecture Beats Alternatives for Cash Management

ApproachFailure Mode in Cash MgmtGraph+LangGraph Solution
Pure Vector RAGHallucinates entity relationships, misses pool structuresGraph provides verified topology; RAG limited to policy text
Text-to-SQLCannot express hierarchical ownership, pool semanticsNative graph traversal expresses treasury ontology directly
Rule Engine OnlyBrittle to new regulations, no natural language interfaceAgents adapt to new policies via RAG; graph absorbs new structures
Single LLM CallNo grounding, no audit trail, no stateful investigationMulti-agent pipeline with checkpointed state and tool-use traces
Dashboard-OnlyNo reasoning over positions, no regulatory cross-referenceConversational interface backed by grounded structural + policy knowledge

7. Compliance & Governance

  • Basel III LCR/NSFR: Graph-traversed liquidity positions feed directly into regulatory ratio calculations. Agent outputs cite specific regulation nodes.

  • Transfer Pricing Documentation: Sweep recommendations include arm’s-length justification sourced from treasury policy RAG.

  • SOX Controls: Every recommendation passes through human approval gate when compliance warnings exist. Full LangGraph trace serves as control evidence.

  • Data Residency: Jurisdiction-tagged nodes enable geo-fenced queries. EU entity data never traverses non-EU graph replicas.

  • Model Risk: Structured state fields (not free-text LLM output) drive downstream systems. LLM is advisory; graph is authoritative.

Conclusion

Cash Management sits at the intersection of structural complexity (entity hierarchies, pool topologies, payment rails) and regulatory density (multi-jurisdictional constraints, evolving policies). Neither pure graph nor pure RAG suffices alone. The integration pattern demonstrated here—graph for ground truth, RAG for policy context, LLM for synthesis, LangGraph for orchestrated stateful reasoning—is the enterprise-grade solution. It transforms cash management from dashboard-driven position monitoring to conversational, compliant, structurally-grounded liquidity optimization. This implementation is for architectural reference. Production cash management systems require validation against your institution’s treasury policy framework, Basel III model risk review, SOX control testing, and coordination with your ALM and regulatory reporting teams. Never allow LLM-generated sweep recommendations to execute automatically without human approval gates and independent compliance validation. All monetary values should use Decimal types in production; floats shown here are for readability.

Membership not found