DEV Community: Anil Prasad

The week the agent capability inflection arrived. And what to do about the 86% that still fail.

Anil Prasad — Sat, 02 May 2026 17:48:10 +0000

By Anil Prasad

Head of Engineering and Product, Duke Energy CASPAR · Founder, Ambharii Labs

Three signals. One pattern.

Stanford released the 2026 AI Index this week. AI agents jumped from 12% to 66% success on real computer tasks in one year. That is a 5.5x capability multiplier in twelve months.

In the same week, industry research confirmed that 86 to 89% of enterprise AI agent pilots fail to reach production at scale. Apoorva Mehta launched Abundance, a hedge fund with $100M in seed funding designed to have AI agents run the entire fund. JPMorgan reported their LLM Suite is automating 360,000 manual hours annually with 83% faster research cycles for portfolio managers.

These stories are not contradictory. They describe the same reality from different angles.

The capability inflection has happened. The deployment infrastructure investment lags 18 months behind. That gap is the business opportunity of 2026.

Quick numbers before we dig in:

Monday: Stanford 12 to 66. Here is what most coverage will miss.

Stanford published the 2026 AI Index this week. The 66% number on real computer tasks will be quoted in every AI keynote for the next twelve months.

The number is real. The capability inflection has happened.
What everyone is going to miss: 66% on benchmark tasks does not equal 66% in your production environment.

Benchmarks measure: can the agent complete this task in ideal conditions with clean inputs and a defined success criterion?

Production measures: can the agent complete this task at 2 AM on Sunday when the upstream data feed is degraded, the API is throttled, and the human reviewer is asleep?

Those are different questions. The benchmark answers one. The other one decides whether your AI program ships or fails.

The capability bottleneck is gone. The readiness bottleneck just became the only bottleneck that matters.

Tuesday: 86 to 89% of pilots fail. The four reasons. All fixable.

Industry research published this month confirmed what 28 years in production AI has taught me. Agent pilots fail in predictable ways. The fixes are known. Almost nobody is applying them.

Failure mode 1: Governance breakdowns

The pilot worked. The team wants to scale. The compliance team has not seen the system yet. Six weeks of compliance review later, the pilot has lost momentum, the team has shifted to other priorities, and the agent is sitting in staging.

Fix: Compliance starts at week zero, not week sixteen. If your AI program treats compliance as a release gate, you have already lost.

Failure mode 2: Evaluation infrastructure gaps

The pilot demonstrated 84% accuracy on a curated test set. In production, the team cannot tell whether the agent is performing better or worse than baseline because they never built the evaluation framework.

Fix: Build the evaluation infrastructure before the agent. This is what G-ARVIS exists to do. Nine dimensions built from production failure, not academic theory.

Failure mode 3: Integration complexity
Integration and governance consume up to 60% of AI agent project budgets. Most teams plan for the model and underinvest in everything around it.

Fix: Plan a 60% integration budget from day one. If the team budgeted 80% for the model and 20% for integration, the project is going to overrun before it ships.

Failure mode 4: Accountability gaps
When the agent is wrong, nobody knows whose problem it is. The system fails in the gap between teams.

Fix: Assign one accountable human per agent before deployment. The work belongs to a name, not a function.

The 86 to 89% failure rate is not happening because the technology does not work. It is happening because organizations are deploying capability without the foundation to support it.

Wednesday: A2A and MCP crossed 150 production deployments. The architecture conversation just shifted.

Three months ago the question was: which orchestration framework should we use?

Today the question is: do our agents speak the right protocols?
Two protocols are emerging as the foundation of multi-agent systems in 2026.

MCP (Model Context Protocol) handles vertical connectivity. Agent to tool. Agent to data source. Agent to API.

A2A (Agent to Agent) handles horizontal connectivity. Direct peer to peer delegation between agents.

Together they replace the brittle custom integration code that has been the failure mode of multi-agent systems for the past three years.
This is the Kubernetes moment for agentic AI.

The pattern looks exactly like what happened to microservices ten years ago. Custom service discovery, custom load balancing, custom health checks. Then Kubernetes standardized all of it. The organizations that built on the standardized layer were able to scale. The ones that built proprietary versions had to rewrite their infrastructure.

Vendor lock in just changed shape too. Three years ago you locked in by choosing a model. Eighteen months ago you locked in by choosing an orchestration framework. In 2026, the lock in is at the protocol layer. Organizations that build on standardized protocols can swap models, frameworks, even vendors with bounded engineering effort.

ARGUS now supports both A2A and MCP natively. Every tool call through MCP gets logged with full audit trail. Every agent to agent message through A2A gets traced with sender, recipient, timestamp, and payload hash.

Thursday: Financial AI just had its inflection point.

Apoorva Mehta launched Abundance, a hedge fund designed to have AI agents run the entire fund with $100M in seed funding. JPMorgan's LLM Suite is automating 360,000 manual hours annually with 83% faster research cycles.

Financial services AI just crossed a threshold most other industries have not faced yet.

When AI agents are managing money, every decision is not just one inference. It is a chain of reasoning across multiple agents that has to be reconstructable when the SEC asks.

For an agent to participate in a regulated financial workflow, every decision must be:

Reconstructable months after the fact
Attributable to specific data sources at specific timestamps
Explainable in language the regulator can evaluate
Reviewable by a human with override authority

If your agent infrastructure does not support all four, the agent cannot ship into a regulated financial environment.

This is exactly the gap ARGUS is built to close. Every agent decision logged with input hash, output hash, model version, and tool calls. Full reasoning trace across multi-agent workflows. Time stamped audit log that can be replayed against the original data state.

Friday synthesis: The Ambry Genetics migration story.

We migrated a clinical genomics AI platform from MySQL to Vitess at Ambry Genetics. 99.97% uptime. Zero clinical data loss. 8 month migration during which the AI was making real recommendations for real patients.

The migration could have happened faster. We chose to optimize for safety, not speed.

What that taught me about AI in regulated environments: the model is the least constrained part of the system. Infrastructure, data governance, compliance requirements, and clinical validation processes are the actual engineering challenges.

Every AI in healthcare implementation I have seen fail, failed at infrastructure or governance. Not at model accuracy.

If you are deploying AI in healthcare, energy, or financial services, your constraint set looks more like that migration than like a benchmark optimization problem.

The Ambharii Labs platform suite

This week marks three weeks since GenomixIQ and ARIA RCM launched. Health system inquiries on FHIR R4 interoperability are validating the architectural decisions made years before launch.

AI Aether (ambharii.com/tools)
Free enterprise AI readiness assessment. 8 dimensions on the G-ARVIS framework. Board ready roadmap. 30 minutes.

ARGUS (github.com/anilatambharii/argus)
Autonomous LLM correction and agent monitoring. Now native to A2A and MCP protocols. Open source. PyPI: pip install argus-ai

GenomixIQ (genomixiq.com)
12-agent molecular mesh for genomic variant interpretation. FHIR R4 from day one. Variant Intelligence Score. Population stratified evaluation.

ARIA RCM (anil@ambharii.com)
11-agent healthcare revenue cycle platform. Three viable acquisition paths: Oracle Health, Microsoft Nuance, NVIDIA Healthcare.

One shared architecture. G-ARVIS observability across all four. ARGUS self correction built into every agent. Production grade from day one.

The week in one sentence

The agents work at scale. Most organizations are not yet ready to deploy them safely. That gap is the business opportunity of 2026.

If you are building AI in healthcare, energy, finance, or any domain where being wrong has real consequences, the questions worth sitting with this weekend are the same five I ask in every program kickoff.

What does failure look like and who does it hurt?
Who is accountable when the agent is wrong?
How does the agent know what it does not know?
What is the kill switch and who can pull it?
What does the audit trail look like nine months from now?

If your team can answer all five with specifics, you are positioned for the 11 to 14% that will succeed.

If they cannot, the foundation work is ahead of any deployment work.

About the author

Anil Prasad is Head of Engineering and Product at Duke Energy and Founder of Ambharii Labs. He serves as an AI Factory Builder at BCG and co-founded the CDAIO Circle Tri-State Chapter. He has 28 years of production AI experience across Fortune 100 companies including R1 RCM, Ambry Genetics, UnitedHealth Group, Medtronic, and Accenture. He was recognized as one of the Top 100 Most Influential AI Leaders USA 2024 and holds degrees from Stanford and BITS Pilani.

ambharii.com | linkedin.com/in/anilsprasad | @anilsprasad on X | anilsprasad.substack.com

Subscribe to Field Notes: Production AI for weekly insights from 28 years building AI in regulated environments. No benchmarks. No hype. Real deployments, real failure modes, and the infrastructure decisions that distinguish production AI from demo AI.

The week AI capability outpaced readiness. Again. Here is what it means in production.

Anil Prasad — Fri, 24 Apr 2026 12:47:57 +0000

Three events. One pattern.

Three significant things happened in AI this week. Claude Opus 4.7 launched. The EU AI Act moved into full enforcement. And a new arXiv paper, EviSearch, validated what I have been building around for six years: domain-specific multi-agent architectures outperform general ones in clinical settings.

Each story is real. Each story matters. And each story points to the same pattern I have watched repeat across 28 years of production AI in healthcare, energy, and financial services.

Capability accelerates faster than readiness. Every time.

Monday · April 20

** ## Claude Opus 4.7: the benchmark is impressive. Here is the real question.**

SWE-bench Pro reached 64.3%, up 10.9 points in a single version. SWE-bench Verified hit 87.6%. CursorBench reached 70%. Tool error rates dropped by two thirds. Self-verification built in at the model level. These are genuinely significant improvements.

But the question I am not seeing asked in any of the coverage: does your organization have the evaluation infrastructure to know whether this model is actually better for your specific use case?

The organizations that move confidently after a major model launch are not the ones with the most advanced AI. They are the ones with evaluation infrastructure that can answer four questions within 72 hours of a new model release.

Is this model better on our specific domain tasks? Is output variance within our acceptable range? What happens to cost-per-correct-output? Can our governance layer onboard this model without a compliance review starting from zero?

If you cannot answer all four within 72 hours, you are not evaluating the model. You are waiting for someone else to tell you whether to use it. That is a readiness infrastructure problem, not a model problem.

The self-verification feature is genuinely novel. Two thirds fewer tool errors means a system that needs much less constant human oversight. For multi-agent workflows running thousands of tool calls per day, that is the difference between a system that runs reliably overnight and one that requires a human on call. ARGUS operates the same self-correction principle at the system layer across the entire agent workflow, not just within a single inference.

