Skip to content

avinaash-sharma/pubsub-realtime-pipeline

BranchesTags

Repository files navigation

(Made GPT to write me the ReadmeFile for the POC - reviewed )

🎯 Realtime Call Intelligence

A production-inspired real-time contact center AI pipeline built on Google Cloud Platform, simulating the core middleware layer of an enterprise contact center architecture.

POC Focus: Layer 3 — Cloud Middleware & Event Pipeline Simulates what happens after a phone call's audio hits the SIP Gateway — transcript streaming, sentiment analysis, real-time storage, and live dashboard updates.

📌 Table of Contents

🎯 What We Wanted to Build

The goal was to understand and implement the Cloud Middleware layer of a real enterprise contact center AI system — specifically the event-driven pipeline that processes live call data.

The reference architecture (enterprise contact center on Google Cloud) processes calls like this:

Phone Call
    → SBC (assigns UCID per caller)
    → SIP Gateway (GCP)
    → Google STT (speech to text)
    → Agent Assist (AI suggestions)
    → Cloud Pub/Sub (event bus)
    → Cloud Run (custom processing)
    → Memorystore + BigQuery (storage)
    → Agent Desktop (Salesforce UI)

We wanted to understand and simulate this pipeline — specifically everything after the SIP Gateway — without needing real telephony hardware.

🏗️ What We Actually Built

A fully working simulation of the middleware layer with:

mock_call.py          →    GCP Pub/Sub          →    processor/main.py
(fake STT output)          (3 topics/subs)            (subscriber + WS server)
                                                              │
                                                   ┌──────────┴──────────┐
                                                   ▼                     ▼
                                              Supabase               WebSocket
                                           (PostgreSQL)             (broadcast)
                                                   │                     │
                                                   └──────────┬──────────┘
                                                              ▼
                                                       dashboard/
                                                  (live UI + call history)

Components Built

Component File What it does
Config loader config.py Loads all .env variables centrally
Data models models.py Pydantic schemas for all event types
DB client database/client.py Async Supabase connection (singleton)
DB writer database/writer.py Async functions to save each event type
Mock publisher publisher/mock_call.py Simulates a full phone call with STT output
Processor processor/main.py Pub/Sub subscriber + FastAPI + WebSocket server
Dashboard dashboard/ Live UI showing transcript, sentiment, call history

✅ What We Achieved

Technical Achievements

  • ✅ Full event-driven pipeline — publisher → Pub/Sub → processor → DB → dashboard
  • ✅ 3 GCP Pub/Sub topics with default subscriptions (call-events, transcript-events, sentiment-events)
  • ✅ Async Python architecture — built async from day one for DB swappability
  • ✅ Decoupled layers — publisher and processor only communicate via Pub/Sub topics, never directly
  • ✅ Real-time dashboard — WebSocket broadcasts every event to connected browser clients instantly
  • ✅ Persistent storage — all calls, transcripts, sentiments stored in Supabase PostgreSQL
  • ✅ Call history — dashboard fetches past calls from Supabase REST API
  • ✅ Config endpoint — dashboard fetches credentials from FastAPI, never hardcoded
  • ✅ Message durability — unprocessed Pub/Sub messages survive processor restarts (up to 7 days)
  • ✅ Foreign key relationships — transcripts and sentiments are always linked to their parent call via ucid
  • ✅ Realtime enabled on all 3 Supabase tables

Conceptual Achievements

  • ✅ Understood GCP project setup — billing, APIs, IAM, service accounts
  • ✅ Understood Pub/Sub — topics, subscriptions, publish/consume, ack/nack
  • ✅ Understood async Python — event loops, threading, run_coroutine_threadsafe
  • ✅ Understood WebSocket — connection manager, broadcast pattern
  • ✅ Understood Repository pattern — DB layer is swappable without touching pipeline code
  • ✅ Understood why decoupling matters — each layer only knows about topics, not other layers

🏛️ Architecture

Data Flow

┌─────────────────────────────────────────────────────────────────┐
│                        YOUR MACHINE                             │
│                                                                 │
│   publisher/mock_call.py                                        │
│   ┌─────────────────────┐                                       │
│   │ simulate_call()     │                                       │
│   │ - generate UCID     │                                       │
│   │ - fake conversation │                                       │
│   │ - analyse sentiment │                                       │
│   └──────────┬──────────┘                                       │
│              │ publish(bytes)                                    │
└──────────────┼──────────────────────────────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────────┐
│                      GOOGLE CLOUD ☁️                            │
│                                                                 │
│  ┌─────────────────┐  ┌──────────────────┐  ┌───────────────┐  │
│  │  call-events    │  │transcript-events │  │sentiment-     │  │
│  │  topic          │  │topic             │  │events topic   │  │
│  │  └─ call-events │  │└─transcript-     │  │└─sentiment-   │  │
│  │     -sub        │  │  events-sub      │  │  events-sub   │  │
│  └────────┬────────┘  └────────┬─────────┘  └──────┬────────┘  │
│           │                    │                    │           │
└───────────┼────────────────────┼────────────────────┼───────────┘
            │                    │                    │
            └────────────────────┼────────────────────┘
                                 │ subscribe (streaming pull)
                                 ▼
