Forem: Sunil Kumar

Shipping AI Features When Headcount Is Frozen, What Is Actually Working

Sunil Kumar — Wed, 29 Apr 2026 07:56:43 +0000

Board and customers are demanding AI features. Your engineering team is still lean from the 2023–24 cuts. Headcount approval isn't coming. Here's what's actually working.

Context: Ailoitte builds AI engineering teams for companies in exactly this situation. Sharing what we see working across multiple teams.

Why Is This Problem So Common Right Now?

The 2023–24 tech layoffs were widespread. Engineering teams at Series B–D SaaS companies cut 30–50% of headcount.

Fast-forward to 2025: competitors who didn't cut as deep — or who maintained AI engineering investment — have shipped features that are now visible to customers.

The board wants AI on the roadmap. Customers are asking. And the engineering team is still operating lean with a headcount freeze still politically in effect.

Why Rebuilding the Team Isn't the Right First Move

For scoped AI feature delivery, rebuilding headcount has three problems:

Timeline. 6–12 months to hire, onboard, and ramp a senior AI engineer. The board's Q3 roadmap expectation and the hiring timeline don't overlap.

Political difficulty. Adding headcount that was just cut is a hard internal conversation. The CFO who approved the cuts is still in the room.

Cost. $200K–$300K annually per AI engineer vs. $30K–$60K for a scoped engagement with a fixed outcome.

What Does an "AI Engineering Arm" Actually Mean?

An AI engineering arm is an external team that operates like an internal one. Not a consultancy. Not a separate product track.

Engineers embedded in your sprint, working in your codebase, attending your standups, shipping to your production environment.

The difference from a typical agency:

Code ships to your production, not a demo environment
IP stays 100% with your company
Engineers are assigned specifically to your team, not rotated across accounts
Start time: 2 weeks, not 6 months

How Does the Budget Conversation Change?

The companies moving fastest on this have reframed the internal approval conversation.

❌ "We need to hire AI engineers" → stalls at headcount approval.

✅ "We need to ship this AI feature — here's the project budget" → routes through a different approval path.

$30K–$60K as a project budget vs. $200K–$300K annually as a hire are evaluated differently by most finance teams — even when the output is equivalent.

What Gets Shipped?

Typical engagements for post-layoff AI delivery:

AI search or assistant layer on an existing SaaS product
RAG-based knowledge system for internal or customer-facing use
AI feature sprint matching a competitor's recently shipped capability
Automation of a specific manual workflow inside the product

The Reframe That Unlocks Movement

The engineering leads making progress right now aren't winning the headcount argument. They're bypassing it — by reframing AI delivery as a project budget conversation instead of a hiring conversation.

The output is the same. The approval path is completely different.

Ailoitte's AI Velocity Pods are structured specifically for this model.

What does your internal conversation around AI headcount look like right now? Drop it in the comments — curious what constraints others are navigating.

$300K vs $60K: The AI Engineer Hiring Gap and Why It Won't Last

Sunil Kumar — Tue, 28 Apr 2026 06:23:25 +0000

Full disclosure: Ailoitte provides offshore AI engineering teams. The market data below is real regardless.

What Does a Senior ML Engineer Actually Cost in 2025?

Location	Total Comp
US	$250K–$350K
India (pre-vetted, production experience)	$38K–$80K

US hiring means competing with FAANG for the same candidates. 1.5 million US software engineering positions are projected to go unfilled through 2028 — and the ML layer is the worst of it.

That $38K–$80K offshore figure isn't for generalists. It's for engineers with real RAG pipeline, agent infrastructure, and ML pipeline work in their history.

That's a 70% cost difference. Not a rounding error.

Is Offshore AI Engineering Actually as Good?

With proper vetting: yes.
Without it: frequently no.

The failure mode is optimising for speed of placement over depth of screen. Here's what good vetting actually looks like:

Has this engineer shipped the specific thing you need — RAG, agents, evaluation infrastructure — in production?
Can they describe a real failure and what they changed?
Is there genuine overlap availability for collaboration?

Engineers who have shipped production AI systems get specific about failures. Engineers who haven't get vague. That distinction is the entire quality filter.

Why Is This Pricing Gap Closing?

Three forces are compressing it simultaneously:

1. FAANG scaling India
Google, Microsoft, Meta — all significantly expanded India engineering in the past 18 months. They recruit from the same pool, and their presence pushes offshore AI engineer salaries up.

2. India's domestic market
Indian unicorns now pay at levels that only US-remote roles matched two years ago. Internal demand is real and growing.

3. Remote work parity
Engineers compare US-remote and offshore-remote directly now. The geographic premium collapses when geography stops mattering.

Estimate: ~18 months for the current spread to close meaningfully. And that's conservative.

How Long Does an Offshore AI Engineer Take to Ramp?

