README.md

metadata

title: CustomerCore API
emoji: 🚀
colorFrom: blue
colorTo: indigo
sdk: docker
app_port: 7860
pinned: false

🧠 CustomerCore

Enterprise AI Customer Support Platform — Multi-Agent Triage, Churn Prediction & Real-Time Observability

🖥️ Live Demo • 📊 MLflow Experiments • 📦 Model Registry • ⚙️ CI/CD

---

📌 What is CustomerCore?

CustomerCore is a production-grade, multi-tenant B2B AI platform that automatically triages customer support tickets using a network of 6 specialized AI agents. It classifies tickets, predicts customer churn risk, detects infrastructure outages, retrieves similar past cases, and routes everything to the right team — all in under 5 seconds.

Not just a prototype — it has CI/CD pipelines, containerized deployment on Hugging Face Spaces, MLOps model tracking via DagsHub, LLM observability via Langfuse, a privacy vault for PII masking, Constitutional AI safety guardrails, and a Human-in-the-Loop review system for high-risk decisions.

---

🔗 Live Infrastructure

Service	Link	What it does
🖥️ Operations Console	HF Space	Live dashboard — submit tickets, view AI triage results
📊 MLflow Experiments	DagsHub	Training runs, metrics, model comparison charts
📦 Model Registry	DagsHub Models	Versioned churn model with Production/Staging labels
📂 Dataset Versioning	DagsHub DVC	Raw + processed datasets tracked via DVC
⚙️ CI/CD Pipelines	GitHub Actions	Lint, test, train, deploy — on every push

---

🏗️ How It All Connects — End to End

This is how a single support ticket flows through the entire system:

sequenceDiagram
    autonumber
    actor User as Client / Dashboard
    participant API as FastAPI Server
    participant Vault as Privacy Vault
    participant Graph as LangGraph 6-Agent Pipeline
    participant LLM as LLM Router (Local/Cloud)
    participant ML as Churn Model (Random Forest)
    participant DB as Supabase PostgreSQL
    participant VDB as ChromaDB / pgvector
    participant Trace as Langfuse + Prometheus

    User->>API: Submit support ticket (JWT auth)
    API->>API: Verify JWT → extract tenant_id + role
    API->>Vault: Mask PII (emails, phones, names, credit cards)
    Vault-->>API: Sanitized text
    API->>DB: Save ticket (status: pending)
    API-->>User: HTTP 202 Accepted
    
    Note over API,Graph: Agent pipeline runs in background
    API->>Graph: Dispatch ticket to 6-agent supervisor
    Graph->>LLM: Classify Agent → categorize + assign priority
    LLM-->>Graph: category=billing, priority=high
    Graph->>DB: Memory Agent → recall past customer interactions
    Graph->>VDB: RAG Agent → BM25 + vector hybrid search
    VDB-->>Graph: Top 5 similar past resolutions
    Graph->>ML: Churn Agent → predict churn probability
    ML-->>Graph: churn_risk=0.73
    Graph->>Graph: Incident Agent → check for outage patterns
    Graph->>Graph: HITL Agent → should human review?
    Graph->>LLM: Generate resolution using retrieved context
    Graph->>Vault: Constitutional Policy check on AI response
    Graph->>DB: Save final triage result + audit log
    Graph->>Trace: Log full trace (latency, tokens, cost, decisions)

---

🧠 The 6-Agent Triage Pipeline

CustomerCore uses LangGraph to orchestrate 6 specialized agents in a parallel + sequential hybrid pipeline with conditional Human-in-the-Loop gating. The graph topology uses a fan-out / fan-in pattern where the Memory, RAG, Churn, and Incident agents execute concurrently after classification, then converge at the HITL gate:

graph LR
    A[📩 Ticket In] --> B[🏷️ Classify]
    B --> C[🧠 Memory]
    B --> D[🔍 RAG]
    B --> E[📉 Churn]
    B --> F[🚨 Incident]
    C & D & E & F --> G{🛑 HITL Check}
    G -->|Safe| H[✅ Finalize]
    G -->|Risky| I[⏸️ Human Review]
    I --> H

#	Agent	What it does	Key detail
1	Classify Agent	Categorizes ticket (Billing, Technical, Security, Account) and assigns priority (Low → Critical)	Uses SLA-Aware LLM Router — local Gemma for routine, cloud frontier for critical
2	Memory Agent	Recalls past interactions for this customer	Uses Mem0 with tenant-scoped identifiers
3	RAG Agent	Hybrid search — BM25 keyword + ChromaDB vector retrieval with Reciprocal Rank Fusion (RRF)	Optional cross-encoder reranking via ms-marco-MiniLM-L-6-v2
4	Churn Agent	Runs customer features through our trained Random Forest model	Features: monthly spend, tenure, ticket count, contract months
5	Incident Agent	Analyzes recent ticket frequency to detect systemic outages	Spike detection across the same tenant
6	HITL Agent	Flags tickets for human review	Triggers when: confidence < 0.65, priority is critical, or safety policy violated

