How I Sent 117K+ Personalised Emails and 10K+ AI Voice Calls in One Day Using GCP Pub/Sub

This is a detailed engineering breakdown of the AI outreach infrastructure I built at Speculo.ai. The system processed 117K+ leads, generated personalised emails and call scripts using LLMs, and dispatched everything autonomously using a four-stage GCP Pub/Sub pipeline with parallel Cloud Run workers. No humans in the loop. $2M+ in attributed revenue. This is exactly how it works.

Full pipeline from lead ingestion to email and voice dispatch. All inter-layer communication goes through Pub/Sub topics.

Key insight:Do not design your message queue around your current load. We initially set max instances to 10 for the enrichment layer and the backlog grew unbounded. The fix was raising max instances to 50 and adding a concurrency limit of 5 per instance. Pub/Sub's backlog metrics in Cloud Monitoring make this easy to tune.

import json
from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
TOPIC = "projects/speculo-prod/topics/leads-raw"

def ingest_lead(lead: dict) -> str:
    if not is_valid(lead):
        publish_dlq(lead, reason="schema_invalid")
        return "rejected"

    lead["icp_score"] = compute_icp_score(lead)
    lead["lead_id"]   = deduplicate_id(lead)   # SHA256 of email+domain

    future = publisher.publish(
        TOPIC,
        data=json.dumps(lead).encode("utf-8"),
        source=lead.get("source", "unknown"),   # message attributes for filtering
    )
    return future.result()  # blocks only until broker acks, ~5ms

import asyncio
from google.cloud import pubsub_v1
from agents import enrich_company, verify_contact

subscriber = pubsub_v1.SubscriberClient()
PULL_SUB   = "projects/speculo-prod/subscriptions/enrichment-sub"
PUSH_TOPIC = "projects/speculo-prod/topics/leads-enriched"

async def process(message: pubsub_v1.types.ReceivedMessage):
    lead = json.loads(message.message.data)
    try:
        company_ctx, contact = await asyncio.gather(
            enrich_company(lead["domain"]),   # Tavily + Gemini summarise
            verify_contact(lead["email"]),     # Hunter.io + MX check
        )
        if not contact["deliverable"]:
            message.nack()   # requeue for DLQ after max attempts
            return

        enriched = {**lead, "company_ctx": company_ctx, "contact": contact}
        publisher.publish(PUSH_TOPIC, json.dumps(enriched).encode())
        message.ack()
    except Exception as e:
        log_error(e, lead["lead_id"])
        message.nack()   # retry with exponential backoff

async def personalise(enriched: dict) -> dict:
    prompt = build_prompt(
        candidate  = HASNAIN_PROFILE,
        company    = enriched["company_ctx"],
        contact    = enriched["contact"],
        news_hook  = enriched["company_ctx"]["recent_news"][0],
    )
    response = await openai.chat.completions.create(
        model           = "gpt-4o",
        messages        = [{"role": "user", "content": prompt}],
        response_format = {"type": "json_object"},
        temperature     = 0.7,
        max_tokens      = 800,
    )
    result = json.loads(response.choices[0].message.content)

    # Hard quality gate — discard and requeue if score < 8
    if result.get("review_score", 0) < 8:
        return requeue_for_repersonalisation(enriched)

    return {
        **enriched,
        "subject"     : humanise(result["subject"]),   # strip AI tells
        "body_html"   : humanise(result["body_html"]),
        "call_script" : result["call_script"],
        "channel"     : determine_channel(enriched),   # email | voice | both
    }

import redis
import time

r = redis.Redis.from_url(os.environ["MEMORYSTORE_URL"])

def acquire_token(api_key: str, rpm_limit: int) -> bool:
    """
    Returns True if token acquired, False if rate limit reached.
    Uses Redis atomic increment with TTL-based window.
    """
    bucket_key = f"ratelimit:{api_key}:{int(time.time() // 60)}"
    count = r.incr(bucket_key)
    if count == 1:
        r.expire(bucket_key, 65)   # slightly over 60s to avoid edge cases
    return count <= rpm_limit

async def call_with_backoff(fn, api_key: str, rpm: int, max_attempts=5):
    for attempt in range(max_attempts):
        if acquire_token(api_key, rpm):
            return await fn()
        wait = min(4 ** attempt, 30)   # 1s, 4s, 16s, 30s cap
        await asyncio.sleep(wait)
    raise RateLimitExhausted(f"Key {api_key[:8]}... exhausted after {max_attempts} attempts")