Two weeks — with structured onboarding.

That means:

Clear documentation
Defined first sprint
3–4 hours of daily overlap
Senior architect available for questions

The 6–8 week ramp stories you hear are almost always disorganised onboarding, not an offshore quality problem.

What Does This Mean Practically?

If you're at Series A–B trying to hire ML engineers:

Factor	Reality
US market	Genuinely difficult without compelling equity at $280K+
Offshore quality	Equivalent with proper vetting
Window for current pricing	~18 months
Ramp time	2 weeks with structured onboarding

Every month of failed US hiring is a month your product isn't shipping.

The Bottom Line

Ailoitte builds AI engineering teams around this model. Full market analysis with sourcing and retention data on Medium →

How long have you been trying to fill an AI engineering role? What has the actual pipeline looked like? Drop it in the comments.

From 62% to 94% RAG Accuracy: The 5 Architecture Changes That Actually Moved the Needle

Sunil Kumar — Mon, 27 Apr 2026 05:24:16 +0000

We measured baseline accuracy across a production RAG system: 62%.

Six weeks later, after five architecture changes and zero model changes: 94%.

Here's exactly what we changed, why each one mattered, and what the numbers looked like before and after each step.

The Setup

Internal knowledge assistant for a mid-market company. Knowledge spread across Confluence, Google Drive, and SharePoint, approximately 4,200 documents total. Users asking natural language questions about internal policies, processes, and product specifications.

Stack at baseline:

Component	Details
LLM	GPT-4o
Embedding model	text-embedding-3-large
Vector store	Pinecone
Chunking	RecursiveCharacterTextSplitter, 1024 tokens, 20% overlap
Retrieval	top-8 by cosine similarity, no re-ranking
Eval	none

This worked well in testing. In production, it hit a ceiling at week three — when real users arrived with queries that went beyond the clean, structured examples we'd tested against.

The first sign was a CTO call. A VP had used the system for a cross-departmental policy question, gotten a confident and wrong answer, and forwarded it to three people. The system had retrieved from one policy document and missed a contradicting clause in another.

Before touching anything, we measured.

Evaluation First — Before Any Changes

150 real queries pulled from production logs. Not test cases we designed, actual user queries from the support logs, stratified across query types.

Reference answers written by domain experts. Automated grading via RAGAS across four dimensions: faithfulness, answer relevance, context precision, context recall. Manual spot-check on the ~15% of answers that fell in scoring edge cases.

Baseline results:

Metric	Score
Overall accuracy	62%
Multi-document queries	41%
Exact-match retrieval	58%
False confidence rate*	68%

*Proportion of wrong answers with no hedging language — the model sounded equally certain whether it was right or wrong.

Without this baseline, none of the following changes would have been verifiable. Build the eval suite first. Always.

Change 1: Semantic Chunking

Impact: 🔴 High

Fixed-window chunking at 1024 tokens truncates documents at arbitrary boundaries. For policy documents, contracts, and multi-page specs, this severs logical relationships between sections, a conditional clause in section 3 that's resolved by a definition in section 7 ends up in two different chunks with no retrieval path connecting them.

We moved to semantic chunking: split on natural topic/section boundaries detected via sentence-level embedding similarity. Consecutive sentences are grouped while their cosine similarity stays above a threshold; a new chunk starts when similarity drops below it.

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def semantic_chunk(sentences, embed_fn, threshold=0.75):
    """
    Split sentences into semantically coherent chunks.

    Args:
        sentences: list of sentence strings
        embed_fn: function that returns an embedding for a string
        threshold: minimum similarity to stay in the same chunk

    Returns:
        list of chunk strings
    """
    chunks = []
    current_chunk = [sentences[0]]

    for i in range(1, len(sentences)):
        prev_emb = embed_fn(sentences[i - 1])
        curr_emb = embed_fn(sentences[i])

        similarity = cosine_similarity(
            np.array(prev_emb).reshape(1, -1),
            np.array(curr_emb).reshape(1, -1)
        )[0][0]

        if similarity >= threshold:
            current_chunk.append(sentences[i])
        else:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i]]

    chunks.append(" ".join(current_chunk))
    return chunks

If you're using LlamaIndex: SemanticSplitterNodeParser with breakpoint_percentile_threshold=85 is a solid starting point. Tune the threshold per document type — policy and legal docs generally benefit from higher thresholds (split less often).

Tradeoffs to expect:

Preprocessing is significantly heavier than fixed-window
Variable chunk lengths complicate token budget management in your context window
Ingestion pipeline needs updating if you have document update automation

For read-heavy internal assistants with low update frequency, the accuracy gain outweighs these costs.

Accuracy improvement on multi-hop queries after this change alone: +31% relative.

Change 2: Hybrid Search with Reciprocal Rank Fusion