When HITL flags a ticket, LangGraph's interrupt_before checkpoint pauses execution. The state is saved in memory. A human operator reviews from the dashboard and calls resume_triage() to continue. The LangGraph MemorySaver checkpointer makes this possible — it stores the full agent state at the interrupt point so execution can resume exactly where it stopped.

Performance gain: Running Memory, RAG, Churn, and Incident agents concurrently instead of sequentially reduced pipeline latency by ~45% (from ~7 seconds to ~3–4 seconds for complex tickets).

---

🔍 RAG — How Knowledge Retrieval Works

CustomerCore doesn't use plain vector search. It uses a 3-layer Graph-RAG system:

graph TD
    Q[User Query] --> V[Layer 1: Vector + BM25 Hybrid Search]
    Q --> G[Layer 2: B2B Knowledge Graph - NetworkX]
    Q --> S[Layer 3: SQL Analytics - DuckDB Gold Marts]
    
    V --> |Top 5 similar tickets| M[Combined Context Builder]
    G --> |Tenant profile + category trends| M
    S --> |Ticket funnel + customer health| M
    M --> LLM[LLM Prompt Injection]

Layer	What it does	Why it matters
Vector + BM25	ChromaDB dense search + BM25 sparse keyword search, merged via Reciprocal Rank Fusion	Dense catches semantics ("frustrated" ≈ "angry"), BM25 catches exact terms ("error code 5012")
Knowledge Graph	NetworkX DiGraph connecting Ticket → Tenant → Category nodes	Answers "Why has acme-corp been escalating?" — pure vector search can't do this
DuckDB Gold	SQL queries on dbt-transformed Parquet marts	Returns structured analytics: ticket funnel, customer health daily, agent performance

Tenant Isolation: Every ChromaDB query has a mandatory where={"tenant_id": current_tenant} filter. BM25 index is partitioned per tenant. Cross-tenant leakage is architecturally impossible.

---

🤖 Models — What We Built vs What We Call

Models We Trained (Custom ML)

Model	Algorithm	Task	Tracked on
Churn Classifier	Random Forest, Logistic Regression, Gradient Boosting	Predict customer churn probability from account features	DagsHub MLflow
Support LLM (QLoRA)	Llama 3 8B Instruct (PEFT / QLoRA)	Predict professional, tenant-specific ticket resolutions	Hugging Face Adapter Registry

• Churn Classifier: The training pipeline (src/ml/train_churn.py) trains all 3 models, compares F1 scores, and auto-promotes the best one to "Production" in the DagsHub Model Registry. Logged artifacts include ROC curves, confusion matrices, and feature importance plots.
• Support LLM (QLoRA): The serverless training pipeline (src/ml/modal_train.py) executes on-demand on an Nvidia A10G GPU using Modal, pulls historical resolved support tickets (raw_text and suggested_resolution) from Supabase, performs QLoRA fine-tuning in 4-bit NF4 double-quantization, and publishes the resulting LoRA adapter (customercore-llama3-adapter) to the Hugging Face Hub.

Models We Call via API (LLM)

Model	Provider	When it's used
Gemma 3 4B	Ollama (runs locally)	Classification, extraction, routine reasoning
Gemini 2.5 Flash	OpenRouter (cloud API)	Fast-tier cloud fallback — when Ollama unavailable
Claude 3.5 Sonnet	OpenRouter (cloud API)	Frontier-tier reasoning for critical/complex tickets

SLA-Aware LLM Router — Smart Model Selection

The router (src/rag/router.py) automatically picks the right model based on task type, priority, and runtime environment:

Task	Low/Medium Priority	High/Critical Priority
Classify	🟢 Local Gemma (~150ms, $0)	🟢 Local Gemma
Extract	🟢 Local Gemma	🟢 Local Gemma
Reason	🟢 Local Gemma	🔵 Cloud Frontier (Claude 3.5 Sonnet)
Action	🔵 Cloud Fast (Gemini 2.5 Flash)	🔵 Cloud Frontier

Result: ~80% of tickets are handled locally at $0 cost and <200ms latency. Only complex/high-risk tickets go to the frontier cloud model.