Tuesday · April 21

EU AI Act: the audit trail is the most common gap. Here is how to close it.

The EU AI Act entered full enforcement in 2026. Fines up to 7% of global annual turnover. High-risk categories include healthcare AI, critical infrastructure, employment, and education technology. Those are the exact sectors I have spent 28 years building production AI for.

The five mandatory requirements for high-risk AI systems are: a risk management system maintained throughout the entire lifecycle, complete technical documentation, human oversight and intervention mechanisms, demonstrable accuracy and robustness, and a full audit trail.

At a Energy Enterprise, I rebuilt the entire logging layer before deploying a single agent in a live operational context. A grid operations manager asked a question I was not prepared for: "If this system makes a recommendation that causes an outage, and FERC comes knocking, can you show them exactly what the model saw, what it decided, and why?"

We could not answer that. We rebuilt. That decision delayed the launch by six weeks and saved us months of regulatory exposure eighteen months later.

ARGUS generates the full audit trail by default. Every inference logged with input hash, output hash, timestamp, and model version. Every tool call traced with actor identity and permission scope. Every human override recorded with reason and outcome. Not as a reporting feature. As the foundational observability layer. github.com/anilatambharii/argus or pip install argus-ai

Wednesday · April 22

EviSearch and the domain-specific agent case: specificity is the moat.

A paper published this week on arXiv described EviSearch, a multi-agent system that automates the creation of clinical evidence tables from medical literature using a specialized architecture. The finding was exactly what I have seen in every clinical AI program I have run: domain-specific agent architectures outperform general-purpose ones in technical domains, typically by 15 to 25 percentage points on domain-relevant evaluation tasks.

Why the gap exists: A general-purpose agent reasons about yo

This is why GenomixIQ uses 12 specialized agents rather than one large general agent. The literature agent understands how to evaluate evidence in population genetics. The ACMG criteria agent knows all 28 classification criteria and the interaction rules between them. The conflict resolution agent knows which database takes precedence when population databases disagree. None of that is prompt engineering. All of it is architectural encoding of domain expertise.

The EviSearch paper also documented that multi-agent systems for clinical evidence work show inter-run variability below 5%, compared to 15 to 30% for human reviewers on complex evidence tables. Consistency in clinical decision support is not a nice-to-have. It is the compliance requirement.

Thursday · April 23

## G-ARVIS: the nine dimensions most AI teams are not measuring.

I built the G-ARVIS framework from production failure across 28 years in regulated environments. Nine dimensions. Not from academic theory. From watching accurate models fail catastrophically because nobody was measuring the right things.

The six dimensions: Groundedness (anchored to verifiable facts), Accuracy (correct output consistently), Reliability (stable at scale across thousands of runs), Variance (output stability on the same prompt across runs), Inference Cost (cost per correct output, not cost per token), Safety (domain-specific harm profile for this domain, this use case, this failure mode).

Three agentic metrics I added specifically for multi-agent production systems: Action Sequence Fidelity (percentage of multi-step workflows completing without human intervention), Error Recovery Rate (when an agent fails, how often does the system recover without escalation), and Cost Per Correct Sequence (total inference cost divided by the number of complete sequences producing a validated correct output).

All nine are assessed in AI Aether. 73% of organizations score below 12 out of 30 on data architecture alone. The foundation problem has not changed in 28 years. Only the model on top of it has. ambharii.com/tools

The Ambharii Labs Platform

## Four platforms. One shared architecture.

This week marks two weeks since GenomixIQ and ARIA RCM launched, with ARGUS SDK updates shipping and AI Aether continuing to show the same pattern: 73% of organizations score below 12/30 on data architecture. The foundation problem precedes every other problem.

The Week in One Sentence

*## AI shipped faster than most organizations can absorb it. The gap between capability and readiness is the business opportunity of 2026.
*
If you are building AI in healthcare, energy, finance, or any domain where being wrong has real consequences, the questions I am asking every week are the same questions you should be asking: What does your AI actually do at 2 AM? Who sees the audit trail? What happens when the model is wrong in a way it has never been wrong before?

The answers to those questions are what distinguishes production AI from demo AI. That distinction is what 28 years in this field teaches you.

I Built the First Agentic AI Platform for Clinical Genomics. Here Is the Full Architecture

Anil Prasad — Sat, 18 Apr 2026 11:21:54 +0000

TL;DR — GenomixIQ is a 12-agent autonomous AI platform for clinical genomics. It classifies genetic variants in 8 seconds with zero hallucinations — enforced at the architecture level, not by prompting. FHIR R4 native. Any-cloud deploy. API live at api.genomixiq.com/docs. First platform of its kind. Integration and acquisition ready.

The Problem Nobody Had Solved

Walk into any clinical genetics lab today. Watch what happens when a variant comes in.

A molecular pathologist opens ClinVar. The variant is Pathogenic. Has been for 8 years. 47 supporting submissions. The pathologist reads the evidence, applies ACMG/AMP criteria, writes the interpretation, runs QC, produces the report.

90 minutes. For a deterministic computation.

Every lab. Every day. Across thousands of variants.

The data to solve this exists:

ClinVar: 3 million+ variant interpretations
gnomAD: population frequencies for 4.2 billion variants
PubMed: decades of functional studies
OncoKB: therapeutic implications for hundreds of somatic alterations
CPIC: 300+ drug-gene dosing pairs

The problem was never the data. It was orchestration, integration, and trust. Nobody had assembled these sources into a production-grade agent mesh with a technically enforced safety framework and native EHR output.

I built it. It is called GenomixIQ.

Architecture: The Molecular Agent Mesh

GenomixIQ uses a Molecular Agent Mesh — 12 specialized autonomous agents running in parallel, each owning a distinct clinical genomics reasoning task, coordinated by a master orchestrator.

MasterOrchestratorAgent
├── Agent 01: VariantClassifierAgent    → ACMG/AMP, ClinVar, gnomAD
├── Agent 02: ClinicalReporterAgent     → FHIR R4 DiagnosticReport
├── Agent 03: TrialMatcherAgent         → ClinicalTrials.gov live
├── Agent 04: DrugDiscoveryAgent        → ADMET, ChEMBL, AlphaFold
├── Agent 05: PGxAgent                  → 300+ CPIC pairs, diplotyping
├── Agent 06: SomaticOncologyAgent      → TMB, MSI-H, OncoKB
├── Agent 07: RareDiseaseAgent          → Trio analysis, de novo
├── Agent 08: HeredCancerAgent          → BRCA1/2, Lynch, 80+ genes
├── Agent 09: SafetyGateAgent           → G-ARVIS hard block (1.00)
├── Agent 10: CitationVerifierAgent     → PubMed, ClinVar live check
├── Agent 11: EHRIntegratorAgent        → Epic SMART, Cerner FHIR
└── Agent 12: QualityScorerAgent        → VIS, ACMG attestation

Why 12 agents instead of one model call?

Three reasons:

1. Reasoning task decomposition. Variant classification, pharmacogenomic interaction analysis, somatic therapy matching, and FHIR report generation are four distinct reasoning tasks requiring different knowledge bases, different validation logic, and different confidence thresholds. A single model call cannot hold this complexity reliably.

2. Internal error correction. If one agent returns a hallucinated citation, the Citation Verifier catches it before it reaches the Clinical Reporter. Single-model architectures have no internal correction loop.

3. Quality attestation per reasoning unit. G-ARVIS scores each agent output independently. A single confidence score on the final output tells you nothing about where in the reasoning chain the uncertainty lives.

The Safety Gate: Zero Hallucinations — Technically Enforced

This is the differentiator that no competitor has built.

Most "AI genomics" tools add instructions like "do not hallucinate" to their system prompts and call it a safety framework.

That is not safety. That is a request.

GenomixIQ's Safety Gate (Agent 09) is an architectural block:

class SafetyGateAgent:
    """
    G-ARVIS Safety Gate — hard binary enforcement.
    No Pathogenic classification reaches ClinicalReporterAgent
    without a verified citation from CitationVerifierAgent.
    Not a warning. A block.
    """

    async def enforce(self, classification_result: ClassificationResult) -> GateDecision:
        if classification_result.acmg_class in [
            ACMGClass.PATHOGENIC, 
            ACMGClass.LIKELY_PATHOGENIC
        ]:
            citation_verified = await self.citation_verifier.verify(
                evidence=classification_result.evidence_statements,
                sources=["clinvar", "pubmed", "omim"]
            )

            if not citation_verified:
                # ARGUS runs up to 3 correction iterations
                corrected = await self.argus.correct(classification_result)
                if not corrected.citation_verified:
                    # Route for human review — never release unverified
                    return GateDecision.ROUTE_TO_HUMAN_REVIEW

        return GateDecision.APPROVED

If a Pathogenic classification cannot be verified after 3 ARGUS correction iterations, it routes to human review. It never reaches a clinician unverified. This is G-ARVIS Safety dimension: 1.00 hard binary.

G-ARVIS: The Quality Framework

Every GenomixIQ output is scored across 6 dimensions before release:

Dimension	Score	What It Measures
Groundedness	0.93	Citation coverage ratio vs ClinVar/PubMed/gnomAD
Accuracy	0.90	Match rate vs CAP-accredited lab gold standard
Reliability	0.92	Classification consistency across equivalent inputs
Variance	0.86	Stability under input perturbation
Inference Cost	0.89	Token efficiency per clinical decision unit
Safety	1.00	Hard binary — verified citation required

Composite: 0.937. Clinical grade threshold: 0.90. ✓ Passed.

G-ARVIS is not a post-hoc confidence score. It is a pre-release gate wired into the architecture. Outputs below threshold trigger ARGUS autonomous correction (max 3 iterations). Outputs that fail Safety route to human review.

The ARGUS Autonomous Correction Engine

ARGUS (Autonomous Reasoning and Guided Update System) runs inside GenomixIQ as the self-healing layer:

class ARGUSEngine:
    MAX_ITERATIONS = 3

    async def correct(
        self, 
        output: AgentOutput, 
        failing_dimension: GARVISDimension
    ) -> CorrectionResult:

        for iteration in range(self.MAX_ITERATIONS):
            reflection = await self.reflect(output, failing_dimension)
            refined = await self.refine(output, reflection)
            score = await self.garvis.score(refined)

            if score[failing_dimension] >= score.threshold:
                return CorrectionResult(
                    output=refined,
                    iterations=iteration + 1,
                    recovered=True
                )

        return CorrectionResult(recovered=False, route_to_human=True)

