When the Nepal Stock Exchange opened on a record‑breaking day last month, our three‑agent LLM system hit a wall: the market summary API throttled, the agents kept looping over contradictory technical indicators, and the latency spiked to over 12 seconds per tick. If you’re running a similar multi‑agent trading loop, you’ll feel that frustration. In this post I’ll show you how to add a semantic volatility router, an async token bucket tuned to provider limits, and a deterministic deadlock arbiter that together trim latency and cost without sacrificing signal quality.
Key Takeaways
- Use a TF‑IDF semantic router to gate expensive LLM calls when market volatility is low.
- Implement an async token bucket that respects tiered API limits and prioritizes risk‑sensitive agents.
- Fallback to a deterministic rule matrix when agents deadlock, preventing infinite retries.
- Validate every agent output with Pydantic to catch silent JSON parsing failures.
- Logging audit events to a file gives you a clear trace of routing decisions and token usage.
The Problem
Our previous post introduced a state‑machine orchestrator for three agents—Risk, Technical, and Valuation—each calling a large language model to debate trade signals. In production, the following pain points surfaced:
- API rate limits were hit during the 11 AM–3 PM NPT window when market data spikes.
- Agents sometimes looped for minutes debating conflicting indicators, exhausting token budgets.
- When the market was sideways, the LLM call was a waste of latency and cost.
- A missing JSON schema check let malformed agent responses slip through, causing downstream crashes.
Data and Sources
We pull two real data streams:
- Live market summary: https://www.nepalstock.com.np/api/nots/market-summary
- Historical floor sheet snapshots: sample_floorsheet.json
Data accessed on 2026‑10‑28.
Loading the Data
We fetch the live market payload and load the historical floor sheet from a local file. Errors are surfaced early so the pipeline can decide to fall back to the last‑known‑good data.
import requests, json, pathlib, logging, time, datetime, asyncio, numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from pydantic import BaseModel, ValidationError, Field
from typing import List, Dict, Any
# Configure logging for audit trail
logging.basicConfig(filename="nepse_agent_audit.log", level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s")
def load_market_summary() -> Dict[str, Any]:
try:
resp = requests.get("https://www.nepalstock.com.np/api/nots/market-summary", timeout=5)
resp.raise_for_status()
return resp.json()
except Exception as e:
logging.error(f"Market summary fetch failed: {e}")
raise
def load_floor_sheet() -> List[Dict[str, Any]]:
try:
path = pathlib.Path("sample_floorsheet.json")
with path.open() as f:
return json.load(f)
except Exception as e:
logging.error(f"Floor sheet load failed: {e}")
raise
The Core Logic
We break the logic into four layers: semantic volatility classification, token budgeting, agent debate, and deterministic fallback. Below is a high‑level view of each component.
1. Semantic Volatility Classification
We vectorise the market summary and compare it against embeddings of historical high‑volatility floor sheets. If the cosine similarity is below a threshold, we skip the LLM loop.
class VolatilityRouter:
def __init__(self, floor_sheets: List[Dict[str, Any]]):
self.vectorizer = TfidfVectorizer(stop_words="english")
# Build a corpus of high‑volatility floor sheet texts
high_vol_texts = [fs["description"] for fs in floor_sheets if fs["volatility"] == "high"]
self.high_vol_matrix = self.vectorizer.fit_transform(high_vol_texts)
def is_high_vol(self, market_summary: Dict[str, Any], threshold: float = 0.4) -> bool:
text = market_summary.get("summary", "")
vec = self.vectorizer.transform([text])
sim = cosine_similarity(vec, self.high_vol_matrix).mean()
return sim >= threshold
2. Asynchronous Token Budget Governor
We implement a simple token bucket that refills at a configurable rate. Each agent type has a priority; risk‑critical agents get first dibs.
class AsyncTokenBucket:
def __init__(self, capacity: int, refill_rate: float): # tokens per second
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.last_refill = time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self, amount: int = 1):
async with self.lock:
now = time.monotonic()
elapsed = now - self.last_refill
refill = int(elapsed * self.refill_rate)
if refill > 0:
self.tokens = min(self.capacity, self.tokens + refill)
self.last_refill = now
if self.tokens < amount:
raise RuntimeError("Token budget exhausted")
self.tokens -= amount
return amount
3. Deterministic Deadlock Arbitration
If the debate loop runs longer than a timeout or the agents keep voting split, we fall back to a hard‑coded rule matrix based on NEPSE circuit‑breaker thresholds.
def deterministic_fallback(market_summary: Dict[str, Any]) -> str:
# Simple rule: if overall % change > 10% then SELL, < -10% then BUY, else HOLD
pct_change = market_summary.get("pct_change", 0)
if pct_change >= 10:
return "SELL"
if pct_change <= -10:
return "BUY"
return "HOLD"
4. Pydantic Output Validation
We enforce a strict schema on the final trade verdict to catch malformed JSON from any agent.
class TradeVerdict(BaseModel):
agent: str = Field(..., description="Agent that produced the verdict")
signal: str = Field(..., regex="^(BUY|SELL|HOLD)$")
confidence: float = Field(..., ge=0, le=1)
Putting It Together
The orchestration layer ties all pieces together. It first classifies volatility, then decides whether to engage the LLM loop. The token bucket ensures we never exceed provider limits, and the deterministic fallback guarantees a verdict even if the debate stalls.
async def run_agents(market_summary: Dict[str, Any], bucket: AsyncTokenBucket) -> List[TradeVerdict]:
agents = ["Risk", "Technical", "Valuation"]
verdicts = []
for agent in agents:
try:
await bucket.acquire(amount=1) # 1 token per agent call
# Simulate an LLM call with a sleep; replace with real API call
await asyncio.sleep(0.5)
# Dummy logic: Risk prefers HOLD, Technical prefers BUY, Valuation prefers SELL
signal = {"Risk": "HOLD", "Technical": "BUY", "Valuation": "SELL"}[agent]
verdicts.append(TradeVerdict(agent=agent, signal=signal, confidence=0.8))
except RuntimeError as e:
logging.warning(f"{agent} could not acquire token: {e}")
continue
return verdicts
def aggregate_verdicts(verdicts: List[TradeVerdict]) -> str:
# Majority vote; if tie, fallback to deterministic rule
signals = [v.signal for v in verdicts]
if signals.count("BUY") > signals.count("SELL"):
return "BUY"
if signals.count("SELL") > signals.count("BUY"):
return "SELL"
return "HOLD"
async def main():
start = time.time()
market = load_market_summary()
floors = load_floor_sheet()
router = VolatilityRouter(floors)
high_vol = router.is_high_vol(market)
if not high_vol:
signal = deterministic_fallback(market)
verdict = {"agent": "Router", "signal": signal, "confidence": 1.0}
logging.info("Low volatility: using deterministic fallback")
else:
bucket = AsyncTokenBucket(capacity=10, refill_rate=5) # 5 tokens/sec
try:
verdicts = await run_agents(market, bucket)
signal = aggregate_verdicts(verdicts)
verdict = {"agent": "Consensus", "signal": signal, "confidence": 0.75}
logging.info("High volatility: LLM debate executed")
except Exception as e:
logging.error(f"Agent debate failed: {e}")
signal = deterministic_fallback(market)
verdict = {"agent": "Fallback", "signal": signal, "confidence": 0.5}
latency = time.time() - start
telemetry = {
"routing": "high" if high_vol else "low",
"tokens_used": 10 - bucket.tokens if high_vol else 0,
"latency_sec": round(latency, 2),
"verdict": verdict,
}
print(json.dumps(telemetry, indent=2))
if __name__ == "__main__":
asyncio.run(main())
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
# dynamic_prompt_router.py
import requests, json, pathlib, logging, time, datetime, asyncio
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from pydantic import BaseModel, ValidationError, Field
from typing import List, Dict, Any
# =======================
# Logging / Audit
# =======================
logging.basicConfig(
filename="nepse_agent_audit.log",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
# =======================
# Data Loading
# =======================
def load_market_summary() -> Dict[str, Any]:
try:
resp = requests.get(
"https://www.nepalstock.com.np/api/nots/market-summary",
timeout=5,
)
resp.raise_for_status()
return resp.json()
except Exception as e:
logging.error(f"Market summary fetch failed: {e}")
raise
def load_floor_sheet() -> List[Dict[str, Any]]:
try:
path = pathlib.Path("sample_floorsheet.json")
with path.open() as f:
return json.load(f)
except Exception as e:
logging.error(f"Floor sheet load failed: {e}")
raise
# =======================
# Semantic Router
# =======================
class VolatilityRouter:
def __init__(self, floor_sheets: List[Dict[str, Any]]):
self.vectorizer = TfidfVectorizer(stop_words="english")
high_vol_texts = [
fs["description"] for fs in floor_sheets if fs["volatility"] == "high"
]
self.high_vol_matrix = self.vectorizer.fit_transform(high_vol_texts)
def is_high_vol(self, market_summary: Dict[str, Any], threshold: float = 0.4) -> bool:
text = market_summary.get("summary", "")
vec = self.vectorizer.transform([text])
sim = cosine_similarity(vec, self.high_vol_matrix).mean()
return sim >= threshold
# =======================
# Async Token Bucket
# =======================
class AsyncTokenBucket:
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate # tokens per second
self.last_refill = time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self, amount: int = 1):
async with self.lock:
now = time.monotonic()
elapsed = now - self.last_refill
refill = int(elapsed * self.refill_rate)
if refill > 0:
self.tokens = min(self.capacity, self.tokens + refill)
self.last_refill = now
if self.tokens < amount:
raise RuntimeError("Token budget exhausted")
self.tokens -= amount
return amount
# =======================
# Deterministic Fallback
# =======================
def deterministic_fallback(market_summary: Dict[str, Any]) -> str:
pct_change = market_summary.get("pct_change", 0)
if pct_change >= 10:
return "SELL"
if pct_change <= -10:
return "BUY"
return "HOLD"
# =======================
# Pydantic Validation
# =======================
class TradeVerdict(BaseModel):
agent: str = Field(..., description="Agent that produced the verdict")
signal: str = Field(..., regex="^(BUY|SELL|HOLD)$")
confidence: float = Field(..., ge=0, le=1)
# =======================
# Agent Debate Simulation
# =======================
async def run_agents(
market_summary: Dict[str, Any], bucket: AsyncTokenBucket
) -> List[TradeVerdict]:
agents = ["Risk", "Technical", "Valuation"]
verdicts = []
for agent in agents:
try:
await bucket.acquire(amount=1) # 1 token per agent call
# Simulate an LLM call with a sleep; replace with real API call
await asyncio.sleep(0.5)
# Dummy logic: Risk prefers HOLD, Technical prefers BUY, Valuation prefers SELL
signal = {"Risk": "HOLD", "Technical": "BUY", "Valuation": "SELL"}[agent]
verdicts.append(TradeVerdict(agent=agent, signal=signal, confidence=0.8))
except RuntimeError as e:
logging.warning(f"{agent} could not acquire token: {e}")
continue
return verdicts
def aggregate_verdicts(verdicts: List[TradeVerdict]) -> str:
signals = [v.signal for v in verdicts]
if signals.count("BUY") > signals.count("SELL"):
return "BUY"
if signals.count("SELL") > signals.count("BUY"):
return "SELL"
return "HOLD"
# =======================
# Orchestrator
# =======================
async def main():
start = time.time()
market = load_market_summary()
floors = load_floor_sheet()
router = VolatilityRouter(floors)
high_vol = router.is_high_vol(market)
if not high_vol:
signal = deterministic_fallback(market)
verdict = {"agent": "Router", "signal": signal, "confidence": 1.0}
logging.info("Low volatility: using deterministic fallback")
else:
bucket = AsyncTokenBucket(capacity=10, refill_rate=5) # 5 tokens/sec
try:
verdicts = await run_agents(market, bucket)
signal = aggregate_verdicts(verdicts)
verdict = {"agent": "Consensus", "signal": signal, "confidence": 0.75}
logging.info("High volatility: LLM debate executed")
except Exception as e:
logging.error(f"Agent debate failed: {e}")
signal = deterministic_fallback(market)
verdict = {"agent": "Fallback", "signal": signal, "confidence": 0.5}
latency = time.time() - start
telemetry = {
"routing": "high" if high_vol else "low",
"tokens_used": 10 - bucket.tokens if high_vol else 0,
"latency_sec": round(latency, 2),
"verdict": verdict,
}
print(json.dumps(telemetry, indent=2))
if __name__ == "__main__":
asyncio.run(main())
Expected Output
When you run the script you should see something like:
{
"routing": "high",
"tokens_used": 3,
"latency_sec": 2.07,
"verdict": {
"agent": "Consensus",
"signal": "BUY",
"confidence": 0.75
}
}
The audit log (nepse_agent_audit.log) will contain timestamps for each routing decision and token acquisition.
Limitations and Tradeoffs
Semantic routing relies on the quality of the historical floor sheet embeddings. If a black‑swan event (e.g., a sudden policy change) produces a market summary that is semantically unlike any past high‑volatility example, the router may misclassify it as low volatility and skip the LLM loop. The token bucket is a simple in‑process implementation; in a distributed system you’d need a shared store (Redis, DynamoDB) to coordinate across workers. Finally, the deterministic fallback is intentionally conservative; it may under‑react to nuanced market signals that a multi‑agent LLM could capture.
Frequently Asked Questions
How do you handle NEPSE's frequent API outages and slow response times during peak trading hours?
We wrap the HTTP call in a timeout and catch exceptions. If the market summary fetch fails, the orchestrator falls back to the last cached snapshot and logs a warning. In production we would add exponential backoff with jitter and a circuit breaker that suppresses retries when the API is persistently down.
What is the memory footprint of keeping historical floor sheet embeddings in-memory alongside the agent execution state?
For the sample dataset (~2 k floor sheets) the TF‑IDF matrix consumes ~20 MB. Scaling to tens of thousands of entries pushes memory to a few hundred MBs, still acceptable for a single‑node service. For larger scale, persist the embeddings to disk or a vector store.
How does this routing pattern scale when expanding from 3 agents to a 10‑agent enterprise topology?
The token bucket naturally handles more agents by increasing capacity and adjusting refill rates. The router’s similarity threshold can be tuned per agent group. However, the latency grows linearly with the number of agents; you may need to shard the debate or parallelize calls across multiple worker processes.
What I'd Change
In a real‑world deployment I’d replace the in‑process token bucket with a Redis‑backed rate limiter so that multiple service instances share the same budget. I’d also switch from TF‑IDF to a lightweight sentence transformer (e.g., Sentence‑BERT) for semantic routing, which captures nuance better than bag‑of‑words. Finally, I’d add a metrics exporter (Prometheus) to surface token usage, latency, and routing decisions, enabling automated alerting for anomalous spikes.