AI Agents  

The Retrieval Strategy Verdict for Financial Anomaly Detection

When it comes to financial anomaly detection (fraud, AML, market manipulation), the short answer is: No single retrieval strategy works in isolation.

Here is the hierarchy of what works, and why the industry has moved beyond simple vector search:

  1. Dense Retrieval (Vector Search): Fails for anomalies. Dense embeddings are great for semantic similarity, but they are "blind" to exact numbers, tickers, and structural relationships. An anomaly is often defined by a specific numerical deviation (e.g., "Volume is 400% above average"), which dense vectors cannot reliably retrieve.

  2. Hybrid Retrieval + Reranking: The best for unstructured text. Combining BM25 (sparse) for exact keyword/ticker matching with Dense (semantic) for context, followed by a Cross-Encoder Reranker, is the gold standard for querying unstructured narratives (emails, SWIFT messages, news).

  3. The Actual Winner: Agentic Multi-Modal Retrieval (The "Tri-Fecta").
    Financial anomalies are rarely just text-based; they are structural and behavioral. The most effective enterprise strategy combines Hybrid Text Retrieval, Graph/Relational Traversal, and Time-Series/Statistical Retrieval, orchestrated by a multi-Agent system.

The Winning Strategy: Context-Aware Agentic Retrieval. The LLM acts as a router, using Hybrid Search for narratives, Graph queries for entity networks, and Statistical lookups for behavioral baselines, synthesizing them into a single anomaly score.

Part 2: Real-Time Use Case

The Use Case: "Project Argus" - Real-Time Insider Trading & Market Manipulation Detection

Scenario: A market surveillance system flags an unusual 800% spike in out-of-the-money call options for ticker $ACME exactly 48 hours before a major earnings announcement.

The Investigation: The system must determine if this is a legitimate market move or insider trading. It cannot just look at the text; it must investigate the actors and their behavior.

  1. Narrative Investigator: Uses Hybrid Search + Reranking to scan unstructured communications (emails/chats) for mentions of $ACME or "earnings".

  2. Network Analyst: Queries the Knowledge Graph to see if the option buyers share hidden links (e.g., same IP address, shared phone numbers, or familial ties to ACME executives).

  3. Behavioral Analyst: Queries the Time-Series database to check if these accounts have a history of trading ahead of corporate events (statistical baseline).

Part 3: Enterprise Multi-Agent Architecture (LangGraph)

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

  • State: SurveillanceState tracks the alert, retrieved text, graph connections, statistical baselines, and the final report.

  • Memory: LangGraph Checkpointer ensures the investigation state is saved. If a human analyst pauses the investigation to gather more data, the graph resumes exactly where it left off.

  • Agents:

    • Supervisor: Routes the investigation flow.

    • Narrative_Investigator: Hybrid Search + Reranking.

    • Network_Analyst: Graph Traversal.

    • Behavioral_Analyst: Time-Series/Statistical Retrieval.

    • Compliance_Synthesizer: Drafts the Market Surveillance Report.

de

Part 4: Code Implementation

Prerequisites

pip install langgraph langchain langchain-openai pydantic

1. Define the State and Memory

from typing import Dict, List, Any, Literal, Optional
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
import os

# Define the shared state for the surveillance graph
class SurveillanceState(TypedDict):
    alert_id: str
    ticker: str
    involved_accounts: List[str]
    messages: List[Dict[str, str]]
    narrative_context: List[str]  # From Hybrid Search
    graph_context: List[str]      # From Graph Traversal
    baseline_context: str         # From Time-Series
    supervisor_decision: Literal["check_narratives", "check_network", "check_behavior", "synthesize", "FINISH"]
    final_report: str

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

2. Mock Enterprise Retrievers (The "Tri-Fecta" Strategy)

class HybridNarrativeRetriever:
    """Simulates BM25 + Dense Vector + Cross-Encoder Reranking"""
    def search(self, ticker: str, accounts: List[str]) -> List[str]:
        # In prod: BM25 matches exact ticker, Dense matches semantic context, 
        # Reranker re-scores top 50 down to top 5.
        return [
            f"[Email - Account {accounts[0]}] 'Hey, did you hear about the $ACME earnings leak? Buy calls now.'",
            f"[Chat - Account {accounts[1]}] 'I have a feeling $ACME is going to beat estimates tomorrow. Loading up.'"
        ]

