By Sagar Shankaran, Founder of CallSphere
Build a comprehensive chat analytics system with conversation metrics collection, conversion tracking, satisfaction scoring, session analysis, and A/B testing frameworks to continuously improve your chat agents.
Key takeaways
Deploying a chat agent without analytics is like launching a website without any traffic tracking. You have no idea whether the agent is helping users, losing them, or frustrating them. Chat analytics gives you the data to answer three fundamental questions: Is the agent working? Where is it failing? What should we improve next?
This guide covers the complete analytics stack: what to track, how to track it, how to score conversations, and how to run experiments to drive improvement.
Every meaningful interaction in a chat session should emit a structured event. Design your event schema to be extensible:
flowchart LR
PR(["PR opened"])
UNIT["Unit tests"]
EVAL["Eval harness<br/>PromptFoo or Braintrust"]
GOLD[("Golden set<br/>200 tagged cases")]
JUDGE["LLM as judge<br/>plus regex graders"]
SCORE["Aggregate score<br/>and per slice"]
GATE{"Score regress<br/>more than 2 percent?"}
BLOCK(["Block merge"])
MERGE(["Merge to main"])
PR --> UNIT --> EVAL --> GOLD --> JUDGE --> SCORE --> GATE
GATE -->|Yes| BLOCK
GATE -->|No| MERGE
style EVAL fill:#4f46e5,stroke:#4338ca,color:#fff
style GATE fill:#f59e0b,stroke:#d97706,color:#1f2937
style BLOCK fill:#dc2626,stroke:#b91c1c,color:#fff
style MERGE fill:#059669,stroke:#047857,color:#fff
from pydantic import BaseModel
from datetime import datetime
from enum import Enum
class EventType(str, Enum):
SESSION_START = "session_start"
SESSION_END = "session_end"
MESSAGE_SENT = "message_sent"
MESSAGE_RECEIVED = "message_received"
TOOL_CALLED = "tool_called"
FALLBACK_TRIGGERED = "fallback_triggered"
ESCALATION_REQUESTED = "escalation_requested"
CONVERSION = "conversion"
FEEDBACK_SUBMITTED = "feedback_submitted"
BUTTON_CLICKED = "button_clicked"
FLOW_STARTED = "flow_started"
FLOW_COMPLETED = "flow_completed"
FLOW_ABANDONED = "flow_abandoned"
class ChatEvent(BaseModel):
event_id: str
session_id: str
user_id: str | None
event_type: EventType
properties: dict = {}
timestamp: datetime
channel: str
class EventCollector:
def __init__(self, db_pool):
self.db = db_pool
self.buffer: list[ChatEvent] = []
self.buffer_size = 50
async def track(self, event: ChatEvent):
self.buffer.append(event)
if len(self.buffer) >= self.buffer_size:
await self.flush()
async def flush(self):
if not self.buffer:
return
events = self.buffer.copy()
self.buffer.clear()
await self.db.executemany(
"""INSERT INTO chat_events (event_id, session_id, user_id,
event_type, properties, timestamp, channel)
VALUES ($1, $2, $3, $4, $5, $6, $7)""",
[(e.event_id, e.session_id, e.user_id, e.event_type.value,
json.dumps(e.properties), e.timestamp, e.channel)
for e in events],
)
Buffer events and flush in batches to avoid per-message database writes, which would add latency to every conversation turn.
Hear it before you finish reading
Talk to a live CallSphere AI voice agent in your browser — 60 seconds, no signup.
Track these metrics to understand agent performance at a glance:
from dataclasses import dataclass
@dataclass
class AgentMetrics:
total_sessions: int
avg_session_duration_seconds: float
avg_messages_per_session: float
resolution_rate: float # Sessions resolved without escalation
escalation_rate: float # Sessions requiring human handoff
fallback_rate: float # Messages triggering fallback
conversion_rate: float # Sessions achieving the goal
avg_first_response_ms: float # Time to first agent response
avg_satisfaction_score: float # From feedback, 1-5
async def calculate_metrics(db, start_date: str, end_date: str) -> AgentMetrics:
sessions = await db.fetch(
"""SELECT
COUNT(DISTINCT session_id) as total_sessions,
AVG(EXTRACT(EPOCH FROM (max_ts - min_ts))) as avg_duration,
AVG(message_count) as avg_messages
FROM (
SELECT session_id,
MIN(timestamp) as min_ts,
MAX(timestamp) as max_ts,
COUNT(*) FILTER (WHERE event_type = 'message_sent') as message_count
FROM chat_events
WHERE timestamp BETWEEN $1 AND $2
GROUP BY session_id
) sub""",
start_date, end_date,
)
rates = await db.fetch(
"""SELECT
COUNT(*) FILTER (WHERE event_type = 'escalation_requested')::float /
NULLIF(COUNT(DISTINCT session_id), 0) as escalation_rate,
COUNT(*) FILTER (WHERE event_type = 'fallback_triggered')::float /
NULLIF(COUNT(*) FILTER (WHERE event_type = 'message_sent'), 0) as fallback_rate,
COUNT(*) FILTER (WHERE event_type = 'conversion')::float /
NULLIF(COUNT(DISTINCT session_id), 0) as conversion_rate
FROM chat_events
WHERE timestamp BETWEEN $1 AND $2""",
start_date, end_date,
)
return AgentMetrics(
total_sessions=sessions[0]["total_sessions"],
avg_session_duration_seconds=sessions[0]["avg_duration"] or 0,
avg_messages_per_session=sessions[0]["avg_messages"] or 0,
resolution_rate=1.0 - (rates[0]["escalation_rate"] or 0),
escalation_rate=rates[0]["escalation_rate"] or 0,
fallback_rate=rates[0]["fallback_rate"] or 0,
conversion_rate=rates[0]["conversion_rate"] or 0,
avg_first_response_ms=0, # Calculated separately
avg_satisfaction_score=0, # From feedback events
)
Beyond aggregate metrics, score individual conversations to identify patterns in good and bad interactions:
async def score_conversation(session_id: str, events: list[ChatEvent]) -> dict:
scores = {
"resolution": 0,
"efficiency": 0,
"sentiment": 0,
"goal_completion": 0,
}
message_count = sum(1 for e in events if e.event_type == EventType.MESSAGE_SENT)
had_fallback = any(e.event_type == EventType.FALLBACK_TRIGGERED for e in events)
had_escalation = any(e.event_type == EventType.ESCALATION_REQUESTED for e in events)
had_conversion = any(e.event_type == EventType.CONVERSION for e in events)
had_feedback = [e for e in events if e.event_type == EventType.FEEDBACK_SUBMITTED]
# Resolution: was the issue handled without escalation?
scores["resolution"] = 0 if had_escalation else 100
# Efficiency: fewer messages for resolution = better
if message_count <= 4:
scores["efficiency"] = 100
elif message_count <= 8:
scores["efficiency"] = 75
elif message_count <= 15:
scores["efficiency"] = 50
else:
scores["efficiency"] = 25
# Goal completion
scores["goal_completion"] = 100 if had_conversion else 0
# Sentiment from user feedback
if had_feedback:
rating = had_feedback[-1].properties.get("rating", 3)
scores["sentiment"] = int((rating / 5) * 100)
overall = sum(scores.values()) / len(scores)
return {"session_id": session_id, "scores": scores, "overall": overall}
For goal-oriented agents like lead qualifiers, track the conversion funnel in TypeScript on the frontend to see where users drop off:
interface FunnelStep {
name: string;
sessionCount: number;
dropoffRate: number;
}
async function buildConversionFunnel(
startDate: string,
endDate: string,
): Promise<FunnelStep[]> {
const response = await fetch(
`/api/analytics/funnel?start=${startDate}&end=${endDate}`,
);
const data: FunnelStep[] = await response.json();
return data;
}
function FunnelChart({ steps }: { steps: FunnelStep[] }) {
const maxCount = steps[0]?.sessionCount || 1;
return (
<div className="funnel">
{steps.map((step, i) => (
<div key={step.name} className="funnel-step">
<div className="bar"
style={{ width: `${(step.sessionCount / maxCount) * 100}%` }}>
<span>{step.name}</span>
<span>{step.sessionCount} sessions</span>
</div>
{i < steps.length - 1 && (
<div className="dropoff">
{step.dropoffRate.toFixed(1)}% drop-off
</div>
)}
</div>
))}
</div>
);
}
Run controlled experiments to measure the impact of changes to prompts, flows, or response strategies:
import hashlib
class ABTestManager:
def __init__(self, db):
self.db = db
def assign_variant(self, session_id: str, test_name: str, variants: list[str]) -> str:
# Deterministic assignment based on session ID
hash_input = f"{test_name}:{session_id}"
hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
variant_index = hash_value % len(variants)
return variants[variant_index]
async def track_exposure(self, session_id: str, test_name: str, variant: str):
await self.db.execute(
"""INSERT INTO ab_test_exposures (session_id, test_name, variant, timestamp)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (session_id, test_name) DO NOTHING""",
session_id, test_name, variant,
)
async def get_results(self, test_name: str) -> dict:
rows = await self.db.fetch(
"""SELECT
e.variant,
COUNT(DISTINCT e.session_id) as sessions,
COUNT(DISTINCT c.session_id) as conversions,
COUNT(DISTINCT c.session_id)::float /
NULLIF(COUNT(DISTINCT e.session_id), 0) as conversion_rate
FROM ab_test_exposures e
LEFT JOIN chat_events c ON e.session_id = c.session_id
AND c.event_type = 'conversion'
WHERE e.test_name = $1
GROUP BY e.variant""",
test_name,
)
return {
"test_name": test_name,
"variants": [dict(r) for r in rows],
}
# Usage in agent initialization
ab = ABTestManager(db)
async def get_system_prompt(session_id: str) -> str:
variant = ab.assign_variant(session_id, "prompt_tone_v2", ["formal", "casual"])
await ab.track_exposure(session_id, "prompt_tone_v2", variant)
prompts = {
"formal": "You are a professional customer service agent. Maintain a formal, courteous tone.",
"casual": "You are a friendly customer service agent. Be warm, conversational, and approachable.",
}
return prompts[variant]
The deterministic hash ensures the same session always gets the same variant, even across reconnections. The LEFT JOIN in the results query ensures sessions without conversions are counted in the denominator.
Still reading? Stop comparing — try CallSphere live.
CallSphere ships complete AI voice agents per industry — 14 tools for healthcare, 10 agents for real estate, 4 specialists for salons. See how it actually handles a call before you book a demo.
Combine all metrics into a monitoring dashboard that updates daily:
from fastapi import APIRouter
router = APIRouter(prefix="/api/analytics")
@router.get("/dashboard")
async def get_dashboard(start: str, end: str):
metrics = await calculate_metrics(db, start, end)
funnel = await build_funnel(db, start, end)
top_fallbacks = await get_top_fallbacks(db, start, end, limit=10)
active_tests = await get_active_ab_tests(db)
return {
"metrics": metrics,
"funnel": funnel,
"top_fallback_topics": top_fallbacks,
"ab_tests": active_tests,
"period": {"start": start, "end": end},
}
It depends on the agent's purpose. For support agents, track resolution rate — the percentage of conversations resolved without human escalation. For sales agents, track conversion rate — the percentage of conversations that achieve the desired outcome (demo booked, email collected). For general knowledge agents, track satisfaction score from post-conversation feedback. Pick one north-star metric and optimize for it.
Ask at the end of the conversation, not during it. Use a simple one-click rating (thumbs up/down or 1-5 stars) rather than a text survey. Make it optional and dismissable. Only ask after conversations longer than 3 messages — single-question interactions do not warrant feedback requests. Aim for a 15-25% response rate; higher than that suggests your prompt is too aggressive.
Run until you have at least 100 conversions per variant for conversion-focused tests, or 500 sessions per variant for engagement metrics. Use a statistical significance calculator — aim for 95% confidence before declaring a winner. For chat agents, this typically takes 1-3 weeks depending on traffic volume. Do not peek at results daily and stop early; this inflates false positive rates.
#Analytics #Metrics #ABTesting #Conversion #ChatAgent #AgenticAI #LearnAI #AIEngineering

