DEV Community: Anil Prasad

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