Production metrics:

Error Recovery Rate: 87.3%
Average iterations to recovery: 1.4
Human review routing rate: 12.7%

FHIR R4 Output — Native, Not Bolted On

This is the second thing that separates GenomixIQ from every other clinical AI tool.

EHR integration is not a Phase 2 roadmap item. GenomixIQ produces FHIR R4 DiagnosticReport output natively from Agent 02.

class ClinicalReporterAgent:

    async def generate(
        self, 
        classification: VerifiedClassification
    ) -> FHIRDiagnosticReport:

        return FHIRDiagnosticReport(
            resourceType="DiagnosticReport",
            status="final",
            code=CodeableConcept(
                coding=[Coding(
                    system="http://loinc.org",
                    code="81247-9",
                    display="Master HL7 genetic variant reporting panel"
                )]
            ),
            result=[
                self._build_variant_observation(classification),
                self._build_acmg_observation(classification),
                self._build_garvis_attestation(classification),
            ],
            conclusion=classification.clinical_interpretation,
            conclusionCode=[
                CodeableConcept(coding=[
                    Coding(
                        system="http://loinc.org",
                        code=classification.acmg_loinc_code
                    )
                ])
            ]
        )

Ready for Epic SMART on FHIR and Cerner Millennium. Zero custom integration work.

Clinical Coverage — 5 Domains

Hereditary Cancer

BRCA1/2, Lynch syndrome (MLH1, MSH2, MSH6, PMS2, EPCAM), Li-Fraumeni (TP53), Cowden (PTEN), hereditary diffuse gastric cancer (CDH1), 80+ hereditary cancer genes with syndrome-specific ACMG logic.

Pharmacogenomics

300+ CPIC Level A/B drug-gene pairs. CYP2D6, CYP2C19, CYP2C9, DPYD, TPMT, SLCO1B1, G6PD, NUDT15. Full diplotype calling with population-adjusted allele frequencies.

Somatic Oncology

TMB calculation, MSI-H assessment, OncoKB therapeutic implication mapping, pan-cancer actionability scoring, FDA-approved and investigational therapy matching.

Rare Disease

Whole exome/genome trio analysis, de novo variant prioritization, pedigree reconstruction from VCF, HPO phenotype-to-gene matching, OMIM disorder linkage.

Drug Discovery

Target-disease association scoring, ADMET prediction, lead compound optimization, AlphaFold-integrated structural impact analysis for variant functional characterization.

Tech Stack

ai_agents:
  llm: Anthropic Claude
  routing: Opus (STAT) → Sonnet (standard) → Haiku (QC)
  quality: G-ARVIS engine
  correction: ARGUS-AI (max 3 iterations)
  orchestration: LangChain + LlamaIndex
  vector_db: Qdrant (7 collections, 1536-dim, tenant-scoped)
    collections:
      - ClinVar
      - gnomAD  
      - OMIM
      - PubMed
      - PharmGKB
      - OncoKB
      - ChEMBL

backend:
  framework: FastAPI
  language: Python 3.11
  orm: SQLAlchemy + Alembic
  validation: Pydantic v2
  auth: Keycloak RBAC + ABAC + JWT

data:
  primary_db: PostgreSQL 16 (RLS, 500-tenant capacity)
  cache: Redis 7
  streaming: Apache Kafka
  analytics: ClickHouse (immutable audit log)
  object_store: Delta Lake / S3 (BAM/VCF/FASTQ/CRAM)

bioinformatics:
  variant_calling: GATK + DeepVariant
  annotation: ANNOVAR + VEP + Ensembl
  structure: AlphaFold API

frontend:
  framework: React 18 + TypeScript
  styling: Tailwind CSS
  visualization: D3.js + IGV.js + Recharts

mlops:
  registry: MLflow
  drift: Evidently
  monitoring: Prometheus + Grafana
  tracing: OpenTelemetry

infrastructure:
  containers: Docker + Kubernetes + Helm
  iac: Terraform
  gitops: ArgoCD
  ci_cd: GitHub Actions + SonarQube + Trivy
  deployment: AWS / Azure / GCP / Oracle Cloud / On-prem

Integration Architecture

GenomixIQ is built integration-ready from day one:

External Systems          GenomixIQ API            Internal Agents
─────────────────         ──────────────           ───────────────
Epic SMART on FHIR ──→   /api/v1/variants  ──→   VariantClassifier
Cerner Millennium  ──→   /api/v1/reports   ──→   ClinicalReporter
Lab LIS            ──→   /api/v1/pgx       ──→   PGxAgent
Research Portal    ──→   /api/v1/trials    ──→   TrialMatcher
Pharma Pipeline    ──→   /api/v1/targets   ──→   DrugDiscovery
                         /api/v1/quality   ──→   QualityScorer
                         /api/v1/batch     ──→   All agents (parallel)

API is publicly testable today: api.genomixiq.com/docs

Try It Right Now

# Test the health check
curl https://api.genomixiq.com/health

# Submit a variant for classification
curl -X POST https://api.genomixiq.com/api/v1/variants \
  -H "Content-Type: application/json" \
  -d '{
    "variant_id": "VCV000017694",
    "gene": "BRCA1",
    "transcript": "NM_007294.4",
    "hgvs_c": "c.5266dupC",
    "genome_build": "GRCh38"
  }'

# Response includes:
# - ACMG classification with criteria applied
# - Variant Intelligence Score (VIS)
# - G-ARVIS quality attestation
# - Verified citations
# - FHIR R4 DiagnosticReport

Terraform modules included for AWS, Azure, GCP, Oracle Cloud, and on-premises Kubernetes.

What This Unlocks for Integration Partners

For EHR vendors (Epic, Oracle Health, Microsoft Nuance, Veeva):
FHIR R4 native output. SMART on FHIR compatible. Plug into your existing genomics module with zero custom integration. G-ARVIS attestation on every result for clinical defensibility.

For health system IT teams:
On-premises Kubernetes deploy. HIPAA-ready architecture with PHI tokenization before any LLM prompt. Full audit trail in ClickHouse. Row-level security across 500-tenant capacity. SOC 2 Type II controls in the pipeline.

For pharma R&D platforms:
Drug Discovery Agent with ChEMBL, DrugBank, and AlphaFold API integration. Target validation, ADMET prediction, lead optimization. REST API + batch endpoints for pipeline integration.

For genomics lab software (Sophia Genetics, Illumina DRAGEN, Fabric Genomics):
VCF ingestion endpoint. Batch classification API. ACMG classification output with full evidence trace. Direct integration point for existing lab workflows.

Benchmarks vs Current Standard of Care

Metric	Manual (current)	GenomixIQ
Time per variant	60–90 min	8 seconds
Cost per variant	$12–$18	$0.023
Citation coverage	Pathologist memory	100% verified
Hallucination rate	N/A (human)	0.00 (hard gate)
FHIR output	Manual entry	Native
Throughput	~6/hour/pathologist	450+/hour
G-ARVIS score	N/A	0.937

What Comes Next

Whole genome population-scale interpretation with federated learning
Direct ClinVar submission pipeline for novel variant classifications
Somatic liquid biopsy with ctDNA quantification
Multi-modal proteomics and epigenomics integration
CAP/CLIA audit package for regulatory inspection readiness

The Acquisition Conversation

GenomixIQ is the first platform to combine:

Production-grade 12-agent clinical genomics mesh
Technically enforced zero-hallucination safety gate
FHIR R4 native output (not a roadmap item)
G-ARVIS — the first AI quality standard for clinical genomics
Any-cloud + on-prem single-command deployment
Live public API with Swagger documentation

Strategic fits include Oracle Health (Cerner), Microsoft (Nuance), NVIDIA (Clara), Illumina, Tempus AI, Sophia Genetics, and health system genomics programs building precision medicine infrastructure.

If you are building in this space — as an engineer, a partner, or an acquirer — the conversation is open.

I Built 11 Autonomous Agents for Healthcare Revenue Cycle. Here Is the Full Architecture.

Anil Prasad — Fri, 10 Apr 2026 19:36:57 +0000

How I Built a Self-Correcting Multi-Agent System for Healthcare — and Why Standard ML Metrics Failed Me

Tags: ai, python, healthcare, machinelearning

Cover image: 04_argus_correction.png

I have been building production AI systems for 28 years. At UnitedHealth Group I ran a 20,000-node Big Data Platform. At R1 RCM I was inside the $4.1B Cloudmed acquisition. At Duke Energy I run AI and product engineering for critical infrastructure.

None of that experience prepared me for the specific engineering problem of building a reliable multi-agent system for healthcare revenue cycle management.

This post is about what I learned, what broke, and what I had to invent to make it work. The code is real. The numbers are from production.

The problem with agentic systems in regulated environments

Most agentic system tutorials show you a single agent calling a few tools and returning a result. That is fine for demos. It is not fine when the agent is making claims submission decisions on a $300M annual revenue stream for a hospital system.

The core issues I ran into, in order of how badly they burned me:

1. LLMs are not deterministic enough for sequential RCM workflows

Give the same clinical note to the same model twice and you will get subtly different ICD-10 code recommendations. In a classification task that is fine — you measure accuracy across a test set. In an agent that is making 14 sequential decisions across a claims workflow, small inconsistencies compound. A slightly different coding recommendation in step 3 changes the prior authorization requirement in step 5, which changes the denial probability score in step 8.

2. Standard metrics do not capture agentic failure modes

Precision and recall tell you nothing about whether the agent followed the right path to get to a correct answer. An agent that approves the right claim after six wrong turns is not a success — it is a future liability. I needed metrics that measured the sequential behavior of the agent across a workflow, not just the final output.

3. PHI in prompts is a HIPAA violation waiting to happen

This one is obvious in theory and surprisingly hard in practice. The moment you build a multi-agent system where context is passed between agents, you have to be extremely deliberate about what is in that context. A naive implementation will leak PHI into prompt context within the first week of real data.

4. There was no observability framework built for agents

Datadog, Arize, WhyLabs — all excellent for ML model monitoring. None of them answer the questions I needed answered: Is this agent's output grounded in the source data? Is it consistent across similar inputs? Is it recovering from failures autonomously or silently degrading?

