Building Real-Time Analytics for Conversational AI

September 5, 2024·15 min

When you're running thousands of concurrent conversations, understanding what's happening in real-time isn't just nice to have—it's essential. Here's how we built analytics that process 3.2 million conversations per day with sub-second latency.

The Challenge

Traditional analytics work great for page views and button clicks. But conversations are different:

Conversations are:

Multi-turn - a single conversation spans dozens of messages
Stateful - context from message 1 affects message 20
Non-linear - users jump between topics
Outcome-oriented - success isn't "viewed page" but "achieved goal"
Real-time - need insights while conversation is happening, not hours later

What customers need to know:

Is my bot working right now?
Why did this conversation fail?
Which topics are users asking about?
Are response times acceptable?
What's my cost per conversation?
Are users satisfied?

And they need answers in seconds, not hours.

Architecture Overview


User Message
    ↓
Lambda: Process Message
    ↓  (logs event)
    ↓
Kinesis Data Stream
    ↓  (real-time processing)
    ↓
Lambda: Aggregate Metrics ──→ DynamoDB (hot data, last 24hrs)
    ↓
Kinesis Firehose ──→ S3 (cold data, historical)
    ↓
Athena (SQL queries on S3)

Hot path: Real-time metrics (last 24 hours) in DynamoDB

Cold path: Historical data in S3, queryable via Athena

Event Schema

Every conversation event follows a consistent schema:

json
{
  "eventId": "evt_abc123",
  "timestamp": "2024-09-05T14:23:11.342Z",
  "appId": "app_xyz789",
  "conversationId": "conv_123456",
  "userId": "user_987654",

  "type": "message_received|message_sent|tool_called|outcome_reached|error",

  "message": {
    "content": "What are your prices?",
    "tokens": 5,
    "model": "gpt-4",
    "latency_ms": 487
  },

  "tool": {
    "name": "search_knowledge_base",
    "duration_ms": 123,
    "success": true
  },

  "outcome": {
    "tier": "useful_outcome",
    "value": 0.99,
    "confidence": 0.95
  },

  "context": {
    "ip": "192.168.1.1",
    "userAgent": "Mozilla/5.0...",
    "referrer": "https://google.com",
    "location": "US-CA-SF"
  }
}

Real-Time Processing

Kinesis Data Stream

Events flow into Kinesis for real-time processing:

Configuration:

Shard count: Auto-scaling (start with 10, scale to 100+)
Retention: 24 hours
Enhanced fan-out: Enabled (multiple consumers)

Why Kinesis:

Handles 3M+ events/day easily
Auto-scales with load
Multiple consumers can read simultaneously
Exactly-once delivery semantics

Aggregation Lambda

Processes events in micro-batches (100 events every 5 seconds):

``javascript


exports.handler = async (event) => {
  const events = event.Records.map(r => JSON.parse(r.kinesis.data));


  // Group by app and 5-minute time window
  const aggregates = {};


  for (const evt of events) {

const key = ${evt.appId}:${getTimeWindow(evt.timestamp)};




    if (!aggregates[key]) {
      aggregates[key] = {
        appId: evt.appId,
        window: getTimeWindow(evt.timestamp),
        totalMessages: 0,
        totalConversations: new Set(),
        avgLatency: [],
        errors: 0,
        outcomes: {},
        tools: {}
      };
    }


    // Accumulate metrics
    aggregates[key].totalMessages++;
    aggregates[key].totalConversations.add(evt.conversationId);
    if (evt.message) {
      aggregates[key].avgLatency.push(evt.message.latency_ms);
    }
    if (evt.type === 'error') {
      aggregates[key].errors++;
    }
    if (evt.outcome) {
      aggregates[key].outcomes[evt.outcome.tier] =
        (aggregates[key].outcomes[evt.outcome.tier] || 0) + 1;
    }
    if (evt.tool) {
      aggregates[key].tools[evt.tool.name] =
        (aggregates[key].tools[evt.tool.name] || 0) + 1;
    }
  }


  // Write to DynamoDB
  for (const aggregate of Object.values(aggregates)) {
    await writeToDynamoDB({

pk: APP#${aggregate.appId},

sk: METRICS#${aggregate.window},


      totalMessages: aggregate.totalMessages,
      totalConversations: aggregate.totalConversations.size,
      avgLatency: average(aggregate.avgLatency),
      p95Latency: percentile(aggregate.avgLatency, 95),
      errorRate: aggregate.errors / aggregate.totalMessages,
      outcomes: aggregate.outcomes,
      tools: aggregate.tools,
      ttl: Date.now() / 1000 + 86400 // 24 hour TTL
    });
  }
};


