DEV Community: Sunil Kumar

4 Engineering Patterns That Cut AI Inference Costs 60–80% Without Touching Output Quality

Sunil Kumar — Mon, 20 Apr 2026 04:21:36 +0000

One of the most common engineering problems I see when teams scale AI from pilot to production: the cloud bill is 5–10× what the pilot suggested it would be.

This isn't a coincidence or a billing error. It's the predictable result of pilot conditions not reflecting production conditions — cleaner inputs, smaller volumes, no monitoring overhead, no retry logic, no agent orchestration multiplication.

Here are the 4 implementation patterns that fix it. All of them can be implemented without changing the AI output quality for the user.

Pattern 1: Semantic Caching

The cheapest LLM call is one you don't make. Semantic caching returns cached responses for queries that are semantically similar to ones you've already answered.

For enterprise AI tools where users ask similar questions repeatedly (HR policy lookup, product specs, internal knowledge queries), cache hit rates of 20–30% are achievable. Each hit is zero cost.

import numpy as np
from openai import OpenAI
from typing import Optional
import hashlib
import json

client = OpenAI()

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.92):
        self.threshold = similarity_threshold
        self.cache: list[dict] = []  # In production: use Redis or Pinecone

    def _embed(self, text: str) -> list[float]:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        a, b = np.array(a), np.array(b)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

    def get(self, query: str) -> Optional[str]:
        if not self.cache:
            return None
        query_embedding = self._embed(query)
        for entry in self.cache:
            similarity = self._cosine_similarity(query_embedding, entry["embedding"])
            if similarity >= self.threshold:
                return entry["response"]  # Cache hit — zero LLM cost
        return None

    def set(self, query: str, response: str) -> None:
        self.cache.append({
            "query": query,
            "embedding": self._embed(query),
            "response": response
        })

# Usage
cache = SemanticCache(similarity_threshold=0.92)

def cached_llm_call(query: str, system_prompt: str) -> dict:
    cached = cache.get(query)
    if cached:
        return {"response": cached, "source": "cache", "cost": 0.0}

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": query}
        ]
    )
    answer = response.choices[0].message.content
    cache.set(query, answer)

    # Calculate approximate cost
    input_tokens = response.usage.prompt_tokens
    output_tokens = response.usage.completion_tokens
    cost = (input_tokens * 0.0000025) + (output_tokens * 0.00001)  # GPT-4o pricing

    return {"response": answer, "source": "llm", "cost": cost}

Expected savings: 15–30% of LLM calls, depending on query repetition rate.

Pattern 2: Query-Complexity-Based Model Routing

Not every query needs GPT-4o. Simple classification, template filling, and routine Q&A can run on smaller models with no quality difference for the user.

from enum import Enum
from dataclasses import dataclass

class QueryComplexity(Enum):
    SIMPLE = "simple"       # → gpt-4o-mini or similar  (~10x cheaper)
    MODERATE = "moderate"   # → gpt-4o
    COMPLEX = "complex"     # → gpt-4o or o1-preview for reasoning

@dataclass
class RoutingDecision:
    complexity: QueryComplexity
    model: str
    rationale: str

class ModelRouter:
    # Signals that suggest higher complexity
    COMPLEX_SIGNALS = [
        "analyse", "compare", "synthesise", "evaluate", "reason",
        "explain why", "what are the implications", "trade-off",
        "multi-step", "infer", "hypothesis"
    ]

    SIMPLE_SIGNALS = [
        "what is", "define", "list", "when", "who", "where",
        "how many", "summarise this", "translate", "format"
    ]

    def route(self, query: str, context_length: int = 0) -> RoutingDecision:
        query_lower = query.lower()

        complex_score = sum(1 for s in self.COMPLEX_SIGNALS if s in query_lower)
        simple_score = sum(1 for s in self.SIMPLE_SIGNALS if s in query_lower)

        # Long context = more likely to need powerful model
        if context_length > 8000 or complex_score >= 2:
            return RoutingDecision(
                complexity=QueryComplexity.COMPLEX,
                model="gpt-4o",
                rationale=f"Complex signals: {complex_score}, context: {context_length}"
            )
        elif simple_score >= 2 and complex_score == 0 and context_length < 2000:
            return RoutingDecision(
                complexity=QueryComplexity.SIMPLE,
                model="gpt-4o-mini",
                rationale=f"Simple signals: {simple_score}"
            )
        else:
            return RoutingDecision(
                complexity=QueryComplexity.MODERATE,
                model="gpt-4o-mini",  # Default to cheaper model for moderate
                rationale="Moderate complexity"
            )

def routed_llm_call(query: str, context: str, system_prompt: str) -> dict:
    router = ModelRouter()
    decision = router.route(query, len(context))

    response = client.chat.completions.create(
        model=decision.model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"{context}\n\nQuery: {query}"}
        ]
    )

    return {
        "response": response.choices[0].message.content,
        "model_used": decision.model,
        "routing_rationale": decision.rationale,
        "tokens_used": response.usage.total_tokens
    }

Expected savings: 30–40% of inference costs for mixed-complexity query distributions.

Pattern 3: Prompt Compression Measurement

Before compressing, measure. The goal is to identify which parts of your prompts are high-information vs redundant.

def measure_prompt_efficiency(
    original_prompt: str,
    compressed_prompt: str,
    test_queries: list[str],
    system_prompt: str
) -> dict:
    """
    Measure: does the compressed prompt produce equivalent outputs?
    Run BEFORE switching to compressed prompt in production.
    """
    original_tokens = len(original_prompt.split()) * 1.3  # rough token estimate
    compressed_tokens = len(compressed_prompt.split()) * 1.3
    token_reduction = (original_tokens - compressed_tokens) / original_tokens

    quality_scores = []
    for query in test_queries[:20]:  # sample evaluation
        orig_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": original_prompt + "\n\n" + query}
            ]
        ).choices[0].message.content

        comp_response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": compressed_prompt + "\n\n" + query}
            ]
        ).choices[0].message.content

        # Simple length/structure similarity as proxy — replace with semantic similarity
        # in production evaluation
        length_ratio = len(comp_response) / max(len(orig_response), 1)
        quality_scores.append(min(length_ratio, 1/length_ratio))

    avg_quality = sum(quality_scores) / len(quality_scores)

    return {
        "token_reduction_pct": f"{token_reduction:.1%}",
        "estimated_monthly_savings_at_1M_calls": f"${token_reduction * 1000 * 0.0000025 * 1_000_000:.0f}",
        "avg_quality_retention": f"{avg_quality:.2%}",
        "recommendation": "SAFE TO DEPLOY" if avg_quality > 0.85 else "REVIEW BEFORE DEPLOYING"
    }

Expected savings: 20–30% of token costs with well-compressed prompts. Run evaluation before switching.

Pattern 4: Cost Monitoring Dashboard (Catch Spiral Early)

The most important pattern: know your cost per query, per model, per feature — in real time.

from datetime import datetime
from collections import defaultdict

class AIcostMonitor:
    # Pricing per 1K tokens (update with current pricing)
    MODEL_COSTS = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "text-embedding-3-small": {"input": 0.00002, "output": 0.0}
    }

    def __init__(self, alert_threshold_daily_usd: float = 100.0):
        self.calls: list[dict] = []
        self.alert_threshold = alert_threshold_daily_usd

    def log_call(self, model: str, input_tokens: int, 
                 output_tokens: int, feature: str):
        pricing = self.MODEL_COSTS.get(model, {"input": 0.003, "output": 0.012})
        cost = (input_tokens/1000 * pricing["input"]) + \
               (output_tokens/1000 * pricing["output"])

        self.calls.append({
            "timestamp": datetime.utcnow().isoformat(),
            "model": model,
            "feature": feature,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost
        })

    def daily_summary(self) -> dict:
        today = datetime.utcnow().date().isoformat()
        today_calls = [c for c in self.calls if c["timestamp"].startswith(today)]

        by_feature = defaultdict(float)
        by_model = defaultdict(float)
        total = sum(c["cost_usd"] for c in today_calls)

        for call in today_calls:
            by_feature[call["feature"]] += call["cost_usd"]
            by_model[call["model"]] += call["cost_usd"]

        alert = total > self.alert_threshold
        return {
            "date": today,
            "total_cost_usd": round(total, 4),
            "alert": alert,
            "alert_message": f"ALERT: Daily cost ${total:.2f} exceeds threshold ${self.alert_threshold}" if alert else None,
            "by_feature": dict(sorted(by_feature.items(), key=lambda x: -x[1])),
            "by_model": dict(sorted(by_model.items(), key=lambda x: -x[1])),
            "projected_monthly": round(total * 30, 2)
        }