Dynamic Local → Cloud Fallback: The LLMClient detects the runtime environment at startup. If SPACE_ID (Hugging Face) or APP_ENV=production is set, local Ollama calls are transparently redirected to openrouter/google/gemini-2.5-flash with zero timeout delay. If local Ollama fails on a development machine (e.g., not running), the same fallback activates automatically.

SLA latency targets per priority: Critical ≤200ms, High ≤500ms, Medium ≤1000ms, Low ≤2000ms. Violations are tracked in Prometheus metrics.

L2 Semantic Cache — Avoid Redundant LLM Calls

On top of model routing, a two-layer semantic cache (src/rag/router.py) intercepts calls before they reach the LLM:

Layer	Storage	Hit condition	Benefit
L1 Exact	Redis in-memory dict	Same prompt hash	Sub-millisecond, $0 cost
L2 Semantic	Redis + embedding similarity	Cosine similarity > 0.92	~85% of near-duplicate tickets served from cache

Cache hit ratio is tracked as a Grafana metric. Each cache hit saves ~$0.002–$0.04 in API costs and 150–800ms of latency.

---

📊 Data Pipeline — How Data Flows

Data Ingestion Strategy

We use the stream-to-database-then-download approach (industry standard):

HuggingFace Datasets (API) 
  → Stream to Redpanda (Kafka-compatible broker)
    → Bronze Consumer (raw events)
      → PII Masking (Presidio + spaCy)
        → Silver Layer (cleaned)
          → dbt transforms (DuckDB)
            → Gold Layer (analytics-ready marts)

Why not download directly? In production, data comes from APIs, webhooks, and real-time events — not static files. Streaming through a message broker (Redpanda) decouples producers from consumers, handles backpressure, and enables replay. Supabase acts as the persistent storage layer (like S3 in AWS architectures).

Datasets Used

Dataset	Rows	License	Language	Purpose
Bitext Customer Support	26,872	CDLA-Sharing-1.0	English	SaaS support Q&A (billing, technical, account)
Bitext Retail Banking	25,545	CDLA-Sharing-1.0	English	Financial support Q&A (card, loan, compliance)
Amazon MASSIVE Intent	34,542	Apache 2.0	DE/FR/ES	Real multilingual customer utterances
Total	~87,000		4 languages

Medallion Architecture (Bronze → Silver → Gold)

Bronze (raw events from Redpanda)
  │
  ├── PII masking applied (Presidio)
  │
Silver (cleaned, PII-scrubbed events)
  │
  ├── dbt transformations (DuckDB adapter)
  │
Gold (analytics-ready marts):
  ├── customer_health_daily     — daily tenant health scores
  ├── ticket_funnel_daily       — intake/processing/resolved funnel
  ├── support_agent_performance — agent resolution metrics
  ├── billing_failure_summary   — payment failure patterns
  ├── incident_severity_hourly  — outage severity tracking
  ├── retention_cohort_metrics  — cohort-based churn analysis
  └── product_adoption_features — feature usage tracking

---

🖥️ The Dashboard (Hugging Face Space)

The live dashboard at huggingface.co/spaces/saibalajiomg/customercore is a single-page app built with vanilla HTML/CSS/JS, served directly from FastAPI.

What you see:

• Session Context Bar — Tenant, Role, and Auth status at the top. Changing either instantly re-generates a JWT and enforces access rules
• AI Triage Pipeline — Submit tickets or use 1-click demo presets (billing issue, outage report, PII leak, etc.)
• Prediction Cards — Priority, Routing Team, Churn Risk %, Outage Detection
• Suggested Resolution — AI-generated response with KB citations from RAG
• HITL Workspace — Role-gated: manager+ sees the flagged ticket review queue; support_agent sees an explicit "Access Denied" panel. Switching roles instantly clears any previously loaded data — no stale reviews persist
• System Health Panel — Service status (Supabase ✅, Redis, ChromaDB)

Understanding the Session Context:

Field	What it means
Tenant	The company/organization (e.g. "acme-corp", "globex", "initech"). All data is isolated per tenant — one tenant can never see another's tickets or triage history
Role	support_agent can submit tickets and view results. manager can additionally access the HITL Workspace to review and approve/override flagged AI decisions
Session / Token	Active JWT token status. The token carries tenant_id + role claims, signed with HS256. Auto-generated on the dashboard for demo purposes — in production this would come from Supabase Auth or Auth0

How Authentication Works:

1. Dashboard lets you pick a tenant + role → auto-generates a JWT (dev mode)
2. JWT payload: {"tenant_id": "acme-corp", "role": "support_agent", "exp": ...}
3. Every API call includes Authorization: Bearer <token>
4. Server verifies signature + expiry → extracts tenant_id → uses it for ALL data isolation
5. In production, tokens would come from Supabase Auth, Auth0, or similar

