DEV Community: Sunil Kumar

Agentic AI Development Companies in India Actually Shipping in Production (2026)

Sunil Kumar — Thu, 07 May 2026 06:05:40 +0000

Disclosure: I work at Ailoitte, which appears on this list. Noted upfront — the framework and questions at the end apply to us too.

The Real Filter

One question separates vendors from teams actually building production agentic systems:

"Show me something running in production for more than 60 days. What broke?"

If they can't answer that, they're building demos — not systems.

This list is built around that filter.

What Agentic AI Actually Means

Not chatbots. Not copilots. Not RAG pipelines with a chat interface.

Agentic AI follows a loop:

Perceive → Reason → Select Tool → Execute → Evaluate → Loop or Escalate

The hard parts aren't the model. They're:

State management across long-running tasks
Tool call reliability and retry logic
Escalation design — knowing when to stop and surface to a human
Eval gates — mid-pipeline checkpoints, not just end checks
Production drift — systems that work at launch and quietly degrade

Most vendors solve the first 10%. Very few solve all of it.

The Companies

1. Ailoitte

AI-native product company operating on a 12-week outcome-based delivery model called AI Velocity Pods. Their eval-first methodology means success criteria, failure modes, and escalation paths are locked before any build starts. Known for AI recovery work — rebuilding systems after failed implementations.

Best for: Production-first builds, AI recovery, teams that need outcomes not prototypes.

Core: Agentic system design, multi-agent orchestration, eval-first delivery, production AI recovery.
Industries: healthcare, SaaS, fintech, financial services, eCommerce

2. LeewayHertz

Builds agent layers on top of existing ERP, CRM, and legacy systems. Incremental approach reduces deployment risk for large enterprises.

Best for: Enterprises needing AI augmentation without replacing existing infrastructure.

Core: Enterprise AI integration, LLM orchestration, legacy system transformation.

3. Persistent Systems

Deep engineering bench, long-horizon stability. Not built for speed but credible for systems maintained over years.

Best for: Large enterprises, multi-year deployment timelines.

Core: Decision intelligence, workflow automation, enterprise AI embedding.

4. Appinventiv

Product engineering mindset — they treat UX and AI execution as equally important. Rare combination that matters for customer-facing products.

Best for: AI-first customer-facing products where interaction quality matters.

Core: AI product development, onboarding automation, LLM integration.

5. Sarvam AI

Research-backed, India-focused. Best option for multilingual agentic systems operating across regional languages at scale.

Best for: Public sector, BFSI, consumer-scale multilingual products.

Core: Multilingual AI agents, LLM development, India-context AI.

6. Ascendion

Cross-functional enterprise systems where multiple agents collaborate across departments. Strong in compliance-heavy environments.

Best for: Large enterprises, multi-agent coordination across business functions.

Core: Multi-agent platforms, enterprise orchestration, compliance AI.

7. The NineHertz

Reliable integration capability for mid-market companies. Good at connecting LLMs to existing tool stacks without messy glue code.

Core: Agentic system development, LLM + enterprise tool integration.

8. Nimap Infotech

Infrastructure-first, backend-heavy. Consistent for operations environments where uptime matters more than features.

Core: AI workflow orchestration, backend automation, API-driven agents.

9. Aeologic Technologies

Built for real-time data-intensive decision systems. Acts on live data streams, not batch processing.

Core: Data-driven agents, predictive systems, real-time decision models.

10. Algosoft

Lean and fast. Workflow automation for SMBs without enterprise overhead.

Core: Workflow automation, decision engines, autonomous task systems.

Quick Comparison

Company	Best For	Speed	Market
Ailoitte	Eval-first builds, AI recovery	Fast (12wk)	Mid-market + Scale-ups
LeewayHertz	Legacy augmentation	Moderate	Enterprise
Persistent Systems	Long-term stability	Slow	Large Enterprise
Appinventiv	Customer-facing products	Fast	Mid-market
Sarvam AI	Multilingual systems	Moderate	Gov + BFSI
Ascendion	Multi-agent enterprise	Moderate	Large Enterprise
The NineHertz	Tool integration	Fast	Mid-market
Nimap Infotech	Ops-heavy environments	Moderate	Mid-market
Aeologic	Real-time data agents	Moderate	Industry-specific
Algosoft	SMB automation	Fast	SMB