The Decision Tree

Query arrives
    │
    ├── Check semantic cache
    │       ├── HIT → return cached response (cost: $0)
    │       └── MISS → continue
    │
    ├── Route by complexity
    │       ├── SIMPLE → gpt-4o-mini (10x cheaper)
    │       ├── MODERATE → gpt-4o-mini
    │       └── COMPLEX → gpt-4o
    │
    ├── Apply compressed prompt template
    │
    ├── Call LLM
    │
    ├── Log to cost monitor
    │
    └── Store in cache for future hits

Implement these 4 patterns in sequence. Start with the monitor (you can't optimise what you can't measure), then caching (biggest immediate impact), then routing, then prompt compression.

What's your current cost per 1,000 queries in production? Curious what numbers people are seeing across different use cases.

Sunil — CEO, Ailoitte. We build cost-optimised AI architectures for funded startups. ailoitte.com

How to Build a Secure Internal AI Assistant That Your Security Team Will Actually Approve

Sunil Kumar — Wed, 15 Apr 2026 05:45:14 +0000

Every enterprise security team is dealing with the same problem: employees are using ChatGPT with company data, and IT has no visibility, no control, and no audit trail.

Banning doesn't work. Employees use the tools that make them productive. If you ban ChatGPT without providing a better alternative, you create a two-tier workforce — the rule-followers who fall behind, and the rule-breakers who deliver.

The actual solution: build an internal AI assistant that is more useful to employees than the public tools they're currently using, deployed with the security controls that make your CISO comfortable.

Here's the full technical architecture.

Why Internal AI Beats Public AI for Employees

ChatGPT answers general questions from general training data. An internal AI answers company-specific questions from company-specific data.

When an employee asks ChatGPT, "What's our refund policy?", they get a generic answer. When they ask an internal AI trained on your actual refund policy docs, they get the exact policy with the relevant exceptions.

This is why employees switch voluntarily. Not policy compliance — better tooling. And better tooling that doesn't require them to paste confidential data into an external service.

The Core Architecture

Employee Query
     │
     ▼
┌─────────────────────────────────┐
│  Company SSO / Auth Layer        │  ← SAML, OAuth2, Azure AD
│  (only authenticated users)     │
└─────────────────────────────────┘
     │
     ▼
┌─────────────────────────────────┐
│  PII Detection & Redaction       │  ← Before query leaves internal network
│  (scan for names, IDs, PHI)     │
└─────────────────────────────────┘
     │ (clean query)
     ▼
┌─────────────────────────────────┐
│  Internal RAG Pipeline           │
│  ├── Vector search (internal docs)│
│  ├── Keyword search (policies)   │
│  └── Structured data lookup     │
└─────────────────────────────────┘
     │ (relevant context)
     ▼
┌─────────────────────────────────┐
│  LLM Inference                   │  ← Hosted internally OR private API
│  (with retrieved context)        │  ← Never raw company data to external
└─────────────────────────────────┘
     │
     ▼
┌─────────────────────────────────┐
│  Audit Logging                   │  ← User ID, query hash, response,
│                                  │     timestamp, data sources accessed
└─────────────────────────────────┘
     │
     ▼
  Response to Employee

1. Auth Integration (Non-Negotiable)

The AI assistant must authenticate through your existing identity provider — no separate accounts, no shared credentials.

from fastapi import FastAPI, Depends, HTTPException
from fastapi. security import OAuth2AuthorizationCodeBearer
import jwt

app = FastAPI()
oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl="https://login.microsoftonline.com/tenant/oauth2/v2.0/authorize",
    tokenUrl="https://login.microsoftonline.com/tenant/oauth2/v2.0/token"
)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, options={"verify_signature": False})
        user_id = payload.get("oid")  # Azure AD object ID
        department = payload.get("department", "unknown")
        return {"user_id": user_id, "department": department}
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid authentication")

@app.post("/query")
async def process_query(
    query: str,
    current_user: dict = Depends(get_current_user)
):
    # Only authenticated users reach this point
    # User context is available for permission-scoping
    return await handle_query(query, current_user)

2. PII Detection Before Any External Call

If you're using a hosted LLM API (OpenAI, Anthropic, etc.), scan every query for PII before it leaves your network. For regulated industries, this is mandatory.

import re
from dataclasses import dataclass
from typing import List

@dataclass
class PIIDetectionResult:
    has_pii: bool
    pii_types: List[str]
    redacted_text: str

class PIIDetector:
    PATTERNS = {
        "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
        "phone": r'\b(\+\d{1,2}\s?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
        "credit_card": r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
        "patient_id": r'\b(MRN|DOB|Patient\s*ID)[:\s]*\S+',  # Healthcare-specific
    }

    def detect_and_redact(self, text: str) -> PIIDetectionResult:
        redacted = text
        found_types = []

        for pii_type, pattern in self.PATTERNS.items():
            matches = re.findall(pattern, text, re.IGNORECASE)
            if matches:
                found_types.append(pii_type)
                redacted = re.sub(pattern, f"[{pii_type.upper()}_REDACTED]", 
                                  redacted, flags=re.IGNORECASE)

        return PIIDetectionResult(
            has_pii=bool(found_types),
            pii_types=found_types,
            redacted_text=redacted
        )

# Usage: scan every query before external API call
detector = PIIDetector()
result = detector.detect_and_redact(user_query)

if result.has_pii:
    # Either block the query, use a redacted version, or route to on-premise model
    query_to_llm = result.redacted_text
    log_pii_detection(current_user, result.pii_types)
else:
    query_to_llm = user_query

3. Department-Scoped RAG

Employees should only get answers from documents they're authorised to access. A sales employee shouldn't retrieve HR disciplinary records. A junior employee shouldn't retrieve board-level financial projections.

class ScopedVectorStore:
    def __init__(self, base_store):
        self.store = base_store

    def search(self, query: str, user: dict, k: int = 5):
        # Build access filter based on user's department and role
        access_filter = self._build_access_filter(user)

        # Only retrieve documents the user is authorised to see
        return self.store.similarity_search(
            query,
            k=k,
            filter=access_filter
        )

    def _build_access_filter(self, user: dict) -> dict:
        # Documents tagged with department restrictions
        # are only returned to users in that department
        return {
            "$or": [
                {"access": "all"},  # Public company docs
                {"access": user["department"]},  # Dept-specific docs
                {"access": f"role:{user.get('role', 'employee')}"}  # Role-based
            ]
        }

4. Audit Logging (Required for Compliance)

Every query must be logged — user, timestamp, query (or hash), data sources accessed, response. This creates the audit trail that HIPAA, GDPR, and SOC 2 require.

import hashlib
import json
from datetime import datetime

class AuditLogger:
    def __init__(self, log_store):
        self.store = log_store

    def log_query(self, user: dict, query: str, 
                  sources: list, response: str, pii_detected: bool):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user["user_id"],
            "department": user["department"],
            "query_hash": hashlib.sha256(query.encode()).hexdigest(),
            # Store hash, not raw query — protects employee privacy
            # while maintaining audit trail
            "sources_accessed": [s["doc_id"] for s in sources],
            "pii_detected": pii_detected,
            "response_length": len(response),
            "session_id": user.get("session_id")
        }
        self.store.append(entry)

    def get_compliance_report(self, start_date: str, end_date: str) -> dict:
        """Produces the report your compliance team will request"""
        entries = self.store.get_range(start_date, end_date)
        return {
            "total_queries": len(entries),
            "unique_users": len(set(e["user_id"] for e in entries)),
            "pii_detection_events": sum(1 for e in entries if e["pii_detected"]),
            "top_document_categories": self._aggregate_sources(entries)
        }

The 3-Week Build Sequence

Week 1: Auth integration + document ingestion + basic RAG pipeline. First queries will be working by the end of the week.

Week 2: PII detection + department-scoped access + audit logging. Security controls complete.

Week 3: UI/UX polish + Slack/Teams integration + user training. Deployed to the pilot group.

The measure of success: employees in the pilot group reduce their ChatGPT usage because the internal tool gives better answers to their actual work questions. Voluntary adoption is the only adoption that lasts.

Sunil — CEO, Ailoitte. We build secure internal AI assistants for regulated industries: 3–4 week deployment, SSO integration, department-scoped access, full audit logging. ailoitte.com

Why Multi-Agent Systems Break in Production, and the 5 Architecture Patterns That Prevent It

Sunil Kumar — Tue, 14 Apr 2026 05:58:05 +0000

I've now debugged enough agent systems that broke in production to see the pattern clearly.

The agent logic is rarely the problem. The prompts work. The individual agents return reasonable outputs when tested in isolation.