---

🔐 Privacy & Security

PII Masking — Privacy Vault

Every ticket goes through PII masking before it hits the database or any LLM:

Input:  "My email is john@acme.com and card 4111-1111-1111-1111, call me at +49 170 1234567"
Output: "My email is [EMAIL_REDACTED] and card [CREDIT_CARD_REDACTED], call me at [PHONE_REDACTED]"

• Uses Microsoft Presidio + spaCy NER running locally in-process
• Detects: emails, phone numbers, credit cards, names, SSNs, IBANs
• Supports reversible tokenization via AES-256-GCM encryption (authorized users can decrypt)
• Even Langfuse traces get PII-masked before being sent to cloud

Constitutional AI Safety Engine

An 8-rule safety engine that checks every AI response before it reaches the customer:

Rule	Severity	What it catches
PII Protection	🔴 CRITICAL	Response leaks raw PII back
No Commitments	🟡 VIOLATION	"We will refund $200 by Friday" — creates legal liability
Language Consistency	⚠️ WARNING	Response in English when ticket was in German
Toxicity Guard	🔴 CRITICAL	Harmful, discriminatory, or abusive language
AI Identity	🔴 CRITICAL	AI denies being an AI when asked directly
Scope Limitation	🟡 VIOLATION	Gives legal, medical, or financial advice
No Hallucination	🟡 VIOLATION	Cites KB articles that don't exist
Competitor Neutral	⚠️ WARNING	Disparages competitors by name

Two execution paths: Fast path (regex, <5ms) catches obvious violations. Slow path (LLM-based, ~500ms) handles nuanced cases like coded language or context-dependent toxicity.

Multi-Tenant Data Isolation

• JWT tokens carry tenant_id — every API request is cryptographically bound to a tenant
• ChromaDB: Metadata filter where={"tenant_id": X} on every query
• BM25 Index: Physically partitioned per tenant (separate corpora)
• Supabase: Row-Level Security (RLS) policies enforce isolation at the database layer
• Role-Based Access Control: support_agent < manager < admin — HITL resume requires manager+
• Frontend RBAC enforcement: The HITL Workspace instantly clears previously loaded reviews and displays an "Access Denied" message the moment the active role is switched to support_agent — no stale data persists across role changes

---

📊 MLOps Pipeline

Training Flow

src/ml/train_churn.py
  ├── Load dataset (DVC-tracked CSV from DagsHub)
  ├── Feature engineering (monthly_spend, tenure_months, ticket_count, contract_months)
  ├── Train 3 models:
  │   ├── Logistic Regression (baseline)
  │   ├── Random Forest (best performer)
  │   └── Gradient Boosting
  ├── Log params + metrics + artifact plots to MLflow @ DagsHub
  ├── Compare F1 scores across all models
  ├── Register best model → DagsHub Model Registry
  └── Auto-promote to "Production" stage

What gets logged to DagsHub MLflow:

• Parameters: n_estimators, max_depth, criterion, test_size, random_state
• Metrics: Accuracy, F1-Score, Recall, Precision, AUC-ROC
• Artifacts: ROC Curve, Confusion Matrix Heatmap, Feature Importance Chart
• Dataset: Registered as an MLflow dataset input for full reproducibility

Dataset Versioning (DVC)

Datasets are tracked with DVC and stored on DagsHub's S3-compatible remote. dvc push / dvc pull to sync. The .dvc files in the repo are pointers to the actual data files.

---

🏛️ Architecture — Local vs Cloud

CustomerCore runs in two modes without changing any application code — the same business logic works in both:

Component	💻 Local Dev Mode	☁️ Cloud Mode (HF Spaces)
API Server	uvicorn on localhost:8080	Docker container on HF Space (port 7860)
Database	SQLite file (customercore.db)	Supabase PostgreSQL (cloud-managed)
Vector Store	In-process ChromaDB (Docker)	Supabase pgvector
LLM (routine)	Ollama → Gemma 3 4B (local GPU/CPU)	Ollama → Gemma 3 4B
LLM (complex)	OpenRouter → Llama 3.1 8B	OpenRouter → Llama 3.1 8B
PII Masking	Local Presidio + spaCy	Local Presidio + spaCy (in-container)
Cache	Redis (Docker container)	Upstash Redis (serverless)
Event Streaming	Redpanda broker (Docker)	Async background tasks
Object Storage	MinIO (S3-compatible, Docker)	DagsHub DVC remote
Metrics	Prometheus + Grafana (Docker)	OpenTelemetry → Grafana Cloud
LLM Tracing	Console debug logs	Langfuse Cloud (full trace UI)
ML Tracking	Local MLflow server	DagsHub MLflow (cloud)
Secrets	.env file	Doppler (secrets management) + HF Space Secrets
Deployment	uvicorn --reload	GitHub Actions → HF Space (auto-deploy)