Key decisions:
5-minute windows - balance between freshness and efficiency
TTL on hot data - auto-delete after 24 hours
Set for unique conversations - accurate conversation counts
P95 latency - better than average for understanding experience

Query API

Customers query metrics through our API:

javascript
// GET /api/analytics/metrics
// ?appId=xyz&start=2024-09-05T00:00:00Z&end=2024-09-05T23:59:59Z


exports.handler = async (event) => {
  const { appId, start, end } = event.queryStringParameters;


  // Query DynamoDB for recent data (last 24hrs)
  if (isRecent(start, end)) {
    const windows = getTimeWindows(start, end, '5m');
    const results = await batchGetDynamoDB(
      windows.map(w => ({

pk: APP#${appId},

sk: METRICS#${w}


      }))
    );
    return aggregateResults(results);
  }


  // Query Athena for historical data
  else {
    const queryId = await athena.startQueryExecution({

QueryString:

SELECT

date_trunc('hour', timestamp) as hour,

COUNT(*) as total_messages,

COUNT(DISTINCT conversation_id) as total_conversations,

AVG(latency_ms) as avg_latency,

APPROX_PERCENTILE(latency_ms, 0.95) as p95_latency,

SUM(CASE WHEN type='error' THEN 1 ELSE 0 END) as errors

FROM conversation_events

WHERE app_id = '${appId}'

AND timestamp BETWEEN '${start}' AND '${end}'

GROUP BY date_trunc('hour', timestamp)

ORDER BY hour

,


      ResultConfiguration: {
        OutputLocation: 's3://analytics-results/'
      }
    });


    return await waitForQueryResults(queryId);
  }
};


Dashboard Metrics

Overview Dashboard

What customers see when they open the dashboard:

Top-line metrics (last 24 hours):
Total conversations
Active users
Average response time
Error rate
Customer satisfaction score

Trends (sparklines showing last 7 days):
Conversations per day
Response time over time
Error rate over time

Current status:
Messages in last 5 minutes (updates every 30 seconds)
Active conversations right now
Current response time

Conversation Explorer

Drill down into individual conversations:

Filters:
Date range
Outcome tier
Response time (slow, normal, fast)
Error status
User satisfaction rating

For each conversation:
Full transcript
Timeline with latency for each turn
Tools called
Outcome reached
Cost breakdown
User feedback (if any)

Topic Analysis

What are users actually talking about?

Implementation:
Extract key phrases from messages using TF-IDF
Group similar topics using clustering
Track topic volume over time