What I built: ARIA and the frameworks around it

ARIA is a hierarchical multi-agent system: one Supervisor agent orchestrating 10 specialist agents across the full RCM workflow. I will not walk through all 11 agents here — the full architecture is in the Medium article linked at the end. What I want to focus on are the three engineering innovations that made it reliable enough for production healthcare.

Innovation 1: G-ARVIS — a 6-dimension observability framework for agents

I defined G-ARVIS to answer the specific observability questions that no existing tool addressed. Six dimensions, scored per agent execution, in real time.

from dataclasses import dataclass
from typing import Optional

@dataclass
class GARVISScore:
    groundedness: float    # Output traceable to source data (0-1)
    accuracy: float        # Factual correctness of output (0-1)
    reliability: float     # Consistency across similar inputs (0-1)
    variance: float        # Stability under edge cases (0-1)
    inference_cost: float  # Token efficiency per correct output (0-1)
    safety: float          # PHI enforcement, HIPAA compliance (0-1)

    @property
    def composite(self) -> float:
        return (
            self.groundedness * 0.20 +
            self.accuracy     * 0.20 +
            self.reliability  * 0.18 +
            self.variance     * 0.17 +
            self.inference_cost * 0.10 +
            self.safety       * 0.15
        )

    @property
    def is_production_ready(self) -> bool:
        # Safety is a hard gate — any PHI violation fails immediately
        if self.safety < 1.0:
            return False
        return self.composite >= 0.85

The weighting is intentional. Groundedness and Accuracy carry the most weight because in healthcare, a hallucinated output is not an annoyance — it is a compliance event. Safety carries 15% but is also a hard gate: any execution that touches PHI in the prompt context fails immediately regardless of the composite score.

Why Variance is the hardest dimension to score

Variance measures output stability under edge cases — ambiguous clinical notes, incomplete payer data, conflicting authorization histories. The challenge is that you can only measure it retrospectively across a population of similar inputs. We use a sliding window of the last 200 similar executions and measure the coefficient of variation on key output fields.

import numpy as np
from collections import deque

class VarianceMonitor:
    def __init__(self, window_size: int = 200):
        self.window = deque(maxlen=window_size)

    def record(self, output_vector: list[float]):
        self.window.append(output_vector)

    def score(self) -> float:
        if len(self.window) < 10:
            return 1.0  # insufficient data, assume stable
        arr = np.array(list(self.window))
        # Coefficient of variation per output dimension
        cv = np.std(arr, axis=0) / (np.mean(arr, axis=0) + 1e-8)
        # Score: 1.0 = perfectly stable, 0.0 = completely unstable
        return float(np.clip(1.0 - np.mean(cv), 0.0, 1.0))

Current production Variance score: 91.7%. This is the dimension I am least satisfied with and where most of our active engineering effort is focused. Target is 95%+.

Innovation 2: Three new agentic metrics

I defined these because ASF, ERR, and CPCS did not exist anywhere I could find, and I needed them.

Action Sequence Fidelity (ASF)

What percentage of agent execution paths match the optimal RCM workflow path? This requires defining the optimal path — which we did by analyzing 50,000 adjudicated claims and extracting the decision sequence that led to first-pass approval with minimum rework.

from difflib import SequenceMatcher

class ASFCalculator:
    def __init__(self, optimal_paths: dict[str, list[str]]):
        # optimal_paths: claim_type -> sequence of agent actions
        self.optimal_paths = optimal_paths

    def calculate(
        self,
        claim_type: str,
        actual_path: list[str]
    ) -> float:
        optimal = self.optimal_paths.get(claim_type, [])
        if not optimal:
            return 1.0  # no baseline, assume correct

        matcher = SequenceMatcher(
            None,
            optimal,
            actual_path
        )
        return matcher.ratio()

    def batch_asf(self, executions: list[dict]) -> float:
        scores = [
            self.calculate(e["claim_type"], e["path"])
            for e in executions
        ]
        return sum(scores) / len(scores) if scores else 0.0

Current production ASF: 91.4%.

Error Recovery Rate (ERR)

When an agent encounters a failure, how often does it recover autonomously? This is straightforward to measure — you track every exception event and whether it resolved within the ARGUS correction loop or escalated to human review.

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ExecutionEvent:
    execution_id: str
    agent_id: str
    timestamp: datetime
    exception_type: Optional[str]
    resolved_autonomously: Optional[bool]
    attempts: int

class ERRTracker:
    def __init__(self):
        self.events: list[ExecutionEvent] = []

    def record(self, event: ExecutionEvent):
        self.events.append(event)

    def calculate_err(
        self,
        window_hours: int = 24
    ) -> float:
        cutoff = datetime.now().timestamp() - (window_hours * 3600)
        recent = [
            e for e in self.events
            if e.timestamp.timestamp() > cutoff
            and e.exception_type is not None
        ]
        if not recent:
            return 1.0

        autonomous = sum(
            1 for e in recent
            if e.resolved_autonomously is True
        )
        return autonomous / len(recent)

Current production ERR: 87.3%.

Cost Per Correct Sequence (CPCS)

Total LLM inference cost for one complete, correct RCM workflow execution. This is your unit economics metric. If CPCS exceeds the margin on the claim being processed, the system is not profitable to operate regardless of how accurate it is.

@dataclass
class SequenceCost:
    execution_id: str
    total_input_tokens: int
    total_output_tokens: int
    model_rates: dict  # model_id -> (input_rate, output_rate) per 1M tokens
    was_correct: bool
    attempts: int

    def total_cost_usd(self) -> float:
        input_cost = sum(
            (self.total_input_tokens / 1_000_000) * rate[0]
            for rate in self.model_rates.values()
        )
        output_cost = sum(
            (self.total_output_tokens / 1_000_000) * rate[1]
            for rate in self.model_rates.values()
        )
        return input_cost + output_cost

class CPCSCalculator:
    def __init__(self):
        self.sequences: list[SequenceCost] = []

    def record(self, seq: SequenceCost):
        self.sequences.append(seq)

    def calculate_cpcs(self) -> float:
        correct = [s for s in self.sequences if s.was_correct]
        if not correct:
            return float('inf')
        total_cost = sum(s.total_cost_usd() for s in correct)
        return total_cost / len(correct)

Current production CPCS: $0.023 per claim end-to-end.

Innovation 3: ARGUS — autonomous self-correction

ARGUS is the layer that makes the system reliable enough for production. The core insight: instead of trying to make an LLM deterministically correct on the first attempt, you build a reflection loop that detects failure, analyzes the failure mode by G-ARVIS dimension, and generates a corrected prompt.

import asyncio
from typing import Any, Callable, Awaitable

@dataclass
class CorrectionResult:
    output: Any
    score: GARVISScore
    attempts: int
    corrected: bool
    escalated: bool

class ARGUSGuard:
    def __init__(
        self,
        max_attempts: int = 3,
        target_composite: float = 0.85,
        safety_threshold: float = 1.0,  # hard gate
        domain: str = "healthcare_rcm",
        phi_safe: bool = True
    ):
        self.max_attempts = max_attempts
        self.target_composite = target_composite
        self.safety_threshold = safety_threshold
        self.domain = domain
        self.phi_safe = phi_safe

    async def execute_with_correction(
        self,
        agent_fn: Callable[..., Awaitable[Any]],
        task: dict,
        scorer: "GARVISScorer"
    ) -> CorrectionResult:

        attempt = 0
        current_task = task.copy()

        while attempt < self.max_attempts:
            output = await agent_fn(current_task)
            score = await scorer.score(output, self.domain)

            # PHI hard gate — fail immediately, do not retry
            if score.safety < self.safety_threshold:
                return CorrectionResult(
                    output=None,
                    score=score,
                    attempts=attempt + 1,
                    corrected=False,
                    escalated=True
                )

            if score.composite >= self.target_composite:
                return CorrectionResult(
                    output=output,
                    score=score,
                    attempts=attempt + 1,
                    corrected=attempt > 0,
                    escalated=False
                )

            # Score below threshold — reflect and refine
            current_task = self._reflect_and_refine(
                original_task=task,
                failed_output=output,
                score=score,
                attempt=attempt
            )
            attempt += 1

        # All attempts exhausted — escalate to human review
        return CorrectionResult(
            output=output,
            score=score,
            attempts=attempt,
            corrected=False,
            escalated=True
        )

    def _reflect_and_refine(
        self,
        original_task: dict,
        failed_output: Any,
        score: GARVISScore,
        attempt: int
    ) -> dict:
        # Identify the weakest dimension and generate
        # a dimension-specific correction signal
        weak_dims = self._weakest_dimensions(score)
        correction_prompt = self._build_correction_prompt(
            original_task,
            failed_output,
            weak_dims,
            attempt
        )
        refined = original_task.copy()
        refined["correction_context"] = correction_prompt
        refined["attempt"] = attempt + 1
        return refined

    def _weakest_dimensions(
        self,
        score: GARVISScore
    ) -> list[str]:
        dims = {
            "groundedness": score.groundedness,
            "accuracy": score.accuracy,
            "reliability": score.reliability,
            "variance": score.variance,
            "inference_cost": score.inference_cost,
        }
        # Return dimensions below 0.85, sorted weakest first
        return sorted(
            [k for k, v in dims.items() if v < 0.85],
            key=lambda k: dims[k]
        )

The _build_correction_prompt method is proprietary — that is where the domain-specific healthcare knowledge lives. But the structure above is fully open in the ARGUS SDK.

The PHI tokenization architecture

This is the part that took the longest to get right. The requirement: agents need full clinical context to make good RCM decisions, but no PHI can appear in any LLM prompt.

import hashlib
import hmac
import re
from typing import Any

