Omnichannel AI Support & Lead Qualification Platform | Technical Architecture Case Study

Production System Architecture Overview

Commercial Production Note: This omnichannel AI customer support and lead qualification engine is actively operated and turnkey-deployed for commercial clients via Assembly Minds.

Enterprise sales and customer support organizations struggle with fragmented inbound communication channels. Customers reach out across WhatsApp, Telegram, and interactive web widgets expecting instantaneous, context-aware answers, while sales teams face delayed lead qualification, manual CRM data entry, and inaccurate routing.

This case study details a high-throughput, production conversational AI system engineered to unify omnichannel inbound streams into a single deterministic processing engine. Built with FastAPI, PostgreSQL 16 with pgvector, Redis, and Celery, the platform delivers semantic knowledge retrieval, real-time qualification scoring, bidirectional CRM synchronization (HubSpot / Salesforce), and zero-friction human-in-the-loop (HITL) agent handoffs.

[ Inbound Users ]
  │
  ├─► WhatsApp (Meta Cloud API) ──┐
  ├─► Telegram (Bot API Webhook) ──┼──► [ Cloudflare Edge / Nginx Reverse Proxy ]
  └─► Web Widget (WebSocket/JWT) ─┘                    │ (HTTPS / WSS)
                                                       ▼
                                      [ FastAPI ASGI Orchestration Core ]
                                         │  (Async Webhook Ingestion)
                                         ├─ Channel Normalization & HMAC Auth
                                         ├─ Session State Manager (Redis 7.2)
                                         │
                 ┌───────────────────────┴───────────────────────┐
                 ▼                                               ▼
     [ Semantic Retrieval Tier ]                    [ Function Calling & CRM Tier ]
     ┌───────────────────────────────┐              ┌─────────────────────────────┐
     │ PostgreSQL 16 + pgvector      │              │ Pydantic v2 Output Validator│
     │ HNSW Index (<=> Cosine Dist)  │              │ Lead Score Evaluator (BANT) │
     │ Metadata-Filtered Chunking    │              │ Celery Worker Queue (Redis) │
     └───────────────────────────────┘              └──────────────┬──────────────┘
                 │                                                 │
                 ▼                                                 ▼
     [ In-Memory Query Cache ]                         [ Bidirectional Sync ]
     (Redis TTL / Semantic Hashing)                 HubSpot CRM / Salesforce / Slack
                 │                                                 │
                 └───────────────────────┬─────────────────────────┘
                                         ▼
                       [ Deterministic Guardrail Engine ]
                         ├─ Confidence Score Assessment (< 0.70)
                         ├─ PII Redaction & Sanitization
                         └─ HITL Escalation Router
                                         │
                      ┌──────────────────┴──────────────────┐
                      ▼                                     ▼
           [ Automated AI Response ]               [ Human Agent Inbox ]
           (Dispatched via Channel API)            (Context Briefing + Live Chat)

Architectural Highlights

Normalized Inbound Ingestion: Ingests asynchronous webhook payloads from WhatsApp Cloud API, Telegram Bot Webhooks, and browser WebSockets into a unified event format (InboundMessageEvent), decoupling channel-specific payload variations from core downstream logic.
Sub-Millisecond Vector Search: PostgreSQL 16 equipped with pgvector utilizing Hierarchical Navigable Small World (HNSW) indexing across 1536-dimensional embeddings. Enforces hard metadata filtering by organization, department, and document access level.
Two-Tier Semantic Caching: Leverages Redis with SHA-256 query fingerprinting and semantic similarity evaluation to bypass LLM inference for recurring operational inquiries, cutting p95 response time to sub-1.2s and reducing API token spend by 42%.
Bidirectional Event-Driven CRM Sync: Dynamic function calling extracts structured BANT (Budget, Authority, Need, Timeline) qualification parameters, enqueuing background Celery tasks that idempotently create contacts, score leads, and assign deal stages in HubSpot and Salesforce.
Fail-Safe Human-in-the-Loop Escalation: Automatically routes conversations to live human support desks whenever retrieval confidence drops below 0.70, sentiment registers negative volatility, or explicit operator escalation is requested—injecting structured conversation summaries directly into Zendesk/Slack.

1. Unified Omnichannel Ingestion & Gateway Layer

Each messaging network presents idiosyncratic payload structures, webhook verification protocols, and rate-limiting constraints. The ingestion layer normalizes all incoming traffic into an immutable internal event model before dispatching to the processing pipeline.

Webhook Verification & HMAC Authentication

Meta WhatsApp Cloud API requires cryptographic payload verification via HMAC-SHA256 headers (X-Hub-Signature-256), while Telegram utilizes secret webhook tokens. The FastAPI gateway verifies these signatures at the edge before allocating compute resources:

