Resilience in Real-Time: Handling Duplicates, Reordering, and Replay in Enterprise Graph RAG

Introduction

In enterprise systems that rely on real-time data streams such as financial trading platforms, IoT sensor networks, or supply chain trackers data rarely arrives in a pristine, linear fashion. Events are often duplicated due to network retries, arrive out of order due to latency variations, or need to be replayed during system recovery. When building Graph RAG (Retrieval-Augmented Generation) systems that ingest this data to build dynamic knowledge graphs, these anomalies can be catastrophic. A duplicate trade event can double-count risk exposure; an out-of-order compliance alert can trigger false positives; and a naive replay can corrupt the historical state of the graph. This article details how we engineered resilience against these three challenges using LangGraph’s persistent state management and provides a complete Proof of Concept (POC).

The Triad of Data Chaos: Duplicates, Out-of-Order, and Replay

1. Duplicate Events

• The Problem: Network glitches cause producers to send the same message multiple times. If the system processes each one, it leads to inflated metrics and corrupted graph edges.

• The Solution: Idempotency Keys. Every event is assigned a unique ID. Before processing, the system checks its persistent memory (checkpoint) to see if this ID has already been handled. If yes, it skips processin1. Duplicate Eventsg but acknowledges receipt.

Out-of-Order Events

• The Problem: Event B (Trade Execution) arrives before Event A (Trade Order). Processing B first creates a "phantom" execution without a corresponding order in the graph.

• The Solution: Event Watermarking & Buffering. The system maintains a logical clock or timestamp. If an event arrives with a timestamp earlier than the current state’s "high-water mark," it is either buffered for reordering or flagged for special handling by a dedicated "Reconciliation Agent."

Replay Scenarios

• The Problem: After a crash, the system must reprocess a batch of historical events. Without care, this replays duplicates and overwrites valid recent state.

• The Solution: State Versioning & Checkpointing. LangGraph’s MemorySaver allows us to restore the graph to a specific point in time. During replay, we use idempotency keys to ensure that events already processed in the restored state are skipped, while missing events are integrated seamlessly.

Real-Time Use Case: High-Frequency Trade Surveillance & Compliance

The Scenario: A fintech firm monitors stock trades for insider trading.

• Event 1: ORDER_PLACED (ID: 101, Time: T1)

• Event 2: TRADE_EXECUTED (ID: 102, Time: T2)

• Anomaly: Due to network lag, TRADE_EXECUTED (ID: 102) arrives before ORDER_PLACED (ID: 101). Then, ORDER_PLACED (ID: 101) is sent twice due to a retry.

The Workflow: Our LangGraph system uses a Guardian Agent to check idempotency keys. It buffers the out-of-order TRADE_EXECUTED until the ORDER_PLACED arrives. It rejects the duplicate ORDER_PLACED. Once both valid events are present, the Graph Builder Agent constructs the trade lineage in the knowledge graph, and the Compliance Agent checks it against RAG-retrieved regulatory policies.

Enterprise Multi-Agent LangGraph Architecture

We use LangGraph’s conditional edges and persistent checkpoints to manage event flow:

1. Guardian Agent: Checks for duplicates (idempotency) and timestamps (ordering).

2. Buffer/Reconciliation Agent: Holds out-of-order events until dependencies are met.

3. Graph Builder Agent: Updates the knowledge graph with validated events.

4. Compliance Agent: Uses RAG to check the updated graph against regulations.

Step-by-Step POC Implementation

Step 1: Defining State, Memory, and Idempotency Logic

We define a state that tracks processed IDs and buffered events.

# backend/graph_state.py
from typing import TypedDict, List, Annotated, Optional, Dict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class TradeSurveillanceState(TypedDict):
    incoming_event: Optional[Dict]  # {id, type, timestamp, data}
    processed_ids: List[str]        # Idempotency tracking
    event_buffer: List[Dict]        # For out-of-order events
    graph_updates: List[str]        # Validated graph changes
    compliance_alert: str
    agent_trace: Annotated[List[str], "Audit trail"]

memory = MemorySaver()

Step 2: Building the Resilient Multi-Agent Workflow

# backend/agents.py
from .graph_state import TradeSurveillanceState, memory

def guardian_agent(state: TradeSurveillanceState):
    """Handles Deduplication and Ordering."""
    event = state["incoming_event"]
    if not event:
        return {"agent_trace": ["Guardian: No new event."]}
      
    eid = event["id"]
    trace = []
  
    # 1. Check for Duplicates (Idempotency)
    if eid in state["processed_ids"]:
        trace.append(f"Guardian: Duplicate event {eid} detected and ignored.")
        return {"agent_trace": trace, "incoming_event": None}
  
    # 2. Check for Out-of-Order (Simple Timestamp Check)
    # In production, compare against a global watermark
    if event["type"] == "TRADE_EXECUTED" and "ORDER_PLACED" not in [e["type"] for e in state["event_buffer"]] and eid != "ORD-101":
         # Simulate buffering if order hasn't arrived yet (simplified logic)
         trace.append(f"Guardian: Event {eid} out of order. Buffering.")
         new_buffer = state["event_buffer"] + [event]
         return {"event_buffer": new_buffer, "agent_trace": trace, "incoming_event": None}
       
    # Mark as processed
    new_ids = state["processed_ids"] + [eid]
    trace.append(f"Guardian: Event {eid} validated and accepted.")
    return {"processed_ids": new_ids, "incoming_event": event, "agent_trace": trace}

