agent-ablation

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