LlamaIndex pipelines fail in ways that look like success. A retriever that returns no relevant chunks still returns a response — the LLM just answers from its parametric memory and sounds confident doing it. An index build that silently skips half your documents still reports completion. A query pipeline with a broken reranker still returns results, just worse ones.
Traditional uptime monitoring doesn't catch this. A ping to your FastAPI endpoint that wraps a LlamaIndex query engine returns 200 regardless of whether the underlying retrieval actually worked. You need application-level monitoring: tracking query latency, retrieval scores, index freshness, and LLM error rates.
This tutorial shows how to add Vigilmon monitoring to a LlamaIndex-powered application — covering API availability, heartbeat monitors for index builds, and custom metrics reporting for query pipeline observability.
What You'll Build
- HTTP monitors for LlamaIndex query API endpoints
- Heartbeat monitors for index build jobs
- Custom metrics reporting for retrieval quality signals
- Alert routing for pipeline degradation versus full outages
Prerequisites
- A Vigilmon account at vigilmon.online
- A Vigilmon API key (Settings → API Keys)
- A working LlamaIndex application (Python, FastAPI or similar)
pip install llama-index requests
Step 1: Add a Health Endpoint to Your Query API
LlamaIndex is typically wrapped in a web API. The first monitoring layer is a standard HTTP health check that Vigilmon can poll.
# app.py
from fastapi import FastAPI
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.core import load_index_from_storage
import time
import os
app = FastAPI()
# Load index at startup
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)
query_engine = index.as_query_engine()
_startup_time = time.time()
@app.get("/health")
async def health():
"""Health check for Vigilmon HTTP monitor."""
return {
"status": "ok",
"uptime_seconds": int(time.time() - _startup_time),
"index_loaded": index is not None,
}
@app.get("/query")
async def query(q: str):
response = query_engine.query(q)
return {"answer": str(response)}
In Vigilmon, create an HTTP monitor:
- Name: LlamaIndex Query API
- URL:
https://your-app.example.com/health - Interval: 2 minutes
- Expected status: 200
- Alert: Slack or PagerDuty depending on production criticality
This catches the API process dying, a bad deploy, or the index failing to load at startup.
Step 2: Add a Deep Health Check for Query Pipeline Validity
A basic /health endpoint only confirms the process is running. A deeper check executes a known test query and validates the response meets expectations.
@app.get("/health/deep")
async def health_deep():
"""
Execute a known test query and validate retrieval works.
Vigilmon polls this at a longer interval (e.g., 10 minutes).
"""
TEST_QUERY = "What is this system about?"
MIN_RESPONSE_LENGTH = 50 # Responses shorter than this suggest retrieval failure
start = time.time()
try:
response = query_engine.query(TEST_QUERY)
duration_ms = int((time.time() - start) * 1000)
response_text = str(response)
if len(response_text) < MIN_RESPONSE_LENGTH:
return {
"status": "degraded",
"reason": "Response too short — retrieval may have failed",
"response_length": len(response_text),
"duration_ms": duration_ms,
}, 503
return {
"status": "ok",
"duration_ms": duration_ms,
"response_length": len(response_text),
}
except Exception as e:
return {
"status": "error",
"reason": str(e),
"duration_ms": int((time.time() - start) * 1000),
}, 500
Add a second Vigilmon HTTP monitor for this endpoint:
- Name: LlamaIndex Query Pipeline - Deep Check
- URL:
https://your-app.example.com/health/deep - Interval: 10 minutes
- Expected status: 200
- Alert: Slack
#ai-opschannel
This catches retrieval failures that the surface-level health check misses.
Step 3: Heartbeat Monitors for Index Build Jobs
Index builds are usually scheduled jobs — nightly full rebuilds, incremental updates after document ingestion, or triggered by data pipeline events. A heartbeat monitor confirms these jobs complete on schedule.
In Vigilmon, create a Heartbeat Monitor for each index build job:
- Name: LlamaIndex Index Build - nightly
- Expected interval: 24 hours
- Grace period: 2 hours
Then ping the heartbeat at the end of your build job:
# scripts/build_index.py
import os
import requests
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
VIGILMON_HEARTBEAT_URL = os.environ["VIGILMON_INDEX_HEARTBEAT_URL"]
def build_index(docs_path: str, storage_path: str):
print(f"Loading documents from {docs_path}")
documents = SimpleDirectoryReader(docs_path).load_data()
print(f"Loaded {len(documents)} documents")
index = VectorStoreIndex.from_documents(documents)
index.storage_context.persist(persist_dir=storage_path)
print(f"Index built and saved to {storage_path}")
return len(documents)
def main():
try:
doc_count = build_index("./data", "./storage")
# Ping heartbeat on success
resp = requests.post(
VIGILMON_HEARTBEAT_URL,
json={
"status": "ok",
"message": f"Index built: {doc_count} documents",
"metadata": {"doc_count": doc_count},
},
timeout=10,
)
print(f"Vigilmon heartbeat: {resp.status_code}")
except Exception as e:
# Report failure to Vigilmon via webhook (separate from heartbeat)
webhook_url = os.environ.get("VIGILMON_WEBHOOK_URL")
if webhook_url:
requests.post(
webhook_url,
json={
"status": "down",
"message": f"Index build failed: {str(e)}",
},
timeout=10,
)
raise
if __name__ == "__main__":
main()
Do not ping the heartbeat on failure — the missing ping is what triggers the Vigilmon alert. Only ping on successful completion.
Step 4: Instrument Query Metrics and Report to Vigilmon
LlamaIndex provides instrumentation hooks. Use them to collect per-query metrics and periodically report aggregates to Vigilmon's webhook.
# monitoring.py
import time
import threading
import requests
import os
from collections import deque
from dataclasses import dataclass, field
from typing import List
VIGILMON_WEBHOOK_URL = os.environ.get("VIGILMON_WEBHOOK_URL")
@dataclass
class QueryMetrics:
duration_ms: float
retrieved_nodes: int
error: bool = False
source_count: int = 0
class MetricsCollector:
def __init__(self, window_size: int = 100):
self._lock = threading.Lock()
self._recent: deque = deque(maxlen=window_size)
def record(self, metrics: QueryMetrics):
with self._lock:
self._recent.append(metrics)
def summarize(self) -> dict:
with self._lock:
if not self._recent:
return {"count": 0}
durations = [m.duration_ms for m in self._recent]
errors = sum(1 for m in self._recent if m.error)
node_counts = [m.retrieved_nodes for m in self._recent if not m.error]
return {
"count": len(self._recent),
"error_rate": errors / len(self._recent),
"p50_latency_ms": sorted(durations)[len(durations) // 2],
"p95_latency_ms": sorted(durations)[int(len(durations) * 0.95)],
"avg_retrieved_nodes": sum(node_counts) / len(node_counts) if node_counts else 0,
}
collector = MetricsCollector()
def report_to_vigilmon():
"""Background thread — reports metrics to Vigilmon every 5 minutes."""
while True:
time.sleep(300)
if not VIGILMON_WEBHOOK_URL:
continue
summary = collector.summarize()
if summary["count"] == 0:
continue
status = "healthy"
if summary["error_rate"] > 0.1:
status = "degraded"
if summary["error_rate"] > 0.5:
status = "down"
try:
requests.post(
VIGILMON_WEBHOOK_URL,
json={
"status": status,
"message": (
f"LlamaIndex: {summary['count']} queries, "
f"p95={summary['p95_latency_ms']:.0f}ms, "
f"error_rate={summary['error_rate']:.1%}"
),
"metadata": summary,
},
timeout=10,
)
except Exception:
pass # Don't let monitoring failure affect the query pipeline
# Start the background reporter
reporter_thread = threading.Thread(target=report_to_vigilmon, daemon=True)
reporter_thread.start()
Instrument your query engine calls to use the collector:
async def query(q: str):
start = time.time()
try:
response = query_engine.query(q)
nodes = response.source_nodes if hasattr(response, 'source_nodes') else []
collector.record(QueryMetrics(
duration_ms=(time.time() - start) * 1000,
retrieved_nodes=len(nodes),
error=False,
source_count=len(nodes),
))
return {"answer": str(response), "sources": len(nodes)}
except Exception as e:
collector.record(QueryMetrics(
duration_ms=(time.time() - start) * 1000,
retrieved_nodes=0,
error=True,
))
raise
Step 5: Monitor Index Freshness
Stale indexes are a common silent failure mode. Documents added to your data source don't appear in query results because the index hasn't been rebuilt.
# Add to your health endpoint
import os
import time
INDEX_STORAGE_PATH = "./storage"
@app.get("/health/index")
async def health_index():
"""Check index freshness — alert if index is older than expected."""
MAX_AGE_HOURS = 26 # Allow some buffer beyond 24h rebuild cycle
try:
# Check the modification time of the index metadata file
metadata_path = os.path.join(INDEX_STORAGE_PATH, "docstore.json")
if not os.path.exists(metadata_path):
return {"status": "down", "reason": "Index storage not found"}, 503
mtime = os.path.getmtime(metadata_path)
age_hours = (time.time() - mtime) / 3600
if age_hours > MAX_AGE_HOURS:
return {
"status": "degraded",
"reason": f"Index is {age_hours:.1f}h old (max {MAX_AGE_HOURS}h)",
"index_age_hours": round(age_hours, 1),
}, 503
return {
"status": "ok",
"index_age_hours": round(age_hours, 1),
}
except Exception as e:
return {"status": "error", "reason": str(e)}, 500
Create a Vigilmon HTTP monitor for this endpoint:
- Name: LlamaIndex Index Freshness
- URL:
https://your-app.example.com/health/index - Interval: 30 minutes
- Expected status: 200
This alerts you before users notice that their questions are returning outdated information.
Step 6: Configure Alert Channels by Severity
Different LlamaIndex failures have different urgency levels:
| Failure type | Severity | Recommended channel |
|---|---|---|
| Query API down (process crash) | Critical | PagerDuty |
| Deep health check fails (retrieval broken) | High | PagerDuty or Slack #incidents |
| Index build missed (heartbeat) | High | Slack #ai-ops |
| Index stale (>26h old) | Medium | Slack #ai-ops |
| Query error rate > 10% | Medium | Slack #ai-ops |
| p95 latency > 5s | Low | Slack #engineering |
In Vigilmon, go to Alerts → Channels and create separate channels for each destination. Then assign each monitor to the appropriate channel.
Monitoring Coverage Summary
| What to monitor | Monitor type | Interval | Alert on | |---|---|---|---| | Query API health (process up) | HTTP | 2 min | Non-200 | | Deep query pipeline health | HTTP | 10 min | Non-200 or slow | | Index freshness | HTTP | 30 min | Age > threshold | | Index build completion | Heartbeat | 24h | Missed ping | | Query error rate | Webhook (metrics) | 5 min | Error rate > 10% | | Query p95 latency | Webhook (metrics) | 5 min | Latency > threshold |
LlamaIndex pipelines fail in subtle ways — stale indexes, empty retrievals, silent LLM errors — that traditional uptime checks completely miss. The monitoring setup in this guide gives you visibility at every layer: process health, pipeline health, index freshness, and query quality signals. When something degrades, Vigilmon tells you before your users do.
Start monitoring your AI pipeline free at vigilmon.online
Tags: #llamaindex #llm #rag #ai #monitoring