import hmac
import hashlib
from fastapi import APIRouter, Request, HTTPException, Header, status
from pydantic import BaseModel, Field
from typing import Optional, Literal

router = APIRouter(prefix="/v1/webhooks", tags=["Webhooks"])

class InboundMessageEvent(BaseModel):
    channel: Literal["whatsapp", "telegram", "webchat"]
    channel_user_id: str = Field(..., description="Unique platform user identifier (phone or chat_id)")
    session_id: str = Field(..., description="Deterministic tenant-scoped session key")
    message_text: str = Field(..., max_length=4096)
    timestamp: int
    metadata: dict = Field(default_factory=dict)

@router.post("/whatsapp")
async def handle_whatsapp_webhook(
    request: Request,
    x_hub_signature_256: Optional[str] = Header(None)
):
    raw_body = await request.body()
    app_secret = request.app.state.settings.WHATSAPP_APP_SECRET.encode()
    
    # Verify cryptographic signature
    if not x_hub_signature_256:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing signature header")
    
    expected_signature = "sha256=" + hmac.new(app_secret, raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected_signature, x_hub_signature_256):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid HMAC signature")

    payload = await request.json()
    # Normalize WhatsApp Graph API v20.0 nested payload
    event = parse_whatsapp_payload(payload)
    if event:
        await request.app.state.orchestrator.dispatch_event(event)
    return {"status": "accepted"}

2. Hybrid Semantic Retrieval Pipeline with pgvector & Redis

Accurate automated support requires grounding in company knowledge bases, warranty policies, API documentation, and pricing tables without hallucinations.

Database Schema & HNSW Indexing

Rather than introducing standalone vector database silos, the architecture embeds pgvector natively into PostgreSQL 16. This preserves relational ACID transactions while supporting cosine vector distance searches accelerated by HNSW indexes:

-- Enable pgvector extension on PostgreSQL 16
CREATE EXTENSION IF NOT EXISTS vector;

-- Knowledge base chunk repository with tenant isolation
CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    document_id UUID NOT NULL,
    category VARCHAR(64) NOT NULL,
    access_tier VARCHAR(32) DEFAULT 'public',
    content TEXT NOT NULL,
    token_count INT NOT NULL,
    metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
    embedding vector(1536) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW cosine index for sub-10ms nearest neighbor queries
CREATE INDEX idx_chunks_embedding_hnsw 
ON document_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Composite index for fast tenant metadata filtering prior to distance evaluation
CREATE INDEX idx_chunks_tenant_category 
ON document_chunks (tenant_id, category, access_tier);

Metadata-Filtered Similarity Query Execution

When an inbound message arrives, an embedding is generated and queried with strict metadata boundaries:

async def query_semantic_knowledge(
    db_session: AsyncSession,
    tenant_id: str,
    query_vector: list[float],
    category_filter: Optional[str] = None,
    distance_threshold: float = 0.35,
    top_k: int = 4
) -> list[dict]:
    """
    Executes cosine distance search on pgvector with strict tenant metadata boundaries.
    Cosine distance = 1 - cosine_similarity. Lower distance implies closer match.
    """
    base_query = """
        SELECT id, content, category, metadata,
               (embedding <=> :query_vec) AS cosine_distance
        FROM document_chunks
        WHERE tenant_id = :tenant_id
    """
    params = {"query_vec": str(query_vector), "tenant_id": tenant_id, "limit": top_k}

    if category_filter:
        base_query += " AND category = :category"
        params["category"] = category_filter

    base_query += """
        AND (embedding <=> :query_vec) <= :threshold
        ORDER BY cosine_distance ASC
        LIMIT :limit;
    """
    params["threshold"] = distance_threshold

    result = await db_session.execute(text(base_query), params)
    return [dict(row._mapping) for row in result.fetchall()]

3. Dynamic Tool Calling & Asynchronous CRM Synchronization

Customer support agents don’t merely recite FAQs; they qualify prospects, calculate pricing tiers, and sync prospective buyers directly into sales pipelines.

Deterministic Qualification Extraction

Using structured function calling with Pydantic v2 models, the model parses user statements into standardized BANT attributes:

from pydantic import BaseModel, Field
from typing import Optional

class LeadQualificationData(BaseModel):
    contact_name: Optional[str] = Field(None, description="Full name of the lead")
    business_email: Optional[str] = Field(None, description="Corporate email address")
    company_name: Optional[str] = Field(None, description="Organization or trading entity")
    budget_range: Optional[str] = Field(None, description="Estimated monthly or annual budget tier")
    timeline_weeks: Optional[int] = Field(None, description="Target deployment timeframe in weeks")
    primary_use_case: str = Field(..., description="Core business pain point or required solution")
    qualification_score: int = Field(..., ge=0, le=100, description="Algorithmic BANT qualification score")
    ready_for_sales_outreach: bool = Field(..., description="Flag determining immediate rep assignment")