The problem is the layer between agents, the orchestration, the shared state management, the failure handling, and the observability. These are the layers most teams don't build until they're already on fire.

Here's what breaks, and the code patterns that prevent it.

Failure Mode 1: Race Conditions in Shared State

Multi-agent systems often share a state object that multiple agents read from and write to. Without explicit locking or immutable state patterns, concurrent agents corrupt each other's context.

import asyncio
from typing import Any
from dataclasses import dataclass, field

# BAD: Shared mutable state — race condition waiting to happen
class AgentState:
    context: dict = {}

# GOOD: Immutable state with explicit update pattern
@dataclass(frozen=True)  
class AgentState:
    context: dict = field(default_factory=dict)
    agent_id: str = ""
    timestamp: float = 0.0

    def with_update(self, key: str, value: Any) -> 'AgentState':
        """Returns new state — never mutates existing"""
        return AgentState(
            context={**self.context, key: value},
            agent_id=self.agent_id,
            timestamp=asyncio.get_event_loop().time()
        )

# State transitions tracked explicitly — no silent overwrites
class StateManager:
    def __init__(self):
        self._states: list[AgentState] = []
        self._lock = asyncio.Lock()

    async def update(self, agent_id: str, key: str, value: Any) -> AgentState:
        async with self._lock:  # Explicit locking on every write
            current = self._states[-1] if self._states else AgentState()
            new_state = current.with_update(key, value)
            self._states.append(new_state)
            return new_state

    def get_trace(self) -> list[AgentState]:
        """Full state history for debugging — not just current state"""
        return self._states.copy()

Failure Mode 2: Cascading Failures Without Circuit Breakers

When Agent A produces low-quality output, and Agent B consumes it as if it were correct, the error compounds. By the time Agent C acts on Agent B's output, the original failure is invisible.

from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable, Any

class OutputQuality(Enum):
    HIGH = "high"       # > 0.85 confidence
    MEDIUM = "medium"   # 0.65–0.85
    LOW = "low"         # < 0.65

@dataclass
class AgentOutput:
    content: str
    quality: OutputQuality
    confidence: float
    agent_id: str

class CircuitBreaker:
    """Prevents cascading failures by blocking low-quality outputs"""

    def __init__(self, min_quality: OutputQuality = OutputQuality.MEDIUM):
        self.min_quality = min_quality
        self._failure_count = 0
        self._threshold = 3  # Open after 3 consecutive failures
        self._is_open = False

    def check(self, output: AgentOutput) -> bool:
        if self._is_open:
            raise RuntimeError(f"Circuit breaker open: too many low-quality outputs from {output.agent_id}")

        if output.quality == OutputQuality.LOW:
            self._failure_count += 1
            if self._failure_count >= self._threshold:
                self._is_open = True
            return False  # Don't pass this output downstream

        self._failure_count = 0  # Reset on success
        return True

def with_quality_gate(min_quality: OutputQuality = OutputQuality.MEDIUM):
    """Decorator: validates output quality before passing to next agent"""
    def decorator(agent_fn: Callable) -> Callable:
        breaker = CircuitBreaker(min_quality)
        async def wrapper(*args, **kwargs) -> Optional[AgentOutput]:
            output = await agent_fn(*args, **kwargs)
            if not breaker.check(output):
                # Explicit fallback — not silent failure
                return AgentOutput(
                    content="[Low confidence — human review required]",
                    quality=OutputQuality.LOW,
                    confidence=output.confidence,
                    agent_id=output.agent_id
                )
            return output
        return wrapper
    return decorator

Failure Mode 3: Silent Errors — No Observability

The single most dangerous failure mode: everything returns success, but the final output is wrong. Nothing alerts. Users discover it.

import time
import uuid
from contextlib import asynccontextmanager

