AI Agents  

Debugging Multi-Agent Memory in Fixed Deposit Orchestration

In enterprise AI, the most dangerous bugs aren’t the ones that crash the system. They are the ones where the agent appears to work correctly 99% of the time, but silently corrupts state in edge cases, leading to regulatory mis-selling or financial loss.

While building our Fixed Deposit Advisory Agent, we encountered a subtle "State Bleed via Mutable Reference" bug. It passed all unit tests, succeeded in happy-path e2e tests, and only manifested when customers asked complex, multi-turn questions involving rate comparisons across different tenures. This article details the bug, the forensic debugging process, and the hardened architecture we deployed to production.

Part 1: The Subtle Bug – Mutable State Bleed in Async Nodes

The Symptom

A customer asks: "Compare the 1-year vs 3-year FD rates for $50,000, and tell me which gives better post-tax returns given my 30% tax bracket."

The agent would correctly retrieve both rates. However, in ~15% of requests, the post-tax calculation for the 1-year FD was computed using the 3-year interest rate, or vice versa. The final response looked plausible but was mathematically wrong.

Why Unit Tests Missed It

Our unit tests mocked the RAG retriever and calculator tool. Each test ran in isolation with fresh state. The bug only appeared under concurrent async execution within a single graph invocation when the supervisor dispatched parallel research tasks.

Root Cause Analysis

The issue was a combination of three factors:

  1. Shared Mutable Default in Pydantic State: Our BankState used a mutable list as a default factory incorrectly.

  2. Async Node Concurrency: LangGraph’s Send() API for parallel fan-out reused state references instead of deep-copying when nodes modified nested objects.

  3. RAG Context Pollution: Both parallel retrieval nodes appended to the same rag_sources list. Due to Python’s reference semantics in async contexts, Node A’s append could be visible to Node B mid-execution, causing the calculator to receive mixed context.

# THE BUGGY CODE
class BankState(BaseModel):
    messages: Annotated[List[BaseMessage], add_messages]
    # BUG: Mutable default shared across concurrent executions
    rag_context: List[dict] = []  
    fd_comparison: dict = {}

When two Send() tasks ran concurrently:

  • Task A retrieves 1-year rate → appends to rag_context

  • Task B retrieves 3-year rate → appends to same rag_context reference

  • Calculator receives merged context → uses wrong rate for tax calc

The Fix: Immutable State Patterns + Explicit Reducers

We applied three corrections:

  1. Never use mutable defaults. Use Field(default_factory=...) or Annotated with custom reducers.

  2. Custom Reducer for RAG Context: Ensure parallel appends are merged safely.

  3. Scoped Sub-State for Parallel Tasks: Each fan-out task gets its own isolated namespace.

#  THE FIXED CODE
from typing import Annotated
from langgraph.graph.message import add_messages
import operator

def merge_rag_context(existing: List[dict], new: List[dict]) -> List[dict]:
    """Deduplicate and safely merge RAG sources from parallel nodes."""
    seen = {doc["doc_id"] for doc in existing}
    merged = list(existing)
    for doc in new:
        if doc["doc_id"] not in seen:
            merged.append(doc)
            seen.add(doc["doc_id"])
    return merged

class BankState(BaseModel):
    messages: Annotated[List[BaseMessage], add_messages]
    # FIX: Custom reducer ensures safe concurrent merges
    rag_context: Annotated[List[dict], merge_rag_context] = Field(default_factory=list)
    # FIX: Structured sub-state prevents cross-contamination
    fd_comparison: Annotated[dict, operator.or_] = Field(default_factory=dict)

Part 2: Real-Time Use Case – Intelligent FD Advisory & Renewal

The Scenario

Customer Raj messages the banking app:

"My $100K FD matures next month. I'm in the 30% tax bracket. Should I renew for 2 years at current rates or split into two 1-year FDs? Also, are there any senior citizen rate benefits I qualify for now that I've turned 60?"

Why This Demands Hardened Orchestration

  1. Multi-Dimensional Comparison: Requires parallel retrieval of multiple rate cards + tax rules.

  2. Personalized Eligibility: Age-based rate uplift requires memory lookup + policy RAG.

  3. Regulatory Precision: Post-tax calculations must cite exact policy clauses.

  4. Actionable Output: Must generate renewal instructions, not just advice.

