How to Build an AI Operations Dashboard with LangChain

Table of Contents

• How to Build an AI Operations Dashboard with LangChain
  • Real-World Use Case and Architecture
  • Prerequisites and Environment Setup
  • Building the Metrics Collection Layer
• metrics_collector.py
• Global metrics buffer - shared across the application
  • Implementing the FastAPI Monitoring Endpoint
• api_server.py
• Allow Streamlit frontend to connect
• In-memory alert rules - use a database in production

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

When your machine learning models are running in production, you need more than just accuracy metrics. You need to know when your inference pipeline is degrading, when your vector database [3] queries are slowing down, and when your API costs are spiking. This tutorial walks through building a real-time AI operations dashboard using LangChain, FastAPI, and Streamlit that monitors model performance, tracks operational metrics, and alerts you to anomalies before they become incidents.

I've built similar systems for teams running LLM-based applications in production, and the patterns here come from actual deployment experience. This isn't a toy example—we'll handle rate limiting, memory management, and real-time streaming.

Real-World Use Case and Architecture

The system we're building solves a specific problem: how do you know your AI pipeline is healthy when you have multiple models, vector stores, and API endpoints running simultaneously? A typical production setup might include:

• An embedding [1] model (like text-embedding-3-small) generating vectors for document retrieval
• A vector database (like LanceDB) handling similarity searches
• An LLM (like GPT-4 or Claude [8]) generating responses
• Multiple API endpoints handling different request types

Each component can fail independently. The embedding service might slow down, the vector store might return stale results, or the LLM might start hallucinating more frequently. Our dashboard needs to capture all these signals in real time.

The architecture uses three layers:

1. Data Collection Layer: LangChain callbacks and custom middleware capture metrics from every component
2. Processing Layer: FastAPI endpoints aggregate and analyze the metrics
3. Visualization Layer: Streamlit dashboard displays real-time charts and alerts

Prerequisites and Environment Setup

You'll need Python 3.10 or later and the following packages:

pip install langchain==0.3.0 langchain-openai [7]==0.2.0 langchain-community==0.3.0
pip install fastapi==0.115.0 uvicorn==0.30.0 streamlit==1.38.0
pip install lancedb==0.11.0 pandas==2.2.0 plotly==5.24.0
pip install pydantic==2.9.0 python-dotenv==1.0.1

Create a .env file for your API keys:

OPENAI_API_KEY=sk-your-key-here
LANCEDB_URI=./data/lancedb

Important: Never commit API keys to version control. Use environment variables or a secrets manager in production.

Building the Metrics Collection Layer

The core of our system is a custom LangChain callback handler that captures timing, token usage, and error information from every chain execution. LangChain's callback system is designed for this exact use case—it hooks into every stage of execution without modifying your chain code.

# metrics_collector.py
import time
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
from collections import deque
from threading import Lock

from langchain.callbacks.base import BaseCallbackHandler
from pydantic import BaseModel, Field

class MetricPoint(BaseModel):
    """Single data point for a metric"""
    timestamp: float = Field(default_factory=time.time)
    value: float
    labels: Dict[str, str] = Field(default_factory=dict)

class MetricsBuffer:
    """Thread-safe buffer for metrics with automatic cleanup"""

    def __init__(self, max_size: int = 10000, ttl_seconds: int = 3600):
        self.max_size = max_size
        self.ttl_seconds = ttl_seconds
        self._buffer: Dict[str, deque] = {}
        self._lock = Lock()

    def add(self, metric_name: str, point: MetricPoint):
        """Add a metric point, cleaning old data if needed"""
        with self._lock:
            if metric_name not in self._buffer:
                self._buffer[metric_name] = deque(maxlen=self.max_size)

            self._buffer[metric_name].append(point)
            self._cleanup(metric_name)

    def _cleanup(self, metric_name: str):
        """Remove points older than TTL"""
        cutoff = time.time() - self.ttl_seconds
        while self._buffer[metric_name] and self._buffer[metric_name][0].timestamp < cutoff:
            self._buffer[metric_name].popleft()

    def get_metrics(self, metric_name: str, 
                    since: Optional[float] = None) -> List[MetricPoint]:
        """Get metrics since a timestamp"""
        with self._lock:
            if metric_name not in self._buffer:
                return []

            if since is None:
                return list(self._buffer[metric_name])

            return [p for p in self._buffer[metric_name] if p.timestamp >= since]

    def get_all_metric_names(self) -> List[str]:
        """Get names of all tracked metrics"""
        with self._lock:
            return list(self._buffer.keys())