Impact: 🔴 High

Semantic/vector search is excellent at concept-level similarity. It's weak at exact matching — product codes, regulation identifiers, version strings, internal project names, anything where a user is looking for a specific term rather than a concept.

We added a BM25 keyword index running parallel to the Pinecone vector index. Retrieved top-20 candidates from each independently, then fused them with Reciprocal Rank Fusion (RRF):

def reciprocal_rank_fusion(
    vector_results: list[str],
    bm25_results: list[str],
    k: int = 60
) -> list[str]:
    """
    Merge two ranked lists using Reciprocal Rank Fusion.

    Args:
        vector_results: doc IDs ranked by vector similarity
        bm25_results: doc IDs ranked by BM25 keyword score
        k: smoothing constant (60 is standard, rarely needs tuning)

    Returns:
        Merged list of doc IDs sorted by fused score (best first)
    """
    scores: dict[str, float] = {}

    for rank, doc_id in enumerate(vector_results):
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    for rank, doc_id in enumerate(bm25_results):
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    return sorted(scores.keys(), key=lambda x: scores[x], reverse=True)

If you're using LangChain: EnsembleRetriever handles this cleanly. Set weights=[0.4, 0.6] for BM25 and vector respectively as a starting point, and adjust based on how many exact-match queries your workload has. The weights matter less than you'd think if you're re-ranking afterward.

from langchain.retrievers import BM25Retriever, EnsembleRetriever

bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 20

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

If your knowledge base contains any structured identifiers, and it almost certainly does, vector-only retrieval is not sufficient. This was the change that fixed the most visible failure class in our case.

Change 3: Cross-Encoder Re-ranking

Impact: 🟡 Medium-High

Vector retrieval and BM25 give you candidate chunks. They rank candidates by how similar they are to the query embedding or how well they match keywords. Neither metric is the same as: how useful is this specific chunk for answering this specific question?

A cross-encoder re-ranker scores each chunk-query pair jointly; it reads both together, not independently, which produces relevance scores much more aligned with what the LLM actually needs.

We used ms-marco-MiniLM-L-6-v2 from sentence-transformers: small enough to run without meaningful infrastructure cost, effective enough to make a real difference.

from sentence_transformers import CrossEncoder
from dataclasses import dataclass

@dataclass
class Chunk:
    text: str
    metadata: dict

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank(query: str, candidates: list[Chunk], top_k: int = 5) -> list[Chunk]:
    """
    Re-rank candidate chunks by relevance to the query.

    Args:
        query: the user's question
        candidates: retrieved chunks from vector + BM25 retrieval
        top_k: how many to pass to the LLM context window

    Returns:
        Top-k chunks ranked by cross-encoder relevance score
    """
    pairs = [(query, chunk.text) for chunk in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, candidates), key=lambda x: x[0], reverse=True)
    return [chunk for _, chunk in ranked[:top_k]]

If you're using LangChain:

from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

model = HuggingFaceCrossEncoder(model_name="cross-encoder/ms-marco-MiniLM-L-6-v2")
compressor = CrossEncoderReranker(model=model, top_n=5)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever  # from Change 2
)

Latency cost: ~200–400ms per query on CPU inference with MiniLM. For an internal assistant where users expect 3–4s response times, this is acceptable. For customer-facing chatbots with sub-1s SLAs, benchmark first and consider skipping re-ranking on queries classified as simple (single-document, low complexity).

Change 4: Source Hierarchy + Metadata Tagging

Impact: 🟡 Medium

When Confluence has a "remote work policy" document updated in March, and SharePoint has a different version updated in January, and both are in your knowledge base — which one does the retriever trust?

Without source hierarchy metadata, the LLM gets both chunks, infers authority from the text, and guesses. It guesses wrong often enough to matter.

We added three metadata fields to every document on ingestion:

from langchain.schema import Document
from datetime import datetime

def create_document_with_metadata(
    text: str,
    source: str,
    source_authority: int,  # 1=primary, 2=secondary, 3=supplementary
    last_updated: datetime,
    domain: str
) -> Document:
    """
    Wrap chunk text with authority metadata for conflict resolution.

    source_authority levels:
        1 = Primary (HR portal, official policy docs, product spec sheets)
        2 = Secondary (team wikis, internal guides, how-to docs)
        3 = Supplementary (Slack archives, meeting notes, draft docs)
    """
    return Document(
        page_content=text,
        metadata={
            "source": source,
            "source_authority": source_authority,
            "last_updated": last_updated.isoformat(),
            "domain": domain
        }
    )

# Retrieval with authority filter — prefer primary sources
vector_retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 20,
        "filter": {"source_authority": {"$lte": 2}}  # primary + secondary only
    }
)