How Cloud Mode Works (HF Spaces):

1. Push to main branch triggers GitHub Actions
2. hf_deploy.yml workflow syncs code to HF Space repo
3. HF builds the Docker image from Dockerfile.hf
4. Container starts: runs database migrations → starts uvicorn → serves dashboard
5. Secrets (Supabase URL, API keys, etc.) are injected as HF Space secrets
6. Supabase handles persistent data — the HF container is stateless
7. Readiness check: /ready endpoint validates Supabase connectivity

Why Some Services Are "Degraded" in Cloud:

In the HF Space, Redis and ChromaDB show as "offline" in the health check — this is expected. The cloud mode uses Supabase pgvector instead of ChromaDB, and Upstash Redis instead of local Redis. The /ready endpoint returns 200 as long as Supabase (the critical service) is connected.

---

⚙️ CI/CD — GitHub Actions

Three workflows run automatically:

Workflow	Trigger	What it does
CI Pipeline (ci.yml)	Push to main or PR	Install deps → Lint with Ruff → Run pytest (307 tests) → Upload failure logs
ML Training (train.yml)	Push to main	Train churn models → Log to DagsHub MLflow → Post CML report as PR comment with metrics
HF Deploy (hf_deploy.yml)	Push to main	Sync entire repo to Hugging Face Space → Triggers Docker rebuild → Live in ~2 min

Note: All 3 workflows trigger on every push to main, including README changes. The ML training re-run ensures model registry stays in sync. If you only want to update docs without triggering training, you can push to a non-main branch first.

---

🔍 Observability Stack

Langfuse — LLM Observability

Every LLM call is traced with a hierarchical structure:

Trace: "ticket-triage" (one per request)
  └── Span: "classify_agent"
       └── Generation: "gemma3-4b" (model, prompt, completion, tokens, cost, latency)
  └── Span: "rag_agent"
       └── Event: "rag_retrieval" (query, num_results, cache_hit)
       └── Generation: "llama-3.1-8b"
  └── Score: constitutional_compliance = 0.95
  └── Score: resolution_quality = 0.82
  └── Score: rag_grounding = 1.0

• PII is stripped from all trace data before sending to Langfuse
• LiteLLM integration: litellm.success_callback = ["langfuse"] auto-traces every LLM call — zero code changes needed
• Graceful degradation: when Langfuse keys are absent, NoOp stubs are used — tests pass offline

Prometheus + Grafana — System Metrics

docker compose -f docker-compose.yml -f docker-compose.monitoring.yml up -d
# Grafana: http://localhost:3001  |  Prometheus: http://localhost:9090

Pre-built dashboard panels:

• 📈 Triage ingestion rate — tickets/second (success vs pending vs error)
• ⏱️ p95 latency profiles — duration distribution by priority
• 🤖 LLM call reliability — volume by model type and status
• 💰 LLM cost tracker — cumulative API spend in USD
• 🎯 Semantic cache hit ratio — L1/L2 cache performance
• 🛑 HITL reason distribution — why tickets are being flagged

OpenTelemetry Collector

The OTel collector (infra/otel-collector.yaml) scrapes Prometheus metrics from the FastAPI /metrics endpoint and forwards them to Grafana Cloud for cloud-mode observability.

---

🌐 Multilingual Support

CustomerCore supports 4 languages out of the box: English, German, French, Spanish.

• Language detection: langdetect library identifies the ticket language
• Translation: Multilingual model translates non-English tickets before classification
• Response language: Constitutional AI policy ensures the response matches the ticket language
• Training data: Amazon MASSIVE dataset provides real customer utterances in DE/FR/ES (not translations — real native utterances)

---

🧪 Testing

Test Categories

Type	Files	What it covers
Unit Tests	13 files in tests/unit/	Every module: API routes, agents, RAG, vault, policy, router, multilingual, supervisor, streaming
Integration Tests	tests/integration/	End-to-end API flows with real HTTP calls
Adversarial Red Team	tests/adversarial_red_team.py	Prompt injection, jailbreak attempts, PII extraction attacks

Running Tests

pytest tests/ -v                    # All tests
pytest tests/unit/ -v               # Unit only
pytest tests/adversarial_red_team.py -v  # Security tests

---

🏗️ Infrastructure

Docker Compose Stack (Local)