class PHITokenizer:
    # Patterns for common PHI types
    PHI_PATTERNS = {
        "MRN":   r"\bMRN[-:\s]?\d{6,10}\b",
        "DOB":   r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b",
        "SSN":   r"\b\d{3}-\d{2}-\d{4}\b",
        "NAME":  r"\b[A-Z][a-z]+\s[A-Z][a-z]+\b",
        "NPI":   r"\bNPI[-:\s]?\d{10}\b",
    }

    def __init__(self, secret_key: bytes):
        self.secret_key = secret_key
        self._token_map: dict[str, str] = {}
        self._reverse_map: dict[str, str] = {}

    def _generate_token(self, phi_value: str, phi_type: str) -> str:
        # Deterministic: same PHI always maps to same token
        raw = f"{phi_type}:{phi_value}"
        token_bytes = hmac.new(
            self.secret_key,
            raw.encode(),
            hashlib.sha256
        ).hexdigest()[:16]
        return f"[{phi_type}_TOKEN_{token_bytes.upper()}]"

    def tokenize(self, text: str) -> str:
        tokenized = text
        for phi_type, pattern in self.PHI_PATTERNS.items():
            matches = re.findall(pattern, tokenized)
            for match in matches:
                token = self._generate_token(match, phi_type)
                self._token_map[match] = token
                self._reverse_map[token] = match
                tokenized = tokenized.replace(match, token)
        return tokenized

    def rehydrate(self, tokenized_text: str) -> str:
        result = tokenized_text
        for token, phi_value in self._reverse_map.items():
            result = result.replace(token, phi_value)
        return result

    def is_phi_clean(self, text: str) -> bool:
        for pattern in self.PHI_PATTERNS.values():
            if re.search(pattern, text):
                return False
        return True

Every prompt that goes to an LLM passes through tokenize() first. Every output that gets committed to the RCM state machine passes through rehydrate() inside the secure perimeter. The is_phi_clean() check is what the G-ARVIS Safety dimension calls before every inference.

Production Safety score: 100%. Zero PHI exposure events.

Install and get started

The ARGUS SDK — G-ARVIS scoring, ASF/ERR/CPCS calculators, PHITokenizer base class, and ARGUSGuard correction loop — is open-core and on PyPI.

pip install argus-ai

from argus_ai import ARGUSGuard, GARVISScorer, PHITokenizer
from argus_ai.metrics import ASFCalculator, ERRTracker, CPCSCalculator

# Wrap any async agent function with self-correction
guard = ARGUSGuard(
    max_attempts=3,
    target_composite=0.85,
    domain="healthcare_rcm",
    phi_safe=True
)

result = await guard.execute_with_correction(
    agent_fn=my_denial_predictor,
    task=claim_task,
    scorer=GARVISScorer()
)

print(f"Score: {result.score.composite:.1%}")
print(f"Attempts: {result.attempts}")
print(f"Escalated: {result.escalated}")

Production results

These are from the live ARIA system, 24-hour rolling average:

Metric	Value
G-ARVIS composite	93.9%
Groundedness	96.2%
Accuracy	94.8%
Reliability	93.1%
Variance	91.7%
Inference Cost	95.3%
Safety	100%
Action Sequence Fidelity	91.4%
Error Recovery Rate	87.3%
Cost Per Correct Sequence	$0.023
Denial rate reduction	38%

What is open vs proprietary

Open (argus-ai on PyPI + GitHub):

ARGUSGuard correction loop
GARVISScorer base framework
PHITokenizer base class
ASF, ERR, CPCS calculators
PulseFlow MLOps pipeline

Proprietary (the ARIA product):

11-agent supervisor hierarchy with RCM domain specialization
Payer policy RAG with live contract updates
Predictive denial scoring model
RCM domain knowledge engine
Multi-tenant deployment infrastructure

Links

GitHub: github.com/anilatambharii/argus-ai
PyPI: pypi.org/project/argus-ai
Platform: ambharii.com/RCM
Full architecture article: medium.com/p/9d0c9f8d662a
Questions or contributions: anil@ambharii.com

If you are building agentic systems in regulated industries and running into the same observability and reliability problems — I would genuinely like to hear from you. The metrics definitions are public. Use them, improve them, tell me what is wrong with them.

Anil Prasad — Founder, Ambharii Labs · Head of Engineering & Product, Duke Energy · Top 100 AI Leaders USA 2024

#HumanWritten #ExpertiseFromField

Building Multi-Agent Systems That Don't Collapse in Production

Anil Prasad — Wed, 08 Apr 2026 18:10:33 +0000

Building Multi-Agent Systems That Don't Collapse in Production
Multi-agent AI deployments grew 327% in four months across 20,000 organizations (Databricks, 2025). Most of those deployments will fail in production. Not because the models are bad. Because the composition is broken.

This post covers three failure modes I've seen repeatedly in regulated production environments, and the engineering patterns that fix them — with real code using ARGUS, the open-source agentic observability framework I built and maintain.

The math that kills multi-agent systems first Before architecture, do this calculation:

pythonimport math

def end_to_end_reliability(agent_reliability: float, num_agents: int) -> float:
return math.pow(agent_reliability, num_agents)

What most teams are actually deploying

print(end_to_end_reliability(0.85, 5)) # → 0.4437
print(end_to_end_reliability(0.90, 5)) # → 0.5905
print(end_to_end_reliability(0.97, 5)) # → 0.8587

The target you need before orchestrating

print(end_to_end_reliability(0.99, 5)) # → 0.9510

The rule: get each single agent to 97%+ before you chain them. Below that, you are engineering a system that fails more than it succeeds.

Failure mode 1: Cascade failures
(See the cascade failure trace diagram above)
Agent A produces a marginally wrong output. Agent B treats it as correct input. Agent C produces a confidently wrong conclusion. No single agent failed — the composition did.
In standard per-agent logging, this is invisible. The per-agent logs all show status: success. Only the final output reveals the failure — after it has already been acted upon.
The fix: inter-agent validation with sampled contracts
pythonfrom argus_ai import AgentTracer, ValidationContract

tracer = AgentTracer(workflow_id="rcm-prior-auth-v2")

class ValidatedAgent:
def init(self, agent_fn, contract: ValidationContract, sample_rate=0.15):
self.agent = agent_fn
self.contract = contract
self.sample_rate = sample_rate

def run(self, input_payload: dict, hop_id: str) -> dict:
    output = self.agent(input_payload)

    # Sample 15% of hops for deep validation
    # 100% validation on high-stakes decision points
    should_validate = (
        random.random() < self.sample_rate
        or input_payload.get("high_stakes", False)
    )

    if should_validate:
        violations = self.contract.check(output)
        tracer.record_hop(
            hop_id=hop_id,
            input=input_payload,
            output=output,
            violations=violations,
            validated=True
        )
        if violations:
            raise ContractViolation(f"hop {hop_id}: {violations}")
    else:
        tracer.record_hop(hop_id=hop_id, input=input_payload,
                          output=output, validated=False)

    return output

Key design decisions here:

15% sample rate on standard hops — cheap enough to run always, catches systematic errors fast
100% validation on high-stakes hops (financial commits, clinical decisions, compliance writes)
Every hop is recorded regardless of whether it was validated — the audit trail is unconditional

Failure mode 2: Context drift
Each agent has a finite context window. As tasks pass between agents, the original intent degrades. By agent 5, the goal may have been silently reinterpreted twice.
This is especially dangerous in regulated domains. If the original intent is a compliance requirement, a drift of even 5% of the specification can create a violation.
The fix: shared state with strict write contracts
pythonfrom argus_ai import SharedStateStore, StateContract
from pydantic import BaseModel
from typing import Optional
import hashlib

class WorkflowIntent(BaseModel):
"""The original goal. Immutable after creation."""
goal_id: str
original_prompt: str
compliance_constraints: list[str]
created_at: str
checksum: str # sha256 of original_prompt + constraints

class AgentWriteContract(BaseModel):
"""What each agent is allowed to write."""
agent_id: str
allowed_write_keys: list[str]
forbidden_write_keys: list[str] = ["original_intent", "goal_id"]

store = SharedStateStore(backend="redis")

def write_with_contract(
agent_id: str,
key: str,
value: any,
contract: AgentWriteContract
) -> None:
if key in contract.forbidden_write_keys:
raise PermissionError(
f"Agent {agent_id} attempted to overwrite protected key: {key}"
)
if key not in contract.allowed_write_keys:
raise PermissionError(
f"Agent {agent_id} attempted to write undeclared key: {key}"
)
store.set(key, value, written_by=agent_id)
The original_intent is write-once. No agent can overwrite the goal. Each agent reads from the store at the start of its hop — it always has access to the original specification, not just what the previous agent passed.

Failure mode 3: Accountability gaps
When the multi-agent workflow fails, which agent do you debug?
Without an end-to-end trace, this question is unanswerable. You have logs from five agents, all showing local success, and a broken final output. That is a crime scene with no chain of custody.
The fix: end-to-end workflow tracing with G-ARVIS scoring
pythonfrom argus_ai import WorkflowTracer, GARVISScorer

Initialize once per workflow run

tracer = WorkflowTracer(
workflow_id="prior-auth-batch-20260408",
g_arvis_dimensions=["groundedness", "accuracy", "reliability",
"variance", "inference_cost", "safety"]
)

Each agent wraps its execution

with tracer.hop("parser", metadata={"model": "claude-sonnet-4-6"}) as hop:
result = parser_agent.run(document)
hop.record(
input_tokens=result.input_tokens,
output_tokens=result.output_tokens,
confidence=result.confidence,
output_hash=hashlib.sha256(
str(result.output).encode()
).hexdigest()
)

After workflow completes — full trace available

report = tracer.finalize()

print(report.end_to_end_success_rate) # 0.943
print(report.weakest_hop) # "validator" — 84.2% pass rate
print(report.g_arvis_scores) # per-dimension scores
print(report.cascade_risk_score) # probability of undetected cascade
The cascade_risk_score is the key metric. It measures the probability that a marginal error in an early hop could propagate undetected to a confident wrong output. If this exceeds 0.15, you have a systemic observability problem regardless of individual agent quality.

Putting it together: the minimal production-ready multi-agent loop
pythonfrom argus_ai import (
AgentTracer, SharedStateStore,
WorkflowTracer, ValidationContract
)

class SupervisorAgent:
def init(self, specialists: dict, tracer: WorkflowTracer):
self.specialists = specialists
self.tracer = tracer
self.store = SharedStateStore()

def run(self, goal: str, constraints: list[str]) -> dict:
    # Write intent once — immutable
    intent = self.store.write_intent(goal, constraints)

    # Decompose
    subtasks = self.decompose(goal)

    results = {}
    for task_id, task in subtasks.items():
        agent = self.specialists[task.agent_type]
        contract = ValidationContract.for_task(task_id)

        with self.tracer.hop(task_id) as hop:
            # Agent always reads original intent from store
            context = {
                "task": task,
                "original_intent": self.store.get_intent(intent.goal_id),
                "prior_results": results  # only pass, never overwrite
            }
            output = agent.run_with_validation(context, contract)
            results[task_id] = output
            hop.record(output)

    return self.synthesize(results, intent)