The heuristic is simple: when two sources conflict on the same topic, source_authority=1 wins. This eliminates the most embarrassing failure class — confident wrong answers sourced from an outdated secondary document when the primary existed and said something different.

This change required no model updates and no prompt changes. It's a data pipeline discipline issue, not an AI problem.

Change 5: Structured Evaluation Suite (The One Most Teams Skip)

Impact: 🔴 Foundational

This is the process change, not the architecture change. And it's the one that made every other change above verifiable rather than speculative.

What we built:

150 representative queries sampled from production logs, not cherry-picked happy-path examples, not queries we designed. Stratified across query types: single-document, multi-document, exact-match, comparative, conditional.
Reference answers written by domain experts for each query, the ground truth the system should produce.
Automated RAGAS grading across faithfulness, answer relevance, context precision, and context recall.
Manual spot-check on ~15% of borderline-scored answers.
CI gate: eval suite runs on every deployment. A drop of >3% on any category blocks the deployment.

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

def run_eval(questions, answers, contexts, ground_truths):
    """
    Run RAGAS evaluation against reference answers.

    Returns a dict with per-metric scores and overall accuracy.
    """
    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,       # list of lists — retrieved chunks per question
        "ground_truth": ground_truths
    }

    dataset = Dataset.from_dict(data)

    result = evaluate(
        dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )

    return result

Without this eval pipeline, you're flying blind. Developers eyeball outputs and call it "seems good." Users report the failures that are loud enough to surface, a heavily biased sample of what's actually failing. The 38% failing quietly never gets fixed.

Final Results

All five changes applied, same LLM throughout:

Metric	Baseline	After	Delta
Overall accuracy	62%	94%	+32pp
Multi-document reasoning	41%	87%	+46pp
Exact-match retrieval	58%	96%	+38pp
False confidence rate	68%	12%	-56pp

The false confidence number is the one I find most significant. 68% of wrong answers originally had zero hedging language — the model sounded equally certain whether it was right or wrong. After fixing the retrieval layer (without touching the model or prompt), that rate dropped to 12%.

Overconfidence wasn't a model behavior problem. It was retrieval quality expressing itself as false certainty. Fix what goes into the context window, and the model's calibration improves without any prompt engineering.

What I'd Do Differently

Start with the eval suite. Not after you get complaints, before you go live. The 4–6 hours of building a 150-query eval set with reference answers will save you weeks of debugging production failures blind.

Build for the full query distribution, not the happy path. Your test suite during development is systematically biased toward queries you designed. Real users will find the edges. Measure in production with real queries before you tune anything.

The model is rarely the bottleneck. If your RAG system is hallucinating, look at retrieval before you look at the LLM. Changing the model is expensive, slow, and usually doesn't fix a retrieval quality problem. Changing the retrieval architecture is faster and the impact is larger.

Still On the List

Things we haven't tested yet on this system:

HyDE (Hypothetical Document Embeddings) for query expansion — benchmarks look promising, mixed reports on real-world gains for knowledge-domain workloads
Query decomposition for explicitly multi-hop questions — breaking compound queries into sub-queries before retrieval
Dynamic eval set refresh — our static 150-query set is starting to feel unrepresentative as the knowledge base and user base evolve

Happy to go deeper on any of these in the comments, or on RAGAS configuration and the source hierarchy schema if there's interest.

Sunil Kumar is CEO of Ailoitte, an AI-native product engineering company that builds production-grade AI systems for startups, enterprises, and regulated industries. Ailoitte's AI Velocity Pods deliver fixed-price, outcome-based software development — including production RAG infrastructure with evaluation pipelines built in from day one.

Hitting a RAG quality wall? Let's talk.

How to Technically Audit a Vendor's AI System - What's Actually Running Under the Hood

Sunil Kumar — Thu, 23 Apr 2026 06:18:04 +0000

I've been asked to audit AI vendor systems several times now — reviewing what the system is actually doing vs what was sold. The patterns I find are consistent enough to write up.

This post is the engineering guide to running that audit. All code below is for authorised testing of systems you have legitimate access to.

Step 1: Inspect the API Call Pattern

The most revealing thing about any AI system is what external API calls it makes. A system making OpenAI calls is using a third-party LLM. A system making no external model calls is either self-hosted or entirely rule-based. The pattern tells you the architecture.

# Run with: mitmproxy -s ai_audit.py
# Note: only use on systems you are authorised to test

import mitmproxy.http
from datetime import datetime
import json

AI_API_DOMAINS = {
    "api.openai.com": "OpenAI",
    "api.anthropic.com": "Anthropic",
    "api.cohere.ai": "Cohere",
    "generativelanguage.googleapis.com": "Google Gemini",
    "api.together.xyz": "Together AI",
    "api.mistral.ai": "Mistral",
}