class KeyPool:
    def __init__(self, keys: list[str], daily_limit: int):
        self.keys         = keys
        self.daily_limit  = daily_limit
        self._idx         = 0

    def acquire(self) -> str | None:
        for _ in range(len(self.keys)):
            key   = self.keys[self._idx % len(self.keys)]
            usage = get_daily_usage(key)          # BigQuery lookup, cached 60s
            if usage < self.daily_limit * 0.9:    # 90% threshold
                self._idx += 1
                return key
            self._idx += 1
        return None   # all keys exhausted — pipeline pauses

KEY_POOL = KeyPool(
    keys        = os.environ["GEMINI_KEYS"].split(","),
    daily_limit = 1000,   # free tier RPD per project
)

from google.cloud import firestore

db = firestore.Client()

async def dispatch_email(message: dict) -> bool:
    lead_id  = message["lead_id"]
    send_ref = db.collection("sent_emails").document(lead_id)

    # Phase 1: claim the send slot (atomic transaction)
    @firestore.transactional
    def claim(txn):
        snap = send_ref.get(transaction=txn)
        if snap.exists:
            return False   # already sent — idempotent skip
        txn.set(send_ref, {"status": "pending", "ts": firestore.SERVER_TIMESTAMP})
        return True

    txn = db.transaction()
    if not claim(txn):
        return False   # another worker already handled this lead

    # Phase 2: actually send
    try:
        await send_via_smtp(message)
        send_ref.update({"status": "sent", "sent_at": firestore.SERVER_TIMESTAMP})
        stream_to_bigquery(message, event="email_sent")
        return True
    except Exception as e:
        send_ref.update({"status": "failed", "error": str(e)})
        raise   # Pub/Sub will redeliver to another worker

from twilio.rest import Client as TwilioClient

twilio = TwilioClient(ACCOUNT_SID, AUTH_TOKEN)

async def initiate_call(lead: dict) -> str:
    call = twilio.calls.create(
        to   = lead["contact"]["phone"],
        from_= WARM_CALLER_ID,
        url  = f"{WEBHOOK_BASE}/voice/start?lead_id={lead['lead_id']}",
        # TwiML webhook streams ElevenLabs audio + handles DTMF
        status_callback = f"{WEBHOOK_BASE}/voice/events",
        machine_detection = "Enable",   # skip answering machines
        machine_detection_timeout = 4,
    )
    stream_to_bigquery(lead, event="call_initiated", call_sid=call.sid)
    return call.sid

# TwiML webhook — served by Cloud Run
@app.post("/voice/start")
async def voice_start(lead_id: str):
    lead   = get_lead(lead_id)
    script = lead["call_script"]
    audio_url = await elevenlabs.generate_audio(
        text  = script["opening"],
        voice = VOICE_ID,   # consistent voice across all calls
    )
    return TwiML.gather_response(audio_url, action="/voice/respond")

# events table — partitioned by DATE(event_ts)
CREATE TABLE speculo.pipeline_events (
  lead_id       STRING NOT NULL,
  event_type    STRING NOT NULL,   -- ingest | enrich | personalise | email_sent | call_started | reply
  event_ts      TIMESTAMP,
  domain        STRING,
  contact_email STRING,
  channel       STRING,
  metadata      JSON,              -- flexible per-event payload
  worker_id     STRING,            -- Cloud Run instance ID for debugging
)
PARTITION BY DATE(event_ts)
CLUSTER BY event_type, domain;

-- Daily funnel query
SELECT
  event_type,
  COUNT(*) AS count,
  COUNT(*) / LAG(COUNT(*)) OVER (ORDER BY stage_order) AS conversion_rate
FROM speculo.pipeline_events
WHERE DATE(event_ts) = CURRENT_DATE()
GROUP BY event_type, stage_order
ORDER BY stage_order;

Building a high-scale AI pipeline?

I have designed and shipped agentic systems that process hundreds of thousands of operations autonomously. If you are building something in this space and want to talk architecture, get in touch.

Written by Hasnain Ali, Senior AI Engineer. Specialising in production LLM systems, agentic pipelines, and GCP infrastructure. Connect on LinkedIn.