# Global metrics buffer - shared across the application
metrics_buffer = MetricsBuffer(max_size=50000, ttl_seconds=7200)

class OpsCallbackHandler(BaseCallbackHandler):
    """
    LangChain callback handler that captures production metrics.
    Tracks latency, token usage, and errors for every chain/LLM call.
    """

    def __init__(self, component_name: str = "default"):
        self.component_name = component_name
        self._start_times: Dict[str, float] = {}

    def on_llm_start(self, serialized: Dict[str, Any], 
                     prompts: List[str], **kwargs) -> None:
        """Capture start time for LLM calls"""
        run_id = kwargs.get('run_id', str(id(prompts)))
        self._start_times[run_id] = time.time()

    def on_llm_end(self, response, **kwargs) -> None:
        """Capture LLM completion metrics"""
        run_id = kwargs.get('run_id', '')
        start_time = self._start_times.pop(run_id, None)

        if start_time is None:
            return

        latency = time.time() - start_time

        # Track latency
        metrics_buffer.add(
            "llm_latency_seconds",
            MetricPoint(
                value=latency,
                labels={"component": self.component_name, "model": response.llm_output.get("model_name", "unknown") if hasattr(response, 'llm_output') else "unknown"}
            )
        )

        # Track token usage if available
        if hasattr(response, 'llm_output') and response.llm_output:
            token_usage = response.llm_output.get('token_usage', {})
            if token_usage:
                total_tokens = token_usage.get('total_tokens', 0)
                metrics_buffer.add(
                    "llm_total_tokens",
                    MetricPoint(
                        value=total_tokens,
                        labels={"component": self.component_name}
                    )
                )

    def on_llm_error(self, error: Exception, **kwargs) -> None:
        """Capture LLM errors"""
        metrics_buffer.add(
            "llm_errors_total",
            MetricPoint(
                value=1,
                labels={
                    "component": self.component_name,
                    "error_type": type(error).__name__
                }
            )
        )

    def on_chain_start(self, serialized: Dict[str, Any], 
                       inputs: Dict[str, Any], **kwargs) -> None:
        """Capture chain execution start"""
        run_id = kwargs.get('run_id', str(id(inputs)))
        self._start_times[f"chain_{run_id}"] = time.time()

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs) -> None:
        """Capture chain completion metrics"""
        run_id = kwargs.get('run_id', '')
        start_time = self._start_times.pop(f"chain_{run_id}", None)

        if start_time is None:
            return

        latency = time.time() - start_time
        metrics_buffer.add(
            "chain_latency_seconds",
            MetricPoint(
                value=latency,
                labels={"component": self.component_name}
            )
        )

The MetricsBuffer uses a thread-safe deque with automatic TTL-based cleanup. This prevents memory leaks in long-running services—a common issue I've seen in production monitoring systems. The maxlen parameter on the deque also prevents unbounded growth if the cleanup thread fails.

Implementing the FastAPI Monitoring Endpoint

Now we need a FastAPI application that serves these metrics and provides alerting logic. The endpoint exposes both raw metrics and aggregated statistics.

# api_server.py
import asyncio
import time
from typing import Dict, List, Optional
from datetime import datetime, timedelta

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from metrics_collector import metrics_buffer, MetricPoint

app = FastAPI(title="AI Ops Dashboard API")

# Allow Streamlit frontend to connect
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in production
    allow_methods=["*"],
    allow_headers=["*"],
)