Part 3: Enterprise Implementation – Fixed Deposits Module

Architecture Diagram

400

Step 1: Typed State with Safe Reducers

from pydantic import BaseModel, Field
from typing import Annotated, List, Literal, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
import operator

class FDRateInfo(BaseModel):
    tenure_months: int
    base_rate: float
    senior_uplift: float = 0.0
    effective_rate: float
    source_doc_id: str
    effective_date: str

class TaxCalculation(BaseModel):
    gross_interest: float
    tax_amount: float
    net_interest: float
    citation: str

class FDComparisonState(BaseModel):
    messages: Annotated[List[BaseMessage], add_messages]
    customer_id: str
    age: int
    tax_bracket: float
    maturity_amount: float
    
    # SAFE: Custom reducer for parallel RAG merges
    retrieved_rates: Annotated[List[FDRateInfo], merge_fd_rates] = Field(default_factory=list)
    
    # SAFE: Structured comparison result
    comparison_result: Optional[dict] = None
    eligibility_flags: dict = Field(default_factory=dict)
    audit_trail: Annotated[List[str], operator.add] = Field(default_factory=list)

Step 2: Parallel Rate Retrieval with Send()

from langgraph.types import Send

async def fd_supervisor(state: FDComparisonState):
    """Fan out parallel rate retrieval for each tenure option."""
    tenures = [12, 24]  # Derived from user query parsing
    
    sends = [
        Send("rate_retriever", {"tenure_months": t, "customer_age": state.age})
        for t in tenures
    ]
    # Also check senior citizen eligibility in parallel
    sends.append(Send("eligibility_checker", {"customer_id": state.customer_id, "age": state.age}))
    
    return sends

async def rate_retriever(state: dict):
    """Isolated retrieval per tenure - NO shared mutable state."""
    tenure = state["tenure_months"]
    age = state["customer_age"]
    
    vectorstore = get_fd_vectorstore()
    docs = await vectorstore.asimilarity_search(
        f"FD rate {tenure} months", 
        k=3,
        filter={"status": "active", "tenure_months": tenure}
    )
    
    # Parse structured rate from RAG output
    rate_info = parse_fd_rate(docs[0], tenure, age)
    
    return {
        "retrieved_rates": [rate_info],  # Reducer handles safe merge
        "audit_trail": [f"Retrieved {tenure}m rate: {rate_info.effective_rate}% from {rate_info.source_doc_id}"]
    }

Step 3: Deterministic Tax Calculator Tool

from langchain_core.tools import tool
from decimal import Decimal, ROUND_HALF_UP

@tool
def calculate_post_tax_fd_returns(
    principal: float, 
    rate: float, 
    tenure_months: int, 
    tax_bracket: float
) -> TaxCalculation:
    """
    Calculates EXACT post-tax FD returns using Decimal arithmetic.
    NEVER use float for banking calculations.
    """
    p = Decimal(str(principal))
    r = Decimal(str(rate)) / Decimal("100")
    t = Decimal(str(tenure_months)) / Decimal("12")
    tax = Decimal(str(tax_bracket))
    
    gross = (p * r * t).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    tax_amt = (gross * tax).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
    net = gross - tax_amt
    
    return TaxCalculation(
        gross_interest=float(gross),
        tax_amount=float(tax_amt),
        net_interest=float(net),
        citation="Income Tax Act Section 194A; Bank FD Policy v4.2 §3.1"
    )

Step 4: Citation Validator Guardrail

This node runs after synthesis but before user delivery. It catches hallucinated citations.

async def citation_validator(state: FDComparisonState):
    """Verify every cited doc_id exists in retrieved_rates."""
    last_msg = state.messages[-1].content
    valid_doc_ids = {r.source_doc_id for r in state.retrieved_rates}
    
    # Extract citations from response
    cited_ids = extract_citations(last_msg)
    
    invalid = cited_ids - valid_doc_ids
    if invalid:
        # SELF-CORRECTION: Strip invalid citations and regenerate
        return {
            "messages": [SystemMessage(
                content=f"CITATION ERROR: {invalid} not in retrieved context. "
                        f"Regenerate response using ONLY: {valid_doc_ids}"
            )],
            "audit_trail": [f"CITATION_VIOLATION: Removed {invalid}"]
        }
    
    return {"audit_trail": ["CITATION_VALIDATED"]}

