In the realm of enterprise financial systems, detecting anomalies in streaming time-series data—such as transaction amounts—is a high-stakes, low-latency challenge. Traditional static thresholds (e.g., "flag anything over $10,000") generate massive false-positive rates. Pure Machine Learning models can be computationally expensive to run on every single micro-transaction. The modern enterprise solution? A Multi-Agent System powered by LangGraph, combining the deterministic speed of Sliding Window Statistical Algorithms with the contextual depth of Retrieval-Augmented Generation (RAG), all tied together with persistent State and Memory.
In this end-to-end guide, we will build a real-time Cross-Border Wire Transfer Fraud Detection system. We will use a sliding window to detect statistical anomalies in real time, and if an anomaly is detected, a multi-agent LangGraph workflow will trigger to investigate the context using RAG before making a final decision.
The Real-World Use Case: Cross-Border Wire Transfers
Imagine an enterprise bank processing thousands of cross-border wire transfers per minute.
The Problem
A corporate client usually sends $50,000 to a known vendor in Germany every Friday. Suddenly, on a Tuesday, a $450,000 transfer is initiated to a new beneficiary in a high-risk jurisdiction.
The Challenge
A simple rule engine might block it, halting legitimate business. A pure LLM approach is too slow and expensive to evaluate every transaction.
The Solution
System 1 (Fast & Deterministic): A sliding window algorithm calculates the statistical deviation of the transaction amount in real time.
System 2 (Slow & Contextual): If the deviation crosses a threshold, a LangGraph Multi-Agent workflow is triggered. It uses RAG to retrieve the client's historical profile, Anti-Money Laundering (AML) policies, and past fraud case studies.
Memory & State: The system remembers the client's historical baseline and past investigations using LangGraph's persistent memory.
The Core Algorithm: Sliding Window Anomaly Detection
For streaming time-series, we cannot recalculate the mean and standard deviation from scratch for every transaction. Instead, we maintain a Sliding Window of the last N transactions for a specific user.
When a new transaction xₜ arrives:
We append xₜ to the window and remove the oldest transaction if the window exceeds size N.
We calculate the rolling mean (μ) and standard deviation (σ).
If Z > Threshold, it is flagged as a statistical anomaly.
Note: In highly skewed financial data, enterprises often use the Modified Z-Score based on Median Absolute Deviation (MAD), but we will use the standard Z-score here for clarity.
End-to-End Python Implementation
We will use LangGraph to orchestrate the agents, LangChain for the RAG pipeline, and NumPy for the sliding window math.
Prerequisites
pip install langgraph langchain-openai langchain-community numpy faiss-cpu
Step 1: Define the State and Memory
In LangGraph, the State is the single source of truth. We will use a Checkpointer to persist this state, meaning the system "remembers" the sliding window and past decisions for each user across different sessions.
import operator
from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# Define the State of our Graph
class TransactionState(TypedDict):
transaction_id: str
user_id: str
amount: float
# The sliding window of past amounts
window_history: List[float]
# Outputs
z_score: float
is_anomaly: bool
rag_context: str
final_decision: str
# Memory tracking
investigation_notes: str
Step 2: The Sliding Window Engine (Node 1)
This node acts as our fast, deterministic filter. It updates the state with the new transaction and calculates the anomaly score.
import numpy as np
WINDOW_SIZE = 20
ANOMALY_THRESHOLD = 2.5 # 2.5 Standard Deviations
def sliding_window_anomaly_node(state: TransactionState) -> Dict[str, Any]:
"""Calculates the Z-score using a sliding window of transaction amounts."""
amount = state["amount"]
window = state.get("window_history", [])
# Update the sliding window
window.append(amount)
if len(window) > WINDOW_SIZE:
window.pop(0) # Remove oldest
# Calculate statistics
if len(window) < 3:
# Not enough data to establish a baseline
return {
"window_history": window,
"z_score": 0.0,
"is_anomaly": False
}
mean = np.mean(window)
std = np.std(window)
# Avoid division by zero
z_score = abs(amount - mean) / std if std > 0 else 0.0
is_anomaly = z_score > ANOMALY_THRESHOLD
return {
"window_history": window,
"z_score": round(z_score, 2),
"is_anomaly": is_anomaly
}
Step 3: The RAG Knowledge Base (Node 2)
When an anomaly is detected, we need context. We will set up a lightweight in-memory Vector Store (FAISS) containing AML policies and historical user profiles.
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
# Initialize Vector Store (In production, use Pinecone, Milvus, or pgvector)
embeddings = OpenAIEmbeddings()
vectorstore = InMemoryVectorStore(embeddings)
# Mocking Enterprise Knowledge Base Data
documents = [
"User U123 is a verified corporate entity. Standard wire limit is $100,000. Exceptions require VP approval.",
"User U456 has a history of high-volume tech supply chain payments. Average transaction is $45,000.",
"AML Policy: Transfers to Jurisdiction X over $50,000 require enhanced due diligence (EDD) and source of funds verification.",
"Past Fraud Case 884: Business Email Compromise (BEC) where attacker changed beneficiary to a new shell company in Jurisdiction Y."
]
vectorstore.add_texts(documents, ids=["doc1", "doc2", "doc3", "doc4"])
def rag_investigation_node(state: TransactionState) -> Dict[str, Any]:
"""Retrieves context from the Vector DB to investigate the anomaly."""
user_id = state["user_id"]
amount = state["amount"]
# Create a contextual query for the retriever
query = f"Transaction anomaly for {user_id} of amount {amount}. Retrieve user profile, limits, and AML policies."
# Retrieve relevant documents
retrieved_docs = vectorstore.similarity_search(query, k=2)
context = "\n".join([doc.page_content for doc in retrieved_docs])
return {"rag_context": context}
Step 4: The Decision Agent (Node 3)
This node uses an LLM to synthesize the statistical anomaly (Z-score) and the RAG context to make a final enterprise decision.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o", temperature=0)
def decision_agent_node(state: TransactionState) -> Dict[str, Any]:
"""LLM Agent synthesizes math and RAG context to make a final decision."""
prompt = ChatPromptTemplate.from_template("""
You are an Enterprise Fraud Investigation Agent.
STATISTICAL ALERT:
Transaction Amount: ${amount}
Z-Score (Deviation from user's rolling average): {z_score}
RETRIEVED CONTEXT (RAG):
{context}
INSTRUCTIONS:
Based on the statistical deviation and the retrieved policies/user history,
make a final decision.
Output ONLY one of the following:
[APPROVE], [FLAG_FOR_REVIEW], or [BLOCK].
Then, provide a brief 1-sentence justification.
""")
chain = prompt | llm
response = chain.invoke({
"amount": state["amount"],
"z_score": state["z_score"],
"context": state["rag_context"]
})
return {
"final_decision": response.content,
"investigation_notes": f"Investigated Z-score {state['z_score']} for {state['user_id']}"
}
Step 5: Routing and Graph Compilation
We need a router to decide whether to trigger the expensive RAG/LLM pipeline or just auto-approve the transaction.
def route_transaction(state: TransactionState) -> str:
"""Conditional edge: Route to RAG if anomaly, else approve."""
if state["is_anomaly"]:
return "investigate_with_rag"
return "auto_approve"
def auto_approve_node(state: TransactionState) -> Dict[str, Any]:
return {
"final_decision": "[APPROVE] Transaction is within normal statistical bounds."
}
# Build the LangGraph
workflow = StateGraph(TransactionState)
# Add Nodes
workflow.add_node("sliding_window_math", sliding_window_anomaly_node)
workflow.add_node("investigate_with_rag", rag_investigation_node)
workflow.add_node("llm_decision", decision_agent_node)
workflow.add_node("auto_approve", auto_approve_node)
# Define Edges
workflow.add_edge(START, "sliding_window_math")
workflow.add_conditional_edges(
"sliding_window_math",
route_transaction,
{
"investigate_with_rag": "investigate_with_rag",
"auto_approve": "auto_approve"
}
)
workflow.add_edge("investigate_with_rag", "llm_decision")
workflow.add_edge("llm_decision", END)
workflow.add_edge("auto_approve", END)
# Compile with Memory (Checkpointer)
# The thread_id will be the user_id, ensuring we remember their sliding window!
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)
Step 6: Simulating the Real-Time Stream
Let's simulate a stream of transactions for a specific user (U456) to see the system in action.
import uuid
def process_stream():
user_id = "U456"
# Simulate a stream of normal transactions to build the baseline
normal_amounts = [45000, 46000, 44000, 45500, 47000, 43000, 45000, 46500]
print("--- Building Baseline (Normal Transactions) ---")
for amt in normal_amounts:
config = {"configurable": {"thread_id": user_id}}
state_update = {
"transaction_id": str(uuid.uuid4()),
"user_id": user_id,
"amount": amt
}
# Invoke graph. Memory ensures the window_history persists!
result = graph.invoke(state_update, config)
print(
f"Amount: ${amt} | "
f"Z-Score: {result['z_score']} | "
f"Decision: {result['final_decision']}"
)
print("\n--- Injecting Anomaly (Potential Fraud) ---")
# Inject a massive anomaly
anomaly_config = {"configurable": {"thread_id": user_id}}
anomaly_update = {
"transaction_id": str(uuid.uuid4()),
"user_id": user_id,
"amount": 250000 # Massive spike compared to ~45k average
}
result = graph.invoke(anomaly_update, config=anomaly_config)
print("\nANOMALY DETECTED")
print(f"Amount: ${anomaly_update['amount']}")
print(f"Z-Score: {result['z_score']}")
print(f"RAG Context Retrieved:\n{result['rag_context']}")
print(f"Final LLM Decision:\n{result['final_decision']}")
print(f"Investigation Notes: {result['investigation_notes']}")
# Run the simulation
if __name__ == "__main__":
# Ensure you have set your OPENAI_API_KEY environment variable
process_stream()
Enterprise Deployment Considerations
While the code above runs in memory, deploying this to a production enterprise environment requires specific architectural adjustments.
State Management (The Sliding Window)
In the demo, the window_history is stored in LangGraph's memory. In a high-throughput enterprise system, you should offload the sliding window state to Redis or Apache Kafka Streams. The LangGraph node would simply query Redis for the user's rolling window, calculate the Z-score, and update Redis.
RAG Vector Database
Replace InMemoryVectorStore with an enterprise-grade vector database like Milvus, Pinecone, or pgvector. Ensure your embeddings model is optimized for financial terminology (e.g., using a fine-tuned BGE model).
Latency & Async Execution
LangGraph supports asynchronous execution. The RAG retrieval and LLM inference should be wrapped in async def nodes to prevent blocking the main transaction processing thread.
Observability
Integrate LangSmith to trace the LangGraph execution. In financial services, you must be able to audit exactly why the LLM made a specific decision, including the exact RAG chunks retrieved and the mathematical Z-score at the time of inference.
Conclusion
By combining a Sliding Window Algorithm with a LangGraph Multi-Agent RAG architecture, we achieve the best of both worlds. The sliding window provides ultra-low latency, deterministic statistical filtering (System 1). When the math detects an anomaly, the LangGraph agents seamlessly take over, utilizing persistent memory and RAG to pull in vital business context (System 2).
This pattern not only drastically reduces false positives in fraud detection but also provides the explainability and auditability required by enterprise compliance and regulatory teams.
Join the conversation! Your thoughts help the community grow.