class MetricSummary(BaseModel):
    """Aggregated metric summary"""
    metric_name: str
    current_value: float
    avg_value: float
    p50_value: float
    p95_value: float
    p99_value: float
    count: int
    time_range_seconds: int

class AlertRule(BaseModel):
    """Configuration for alert thresholds"""
    metric_name: str
    warning_threshold: float
    critical_threshold: float
    window_seconds: int = 300  # 5 minute window

# In-memory alert rules - use a database in production
alert_rules: Dict[str, AlertRule] = {}

@app.get("/api/v1/metrics/{metric_name}")
async def get_metric(
    metric_name: str,
    since: Optional[float] = Query(None, description="Unix timestamp to filter from")
):
    """Get raw metric data points"""
    points = metrics_buffer.get_metrics(metric_name, since=since)
    return {
        "metric_name": metric_name,
        "points": [p.model_dump() for p in points],
        "count": len(points)
    }

@app.get("/api/v1/metrics/{metric_name}/summary")
async def get_metric_summary(
    metric_name: str,
    window_seconds: int = Query(300, description="Time window in seconds")
):
    """Get aggregated statistics for a metric"""
    since = time.time() - window_seconds
    points = metrics_buffer.get_metrics(metric_name, since=since)

    if not points:
        return MetricSummary(
            metric_name=metric_name,
            current_value=0.0,
            avg_value=0.0,
            p50_value=0.0,
            p95_value=0.0,
            p99_value=0.0,
            count=0,
            time_range_seconds=window_seconds
        )

    values = sorted([p.value for p in points])
    n = len(values)

    return MetricSummary(
        metric_name=metric_name,
        current_value=values[-1],
        avg_value=sum(values) / n,
        p50_value=values[int(n * 0.5)],
        p95_value=values[int(n * 0.95)],
        p99_value=values[int(n * 0.99)],
        count=n,
        time_range_seconds=window_seconds
    )

@app.get("/api/v1/metrics")
async def list_metrics():
    """List all tracked metric names"""
    return {"metrics": metrics_buffer.get_all_metric_names()}

@app.post("/api/v1/alerts/rules")
async def create_alert_rule(rule: AlertRule):
    """Create or update an alert rule"""
    alert_rules[rule.metric_name] = rule
    return {"status": "created", "rule": rule.model_dump()}

@app.get("/api/v1/alerts/check")
async def check_alerts():
    """Check all alert rules against current metrics"""
    alerts = []

    for metric_name, rule in alert_rules.items():
        since = time.time() - rule.window_seconds
        points = metrics_buffer.get_metrics(metric_name, since=since)

        if not points:
            continue

        # Calculate average over the window
        avg_value = sum(p.value for p in points) / len(points)

        if avg_value >= rule.critical_threshold:
            alerts.append({
                "metric": metric_name,
                "severity": "critical",
                "value": avg_value,
                "threshold": rule.critical_threshold,
                "timestamp": time.time()
            })
        elif avg_value >= rule.warning_threshold:
            alerts.append({
                "metric": metric_name,
                "severity": "warning",
                "value": avg_value,
                "threshold": rule.warning_threshold,
                "timestamp": time.time()
            })

    return {"alerts": alerts, "checked_at": time.time()}