Step 5: Compile with Persistence & Interrupts

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

checkpointer = AsyncPostgresSaver.from_conn_string(FD_DB_URI)

workflow = StateGraph(FDComparisonState)

workflow.add_node("supervisor", fd_supervisor)
workflow.add_node("rate_retriever", rate_retriever)
workflow.add_node("eligibility_checker", eligibility_checker)
workflow.add_node("calculator", calculator_node)
workflow.add_node("synthesizer", comparison_synthesizer)
workflow.add_node("citation_validator", citation_validator)
workflow.add_node("human_review", human_review_node)  # For large renewals

# Fan-out edges handled by supervisor returning Send()
workflow.add_edge("rate_retriever", "calculator")
workflow.add_edge("eligibility_checker", "calculator")
workflow.add_edge("calculator", "synthesizer")
workflow.add_edge("synthesizer", "citation_validator")

# Conditional: self-correct or proceed
workflow.add_conditional_edges("citation_validator", 
    lambda s: "synthesizer" if s.messages[-1].type == "system" else "human_review"
)

# Human approval for renewals > $50K
workflow.add_interrupt_before(["human_review"])
workflow.add_edge("human_review", END)

app = workflow.compile(checkpointer=checkpointer)

Part 4: Testing Strategy That Caught the Bug

After fixing the state bleed, we added specific tests to prevent regression:

Concurrent State Isolation Test

@pytest.mark.asyncio
async def test_parallel_rate_retrieval_isolation():
    """Verify concurrent Send() tasks don't pollute each other's RAG context."""
    config = {"configurable": {"thread_id": "test-isolation"}}
    
    # Run 20 concurrent invocations to stress-test reducer
    results = await asyncio.gather(*[
        app.ainvoke({
            "messages": [("user", "Compare 1Y vs 3Y FD for $100K")],
            "customer_id": f"CUST-{i}",
            "age": 60,
            "tax_bracket": 0.3,
            "maturity_amount": 100000
        }, config={**config, "configurable": {"thread_id": f"test-{i}"}})
        for i in range(20)
    ])
    
    for result in results:
        rates = result["retrieved_rates"]
        assert len(rates) == 2, f"Expected 2 rates, got {len(rates)}"
        tenures = {r.tenure_months for r in rates}
        assert tenures == {12, 36}, f"Tenure contamination detected: {tenures}"
        
        # Verify no duplicate doc_ids (reducer working)
        doc_ids = [r.source_doc_id for r in rates]
        assert len(doc_ids) == len(set(doc_ids)), "Duplicate RAG sources found"

Decimal Precision Assertion

def test_tax_calculation_decimal_precision():
    """Float arithmetic causes penny errors. Must use Decimal."""
    result = calculate_post_tax_fd_returns.invoke({
        "principal": 100000, "rate": 7.25, 
        "tenure_months": 24, "tax_bracket": 0.3
    })
    # Exact expected value pre-computed with Decimal
    assert result.net_interest == 10150.00  
    assert isinstance(result.gross_interest, float)  # Serialized as float
    # But internal computation was Decimal (verified via audit)

Key Takeaways for Enterprise Agent Builders

LessonApplication
Mutable defaults are landminesAlways use Field(default_factory=...) + custom reducers for lists/dicts in state
Parallel ≠ IndependentSend() shares parent state; design nodes to write to reducer-managed fields only
Floats are forbidden in financeUse Decimal in tools; serialize to float only at API boundary
Citations need validationPost-hoc guardrail catching hallucinated refs is cheaper than perfect retrieval
Test concurrency explicitlySingle-threaded tests won’t catch state bleed; use asyncio.gather stress tests
Audit trail as first-class stateEvery node appends to audit_trail; enables post-incident forensics without log diving

The state bleed bug taught us a fundamental truth: LangGraph’s flexibility is also its greatest risk surface. In banking, you must trade some of that flexibility for deterministic safety through typed state, custom reducers, and adversarial testing. The extra engineering cost pays for itself the first time it prevents a mis-sold FD product from reaching a customer.

Membership not found