agent-ablation (Python)

Leave-one-out ablation testing, backward elimination, and ROI evaluation for multi-agent systems in Python. You have a set of per-agent findings (scores, confidences, telemetry) and a function or model that turns those findings into a verdict. agent-ablation answers the key production questions:

1. Load-Bearing Influence: Which agents' findings actually changed the verdict, and which were along for the ride?
2. Cost & Token ROI: How many dollars and tokens did each specialist burn per verdict flip? Does that small accuracy bump justify the API bill?
3. Correlated Agents & Pruning: Which redundant agents can be safely pruned via greedy backward elimination without breaking the final verdict?
4. Protective vs. Harmful Signals: When ground truth is provided, did an agent's presence prevent an error (protective), or did it cause a hallucination/false positive (harmful)?
5. Stochastic & Async LLM Judges: Handles async decisions and repeated sampling with majority voting to filter out LLM temperature variance.

Zero runtime dependencies. Pure Python 3.9+.

Install

pip install agent-ablation

Core Features & Usage

1. Basic Leave-One-Out Ablation

from typing import List
from agent_ablation import Finding, run_ablation

def decide(findings: List[Finding]) -> str:
    # Noisy-OR combination
    survival = 1.0
    for f in findings:
        survival *= (1.0 - f.score / 100.0)
    risk = 1.0 - survival

    if risk >= 0.7:
        return "decline"
    if risk <= 0.3:
        return "approve"
    return "escalate"

findings = [
    Finding(agent_id="transaction_pattern", score=25),
    Finding(agent_id="identity_signal", score=90),
    Finding(agent_id="network_analysis", score=20),
]

result = run_ablation(findings, decide)

print(result.baseline)           # "decline"
print(result.load_bearing_ratio) # 0.33 (1 out of 3 agents flipped the outcome)
for p in result.per_agent:
    print(p.removed_agent_id, "->", p.verdict_without, "(load-bearing)" if p.changed else "")

2. Cost & Token ROI Analysis ("Cost per Verdict Flip")

Pass telemetry (cost, tokens, latency_ms) inside your findings. batch_ablation computes the exact ROI metrics and identifies expensive agents with low decision impact:

from agent_ablation import Finding, batch_ablation, format_markdown_report

cases = [
    [
        Finding(agent_id="expensive_reasoner", score=90, cost=0.15, tokens=3000),
        Finding(agent_id="cheap_heuristic", score=10, cost=0.002, tokens=50),
    ],
    [
        Finding(agent_id="expensive_reasoner", score=20, cost=0.15, tokens=3000),
        Finding(agent_id="cheap_heuristic", score=85, cost=0.002, tokens=50),
    ],
]

results, summary = batch_ablation(cases, decide)

print(summary.roi.agents["expensive_reasoner"].cost_per_verdict_flip) # Cost per decision flip
print(summary.roi.recommendations) # Automated pruning/downgrade advice

# Format into a Markdown report for PRs or documentation
print(format_markdown_report(summary))

3. Async & Stochastic Decision Functions (LLM-as-a-Judge)

When your decision step is an async LLM call with temperature, use run_ablation_async or batch_ablation_async. Set samples=k to take a majority-vote consensus across runs to eliminate sampling noise:

import asyncio
from agent_ablation import Finding, run_ablation_async

async def llm_supervisor_decide(findings: List[Finding]) -> str:
    res = await call_llm_judge(findings)
    return res["verdict"]

result = await run_ablation_async(
    findings,
    llm_supervisor_decide,
    samples=5,  # Runs 5 samples per ablation to filter out temperature noise
)

4. Greedy Backward Elimination & Minimal Viable Panel

If you have correlated or redundant agents (e.g., three critics looking at the same context), simple leave-one-out might mark all of them as not load-bearing because the others compensate.

run_backward_elimination iteratively eliminates agents one-by-one until removing any further agent flips the verdict, revealing the minimal viable panel:

from agent_ablation import run_backward_elimination

result = run_backward_elimination(all_specialists, decide)

print(result.minimal_agent_ids)   # ["critic_1", "security_auditor"]
print(result.eliminated_agent_ids) # ["critic_2", "critic_3", "scout_noisy"]
print(result.steps)               # Step-by-step elimination trace

For detecting 2nd-order joint dependencies, run_pairwise_ablation(findings, decide) evaluates all pairs $(A, B)$ to catch cases where neither agent alone is load-bearing, but removing both together flips the outcome.

5. Ground-Truth & Net Accuracy Impact ("Protective vs. Harmful")

Supply ground truth labels in batch_ablation to measure whether an agent's load-bearing presence actually improved accuracy or injected errors / hallucinations:

results, summary = batch_ablation(
    cases,
    decide,
    ground_truth=["approve", "decline", "approve", "escalate"],
)

# Per-agent stats:
# - Protective: removing the agent caused a correct verdict to become incorrect
# - Harmful: removing the agent fixed an incorrect verdict
print(summary.per_agent_stats["hallucinating_agent"].role)               # "Harmful"
print(summary.per_agent_stats["hallucinating_agent"].net_accuracy_impact) # -0.25

Framework Adapters

Zero-dependency adapters to map telemetry and messages from popular agent frameworks directly into Finding:

LangGraph / LangChain

from agent_ablation import from_langgraph_messages

findings = from_langgraph_messages(
    state["messages"],
    score_of=lambda msg: msg["content"]["score"],
    confidence_of=lambda msg: msg["content"].get("confidence"),
)

CrewAI

from agent_ablation import from_crewai_tasks

findings = from_crewai_tasks(
    crew_output.tasks_output,
    score_of=lambda task: task.json_dict.get("score", 0),
)

AutoGen

from agent_ablation import from_autogen_messages

findings = from_autogen_messages(
    chat_history,
    score_of=lambda msg: msg["content"]["risk_score"],
)

Vercel AI SDK / Trace Steps

from agent_ablation import from_ai_sdk_steps

findings = from_ai_sdk_steps(
    steps,
    score_of=lambda step: step["result"]["score"],
)

Generic Custom Records

from agent_ablation import from_records

findings = from_records(
    custom_audit_records,
    agent_id=lambda r, idx: r.specialist_id,
    score_of=lambda r, idx: r.risk_score,
    confidence_of=lambda r, idx: r.confidence_level,
)

Worked Example: SentryMesh 33% Multi-Signal Finding

SentryMesh is a four-specialist multi-agent fraud investigation system. Its eval harness runs an ablation over its 23-case bank and reports: of 9 cases auto-resolved without human escalation, only 3 survive removal of their single loudest specialist — 6 collapse to escalate.

tests/test_ablation.py reproduces all 6 cases verbatim with agent-ablation.

Architectural Note: Leaf Ablation vs. DAG Subgraph Replay

• Leaf Finding Ablation (This Package): Best for parallel / fan-out / fan-in panels where specialists independently produce findings that feed a decision gate. Because findings are generated independently, dropping an item at decide() measures causal weight with zero LLM re-invocation cost.
• Sequential DAG Replay: If your pipeline is sequential (Agent A feeds intermediate prompt context to Agent B), removing Agent A at the final decision gate misses that Agent B's output already reflects Agent A. Measuring sequential pipelines requires replaying downstream subgraphs or injecting mock messages into the trace.

API Summary

License

MIT © Ayush Verma — ayushv3533e@gmail.com