@app.get("/api/v1/health")
async def health_check():
    """Basic health check endpoint"""
    return {
        "status": "healthy",
        "timestamp": time.time(),
        "metrics_count": len(metrics_buffer.get_all_metric_names())
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

The alerting system uses a sliding window approach—it checks the average value over the last N seconds rather than individual spikes. This prevents alert fatigue from transient anomalies. In production, you'd want to add hysteresis and cooldown periods to avoid flapping alerts.

Building the Streamlit Dashboard

The frontend displays real-time charts and alerts using Streamlit's built-in caching and auto-refresh capabilities.

# dashboard.py
import time
from datetime import datetime
from typing import Dict, List, Optional

import streamlit as st
import pandas as pd
import plotly.graph_objects as go
import plotly.express as px
import requests

# Configuration
API_BASE_URL = "http://localhost:8000/api/v1"
REFRESH_INTERVAL = 5  # seconds

st.set_page_config(
    page_title="AI Ops Dashboard",
    page_icon="📊",
    layout="wide",
    initial_sidebar_state="expanded"
)

st.title("AI Operations Dashboard")
st.markdown("Real-time monitoring for LLM pipelines")

# Sidebar controls
with st.sidebar:
    st.header("Controls")
    auto_refresh = st.checkbox("Auto-refresh", value=True)
    refresh_rate = st.slider("Refresh rate (seconds)", 1, 30, REFRESH_INTERVAL)

    st.header("Alert Configuration")
    with st.form("alert_form"):
        metric_name = st.text_input("Metric name", "llm_latency_seconds")
        warning_threshold = st.number_input("Warning threshold", value=2.0)
        critical_threshold = st.number_input("Critical threshold", value=5.0)
        window_seconds = st.number_input("Window (seconds)", value=300)

        if st.form_submit_button("Create Alert Rule"):
            response = requests.post(
                f"{API_BASE_URL}/alerts/rules",
                json={
                    "metric_name": metric_name,
                    "warning_threshold": warning_threshold,
                    "critical_threshold": critical_threshold,
                    "window_seconds": window_seconds
                }
            )
            if response.status_code == 200:
                st.success(f"Alert rule created for {metric_name}")
            else:
                st.error(f"Failed to create alert: {response.text}")

# Main dashboard layout
col1, col2 = st.columns(2)

def fetch_metric_data(metric_name: str, window_seconds: int = 300) -> Optional[pd.DataFrame]:
    """Fetch metric data from API and return as DataFrame"""
    try:
        response = requests.get(
            f"{API_BASE_URL}/metrics/{metric_name}",
            params={"since": time.time() - window_seconds}
        )
        if response.status_code != 200:
            return None

        data = response.json()
        if not data["points"]:
            return None

        df = pd.DataFrame(data["points"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="s")
        return df

    except requests.exceptions.ConnectionError:
        return None

def create_time_series_chart(df: pd.DataFrame, metric_name: str) -> go.Figure:
    """Create a Plotly time series chart"""
    fig = go.Figure()

    fig.add_trace(go.Scatter(
        x=df["timestamp"],
        y=df["value"],
        mode="lines+markers",
        name=metric_name,
        line=dict(width=2),
        marker=dict(size=4)
    ))

    fig.update_layout(
        title=f"{metric_name} Over Time",
        xaxis_title="Time",
        yaxis_title="Value",
        height=400,
        margin=dict(l=40, r=40, t=40, b=40)
    )

    return fig

# Main loop with auto-refresh
placeholder = st.empty()

while True:
    with placeholder.container():
        # Fetch metrics list
        try:
            metrics_response = requests.get(f"{API_BASE_URL}/metrics")
            if metrics_response.status_code == 200:
                metric_names = metrics_response.json()["metrics"]
            else:
                metric_names = []
        except requests.exceptions.ConnectionError:
            st.error("Cannot connect to API server. Make sure it's running on port 8000.")
            metric_names = []

        # Display metrics in columns
        cols = st.columns(3)
        for idx, metric_name in enumerate(metric_names[:6]):  # Show first 6 metrics
            with cols[idx % 3]:
                df = fetch_metric_data(metric_name, window_seconds=300)
                if df is not None and not df.empty:
                    fig = create_time_series_chart(df, metric_name)
                    st.plotly_chart(fig, use_container_width=True)

                    # Show summary stats
                    summary_response = requests.get(
                        f"{API_BASE_URL}/metrics/{metric_name}/summary",
                        params={"window_seconds": 300}
                    )
                    if summary_response.status_code == 200:
                        summary = summary_response.json()
                        st.metric(
                            label=f"Current {metric_name}",
                            value=f"{summary['current_value']:.3f}",
                            delta=f"{summary['avg_value']:.3f} avg"
                        )
                else:
                    st.info(f"No data for {metric_name}")

        # Alert section
        st.header("Active Alerts")
        try:
            alerts_response = requests.get(f"{API_BASE_URL}/alerts/check")
            if alerts_response.status_code == 200:
                alerts = alerts_response.json()["alerts"]
                if alerts:
                    for alert in alerts:
                        severity_color = "🔴" if alert["severity"] == "critical" else "🟡"
                        st.warning(
                            f"{severity_color} **{alert['metric']}**: "
                            f"Value {alert['value']:.3f} exceeds "
                            f"{alert['severity']} threshold of {alert['threshold']}"
                        )
                else:
                    st.success("No active alerts")
        except requests.exceptions.ConnectionError:
            st.error("Cannot check alerts - API unavailable")

        # Health check
        st.header("System Health")
        try:
            health_response = requests.get(f"{API_BASE_URL}/health")
            if health_response.status_code == 200:
                health = health_response.json()
                st.json(health)
        except requests.exceptions.ConnectionError:
            st.error("Health check failed")

    if not auto_refresh:
        break

    time.sleep(refresh_rate)

Pitfalls and Production Tips

After running this system in production, here are the issues you'll actually encounter:

1. Memory Management for Metrics Buffers

The deque-based buffer works well for moderate traffic, but under high load (1000+ requests/second), you'll hit memory limits. I've seen this happen when a sudden traffic spike fills the buffer faster than the cleanup can run. Solution: implement a separate cleanup thread that runs every 60 seconds regardless of traffic, and use psutil to monitor the process memory:

import psutil
import threading

def memory_cleanup_thread():
    while True:
        process = psutil.Process()
        memory_mb = process.memory_info().rss / 1024 / 1024
        if memory_mb > 500:  # 500MB threshold
            # Force cleanup of oldest 25% of data
            for metric_name in metrics_buffer.get_all_metric_names():
                with metrics_buffer._lock:
                    buffer = metrics_buffer._buffer[metric_name]
                    cutoff = len(buffer) // 4
                    for _ in range(cutoff):
                        buffer.popleft()
        time.sleep(60)

cleanup_thread = threading.Thread(target=memory_cleanup_thread, daemon=True)
cleanup_thread.start()

2. API Rate Limiting with OpenAI

Your callback handler will capture latency spikes that are actually caused by OpenAI rate limiting, not your code. Add a separate metric for rate limit errors:

def on_llm_error(self, error: Exception, **kwargs):
    error_name = type(error).__name__
    if "RateLimitError" in error_name:
        metrics_buffer.add(
            "rate_limit_hits",
            MetricPoint(value=1, labels={"component": self.component_name})
        )

3. Streamlit Performance with Large Datasets

Streamlit reruns the entire script on each interaction. With 50,000+ metric points, this becomes unusable. Use st.cache_data with a TTL to avoid refetching data on every rerun:

@st.cache_data(ttl=10)  # Cache for 10 seconds
def fetch_metric_data_cached(metric_name: str, window_seconds: int):
    return fetch_metric_data(metric_name, window_seconds)

4. Timezone Handling

All timestamps in the system use Unix timestamps (UTC). When displaying in Streamlit, convert to local time explicitly:

df["local_time"] = df["timestamp"].dt.tz_localize("UTC").dt.tz_convert("America/New_York")

What's Next

This dashboard gives you visibility into your AI pipeline's operational health, but it's just the starting point. Consider adding:

• Anomaly detection: Use statistical methods (like Z-score or moving averages) to automatically detect unusual patterns without manual thresholds
• Cost tracking: Integrate with OpenAI's usage API to track per-model costs and set budget alerts
• A/B testing infrastructure: Route traffic between different model versions and compare metrics side-by-side

The complete code is available on GitHub (replace with actual repo). For more on building production AI systems, check out our guides on monitoring LLM applications and optimizing vector database performance.

Remember: the best monitoring system is one that alerts you before your users notice a problem. Start with the metrics that matter most to your application, then iterate based on what you learn from production incidents.

---