docker compose up -d    # Start everything

Service	Port	Purpose
Redpanda	9092	Kafka-compatible message broker
Redpanda Console	8080	Visual dashboard for topics/messages
MinIO	9000/9001	S3-compatible object storage
ChromaDB	8000	Vector database for RAG
Redis	6379	Semantic cache + rate limiter
OTel Collector	4317/4318	Telemetry pipeline

Kubernetes (Production-Ready)

K8s manifests in infra/k8s/:

• namespace.yaml — customercore namespace
• fastapi-deployment.yaml — 2-replica deployment with health probes
• secrets.yaml — Doppler-injected secrets
• ingress.yaml — NGINX ingress for external access
• pdb.yaml — Pod Disruption Budget (min 1 available)

Can be deployed to a local Kind cluster (infra/kind-config.yaml) or any cloud K8s.

---

🔑 Secrets Management

All secrets are managed via Doppler (cloud secrets manager):

doppler run -- uvicorn src.api.main:app    # Injects all secrets as env vars
doppler run -- python src/ml/train_churn.py # Same for training

Required secrets (set in Doppler or as env vars):

• SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY
• OPENROUTER_API_KEY (for cloud LLM calls)
• LITELLM_MASTER_KEY (also used as JWT signing key)
• LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY
• MLFLOW_TRACKING_USERNAME, MLFLOW_TRACKING_PASSWORD (DagsHub)
• HF_TOKEN (for Hugging Face deployment)
• UPSTASH_REDIS_URL (cloud Redis)

---

📁 Project Structure

CustomerCore/
├── src/
│   ├── agent/                        # LangGraph multi-agent triage
│   │   ├── supervisor.py             # StateGraph: 6 nodes + HITL interrupt + MemorySaver
│   │   ├── state.py                  # AgentState TypedDict (shared state across all agents)
│   │   ├── schemas.py                # TriageOutput Pydantic model (validated output)
│   │   └── nodes/
│   │       ├── classify_agent.py     # Ticket categorization + priority assignment
│   │       ├── memory_agent.py       # Mem0 tenant-scoped customer memory recall
│   │       ├── rag_agent.py          # Hybrid retrieval + resolution generation
│   │       ├── churn_agent.py        # ML model inference for churn prediction
│   │       ├── incident_agent.py     # Outage spike detection
│   │       └── hitl_agent.py         # Human-in-the-loop gating logic
│   ├── api/                          # FastAPI application layer
│   │   ├── main.py                   # App factory, CORS, lifespan events
│   │   ├── auth.py                   # JWT verification + RBAC + tenant extraction
│   │   ├── ui.py                     # SPA dashboard (HTML/CSS/JS served from Python)
│   │   └── routers/
│   │       ├── triage.py             # POST /triage, GET /triage/{id}, POST /triage/{id}/resume
│   │       ├── health.py             # /health (liveness), /ready (readiness)
│   │       ├── metrics.py            # Prometheus /metrics endpoint
│   │       └── stream.py             # SSE streaming for real-time triage status
│   ├── rag/                          # Retrieval-Augmented Generation engine
│   │   ├── hybrid_retriever.py       # BM25 + ChromaDB + Reciprocal Rank Fusion + cross-encoder reranking
│   │   ├── graph_rag.py              # 3-layer Graph-RAG: Vector + NetworkX Graph + DuckDB SQL
│   │   ├── router.py                 # SLA-aware multi-model LLM router with cost tracking
│   │   ├── llm_client.py             # LiteLLM wrapper: Ollama (local) + OpenRouter (cloud)
│   │   └── multilingual.py           # Language detection + cross-lingual translation
│   ├── ml/
│   │   ├── train_churn.py            # Multi-model training + MLflow + auto-promote to Production
│   │   └── deploy_to_hf.py           # Automated Hugging Face Space code sync
│   ├── responsible_ai/
│   │   ├── privacy_vault.py          # AES-256-GCM PII masking (Presidio + spaCy NER)
│   │   ├── constitutional_policy.py  # 8-rule Constitutional AI safety engine
│   │   ├── audit_log.py              # Durable audit trail (every decision logged)
│   │   ├── key_manager.py            # Encryption key lifecycle management
│   │   └── model_cards/              # ML model documentation cards
│   ├── db/
│   │   ├── repository.py             # Supabase CRUD (tickets, customers, audit events)
│   │   └── migrations.py             # Auto-DDL: creates tables on startup if missing
│   ├── monitoring/
│   │   └── langfuse_tracer.py        # Full LLM observability: Trace → Span → Generation → Score
│   ├── streaming/
│   │   ├── data_loader.py            # Downloads 87K records from HuggingFace → Redpanda
│   │   ├── producers/                # Synthetic event generators for testing
│   │   ├── producer_helper.py        # Kafka producer wrapper
│   │   ├── bronze_consumer.py        # Raw event consumer from Redpanda
│   │   ├── bronze_to_silver.py       # PII masking + data cleaning pipeline
│   │   ├── triage_consumer.py        # Triggers agent pipeline from stream events
│   │   └── minio_setup.py            # S3 bucket initialization
│   └── dbt/                          # Data transformations (DuckDB adapter)
│       ├── dbt_project.yml
│       ├── profiles.yml
│       └── models/gold/              # 8 Gold mart models (health, funnel, agents, billing, etc.)
├── tests/
│   ├── unit/                         # 13 test files, 300+ tests
│   ├── integration/                  # E2E API tests
│   ├── adversarial_red_team.py       # Prompt injection + jailbreak test suite
│   └── conftest.py                   # Shared fixtures
├── infra/
│   ├── k8s/                          # Kubernetes: deployment, ingress, PDB, secrets
│   ├── monitoring/                   # Grafana dashboard JSON + Prometheus config
│   ├── kind-config.yaml              # Local multi-node K8s cluster
│   └── otel-collector.yaml           # OpenTelemetry pipeline config
├── .github/workflows/
│   ├── ci.yml                        # Lint (Ruff) + Test (pytest) pipeline
│   ├── train.yml                     # ML training + CML report
│   └── hf_deploy.yml                 # Auto-deploy to Hugging Face Spaces
├── docker-compose.yml                # Full local stack (Redpanda, MinIO, ChromaDB, Redis, OTel)
├── docker-compose.monitoring.yml     # Prometheus + Grafana
├── Dockerfile                        # Full production image
├── Dockerfile.hf                     # Slim image for Hugging Face Spaces
├── requirements.txt                  # Production dependencies
├── requirements-ci.txt               # CI/test dependencies (CPU-only PyTorch)
└── data/                             # DVC-tracked datasets (.dvc pointer files)