class AIVendorInspector:
    def __init__(self):
        self.calls = []

    def response(self, flow: mitmproxy.http.HTTPFlow):
        host = flow.request.pretty_host
        ai_provider = next(
            (name for domain, name in AI_API_DOMAINS.items() if domain in host),
            None
        )
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "host": host,
            "path": flow.request.path,
            "ai_provider": ai_provider,
            "request_bytes": len(flow.request.content),
        }
        if ai_provider and flow.request.content:
            try:
                body = json.loads(flow.request.content)
                entry["model"] = body.get("model", "unknown")
            except Exception:
                pass
        self.calls.append(entry)

    def verdict(self) -> dict:
        total = len(self.calls)
        ai_calls = [c for c in self.calls if c["ai_provider"]]
        ratio = len(ai_calls) / max(total, 1)
        providers = list(set(c["ai_provider"] for c in ai_calls))
        models = list(set(c.get("model","?") for c in ai_calls if c.get("model")))

        if ratio == 0:
            assessment = "NO EXTERNAL AI DETECTED — system appears rule-based or self-hosted"
        elif ratio < 0.10:
            assessment = "MINIMAL AI — LLM called in <10% of operations, likely for formatting only"
        elif ratio < 0.50:
            assessment = "PARTIAL AI — AI used in some operations; investigate which decisions it makes"
        else:
            assessment = "SIGNIFICANT AI — External model calls central to operation"

        return {
            "total_requests": total,
            "ai_api_calls": len(ai_calls),
            "ai_call_ratio_pct": round(ratio * 100, 1),
            "providers_detected": providers,
            "models_detected": models,
            "assessment": assessment,
        }

addons = [AIVendorInspector()]

Key signal: If external model API calls are zero or minimal, the intelligence is rule-based, not learned. If you see only OpenAI calls for formatting at the end of a chain — that's Pattern 1 (rules engine + GPT wrapper).

Step 2: Semantic Consistency Testing

Genuine NLP models produce consistent outputs across semantically equivalent inputs phrased differently. Keyword-matching systems show high variance on paraphrases because they match surface patterns, not meaning.

import random
import string
from typing import Callable

def paraphrase_variants(text: str, n: int = 20) -> list[str]:
    """
    Generates semantically equivalent variants of an input.
    High output variance across these → keyword-based system.
    Low variance → genuine semantic understanding.
    """
    words = text.split()
    variants = []

    strategies = [
        # Add filler words
        lambda t: f"please {t}",
        lambda t: f"I need to {t}",
        lambda t: f"can you help me {t}",
        # Swap non-critical adjacent words
        lambda t: " ".join(t.split()[1:] + [t.split()[0]]) if len(t.split()) > 2 else t,
        # Add trailing context
        lambda t: f"{t} as soon as possible",
        lambda t: f"{t} for our team",
        # Common abbreviations
        lambda t: t.replace("and", "&").replace("with", "w/"),
        # Introduce minor typo (transposition)
        lambda t: t[:5] + t[6] + t[5] + t[7:] if len(t) > 8 else t,
    ]

    for _ in range(n):
        strategy = random.choice(strategies)
        variants.append(strategy(text))

    return list(set(variants))[:n]

def test_semantic_consistency(
    model_fn: Callable[[str], float],
    test_inputs: list[str],
    variance_threshold: float = 0.18
) -> dict:
    """
    model_fn: takes text input, returns a confidence/score float.
    Returns whether the model behaves like keyword matching or semantic understanding.
    """
    results = []
    for input_text in test_inputs:
        variants = paraphrase_variants(input_text, n=15)
        scores = [model_fn(v) for v in variants]
        score_range = max(scores) - min(scores)
        results.append({
            "original": input_text,
            "score_variance": round(score_range, 3),
            "min": round(min(scores), 3),
            "max": round(max(scores), 3),
            "signal": "KEYWORD_BASED" if score_range > variance_threshold else "SEMANTIC",
        })

    keyword_count = sum(1 for r in results if r["signal"] == "KEYWORD_BASED")
    overall = "KEYWORD_MATCHING" if keyword_count > len(results) * 0.5 else "SEMANTIC_UNDERSTANDING"

    return {
        "inputs_tested": len(results),
        "keyword_signal_count": keyword_count,
        "overall_verdict": overall,
        "details": results,
    }

Step 3: Benchmark Validation — Their Set vs Your Data

import numpy as np
from typing import Callable