class AgentTracer:
    """Distributed tracing for multi-agent systems"""

    def __init__(self, trace_id: str = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.spans: list[dict] = []

    @asynccontextmanager
    async def span(self, agent_id: str, operation: str):
        span_id = str(uuid.uuid4())
        start_time = time.time()
        span_data = {
            "span_id": span_id,
            "trace_id": self.trace_id,
            "agent_id": agent_id,
            "operation": operation,
            "start_time": start_time,
            "status": "running"
        }
        self.spans.append(span_data)

        try:
            yield span_data
            span_data["status"] = "success"
        except Exception as e:
            span_data["status"] = "error"
            span_data["error"] = str(e)
            raise
        finally:
            span_data["duration_ms"] = (time.time() - start_time) * 1000

    def get_full_trace(self) -> dict:
        """Everything every agent did — for debugging what went wrong"""
        return {
            "trace_id": self.trace_id,
            "total_duration_ms": sum(s["duration_ms"] for s in self.spans if "duration_ms" in s),
            "agent_calls": len(self.spans),
            "errors": [s for s in self.spans if s["status"] == "error"],
            "spans": self.spans
        }

Failure Mode 4: Infinite Loops Without Termination Conditions

from typing import Optional
import asyncio

async def run_agent_with_limits(
    agent_fn: Callable,
    max_iterations: int = 10,
    timeout_seconds: float = 30.0,
    termination_check: Optional[Callable] = None
) -> AgentOutput:
    """Always has a way out — no infinite loops"""

    for iteration in range(max_iterations):
        try:
            result = await asyncio.wait_for(
                agent_fn(),
                timeout=timeout_seconds
            )

            # Check termination condition if provided
            if termination_check and termination_check(result):
                return result

            # If no termination check, return on first success
            if not termination_check:
                return result

        except asyncio.TimeoutError:
            if iteration == max_iterations - 1:
                return AgentOutput(
                    content="[Timeout — escalating to human review]",
                    quality=OutputQuality.LOW,
                    confidence=0.0,
                    agent_id="timeout_handler"
                )

    # Hit max iterations without satisfying termination condition
    return AgentOutput(
        content="[Max iterations reached — escalating]",
        quality=OutputQuality.LOW,
        confidence=0.0,
        agent_id="iteration_limit_handler"
    )

The Human-in-the-Loop Escalation Pattern

from typing import Callable, Optional

class HumanInTheLoop:
    """Defines exactly when and how humans enter the agent workflow"""

    def __init__(self, 
                 confidence_threshold: float = 0.70,
                 escalation_fn: Callable = None):
        self.threshold = confidence_threshold
        self.escalation_fn = escalation_fn or self._default_escalation

    async def check(self, output: AgentOutput) -> AgentOutput:
        if output.confidence < self.threshold:
            await self.escalation_fn(output)
            return AgentOutput(
                content=f"[Pending human review — confidence {output.confidence:.0%}]",
                quality=OutputQuality.LOW,
                confidence=output.confidence,
                agent_id="human_in_the_loop"
            )
        return output

    async def _default_escalation(self, output: AgentOutput):
        # Log, alert, create ticket — whatever your workflow requires
        print(f"ESCALATION: Agent {output.agent_id} confidence {output.confidence:.0%} below threshold")

The Production Readiness Checklist

Before shipping any multi-agent system to production:

□ Immutable or explicitly locked shared state
□ Circuit breakers between the agent output and the next agent input  
□ Distributed tracing covering every agent call
□ Max iteration + timeout limits on every agent
□ Explicit fallback output for every failure mode
□ Human-in-the-loop escalation with defined confidence thresholds
□ Output validation schema before any agent result is served to users
□ Load testing at 5× expected concurrent agent invocations

If any of these are missing, do not ship. Add it in week 1 of the project, not week 10 of debugging.

Ailoitte builds production-grade agent systems for funded startups, orchestration, observability, guardrails, and human-in-the-loop. 6–8 week delivery. ailoitte.com

A Technical Framework for Deciding Which Business Workflows to Automate with AI First

Sunil Kumar — Mon, 13 Apr 2026 07:21:31 +0000

Every week, I talk to engineering leads who've been handed the same brief: "Figure out where we should be using AI."

It sounds like a strategy question. It's actually an engineering question in disguise — because the right answer depends on factors that non-technical leadership can't easily evaluate: data availability, task structure, output verifiability, integration complexity.

Here's the framework we use to answer it systematically.

The 4-Dimension Workflow Scoring Matrix

For any workflow you're evaluating, score it on these 4 dimensions on a 1–5 scale:

workflow_score = {
    "task_structure": 0,       # How structured and repeatable is the task?
    "data_availability": 0,    # Do we have historical data to train/guide the AI?
    "output_verifiability": 0, # Can we verify if the AI output is correct without deep expertise?
    "volume_frequency": 0,     # How often does this task occur?
}

# Scoring rubric
TASK_STRUCTURE_GUIDE = {
    1: "Highly ambiguous — requires accumulated domain judgment",
    3: "Mixed — structured core with some judgment calls",
    5: "Fully structured — clear decision criteria or rule-based."
}

DATA_AVAILABILITY_GUIDE = {
    1: "No historical data exists",
    3: "Some data, partially structured",
    5: "Rich historical data, well-labelled, production-quality."
}

OUTPUT_VERIFIABILITY_GUIDE = {
    1: "Near impossible to verify without deep domain expertise",
    3: "Can sample-review ~20% of outputs",
    5: "Automated verification possible (format, schema, rules)."
}

VOLUME_FREQUENCY_GUIDE = {
    1: "< 10 instances/month",
    3: "50–200 instances/month",
    5: "> 1000 instances/month or continuous."
}

# Weighted final score
def ai_readiness_score(scores):
    weights = {
        "task_structure": 0.30,
        "data_availability": 0.30,
        "output_verifiability": 0.25,
        "volume_frequency": 0.15
    }
    return sum(scores[k] * weights[k] for k in scores)

# Interpretation
# 3.5+ → Strong AI candidate. Start here.
# 2.5–3.5 → Viable, but address the low-scoring dimension first
# < 2.5 → Not yet ready. Data or structure work is needed before AI.

Example: Scoring 5 Common Workflows

Workflow                    | Structure | Data | Verify | Volume | Score
---------------------------|-----------|------|--------|--------|------
Invoice data extraction    |     5     |  4   |   5    |   4    |  4.55  ← Start here
Customer support triage    |     4     |  4   |   3    |   5    |  3.85  ← Strong
Employee onboarding Q&A    |     4     |  3   |   4    |   3    |  3.55  ← Good
Contract risk flagging     |     3     |  3   |   2    |   3    |  2.80  ← Later
Strategic pricing decisions|     1     |  2   |   1    |   2    |  1.45  ← Not AI

Invoice extraction and support triage score the highest. Contract risk flagging is interesting, but the output verifiability gap makes it premature. Strategic pricing stays with humans.

The Pre-Build Data Readiness Checklist

Before scoping any AI implementation, verify:

data_readiness = {
    "volume": "500+ historical examples of the task completed?",
    "quality": "Historical data consistent and trustworthy (not riddled with exceptions)?",
    "labels": "Do we have correct outputs for historical inputs (for supervised approaches)?",
    "recency": "Data representative of current conditions (not stale from 2+ years ago)?"
}
# Any 'No': address the data problem before scoping the AI.
# AI quality is bounded by training/retrieval data quality — no model selection fixes bad data.

The Two Infrastructure Questions Nobody Asks Until Week 8

Where does the AI output go?

If outputs need to integrate with a legacy system (SAP, Salesforce, or an older ERP), the integration layer is typically 40–60% of the total build effort. Map downstream systems before scoping the AI component.

What happens when the AI is wrong?

Every AI system has a failure rate. Define your acceptable error rate before building. What does a wrong output cost? Can you spot-check a sample? How do you handle low-confidence outputs? These are week-1 architecture decisions, not week-8 discoveries.

The One-Page Prioritisation Output

After scoring every workflow:

AI READINESS SCORECARD — [Company Name]
=========================================
TOP 3 AI PRIORITY WORKFLOWS:
1. Invoice Data Extraction        Score: 4.55  ROI: HIGH
   Data ready: YES  Integration: 3 weeks  First value: 4 weeks

2. Customer Support Triage        Score: 3.85  ROI: MEDIUM-HIGH  
   Data ready: PARTIAL  Gap: Need 200 labelled examples

3. Employee Onboarding Q&A        Score: 3.55  ROI: MEDIUM
   Data ready: YES (Notion + Confluence)  Integration: 2 weeks

NOT YET READY (address these data gaps first):
- Contract Risk Flagging (output verifiability gap)
- Strategic Pricing (judgment-based, insufficient structure)

Run this before you talk to any AI vendor. Know your scores before you know the solution.

How We Build Production-Grade AI MVPs in 8 Weeks: The Technical Sprint Framework

Sunil Kumar — Thu, 09 Apr 2026 05:46:32 +0000

I built Ailoitte because I watched too many good startup ideas die in development.

Not because the founders had bad ideas. Not because the market wasn't ready. Because the build took too long, burned too much cash, and by the time the MVP was live, if it ever got live, the runway was gone.

The US agency model for startup MVPs is structurally broken. $170K–$280K and 12–16 weeks for a first version that founders then have to maintain themselves. That timeline kills startups that would have survived if they'd shipped 6 weeks earlier.

So I want to share exactly how we built a model that gets production-grade AI-powered MVPs live in 8 weeks, not as marketing copy, but as the actual engineering and process structure we use.

The Core Problem: Sequential Workflows Kill Timeline

Traditional agency builds run sequentially:

Discovery → Design → Architecture → Frontend → Backend → Integration → QA → Deploy

Each phase waits for the previous one. Handoffs create delays. Requirements change mid-phase. Scope grows. Timeline extends.

At Ailoitte, we run a parallel sprint model:

Week 1: Architecture + Scope lock (all roles simultaneously)
Weeks 2–6: Parallel build tracks
  Track A: Core backend + AI layer
  Track B: Frontend + UX
  Track C: Data pipeline + integrations
  (Daily integration points prevent divergence)
Week 7: Integration + QA + Load testing
Week 8: Production deployment + handover

Everything runs simultaneously from week 2. The daily integration point catches conflicts before they become expensive.

The AI-First Development Stack That Makes 8 Weeks Possible

This is the part most people are skeptical about. "3× velocity" sounds like marketing. Here's the technical reality:

Code generation with structured agents:

# Our internal tooling structure (simplified)
class SprintAgent:
    def generate_component(self, spec: ComponentSpec, context: CodebaseContext) -> GeneratedCode:
        """
        Takes a scoped spec + existing codebase context
        Returns: implementation + tests + documentation

        Human review gates: architecture decisions, security-critical paths, 
        third-party integrations, data model changes

        AI handles: boilerplate, CRUD operations, standard API patterns,
        test generation, documentation, refactoring
        """

The key: AI handles high-spec, low-judgment work. Humans own the judgment calls. The split for a typical MVP:

AI-handled (3× velocity applicable): ~65% of total code volume
Human-owned (standard velocity): ~35% — the architectural decisions, security logic, complex business rules

Blended velocity: approximately 2.8–3.2× vs traditional workflow. That's where the 8-week timeline comes from.

AI-assisted code review:

Every PR goes through an AI pre-review before human review. This catches:

Security anti-patterns
Performance issues
Consistency violations
Missing error handling

Human reviewers see cleaner PRs. Review time drops by ~40%.

The Scope Lock Protocol — Week 1 Non-Negotiable

The most expensive mistake in MVP builds: scope changes in week 5.

Week 1 at Ailoitte has a specific output: a locked scope document that includes:

## Week 1 Deliverables (non-negotiable before Week 2 build starts)

### Core user journeys (maximum 3 for an 8-week MVP)
- Journey 1: [specific user action → specific system response]
- Journey 2: ...
- Journey 3: ...

### Technical stack decision (final)
- Frontend: [decision + rationale]
- Backend: [decision + rationale]  
- AI layer: [model(s) + rationale]
- Infrastructure: [cloud + services]

### Out of scope (explicitly listed)
- [Feature 1]: Phase 2
- [Feature 2]: Phase 2

### Definition of done
- Users can complete all 3 core journeys without error
- P95 latency ≤ [agreed threshold]ms at [agreed] concurrent users
- Deployed to production, not staging

When scope creep appears in week 4 (it always does), the locked scope document is the reference. "That's Phase 2" has documentation behind it.

What "Production-Grade" vs "Demo-Grade" Actually Means

The deliverable is not a prototype or a polished demo. By the end of week 8:

Requirement	Our standard
Error handling	Every failure path has a designed user experience
Load testing	Tested at 2× expected peak traffic
Monitoring	Uptime + error rate + key business metrics alerting
Auth	Production auth with proper session management
Data	Proper backup, retention, and privacy policies
Documentation	API docs, deployment runbook, architecture diagram

A US agency might deliver some of these. The 8-week sprint delivers all of them because they're scoped into week 1.

The Daily Standup Structure (EST Overlap Matters)

We work with a 3-hour EST overlap window daily (8–11 AM EST = 6:30–9:30 PM IST). Every working day:

15-minute standup: what shipped yesterday, what's in progress today, blockers
Async communication: everything documented in Notion, Slack, or Linear — no information locked in someone's head
CEO access: founders can Slack the engineering lead directly. No PM layer between founder and engineer.

This is the communication model that makes offshore work feel local.

The Economics (Honest Comparison)

US Agency (Mid-Market)          Ailoitte 8-Week Sprint
━━━━━━━━━━━━━━━━━━━━━━         ━━━━━━━━━━━━━━━━━━━━
Timeline: 12–16 weeks           Timeline: 8 weeks
Cost: $170K–$280K               Cost: $35K–$65K
IP Transfer: at completion       IP Transfer: day 1
Communication: PM layer         Communication: direct to engineers
Velocity: 1×                    Velocity: 3× (AI-first)
Runway burned: 3–4 months       Runway burned: 2 months

The 60% cost difference is real. The 4–8 week timeline difference is real.

But the number that matters most to a pre-seed founder: the runway difference.

If you have 12 months of runway and an agency takes 16 weeks, you have 9 months left when you finally have a product to show. If Ailoitte takes 8 weeks, you have 10 months left — with a working product in market 8 weeks earlier.

That 2-month + 8-week difference can be the difference between surviving to your next raise and not.

What We've Built in 8 Weeks (3 Examples)

AI-powered document processing pipeline for a legal tech startup: uploaded PDFs → extracted structured data → review dashboard → audit trail. Built for compliance from day 1.
Customer-facing AI assistant + internal CRM integration for a SaaS startup: RAG over product documentation, Salesforce integration, escalation routing. Deployed to 200 beta users in week 8.
Multi-tenant logistics visibility dashboard with AI anomaly detection: real-time shipment data, predictive delay alerts, and carrier performance scoring. Production on AWS with a full monitoring stack.

None of these were demos handed over as zip files. All were production systems on real infrastructure.

If you're a founder evaluating MVP build options right now, or an engineer who's worked inside an 8-week sprint model, I'd genuinely love to hear what worked and what didn't.

What's the shortest credible MVP build timeline you've seen for a production-quality system?

Why Startups Are Beating FAANG at AI Shipping Speed: It's Not About Hiring More ML Engineers

Sunil Kumar — Wed, 08 Apr 2026 06:47:48 +0000

The US has 1.5 million unfilled software engineering positions projected through 2028. Senior ML engineers command $250,000–$350,000 in total comp. And you're a Series A startup trying to hire your first ML engineer while competing against Amazon and OpenAI.

This is not a sourcing problem. This is a structural market problem.

The startups moving fastest on AI aren't winning because they hired better. Most of them changed the model entirely.

The Real Cost of the ML Hiring Process

Let's do the actual math that most teams don't calculate:

Time cost:

Average time to hire a senior ML engineer: 6+ months
Weeks of ML roadmap blocked: 26
Features not shipped: dependent on roadmap, but almost certainly significant

Financial cost:

Recruiter fee (typically 20–25% of first-year salary): $50,000–$80,000
Engineering manager time on interviews (40+ hours): $5,000–$10,000 at loaded cost
First-year total comp: $250,000–$350,000
Year 1 cost to hire and employ: $300,000–$440,000

The hidden cost: 26 weeks during which your AI roadmap wasn't moving and your competitors' was.

What "AI-First Engineering" Actually Means (Not Just Copilot)

There's a meaningful distinction between a developer who uses GitHub Copilot and an engineer who builds in an AI-first workflow.

Standard developer + Copilot:

Autocomplete and basic code suggestion
Marginal velocity improvement: 1.5–2×
Still requires significant manual implementation for novel problems

AI-first engineering workflow:

Planning → AI-assisted architecture review + spec generation
Implementation → Multi-agent code generation with human review gates
Testing → AI-generated test cases + automated evaluation loops
Debugging → Semantic search across codebase + AI root cause analysis
Documentation → Auto-generated from code + human refinement
Code review → AI pre-review flags issues before a human reviewer sees them

The velocity difference for appropriate tasks: 10–20× vs. a traditional workflow. Not everywhere — but on the high-repetition, high-specification tasks that make up 60–70% of AI feature development, the gap is significant.

What "Productive in 2 Weeks" Actually Requires

For an external engineer to be genuinely productive on your codebase within 2 weeks, specific conditions need to be true:

Week 1 onboarding checklist:
□ Complete codebase access + architecture walkthrough (day 1–2)
□ Development environment setup with AI tooling configured (day 1)
□ First PR in review by end of week 1
□ Daily stand-up overlap with your team (30 min minimum)
□ Clear first deliverable scoped before they start

Week 2 velocity check:
□ PRs being merged without major rework
□ Asking domain questions, not tooling questions
□ Contributing to technical decisions, not just implementing specs

If these conditions aren't met, "productive in 2 weeks" is marketing language. Verify the specific team's onboarding process before committing.

The Economics Comparison

Option A: US ML Engineer Hire
├── Time to start: 6+ months
├── Year 1 fully-loaded cost: $300K–$440K
├── Ongoing: $250K–$350K/year
├── Risk: They leave for FAANG in 18 months
└── Equity dilution: Typically 0.1–0.5% for senior ML hire

Option B: AI-First Team (pre-vetted, productivity-focused)
├── Time to start: 2 weeks
├── Monthly cost: $25K–$40K/month
├── Annualized: $300K–$480K (comparable)
├── But: No 6-month wait, no recruiter fee, no ramp time
└── And: Scales up/down with your roadmap needs

The total cost over 12 months can be similar. The velocity difference in months 1–6 is not.

When Hiring Still Makes Sense

This isn't an argument against hiring ML engineers. It's an argument against letting the hiring process be the bottleneck for your AI roadmap.

Hire when:

You're building proprietary ML systems that require deep, continuous domain expertise
You have stable, long-term ML infrastructure work (not feature development)
You're at Series B+ and building an internal ML platform
The specific work requires on-site access or regulatory clearance

Use an AI-first team when:

You need to start shipping AI features in weeks, not months
Your ML needs are feature-driven (RAG, agents, integrations, inference pipelines)
Your roadmap is evolving, and you need flexibility to scale the team up or down
The opportunity cost of a 6-month hiring cycle is unacceptable

The Engineering Question Worth Asking

If you're evaluating the AI-first team approach, what does your onboarding process look like? How do they handle knowledge transfer? What's the escalation path when technical decisions require domain expertise you haven't transferred yet?

The answer to those questions tells you more about whether the model works than any velocity claim.

What's your team's current approach when ML hiring is taking too long? Curious whether others have found alternative models that worked.

Sunil writes about production AI engineering from the Ailoitte team — AI-first engineering teams for startups building AI products.

Why AI Systems Break in Production (And the 5 Architecture Decisions That Prevent It)

Sunil Kumar — Tue, 07 Apr 2026 05:59:33 +0000

After working on production AI systems across fintech, healthcare, and SaaS, I've seen this pattern repeat so consistently that it now has a name in our team: the week-6 demo gap.

The AI demo worked perfectly. Six weeks after launch, users started reporting wrong outputs. Nobody could explain why, because the system was never built to explain why.

Here's what causes it, and the 5 architecture decisions that prevent it.

The Demo Is Not the Product

Every AI demo uses carefully selected examples where the system performs well. Production users are unpredictable — they hit exactly the edge cases the demo never surfaced.

This isn't dishonesty on the part of the development team. It's the natural result of showcasing a system under optimal conditions rather than operating it under production conditions.

The gap:

Demo inputs: curated, cleaned, representative of the "easy 80%"
Production inputs: unpredictable, messy, often the "hard 20%" that breaks the system

The 5 Architecture Decisions That Determine Outcome

1. Eval Framework — Built Before Application Code

# Minimal eval framework structure
eval_config = {
    "test_set_path": "./eval/production_samples_500.jsonl",  
    "metrics": ["precision_at_5", "factual_accuracy", "format_compliance"],
    "regression_threshold": {
        "precision_at_5": 0.05,    # max allowed drop before blocking
        "factual_accuracy": 0.03,
        "format_compliance": 0.02
    },
    "human_eval_sample_rate": 0.02  # 2% of production calls sampled
}

Without this: you make a change, it looks good on 10 examples, you ship it. Two weeks later: users report a regression that wasn't in your 10 examples. With this: every change is validated against 500 representative labelled examples before shipping.

Build this in week 1. Not week 10.

2. Confidence Thresholding — Route Low-Confidence Outputs

def route_ai_response(response, context):
    confidence = calculate_confidence(response, context)

    if confidence >= 0.85:
        return {"status": "serve", "response": response}
    elif confidence >= 0.70:
        return {"status": "serve_with_caveat", 
                "response": response,
                "disclaimer": "Note: This response may have lower accuracy on this specific query."}
    else:
        return {"status": "fallback",
                "response": "I don't have confident information on this specific question.",
                "escalate": True}

The system should know when it doesn't know. This is not an optional quality-of-life feature. In regulated industries (fintech, healthcare), a system that presents guesses as facts with equal confidence is a compliance risk.

3. Graceful Degradation — Design Every Failure Path

async def ai_feature_handler(user_query: str, context: dict) -> AIResponse:
    try:
        response = await ai_client.generate(user_query, context, timeout=4.0)

        if not validate_response_format(response):
            return fallback_response("format_invalid")

        if get_confidence(response) < CONFIDENCE_THRESHOLD:
            return fallback_response("low_confidence")

        return AIResponse(content=response, status="success")

    except TimeoutError:
        return fallback_response("timeout")
    except RateLimitError:
        return fallback_response("rate_limited") 
    except Exception as e:
        log_ai_error(e, user_query, context)
        return fallback_response("unexpected_error")

def fallback_response(reason: str) -> AIResponse:
    """Always return something useful, never break silently."""
    messages = {
        "timeout": "Taking longer than expected. Please try again in a moment.",
        "low_confidence": "I don't have enough information to answer this confidently.",
        "rate_limited": "High demand right now. Please retry in 30 seconds.",
        "format_invalid": "Unable to process response. Please rephrase your question.",
        "unexpected_error": "Something went wrong. Our team has been notified."
    }
    return AIResponse(content=messages.get(reason, "An error occurred."), status="fallback", reason=reason)

The failure path needs as much design attention as the success path. In most systems we audit, failure handling is an afterthought.

4. Retrieval Quality Monitoring — Separate from Generation Quality

class RetrievalMonitor:
    def log_retrieval_event(self, query, retrieved_chunks, latency_ms):
        # Track separately from generation quality
        self.metrics.record({
            "query_hash": hash(query),
            "retrieval_latency_ms": latency_ms,
            "chunks_returned": len(retrieved_chunks),
            "avg_relevance_score": mean([c.score for c in retrieved_chunks]),
            "top_relevance_score": max([c.score for c in retrieved_chunks]),
            "low_confidence_retrieval": max([c.score for c in retrieved_chunks]) < 0.7
        })

    def alert_on_degradation(self, window_minutes=60):
        recent_events = self.get_events_in_window(window_minutes)
        low_confidence_rate = sum(1 for e in recent_events if e["low_confidence_retrieval"]) / len(recent_events)

        if low_confidence_rate > 0.15:  # >15% of queries returning low-confidence retrievals
            self.send_alert(f"Retrieval quality degraded: {low_confidence_rate:.1%} low-confidence rate in last {window_minutes}min")

Retrieval and generation fail independently. A system can have good generation quality on easy queries and silently terrible retrieval on hard queries. End-to-end metrics don't surface this. You need separate retrieval monitoring.

5. Model Version Pinning — No Surprise Breaking Changes

# ai_config.yaml
models:
  primary:
    provider: openai
    model: "gpt-4o-2024-08-06"  # Pinned — not "gpt-4o" (auto-updates)
    fallback: "gpt-4o-mini-2024-07-18"
  embedding:
    provider: openai
    model: "text-embedding-3-small"
    version: "2"

deployment:
  auto_upgrade: false
  change_management: required
  test_before_upgrade: true

"Latest" is not a production model version. Pin everything. Test model upgrades in staging with your eval suite before promoting to production.

The One Question That Tests All 5

Before signing any AI development contract, ask the vendor:

"Can you run your demo on 20 inputs I select, including our messiest real-world examples?"

Teams who've built for production say yes immediately.

Teams who've built impressive demos find qualifications: "we'd need to clean it first," "that's a slightly different use case," "we'll address that in phase 2."

Those qualifications are the answer.

Summary

Gap	Prevention
No eval framework	Build before week 1 application code
No confidence handling	Implement thresholding + routing
No graceful degradation	Design every failure path
No retrieval monitoring	Separate retrieval metrics from generation
Model version surprises	Pin all model versions

What production AI gap has been hardest to catch before it affected users? Drop your experience in the comments, genuinely useful to compare patterns across different domains.

Sunil writes about production AI engineering from the Ailoitte team, we build 12-week AI Velocity Pods engagements for fintech, healthcare, and SaaS companies.

Building Model-Agnostic AI Architecture: The Pattern That Future-Proofs Your System

Sunil Kumar — Mon, 06 Apr 2026 06:13:26 +0000

Why this matters

AI model prices dropped 80% between 2023 and 2025. Three major model releases changed the capability/cost tradeoff significantly. New providers entered the market. Existing providers deprecated model versions with 6 months notice.

If your application calls openai.chat.completions.create() directly in 40 places, every one of these market changes is a refactoring project.

Here's the architecture pattern we use across every production AI system we build at Ailoitte.

The 4-Layer Model-Agnostic Pattern

Layer 1: The Unified Interface

Application code never imports a provider SDK directly. Every AI call goes through a single interface:

interface AICallConfig {
  task: 'classification' | 'generation' | 'extraction' | 'embedding';
  priority: 'speed' | 'quality' | 'cost';
  maxTokens?: number;
  temperature?: number;
  stream?: boolean;
}

interface NormalisedAIResponse {
  content: string;
  tokensUsed: { input: number; output: number; total: number };
  modelUsed: string;
  latencyMs: number;
  confidence?: number;
  cached: boolean;
}

async function callAI(
  prompt: string,
  context: Record,
  config: AICallConfig
): Promise {
  const provider = await routeToProvider(config);
  const rawResponse = await provider.complete(prompt, context, config);
  return normaliseResponse(rawResponse, provider.name, config);
}

No provider import. No model name. No API key. Just your interface.

Layer 2: Configuration-Driven Routing

The routing config lives in a JSON file — not in code:

{
  "routing": {
    "classification": {
      "cost": "gpt-4o-mini",
      "quality": "claude-3-5-sonnet-20241022",
      "speed": "gpt-4o-mini"
    },
    "generation": {
      "cost": "gpt-4o-mini",
      "quality": "claude-3-5-sonnet-20241022",
      "speed": "gpt-4o"
    },
    "extraction": {
      "cost": "gpt-4o-mini",
      "quality": "gpt-4o",
      "speed": "gpt-4o-mini"
    }
  },
  "fallback_chain": ["gpt-4o", "claude-3-5-sonnet-20241022", "gpt-4o-mini"],
  "fallback_trigger": ["rate_limit", "timeout", "server_error"]
}

Changing the model for any task type = updating this JSON. No code deployment.

Layer 3: Response Normalisation

OpenAI, Anthropic, and other providers return different response shapes. The normalisation layer abstracts this:

function normaliseResponse(
  raw: OpenAIResponse | AnthropicResponse | GeminiResponse,
  providerName: string,
  config: AICallConfig
): NormalisedAIResponse {
  if (providerName === 'openai') {
    return {
      content: raw.choices[0].message.content,
      tokensUsed: {
        input: raw.usage.prompt_tokens,
        output: raw.usage.completion_tokens,
        total: raw.usage.total_tokens
      },
      modelUsed: raw.model,
      latencyMs: raw._latencyMs, // injected by provider wrapper
      cached: false
    };
  }
  if (providerName === 'anthropic') {
    return {
      content: raw.content[0].text,
      tokensUsed: {
        input: raw.usage.input_tokens,
        output: raw.usage.output_tokens,
        total: raw.usage.input_tokens + raw.usage.output_tokens
      },
      modelUsed: raw.model,
      latencyMs: raw._latencyMs,
      cached: false
    };
  }
  // Add more providers here without touching application code
}

Layer 4: Automatic Fallback Routing

async function routeToProvider(config: AICallConfig): Promise {
  const routingConfig = await loadRoutingConfig();
  const primaryModelName = routingConfig.routing[config.task][config.priority];

  try {
    const provider = providerRegistry.get(primaryModelName);
    await provider.healthCheck(); // lightweight check
    return provider;
  } catch (e) {
    if (isRetryableError(e)) {
      // Route to next in fallback chain
      return getNextFallbackProvider(primaryModelName, routingConfig.fallback_chain);
    }
    throw e;
  }
}

Real-World Result

One client wanted to migrate 60% of their API calls from GPT-4 to Claude 3.5 Sonnet when Anthropic's pricing dropped significantly. Because they'd built with this pattern:

Updated the routing config JSON (2 minutes)
Ran the eval set against the new routing (2 hours)
Confirmed quality parity on their specific tasks (passed)
Deployed the config change (10 minutes)

Zero application code changes. Monthly savings: $9,200.

The Setup Cost

This pattern adds approximately 3–5 days of initial engineering work, depending on how many providers you want to support. In the current AI market, that investment pays back within the first provider change you need to make.

Written by Sunil, Content Lead at Ailoitte — we build production AI systems for fintech, healthcare, SaaS, and logistics companies. We publish technical content on what actually works in production AI, not just what tutorials teach.

Token Cost Optimization in Production LLMs: 3 Approaches With Real Numbers

Sunil Kumar — Thu, 02 Apr 2026 07:15:22 +0000

We were burning $4,100/month on inference for one fintech client. Here's the three-part stack that cut it to $1,560, without touching the model.
LLM inference costs are the silent budget killer of production AI. You see a demo that costs pennies to run. You ship it, users arrive, the corpus grows, query complexity rises — and suddenly you're looking at a cloud bill that nobody planned for.

We hit this on a fintech client's internal compliance Q&A system. At launch: ~2,000 queries/day, average prompt length 1,800 tokens, GPT-4 for everything. Monthly inference bill: $4,100. Three months post-launch: 6,000 queries/day, average prompt ballooning to 2,400 tokens from accumulated context. Projected bill: $13,000/month. Nobody had modelled for usage growth.

Here's the three-layer optimization stack we implemented, with exact numbers from that engagement.

01 Prompt compression — trim the fat before it hits the model
The most direct lever: reduce the token count of every prompt before it reaches the inference endpoint. This sounds obvious. Most teams don't do it because the naive approach (just truncate) destroys quality. The right approach uses semantic compression.

We used LLMLingua from Microsoft Research, a small model that compresses prompts by removing tokens that are statistically low-information relative to the query, while preserving semantic content. On our fintech client's prompts, we achieved 38% compression with less than 3% degradation in answer quality on the golden dataset.

The latency cost of compression is ~120ms on the CPU. For our use case (internal tool, not real-time), this was acceptable. If you're building a customer-facing product where P95 latency matters, benchmark this carefully — it may not always be worth it.

✓ On 2,400-token average prompts, 38% compression saves ~912 tokens per query. At $0.03/1K tokens (GPT-4), that's $0.027/query. At 6,000 queries/day: ~$162/day, ~$4,860/month, from compression alone.

02 Intelligent model routing — not everything needs GPT-4
The second insight sounds simple, but gets skipped: most queries in a production system don't require your most expensive model. Simple factual lookups, short-answer questions, classification tasks — these can be handled by a cheaper model with no perceptible quality difference to the user.

We built a lightweight router that classifies incoming queries by complexity before they hit the inference endpoint. Simple queries go to GPT-3.5-turbo (or equivalent). Complex, multi-hop, or reasoning-heavy queries go to GPT-4. The classification itself is done with a fine-tuned small model (300M parameters) that adds ~15ms of latency.

In our fintech client's query distribution, 61% of queries were classifiable as "simple" (lookup, boolean, date-retrieval). Routing those to GPT-3.5-turbo reduced cost per query by ~93% on that segment, which was 61% of all queries. Blended cost reduction: ~57%.

⚠ Do not use a threshold below 0.80 for your complexity classifier. At 0.70, we saw too many complex queries slipping through to the cheaper model, which produced noticeably lower quality answers on multi-part compliance questions. Trust the uncertainty — if it's not clearly simple, route up.

03 Semantic caching — stop paying for identical questions
In any production deployment with hundreds or thousands of users, a meaningful percentage of queries are semantically identical even if lexically different. "What's the KYC requirement?" and "Can you explain the know-your-customer process?" are the same query. Without a cache, you pay full inference cost for both.

Semantic caching embeds incoming queries and compares them against a cache index. If a semantically similar query exists (above a cosine similarity threshold), you return the cached response. No model call required.

On the fintech compliance system, cache hit rate stabilised at 34% after two weeks. That's 34% of queries returning a cached answer with zero inference cost. With the combination of all three approaches, compression, routing, and caching, here's what the numbers looked like:

→Implementation order matters
If you're implementing these on an existing system, do them in this order:

Semantic caching first — zero infrastructure complexity, immediate payoff on any system with repeated query patterns. Measurable in 72 hours.
Model routing second — requires building or fine-tuning a classifier, but the ROI is significant if your query distribution is mixed-complexity (most are).
Prompt compression third — most engineering effort, requires calibration against your golden dataset. Worth it at scale, but don't start here.

✓ Before you implement any of these: instrument everything. If you don't have per-query token counts, model selection, and cache hit rate logged today, you're flying blind. Add logging first. Optimize second.

If your team is burning more than $1,000/month on inference and you haven't implemented semantic caching yet, that's your fastest win. The model routing classifier takes longer to build but pays back disproportionately if your query mix is heterogeneous.

What optimization approaches is your team using? Drop a comment, I'm specifically curious whether anyone's had success with speculative decoding or prefix caching at the infrastructure level.

We run production AI delivery engagements at Ailoitte, if you're wrestling with runaway inference costs, the architecture decisions are usually fixable without changing the model.

Why RAG Pipelines Fail at Production Scale (And What We Fixed)

Sunil Kumar — Wed, 01 Apr 2026 11:07:35 +0000

5 failure modes we hit building 12+ production RAG systems, and the architectural fixes that actually worked.

I've spent the last 14 months building production AI systems for fintech, healthcare, and SaaS clients. Of the 12+ RAG pipelines we've shipped, every single one failed in a different way than it did in staging.

Not broke. Failed. Silently degraded. Answered confidently and wrong. Retrieved the right document but extracted the wrong passage. Worked at 10 queries per minute and collapsed at 100.

Here's what we kept hitting, and what we fixed.

01 Naive chunking destroys retrieval quality
The default in most RAG tutorials is fixed-size chunking, splitting every document into 512-token chunks, embedding them, and then done. It works in demos. In production, it silently kills accuracy.

The problem: semantic meaning doesn't respect token boundaries. A contract clause that spans 600 tokens gets split in the middle. A medical report with a critical finding in the second half of a paragraph gets separated from its context. The retriever finds half the answer, and the LLM hallucinates the rest.

❌ Fixed-size chunking at 512 tokens: retrieval precision dropped to 54% on our healthcare client's policy documents after go-live.

What we switched to: a parent-child chunking strategy with semantic boundary detection.

Retrieval precision went from 54% to 81% on the same document set. The LLM gets the full semantic unit, not a fragment.

02 The wrong embedding model for your domain
Most teams default to text-embedding-ada-002 or a generic SBERT model. These are fine for general English. They're inadequate for financial filings, clinical notes, or legal language.

We had a fintech client whose RAG system was scoring 0.87 cosine similarity on retrieved passages, but the answers were wrong 40% of the time. The model was retrieving chunks that were superficially similar in language but semantically different in context. "Risk" in a compliance document does not mean the same thing as "risk" in an earnings call.

The fix: switch to a domain-adapted or domain-fine-tuned embedding model. For finance, BGE-financial or FinBERT embeddings. For clinical, ClinicalBERT or BioBERT as an embedding base. For general enterprise, a hybrid approach:

The instruction asymmetry matters. BGE models were trained with different prefixes for queries vs documents. Skip it, and you lose 8–12% recall on domain-specific content.

03 No reranking layer — cosine similarity isn't relevance
Vector similarity retrieves semantically proximate chunks. But proximity ≠ relevance to the specific question. You need a reranker.

Without a reranker, the top-k retrieved chunks are sorted by embedding similarity, which doesn't account for query-specific intent, negation, or specificity. We consistently saw the most relevant chunk sitting at position 4 or 5 in the retrieval output, behind noisier but "closer" matches.

04 Context window mismanagement at scale
At low query volumes, stuffing 8 retrieved chunks into the prompt works. At production scale with concurrent requests, you hit three problems: cost explosion, latency spikes, and more insidiously, the Lost in the Middle problem.

Research consistently shows that LLMs have lower recall for information buried in the middle of long contexts. If your most relevant chunk ends up at position 3 of 8 in the context, the model may not weight it appropriately.

Our production pattern now:

Retrieve 20 candidates from the vector store
Rerank to top 5
Apply context compression to reduce token count by ~60%
Place the most relevant chunk first and last (primacy + recency bias in LLMs)

05 No evaluation infrastructure — flying blind
This is the one that hurts the most to admit: most of the RAG systems we inherited had zero evaluation framework. They were shipped, deemed "working" based on informal testing, and degraded silently over weeks as the document corpus grew or the query distribution shifted.

You need three things before you go to production:

A golden dataset — 50–100 question/answer pairs manually verified against your document corpus
RAGAS metrics — faithfulness, answer relevancy, context precision, context recall
A weekly eval run — automated, tracked in a dashboard, with alerts if any metric drops more than 5%

✅ Once you have RAGAS running, you can actually compare chunking strategies, embedding models, and reranker configs quantitatively. It turns RAG tuning from guesswork into engineering.

What the fixed architecture looks like
After applying all five fixes on a healthcare SaaS client's policy document RAG system, here's what the numbers looked like:

The full production RAG stack now looks like: semantic chunking → domain-adapted embeddings → hybrid search (vector + BM25) → cross-encoder reranking → context compression → LLM with structured output + RAGAS eval loop.

Each layer adds ~50–150ms of latency. The tradeoff is worth it when the cost of a hallucinated answer in a healthcare or fintech context is a support ticket, a compliance issue, or a lost contract.

If you've hit any of these, or if your RAG system works great in staging and degrades in production, drop a comment. I'm collecting failure patterns across verticals right now and would love to hear what you're seeing.

We run a technical AI delivery practice called Ailoitte, if you're rebuilding a broken RAG pipeline and want to talk architecture, reach out.

How We Ship Production AI in 12 Weeks: The Architecture That Actually Works

Sunil Kumar — Thu, 26 Mar 2026 07:17:30 +0000

If you've tried shipping an AI feature to production recently, you know the gap between "demo works in staging" and "prod-stable under real load" is enormous.
This post is about the architecture decisions that close that gap, specifically, the five engineering phases we've converged on after shipping production AI across 14+ industries. No fluff, just the decisions that matter.

The 4 Engineering Failure Modes That Kill AI Timelines
Before the framework, the failure modes. These are not theoretical, every one of them has caused a production incident or a blown timeline in the last 18 months.

1. Token cost explosions in agentic loops
Single-turn LLM calls are predictable. Agentic loops, where an AI takes sequential actions, calls tools, and iterates, are not. Without per-workflow token budgets, you're running an infinite loop on a metered connection.
Here's what unguarded agentic architecture looks like:

We diagnosed a production chatbot burning $400/day per enterprise client. Nobody noticed until month 3, by which point, the feature was destroying margin in real time. The fix:

2. RAG without domain boundaries
The naive RAG setup: dump all your enterprise data into a vector store, let the LLM retrieve whatever it wants. This produces authoritative hallucinations, outputs that are coherent, confident, and wrong because they're blending context from unrelated domains.

Domain-Driven Design applies directly to AI service layers. The principle: an AI workflow accesses only the data collections relevant to its task category. Full stop.

The benefits compound: smaller context windows (lower cost), easier compliance auditing (you know exactly what data informed every decision), and a dramatically reduced hallucination surface area.

3. No observability in production
You are not done shipping when the feature passes staging tests. Production AI requires active monitoring that most teams treat as a post-launch concern. It isn't.

The minimum viable observability stack for production AI:
• Hallucination detection — compare outputs against retrieved source context; flag divergence above a threshold

• Drift detection — monitor output distribution over time; model behavior changes as training data ages

• HITL checkpoints — for high-stakes decisions (loan approvals, patient triage, compliance flags), human review before action

• Decision logs — structured record of: input, retrieved context, model output, confidence score, action taken. Forensic trail for every decision

The LLM landscape shifts quarterly. Lock-in to a single provider is technical debt that compounds with every model release you can't migrate to.

The 5-Phase Delivery Framework

The Billing Model Is an Architectural Decision
This sounds like a business detail. It isn't. The billing model determines every engineering incentive in the engagement.
Under hourly billing: no structural reason to ship faster, optimize token costs, or build durable monitoring. Every inefficiency is revenue. Every extra sprint is billable.

Under outcome-based contracts: speed becomes a margin driver. Token optimization saves the delivery team money. Durable architecture reduces support load. Every incentive aligns with delivery quality.

The market data: seat/hourly AI pricing dropped 21% to 15% of engagements in 2025. Outcome-based surged 27% to 41%.

One More Thing: The Compounding Data Moat
Every production AI deployment generates proprietary training signals, correction patterns, user interactions, and edge cases. These compounds.

An enterprise that deployed in Q1 has 3 quarters of proprietary production data by Q4. A competitor still in planning cycles has none. That data gap doesn't close with a better model selection. It closes slowly, with earlier deployment.

The fastest path to closing it is shipping. This is the whole argument for Velocity PODs.

What's your current production AI stack?
Specifically curious what others are using for observability and hallucination detection in production.
LangSmith? Custom? Something else? Drop it in the comments.

Why Your $600K AI Hiring Cycle Is Costing You More Than Just Money

Sunil Kumar — Wed, 25 Mar 2026 07:52:47 +0000

82% of enterprises are running active AI PoCs. Fewer than 4% reach production-wide deployment. The gap isn't talent or budget, it's delivery architecture.

I want to talk about something most AI delivery postmortems won't say out loud: the traditional hire-and-build model is structurally broken for AI systems in 2026.

Not because the engineers aren't good. Because the incentive structures, team compositions, and billing models were designed for a world where software systems were deterministic.

AI systems aren't.

The Math Behind the $600K Figure

A senior AI/ML engineer in 2026 costs $180K+ base. Recruiter fee at 20%: $36K. Time-to-hire in the current market: 3–6 months. Onboarding ramp on LLM-specific tooling: another 1–3 months.

Now build your minimum viable AI delivery team:

AI/LLM Engineer: ~$180K
MLOps Specialist: ~$160K
Data Engineer: ~$140K

That's $480K/year in salaries alone — before tooling, cloud costs, or the first PR is merged.

Before a single production model has been trained on your domain data.

The Capability-Delivery Chasm (Why PoCs Fail in Production)

Here's a pattern every AI engineer reading this has probably seen:

PoC in sandbox → Works in demo → Breaks on production load

The PoC was built fast, by generalists learning LLM orchestration on the job, optimizing for demo performance rather than production stability.

What's missing at handoff:

Hallucination monitoring
Token cost guardrails
Drift detection
Audit trail / HITL checkpoints for regulated decisions
Observability stack
Model-agnostic architecture (so you're not locked to one provider)

These aren't afterthoughts. In production AI, these ARE the system.

The Compute Waste Problem (3–10x Cost Multiplier)

This one stings because it's invisible until the cloud bill arrives.

Generalist developers default to:

Full-context retrieval on every query
No prompt caching
Unstructured prompts that balloon token usage
No cost ceiling monitoring per workflow

One agentic workflow without token guardrails can generate a $50K monthly API bill overnight. A real healthcare SaaS deployment we audited had $11K/month in unnecessary API spend traced directly to unstructured prompts and full-context retrieval on every call.

The fix was architectural, not model-related. Applied in the first sprint.

What an AI POD Actually Is (vs. Staff Aug)

The term "AI POD" gets used loosely, so let me be precise:

AI POD = pre-assembled, cross-functional delivery unit

AI/LLM Engineer
MLOps Specialist
Data Engineer
Domain Architect
QA Specialist

Contracted on defined deliverables with production-stable AI as the exit criterion. Not hours. Not headcount. Outcomes.

The key distinction from staff augmentation: a POD ships the monitoring stack, observability layer, and IP transfer as required deliverables, not optional line items.

The Delivery Sequence That Actually Works

Start with data, not models:

Step 1: Data Landscape Audit
Map every silo. Define ingestion architecture. Identify what the AI can touch and what it shouldn't. Skipping this step produces confident hallucinations, the worst kind.

Step 2: Domain-Driven Service Boundaries
Apply DDD to the AI service layer. Tight boundaries reduce hallucination surface area, attack surface, and make compliance auditing tractable.

Step 3: Model-Agnostic RAG Build
Build the retrieval layer on open frameworks, LangChain, LlamaIndex. The LLM landscape shifts every quarter. Locking into a single provider is compounding technical debt.

Step 4: Token Optimization + Guardrails
Prompt caching, structured retrieval, cost ceiling monitoring, and token budget guardrails per workflow. This is what separates a POD from a staff aug arrangement.

Step 5: Observability Stack + IP Transfer
Hallucination monitoring, drift detection, HITL checkpoints, automated decision logs. Full IP transfer, every model, config, codebase, and client retains everything.

The Billing Model Problem

Under hourly billing, the vendor has no structural incentive to:

Ship faster
Optimize token costs
Build monitoring layers

Every extra hour is revenue. Every inefficiency is a billable line item. AI work is non-linear; an optimized prompt can replace forty API calls. Hourly billing rewards the forty-call path.

Outcome-based billing resolves this. The POD is contracted to ship a production-stable system. Token efficiency and monitoring aren't optional; they're part of what "shipped" means.

The question isn't whether to use AI. That decision was made two years ago.

The question is: how many more 6-month delivery cycles can you absorb while a competitor ships quarterly?