┌─────────────────────────────────────────────────────────────────┐
│                        YOUR MACHINE                             │
│                                                                 │
│   processor/main.py                                             │
│   ┌──────────────────────────────────────────┐                  │
│   │ Thread 1 (main): FastAPI + uvicorn        │                  │
│   │   GET  /health                            │                  │
│   │   GET  /config                            │                  │
│   │   WS   /ws  ←── dashboard connects here  │                  │
│   │                                           │                  │
│   │ Thread 2: async event loop                │                  │
│   │   save_call()       → Supabase            │                  │
│   │   save_transcript() → Supabase            │                  │
│   │   save_sentiment()  → Supabase            │                  │
│   │   broadcast()       → all WS clients      │                  │
│   │                                           │                  │
│   │ Thread 3: Pub/Sub subscribers             │                  │
│   │   handle_call_event()                     │                  │
│   │   handle_transcript_event()               │                  │
│   │   handle_sentiment_event()                │                  │
│   └──────────────────────────────────────────┘                  │
│              │                      │                           │
│              ▼                      ▼                           │
└─────────────────────────────────────────────────────────────────┘
               │                      │
               ▼                      ▼
┌──────────────────────┐   ┌──────────────────────────────────────┐
│   SUPABASE ☁️        │   │         YOUR BROWSER                 │
│                      │   │                                      │
│  calls table         │   │  dashboard/index.html                │
│  transcripts table   │   │  ┌────────────────────────────────┐  │
│  sentiments table    │   │  │ 📞 Active Call Info            │  │
│                      │   │  │ 💬 Live Transcript Feed        │  │
│  realtime enabled ✅  │   │  │ 🎭 Sentiment Indicator        │  │
└──────────────────────┘   │  │ 📋 Call History Table         │  │
                           │  └────────────────────────────────┘  │
                           └──────────────────────────────────────┘

Database Schema

calls
├── id          UUID (primary key, auto-generated)
├── ucid        TEXT UNIQUE (business key, e.g. "CALL-A1B2C3D4")
├── agent_id    TEXT
├── caller_id   TEXT
├── started_at  TIMESTAMPTZ
├── ended_at    TIMESTAMPTZ (null until call ends)
└── status      TEXT ("active" | "completed")

transcripts
├── id          UUID (primary key)
├── ucid        TEXT → references calls(ucid)
├── speaker     TEXT ("agent" | "caller")
├── text        TEXT
├── confidence  FLOAT (0.0 - 1.0)
└── created_at  TIMESTAMPTZ

sentiments
├── id          UUID (primary key)
├── ucid        TEXT → references calls(ucid)
├── sentiment   TEXT ("positive" | "neutral" | "negative")
├── score       FLOAT (-1.0 to 1.0)
└── created_at  TIMESTAMPTZ

🛠️ Tech Stack

Layer Technology Why
Event Bus Google Cloud Pub/Sub Decoupled, durable, auto-scales
Backend Python 3.11 + FastAPI Async-first, clean API, WebSocket support
Web Server Uvicorn ASGI server for FastAPI
Data Validation Pydantic v2 Runtime type validation (like TypeScript at runtime)
Database Supabase (PostgreSQL) Free tier, built-in realtime, REST API
GCP Auth Service Account + JSON key Credentials for GCP APIs
Frontend Vanilla JS + CSS No framework overhead for a POC

📁 Project Structure

realtime-call-intelligence/
│
├── .env                        # Environment variables (never commit!)
├── .gitignore                  # Ignores credentials.json, .env, venv/
├── credentials.json            # GCP service account key (never commit!)
├── requirements.txt            # Python dependencies
│
├── config.py                   # Centralised env loader (like constants.ts)
├── models.py                   # Pydantic event schemas (like TypeScript interfaces)
│
├── publisher/
│   ├── __init__.py
│   └── mock_call.py            # Simulates STT + sentiment events
│
├── processor/
│   ├── __init__.py
│   └── main.py                 # Pub/Sub subscriber + FastAPI + WebSocket server
│
├── database/
│   ├── __init__.py
│   ├── client.py               # Async Supabase singleton client
│   └── writer.py               # Async DB write functions
│
└── dashboard/
    ├── index.html              # Dashboard UI
    ├── style.css               # Dark theme styling
    └── app.js                  # WebSocket client + Supabase REST calls