def compare_benchmarks(
    model_fn: Callable,  # vendor's inference endpoint
    vendor_test_set: list[dict],    # [{"input": ..., "label": ...}]
    your_production_sample: list[dict],  # same format, from your actual data
) -> dict:
    """
    Compares model performance on vendor's curated test set vs your real data.
    Large delta = demo-tuned. Small delta = robust model.
    """
    def evaluate(dataset):
        correct = []
        for item in dataset:
            pred = model_fn(item["input"])
            correct.append(1 if pred == item["label"] else 0)
        return sum(correct) / len(correct) if correct else 0.0

    vendor_acc = evaluate(vendor_test_set)
    prod_acc = evaluate(your_production_sample)
    delta = vendor_acc - prod_acc

    return {
        "vendor_test_accuracy": round(vendor_acc, 4),
        "your_production_accuracy": round(prod_acc, 4),
        "accuracy_delta": round(delta, 4),
        "demo_tuning_risk": (
            "HIGH — significant gap; model likely optimised for demo scenarios"
            if delta > 0.15 else
            "MEDIUM — some gap; worth investigating the test set composition"
            if delta > 0.08 else
            "LOW — performance consistent across datasets"
        ),
        "recommendation": (
            "REQUEST vendor's test set composition. "
            "Insist on production SLA based on YOUR data, not their benchmark."
            if delta > 0.08 else
            "Benchmark appears robust. Still request right-to-audit clause in contract."
        ),
    }

The Audit Report — What to Produce

After these three tests, you have the evidence for a vendor conversation:

AI VENDOR TECHNICAL AUDIT
==========================
Vendor + Product: [name]
Audit Date: [date]
System Under Test: [description]

FINDINGS:
1. API Inspection
   Total requests observed: [X]
   External AI API calls: [Y] ([Z]%)
   Providers detected: [list]
   Assessment: [from verdict()]

2. Semantic Consistency Test
   Inputs tested: [N]
   Keyword-pattern signals: [count]
   Overall: [KEYWORD_MATCHING / SEMANTIC_UNDERSTANDING]

3. Benchmark Comparison
   Vendor test accuracy: [X%]
   Your production accuracy: [Y%]
   Delta: [Z%]
   Demo-tuning risk: [level]

SUMMARY:
[One paragraph honest assessment of what the system is actually doing]

CONTRACTUAL RECOMMENDATIONS:
- Require production accuracy SLA tied to YOUR data, not vendor benchmarks
- Insist on right-to-audit the model architecture
- Add exit clause if performance benchmarks not met within 6 months
- Request quarterly performance reports on production metrics

This report gives you the evidence to renegotiate, exit, or make a justified case for rebuilding with a vendor who can actually deliver.

Sunil — CEO, Ailoitte. We run AI Stack Audits for companies re-evaluating enterprise AI vendors. 2-day turnaround. ailoitte.com

EU AI Act Technical Compliance - What Your Engineering Team Needs to Build Before August 2026

Sunil Kumar — Wed, 22 Apr 2026 06:24:57 +0000

August 2, 2026, is when the EU AI Act becomes fully enforceable for high-risk AI applications.

If your team hasn't started: this post is the engineering implementation guide for the four technical requirements — audit logging, explainability, bias monitoring, and data lineage, all with code.

The regulation applies to you if your AI touches EU users and operates in healthcare, fintech, HR tech, legal tech, or education — regardless of where your company is incorporated.

Requirement 1: Comprehensive Audit Logging

Every AI decision in a high-risk system must be logged: inputs, outputs, context, model version, timestamp. For every decision. Queryable. 5-year minimum retention.

import hashlib, json, uuid
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Any, Optional

@dataclass
class AIDecisionRecord:
    decision_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    model_id: str = ""
    model_version: str = ""
    feature_name: str = ""         # which product feature made this call
    risk_category: str = ""        # e.g. "high_risk_hr", "high_risk_fintech"
    input_hash: str = ""           # SHA-256 of raw input — audit trail without storing PII
    input_type_summary: str = ""   # e.g. "cv_document:3420_chars"
    output_summary: str = ""       # brief of what was decided/returned
    confidence: Optional[float] = None
    user_id_pseudonym: str = ""    # pseudonymised — GDPR + AI Act compatible
    explanation_logged: bool = False