class GraphNetworkRetriever:
    """Simulates Neo4j Knowledge Graph Traversal"""
    def search(self, accounts: List[str]) -> List[str]:
        # In prod: Cypher query to find shared nodes (IPs, phones, addresses)
        return [
            f"[Graph] Account {accounts[0]} and Account {accounts[1]} share the same registered IP address (192.168.1.45).",
            f"[Graph] Account {accounts[0]} is linked to 'John Doe', who is a registered vendor for ACME Corp."
        ]

class TimeSeriesBaselineRetriever:
    """Simulates Statistical/Time-Series Retrieval"""
    def search(self, accounts: List[str], ticker: str) -> str:
        # In prod: Queries InfluxDB/TimescaleDB for historical trading baselines
        return (
            f"[Baseline] Accounts {accounts} typically trade < $5,000/month. "
            f"Current $ACME options volume is $450,000 (9,000% deviation from 30-day rolling average). "
            f"Win rate on pre-earnings trades: 100% (3/3)."
        )

hybrid_retriever = HybridNarrativeRetriever()
graph_retriever = GraphNetworkRetriever()
ts_retriever = TimeSeriesBaselineRetriever()

3. Define the Agent Nodes

def supervisor_node(state: SurveillanceState) -> dict:
    """Decides the next investigative step."""
    prompt = f"""You are the Market Surveillance Supervisor.
    Current Investigation State:
    - Narratives retrieved: {len(state.get('narrative_context', []))}
    - Graph links found: {len(state.get('graph_context', []))}
    - Baseline retrieved: {bool(state.get('baseline_context'))}
    
    Route the investigation:
    1. If no narratives, choose 'check_narratives'.
    2. If no graph links, choose 'check_network'.
    3. If no baseline, choose 'check_behavior'.
    4. If all are present, choose 'synthesize'.
    """
    response = llm.invoke([SystemMessage(content=prompt)])
    
    # Simple parsing for demo (Use structured output in prod)
    text = response.content.lower()
    if "check_narratives" in text and not state.get('narrative_context'):
        decision = "check_narratives"
    elif "check_network" in text and not state.get('graph_context'):
        decision = "check_network"
    elif "check_behavior" in text and not state.get('baseline_context'):
        decision = "check_behavior"
    elif "synthesize" in text:
        decision = "synthesize"
    else:
        decision = "synthesize" # Fallback
        
    return {"supervisor_decision": decision}

def narrative_investigator_node(state: SurveillanceState) -> dict:
    """Executes Hybrid Search + Reranking on unstructured comms."""
    context = hybrid_retriever.search(state['ticker'], state['involved_accounts'])
    return {"narrative_context": context, "supervisor_decision": "check_network"}

def network_analyst_node(state: SurveillanceState) -> dict:
    """Executes Graph Traversal to find hidden entity links."""
    context = graph_retriever.search(state['involved_accounts'])
    return {"graph_context": context, "supervisor_decision": "check_behavior"}

def behavioral_analyst_node(state: SurveillanceState) -> dict:
    """Executes Time-Series retrieval to establish behavioral baselines."""
    context = ts_retriever.search(state['involved_accounts'], state['ticker'])
    return {"baseline_context": context, "supervisor_decision": "synthesize"}

def compliance_synthesizer_node(state: SurveillanceState) -> dict:
    """Synthesizes all multi-modal context into a formal report."""
    prompt = f"""You are a Senior Market Surveillance Officer. Draft a formal Market Manipulation Report.
    
    Alert ID: {state['alert_id']}
    Ticker: {state['ticker']}
    Involved Accounts: {state['involved_accounts']}
    
    1. Unstructured Narratives (Hybrid Search):
    {chr(10).join(state['narrative_context'])}
    
    2. Entity Network (Graph Traversal):
    {chr(10).join(state['graph_context'])}
    
    3. Behavioral Baseline (Time-Series):
    {state['baseline_context']}
    
    Conclusion: Based on the Tri-Fecta of evidence, is this insider trading? 
    Provide a professional, actionable recommendation for the Compliance Committee.
    """
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"final_report": response.content, "supervisor_decision": "FINISH"}