⚙️ Setup Guide

Prerequisites

Tool Version Install
Python 3.11 brew install python@3.11
gcloud CLI latest cloud.google.com/sdk
Git any pre-installed on Mac

Step 1 — Google Cloud Setup

1a. Create GCP Project

  1. Go to console.cloud.google.com
  2. Create new project → name it contact-center-poc
  3. Note the Project ID (e.g. contact-center-poc-494117)

1b. Link Billing

gcloud billing projects link YOUR_PROJECT_ID \
  --billing-account=YOUR_BILLING_ACCOUNT_ID

1c. Enable APIs

gcloud services enable \
  pubsub.googleapis.com \
  run.googleapis.com \
  bigquery.googleapis.com \
  storage.googleapis.com \
  secretmanager.googleapis.com \
  --project=YOUR_PROJECT_ID

Or via UI: console.cloud.google.com/apis/library Search and enable: Cloud Pub/Sub, Cloud Run, BigQuery, Cloud Storage, Secret Manager

1d. Create Service Account

Via UI: console.cloud.google.com/iam-admin/serviceaccounts

  1. Click + Create Service Account
  2. Name: contact-center-sa
  3. Assign roles: Pub/Sub Admin, BigQuery Admin, Cloud Run Admin
  4. Go to Keys tab → Add Key → JSON → download
  5. Rename downloaded file to credentials.json
  6. Place in project root

1e. Create Pub/Sub Topics

Via UI: console.cloud.google.com/cloudpubsub/topic/list

Create these 3 topics (check "Add a default subscription" for each):

Topic ID Subscription (auto-created)
call-events call-events-sub
transcript-events transcript-events-sub
sentiment-events sentiment-events-sub

Step 2 — Supabase Setup

  1. Go to supabase.com → create new project

  2. Name: contact-center-poc, Region: South Asia (Mumbai)

  3. Go to Settings → API and note down:

    • Project URL
    • anon public key
    • service_role secret key

  4. Go to SQL Editor → run this:

CREATE TABLE calls (
  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
  ucid TEXT NOT NULL UNIQUE,
  agent_id TEXT NOT NULL,
  caller_id TEXT NOT NULL,
  started_at TIMESTAMPTZ DEFAULT NOW(),
  ended_at TIMESTAMPTZ,
  status TEXT DEFAULT 'active'
);