What it shows:
Top 20 topics this week
Trending topics (sudden spikes)
Topic sentiment (positive/negative)
Unresolved topics (user asked but bot couldn't answer)

Cost Analytics

Critical for outcome-based pricing:

Metrics:
Cost per conversation by outcome tier
LLM API costs per conversation
Tool execution costs
Total spend vs. budget

Forecasting:
Projected month-end spend
Cost trend (increasing/decreasing)
Cost per customer acquisition
ROI by conversation type

Performance Optimizations

1. Pre-aggregation

Don't calculate metrics on every query. Pre-aggregate:

javascript
// Bad: Calculate on query
SELECT AVG(latency_ms) FROM events
WHERE app_id = 'xyz' AND timestamp > now() - interval '24 hours'


// Good: Read pre-aggregated
SELECT avg_latency FROM aggregates
WHERE pk = 'APP#xyz' AND sk BETWEEN 'METRICS#2024-09-04' AND 'METRICS#2024-09-05'


Result: Query time reduced from 2.3s to 45ms.

2. Caching

Cache expensive queries:

javascript

const cacheKey = metrics:${appId}:${start}:${end};


const cached = await redis.get(cacheKey);
if (cached) return JSON.parse(cached);


const results = await queryMetrics(appId, start, end);
await redis.setex(cacheKey, 300, JSON.stringify(results)); // 5 min cache
return results;


Result: 80% of queries served from cache.

3. Sampling for High-Volume Apps

For apps with >10,000 conversations/day:

javascript
// Sample 10% of events for analytics
if (Math.random() < 0.1) {
  await sendToKinesis(event);
}


But always track critical events (errors, outcomes).

Result: 90% reduction in processing costs for high-volume apps, negligible accuracy impact.

4. Incremental Updates

Dashboard updates every 30 seconds with only new data:

javascript
// Client tracks last update
const lastUpdate = localStorage.getItem('lastUpdate') || now() - 3600000;


// Server returns only new data
const newEvents = await queryEvents({
  appId,
  since: lastUpdate
});


// Client merges with existing data
mergeMetrics(existingMetrics, newEvents);


Result: Dashboard updates in <200ms instead of reloading everything.

Alerting

Automated alerts based on metrics:

Alert Types

1. Error rate spike
javascript
if (errorRate > 5% && previousErrorRate < 2%) {
  alert({
    severity: 'high',
    message: 'Error rate spiked from 1.2% to 6.7%',
    appId: 'xyz',
    actions: ['View errors', 'Disable app', 'Call on-call engineer']
  });
}


2. Slow responses
javascript
if (p95Latency > 5000 && previousP95 < 2000) {
  alert({
    severity: 'medium',
    message: 'Response time p95 increased from 1.8s to 5.2s'
  });
}


3. Cost spike
javascript
if (todaySpend > yesterdaySpend * 2) {
  alert({
    severity: 'medium',
    message: 'Spending is 2.3x yesterday ($127 vs $55)'
  });
}


4. No traffic
javascript
if (messagesInLast10Min === 0 && usualTraffic > 100) {
  alert({
    severity: 'high',
    message: 'No messages received in 10 minutes (usually 120/10min)'
  });
}


Delivery

Email - digest once per day
Slack - critical alerts immediately
Dashboard - all alerts shown in app
SMS - only for severity='critical'

Privacy and Compliance

Analytics on conversations require careful handling:

PII Redaction

Before logging:
javascript
function redactPII(message) {
  return message
    .replace(/\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi, '[EMAIL]')
    .replace(/\b\d{3}-\d{2}-\d{4}\b/g, '[SSN]')
    .replace(/\b\d{16}\b/g, '[CARD]')
    .replace(/\b\d{3}-\d{3}-\d{4}\b/g, '[PHONE]');
}

Data Retention

Hot data (DynamoDB): 24 hours, then auto-deleted
Warm data (S3): 90 days, then archived to Glacier
Cold data (Glacier): 7 years for compliance

Customer Controls

Customers can:

Disable analytics entirely
Exclude specific fields (e.g., message content)
Export all their data
Request deletion

Current Scale

As of September 2024:

3.2M events/day processed in real-time
15,000 applications tracked
<1 second from event to dashboard
99.98% delivery success (event → analytics)
<200ms dashboard load time

What's Next

We're working on:

Predictive analytics - forecast conversation volume and costs
Anomaly detection - automatically flag unusual patterns
Conversation replay - step through conversations like a debugger
A/B testing - compare different prompts/configurations
Custom dashboards - customers build their own views

For Developers

Want to build similar analytics?

Key lessons:

1. Design events first - good schema makes everything easier

2. Pre-aggregate everything - don't calculate on query

3. Separate hot and cold paths - different tools for different time ranges

4. Cache aggressively - most queries are repeated

5. Sample when needed - 100% accuracy isn't always necessary

Tools we recommend:

Kinesis for real-time streaming
DynamoDB for hot data
S3 + Athena for cold data
CloudWatch for system metrics
Grafana for custom dashboards

Read our docs → | See API reference →

---

*Questions about our analytics architecture? Email hello@ilq.ai*