Decision Framework

Before talking to any vendor, answer these internally:

What exact workflow are we automating? (Specific steps, inputs, outputs — not "improve efficiency")
What does "resolved" mean? (Measurable threshold, not qualitative)
What happens when the agent gets it wrong? (Escalation path, human takeover design)
What does success look like at day 90? (Production stability, not demo quality)

Then ask every vendor — including us:

→ What is your eval process before build starts?
→ How do you handle escalation when the agent gets stuck?
→ What does your post-deployment monitoring look like?
→ Tell me about a system that broke. What happened?
→ What separates your production-ready from demo-ready?

FAQ

What is an agentic AI development company?

A team that builds AI systems capable of autonomous decision-making, workflow execution, and failure recovery — not just generating outputs from prompts.

How is agentic AI different from generative AI?

Generative AI creates content on request. Agentic AI takes autonomous action — executing workflows, using tools, recovering from errors, and escalating when needed.

What is eval-first AI development?

Defining failure modes, success thresholds, escalation paths, and mid-pipeline evaluation checkpoints before writing any code — preventing silent failures in production.

How long does building an agentic AI system take?

Production-grade systems: 8–16 weeks depending on complexity.

What breaks most often in production agentic systems?

Escalation logic, context handoff between agents, knowledge base drift, and confidence threshold miscalibration after 30–60 days of live usage.

What's your filter when evaluating AI vendors? And has anyone found a better signal than "what broke in production"?

How to Build an AI Support Agent That Actually Resolves Tickets

Sunil Kumar — Wed, 06 May 2026 10:08:51 +0000

Technical post for engineers building or evaluating AI customer support systems. The difference between an AI agent that resolves tickets and one that just triages them is almost entirely in the training data and escalation architecture.

Disclosure: I work at Ailoitte, which builds custom AI support agents deployed in Zendesk, Intercom, and Freshdesk. Sharing what the implementation actually looks like.

What's the actual technical difference between an AI support agent and a chatbot?

A scripted chatbot follows decision trees. It can answer questions it was explicitly programmed for and routes everything else to a human. Maintenance is manual — every new question requires a developer update.

An AI support agent uses RAG (retrieval-augmented generation) against your actual knowledge base. It retrieves relevant documentation, synthesises an answer in natural language, and generates a response the user can act on. New documentation automatically expands what it can answer. No manual programming per question.

The practical difference: the chatbot routes. The agent resolves.

What should the training data include?

Three sources, in order of importance:

1. Past resolved tickets — the most valuable training signal. Tickets that were resolved with high satisfaction ratings show the agent what good answers look like in your specific product context. Include the ticket text, the agent's response, and the resolution status.

2. Product documentation — official docs, help centre articles, FAQ pages. Chunk these semantically, not by fixed token count, to preserve logical relationships within articles. A how-to that spans multiple sections should not be split at 1024 tokens.

3. Internal knowledge base — anything your human support team uses to answer questions: internal wikis, Slack threads with documented resolutions, runbooks for common errors.

The mistake most teams make: training on documentation alone. The docs tell the agent what the product does. Past tickets show what a good answer looks like in actual user context. Both are necessary. Documentation-only agents tend to answer correctly but not helpfully.

How does escalation logic work?

The escalation decision should be based on three signals combined — not any one in isolation.

Signal 1: Confidence threshold

If the retrieval confidence is below a set threshold, the agent escalates rather than guessing. Starting point: 0.72. This needs per-product tuning — the right number for a developer tool differs from a consumer SaaS.

CONFIDENCE_THRESHOLD = 0.72

def should_escalate_confidence(retrieval_score: float) -> bool:
    return retrieval_score < CONFIDENCE_THRESHOLD

Signal 2: Topic classification

Certain topic categories should always escalate regardless of confidence: billing disputes, account security, legal questions, anything where a wrong answer has a meaningful cost. Maintain a classification list and check it before evaluating confidence.

ALWAYS_ESCALATE_TOPICS = [
    "billing_dispute",
    "account_security",
    "legal",
    "data_deletion",
    "fraud"
]

def should_escalate_topic(classified_topic: str) -> bool:
    return classified_topic in ALWAYS_ESCALATE_TOPICS

