Hybrid Chunking and Multi-Agent LangGraph for Financial Investigations

Part 1: Chunking Strategy for Financial Data

In enterprise financial systems, naive "fixed-size" chunking (e.g., splitting every 500 tokens) is catastrophic. It severs table headers from their data, breaks legal clauses in half, and destroys the chronological flow of transaction narratives.

To solve this, we use a Dual-Track Chunking Strategy tailored to the distinct natures of Financial Documents and Transaction Narratives.

1. Financial Documents (10-Ks, Contracts, KYC Policies)

Strategy: Hierarchical / Structure-Aware Chunking (Parent-Child)

How it works: We parse documents into Markdown or HTML, preserving the DOM/Header hierarchy (e.g., H1 > H2 > H3). We chunk at the lowest logical level (paragraphs or clauses) but retain a "Parent" chunk (the entire section) in the vector database.
Why: Financial documents rely heavily on context. If a vector search retrieves a specific clause about "wire transfer limits," the LLM needs the surrounding section (the Parent) to understand the exceptions and definitions. This prevents hallucinations and ensures legal/compliance accuracy. We also use specialized table-extraction chunking to keep tabular data intact.

2. Transaction Narratives (SWIFT messages, Ledger notes, AML alerts)

Strategy: Event-Driven / Metadata-Enriched Chunking

How it works: Transaction narratives are usually short, dense, and entity-heavy (e.g., "Wire transfer to offshore entity for consulting services"). Instead of chunking the text, we chunk by Transaction Event. The "chunk" is the narrative text, but it is heavily augmented with structured metadata: sender_id, receiver_id, amount, currency, timestamp, and purpose_code.
Why: Transaction narratives require Hybrid Search (Vector similarity + strict Metadata filtering). If an investigator asks, "Show me transactions over $10,000 to the Cayman Islands," a pure vector search will fail. By chunking by event and attaching metadata, the retrieval system can execute a SQL/NoSQL filter first, and then use semantic search only on the filtered narrative text to understand the context.

Part 2: Real-Time Use Case & Architecture

The Use Case: Automated AML (Anti-Money Laundering) Investigation

Scenario: A high-value wire transfer triggers an AML alert. The system must investigate the client's KYC documents (to check their stated business purpose) and their transaction history (to check for structuring or suspicious narratives). Finally, it must draft a Suspicious Activity Report (SAR) recommendation.

The Multi-Agent Architecture (LangGraph)

We use a Supervisor Multi-Agent Architecture with shared state and persistent memory.

Supervisor Agent: Routes the investigation. Decides if more document research or transaction research is needed.
Document Researcher Agent: Uses the Hierarchical Retriever to find KYC/Policy context.
Transaction Investigator Agent: Uses the Metadata-Filtered Retriever to analyze transaction narratives.
Compliance Writer Agent: Synthesizes the findings into a formal SAR draft.

State & Memory:

State: A centralized InvestigationState tracks the alert details, retrieved context, agent thoughts, and the final report.
Memory: We use LangGraph's Checkpointer (backed by Postgres/Redis in production) for short-term thread memory, and Vector/SQL databases for long-term enterprise knowledge.

Part 3: Code Implementation

Below is the end-to-end Python implementation using langgraph, langchain, and pydantic.

Prerequisites

pip install langgraph langchain langchain-openai pydantic faiss-cpu

1. Define the State and Memory

from typing import List, Dict, Any, Literal, Annotated
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# Define the shared state for the multi-agent graph
class InvestigationState(BaseModel):
    alert_id: str
    client_id: str
    transaction_details: Dict[str, Any]
    messages: List[Dict[str, str]] = Field(default_factory=list)
    doc_context: List[str] = Field(default_factory=list)
    txn_context: List[str] = Field(default_factory=list)
    supervisor_decision: Literal["research_docs", "research_txns", "write_report", "FINISH"] = "research_docs"
    final_report: str = ""

    class Config:
        arbitrary_types_allowed = True

2. Mock Enterprise Retrievers (Reflecting our Chunking Strategy)

class FinancialDocRetriever:
    """Simulates Hierarchical/Parent-Child Retrieval for KYC/Policy Docs"""
    def search(self, query: str, client_id: str) -> List[str]:
        # In reality, this queries a Vector DB (like FAISS/Pinecone) 
        # and fetches the Parent chunk for context.
        return [
            f"[KYC Policy] Client {client_id} is registered as an Import/Export business. Allowed transaction limit: $50,000/month.",
            f"[Contract Clause] Wire transfers to non-treaty countries require secondary approval from the Compliance Officer."
        ]

class TransactionNarrativeRetriever:
    """Simulates Metadata-Filtered Retrieval for Transaction Narratives"""
    def search(self, query: str, filters: Dict[str, Any]) -> List[str]:
        # In reality, this applies SQL/NoSQL filters (amount > X, date = Y)
        # and then runs semantic search on the narrative text.
        return [
            f"[TXN-992] $45,000 to 'Offshore Consulting Ltd'. Narrative: 'Payment for Q3 advisory services'. (Note: Client usually transacts < $5,000).",
            f"[TXN-993] $4,900 to 'Offshore Consulting Ltd'. Narrative: 'Consulting fees'. (Potential structuring detected)."
        ]

doc_retriever = FinancialDocRetriever()
txn_retriever = TransactionNarrativeRetriever()

3. Define the Agent Nodes

from langchain_openai import ChatOpenAI
import os