Idempotent Background Task Dispatch (Celery + Redis)

Direct synchronous HTTP calls to third-party CRM APIs introduce latency spikes and network failures that degrade the conversational experience. Instead, verified lead payloads are committed to an asynchronous queue:

from celery import Celery
import requests

celery_app = Celery("crm_sync", broker="redis://prod_cache:6379/1")

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def sync_lead_to_hubspot(self, tenant_id: str, lead_dict: dict, conversation_history: list):
    """
    Idempotently upserts lead contact records and logs qualification notes in HubSpot CRM.
    """
    try:
        api_token = get_tenant_secret(tenant_id, "HUBSPOT_ACCESS_TOKEN")
        headers = {"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"}
        
        # Upsert contact by business email
        payload = {
            "properties": {
                "email": lead_dict["business_email"],
                "firstname": lead_dict.get("contact_name", "").split(" ")[0],
                "company": lead_dict.get("company_name", ""),
                "lead_qualification_score": str(lead_dict["qualification_score"]),
                "ai_notes": f"Primary Need: {lead_dict['primary_use_case']} | Timeline: {lead_dict.get('timeline_weeks')} weeks"
            }
        }
        res = requests.post("https://api.hubapi.com/crm/v3/objects/contacts", json=payload, headers=headers, timeout=10)
        res.raise_for_status()
        return {"status": "synced", "lead": lead_dict["business_email"]}
    except requests.exceptions.RequestException as exc:
        raise self.retry(exc=exc)

4. Deterministic Guardrails & Human-in-the-Loop (HITL) Handoff

Autonomous AI systems must recognize their operational boundaries. When faced with complex disputes, low retrieval confidence, or high-value enterprise accounts requesting human touch, the platform initiates deterministic live-agent handoff.

[ User Interaction ] ──► [ Intent & Confidence Evaluator ]
                                   │
              ┌────────────────────┴────────────────────┐
              │ Cosine Distance > 0.35 OR Escalation Flag│
              ▼                                         ▼
      [ High Confidence ]                       [ Low Confidence / Request ]
              │                                         │
    [ Stream AI Response ]                     [ Initiate HITL Handoff ]
              │                                         │
              │                                         ├─ Generate 3-Sentence Conversation Digest
              │                                         ├─ Freeze Session AI Auto-Responder
              │                                         ├─ Push Event to Zendesk / Live Agent Inbox
              │                                         └─ Dispatch Handover Notice to User

Automated Live-Agent Dispatch Logic

When a conversation triggers escalation, the session state in Redis is toggled to HUMAN_CONTROL, preventing contradictory responses while alerting the human triage team:

async def trigger_human_escalation(
    redis_client,
    channel_adapter,
    session_id: str,
    channel_user_id: str,
    reason: str,
    transcript: list[dict]
):
    # 1. Update session state to human-control mode with 2-hour TTL
    await redis_client.set(f"session:{session_id}:mode", "HUMAN_CONTROL", ex=7200)
    
    # 2. Synthesize concise briefing for the human agent
    summary_digest = await generate_context_briefing(transcript)
    
    # 3. Post notification to agent dashboard / internal escalation channel
    await notify_agent_queue(
        session_id=session_id,
        user_id=channel_user_id,
        reason=reason,
        briefing=summary_digest
    )
    
    # 4. Inform user gracefully on their native channel
    courtesy_msg = "I've connected our senior team to review your request. An agent will reply directly here shortly."
    await channel_adapter.send_message(channel_user_id, courtesy_msg)

5. Production Metrics & Business Impact

Since deployment into live enterprise production environments across multi-tier accounts, the platform has demonstrated marked improvements in customer support response latency and sales velocity:

78% Automated Resolution Rate: Routine support tickets, order inquiries, and technical FAQs resolved autonomously without agent intervention.
Sub-1.2s Response Latency: End-to-end round trip time (from message arrival to dispatch across WhatsApp/Telegram webhooks) maintained below 1,200ms.
3x Acceleration in Inbound Routing: Hot qualified inbound leads (BANT score > 80) synchronized to HubSpot and alerted to sales executives in under 4 seconds, increasing demo bookings by 41%.
Zero Hallucination Tolerance: Cosine distance cutoff thresholds (<= 0.35) ensure unverified claims are suppressed in favor of instant human handoff.
24/7 Global Availability: Fault-tolerant containerized architecture delivering 99.95% uptime across high-volume marketing campaigns.