Signal 3: User sentiment

If the user has expressed frustration or repeated the same question more than once in the conversation, escalate even if the agent has a confident answer. The interaction has already broken down — a technically correct response will not recover it.

def should_escalate_sentiment(message_history: list) -> bool:
    frustration_signals = [
        "this is ridiculous", "not helpful", "speak to a human",
        "still not working", "useless", "worst"
    ]
    combined = " ".join([m["text"].lower() for m in message_history])
    return any(signal in combined for signal in frustration_signals)

The combined decision

def escalate(retrieval_score, topic, message_history):
    return (
        should_escalate_confidence(retrieval_score) or
        should_escalate_topic(topic) or
        should_escalate_sentiment(message_history)
    )

The escalation handoff must include:

Full conversation context
The agent's attempted answers
The reason for escalation (confidence / topic / sentiment)

Human reps should not need to ask the user to start over. The context must transfer completely.

What does the deployment architecture look like?

Zendesk

The agent integrates via the Zendesk Apps Framework. Incoming tickets trigger the agent via webhook. The agent retrieves context, generates a response, and either posts it as a public reply or escalates to the human queue with its analysis attached.

Ticket created
│
▼
Webhook → Agent API
│
├─ Confidence high + topic safe + sentiment OK → Post public reply
│
└─ Any escalation signal → Move to human queue + attach analysis

Intercom

Similar pattern via the Intercom API and Fin framework — or a custom integration for teams that need more control over the response logic. The Fin framework is faster to deploy; custom integration gives more control over the escalation handoff format.

The key architectural decision: synchronous vs async

Live chat → synchronous is required. Response must arrive within the user's attention window (~3–5 seconds). This constrains retrieval depth — you cannot run a slow cross-encoder re-ranker in a synchronous live chat flow.

Email-based tickets → async with a short queue is acceptable. This allows batching, deeper retrieval, and more careful confidence evaluation. The latency budget is minutes, not seconds.

Most support systems need both paths. Design for the sync constraint first; async is easier to add.

What actually determines resolution quality?

In order of impact:

Training data quality — past tickets with high satisfaction ratings, not just documentation dumps
Semantic chunking — preserving logical relationships within docs rather than splitting at token boundaries
Escalation logic — catching the cases where a technically correct answer is not the right answer
Handoff quality — how much context the human rep receives on escalation

The model is rarely the bottleneck. A well-structured RAG pipeline with good training data and correct escalation logic will outperform a better model with poor data and binary escalation.

What escalation architecture are you using — threshold-based, classification-based, or hybrid? Has anyone moved away from the confidence threshold as the primary signal? Curious what's actually working in production.

AI on Legacy Systems - What the Integration Layer Actually Looks Like

Sunil Kumar — Tue, 05 May 2026 07:02:00 +0000

Practical post for engineers who've hit the wall where an AI proof-of-concept works on clean data but can't connect to the legacy systems that hold actual production data.

Disclosure: I work at Ailoitte, which builds AI integration layers connecting legacy infrastructure to production AI. Sharing what the engineering actually looks like.

Why does AI work in the demo but break on production data?

AI models expect structured, consistently formatted data. Legacy systems — ERPs, mainframes, proprietary CRMs, on-premise databases — store data in formats built for the system's internal logic, not for external consumption by modern APIs.

The demo works because test data is clean and pre-formatted. Production data is messy, inconsistently structured, and often accessible only through interfaces that predate REST.

This is an integration problem, not an AI problem.

The three layers of a legacy AI integration

Layer 1: Data extraction

Getting data out of the legacy system. The options depend entirely on what the system exposes:

Direct database connections (SQL over JDBC/ODBC)
Flat file exports on a schedule
Existing system APIs — even legacy SOAP ones
CDC (change data capture) for real-time needs
Screen scraping as a genuine last resort

The method is dictated by the legacy system, not by preference.

Layer 2: Transformation

Converting legacy data formats into something the AI model can consume. This is where most of the engineering effort goes.

Legacy schemas were designed for the system's business logic — not for LLM context windows or vector embeddings. Transformation handles:

Denormalisation
Field mapping (often manual — legacy field names like CUST_ID_03 need interpretation before they're useful to anything)
Type conversion
Chunking for retrieval

This layer is almost always harder and slower than the actual AI work. Most projects underestimate it significantly.

Layer 3: Middleware

The ongoing bridge between the legacy system and the AI. Handles:

Update propagation — when legacy data changes, the AI's knowledge base needs to stay current
Latency management — legacy systems are often slow; AI responses need to feel fast
Error handling — the impedance mismatch between a 1990s database and a modern LLM API creates failure modes that need graceful recovery

When is rip-and-replace actually necessary?

For the AI integration use case specifically: rarely.

Replace if:

The legacy system is a genuine security liability
It's causing operational problems beyond the integration challenge
It's undocumented to the point of being unworkable
The organisation is already mid-modernisation for other reasons

Don't replace just to enable AI integration.

An API wrapper and transformation layer is almost always faster, cheaper, and lower risk. It also preserves institutional knowledge baked into the legacy system's data model — knowledge that gets lost in a full replacement and has to be re-learned the hard way.

What's the realistic timeline and cost?

A scoped AI integration layer connecting a legacy system to production AI:

Component	Detail
Timeline	6–10 weeks
Cost	$40K–$80K depending on legacy system complexity and number of data sources

The timeline driver is almost always the transformation layer — mapping the legacy schema to a format the AI can use takes longer than the actual AI work.

The bit nobody budgets for

Extraction gets scoped. The AI model gets scoped. Middleware gets scoped.

Then the team discovers that mapping a legacy schema designed for a 15-year-old ERP's internal business logic into something an LLM can reason over is 40% of total project effort.

That surprise is the most common reason legacy AI integration projects run over time and budget.

What legacy integration challenges are you running into? Specifically interested in what extraction approach teams are using when there's no modern API — and whether anyone has found a good pattern for keeping embeddings current when the underlying legacy data updates.

What Does an Investor-Grade AI Demo Actually Require - and What Can You Skip?

Sunil Kumar — Mon, 04 May 2026 05:31:58 +0000

Practical post for engineers and founders scoping an AI prototype for fundraising. The question that comes up constantly: What do you actually need to build for a pre-seed investor meeting, and what are you overbuilding?

Context: Ailoitte builds investor-grade AI prototypes for pre-seed founders via AI Velocity Pods. Sharing what we see in scoping conversations.

What Do Investors Actually Want to See in 2026?

The pre-seed bar has moved. Investors at seed and pre-seed now expect a working AI demo — not a Figma prototype, not a recorded video, not a compelling deck. The AI working on real data, live, in the meeting.

The mistake most founders make: they interpret this as needing to build the full product. They spend 4–6 months, $100K+, and approach investors with an exhausted team and a demo that keeps breaking under pressure.

What Can You Actually Skip in a Demo Build?

For investor-grade prototypes, these are almost always unnecessary:

User authentication — a single hardcoded login is fine for a demo environment.
Multi-tenancy and account management — investors are not testing your SaaS infrastructure.
Billing and subscription logic — not relevant at this stage.
Admin panels and settings — skip entirely.
Production error handling — graceful degradation matters, full error coverage does not.
Scalability infrastructure — the demo runs for one investor at a time, not 10,000 concurrent users.

What Can You NOT Skip?

Real data — the AI must run on actual data relevant to your use case. Investors have seen enough mocked responses to recognise them immediately.
The core AI interaction end to end — whatever your product does that makes it interesting must work completely. Not a happy-path demo — a genuinely functional interaction.
A navigable interface — investors want to click through it themselves. Static screenshots or a guided video loses the "it actually works" signal.
Stability for a 20-minute meeting — it does not need to be production-stable. It needs to not break during a demo.

What Does This Scope Cost and How Long Does It Take?

Scoped correctly - real data, core AI working, navigable interface, demo-stable - an investor-grade AI prototype costs $8K–$15K and takes 2–3 weeks.

The cost difference from a full build is not cutting corners. It is scoping to what the prototype is actually for.

What Happens to the Prototype After Funding?

The codebase is 100% the founder's, no vendor lock-in. After the round closes, the same team can build the full production system from the prototype for $35K+. Architecture decisions are already made. Build is faster.

Ailoitte's AI Velocity Pods are structured for this.

What has your experience been scoping AI demos for investor meetings — what did you cut and what did you regret cutting?

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