def graph_builder_agent(state: TradeSurveillanceState):
    """Updates the Knowledge Graph with validated events."""
    event = state["incoming_event"]
    if not event:
        return {"agent_trace": ["Graph Builder: No event to process."]}
      
    update = f"Added Node: {event['type']} (ID: {event['id']}) -> Graph"
    trace = [f"Graph Builder: Updated graph with {event['id']}."]
  
    # Process any buffered events that are now ready
    final_buffer = []
    for buf_event in state["event_buffer"]:
        # Simple logic: if we just processed an ORDER, process buffered EXECUTIONS
        if event["type"] == "ORDER_PLACED" and buf_event["type"] == "TRADE_EXECUTED":
             update += f" | Added Node: {buf_event['type']} (ID: {buf_event['id']}) from Buffer"
             trace.append(f"Graph Builder: Processed buffered event {buf_event['id']}.")
             # Add buffered ID to processed list
             state["processed_ids"].append(buf_event["id"])
        else:
             final_buffer.append(buf_event)
           
    return {
        "graph_updates": state["graph_updates"] + [update],
        "event_buffer": final_buffer,
        "agent_trace": state["agent_trace"] + trace
    }

def compliance_agent(state: TradeSurveillanceState):
    """RAG-based Compliance Check."""
    # Mock RAG Retrieval
    policy = "Regulation X: All trades must have a pre-existing order within 5 seconds."
  
    alert = "COMPLIANCE CLEAR"
    # Simple check: if we have both order and execution in updates, it's likely valid
    if len(state["graph_updates"]) >= 2:
        alert = "COMPLIANCE CLEAR: Valid Trade Lineage Detected."
    else:
        alert = "REVIEW REQUIRED: Incomplete Lineage."
      
    trace = [f"Compliance Agent: Checked against policy. Result: {alert}"]
    return {"compliance_alert": alert, "agent_trace": state["agent_trace"] + trace}

# Routing Logic
def route_after_guardian(state: TradeSurveillanceState):
    if state["incoming_event"]:
        return "build_graph"
    return END

# Build Graph
workflow = StateGraph(TradeSurveillanceState)
workflow.add_node("guardian", guardian_agent)
workflow.add_node("build_graph", graph_builder_agent)
workflow.add_node("compliance", compliance_agent)

workflow.set_entry_point("guardian")
workflow.add_conditional_edges("guardian", route_after_guardian, {
    "build_graph": "build_graph",
    END: END
})
workflow.add_edge("build_graph", "compliance")
workflow.add_edge("compliance", END)

app = workflow.compile(checkpointer=memory)

Step 3: The FastAPI Backend

# backend/main.py
from fastapi import FastAPI
from pydantic import BaseModel
from .agents import app
from typing import Optional

app_api = FastAPI(title="Resilient Trade Surveillance POC")

class TradeEvent(BaseModel):
    id: str
    type: str  # ORDER_PLACED, TRADE_EXECUTED
    timestamp: float
    data: dict
    thread_id: str = "surveillance_thread_01"

@app_api.post("/process-event")
async def process_event(event: TradeEvent):
    config = {"configurable": {"thread_id": event.thread_id}}
  
    # Retrieve current state to maintain continuity
    initial_state = {
        "incoming_event": event.dict(),
        "processed_ids": [],
        "event_buffer": [],
        "graph_updates": [],
        "compliance_alert": "",
        "agent_trace": []
    }
  
    # In a real app, we would load the previous state from memory here
    # For demo, we invoke fresh but pass the event
  
    final_state = app.invoke(initial_state, config)
    return {
        "status": "PROCESSED",
        "alert": final_state["compliance_alert"],
        "graph_updates": final_state["graph_updates"],
        "trace": final_state["agent_trace"]
    }

Step 4: The Streamlit Frontend

# frontend/app.py
import streamlit as st
import requests
import time

st.set_page_config(page_title="Resilient Event Processing", layout="wide")
st.title("  Enterprise Event Resilience: Duplicates, Ordering & Replay")

st.sidebar.header("Simulate Event Stream")
event_type = st.sidebar.selectbox("Event Type", ["ORDER_PLACED", "TRADE_EXECUTED"])
event_id = st.sidebar.text_input("Event ID", "ORD-101")
is_duplicate = st.sidebar.checkbox("Simulate Duplicate?")
thread_id = "surv_thread_01"

if st.sidebar.button("Send Event"):
    if is_duplicate:
        st.warning(f"Sending Duplicate Event: {event_id}")
  
    payload = {
        "id": event_id,
        "type": event_type,
        "timestamp": time.time(),
        "data": {"symbol": "AAPL", "qty": 100},
        "thread_id": thread_id
    }
  
    response = requests.post("http://localhost:8000/process-event", json=payload)
  
    if response.status_code == 200:
        data = response.json()
        st.success("Event Processed")
        st.json(data)
    else:
        st.error("Error processing event")

Conclusion

Handling the chaos of real-time data streams is not optional for enterprise Graph RAG systems it is a foundational requirement. By leveraging LangGraph’s persistent state, we can implement robust idempotency checks to kill duplicates, buffering logic to handle out-of-order events, and checkpointing to support safe replays. This architecture ensures that our knowledge graph remains accurate and compliant, even when the underlying data feed is messy. The result is a resilient surveillance system that trusts its data, protects its state, and delivers reliable insights in real-time.