---

🛠️ Full Tech Stack

Category	Tools
Language	Python 3.12
API Framework	FastAPI, Uvicorn, Pydantic v2, SlowAPI (rate limiting)
AI Agents	LangGraph (StateGraph, MemorySaver, interrupt checkpoints)
LLM Orchestration	LangChain, LiteLLM (unified interface for local + cloud)
LLMs	Ollama (Gemma 3 4B — local), OpenRouter (Llama 3.1 8B — cloud)
ML Training	scikit-learn, MLflow, DagsHub, DVC
RAG	ChromaDB (dense), rank-bm25 (sparse), Reciprocal Rank Fusion, sentence-transformers (reranker)
Graph RAG	NetworkX (knowledge graph), DuckDB (SQL analytics)
NLP / PII	spaCy (NER), Microsoft Presidio (PII detection), langdetect
Database	Supabase (PostgreSQL + pgvector + RLS), SQLite (local fallback), DuckDB (analytics)
Auth	PyJWT (HS256), RBAC, Supabase Row-Level Security
Event Streaming	Redpanda (Kafka-compatible, no JVM, no ZooKeeper)
Object Storage	MinIO (local S3), DagsHub DVC remote (cloud)
Caching	Redis / Upstash Redis — L1 exact cache + L2 semantic cache (cosine similarity > 0.92)
Observability	Prometheus, Grafana, OpenTelemetry Collector, Langfuse
Infrastructure	Docker, Docker Compose, Kubernetes (Kind), Hugging Face Spaces
CI/CD	GitHub Actions, CML (Continuous Machine Learning), Ruff (linter)
Data Transforms	dbt (DuckDB adapter) — 8 Gold mart models
Security	AES-256-GCM encryption, Constitutional AI (8-rule safety engine), frontend RBAC enforcement
Secrets	Doppler (cloud secrets manager), Hugging Face Space Secrets
Logging	structlog (structured JSON logs)

---

🚀 Quick Start

1. Clone & Install

git clone https://github.com/saibalajinamburi/CustomerCore.git
cd CustomerCore
python -m venv .venv
.venv\Scripts\activate        # Windows
pip install -r requirements.txt

2. Set Environment Variables

Create a .env file or use Doppler:

# Required for cloud features (optional for local-only mode)
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_ANON_KEY=your-anon-key
LITELLM_MASTER_KEY=your-key      # Also used as JWT signing secret
OPENROUTER_API_KEY=your-key       # For cloud LLM calls

3. Start Local Infrastructure

docker compose up -d              # Redpanda, ChromaDB, Redis, MinIO, OTel