# Initialize LLM (Ensure OPENAI_API_KEY is set in your environment)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def supervisor_node(state: InvestigationState) -> dict:
    """Decides the next step in the investigation."""
    prompt = f"""You are the AML Investigation Supervisor.
    Current State:
    - Docs retrieved: {len(state.doc_context)}
    - Txns retrieved: {len(state.txn_context)}
    
    Decide the next action. If we lack policy context, choose 'research_docs'. 
    If we lack transaction history, choose 'research_txns'. 
    If we have enough context, choose 'write_report'.
    """
    response = llm.invoke([SystemMessage(content=prompt)])
    
    # Simple parsing for the demo (in prod, use structured output / tool calling)
    decision = "write_report"
    if "research_docs" in response.content.lower() and len(state.doc_context) == 0:
        decision = "research_docs"
    elif "research_txns" in response.content.lower() and len(state.txn_context) == 0:
        decision = "research_txns"
        
    return {"supervisor_decision": decision}

def doc_researcher_node(state: InvestigationState) -> dict:
    """Retrieves hierarchical financial document chunks."""
    query = f"KYC profile and policy limits for client {state.client_id}"
    chunks = doc_retriever.search(query, state.client_id)
    
    # Update state with retrieved context
    new_docs = state.doc_context + chunks
    return {"doc_context": new_docs, "supervisor_decision": "research_txns"}

def txn_investigator_node(state: InvestigationState) -> dict:
    """Retrieves metadata-filtered transaction narratives."""
    filters = {"client_id": state.client_id, "amount": state.transaction_details.get("amount", 0)}
    query = "Suspicious narrative patterns or structuring"
    chunks = txn_retriever.search(query, filters)
    
    new_txns = state.txn_context + chunks
    return {"txn_context": new_txns, "supervisor_decision": "write_report"}

def compliance_writer_node(state: InvestigationState) -> dict:
    """Synthesizes findings into a final SAR report."""
    prompt = f"""You are an expert AML Compliance Officer. Draft a Suspicious Activity Report (SAR).
    
    Alert ID: {state.alert_id}
    Client ID: {state.client_id}
    Trigger Transaction: {state.transaction_details}
    
    KYC/Policy Context:
    {chr(10).join(state.doc_context)}
    
    Transaction History Context:
    {chr(10).join(state.txn_context)}
    
    Provide a professional, concise SAR recommendation detailing why this is suspicious based on the context.
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"final_report": response.content, "supervisor_decision": "FINISH"}

4. Build and Compile the LangGraph

def route_supervisor(state: InvestigationState) -> str:
    """Conditional edge to route based on Supervisor's decision."""
    decision = state.supervisor_decision
    if decision == "research_docs":
        return "doc_researcher"
    elif decision == "research_txns":
        return "txn_investigator"
    elif decision == "write_report":
        return "compliance_writer"
    else:
        return "end"

# Initialize the Graph
workflow = StateGraph(InvestigationState)

# Add Nodes
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("doc_researcher", doc_researcher_node)
workflow.add_node("txn_investigator", txn_investigator_node)
workflow.add_node("compliance_writer", compliance_writer_node)

# Set Entry Point
workflow.set_entry_point("supervisor")

# Add Edges
workflow.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "doc_researcher": "doc_researcher",
        "txn_investigator": "txn_investigator",
        "compliance_writer": "compliance_writer",
        "end": END
    }
)

# All worker nodes route back to the supervisor to re-evaluate
workflow.add_edge("doc_researcher", "supervisor")
workflow.add_edge("txn_investigator", "supervisor")
workflow.add_edge("compliance_writer", END)

# Compile with Memory (Checkpointer)
# In production, use PostgresSaver or RedisSaver for persistent enterprise memory
memory = MemorySaver() 
graph = workflow.compile(checkpointer=memory)

5. Execute the Investigation

if __name__ == "__main__":
    # Simulate an incoming AML Alert
    initial_state = InvestigationState(
        alert_id="AML-2026-8839",
        client_id="CORP-9921",
        transaction_details={
            "amount": 45000,
            "currency": "USD",
            "destination": "Cayman Islands",
            "narrative": "Q3 advisory services"
        }
    )
    
    # Config for memory (thread_id allows resuming this exact investigation later)
    config = {"configurable": {"thread_id": "AML-2026-8839"}}
    
    print("--- Starting AML Investigation Graph ---")
    
    # Stream the execution to see the agent routing in real-time
    for event in graph.stream(initial_state.dict(), config):
        for node_name, node_output in event.items():
            print(f"\n[Node: {node_name}]")
            if node_name == "supervisor":
                print(f"Decision: {node_output.get('supervisor_decision')}")
            elif node_name == "compliance_writer":
                print("\n=== FINAL SAR REPORT ===")
                print(node_output.get('final_report'))
                print("========================\n")

Part 4: Enterprise Considerations for Production

To move this from a prototype to a production-grade enterprise system, implement the following:

Persistent Memory (Checkpointer): Replace MemorySaver with langgraph.checkpoint.postgres.PostgresSaver. This ensures that if an investigation is paused, or if a human compliance officer needs to step in and ask the agent a follow-up question via a chat UI, the exact state and context are preserved.
Human-in-the-Loop (Interrupts): Use LangGraph's interrupt_before parameter on the compliance_writer node. This pauses the graph, sends the draft SAR to a human officer's dashboard, and waits for their approval or edits before finalizing the state.
Guardrails & PII Redaction: Financial narratives contain PII (Names, Account Numbers). Implement a pre-processing node that uses a library like presidio to mask PII before it hits the LLM context window, ensuring compliance with GDPR/CCPA.
Evaluation (LangSmith): Connect the LangGraph to LangSmith. Trace every node execution to measure retrieval latency, LLM token usage, and accuracy. This is critical for auditing AI decisions in financial compliance.