CREATE TABLE transcripts (
  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
  ucid TEXT NOT NULL REFERENCES calls(ucid),
  speaker TEXT NOT NULL,
  text TEXT NOT NULL,
  confidence FLOAT DEFAULT 1.0,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE sentiments (
  id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
  ucid TEXT NOT NULL REFERENCES calls(ucid),
  sentiment TEXT NOT NULL,
  score FLOAT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

ALTER PUBLICATION supabase_realtime ADD TABLE calls;
ALTER PUBLICATION supabase_realtime ADD TABLE transcripts;
ALTER PUBLICATION supabase_realtime ADD TABLE sentiments;

Step 3 — Local Setup

# Clone repo
git clone https://github.com/yourusername/realtime-call-intelligence
cd realtime-call-intelligence

# Create virtual environment with Python 3.11
python3.11 -m venv venv
source venv/bin/activate  # Mac/Linux
# venv\Scripts\activate   # Windows

# Install dependencies
pip install -r requirements.txt

Step 4 — Environment Variables

Create .env in the project root:

# ── Google Cloud ──────────────────────────────
GOOGLE_APPLICATION_CREDENTIALS=./credentials.json
GCP_PROJECT_ID=your-project-id

# ── Pub/Sub Topics ────────────────────────────
PUBSUB_CALL_TOPIC=call-events
PUBSUB_TRANSCRIPT_TOPIC=transcript-events
PUBSUB_SENTIMENT_TOPIC=sentiment-events

# ── Supabase ──────────────────────────────────
SUPABASE_URL=https://xxxxxxxxxxxx.supabase.co
SUPABASE_SERVICE_KEY=your-service-role-key
SUPABASE_ANON_KEY=your-anon-key

Step 5 — Verify Setup

# Test config loads correctly
python config.py
# Expected: no output (all vars found)

# Test Supabase connection
python -m database.client
# Expected: ✅ Supabase async connected successfully!

# Test DB writes
python -m database.writer
# Expected: all 4 operations succeed, check Supabase table editor

🚀 Running the Project

⚠️ Always start the processor BEFORE the publisher to avoid message ordering issues.

Terminal 1 — Start the Processor

source venv/bin/activate
python -m processor.main

Expected output:

📡 Listening on 3 Pub/Sub subscriptions
🔌 WebSocket server: ws://localhost:8000/ws
🏥 Health check:     http://localhost:8000/health

Verify endpoints:

Terminal 2 — Open Dashboard

Open dashboard/index.html in your browser via Live Server or:

cd dashboard && python -m http.server 3000

Then visit: http://localhost:3000

You should see the green connected dot in the header.

Terminal 3 — Run a Mock Call

source venv/bin/activate
python -m publisher.mock_call

Watch the dashboard update live — transcript lines appear one by one, sentiment updates in real time, and call history loads from Supabase.

🐛 Edge Cases & Known Behaviours

1. Message Ordering Issue

Problem: If publisher runs before processor starts, Pub/Sub queues messages. When processor starts, it consumes them out of order — transcripts arrive before call_started, causing foreign key violations.

Why: Pub/Sub doesn't guarantee strict ordering across topics.

Fix: Always start processor first. If ordering issues occur, purge subscriptions via: console.cloud.google.com/cloudpubsub/subscription/list → select subscription → Purge messages

2. Old Messages on Processor Restart

Problem: When processor restarts, it consumes ALL unacknowledged messages from previous sessions (Pub/Sub stores them for 7 days).

Why: This is actually a feature — messages aren't lost if processor goes down.

Fix: This is expected behaviour. If you want a clean start, purge subscriptions manually.

3. Sentiment Keyword Matching is Basic

Problem: The sentiment analyser uses keyword matching — context-free. "I need help" incorrectly scores as negative because of the word "help".

Why: We deliberately kept sentiment simple to focus on the pipeline architecture.

Fix (future): Replace analyse_sentiment() in mock_call.py with Google Natural Language API:

from google.cloud import language_v1
client = language_v1.LanguageServiceClient()

4. No Auth on WebSocket

Problem: Any client can connect to ws://localhost:8000/ws and receive all call data.

Why: POC — auth adds complexity without educational value here.

Fix (future): Add JWT token validation in websocket_endpoint():

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str):
    if not verify_token(token):
        await websocket.close(code=1008)
        return

5. All Dashboards See All Calls

Problem: Every connected dashboard sees every call, regardless of which agent is handling it.

Why: broadcast() sends to all clients — no filtering by agent.

Fix (future): Track agent ID per WebSocket connection and use send_to_agent() instead of broadcast().

6. Virtual Environment Not Active

Problem: ModuleNotFoundError: No module named 'google' when running scripts.

Fix: Always activate venv first:

source venv/bin/activate  # Mac/Linux

Or set VS Code interpreter to ./venv/bin/python via Cmd+Shift+P → Python: Select Interpreter.

7. Python 3.14 pip Bug

Problem: ModuleNotFoundError: No module named 'pip._vendor.packaging' when using Python 3.14.

Why: Python 3.14 has a broken pip at time of writing.

Fix: Use Python 3.11 explicitly:

brew install python@3.11
python3.11 -m venv venv

🚀 What's Next

Layer 3 Improvements (Immediate)

  • Real Google STT — feed actual .wav audio files through google-cloud-speech instead of hardcoded text
  • Real sentiment — replace keyword matching with Google Natural Language API
  • Per-agent WebSocket — filter events by agent_id so each dashboard only sees its own call
  • JWT auth on WebSocket endpoint
  • Deploy processor to Cloud Run — containerise with Docker, deploy to GCP
  • Retry logic — if call_started hasn't arrived yet, queue transcripts and retry

Layer 4 — Telephony Integration

  • Asterisk/FreeSWITCH — simulate SBC and SIPREC streaming
  • Real SIP Gateway — route audio to GCP's telephony stack
  • UCID generation — replicate how SBC assigns UCIDs in real calls

Layer 1 — Agent Desktop

  • Rebuild dashboard in React — component-based, TypeScript
  • Knowledge Assist panel — surface relevant articles based on transcript
  • AI Coach suggestions — use Gemini to suggest responses to agent in real time
  • Call summarization — generate summary when call ends
  • Salesforce integration — embed dashboard in Salesforce Agent Desktop

Infrastructure

  • GitHub Actions CI/CD — auto-deploy processor to Cloud Run on push
  • Terraform — infrastructure as code for all GCP resources
  • Monitoring — Cloud Monitoring + alerting for processor health

📚 References

👤 Author

Avinash Sharma — Senior Frontend Engineer

  • Currently exploring agentic AI development and GCP infrastructure
  • This POC is part of a larger learning roadmap toward production-grade agentic systems

Built as a personal learning POC.

About

No description, website, or topics provided.

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages