The Batch Loader Fallacy in Financial RAG
LangChain’s document loader ecosystem is fundamentally batch-oriented. PyPDFLoader, CSVLoader, and UnstructuredLoader assume documents arrive as complete, static files. This assumption collapses in financial services where data arrives as high-velocity streams: Bloomberg terminal feeds, SEC EDGAR filings, broker-dealer trade confirmations, and real-time market microstructure data. In these environments, waiting for a "complete document" means missing the alpha window entirely. A 10-K filing is a discrete artifact; a stream of earnings revisions, insider transactions, and options flow is not. You need loaders that treat streaming data as first-class documents with temporal semantics, partial state management, and backpressure-aware ingestion—all integrated into a stateful multi-agent RAG system. This article demonstrates building custom streaming document loaders for a real-time institutional trading desk, integrated into a LangGraph multi-agent architecture with persistent memory and state.
Real-Time Use Case: Institutional Event-Driven Trading Desk
The Scenario
A quantitative hedge fund’s event-driven desk trades around corporate events. Analysts must synthesize signals from four concurrent high-velocity streams:
SEC EDGAR Stream: 8-K filings, Form 4 insider transactions (arrives within seconds of filing)
Earnings Revision Stream: Broker estimates updated in real-time via Refinitiv API
Options Flow Stream: Unusual block trades detected by proprietary scanner
News Wire Stream: PR Newswire/Business Wire filtered for covered universe
An analyst asks: "What's driving the sudden SPX put skew in NVDA? Cross-reference with any recent insider activity and estimate revisions."
The system must ingest, index, and reason over data arriving at 50-200 events/second while maintaining conversation state across multiple analyst queries.
Why Standard Loaders Fail
| Requirement | Batch Loader Limitation | Streaming Loader Solution |
|---|---|---|
| Sub-second latency | Wait for file completion | Process events as they arrive |
| Partial document assembly | All-or-nothing loading | Incremental chunking with sequence tracking |
| Temporal ordering | File modification time only | Event-level timestamps with causal ordering |
| Backpressure handling | OOM on burst traffic | Bounded buffers with adaptive throttling |
| Deduplication across streams | Post-hoc dedup | Content-hash dedup at ingestion |
| State continuity | Stateless reload | Cursor-based resumption after restart |
| Schema evolution | Fixed schema per loader | Adaptive parsing with version detection |
Architecture: Streaming Loaders Inside LangGraph
Implementation
Step 1: Define Streaming-Aware State
Standard RAG state assumes static documents. Streaming RAG requires cursors, sequence numbers, and buffer health metrics.
from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from datetime import datetime
import operator
class StreamCursor(TypedDict):
"""Resumable position marker for each data stream."""
stream_id: str
last_sequence: int
last_timestamp: datetime
checkpoint_hash: str # For integrity verification
class IngestedEvent(TypedDict):
"""Normalized event envelope across all streams."""
event_id: str
stream_id: str
sequence: int
timestamp: datetime
ticker: str
event_type: str # "8k_filing", "estimate_revision", "options_block", "news_wire"
content: str
metadata: Dict[str, Any]
content_hash: str
class TradingDeskState(TypedDict):
messages: Annotated[list, add_messages]
# Resumable cursors for each stream
stream_cursors: Dict[str, StreamCursor]
# Recently ingested events (bounded window)
ingested_events: Annotated[List[IngestedEvent], operator.add]
# Buffer health metrics for backpressure
buffer_health: Dict[str, Dict[str, Any]]
# Cross-stream signal cache (avoids re-computation)
signal_cache: Dict[str, Any]
# Current analyst query
current_query: str
# Generated trade thesis
trade_thesis: Optional[str]
# Audit trail
audit_trail: Annotated[List[Dict], operator.add]Step 2: Base Streaming Loader Infrastructure
All custom loaders inherit from this base class that handles the hard parts: async iteration, backpressure, deduplication, and cursor management.
import asyncio
import hashlib
from abc import ABC, abstractmethod
from collections import deque
from typing import AsyncIterator, Set, Callable, Optional
from langchain_core.documents import Document
class AsyncEventBuffer:
"""
Backpressure-aware bounded buffer for streaming events.
Prevents OOM during burst traffic while maintaining ordering.
"""
def __init__(self, max_size: int = 10000, flush_threshold: float = 0.8):
self._buffer: deque = deque(maxlen=max_size)
self._max_size = max_size
self._flush_threshold = flush_threshold
self._lock = asyncio.Lock()
self._seen_hashes: Set[str] = set()
self._dropped_count = 0
@property
def utilization(self) -> float:
return len(self._buffer) / self._max_size
@property
def should_flush(self) -> bool:
return self.utilization >= self._flush_threshold
async def put(self, event: IngestedEvent) -> bool:
"""Add event with dedup. Returns False if dropped due to backpressure."""
async with self._lock:
# Content-hash deduplication across streams
if event["content_hash"] in self._seen_hashes:
return True # Duplicate, silently skip
if len(self._buffer) >= self._max_size:
self._dropped_count += 1
return False # Backpressure: drop oldest
self._buffer.append(event)
self._seen_hashes.add(event["content_hash"])
# Evict old hashes to prevent unbounded growth
if len(self._seen_hashes) > self._max_size * 2:
# Keep only hashes for events still in buffer
active_hashes = {e["content_hash"] for e in self._buffer}
self._seen_hashes = active_hashes
return True
async def drain(self) -> List[IngestedEvent]:
"""Atomically drain all buffered events."""
async with self._lock:
events = list(self._buffer)
self._buffer.clear()
return events
def health_metrics(self) -> Dict[str, Any]:
return {
"utilization": self.utilization,
"buffered": len(self._buffer),
"dropped": self._dropped_count,
"unique_hashes": len(self._seen_hashes)
}
class BaseStreamingLoader(ABC):
"""
Abstract base for all streaming financial data loaders.
Handles async iteration, cursor persistence, and normalization.
Subclasses implement only source-specific parsing.
"""
def __init__(
self,
stream_id: str,
buffer: AsyncEventBuffer,
cursor_store: Any, # Redis/Postgres cursor persistence
vectorstore: Any,
batch_size: int = 50,
poll_interval_ms: int = 100
):
self.stream_id = stream_id
self.buffer = buffer
self.cursor_store = cursor_store
self.vectorstore = vectorstore
self.batch_size = batch_size
self.poll_interval_ms = poll_interval_ms
self._running = False
@abstractmethod
async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
"""Source-specific: fetch raw events since cursor position."""
...
@abstractmethod
def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
"""Source-specific: normalize raw event to common schema."""
...
def _compute_content_hash(self, content: str, ticker: str, event_type: str) -> str:
"""Deterministic hash for cross-stream deduplication."""
canonical = f"{ticker}|{event_type}|{content.strip()}"
return hashlib.sha256(canonical.encode()).hexdigest()[:32]
async def get_cursor(self) -> StreamCursor:
"""Load persisted cursor or initialize."""
cursor = await self.cursor_store.get(f"cursor:{self.stream_id}")
if cursor is None:
return {
"stream_id": self.stream_id,
"last_sequence": 0,
"last_timestamp": datetime.min,
"checkpoint_hash": ""
}
return cursor
async def save_cursor(self, cursor: StreamCursor):
"""Persist cursor for resumability."""
await self.cursor_store.set(f"cursor:{self.stream_id}", cursor)
async def ingest_loop(self):
"""Main ingestion loop with backpressure awareness."""
self._running = True
cursor = await self.get_cursor()
while self._running:
try:
# Fetch new events from source
raw_events = await self._fetch_batch(cursor)
if not raw_events:
await asyncio.sleep(self.poll_interval_ms / 1000)
continue
# Normalize and buffer
accepted = 0
for i, raw in enumerate(raw_events):
seq = cursor["last_sequence"] + i + 1
event = self._normalize_event(raw, seq)
success = await self.buffer.put(event)
if success:
accepted += 1
cursor = {
"stream_id": self.stream_id,
"last_sequence": seq,
"last_timestamp": event["timestamp"],
"checkpoint_hash": event["content_hash"]
}
# Flush to vectorstore when buffer is hot
if self.buffer.should_flush:
events = await self.buffer.drain()
docs = [
Document(
page_content=e["content"],
metadata={
"event_id": e["event_id"],
"stream_id": e["stream_id"],
"ticker": e["ticker"],
"event_type": e["event_type"],
"timestamp": e["timestamp"].isoformat(),
"sequence": e["sequence"],
"content_hash": e["content_hash"]
}
)
for e in events
]
await self.vectorstore.aadd_documents(docs)
# Persist cursor after successful batch
await self.save_cursor(cursor)
except Exception as e:
# Log but don't crash; retry on next cycle
print(f"[{self.stream_id}] Ingestion error: {e}")
await asyncio.sleep(1)
async def stop(self):
self._running = FalseStep 3: Concrete Loader — SEC EDGAR Stream
import aiohttp
from lxml import etree
class EDGARStreamLoader(BaseStreamingLoader):
"""
Streams SEC 8-K and Form 4 filings from EDGAR's Atom feed.
Handles XML parsing, exhibit extraction, and rate limiting.
"""
EDGAR_ATOM_URL = "https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&type=8-K&dateb=&owner=include&count=100&search_text=&action=getcompany&output=atom"
RATE_LIMIT_DELAY = 0.1 # SEC allows 10 req/sec
def __init__(self, buffer, cursor_store, vectorstore, **kwargs):
super().__init__("edgar_stream", buffer, cursor_store, vectorstore, **kwargs)
self._session: Optional[aiohttp.ClientSession] = None
self._rate_limiter = asyncio.Semaphore(10)
async def _get_session(self) -> aiohttp.ClientSession:
if self._session is None or self._session.closed:
headers = {"User-Agent": "TradingDesk RAG v1.0 [email protected]"}
self._session = aiohttp.ClientSession(headers=headers)
return self._session
async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
async with self._rate_limiter:
session = await self._get_session()
params = {
"start": cursor["last_sequence"],
"count": self.batch_size
}
async with session.get(self.EDGAR_ATOM_URL, params=params) as resp:
resp.raise_for_status()
xml_content = await resp.text()
# Parse Atom feed entries
root = etree.fromstring(xml_content.encode())
nsmap = {"atom": "http://www.w3.org/2005/Atom"}
entries = root.findall("atom:entry", nsmap)
results = []
for entry in entries:
title = entry.findtext("atom:title", "", nsmap)
link = entry.find("atom:link", nsmap).get("href")
updated = entry.findtext("atom:updated", "", nsmap)
# Extract CIK and form type from title
# Format: "1000275 - 8-K - NVIDIA CORP"
parts = title.split(" - ")
cik = parts[0].strip() if len(parts) > 0 else ""
form_type = parts[1].strip() if len(parts) > 1 else ""
company = parts[2].strip() if len(parts) > 2 else ""
results.append({
"cik": cik,
"form_type": form_type,
"company": company,
"filing_url": link,
"filed_at": updated,
"raw_xml": etree.tostring(entry, encoding="unicode")
})
await asyncio.sleep(self.RATE_LIMIT_DELAY)
return results
def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
content = f"SEC {raw['form_type']} Filing: {raw['company']} (CIK: {raw['cik']})\nFiled: {raw['filed_at']}\nURL: {raw['filing_url']}"
# Map company name to ticker (in production, use a lookup service)
ticker = self._resolve_ticker(raw["company"])
return {
"event_id": f"edgar_{raw['cik']}_{sequence}",
"stream_id": "edgar_stream",
"sequence": sequence,
"timestamp": datetime.fromisoformat(raw["filed_at"].replace("Z", "+00:00")),
"ticker": ticker,
"event_type": "8k_filing" if "8-K" in raw["form_type"] else "form4",
"content": content,
"metadata": {
"cik": raw["cik"],
"form_type": raw["form_type"],
"filing_url": raw["filing_url"]
},
"content_hash": self._compute_content_hash(content, ticker, raw["form_type"])
}
def _resolve_ticker(self, company_name: str) -> str:
# Production: use SEC company tickers dataset or mapping service
mappings = {"NVIDIA CORP": "NVDA", "APPLE INC": "AAPL", "TESLA INC": "TSLA"}
return mappings.get(company_name.upper(), "UNKNOWN")
async def close(self):
if self._session and not self._session.closed:
await self._session.close()Step 4: Concrete Loader — Options Flow Stream
class OptionsFlowLoader(BaseStreamingLoader):
"""
Streams unusual options block trades from proprietary scanner.
Handles WebSocket connection, binary protocol decoding, and reconnection.
Demonstrates non-HTTP streaming source integration.
"""
def __init__(self, ws_url: str, api_key: str, buffer, cursor_store, vectorstore, **kwargs):
super().__init__("options_flow", buffer, cursor_store, vectorstore, **kwargs)
self.ws_url = ws_url
self.api_key = api_key
self._ws = None
async def _fetch_batch(self, cursor: StreamCursor) -> List[Dict]:
"""WebSocket streams don't use cursors; we track by sequence."""
import websockets
if self._ws is None or self._ws.closed:
self._ws = await websockets.connect(
self.ws_url,
extra_headers={"Authorization": f"Bearer {self.api_key}"}
)
# Resume from last sequence
await self._ws.send(f'{{"resume_from": {cursor["last_sequence"]}}}')
events = []
try:
for _ in range(self.batch_size):
msg = await asyncio.wait_for(self._ws.recv(), timeout=1.0)
data = json.loads(msg)
events.append(data)
except asyncio.TimeoutError:
pass # No more events in this window
except websockets.ConnectionClosed:
self._ws = None # Will reconnect next cycle
return events
def _normalize_event(self, raw: Dict, sequence: int) -> IngestedEvent:
side = raw.get("side", "unknown").upper()
ticker = raw.get("symbol", "UNKNOWN")
strike = raw.get("strike", 0)
expiry = raw.get("expiry", "")
premium = raw.get("premium", 0)
volume = raw.get("volume", 0)
oi = raw.get("open_interest", 0)
content = (
f"OPTIONS BLOCK: {ticker} {expiry} ${strike} {side}\n"
f"Volume: {volume:,} | OI: {oi:,} | Premium: ${premium:,.0f}\n"
f"Vol/OI Ratio: {volume/max(oi,1):.2f} | "
f"Notional: ${premium * volume * 100:,.0f}"
)
event_type = "options_block_put" if side == "BUY" and raw.get("type") == "P" else "options_block_call"
return {
"event_id": f"opt_{ticker}_{sequence}",
"stream_id": "options_flow",
"sequence": sequence,
"timestamp": datetime.fromtimestamp(raw.get("ts", 0)),
"ticker": ticker,
"event_type": event_type,
"content": content,
"metadata": {
"strike": strike, "expiry": expiry, "side": side,
"volume": volume, "open_interest": oi, "premium": premium
},
"content_hash": self._compute_content_hash(content, ticker, event_type)
}Step 5: Integrate Streaming Loaders into LangGraph Agents
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
# Initialize shared infrastructure
shared_buffer = AsyncEventBuffer(max_size=50000)
cursor_store = RedisCursorStore(redis_url="redis://localhost:6379")
vectorstore = PGVectorStore(connection_string="postgresql://...")
# Start streaming loaders as background tasks
loaders = [
EDGARStreamLoader(shared_buffer, cursor_store, vectorstore, batch_size=20),
OptionsFlowLoader("wss://flow.scanner.io/ws", API_KEY, shared_buffer, cursor_store, vectorstore),
# EarningsStreamLoader(...), NewsWireLoader(...)
]
loader_tasks = [asyncio.create_task(loader.ingest_loop()) for loader in loaders]
# --- INGESTION VALIDATION NODE ---
async def ingestion_validation_node(state: TradingDeskState) -> dict:
"""Validates buffer health and routes stale/broken streams."""
health = shared_buffer.health_metrics()
alerts = []
if health["utilization"] > 0.95:
alerts.append("CRITICAL: Buffer near capacity, events being dropped")
if health["dropped"] > 100:
alerts.append(f"WARNING: {health['dropped']} events dropped since last check")
# Check stream freshness
for stream_id, cursor in state.get("stream_cursors", {}).items():
age = (datetime.utcnow() - cursor["last_timestamp"]).total_seconds()
if age > 300: # 5 min staleness threshold
alerts.append(f"STALE: {stream_id} last updated {age:.0f}s ago")
return {
"buffer_health": health,
"audit_trail": [{"event": "health_check", "alerts": alerts, "metrics": health}]
}
# --- CROSS-STREAM ANALYSIS NODE ---
analysis_prompt = ChatPromptTemplate.from_messages([
("system", """You are an event-driven trading analyst.
Synthesize signals from multiple concurrent financial streams.
Always cite event_id and timestamp for every claim.
Identify temporal causality: which event likely triggered another.
Flag conflicting signals explicitly."""),
("human", "Query: {query}\n\nRecent Events:\n{events_context}\n\nBuffer Health: {health}")
])
async def cross_stream_analysis_node(state: TradingDeskState) -> dict:
# Build temporally-ordered context from ingested events
events_sorted = sorted(state["ingested_events"], key=lambda e: e["timestamp"], reverse=True)
events_context = "\n\n".join(
f"[{e['event_id']} | {e['timestamp'].isoformat()} | {e['stream_id']}]\n"
f"{e['ticker']}: {e['content']}"
for e in events_sorted[:30] # Top 30 most recent
)
response = await analysis_prompt | llm | (lambda x: x.content)
thesis = await response.ainvoke({
"query": state["current_query"],
"events_context": events_context,
"health": json.dumps(state["buffer_health"])
})
return {"trade_thesis": thesis}
# --- BUILD GRAPH ---
workflow = StateGraph(TradingDeskState)
workflow.add_node("validate_ingestion", ingestion_validation_node)
workflow.add_node("analyze", cross_stream_analysis_node)
workflow.add_edge(START, "validate_ingestion")
workflow.add_conditional_edges(
"validate_ingestion",
lambda s: "analyze" if not any("CRITICAL" in a for a in s.get("audit_trail", [{}])[0].get("alerts", [])) else END,
{"analyze": "analyze", END: END}
)
workflow.add_edge("analyze", END)
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = workflow.compile(checkpointer=checkpointer)Step 6: Execute with Persistent Memory
config = {"configurable": {"thread_id": "trading-desk-nvda-20240805"}}
result = await app.ainvoke({
"current_query": "What's driving NVDA put skew? Cross-reference insider activity and estimate revisions.",
"messages": [],
"stream_cursors": {},
"ingested_events": [],
"buffer_health": {},
"signal_cache": {},
"trade_thesis": None,
"audit_trail": []
}, config=config)
print(result["trade_thesis"])
print(f"\nBuffer Utilization: {result['buffer_health']['utilization']:.1%}")
print(f"Events Ingested This Turn: {len(result['ingested_events'])}")Key Design Principles for Streaming Loaders
1. Decouple Fetch Rate from Processing Rate
The AsyncEventBuffer acts as a shock absorber. Sources may burst at 500 events/sec while vector indexing sustains only 50/sec. The buffer absorbs bursts and signals backpressure when full. Never let source velocity dictate processing cadence.
2. Make Cursors First-Class State
Batch loaders treat position as an implementation detail. Streaming loaders treat cursors as resumable, auditable, versioned state. Every cursor includes a checkpoint hash so you can detect corruption or tampering after restart.
3. Normalize Before Indexing
Each stream has unique schemas. Normalize to a common IngestedEvent envelope before buffering. This enables cross-stream deduplication, unified retrieval, and consistent metadata filtering in the vector store.
4. Hash Everything
Content hashes serve triple duty: deduplication across streams, integrity verification for cursors, and citation anchors for analyst responses. Compute hashes deterministically (ticker|event_type|content) so identical events from different sources produce identical hashes.
5. Monitor Buffer Health as Graph State
Buffer utilization isn't just an ops metric—it's analytical context. When the analysis agent sees 95% utilization, it knows recent events may be incomplete and should qualify its conclusions accordingly. This is impossible with batch loaders.
Conclusion
High-velocity financial RAG demands rethinking document loading from a batch operation to a streaming primitive. The custom loaders demonstrated here—EDGAR filings, options flow, earnings revisions—are not merely faster versions of existing loaders. They are architecturally different: they maintain state, enforce backpressure, normalize across heterogeneous sources, and expose their operational health as first-class graph state. When integrated into a LangGraph multi-agent system, streaming loaders transform RAG from a question-answering tool into a real-time situational awareness platform. The ingestion validation agent monitors data freshness, the analysis agent reasons over temporally-ordered cross-stream events, and the entire system resumes seamlessly after failures. The fundamental insight is that in streaming RAG, the loader is not a preprocessing step—it is a continuous agent. Treat it accordingly: give it state, give it error handling, give it observability, and integrate it deeply into your graph's reasoning lifecycle. The alpha is in the stream; make sure your architecture can actually drink from it.
Join the conversation! Your thoughts help the community grow.