Three things this loop enforces that most implementations skip:

Every agent reads the original intent — not just what the previous agent passed
Every hop is traced unconditionally — validation is sampled, tracing is not
The supervisor synthesizes from all hop results — not just the last agent's output

Install and try it
bashpip install argus-ai
python# Minimal smoke test
from argus_ai import AgentTracer

tracer = AgentTracer(workflow_id="test-001")

with tracer.hop("my-first-agent") as hop:
output = {"result": "hello", "confidence": 0.94}
hop.record(output=output, confidence=0.94)

print(tracer.finalize().summary())
Full docs and examples at github.com/anilatambharii/argus-ai.
The G-ARVIS scoring engine and SDK are fully open-source. The autonomous correction agents (self-healing workflows) are in the Pro tier.

Check your agentic readiness before you deploy
The AI Aether Platform runs a G-ARVIS-based readiness assessment across 8 dimensions — observability maturity, governance posture, agentic infrastructure, and more. Takes 10 minutes. Gives you a baseline before you commit architecture decisions that cost months to reverse.
CDAIO Circle members: use code CDAIO2026 for Pro access.

I write about production AI engineering from regulated-industry deployments (healthcare, energy, financial services). Follow for more patterns from the field.

AgenticAI #MLOps #Python #ProductionAI #HumanWritten #ExpertiseFromField

How I Built a Self-Correcting Multi-Agent System for Healthcare - and Why Standard ML Metrics Failed Me

Anil Prasad — Tue, 07 Apr 2026 21:48:54 +0000

How I Built a Self-Correcting Multi-Agent System for Healthcare — and Why Standard ML Metrics Failed Me

Tags: ai, python, healthcare, machinelearning

Cover image: 04_argus_correction.png

None of that experience prepared me for the specific engineering problem of building a reliable multi-agent system for healthcare revenue cycle management.

This post is about what I learned, what broke, and what I had to invent to make it work. The code is real. The numbers are from production.

The problem with agentic systems in regulated environments

The core issues I ran into, in order of how badly they burned me:

1. LLMs are not deterministic enough for sequential RCM workflows

2. Standard metrics do not capture agentic failure modes

3. PHI in prompts is a HIPAA violation waiting to happen

4. There was no observability framework built for agents

What I built: ARIA and the frameworks around it

Innovation 1: G-ARVIS — a 6-dimension observability framework for agents

I defined G-ARVIS to answer the specific observability questions that no existing tool addressed. Six dimensions, scored per agent execution, in real time.

from dataclasses import dataclass
from typing import Optional

@dataclass
class GARVISScore:
    groundedness: float    # Output traceable to source data (0-1)
    accuracy: float        # Factual correctness of output (0-1)
    reliability: float     # Consistency across similar inputs (0-1)
    variance: float        # Stability under edge cases (0-1)
    inference_cost: float  # Token efficiency per correct output (0-1)
    safety: float          # PHI enforcement, HIPAA compliance (0-1)

    @property
    def composite(self) -> float:
        return (
            self.groundedness * 0.20 +
            self.accuracy     * 0.20 +
            self.reliability  * 0.18 +
            self.variance     * 0.17 +
            self.inference_cost * 0.10 +
            self.safety       * 0.15
        )

    @property
    def is_production_ready(self) -> bool:
        # Safety is a hard gate — any PHI violation fails immediately
        if self.safety < 1.0:
            return False
        return self.composite >= 0.85

Why Variance is the hardest dimension to score

import numpy as np
from collections import deque

class VarianceMonitor:
    def __init__(self, window_size: int = 200):
        self.window = deque(maxlen=window_size)

    def record(self, output_vector: list[float]):
        self.window.append(output_vector)

    def score(self) -> float:
        if len(self.window) < 10:
            return 1.0  # insufficient data, assume stable
        arr = np.array(list(self.window))
        # Coefficient of variation per output dimension
        cv = np.std(arr, axis=0) / (np.mean(arr, axis=0) + 1e-8)
        # Score: 1.0 = perfectly stable, 0.0 = completely unstable
        return float(np.clip(1.0 - np.mean(cv), 0.0, 1.0))

Current production Variance score: 91.7%. This is the dimension I am least satisfied with and where most of our active engineering effort is focused. Target is 95%+.

Innovation 2: Three new agentic metrics

I defined these because ASF, ERR, and CPCS did not exist anywhere I could find, and I needed them.

Action Sequence Fidelity (ASF)

from difflib import SequenceMatcher

class ASFCalculator:
    def __init__(self, optimal_paths: dict[str, list[str]]):
        # optimal_paths: claim_type -> sequence of agent actions
        self.optimal_paths = optimal_paths

    def calculate(
        self,
        claim_type: str,
        actual_path: list[str]
    ) -> float:
        optimal = self.optimal_paths.get(claim_type, [])
        if not optimal:
            return 1.0  # no baseline, assume correct

        matcher = SequenceMatcher(
            None,
            optimal,
            actual_path
        )
        return matcher.ratio()

    def batch_asf(self, executions: list[dict]) -> float:
        scores = [
            self.calculate(e["claim_type"], e["path"])
            for e in executions
        ]
        return sum(scores) / len(scores) if scores else 0.0

Current production ASF: 91.4%.

Error Recovery Rate (ERR)

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ExecutionEvent:
    execution_id: str
    agent_id: str
    timestamp: datetime
    exception_type: Optional[str]
    resolved_autonomously: Optional[bool]
    attempts: int

class ERRTracker:
    def __init__(self):
        self.events: list[ExecutionEvent] = []

    def record(self, event: ExecutionEvent):
        self.events.append(event)

    def calculate_err(
        self,
        window_hours: int = 24
    ) -> float:
        cutoff = datetime.now().timestamp() - (window_hours * 3600)
        recent = [
            e for e in self.events
            if e.timestamp.timestamp() > cutoff
            and e.exception_type is not None
        ]
        if not recent:
            return 1.0

        autonomous = sum(
            1 for e in recent
            if e.resolved_autonomously is True
        )
        return autonomous / len(recent)

Current production ERR: 87.3%.

Cost Per Correct Sequence (CPCS)

@dataclass
class SequenceCost:
    execution_id: str
    total_input_tokens: int
    total_output_tokens: int
    model_rates: dict  # model_id -> (input_rate, output_rate) per 1M tokens
    was_correct: bool
    attempts: int

    def total_cost_usd(self) -> float:
        input_cost = sum(
            (self.total_input_tokens / 1_000_000) * rate[0]
            for rate in self.model_rates.values()
        )
        output_cost = sum(
            (self.total_output_tokens / 1_000_000) * rate[1]
            for rate in self.model_rates.values()
        )
        return input_cost + output_cost

class CPCSCalculator:
    def __init__(self):
        self.sequences: list[SequenceCost] = []

    def record(self, seq: SequenceCost):
        self.sequences.append(seq)

    def calculate_cpcs(self) -> float:
        correct = [s for s in self.sequences if s.was_correct]
        if not correct:
            return float('inf')
        total_cost = sum(s.total_cost_usd() for s in correct)
        return total_cost / len(correct)

Current production CPCS: $0.023 per claim end-to-end.

Innovation 3: ARGUS — autonomous self-correction

import asyncio
from typing import Any, Callable, Awaitable

@dataclass
class CorrectionResult:
    output: Any
    score: GARVISScore
    attempts: int
    corrected: bool
    escalated: bool

class ARGUSGuard:
    def __init__(
        self,
        max_attempts: int = 3,
        target_composite: float = 0.85,
        safety_threshold: float = 1.0,  # hard gate
        domain: str = "healthcare_rcm",
        phi_safe: bool = True
    ):
        self.max_attempts = max_attempts
        self.target_composite = target_composite
        self.safety_threshold = safety_threshold
        self.domain = domain
        self.phi_safe = phi_safe

    async def execute_with_correction(
        self,
        agent_fn: Callable[..., Awaitable[Any]],
        task: dict,
        scorer: "GARVISScorer"
    ) -> CorrectionResult:

        attempt = 0
        current_task = task.copy()

        while attempt < self.max_attempts:
            output = await agent_fn(current_task)
            score = await scorer.score(output, self.domain)

            # PHI hard gate — fail immediately, do not retry
            if score.safety < self.safety_threshold:
                return CorrectionResult(
                    output=None,
                    score=score,
                    attempts=attempt + 1,
                    corrected=False,
                    escalated=True
                )

            if score.composite >= self.target_composite:
                return CorrectionResult(
                    output=output,
                    score=score,
                    attempts=attempt + 1,
                    corrected=attempt > 0,
                    escalated=False
                )

            # Score below threshold — reflect and refine
            current_task = self._reflect_and_refine(
                original_task=task,
                failed_output=output,
                score=score,
                attempt=attempt
            )
            attempt += 1

        # All attempts exhausted — escalate to human review
        return CorrectionResult(
            output=output,
            score=score,
            attempts=attempt,
            corrected=False,
            escalated=True
        )

    def _reflect_and_refine(
        self,
        original_task: dict,
        failed_output: Any,
        score: GARVISScore,
        attempt: int
    ) -> dict:
        # Identify the weakest dimension and generate
        # a dimension-specific correction signal
        weak_dims = self._weakest_dimensions(score)
        correction_prompt = self._build_correction_prompt(
            original_task,
            failed_output,
            weak_dims,
            attempt
        )
        refined = original_task.copy()
        refined["correction_context"] = correction_prompt
        refined["attempt"] = attempt + 1
        return refined

    def _weakest_dimensions(
        self,
        score: GARVISScore
    ) -> list[str]:
        dims = {
            "groundedness": score.groundedness,
            "accuracy": score.accuracy,
            "reliability": score.reliability,
            "variance": score.variance,
            "inference_cost": score.inference_cost,
        }
        # Return dimensions below 0.85, sorted weakest first
        return sorted(
            [k for k, v in dims.items() if v < 0.85],
            key=lambda k: dims[k]
        )

The _build_correction_prompt method is proprietary — that is where the domain-specific healthcare knowledge lives. But the structure above is fully open in the ARGUS SDK.

The PHI tokenization architecture

This is the part that took the longest to get right. The requirement: agents need full clinical context to make good RCM decisions, but no PHI can appear in any LLM prompt.