4. Run the API

uvicorn src.api.main:app --host 0.0.0.0 --port 8080 --reload

5. Open Dashboard

• Dashboard: http://localhost:8080
• Swagger API Docs: http://localhost:8080/docs

6. (Optional) Load Training Data

python -m src.streaming.data_loader --sources all --limit 500

7. (Optional) Train ML Models

doppler run -- python -X utf8 src/ml/train_churn.py

8. Run Tests

pytest tests/ -v

9. (Optional) Run Serverless GPU LLM Fine-Tuning

doppler run -- modal run src/ml/modal_train.py

---

⚠️ Challenges Faced & How We Solved Them

Challenge	Root Cause	Solution
Langfuse SDK v4 broke LiteLLM	LiteLLM v1.85 calls langfuse.version and Langfuse.trace() which don't exist in SDK v4	Wrote monkeypatches in langfuse_tracer.py that shim the old API onto the new SDK — mock_trace(), mock_span_method(), mock_generation_method()
CI failing on PyTorch	Full PyTorch is 2GB+ and slows CI to 15 min	Created requirements-ci.txt with CPU-only PyTorch from download.pytorch.org/whl/cpu
HF Space shows "degraded"	Health check expects Redis + ChromaDB locally, but cloud uses Upstash + pgvector	Modified /ready endpoint to return 200 as long as Supabase (critical path) is connected
Cross-tenant data leakage risk	Naive vector search returns results from all tenants	Enforced where={"tenant_id": X} on every ChromaDB query + physically partitioned BM25 index per tenant
PII in LLM traces	Langfuse receives full prompt text which could contain customer PII	Added _mask_pii() regex pipeline that strips emails, phones, SSNs, IBANs, credit cards before any data leaves the process
UTF-8 encoding on Windows	Python on Windows defaults to CP1252 — MLflow logging fails with emoji/special chars	Set PYTHONIOENCODING=utf-8 and use python -X utf8 flag
CML reports on every push	ML training workflow triggers even on README changes	Accepted trade-off: ensures model registry is always in sync. Can be scoped to src/ml/** paths if needed
Database schema drift	New columns added in code but HF Space container has old schema	Added migrations.py that runs ALTER TABLE ADD COLUMN IF NOT EXISTS on every startup
Supabase direct port 5432 blocked in cloud	IPv6-only resolution on direct PostgreSQL connections times out on HF Spaces (IPv4 only)	Switched migrations to Supabase Connection Pooler on port 6543 — IPv4-compatible, stable in all cloud environments
Ollama unavailable in cloud containers	HF Spaces don't ship with Ollama — local model calls time out after several seconds	LLMClient detects SPACE_ID / APP_ENV=production at startup and transparently redirects local tier calls to openrouter/google/gemini-2.5-flash
Stale HITL data persists after role switch	Switching from manager to support_agent in the demo left previously fetched review rows visible	Added an immediate role guard in generateToken() that clears the HITL table and shows an "Access Denied" banner the instant the role changes to support_agent
Misleading DB timeout error messages	When Supabase was unreachable, the UI catch block printed "Ensure role is manager" — wrong root cause	Updated catch blocks to distinguish NetworkError / ENOTFOUND from auth errors and display an accurate "Database unreachable" message with guidance
Gated base model downloads on HF	Hugging Face gated models (like meta-llama/Meta-Llama-3-8B-Instruct) returned a 403 Forbidden error on serverless containers without user authorization	Switched base model to NousResearch/Meta-Llama-3-8B-Instruct, a public non-gated clone with identical architecture and weights
Missing dependency in Python 3.12 container	The remote training process crashed with ModuleNotFoundError: No module named 'setuptools'	Added setuptools to the container pip_install settings in modal_train.py for Python 3.12 compatibility
Schema mapping & status issues in training script	Direct queries failed due to non-existent columns (text) and incorrect status mapping (completed vs check constraints)	Replaced the column selection with raw_text and changed the status filter criteria to 'complete'

---

🔒 EU AI Act Compliance

Article	Requirement	Implementation
Art. 10	Data governance & PII masking	Privacy Vault: Presidio + spaCy masks all PII before storage/LLM calls. AES-256-GCM reversible tokenization
Art. 12	Audit traceability	Every decision logged to audit_events table with timestamp, tenant, input, output, model used
Art. 14	Human oversight	LangGraph interrupt checkpoint pauses high-risk decisions for human review before execution
Art. 15	Fairness & security	Adversarial red-team tests, model cards, 8-rule Constitutional AI safety engine, RBAC

---

📄 License

MIT License — see LICENSE for details.