4. Build and Compile the LangGraph

def route_supervisor(state: SurveillanceState) -> str:
    """Conditional edge routing based on Supervisor's decision."""
    decision = state["supervisor_decision"]
    routing_map = {
        "check_narratives": "narrative_investigator",
        "check_network": "network_analyst",
        "check_behavior": "behavioral_analyst",
        "synthesize": "compliance_synthesizer",
        "FINISH": "end"
    }
    return routing_map.get(decision, "end")

# Initialize Graph
workflow = StateGraph(SurveillanceState)

# Add Nodes
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("narrative_investigator", narrative_investigator_node)
workflow.add_node("network_analyst", network_analyst_node)
workflow.add_node("behavioral_analyst", behavioral_analyst_node)
workflow.add_node("compliance_synthesizer", compliance_synthesizer_node)

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

# Add Conditional Edges from Supervisor
workflow.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {
        "narrative_investigator": "narrative_investigator",
        "network_analyst": "network_analyst",
        "behavioral_analyst": "behavioral_analyst",
        "compliance_synthesizer": "compliance_synthesizer",
        "end": END
    }
)

# Worker nodes route back to supervisor to re-evaluate state
workflow.add_edge("narrative_investigator", "supervisor")
workflow.add_edge("network_analyst", "supervisor")
workflow.add_edge("behavioral_analyst", "supervisor")
workflow.add_edge("compliance_synthesizer", END)

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

5. Execute the Real-Time Investigation

if __name__ == "__main__":
    # Simulate an incoming Market Surveillance Alert
    initial_state = {
        "alert_id": "SURV-2026-991",
        "ticker": "$ACME",
        "involved_accounts": ["ACC-881", "ACC-882"],
        "messages": [],
        "narrative_context": [],
        "graph_context": [],
        "baseline_context": "",
        "supervisor_decision": "check_narratives",
        "final_report": ""
    }
    
    # Config for memory (thread_id allows resuming this exact investigation)
    config = {"configurable": {"thread_id": "SURV-2026-991"}}
    
    print("--- Starting Project Argus Market Surveillance Graph ---\n")
    
    # Stream the execution
    for event in graph.stream(initial_state, config):
        for node_name, node_output in event.items():
            print(f"[Node Executed: {node_name}]")
            if node_name == "supervisor":
                print(f" -> Routing to: {node_output.get('supervisor_decision')}\n")
            elif node_name == "compliance_synthesizer":
                print("\n" + "="*60)
                print("FINAL MARKET SURVEILLANCE REPORT:")
                print("="*60)
                print(node_output.get('final_report'))
                print("="*60 + "\n")

Part 5: Enterprise Production Considerations

To deploy this in a Tier-1 financial institution, implement the following:

  1. Structured Output for Routing: In the supervisor_node, replace string parsing with LangChain's with_structured_output (Pydantic model). This guarantees the supervisor returns a valid enum for routing, preventing graph crashes.

  2. Persistent Checkpointing (PostgresSaver): Replace MemorySaver with langgraph.checkpoint.postgres.PostgresSaver. Market investigations take days. Persistent memory allows a human analyst to log in on Day 2, ask the agent "What did we find on the graph?", and the agent will recall the exact state without re-running the expensive retrievals.

  3. Human-in-the-Loop (Interrupts): Add interrupt_before=["compliance_synthesizer"] in the compile() method. Financial regulations require human sign-off before filing a Suspicious Activity Report (SAR) or freezing accounts. The graph should pause, present the drafted report to a UI, and await human approval.

  4. Retrieval Latency & Caching: Graph traversals and Hybrid searches can be slow. Implement a Redis caching layer for the Network_Analyst and Behavioral_Analyst nodes. If the graph relationships for ACC-881 were calculated yesterday, serve them from cache.

  5. Auditability (LangSmith): Wrap the execution in LangSmith. In market surveillance, you must prove to regulators (e.g., SEC, FCA) exactly why an alert was generated. LangSmith traces provide an immutable log of the retrieved narratives, graph links, and statistical baselines used to make the final decision.

Membership not found