class EUAIActAuditLogger:
    """
    Logs AI decisions in EU AI Act-compliant format.
    Back end: any append-only, tamper-evident store (CloudWatch, BigQuery, etc.)
    """

    def __init__(self, storage):
        self.storage = storage  # inject your storage implementation

    def log(
        self,
        model_id: str,
        model_version: str,
        feature_name: str,
        risk_category: str,
        raw_input: Any,
        output_summary: str,
        user_id: str,
        confidence: Optional[float] = None,
        explanation: Optional[str] = None,
    ) -> str:
        record = AIDecisionRecord(
            model_id=model_id,
            model_version=model_version,
            feature_name=feature_name,
            risk_category=risk_category,
            input_hash=hashlib.sha256(
                json.dumps(raw_input, default=str).encode()
            ).hexdigest(),
            input_type_summary=self._type_summary(raw_input),
            output_summary=output_summary[:500],  # cap length in log
            confidence=confidence,
            user_id_pseudonym=hashlib.sha256(
                f"ailoitte_salt_{user_id}".encode()
            ).hexdigest()[:16],
            explanation_logged=explanation is not None,
        )
        self.storage.append(asdict(record))
        return record.decision_id

    def _type_summary(self, raw_input: Any) -> str:
        if isinstance(raw_input, str):
            return f"text:{len(raw_input)}_chars"
        if isinstance(raw_input, dict):
            return f"structured:{list(raw_input.keys())[:5]}"
        return f"type:{type(raw_input).__name__}"

    def compliance_report(self, start: str, end: str) -> dict:
        """Pull audit log slice for regulatory inspection."""
        records = self.storage.query_range(start, end)
        return {
            "period": {"from": start, "to": end},
            "total_decisions": len(records),
            "by_feature": self._group(records, "feature_name"),
            "by_risk_category": self._group(records, "risk_category"),
            "explanation_coverage_pct": round(
                sum(1 for r in records if r["explanation_logged"]) / max(len(records), 1) * 100, 1
            ),
        }

    def _group(self, records, key):
        from collections import Counter
        return dict(Counter(r[key] for r in records))

Key point: Build this before shipping any production AI decision feature. Retrofitting it into an existing system typically costs 3–4× more engineering time.

Requirement 2: Explainability

You must be able to explain any high-risk AI output in human-readable terms on regulatory request. Two approaches depending on model type:

For structured ML (gradient boosting, tabular neural networks):

import shap, numpy as np

class StructuredMLExplainer:
    def __init__(self, model, feature_names: list[str]):
        self.model = model
        self.features = feature_names
        self.explainer = None

    def fit(self, background_data: np.ndarray):
        self.explainer = shap.TreeExplainer(self.model)

    def explain(self, input_row: np.ndarray, top_n: int = 5) -> dict:
        vals = self.explainer.shap_values(input_row)
        if isinstance(vals, list):
            vals = vals[1]  # positive class for binary classifiers
        flat = vals[0] if vals.ndim > 1 else vals
        ranked = sorted(zip(self.features, flat), key=lambda x: abs(x[1]), reverse=True)[:top_n]
        return {
            "top_factors": [
                {
                    "factor": name,
                    "direction": "increased risk" if v > 0 else "decreased risk",
                    "magnitude": "high" if abs(v) > 0.10 else "medium" if abs(v) > 0.04 else "low",
                }
                for name, v in ranked
            ],
            "plain_english": self._plain(ranked),
            "compliance_basis": "EU AI Act Art. 13 — generated via SHAP feature attribution",
        }

    def _plain(self, ranked):
        pos = [n for n, v in ranked if v > 0][:2]
        neg = [n for n, v in ranked if v < 0][:2]
        parts = []
        if pos: parts.append(f"Factors that raised this score: {', '.join(pos)}")
        if neg: parts.append(f"Factors that lowered this score: {', '.join(neg)}")
        return ". ".join(parts)

For LLM-based systems (RAG, generative AI):

SHAP doesn't apply to transformer models. The current practical approach: log the context that produced the output, so you can say "this decision was based on these retrieved documents."

@dataclass
class LLMDecisionProvenance:
    """
    Stores the 'why' for an LLM decision — the closest EU AI Act-compliant
    explainability for generative systems currently available.
    """
    decision_id: str
    model_id: str
    prompt_template_id: str          # which prompt template was used
    retrieved_sources: list[dict]    # [{"doc_id": ..., "title": ..., "excerpt": ...}]
    user_query_hash: str             # hash of original user query
    output_length: int
    timestamp: str

    def to_explanation(self) -> str:
        sources = ", ".join(s["title"] for s in self.retrieved_sources[:3])
        return (
            f"This output was generated based on the following internal sources: {sources}. "
            f"It used prompt template '{self.prompt_template_id}'. "
            f"Full source text available in audit log reference {self.decision_id}."
        )

Requirement 3: Bias Monitoring Pipeline

Ongoing monitoring across protected characteristics — not just a pre-launch test.

from dataclasses import dataclass
from datetime import datetime
from collections import defaultdict
from typing import Optional

@dataclass
class BiasEvaluationReport:
    model_id: str
    model_version: str
    eval_date: str
    characteristic: str
    group_positive_rates: dict
    disparity_ratio: float          # min_rate / max_rate (1.0 = perfect parity)
    threshold: float                # 0.80 = EU standard (80% rule)
    compliant: bool
    action_required: Optional[str]