Written by
Sagar Shankaran· Founder, CallSphere
LinkedInSagar Shankaran is the founder of CallSphere, where he builds production AI voice and chat agents deployed across healthcare, hospitality, real estate, and home services. He writes about agentic AI, LLM engineering, and shipping voice agents that handle real calls in production.
See how AI voice agents work for your industry. Live demo available -- no signup required.
A data-led look at how small businesses from Lusaka to Kitwe are losing revenue to missed calls, and how CallSphere AI voice and chat agents answer every enquiry 24/7 in English, Bemba, and Nyanja.
Sentiment is not a single number per call - it is a curve. The shape (started positive, dropped at minute 4, recovered) tells you what your AI did wrong. Here is the per-utterance sentiment pipeline and the dashboards we ship by vertical.
How a Seattle SaaS team used the Vercel AI SDK 5 agent loop to build an in-product onboarding agent that converts trial users at measurably higher rates.
LangChain v1 + LangGraph v1 in JS, paired with Ollama, gives you a fully local chat agent with tools, memory, and structured output. No OpenAI key required.
A/B testing LLM features needs different metrics than traditional A/B. The 2026 patterns for sound LLM experimentation in production.
pg_duckdb embeds DuckDB inside Postgres so transactional and analytic queries share the same database. AI dashboards that took 90 sec on Postgres run in <1 sec via DuckDB — without leaving Postgres.
© 2026 CallSphere Inc. All rights reserved.
Made within San Francisco
Watch how CallSphere handles real customer calls, schedules appointments, and processes payments — live.
Try Live DemoBook a DemoCalculate Your ROI