import hashlib
import hmac
import re
from typing import Any

class PHITokenizer:
    # Patterns for common PHI types
    PHI_PATTERNS = {
        "MRN":   r"\bMRN[-:\s]?\d{6,10}\b",
        "DOB":   r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b",
        "SSN":   r"\b\d{3}-\d{2}-\d{4}\b",
        "NAME":  r"\b[A-Z][a-z]+\s[A-Z][a-z]+\b",
        "NPI":   r"\bNPI[-:\s]?\d{10}\b",
    }

    def __init__(self, secret_key: bytes):
        self.secret_key = secret_key
        self._token_map: dict[str, str] = {}
        self._reverse_map: dict[str, str] = {}

    def _generate_token(self, phi_value: str, phi_type: str) -> str:
        # Deterministic: same PHI always maps to same token
        raw = f"{phi_type}:{phi_value}"
        token_bytes = hmac.new(
            self.secret_key,
            raw.encode(),
            hashlib.sha256
        ).hexdigest()[:16]
        return f"[{phi_type}_TOKEN_{token_bytes.upper()}]"

    def tokenize(self, text: str) -> str:
        tokenized = text
        for phi_type, pattern in self.PHI_PATTERNS.items():
            matches = re.findall(pattern, tokenized)
            for match in matches:
                token = self._generate_token(match, phi_type)
                self._token_map[match] = token
                self._reverse_map[token] = match
                tokenized = tokenized.replace(match, token)
        return tokenized

    def rehydrate(self, tokenized_text: str) -> str:
        result = tokenized_text
        for token, phi_value in self._reverse_map.items():
            result = result.replace(token, phi_value)
        return result

    def is_phi_clean(self, text: str) -> bool:
        for pattern in self.PHI_PATTERNS.values():
            if re.search(pattern, text):
                return False
        return True

Production Safety score: 100%. Zero PHI exposure events.

Install and get started

The ARGUS SDK — G-ARVIS scoring, ASF/ERR/CPCS calculators, PHITokenizer base class, and ARGUSGuard correction loop — is open-core and on PyPI.

pip install argus-ai

from argus_ai import ARGUSGuard, GARVISScorer, PHITokenizer
from argus_ai.metrics import ASFCalculator, ERRTracker, CPCSCalculator

# Wrap any async agent function with self-correction
guard = ARGUSGuard(
    max_attempts=3,
    target_composite=0.85,
    domain="healthcare_rcm",
    phi_safe=True
)

result = await guard.execute_with_correction(
    agent_fn=my_denial_predictor,
    task=claim_task,
    scorer=GARVISScorer()
)

print(f"Score: {result.score.composite:.1%}")
print(f"Attempts: {result.attempts}")
print(f"Escalated: {result.escalated}")

Production results

These are from the live ARIA system, 24-hour rolling average:

Metric	Value
G-ARVIS composite	93.9%
Groundedness	96.2%
Accuracy	94.8%
Reliability	93.1%
Variance	91.7%
Inference Cost	95.3%
Safety	100%
Action Sequence Fidelity	91.4%
Error Recovery Rate	87.3%
Cost Per Correct Sequence	$0.023
Denial rate reduction	38%

What is open vs proprietary

Open (argus-ai on PyPI + GitHub):

ARGUSGuard correction loop
GARVISScorer base framework
PHITokenizer base class
ASF, ERR, CPCS calculators
PulseFlow MLOps pipeline

Proprietary (the ARIA product):

11-agent supervisor hierarchy with RCM domain specialization
Payer policy RAG with live contract updates
Predictive denial scoring model
RCM domain knowledge engine
Multi-tenant deployment infrastructure

78% of PyTorch Models Never Reach Production. I Built the Fix.

Anil Prasad — Sat, 04 Apr 2026 18:26:23 +0000

78% of PyTorch Models Never Reach Production. I Built the Fix.

After 28 years shipping AI at scale, I got tired of watching good models die on the way to production.

By Anil S. Prasad — Founder, Ambharii Labs | Head of Engineering & Product, Duke Energy | Top 100 Most Influential AI Leaders USA 2024

There is a number that has followed me across every organization I have worked in. UnitedHealth Group, Medtronic, Ambry Genetics, R1 RCM, Duke Energy. The number is 78.

Seventy-eight percent of PyTorch models built in research never make it to production.

This is not a data science problem. The data scientists are talented. The models are good. The math works.

The problem is everything around the model. The audit trail that regulators demand. The compliance framework that legal requires. The drift detection that ops needs at 3am. The fairness analysis that the board is now asking about. The explanation that a clinician, an underwriter, or a grid operator needs before they trust the output.

None of that is in PyTorch. And nobody was building it.

So I did.

Introducing TorchForge

TorchForge is an open source enterprise governance wrapper for PyTorch. You take any model you have already built and wrap it in four lines of code. What comes back is the same model, same weights, same architecture, with a full production governance layer running underneath it.

Two-point-five percent overhead. That is all it costs.

from torchforge import ForgeModel, ForgeConfig

config = ForgeConfig(
    model_name="credit_risk_v2",
    version="1.0.0",
    enable_governance=True,
    compliance_framework="NIST_RMF_1.0"
)

model = ForgeModel(your_pytorch_model, config)
output = model(x)

# Audit trail: live.
# Drift detection: live.
# Compliance reporting: live.
# That's it.

No refactoring. No retraining. No new infrastructure team required.

What You Get Out of the Box

I want to be precise here because this is where most governance tools either overpromise or underdeliver.

NIST AI RMF 1.0 compliance tracking. Every inference is logged against the seven functions of the NIST Risk Management Framework. Govern, Map, Measure, Manage. The report generates automatically. When a regulator asks for your AI risk documentation, you export it in one command.

Real-time drift detection with automatic alerts. TorchForge monitors input distribution and output distribution on every inference pass. When drift exceeds configurable thresholds, it fires alerts to Slack, PagerDuty, or any webhook you point it at. No separate monitoring pipeline. No Evidently setup. No manual dashboards.

Bias and fairness analysis on every prediction. Demographic parity, equalized odds, individual fairness metrics run as part of the inference pass. Not as a post-hoc audit you remember to do quarterly. On every prediction. Because bias does not wait for your audit schedule.

Full audit trail from training to deployment. Every model version, every config change, every inference batch is logged with timestamp, input hash, output, confidence scores, and the governance metadata. Immutable. Queryable. Exportable.

One-click deployment to five clouds. AWS, Azure, GCP, Kubernetes, Oracle Cloud. The deployment module generates the Terraform, the Helm chart, and the GitHub Actions pipeline. Your ops team gets a clean artifact, not a Jupyter notebook printed to PDF.

A/B testing with gradual rollout. Define your champion and challenger models. Set a traffic split. TorchForge handles the routing, collects the performance metrics, and helps you decide when to promote. No feature flags library required.

Why I Built This Now

Three things converged in 2025 that made this the right moment.

First, the regulatory pressure on AI is no longer hypothetical. The EU AI Act is in force. US federal agencies have issued AI governance guidance. State-level bills are passing. Every organization I talk to is scrambling to answer the same question: how do we prove our AI is trustworthy? TorchForge makes that question answerable.

Second, PyTorch won. It is the dominant research framework and increasingly the production framework. ONNX and TorchScript made serving easier. But nobody solved governance at the framework layer. Everyone solved it at the infrastructure layer, which means it is always bolted on, never built in.

Third, I kept meeting talented ML engineers who had the same story. They built something that worked. Leadership approved it. Then it went to compliance, and it sat there for six months because nobody could answer the audit questions. TorchForge is the answer you hand compliance the day the model is ready.

The Performance Story

I know what you are thinking. Governance overhead sounds expensive.

Here is what we measured in production-equivalent workloads:

Operation	TorchForge	Pure PyTorch	Overhead
Forward pass	12.3ms	12.0ms	2.5%
Training step	45.2ms	44.8ms	0.9%
Inference batch	8.7ms	8.5ms	2.3%

Two-point-five percent on a forward pass. On a GPU cluster running ten thousand inferences per second, that is 250 extra milliseconds of total compute per second across the cluster. In exchange for full NIST compliance, continuous drift detection, bias monitoring, and a complete audit trail.

That is not a trade-off. That is a deal.

Who This Is For

If you are a solo researcher building hobby projects, TorchForge is overkill. Use it anyway if you want to learn the patterns, but it is not built for you.

TorchForge is built for three audiences.

ML engineers at regulated companies. Healthcare, financial services, energy, insurance. If your model touches a human life, a financial decision, or critical infrastructure, you need this. The compliance cost of not having it is orders of magnitude higher than the 2.5% overhead.

ML platform teams at growth-stage companies. You are scaling from one model to fifty. You need standardization. TorchForge is the standard. Every team wraps their model the same way, and you get a unified governance view across the entire portfolio.

AI consultants and system integrators. When you deliver a PyTorch model to a client and it comes with TorchForge, you are delivering a production-ready artifact, not a prototype. That changes the conversation about what you charge and what the client owns.

The Open Core Model

TorchForge is MIT-licensed. The core is free and always will be. I believe governance tooling should be open because the alternative is that only large companies with large procurement budgets can ship trustworthy AI. That is a bad outcome for the field.

The enterprise platform, the autonomous correction agents, the multi-tenant dashboard, the SLA-backed support, the private deployment options, those are available through Ambharii Labs. If you need them, you know where to find me.

But the open core does everything I described above. The compliance tracking, drift detection, bias analysis, audit trail, deployment tooling, A/B testing framework. All of it. No license key required.

Try It in Three Minutes

pip install torchforge

import torch
import torch.nn as nn
from torchforge import ForgeModel, ForgeConfig

# Your existing model, unchanged
class YourModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, 2)
        )
    def forward(self, x):
        return self.net(x)

# Wrap it
config = ForgeConfig(
    model_name="my_first_governed_model",
    version="1.0.0",
    enable_governance=True,
    compliance_framework="NIST_RMF_1.0"
)

model = ForgeModel(YourModel(), config)

# Run inference — governance is automatic
x = torch.randn(4, 128)
output = model(x)

# Export your compliance report
model.export_compliance_report("./compliance_report.json")

That is three minutes from install to your first NIST-compliant inference.

The live demo runs on Hugging Face Spaces at no cost to you. Go to huggingface.co/spaces/AmbhariiLabs/torchforge-demo and run a governed inference in your browser before you install anything.