class BiasMonitor:
    """Run after every model update and on a monthly schedule in production."""

    EU_PROTECTED = ["gender", "age_group", "ethnicity", "disability_status", "religion"]
    DISPARITY_THRESHOLD = 0.80   # 80% rule — industry standard, aligns with EU law

    def evaluate(
        self,
        model_id: str,
        model_version: str,
        predictions: list[int],
        demographics: dict[str, list],  # {"gender": [...], "age_group": [...]}
    ) -> list[BiasEvaluationReport]:
        reports = []
        for char in self.EU_PROTECTED:
            if char not in demographics:
                continue
            groups = demographics[char]
            rates = defaultdict(list)
            for pred, group in zip(predictions, groups):
                if group:
                    rates[group].append(pred)
            group_positive_rates = {
                g: round(sum(preds) / len(preds), 4)
                for g, preds in rates.items()
                if preds
            }
            if len(group_positive_rates) < 2:
                continue
            vals = list(group_positive_rates.values())
            ratio = round(min(vals) / max(vals), 4) if max(vals) > 0 else 1.0
            compliant = ratio >= self.DISPARITY_THRESHOLD
            reports.append(BiasEvaluationReport(
                model_id=model_id,
                model_version=model_version,
                eval_date=datetime.utcnow().isoformat(),
                characteristic=char,
                group_positive_rates=group_positive_rates,
                disparity_ratio=ratio,
                threshold=self.DISPARITY_THRESHOLD,
                compliant=compliant,
                action_required=None if compliant else "BIAS_REMEDIATION_REQUIRED — file incident report",
            ))
        return reports

    def regulatory_summary(self, reports: list[BiasEvaluationReport]) -> dict:
        return {
            "evaluation_date": datetime.utcnow().isoformat(),
            "overall_status": "COMPLIANT" if all(r.compliant for r in reports) else "NON_COMPLIANT",
            "characteristics_evaluated": len(reports),
            "characteristics_passing": sum(1 for r in reports if r.compliant),
            "details": [
                {
                    "characteristic": r.characteristic,
                    "disparity_ratio": r.disparity_ratio,
                    "status": "PASS" if r.compliant else "FAIL",
                    "action": r.action_required,
                }
                for r in reports
            ],
            "next_scheduled_evaluation": "30 days",
            "standard_applied": "EU AI Act Annex IV + 80% disparate impact rule",
        }

Requirement 4: Data Lineage Documentation

Maintain a data card for every training dataset. Minimum fields:

@dataclass
class DataCard:
    dataset_name: str
    dataset_version: str
    source_description: str
    collection_date_range: str
    total_records: int
    demographic_composition: dict    # known breakdown by protected characteristics
    known_biases: list[str]          # documented gaps or known issues
    quality_assessment: str
    usage_restrictions: list[str]
    last_updated: str
    maintained_by: str

# Example — fills in the EU AI Act Annex IV data documentation requirement
cv_screening_training_data = DataCard(
    dataset_name="cv_screening_v3",
    dataset_version="3.2.1",
    source_description="Historical hiring decisions from 2018–2023, 6 partner companies",
    collection_date_range="2018-01-01 to 2023-12-31",
    total_records=84_000,
    demographic_composition={
        "gender": {"male": 0.62, "female": 0.35, "other/unknown": 0.03},
        "age_group": {"18-30": 0.28, "31-45": 0.41, "46+": 0.31},
        "ethnicity": "not collected — gap documented",
    },
    known_biases=[
        "Gender imbalance in technical roles (62% male in engineering data)",
        "Ethnicity data not collected — bias in this dimension unmeasurable",
        "Seniority overrepresentation from partner companies in financial services",
    ],
    quality_assessment="Reviewed 2024-03. Manual spot-check of 500 records. Outlier rate: 2.1%.",
    usage_restrictions=["HR screening models only", "Not for compensation decisions"],
    last_updated="2024-03-15",
    maintained_by="ML Platform Team",
)

Implementation Order

Week 1: Audit logging
    Build the logging layer FIRST — everything else depends on it.
    Design the schema. Hook into every AI decision path. Set up storage.

Week 2: Explainability
    Structured ML → SHAP integration + plain-English generator
    LLM systems → provenance logging + source attribution

Week 3: Bias monitoring
    Build evaluation pipeline. Define acceptable thresholds.
    Set up monthly scheduled runs + alerting on threshold breach.

Week 4: Data lineage
    Write data cards for all training datasets.
    Set up update procedures for future dataset changes.

Weeks 5–6: Internal review + documentation
    EU AI Act Article 11 technical documentation.
    Conformity assessment (required for highest-risk systems).
    Registration in EU database if required for your category.

One question I'm genuinely still working through: for LLM-based high-risk systems, is provenance logging (the retrieved sources approach above) what regulators will actually accept as "explainability," or is there a more rigorous approach emerging? Curious what others building in regulated spaces have encountered.

Sunil — CEO, Ailoitte. We build EU AI Act compliance infrastructure for funded startups. 4 week delivery. ailoitte.com

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.