What Comes Next

The roadmap for TorchForge is driven by what I see breaking in production, not by what sounds impressive in a conference talk.

Q2 2026: Federated learning support with differential privacy guarantees. For healthcare and financial services teams who cannot centralize training data.

Q3 2026: LLM governance extension. The same wrapper pattern applied to fine-tuned language models. Hallucination rate tracking, toxicity monitoring, prompt injection detection.

Q4 2026: Cross-framework support. The governance layer decoupled from PyTorch so it can wrap TensorFlow, JAX, and ONNX models with the same four-line interface.

Everything will stay open core. That is not a marketing promise. It is the design constraint I set before writing the first line of code.

Links

GitHub: github.com/anilatambharii/torchforge

PyPI: pip install torchforge

Live demo: huggingface.co/spaces/AmbhariiLabs/torchforge-demo

Enterprise: ambharii.com

Connect: linkedin.com/in/anilsprasad

Built by Anil S. Prasad — Founder, Ambharii Labs. 28 years of production AI across UnitedHealth Group, Medtronic, Duke Energy, Ambry Genetics, and R1 RCM. Co-Founder of the CDAIO Circle Tri-State Chapter. Stanford and BITS Pilani.

#HumanWritten #ExpertiseFromField #PyTorch #MLOps #EnterpriseAI #AIGovernance #OpenSource #NIST #ProductionAI

Cross-posting note: This article is published on Medium (@anilAmbharii). Canonical version lives at medium.com. If you found this on DEV.to, Substack, or LinkedIn, follow me there for more field notes from 28 years of production AI.

I open sourced a production MLOps pipeline. Here is what it took to get it to PyPI and Hugging Face in one day.

Anil Prasad — Thu, 02 Apr 2026 00:27:44 +0000

I have been running ML pipelines in production for few years. Tens of millions of predictions a day, real money on the line, no tolerance for guesswork.

PulseFlow started as something I built for myself. A reference architecture I kept recreating from scratch at every company because nothing open source matched what production actually demands.

Today I packaged it, published it to PyPI, and put a live demo on Hugging Face. Here is what it covers and how to run it in under ten minutes.

What PulseFlow is

A production-grade MLOps pipeline you can clone and run immediately. Not a tutorial. Not a toy dataset. A real stack.

pip install pulseflow-mlops

Five components wired together:

ETL pipeline: ingestion and preprocessing with Pandas and SQLAlchemy
Training pipeline: model training with MLflow experiment tracking
Deployment service: FastAPI microservice for real-time inference
Orchestration: Apache Airflow DAGs for end-to-end automation
Full Docker Compose stack: one command to run everything

The architecture

Every enterprise ML system I have built follows the same pattern. Raw data in, predictions out, everything in between observable and reproducible.

Raw Data → ETL → Feature Store → Training → MLflow Registry → FastAPI → Clients
                                                    ↑
                                              Airflow Scheduler

PulseFlow makes this concrete with actual code, not diagrams.

Run it locally in four commands

git clone https://github.com/anilatambharii/PulseFlow.git
cd PulseFlow
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

Then run each stage:

python etl/data_ingestion.py
python etl/data_preprocessing.py
python training/train_model.py
uvicorn deployment.app.main:app --reload

MLflow logs to ./mlruns locally. No server required. If you want the full UI:

mlflow ui --port 5000

Or bring up the complete stack:

docker-compose up --build

Why I built this as open source

Three reasons.

First, I kept seeing junior engineers spend weeks building pipeline scaffolding that should take days. PulseFlow collapses that to a git clone.

Second, enterprise ML has a credibility problem with open source. Most OSS ML projects are notebooks or toy pipelines. PulseFlow is the kind of code I would put in front of a Duke Energy production environment.

Third, I am building ARGUS-AI alongside this. ARGUS is an LLM observability platform that evaluates every model output across six dimensions: Groundedness, Accuracy, Reliability, Variance, Inference Cost, Safety. PulseFlow is what you run your models through. ARGUS is how you know they are not degrading in production.

They compose. PulseFlow trains and serves. ARGUS monitors and evaluates.

What is in the repo

PulseFlow/
├── etl/                  # Data ingestion and preprocessing
├── training/             # Model training with MLflow tracking
├── deployment/           # FastAPI inference service
├── airflow/              # Orchestration DAGs
├── models/               # Model artifacts
├── ci_cd/                # GitHub Actions workflows
├── docker-compose.yml    # Full stack in one command
└── pyproject.toml        # pip install pulseflow-mlops

Live demo on Hugging Face

You can run the full ETL, training, and inference pipeline without installing anything:

PulseFlow MLOps Demo on Hugging Face Spaces

Three tabs. Load sample data, configure hyperparameters, run inference against the FastAPI endpoint simulation. All in the browser.

The production gap no one talks about

Most MLOps content stops at "train a model and log it to MLflow." That is maybe 20 percent of what production demands.

The other 80 percent:

What happens when your data source schema changes at 2 AM?
How do you roll back a model that passed validation but is failing on live traffic?
Who gets paged when inference latency exceeds SLA?
How do you prove to your compliance team that the model version in production matches what was approved?

PulseFlow gives you the structural patterns to answer all of these. It does not answer them for you because every organization's answers are different. But it gives you the right skeleton.

What I am adding next

LangChain integration for LLM pipeline orchestration
ARGUS-AI integration for automatic G-ARVIS scoring on inference outputs
Kubernetes deployment manifests (production-grade, not tutorials)
Prometheus metrics endpoint on the FastAPI service

Connect

GitHub: github.com/anilatambharii/PulseFlow
PyPI: pypi.org/project/pulseflow-mlops
ARGUS-AI (the observability layer): github.com/anilatambharii/argus-ai
Hugging Face: huggingface.co/AmbhariiLabs
LinkedIn newsletter Field Notes: Production AI: linkedin.com/in/anilsprasad

If you are building ML systems in production and running into the gaps PulseFlow addresses, reach out. This is open source because I want it to be the reference architecture the community builds on.

28 years of production AI. All opinions are mine. All lessons were expensive.

HumanWritten #ExpertiseFromField


---

## Step 4 — Publish settings

- **Series:** Leave blank for now (or create "Ambharii Labs Open Source" series later)
- **Schedule:** Publish immediately — Tuesday 9 AM ET is ideal but today is fine given the momentum
- Click **Publish**

---

## Step 5 — After publishing, copy the URL and do these immediately

https://dev.to/anilatambharii/your-new-article-slug

argus-llm is now on PyPI — production LLM observability in one pip install

Anil Prasad — Sat, 28 Mar 2026 20:01:46 +0000

argus-llm is now on PyPI — production LLM observability in one pip install

Hi DEV community — Anil Prasad here, Head of Engineering at Duke Energy
and Founder of Ambharii Labs. Just published argus-llm, an open-source
LLM observability framework built from 28 years of production AI experience.

Would love feedback from practitioners here.

pip install argus-llm

OpenTelemetry Traces Your LLM. It Does Not Fix It.

Anil Prasad — Wed, 25 Mar 2026 19:35:36 +0000

The DEV community is buzzing about OpenTelemetry standardizing LLM tracing. That is a real win. Spans, traces, semantic conventions for gen AI — all of it matters. I have been watching this space for a while.

But I want to say something that production experience has drilled into me.

Observability without correction is a dashboard full of problems you are still solving manually.

What Tracing Gives You
OpenTelemetry for LLMs gives you visibility into:

Latency per call
Token consumption
Span trees across your agent chain
Model inputs and outputs at each step

That is genuinely useful. I am not dismissing it.
But here is what it does not give you:

Detection that the output is hallucinated before it reaches your user
Automatic retry with a corrected prompt when groundedness fails
Cost circuit breakers that fire before your inference bill explodes
Safety flags that block a response instead of just logging that it was bad

You are still the correction layer. You are the human staring at a Grafana dashboard at 2am deciding whether to roll back a prompt.

What Production AI Actually Looks Like

I spent the last two years building AI systems at scale inside regulated industries. Healthcare revenue cycle. Power grid intelligence. Genomics pipelines. These are not playgrounds.

The pattern I kept seeing: teams invest heavily in logging and tracing. They build beautiful dashboards. And then when the LLM misbehaves in production, the process to correct it is still manual, slow, and incident-driven.

The gap is not observability. The gap is autonomous correction at the output layer.

Nobody had shipped that as a product. So I built it.

ARGUS: Autonomous Runtime Guardian for Unified Systems

ARGUS is an open-source LLM observability platform that goes one layer further than tracing. It evaluates six dimensions of LLM output in real time:

For agentic systems specifically, ARGUS adds three more signals:

ASF Agent Success Fraction — what percentage of agent tasks complete successfully
ERR Error Recovery Rate — how well the agent recovers from tool failures
CPCS Cost Per Completed Subtask — real cost accountability at the task level

When a dimension fails a threshold, ARGUS does not just log it. It triggers a correction loop.

The Architecture in One Diagram

LLM Call → ARGUS Eval Layer → Pass/Fail per Dimension
↓
Fail Detected
↓
Autonomous Correction Loop
(prompt rewrite + retry)
↓
Corrected Output → Your App

The observability layer is the open-source core (argus-ai on PyPI). The autonomous correction loop is the proprietary layer being built for enterprise deployment.

Why OpenTelemetry + ARGUS Is the Right Stack
I am not building against OpenTelemetry. The right mental model is:

OpenTelemetry = infrastructure observability, distributed tracing, the plumbing
ARGUS = semantic output evaluation, correction, quality guarantee

They compose. You can pipe ARGUS evaluation results as spans into your OTel collector. You get the full picture: infrastructure health AND output quality in the same trace.

What I Learned the Hard Way

At R1 RCM I led the engineering work that contributed to a $4.1B acquisition. The AI systems underpinning that work processed millions of healthcare claims. When the LLM got it wrong, it was not just a metric. It was a denied claim, a delayed payment, a patient impact.

Tracing told us what happened. Correction prevented why it would happen again.

That difference is what drove me to build ARGUS.

Get Started
pip install argus-ai

GitHub: github.com/anilatambharii/argus-ai
PyPI: pypi.org/project/argus-ai

If you are building LLM systems in production and want to collaborate, reach out. This is open-source and I want it to be the standard evaluation layer the community builds on.

25 years of production AI. All opinions are mine. All lessons were expensive.

HumanWritten #ExpertiseFromField