DEV Community: Nilofer 🚀

MoE Cost Analyzer: Benchmark Dense vs Mixture-of-Experts LLMs on Your Own Prompts with Real API Calls

Nilofer 🚀 — Mon, 20 Jul 2026 15:57:55 +0000

Every team running LLMs in production eventually asks the same question: should we switch to a Mixture-of-Experts model? The MoE architecture activates fewer parameters per token, which means cheaper inference on shared infrastructure. But switching blindly is a gamble. Your prompts, your traffic volume, and your latency SLA determine whether the tradeoff actually pays off.

MoE Cost Analyzer is a benchmarking tool that answers that question with hard numbers rather than theory. It takes a JSON file of your real production prompts, runs live API calls against both a dense and a MoE model via OpenRouter, measures actual latency and cost per query, and returns a concrete recommendation against SLA thresholds you define. No simulations. No estimates. Real calls, real results, one decision.

To validate the tool, it was run on a 100-task sentiment-analysis benchmark against two Gemma variants: the dense google/gemma-4-31b-it and the MoE google/gemma-4-26b-a4b-it. Both models received identical prompts via live OpenRouter API calls. Every response was timed, every token counted, every dollar logged. The results below come from those 200 real API calls, with zero errors.

The Report: Validated Live Results

Benchmark: 100 sentiment-analysis tasks
API: OpenRouter (live calls, not simulated)
Total calls: 200 (100 per model) | Errors: 0

Verdict: USE MoE. It met both SLA targets (latency < 3 000 ms, cost < $0.02/1K queries).

Reading the Results

The MoE model is cheaper because it activates fewer parameters per token.
Both models produced identical token counts (4 939 total). The output quality is equivalent, but the MoE model costs 20% less per token on OpenRouter because inference is cheaper on sparse architectures.

The MoE model is faster at median load (P50: -27%) but the gap narrows under stress (P95: -13%).
P50 shows typical behaviour: MoE is noticeably snappier. P95 shows tail latency under OpenRouter's shared infrastructure, and both models slow down similarly under peak load, so MoE still wins but by a smaller margin. For latency-sensitive SLAs check your own P95 against your threshold.

Token count is identical. Output quality is maintained.
Average prompt tokens (~49) and completion tokens (~20) were the same across both models, confirming the architecture change does not degrade output length or structure.

Zero errors across 200 live API calls.
The retry logic (3x exponential back-off on 429/5xx) was never triggered. Both models were stable throughout.

Extrapolated Production Impact

Based on real measured average of ~49.4 tokens/query.

Your numbers will differ, which is exactly why you should run this tool on your own prompts.

The Tool: Run It on Your Own Workload

The report above is a fixed snapshot. Your production prompts are longer, your SLA is different, your traffic pattern is unique. Run the analyzer on your own benchmark to get a recommendation that actually applies to you.

Setup

git clone https://github.com/dakshjain-1616/MoE-Cost-Analyzer.git
cd MoE-Cost-Analyzer
pip install -r requirements.txt
cp .env.example .env
# add your key:  OPENROUTER_API_KEY=sk-or-v1-...

Prepare your benchmark

Create a JSON file with your real production prompts:

{
  "name": "My Production Benchmark",
  "tasks": [
    {
      "id": "task_001",
      "prompt": "Extract all dates mentioned in this support ticket: ...",
      "expected_labels": ["2026-01-15", "2026-02-03"]
    },
    {
      "id": "task_002",
      "prompt": "Classify the urgency of this message as Low / Medium / High: ...",
      "expected_labels": ["High"]
    }
  ]
}

Use 50–200 representative prompts. The included benchmark.json has 100 sentiment tasks you can use as a starting point.

Run

# Live run against OpenRouter
python analyze.py my_benchmark.json

# With custom SLA thresholds
python analyze.py my_benchmark.json \
  --sla-latency-ms 1500 \
  --sla-cost-per-1k 0.005 \
  --output my_results.csv

# Dry-run (no API calls, uses simulated data — good for testing the pipeline)
python analyze.py benchmark.json --dry-run

What You Get

1. A Rich terminal table: colour-coded decision matrix printed to stdout:

╭────────────────────────┬──────────────────────┬──────────────────────┬──────────╮
│ Metric                 │ Dense                │ MoE                  │ MoE vs Dense │
├────────────────────────┼──────────────────────┼──────────────────────┼──────────┤
│ Avg Latency (ms)       │ 1721.3               │ 1283.2               │ -25.5%       │
│ P50 Latency (ms)       │ 833.2                │ 605.1                │ -27.3%       │
│ P95 Latency (ms)       │ 6748.1               │ 5879.2               │ -12.9%       │
│ Avg Cost / Query (USD) │ $0.000005            │ $0.000004            │ -20.0%       │
│ Total Cost (USD)       │ $0.0005              │ $0.0004              │ -20.0%       │
│ Total Tokens           │ 4939                 │ 4939                 │ 0.0%         │
│ Error Rate             │ 0.0%                 │ 0.0%                 │ N/A          │
╰────────────────────────┴──────────────────────┴──────────────────────┴──────────╯

Summary: MoE is 20.0% cheaper and 25.5% faster than the dense model.

✓  Recommendation: USE MoE — latency 1283ms < 3000ms SLA and cost $0.0040/1K < $0.0200/1K SLA

2. A per-query CSV: every task, every model, every metric:

task_id,model_id,latency_ms,prompt_tokens,completion_tokens,total_tokens,cost_usd,error
task_001,google/gemma-4-26b-a4b-it,816.6,60,20,80,0.0000064,
task_001,google/gemma-4-31b-it,1384.2,60,20,80,0.0000080,
task_002,google/gemma-4-26b-a4b-it,897.0,60,20,80,0.0000064,

Load it in pandas for deeper analysis:

import pandas as pd
df = pd.read_csv("results.csv")
print(df.groupby("model_id")[["latency_ms", "cost_usd"]].describe())

3. A recommendation string: one of three outcomes:

✓ USE MoE: MoE meets both your latency and cost SLAs. Switch.

~ MARGINAL: MoE meets one SLA but not both. Review the failing dimension.

✗ STICK WITH DENSE: MoE fails one or both SLAs at your thresholds. Do not switch yet.

CLI Reference

python analyze.py <benchmark_file> [options]

Options:
  --dense-model       Dense model ID       (default: google/gemma-4-31b-it)
  --moe-model         MoE model ID         (default: google/gemma-4-26b-a4b-it)
  --sla-latency-ms    Avg latency cap (ms) (default: 2000)
  --sla-cost-per-1k   Cost cap per 1K reqs (default: 0.01)
  --output            CSV output path      (default: results.csv)
  --dry-run           Simulate API calls, no cost incurred

Edge Cases and Reliability

Empty benchmark file: Prints "No tasks to evaluate." and exits cleanly.

Invalid model ID: Prints "Error: Invalid model ID" and exits with code 1.

Missing API key: Suggests --dry-run and exits with code 1.

Rate limit (429): Retries up to 3x with exponential back-off (1s, 2s, 4s).

Server error (5xx): Same retry logic.

Partial failures: Error rate shown in matrix; successful tasks still analysed.

How It Works

your_benchmark.json
        │
        ▼
   analyze.py  ─── validates model IDs + task list
        │
        ▼
   runner.py   ─── asyncio.gather (semaphore = 5 concurrent)
                       ├── google/gemma-4-31b-it   ──► OpenRouter API
                       └── google/gemma-4-26b-a4b-it ─► OpenRouter API
        │
        ▼
  analyzer.py  ─── per-model stats: avg / p50 / p95 latency, cost
               ─── decision matrix DataFrame
               ─── recommend() against your SLA thresholds
               ─── save_csv()
        │
        ▼
  reporter.py  ─── Rich terminal table + coloured recommendation

Concurrent requests are capped at 5 to stay within OpenRouter rate limits. Adjust asyncio.Semaphore(5) in src/runner.py if your plan allows higher throughput.

Pricing

Pricing is approximate and subject to change. Check OpenRouter for current rates. Update src/pricing.py when OpenRouter rates change. The ModelPricing dataclass also stores context_window and model type for future extensions.

Tests

pytest tests/ -v
# 18 passed in 5.11s

test_analyzer.py: Stats computation, decision matrix shape, recommendation logic, error-rate tracking.

test_runner.py: Pricing calculation, model ID validation, cost ordering.

test_cli.py: Empty benchmark → clean exit, invalid model ID → error message, dry-run → CSV written.

Project Structure

moe-cost-analyzer/
├── analyze.py              ← entry point
├── benchmark.json          ← 100-task example benchmark
├── requirements.txt
├── .env.example
├── conftest.py
├── docs/
│   └── benchmark_results.svg  ← chart generated from live results
├── src/
│   ├── pricing.py          ← ModelPricing dataclass + compute_cost()
│   ├── runner.py           ← async OpenRouter calls, retry logic, dry-run
│   ├── analyzer.py         ← stats, decision matrix, CSV export
│   └── reporter.py         ← Rich terminal output
└── tests/
    ├── test_analyzer.py
    ├── test_runner.py
    ├── test_cli.py
    └── fixtures/
        └── empty_benchmark.json

How I Built This Using NEO

This project was built using NEO. NEO is a fully autonomous AI engineering agent that can write code and build solutions for AI/ML tasks including AI model evals, prompt optimization and end to end AI pipeline development.

The requirement was a tool that runs live API calls against both a dense and a MoE model on a user-supplied benchmark, measures real latency and cost, and returns a data-backed recommendation against configurable SLA thresholds. NEO planned and produced the files in this repository: the CLI entry point in analyze.py, the async OpenRouter runner with retry logic and dry-run mode in src/runner.py, the stats and decision matrix engine in src/analyzer.py, the Rich terminal reporter in src/reporter.py, the pricing dataclass in src/pricing.py, three test files covering stats computation, pricing, and CLI behavior, and the 100-task example benchmark in benchmark.json.

The result is a fully working analyzer that takes any JSON benchmark, makes live API calls to both models concurrently, and returns a colour-coded terminal table, a per-query CSV, and a single recommendation string in seconds.

How You Can Use This With NEO

Run it on your own production prompts before switching models.
Replace benchmark.json with 50–200 of your actual production prompts, set your own --sla-latency-ms and --sla-cost-per-1k thresholds, and get a recommendation grounded in your real workload rather than a generic benchmark.

Use --dry-run to test the pipeline without spending API budget.
The dry-run mode uses simulated data and produces the same CSV and terminal output as a live run, making it safe to validate the setup and SLA threshold logic before committing to real API calls.

Load the per-query CSV in pandas for deeper analysis.
Every task, every model, every latency and cost measurement is written to a CSV. Grouping by model_id and running describe() on latency_ms and cost_usd surfaces distribution details that the summary table alone does not show.

Point it at any two OpenRouter models, not just Gemma.
The --dense-model and --moe-model flags accept any valid OpenRouter model ID. The same benchmark, the same SLA thresholds, and the same recommendation logic work across any pair of models you want to compare.

Final Notes

The question "should we switch to a MoE model?" comes up every time a team hits a cost or latency wall. The answer is almost always "it depends on your prompts", which is why a tool that runs on your actual workload and returns a gated recommendation is more useful than any benchmark you did not run yourself.

The code is at https://github.com/dakshjain-1616/MoE-Cost-Analyzer
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Infra-Sizer: A CLI That Recommends the Right GPU Infrastructure for Any LLM Deployment

Nilofer 🚀 — Fri, 17 Jul 2026 06:23:05 +0000

"What hardware do I need to serve this model?" is the first question every team asks before deployment. The answer requires juggling model size, throughput benchmarks, AWS pricing, batching math, and serving framework tradeoffs simultaneously.

infra-sizer answers it in seconds. It is a CLI that takes a model ID, expected request volume, latency SLA, and context length, then outputs a concrete GPU recommendation: instance type, count, serving framework, batching config, and monthly cost estimate. Built autonomously using NEO.

The Problem

Getting the hardware wrong is expensive in both directions.

Overprovision: Wasted $10k+/month on idle H100s.

Underprovision: Latency SLA breaches and on-call pages at 3am.

Wrong framework: vLLM vs TGI vs llama.cpp have 5x throughput differences for certain workloads.

Infra-Sizer gives you the right answer in seconds, not hours of spreadsheet work.

Architecture

The tool moves through five steps: the CLI parses inputs and fetches model metadata from the HuggingFace API, the GPU database looks up throughput benchmarks and AWS on-demand pricing across 6 GPU tiers, the planner sends the constraints and data to Llama-4-Maverick via OpenRouter for LLM-assisted reasoning, and the output is a Rich recommendation table covering GPU tier, instance count, framework, batch size, cost, and confidence score. If no API key is set, a rule-based fallback produces the recommendation without the LLM.

How It Works

Parse CLI args: model ID, requests-per-second, latency SLA (seconds), context length
Fetch model metadata from HuggingFace API: parameter count, architecture, dtype
Query GPU database: look up throughput benchmarks and AWS on-demand pricing for 6 GPU tiers
LLM reasoning: send constraints and data to Llama-4-Maverick via OpenRouter, which reasons over tradeoffs and recommends a config
Rule-based fallback: if OPENROUTER_API_KEY is unset, a deterministic algorithm produces a recommendation without LLM

GPU Database Coverage

Installation

git clone https://github.com/neo-engine/infra-sizer.git
cd infra-sizer
pip install -r requirements.txt

Requirements: Python 3.11+, openai, requests, rich, pandas

Usage

python main.py --model Qwen/Qwen3-32B --rps 50 --latency-sla 2.0 --context 8192

Arguments

Environment Variables

export OPENROUTER_API_KEY=your_key_here  # Enables LLM-assisted reasoning

Without the key, the tool falls back to rule-based recommendations. No crash, still useful.

Output

╔══════════════════════════════════════════════════════════════╗
║          Infra-Sizer — Infrastructure Recommendation         ║
╠══════════════════════════════════╦═══════════════════════════╣
║ GPU Tier                         ║ A100-80G                  ║
║ Instance Count                   ║ 2                         ║
║ Serving Framework                ║ vLLM                      ║
║ Recommended Batch Size           ║ 32                        ║
║ Max Concurrent Requests          ║ 64                        ║
║ Estimated Monthly Cost (AWS)     ║ $32,058                   ║
║ Confidence Score                 ║ 0.85                      ║
╚══════════════════════════════════╩═══════════════════════════╝

Test Cases

# Small model, low load → CPU or single L4
python main.py --model 3B --rps 5 --latency-sla 10.0 --context 2048

# Large model, high load, tight SLA → multi-GPU H100
python main.py --model 70B --rps 100 --latency-sla 1.0 --context 4096

# Impossible constraint → "No viable configuration" + closest feasible
python main.py --model 1T --rps 1000 --latency-sla 0.1 --context 128000

File Structure

infra-sizer/
├── main.py              # CLI entrypoint
├── hf_metadata.py       # HuggingFace model card fetcher
├── gpu_db.py            # GPU performance database + cost calculator
├── planner.py           # LLM-assisted recommendation engine (OpenRouter)
├── data/
│   ├── gpu_specs.json   # Throughput benchmarks per GPU per model size
│   └── aws_pricing.json # AWS on-demand pricing per instance type
├── requirements.txt
└── README.md

Why This Exists

"Just throw it on an A100" is the most expensive mistake in AI infrastructure. The right hardware choice depends on model size, throughput requirements, latency SLA, and cost constraints, all at once. This tool does that math in seconds so you can make an informed decision before signing a cloud contract.

How I Built This Using NEO

The requirement was a CLI that takes an LLM model ID and deployment constraints, then outputs a concrete GPU infrastructure recommendation backed by real throughput benchmarks and AWS pricing. NEO planned and produced the files in this repository: the CLI entrypoint in main.py, the HuggingFace metadata fetcher, the GPU performance database and cost calculator, the LLM-assisted planner using Llama-4-Maverick via OpenRouter with a rule-based fallback, and the two data files covering GPU throughput benchmarks and AWS on-demand pricing across 6 GPU tiers.

The result is a fully working CLI that takes four inputs and returns a concrete infrastructure recommendation with GPU tier, instance count, serving framework, batch size, monthly cost, and confidence score, in seconds.

How You Can Use This With NEO

Run it before signing a cloud contract.
Pass in the model ID, expected RPS, latency SLA, and context length and get a concrete GPU recommendation with a monthly cost estimate before committing to any instance type. Takes seconds instead of hours of spreadsheet work.

Test impossible constraints to find the closest feasible configuration.
Passing an infeasible combination like a 1T-parameter model at 1000 RPS with a 0.1-second SLA returns "No viable configuration" alongside the closest feasible option, so you know exactly where the constraint breaks before you hit it in production.

Use it without an API key for quick rule-based estimates.
Without OPENROUTER_API_KEY set, the tool falls back to a deterministic algorithm and still produces a recommendation. Useful for quick checks in environments where external API calls are not allowed.

Use it as a pre-deployment checklist step.
Before any model goes to production, running Infra-Sizer with the actual expected load validates whether the planned instance type and count will meet the latency SLA, with a confidence score attached to the recommendation.

Final Notes

Every team deploying an LLM in production faces the same sizing problem, and most solve it the same way: by guessing, overprovisioning, and paying for it. Infra-Sizer makes that decision explicit, data-backed, and fast.

The code is at https://github.com/dakshjain-1616/infra-sizer
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

NeuralUCB Router: An OpenAI-Compatible API Proxy That Routes LLM Requests Using a Multi-Armed Bandit Algorithm

Nilofer 🚀 — Thu, 16 Jul 2026 18:55:37 +0000

Every team using LLMs in production hits the same wall: you don't need GPT-4 for every request, but you don't know in advance which ones actually need it. A user asking "what is 2+2?" and a user asking "find the bug in this recursive async Rust function" both hit the same /v1/chat/completions endpoint. Route them both to GPT-4o and you're paying $15 per million tokens for arithmetic. Route them both to a local model and quality collapses on complex work.

NeuralUCB Router solves this. It is a production-quality OpenAI-compatible API proxy that uses the NeuralUCB multi-armed bandit algorithm to dynamically route requests to the cheapest model while retaining 92%+ quality. Built autonomously using NEO.

The Problem

The naive solutions all fail:

Always use GPT-4o: $1,500 per month for 100K queries, most of it wasted on simple tasks.

Always use cheap or local model: Quality collapses on hard reasoning, code, and math.

Hard-coded if/else rules: Brittle. You cannot enumerate every query type, and rules do not adapt.

Random routing: No learning. Permanently suboptimal, ignores what is actually working.

Fine-tune a classifier: Requires labeled data, offline training, separate deployment, manual updates.

The root issue: routing is a sequential decision problem under uncertainty. Each query is different, the right model depends on context you can only partially observe, and you need to keep learning as usage patterns shift.

The Solution: NeuralUCB Bandit Routing

NeuralUCB Router frames LLM selection as a multi-armed bandit problem: the same class of problem used in ad auctions, clinical trials, and recommendation systems. Each LLM is an "arm". Each request is a "round". The goal: maximize cumulative reward (quality divided by cost) over time.

A classifier needs labeled training data telling it "this query → use GPT-4". A bandit generates its own training signal by trying models and observing outcomes. It learns online, from live traffic, with no pre-labeled dataset.

How NeuralUCB works for each incoming request:

1. Extract context features
   [query_length, token_entropy, is_code, is_math,
    time_of_day, session_cost, recent_quality, ...]

2. For each LLM arm, estimate:
   UCB score = predicted_reward(context) + exploration_bonus

   Where:
   - predicted_reward  = 2-layer MLP(context)   ← "exploitation"
   - exploration_bonus = λ · sqrt(gᵀ Z⁻¹ g)     ← "exploration"
     g = gradient of MLP w.r.t. last layer
     Z = covariance matrix (Sherman-Morrison updates, O(n²) not O(n³))

3. Route to arm with highest UCB score

4. Observe quality + cost → compute reward → update MLP weights + Z matrix

The UCB bonus is the key insight. A model that hasn't been tried much has high uncertainty, which means a high exploration bonus and more selections until its true value is known. Once well-characterized, the bonus shrinks and the MLP prediction dominates.

What happens after convergence (~500 requests):

Simple factual queries   →  local Ollama/Llama3    (free,   ~98% of the time)
Code generation          →  GPT-4o-mini            ($0.0006, ~87% of the time)
Complex reasoning/debug  →  GPT-4o                 ($0.015,  ~71% of the time)
Creative writing         →  Claude Haiku            ($0.0003, ~82% of the time)

The result: 88% cost reduction versus always-GPT-4o, while retaining 8.9/10 average quality.

Cost Savings

Key Features

Dynamic Model Routing: Automatically selects the best model for each request based on context
Cost Optimization: Routes 80% of requests to free/cheap models, saving 75%+ on API costs
Quality Retention: Maintains 92%+ quality by using expensive models for complex tasks
OpenAI-Compatible: Drop-in replacement for OpenAI API (/v1/chat/completions, /v1/models)
Real-time Dashboard: Streamlit dashboard with auto-refresh, cost savings meter, routing heatmap
Multiple Backends: Supports Ollama (local), OpenAI, Anthropic providers
Audit Logging: SQLite audit log with CSV/Parquet export for analysis

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    NeuralUCB Router                             │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  User Request → /v1/chat/completions                            │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  Context Extraction                                     │    │
│  │  • Prompt length, task type, temporal features          │    │
│  └─────────────────────────────────────────────────────────┘    │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  NeuralUCB Bandit                                       │    │
│  │  • 2-layer MLP per model (hidden=64)                    │    │
│  │  • UCB = f(x) + λ√(gᵀZ⁻¹g)                              │    │
│  │  • Sherman-Morrison rank-1 updates                      │    │
│  └─────────────────────────────────────────────────────────┘    │
│       │                                                         │
│       ▼                                                         │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  Model Selection                                        │    │
│  │  ┌─────┐ ┌─────┐ ┌─────┐                                │    │
│  │  │LLM A│ │LLM B│ │LLM C│ ← Cheapest + Quality           │    │
│  │  └─────┘ └─────┘ └─────┘                                │    │
│  └─────────────────────────────────────────────────────────┘    │
│       │                                                         │
│       ▼                                                         │
│  Response with audit logging                                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Quick Start

Docker Deployment

# Clone repository
git clone https://github.com/dakshjain-1616/neuralucb-router.git
cd neuralucb-router

# Set API keys
export OPENAI_API_KEY=your-key
export ANTHROPIC_API_KEY=your-key

# Start services
docker-compose up -d

# Access router
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{"model": "auto", "messages": [{"role": "user", "content": "Hello"}]}'

# Access dashboard
open http://localhost:8501

Local Installation

# Install dependencies
pip install -r requirements.txt

# Run router
python -m router.proxy.server

# Run dashboard
streamlit run dashboard/app.py --port 8501

Configuration

Edit configs/example_config.yaml:

models:
  - name: llama3.2-1b
    provider: ollama
    base_url: http://localhost:11434
    cost_per_1k: 0.0
  - name: gpt-4o-mini
    provider: openai
    cost_per_1k: 0.00015
  - name: claude-haiku-4-5
    provider: anthropic
    cost_per_1k: 0.00025

bandit:
  hidden_dim: 64
  lambda_param: 1.0
  exploration_weight: 0.1
  learning_rate: 0.01

Bandit Convergence

Model selection distribution across 1,000 requests:

The bandit learns to route 80% of requests to free Llama3.2-1b while maintaining quality on complex tasks.

Technical Details

NeuralUCB Algorithm

Architecture: 2-layer MLP (hidden=64, ReLU activation, Sigmoid output) per model arm
UCB Formula: UCB(model) = f(x; θ_m) + λ * sqrt(gᵀ Z⁻¹ g)
f(x; θ_m): MLP prediction for model m
g: Gradient of f with respect to θ_m
Z: Covariance matrix updated via Sherman-Morrison
λ: Exploration weight
Updates: Sherman-Morrison rank-1 updates for Z⁻¹ (O(n²) vs O(n³) for full inverse)

Context Features (15 dimensions)

prompt_length: Normalized token count
task_type: One-hot [code, math, creative, qa, chat, other]
avg_token_length: Average token length
has_system_prompt: Binary flag
time_of_day: Sin/cos encoding
latency_ema: Per-model exponential moving average

Reward Calculation

Two modes:

Embedding Cosine Similarity: Compare response to reference using sentence-transformers
LLM-as-Judge: Use LLM to score quality (1-5 scale)

Reward = quality / cost_per_1k (normalized)

Testing

# Run all tests
pytest tests/ -v

# Run specific test
pytest tests/test_neural_ucb.py -v
pytest tests/test_context.py -v
pytest tests/test_proxy.py -v

All tests mock LLM calls. No API keys required for testing.

Dashboard Features

Auto-refresh: Updates every 2 seconds
Cost Savings Meter: Cumulative cost tracking
Model Performance Bar Chart: Selections and average rewards
Routing Heatmap: Visualize routing decisions over time
Recent Decisions Table: Last 20 routing events with metrics

API Endpoints

Project Structure

neuralucb-router/
├── router/bandit/
│   ├── neural_ucb.py     # NeuralUCB bandit algorithm
│   ├── context.py        # Context feature extraction
│   ├── reward.py         # Reward calculation
│   └── __init__.py
├── router/backends/
│   ├── base.py           # Backend provider ABC
│   ├── ollama.py         # Ollama integration
│   ├── openai_backend.py # OpenAI integration
│   ├── anthropic_backend.py # Anthropic integration
│   └── __init__.py
├── router/proxy/
│   ├── server.py         # FastAPI server
│   ├── middleware.py     # Auth/rate limiting
│   ├── openai_schema.py  # Pydantic schemas
│   └── __init__.py
├── router/audit/
│   ├── logger.py         # SQLite audit logging
│   ├── export.py         # CSV/Parquet export
│   └── __init__.py
├── router/config.py      # Configuration loader
├── dashboard/app.py      # Streamlit dashboard
├── tests/
│   ├── test_neural_ucb.py
│   ├── test_context.py
│   ├── test_proxy.py
│   └── __init__.py
├── configs/
│   ├── example_config.yaml
├── hf_export/
│   ├── README.md
│   ├── config.json
│   ├── router_state.json
│   ├── push_to_hub.py
├── infographics/
│   ├── routing_flow.txt
│   ├── cost_comparison.txt
│   ├── bandit_convergence.txt
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── pyproject.toml
└── README.md

Export to Hugging Face

from hf_export.push_to_hub import export_to_hub

export_to_hub(
    repo_id="your-username/neuralucb-router",
    config_path="configs/example_config.yaml",
    state_path="hf_export/router_state.json"
)

Comparison

How I Built This Using NEO

The requirement was a production-quality OpenAI-compatible API proxy that uses the NeuralUCB multi-armed bandit algorithm to dynamically route LLM requests based on context, cutting costs while retaining quality. NEO planned and produced the files in this repository: the bandit algorithm, context extraction, and reward calculation modules, three backend integrations for Ollama, OpenAI, and Anthropic, a FastAPI proxy server with middleware and Pydantic schemas, a SQLite audit logger with CSV and Parquet export, a Streamlit dashboard, a three-suite test setup, HuggingFace export tooling, Docker and compose files, and the configuration system.

The result is a fully working intelligent router that learns from live traffic, converges after roughly 500 requests, routes 80% of requests to free or cheap models, and delivers 8.9/10 quality at 12% of the cost of always using GPT-4o.

How You Can Use This With NEO

Drop it in as a zero-config OpenAI proxy and immediately start cutting costs.
Since the router exposes a fully OpenAI-compatible /v1/chat/completions endpoint, any existing application pointing at the OpenAI API needs only its base URL changed to start routing intelligently. No code changes required beyond that one configuration line.

Add your own models by editing a single YAML file.
Any provider supported by Ollama, OpenAI, or Anthropic can be added to configs/example_config.yaml with a name, provider, base URL, and cost per 1K tokens. The bandit picks it up on the next start and begins exploring it automatically alongside existing arms.

Use the audit log to understand exactly how your traffic is being routed.
Every routing decision is written to SQLite with full metrics. The /v1/router/audit endpoint and the CSV/Parquet export let teams pull this data into any analytics tool and see which models are handling which query types, what quality scores they are producing, and where cost savings are actually coming from.

Run the Streamlit dashboard as a live cost monitoring layer for your team.
The dashboard auto-refreshes every 2 seconds and shows cumulative cost savings, model selection distribution, and the last 20 routing decisions with metrics. Useful for keeping non-technical stakeholders informed without building a separate reporting tool.

Final Notes

Most teams end up on one of two bad equilibria: spending too much on a frontier model for every request, or accepting degraded quality from a cheap model across the board. NeuralUCB Router sits between those two by learning which requests actually need the expensive model and routing everything else to the cheapest option that can handle it.

The code is at https://github.com/dakshjain-1616/neuralucb-router
The model is on HuggingFace at https://huggingface.co/daksh-neo/neuralucb-router
Install via pip: pip install neuralucb-router
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

AI Cost Optimizer CLI: Real-Time AI Cost Tracking, Model Routing, and Shadow AI Detection from the Terminal

Nilofer 🚀 — Wed, 15 Jul 2026 13:47:44 +0000

Engineering teams running AI at scale eventually hit the same wall: no one knows exactly how much each model call costs, which GPU is idle, whether a cheaper model would have done the job just as well, or whether someone in the organization is quietly using an unauthorized AI service. The answers exist in the data, but nothing surfaces them without instrumentation. By the time costs are visible, the spend has already happened.

AI Cost Optimizer CLI brings that visibility into the terminal before it becomes a problem. It is a command-line toolkit designed for engineering teams who use AI and LLM APIs at scale: real-time cost tracking at the token level, GPU utilization monitoring, intelligent model routing, caching analysis, shadow AI detection, and comprehensive reporting, all from a single CLI with no external API keys and no cloud dependencies.

What This Tool Does

Six capabilities work together from one command entry point.

The cost tracker logs every API call with precise attribution: prompt tokens multiplied by input price plus completion tokens multiplied by output price, per call, per session, per project, stored in a local SQLite database. The GPU monitor queries NVML for live metrics (utilization, memory, temperature, power draw) or generates realistic random-walk simulations when no NVIDIA hardware is present. The model router classifies task complexity using keyword analysis and recommends the cost-optimal model, routing low-complexity tasks to SLMs where the savings range from 60 to 90 percent versus frontier models. The caching advisor analyzes prompt patterns and estimates dollar savings across four caching strategies with daily, monthly, and yearly projections. The shadow AI scanner parses DNS packet captures to detect unauthorized AI API usage across 25+ known domains. The report generator aggregates all of this into a single actionable output: current cost, optimized cost, potential savings, and the routing change that gets there.

A pricing database of 25+ models across 9 providers is built in. No external API keys are needed. No cloud dependencies. GPU monitoring works in simulation mode when no NVIDIA GPU is present.

Architecture

The CLI is built on click, with each subcommand routing to an independent backend module:

┌──────────────────────────────────────────────────────────────────────────┐
│                           neo-cost CLI (click)                            │
│                                                                          │
│   cost     gpu      route     cache     shadow    report    monitor      │
│   ─────   ─────   ───────   ───────   ───────   ───────   ───────      │
└──────────────────────────────────────────────────────────────────────────┘
         │         │         │          │         │          │
         ▼         ▼         ▼          ▼         ▼          ▼
┌─────────────┐ ┌─────────┐ ┌──────────┐ ┌─────────────┐ ┌────────┐
│ CostTracker │ │GPUMonit │ │ModelRoute│ │CachingAdvis │ │ShadowAI│
│             │ │ or      │ │ r        │ │ or          │ │Scanner │
│  ┌───────┐  │ │         │ │          │ │             │ │        │
│  │SQLite │  │ │pynvml + │ │PricingDB │ │PromptAnalys │ │dpkt +  │
│  │  DB   │  │ │ Demo    │ │ +        │ │ is +        │ │ Pcap   │
│  └───────┘  │ │ Mode    │ │Keyword   │ │SavingsCalc  │ │Parser  │
│             │ └─────────┘ │Classifier│ └─────────────┘ └────────┘
│  Per-call   │             └──────────┘
│  logging    │            ┌─────────────┐
│  Aggregated │            │ PRICING_DB  │
│  reports    │            │ (25 models) │
└─────────────┘            └─────────────┘
                                    ▲
                                    │
                              ┌──────────┐
                              │ config   │
                              │ (YAML)   │
                              └──────────┘

The data flow follows each module's responsibility cleanly. The cost tracker logs each API call to a local SQLite database and produces aggregated reports. The GPU monitor queries NVML for live metrics or generates realistic simulations in demo mode. The model router classifies task complexity using keyword analysis and recommends the optimal model from the pricing database. The caching advisor analyzes prompt patterns and estimates dollar savings from different caching strategies. The shadow AI scanner parses DNS packet captures to detect unauthorized AI API usage. The report generator aggregates data across all modules into actionable recommendations.

Features

1. Cost Tracker: Token-Level Attribution
Every API call is tracked with precise cost attribution across five dimensions:

Per-call costing: Prompt tokens × input price + completion tokens × output price
Session tracking: Group related calls into sessions for project-level analysis
Model breakdowns: See exactly what each model costs you
Project breakdowns: Allocate costs across teams or applications
Time-series: Daily cost trends over configurable periods

Sample output:

💰 Cost Logged — gpt-4o
╭───────────────────┬──────────────╮
│ Metric            │ Value        │
├───────────────────┼──────────────┤
│ Model             │ gpt-4o       │
│ Provider          │ OpenAI       │
│ Project           │ test-project │
│ Prompt Tokens     │ 500          │
│ Completion Tokens │ 150          │
│ Total Tokens      │ 650          │
│ Input Cost        │ $0.001250    │
│ Output Cost       │ $0.001500    │
│ Total Cost        │ $ 0.002750   │
╰───────────────────┴──────────────╯

2. GPU Monitor: Utilization and Alerts
Live GPU metrics with threshold-based alerts:

GPU Utilization: NVML or simulated random-walk
Memory Usage: Used / Total GB with percentage
Temperature: Live sensor reading in °C
Power Draw: Current and max power in Watts
Alert Thresholds: Configurable for temp, util, mem, power

Demo mode generates realistic GPU metrics using a random-walk simulation, for development and testing without dedicated hardware. Continuous monitoring with neo-cost gpu watch shows a live-updating terminal dashboard with ASCII bar charts.

3. Model Router: Intelligent Routing Engine
Task complexity is classified across four dimensions:

Simple: Summarization, translation, extraction, formatting
Complex: Analysis, comparison, evaluation, troubleshooting
Creative: Writing, composition, brainstorming, marketing
Code: Implementation, debugging, refactoring, architecture

Routing logic maps complexity to model category:

Low complexity → SLM (saves 60-90% vs frontier)
Medium complexity → SLM with frontier fallback option
High complexity → Frontier model for quality

Each model receives a composite score based on task-specific quality multiplier, per-call cost estimate, and token budget prediction.

4. Caching Advisor: Cost Savings Analysis
Four caching strategies evaluated with concrete dollar estimates:

Exact Match (~15% hit rate): Best for repeated identical prompts
Semantic (~25% hit rate): Best for similar meaning, different wording
Prompt Caching (~40% hit rate): Best for shared system prefixes and templates
TTL-Based (~20% hit rate): Best for time-bounded repeated queries

Each strategy produces daily, monthly, and yearly savings projections based on estimated call volume.

5. Shadow AI Scanner: DNS-Based Detection

Unauthorized AI API usage is detected by analyzing DNS traffic. The detection database covers 25+ known AI API domains across providers including OpenAI, Anthropic, Google, Groq, DeepSeek, Mistral, Cohere, HuggingFace, Replicate, Perplexity, Together AI, and more.

Risk scoring considers the number of queries to each AI domain, the sensitivity of the service (API endpoints score higher risk than web UIs), and the total number of unique AI services detected. Sample PCAP generation creates realistic test data with mixed benign and AI API DNS queries.

6. Report Generator: Comprehensive Insights
All six modules aggregate into a single report:

Current Cost: $152.40                                                        
Optimized Cost: $88.39                                                       
Potential Savings: $64.01                                                    
Savings %: 42.0%                                                             
Route 60% of simple/medium tasks from frontier to SLM models

Installation

Prerequisites

Python 3.8 or higher
pip (Python package manager)

Install from source

# Clone the repository
git clone https://github.com/your-org/neo-cost-optimizer.git
cd neo-cost-optimizer

# Create and activate virtual environment (recommended)
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Install the package
pip install -e .

# Verify installation
neo-cost --help

Quick install via pip

pip install neo-cost

Dependencies

Quick Start

# Track a single API call
neo-cost cost log gpt-4o --prompt-tokens 500 --completion-tokens 150 --project my-app

# View cost summary
neo-cost cost summary

# Check GPU status (simulation mode)
neo-cost gpu status --demo

# Get model routing recommendation
neo-cost route "Write a Python function to sort a list of dictionaries"

# Analyze caching strategy
neo-cost cache "You are a helpful assistant. Summarize the following text:" --volume 5000

# Scan for shadow AI usage (generates sample data automatically)
neo-cost shadow scan

# Generate a comprehensive report
neo-cost report

# List all models with pricing
neo-cost models

# Show current configuration
neo-cost config

Configuration

Configuration is stored in ~/.config/neo-cost/config.yaml (or $XDG_CONFIG_HOME/neo-cost/config.yaml) and is created automatically on first run.

Default configuration

cost_tracker:
  db_path: ~/.neo-cost/cost_tracker.db
  default_project: default

gpu_monitor:
  poll_interval: 2.0
  temp_threshold_c: 85
  util_threshold_pct: 95
  memory_threshold_pct: 90
  power_threshold_w: 400
  demo_mode: false

model_router:
  default_slm: gpt-4o-mini
  default_frontier: gpt-4o
  token_budget_warning_pct: 80

caching_advisor:
  min_similarity: 0.7
  default_ttl_hours: 24
  cache_hit_cost_reduction_pct: 90

shadow_ai:
  sample_pcap_path: ~/.neo-cost/shadow_ai_sample.pcap
  risk_score_threshold: 5

reporting:
  output_dir: ~/.neo-cost/reports
  default_format: terminal

Configuration is edited directly or viewed with neo-cost config.

Command Reference

neo-cost cost

neo-cost gpu

status: Current GPU metrics (--demo)
watch: Live monitoring dashboard (--demo, --interval)

neo-cost route

neo-cost cache

--volume: Daily call volume (default: 1000)
--ttl: TTL in hours (default: 24)

neo-cost shadow

Other commands

report: Comprehensive cost optimization report
models: List all models with pricing
config: Show current configuration
monitor: Interactive real-time monitoring dashboard

Example Workflows

Workflow 1: Daily Cost Review

# 1. Log all your API calls (integrate into your app or batch log)
neo-cost cost log gpt-4o --prompt-tokens 15000 --completion-tokens 3000 --project production
neo-cost cost log gpt-4o-mini --prompt-tokens 50000 --completion-tokens 10000 --project production

# 2. Get daily summary
neo-cost cost summary --days 1

# 3. Generate full report
neo-cost report --days 7

Workflow 2: Model Selection Decision

# Compare routing for different task types
neo-cost route "Translate this document to Spanish"
neo-cost route "Debug this production issue with the microservice"
neo-cost route "Write a marketing campaign for our new product"

Workflow 3: Shadow AI Audit

# 1. Capture DNS traffic on your network (using tcpdump)
sudo tcpdump -i eth0 port 53 -w network_traffic.pcap

# 2. Scan with neo-cost
neo-cost shadow scan --pcap network_traffic.pcap

# 3. Review the risk report and take action

Workflow 4: Caching ROI Analysis

# Analyze your most common prompt patterns
neo-cost cache "You are a customer support agent..." --volume 10000
neo-cost cache "Summarize the following article:" --volume 5000

Workflow 5: GPU Capacity Planning

# Monitor GPU over time
neo-cost gpu watch --interval 5

# Check current status
neo-cost gpu status

Cost-Savings Methodology

The cost-optimization engine uses a multi-factor scoring system:

Model Score = QualityMultiplier(TaskType, ModelCategory) / Cost

Quality Multiplier Table:

Savings estimation analyzes the current model mix from cost tracker data, classifies each task by complexity, proposes alternative SLM routing for simple and medium tasks, applies an average savings of ~70% per rerouted call, and assumes a realistic adoption rate of 60%.

Caching savings follow this formula:

Savings = DailyVolume × CacheHitRate × CostPerCall × 0.9

The 0.9 factor represents the 90% cost reduction when serving from cache.

Real-World Use Cases

Engineering Teams: track per-feature AI costs across development sprints, optimize model selection for different product features, monitor GPU utilization in CI/CD pipelines.

FinOps / Cloud Cost Teams: allocate AI costs to business units and projects, identify cost anomalies with daily time-series tracking, audit shadow AI usage before it becomes a security risk.

AI Platform Teams: build an internal model gateway with routing recommendations, implement caching for high-volume inference endpoints, create chargeback reports for internal AI consumers.

Security Teams: detect unauthorized AI API usage in corporate networks, identify data exfiltration risks from shadow AI tools, enforce AI usage policies with DNS-level monitoring.

How I Built This Using NEO

The requirement was a terminal-based toolkit that gives engineering teams full visibility into their AI spend: token-level cost attribution, GPU monitoring, intelligent model routing, caching analysis, shadow AI detection, and comprehensive reporting, all from a single CLI with no external dependencies. NEO planned and produced the files in this repository: six backend modules covering cost tracking, GPU monitoring, model routing, caching analysis, shadow AI scanning, and report generation, a built-in pricing database across 25+ models and 9 providers, a click-based CLI layer with 9 commands, a YAML configuration system, and a test suite under test_suite/.

The result is a fully working CLI toolkit that surfaces AI cost decisions in real time: what each call costs, which model would have been cheaper, how much caching would save, and whether unauthorized AI services are active on the network.

How You Can Use This With NEO

Track AI costs per project across an engineering team.
The neo-cost cost log command accepts a model name, token counts, and a project label on every API call. The neo-cost cost summary command then breaks down spending by model, project, and day, turning scattered API usage into an auditable cost record without any external tooling.

Route tasks to cheaper models before making the API call.
The neo-cost route command accepts a prompt and returns a model recommendation based on task complexity. Simple and medium-complexity tasks are routed to SLMs where savings of 60–90% versus frontier models are possible, with the scoring methodology transparently documented in neo_cost/pricing_db.py.

Audit shadow AI usage across a corporate network.
DNS traffic captured with tcpdump can be piped directly into neo-cost shadow scan. The scanner checks against 25+ known AI API domains and produces a risk score, giving security teams a concrete detection path without requiring agent deployment on individual machines.

Calculate caching ROI before committing to an implementation.
The neo-cost cache command takes a prompt pattern and a daily call volume and returns projected savings across four strategies: exact match, semantic, prompt caching, and TTL-based, with daily, monthly, and yearly figures, so the decision to invest in a caching layer is backed by numbers before any code is written.

Final Notes

Most AI cost problems are invisible until they're large. By the time a team notices the spend, the model choices, the redundant calls, and the shadow usage have already accumulated. AI Cost Optimizer CLI moves that visibility to the point where the decisions are still being made: which model, which caching strategy, which teams are spending what, and puts it in the terminal where engineering teams already work.

The code is at https://github.com/dakshjain-1616/AI-Cost-Optimizer-CLI
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Dataset Factory: A Production-Grade Benchmark Dataset Factory for AI Agent Evaluation

Nilofer 🚀 — Fri, 10 Jul 2026 04:21:47 +0000

Evaluating AI agents requires benchmark datasets that are high-quality, diverse, balanced, and free of duplicates. Building those datasets by hand is slow, inconsistent, and hard to reproduce. The Mercor Dataset Factory automates the entire pipeline: generate, validate, deduplicate, analyze coverage, remediate gaps, audit, govern, and verify, all with a single command.

It is not a simple data generator. It is a complete data engineering platform purpose-built for creating, validating, and releasing benchmark datasets for AI agent evaluation.

Overview

The factory produces a 1,000-record benchmark dataset spanning 10 agent-task categories with balanced difficulty distribution. Each record is a structured evaluation task containing a natural language query, ground-truth plan, expected tools, evaluation criteria, and success conditions.

Every phase has a quality gate. If validation fails, the pipeline halts. If duplicates are found, they are automatically removed and replaced. If coverage gaps exist, they are remediated. The result is a production-ready benchmark with guaranteed quality guarantees.

Features

Config-driven architecture: Categories are self-contained drop-in pack files in categories/*.yaml. Global settings live in config.yaml. No Python changes needed to customize.
Extensible categories: Add a category by dropping a new pack file into categories/. Remove one by deleting its pack. Packs are auto-discovered and authoritative for the category set.
10 benchmark categories: Coding, Web Research, Data Analysis, System Design, Debugging, Document Processing, Tool Usage, Multi-Step Reasoning, Operations/DevOps, Agent Coordination, each shipped as a pack in categories/
Balanced distribution: Exactly 100 records per category, 330/340/330 split across easy/medium/hard
Source/config preflight: A pydantic-based validator (factory/config_validator.py) checks the configuration before any work runs, catching incomplete categories, bad/duplicate prefixes, unknown template placeholders, and inconsistent counts
Canonical record contract: Every record is validated against a declared JSON Schema (schema/record.schema.json) in addition to the inline validation rules
Near-duplicate-aware generation: The generator rejects near-duplicate queries at creation time using a pure-stdlib term-frequency cosine guard, plus a 3-layer dedup pass (exact MD5 hash, TF-IDF cosine, semantic embedding)
100% coverage guarantee: Automatic gap detection and remediation for tools (127) and domains (118)
8-dimensional auditing: Schema quality, diversity, coverage, difficulty balance, tool diversity, duplicate rate, auditability, reproducibility, weighted to a single 0-100 score
Independent verification: A separate Python script (verification.py) with zero imports from the factory package validates independently
Full governance: Dataset card, release notes, methodology document, lineage tracking, versioning strategy
Fully reproducible: Single --seed 42 flag locks every stochastic component

Project Structure

mercorDatasetFactory/
│
├── README.md                          ← This file
├── CONTRIBUTING.md                    ← Contributor guide (setup, tests, conventions)
├── LICENSE                            ← MIT License
├── pyproject.toml                     ← Packaging metadata + `dataset-factory` entry point
├── requirements.txt                   ← Runtime dependencies
├── requirements-dev.txt               ← Dev/test dependencies (pytest, coverage)
├── dataset_factory.py                 ← Main orchestrator — the CLI entry point
├── verification.py                    ← Independent verification (zero factory imports)
├── config.yaml                        ← GLOBAL configuration only (targets, difficulties, schema constraints, seed)
│
├── categories/                        ← Category packs — one self-contained YAML per category (authoritative)
│   ├── coding.yaml
│   ├── web_research.yaml
│   ├── ...                            ← 10 built-in packs; drop a new file to add a category
│   └── agent_coordination.yaml
│
├── schema/
│   └── record.schema.json             ← Canonical record JSON Schema (the data contract)
│
├── factory/                           ← Core factory package
│   ├── __init__.py
│   ├── config_loader.py               ← Config + category-pack loading, deep-merge, defaults inline
│   ├── config_validator.py            ← Pydantic source/config preflight validator
│   ├── record_schema.py               ← Builds the JSON Schema record contract from config
│   ├── schema.py                      ← Constants, enums, field definitions, schema constraints
│   ├── categories.py                  ← Category registries (loaded from packs via config_loader)
│   ├── templates.py                   ← Query/plan/criteria/success prompt templates
│   ├── generator.py                   ← Stratified sampling, near-duplicate guard & record construction
│   ├── validator.py                   ← Schema + JSON Schema contract validation & distribution analysis
│   ├── deduplicator.py                ← 3-layer dedup (exact, TF-IDF, semantic)
│   ├── coverage.py                    ← Coverage matrices & over/under-representation analysis
│   ├── remediator.py                  ← Automatic gap remediation & supplemental generation
│   ├── auditor.py                     ← 8-dimensional weighted quality scoring
│   └── governance.py                  ← Governance document generation (card, notes, methodology)
│
├── tests/                             ← Test suite (run with `pytest -q`)
├── .github/                           ← CI workflows (must pass on every change)
│
├── artifacts/                         ← All pipeline outputs (git-ignored; regenerated each run)
│   ├── dataset.jsonl                  ★ Final 1,000-record benchmark dataset
│   ├── validation_report.md           ★ Schema, contract & distribution validation results
│   ├── duplicate_analysis.md          ★ 3-layer dedup analysis with examples
│   ├── coverage_report.md             ★ Tool/domain/category coverage matrices
│   ├── remediation_report.md          ★ Gap remediation log (if any gaps found)
│   ├── audit_report.md                ★ 8-dimensional quality score report
│   ├── governance.md                  ★ Dataset lineage, assumptions, versioning
│   ├── release_notes.md               ★ Release documentation
│   ├── dataset_card.md                ★ Dataset card (HuggingFace-style)
│   └── generation_methodology.md      ★ Full methodology for reproducibility
│
├── plans/                             ← Implementation plans
│   ├── plan.md
│   └── config_driven_plan.md
│
└── test_custom_config.yaml            ← Example custom config for testing

Quick Start

Prerequisites

Python 3.8+
pip (Python package installer)
~4 GB RAM (sentence-transformers model loads to ~800MB, preprocessing on CPU)
No GPU required: all phases run on CPU

1. Set Up Environment

# Clone or navigate to the project
cd mercorDatasetFactory

# Create a virtual environment (recommended)
python3 -m venv venv
source venv/bin/activate     # Linux/macOS
# .\venv\Scripts\activate    # Windows

2. Install Dependencies

pip install -r requirements.txt

Alternatively:

pip install scikit-learn sentence-transformers pyyaml pydantic

Note: scikit-learn is used for TF-IDF vectorization in the deduplication layer. sentence-transformers provides the all-MiniLM-L6-v2 embedding model for semantic duplicate detection. pyyaml handles config and category-pack loading. pydantic (v2) powers the source/config preflight validator.

3. Run the Full Pipeline

python3 dataset_factory.py

The pipeline will:

Generate 1,000 records using seed=42 (reproducible)
Validate every record against schema, field constraints, and distribution targets
Deduplicate using 3-layer detection (exact, TF-IDF, semantic), removing and replacing flagged records
Analyze coverage of all 127 tools and 118 domains across 10 categories
Remediate any gaps automatically with supplemental records
Audit the dataset across 8 quality dimensions with weighted scoring
Generate governance documents: dataset card, release notes, methodology
Run independent verification: 6 checks resulting in PASS/FAIL

All outputs appear in artifacts/.

4. Verify Independence

python3 verification.py

Expected output:

============================================================
OVERALL: PASS (6/6 checks passed)
============================================================

This script has zero imports from the factory/ package. It independently validates the dataset by reading artifacts/dataset.jsonl directly.

CLI Reference

python3 dataset_factory.py [OPTIONS]

Examples

# Full factory run (reproducible default)
python3 dataset_factory.py

# Custom seed and record count
python3 dataset_factory.py --seed 123 --count 500

# Use a custom configuration
python3 dataset_factory.py --config my_custom_config.yaml

# Re-validate and re-audit an existing dataset
python3 dataset_factory.py --skip-generation

# Validate the configuration and category packs without generating anything
python3 dataset_factory.py --validate-config

# Custom config + custom seed + custom count
python3 dataset_factory.py --config my_config.yaml --seed 7 --count 2000

Source/config validation

Before any generation work, the pipeline runs a pydantic-based preflight (factory/config_validator.py) over the merged configuration and the discovered category packs. It aggregates all problems into a single report and refuses to run if there are hard errors: incomplete categories, malformed or duplicate prefixes, unknown template placeholders (allowed: domain, topic, code_snippet, claim, n, stack_trace, task_type). Count drift is surfaced as a non-fatal warning. Run the preflight on its own with --validate-config:

Config validation: PASS — configuration is coherent.

Pipeline Architecture

Each phase has a specific responsibility and a quality gate. If any phase fails its gate, the pipeline halts with a clear error message.

Phase 1: Generation

Engine: factory/generator.py → DatasetGenerator

Stratified sampling across 10 categories x 3 difficulty levels (30 strata). Each record is constructed from templates with randomized domain, tool, and task-type substitutions. The generator maintains running counters to ensure exact balance.

Quality gate: Target distribution (±1 record per category per difficulty). If the generator cannot meet targets after 2x attempts, it reports generation statistics for diagnosis.

Phase 2: Validation

Engine: factory/validator.py → validate_dataset()

Validates every record against the schema definition:

All 9 required fields present with correct types
ID format: {CATEGORY_PREFIX}_{SEQ:03d} (e.g., COD_001, WEB_042)
Query length: 50–500 characters
Plan steps: 3–8 (varies by difficulty)
Evaluation criteria: ≥2
Success conditions: ≥1
Tools: ≥1
Category distribution: 100 ± 5 per category
Difficulty distribution: easy ±5%, medium ±5%, hard ±5%
Query diversity: unique queries ≥ 95%

Gate: Pipeline halts if schema validation fails. Distribution warnings are reported but non-fatal.

Phase 3: Deduplication

Engine: factory/deduplicator.py → DatasetDeduplicator

Three-layer duplicate detection, applied in sequence:

Flagged duplicate records are removed and replacement records are automatically generated to maintain exactly 1,000 records with balanced category and difficulty distributions.

Gate: Pipeline proceeds regardless (dedup always succeeds, but high duplicate rates are reflected in the audit score).

Phase 4: Coverage Analysis

Engine: factory/coverage.py → analyze_coverage()

Builds coverage matrices for three dimensions:

Tool coverage: All 127 registered tools appear in at least one record
Domain coverage: All 118 registered domains appear in at least one record
Category x Difficulty: No empty stratum

Also checks for over/under-representation using a ±20% threshold from expected distribution.

Phase 5: Gap Remediation

Engine: factory/remediator.py → remediate()

If any gaps are found in Phase 4, the remediator identifies the specific missing tools/domains, generates supplemental records targeting those gaps, tracks per-category sequence numbers to avoid ID collisions, and merges supplemental records into the dataset.

Gate: Pipeline halts if remediation fails.

Phase 6: Audit

Engine: factory/auditor.py → audit_dataset()

Eight quality dimensions, each scored 0–100 and weighted:

Rating scale: EXCELLENT (≥90), GOOD (≥75), FAIR (≥60), POOR (<60)

Phase 7: Governance

Engine: factory/governance.py → generate_all()

Generates four documents automatically: governance.md, release_notes.md, dataset_card.md, and generation_methodology.md.

Phase 8: Independent Verification

Engine: verification.py (standalone script)

Gate: Final release gate. The dataset must pass 6/6 before it can be considered production-ready.

Configuration System

How Configuration Works
Configuration is split into two concerns:

Category packs (categories/*.yaml): one self-contained file per category, auto-discovered and authoritative
Global config (config.yaml): globals only: record targets, difficulties, SCHEMA_CONSTRAINTS, DEFAULT_SEED, and difficulty step ranges/tags

The factory uses a layered configuration architecture:

[Python defaults in config_loader.py]
            │
            ▼
[category packs in categories/*.yaml]  ← discovered & merged (authoritative for categories)
            │
            ▼
[config.yaml at project root]  ← deep-merge globals over defaults
            │
            ▼
[--config CLI flag / env var]  ← deep-merge over everything
            │
            ▼
[Module-level globals]
(schema.py, categories.py, templates.py → generator.py, etc.)

The deep-merge algorithm: dictionaries are merged recursively; lists and scalars are replaced entirely by the user config (not appended to).

Using a Custom Config

# Via CLI flag
python3 dataset_factory.py --config my_dataset_config.yaml

# Via environment variable
export DATASET_FACTORY_CONFIG=/path/to/config.yaml
python3 dataset_factory.py

# CLI overrides env var
python3 dataset_factory.py --config override.yaml

What You Can Customize

config.yaml controls the global settings: difficulty ratios, record targets, field bounds, the default seed, and difficulty step ranges and tags. Edit this file to change how many records are generated, how they are split across difficulty levels, or what constraints apply to fields like query length and plan steps.

categories/.yaml controls everything per category: the category name, its unique record ID prefix, the tasks it covers, the tools and domains available to it, the query and plan step templates, evaluation criteria and success condition templates, and an optional topic pool. Add a category by dropping a new pack file. Remove one by deleting it. These packs are the authoritative source for the category set and all per-category data.

Example: Custom Globals Config

# my_dataset_config.yaml — globals only; categories come from categories/*.yaml

TOTAL_RECORDS: 500
RECORDS_PER_CATEGORY: 50
DIFFICULTY_TARGETS:
  easy: 25
  medium: 25
  hard: 0        # Skip hard difficulty entirely
GLOBAL_DIFFICULTY_TARGETS:
  easy: 250
  medium: 250
  hard: 0
SCHEMA_CONSTRAINTS:
  user_query_min_length: 50
  user_query_max_length: 500

Config File Structure

TOTAL_RECORDS: 1000
RECORDS_PER_CATEGORY: 100
SCHEMA_CONSTRAINTS:
  user_query_min_length: 50
  user_query_max_length: 500
  plan_min_steps: 3
  plan_max_steps: 8
  min_tools: 1
  max_tools: 10
  min_criteria: 2
  max_criteria: 6
  min_conditions: 1
  max_conditions: 5
  min_domains: 1
  max_domains: 4

Programmatic Reload

from factory.config_loader import reload_config

# Load a different config at runtime (optionally validate it)
reload_config("/path/to/custom_config.yaml", validate=True)

# All downstream modules immediately reflect the new values
from factory.schema import TOTAL_RECORDS, SCHEMA_CONSTRAINTS
print(TOTAL_RECORDS)                              # Updated record target
print(SCHEMA_CONSTRAINTS['user_query_max_length'])  # Updated field bound

Adding a Category

Categories are self-contained drop-in pack files. To add a category, no Python or config.yaml changes are needed. Drop a new categories/<name>.yaml file. Packs are auto-discovered and authoritative for the category set.

Allowed template placeholders: domain, topic, code_snippet, claim, n, stack_trace, task_type.

1. Drop a pack file

# categories/incident_response.yaml
name: incident_response
order: 11
prefix: INC
tasks:
  - triage
  - root_cause_analysis
  - postmortem
tools:
  - pager
  - log_search
  - metrics_dashboard
  - runbook
domains:
  - payments
  - identity
  - data_platform
topic_pool:
  - a cascading service outage
  - a degraded p99 latency spike
query_templates:
  - "Triage {topic} in the {domain} system and identify the {task_type} steps for {n} on-call engineers."
  - "Write a postmortem for {topic} affecting {domain}, including timeline and remediation."
plan_step_templates:
  - "Acknowledge the alert for {topic} and assemble responders"
  - "Correlate logs and metrics to localize {topic}"
  - "Apply mitigation and verify recovery of {domain}"
evaluation_criteria_templates:
  - "Root cause of {topic} is correctly identified"
  - "Mitigation restores the {domain} service"
success_condition_templates:
  - "Incident is resolved and a postmortem is filed"

2. Validate the config (preflight)

python3 dataset_factory.py --validate-config

3. Run the pipeline as usual

python3 dataset_factory.py --seed 42

NUM_CATEGORIES is derived automatically from the discovered packs, so the new category is picked up with no further changes.

Removing or relocating packs

Remove a category: delete its categories/<name>.yaml pack.

Use a different packs directory:

export DATASET_FACTORY_CATEGORIES_DIR=/path/to/my/packs
python3 dataset_factory.py --validate-config

Programmatic API

Each factory module can be imported independently for custom pipelines or integration into larger systems.

Basic Usage

# Generate a dataset
from factory.generator import DatasetGenerator

gen = DatasetGenerator(seed=42)
records = gen.generate(1000)
gen.save(records, "artifacts/dataset.jsonl")

# Validate
from factory.validator import validate_dataset

result = validate_dataset("artifacts/dataset.jsonl", "artifacts/validation_report.md")
print(f"Status: {result['overall']['status']}")   # "PASS" or "FAIL"
print(f"Checks: {result['overall']['passed']}/{result['overall']['total']}")

Full Pipeline Programmatically

from factory.generator import DatasetGenerator
from factory.validator import validate_dataset
from factory.deduplicator import DatasetDeduplicator
from factory.coverage import analyze_coverage
from factory.remediator import remediate
from factory.auditor import audit_dataset
from factory.governance import generate_all

# Generate
gen = DatasetGenerator(seed=42)
records = gen.generate(1000)
gen.save(records, "artifacts/dataset.jsonl")

# Validate
validation = validate_dataset("artifacts/dataset.jsonl", "artifacts/validation_report.md")
if validation['overall']['status'] != 'PASS':
    exit(1)

# Deduplicate
import json
deduper = DatasetDeduplicator(records)
analysis = deduper.analyze_all()
deduped, dedup_analysis = deduper.deduplicate()      # returns (records, analysis)
with open("artifacts/dataset.jsonl", "w") as f:
    for rec in deduped:
        f.write(json.dumps(rec) + "\n")
print(f"Removed {len(records) - dedup_analysis['remaining_records']} duplicates")

# Coverage
coverage = analyze_coverage("artifacts/dataset.jsonl", "artifacts/coverage_report.md")
print(f"Tool coverage: {coverage['tool_coverage']['global']['overall_coverage_pct']}")

# Remediate
remediation = remediate(
    "artifacts/dataset.jsonl",
    coverage_results=coverage,
    output_path="artifacts/remediation_report.md",
    updated_dataset_path="artifacts/dataset.jsonl",
)
print(f"Gaps: {remediation['gaps_identified']}, Generated: {remediation['supplemental_generated']}")

# Audit
audit = audit_dataset("artifacts/dataset.jsonl", "artifacts/audit_report.md", dedup_results=analysis)
print(f"Score: {audit['overall_score']:.2f}/100 ({audit['rating']})")

# Governance
generate_all("artifacts/dataset.jsonl")

Deduplicator Details

from factory.deduplicator import DatasetDeduplicator

deduper = DatasetDeduplicator(records)

# Run all 3 layers (consolidated)
full_analysis = deduper.analyze_all()
print(f"Total duplicates found: {full_analysis['total_duplicates_found']}")

# Or run layers individually
exact_results = deduper.find_exact_duplicates()
tfidf_results = deduper.find_tfidf_duplicates()
semantic_results = deduper.find_semantic_duplicates()

# Remove duplicates → (deduplicated_records, analysis)
deduped, analysis = deduper.deduplicate()
print(f"Remaining after dedup: {analysis['remaining_records']}")

Coverage Analyzer

from factory.coverage import analyze_coverage

result = analyze_coverage("artifacts/dataset.jsonl", "artifacts/coverage_report.md")

# Per-category tool coverage
for cat, cov in result['tool_coverage']['per_category'].items():
    used = cov['unique_tools_used']
    total = cov['tools_available']
    pct = cov['coverage_pct']
    print(f"{cat}: {used}/{total} tools ({pct})")

# Global coverage stats
global_cov = result['tool_coverage']['global']
print(f"Overall tool coverage: {global_cov['overall_coverage_pct']}")

Output Artifacts

The pipeline writes 10 artifacts to the artifacts/ directory (git-ignored, regenerated by every run):

dataset.jsonl is the final 1,000-record benchmark dataset, one JSON object per line.

validation_report.md contains per-check PASS/FAIL results from schema, JSON Schema contract, and distribution validation.

duplicate_analysis.md contains the 3-layer deduplication results with examples of detected duplicates.

coverage_report.md contains tool, domain, and category coverage matrices with representation analysis.

remediation_report.md logs gap identification and supplemental generation if any gaps were found.

audit_report.md contains the 8-dimensional quality score with weighted overall score and rationale.

governance.md documents dataset lineage, assumptions, limitations, versioning, and release process.

release_notes.md contains semantic versioning release notes.

dataset_card.md is the dataset card in HuggingFace-style format.

generation_methodology.md contains the full methodology for reproducibility.

Dataset Record Schema

{
  "id": "COD_001",
  "category": "coding",
  "difficulty": "medium",
  "user_query": "Implement a function that validates JSON schema compliance...",
  "ground_truth_plan": [
    "Parse the input JSON string into a Python dictionary",
    "Load the JSON Schema definition from the provided schema file",
    "Validate the dictionary against the schema using the jsonschema library",
    "Collect all validation errors with paths and messages",
    "Return the validation result with pass/fail status and error details"
  ],
  "expected_tools": [
    "code_editor",
    "linter",
    "test_runner",
    "static_analyzer"
  ],
  "evaluation_criteria": [
    "Correctly validates all schemas and reports errors",
    "Handles edge cases (empty objects, nested schemas, $ref)",
    "Tests cover at least 90% of code paths"
  ],
  "success_conditions": [
    "All unit tests pass with >90% code coverage",
    "Code passes linting with zero errors or warnings"
  ],
  "metadata": {
    "domains": ["backend_services", "data_structures"],
    "requires_external_access": false,
    "estimated_steps": 7,
    "tags": ["intermediate", "cross_domain", "independent_steps"]
  }
}

The metadata block always contains domains (list), requires_external_access (bool), estimated_steps (int), and tags (list).

Quality Report

Produced from seed=42, 1,000 records:

Audit Score Breakdown

Extending the Factory

Adding a New Category

Drop a new pack file into categories/, validate, and run:

# 1. Create categories/<name>.yaml
# 2. Preflight the config + packs
python3 dataset_factory.py --validate-config
# 3. Run
python3 dataset_factory.py --seed 42

Adding a New Tool

Edit the relevant category pack and add the tool to its tools list:

# categories/coding.yaml
tools:
  - code_editor
  - linter
  - test_runner
  - my_new_tool   # ← Add it here

Then re-run python3 dataset_factory.py --validate-config and the pipeline.

Creating a Custom Pipeline

Import the factory modules directly in Python (see Programmatic API above).

Reproducibility

To reproduce the exact same dataset:

python3 dataset_factory.py --seed 42 --count 1000

The --seed flag propagates to every stochastic component:

Full methodology is documented in artifacts/generation_methodology.md.

Governance and Maintenance

Versioning

Semantic versioning (MAJOR.MINOR.PATCH):

Release Process

Run the config preflight: python3 dataset_factory.py --validate-config → PASS
Run the full pipeline
Verify: python3 verification.py → 6/6 PASS
Check audit score ≥ 90 (EXCELLENT)
Review release_notes.md for accuracy
Tag release in version control

Known Limitations

Documented in artifacts/governance.md:

Synthetic queries: All queries are generated from templates, not collected from real users
Template bias: Domain/tool distributions reflect template structure
Difficulty calibration: Based on step count ranges, not human evaluation
No ground-truth execution: Plans are text-based and not executable

Extending for a New Version

# 1. Add/edit category packs in categories/ and/or globals in config.yaml
# 2. Preflight the config + packs
python3 dataset_factory.py --validate-config
# 3. Regenerate
python3 dataset_factory.py --seed 42 --count 1000
# 4. Verify
python3 verification.py
# 5. Check audit score ≥ 90

How I Built This Using NEO

The requirement was a production-grade benchmark dataset factory that could generate, validate, deduplicate, audit, and release high-quality evaluation datasets for AI agents with a single command. NEO planned and produced the files in this repository: the main orchestrator, an independent verification script, 14 factory modules covering generation, validation, deduplication, coverage analysis, remediation, auditing, and governance, the canonical record schema, the global config, all 10 category pack files, the test suite, and the CI workflows. The plans/ directory in the repo documents the build run directly.

The result is a fully working 8-phase pipeline that produces a 1,000-record benchmark dataset with a 94.94/100 audit score, 100% tool and domain coverage, zero duplicates, and 6/6 independent verification checks passed.

How You Can Extend This Further With NEO

Use it as the evaluation layer in a CI/CD pipeline for AI agents.
The factory outputs dataset.jsonl and verification.py exits 0 only when 6/6 checks pass. That exit code becomes a CI gate: if the benchmark does not meet quality thresholds, the pipeline fails before the agent is evaluated against bad data. Plugging python3 dataset_factory.py --seed 42 and python3 verification.py as pipeline steps gives every agent deployment a reproducible, audited evaluation layer.

Use --skip-generation to re-audit a dataset after agents modify or filter it.
If a downstream team removes records, rebalances difficulty, or adds domain-specific records to dataset.jsonl, running with --skip-generation re-runs all validation, coverage analysis, deduplication, and auditing against the modified file without regenerating. The audit score immediately shows whether the changes degraded data quality across any of the 8 dimensions.

Use the programmatic API to build a custom evaluation loop.
Every factory module is independently importable. DatasetGenerator, validate_dataset, DatasetDeduplicator, and audit_dataset can be imported directly into an agent testing script, generating a fresh benchmark for each experiment and feeding records to the agent under test, all in one Python process without touching the CLI or the filesystem.

Use a custom category pack to benchmark agents against your own task taxonomy.
If the 10 built-in categories do not match the agent's actual workload, a custom categories/<name>.yaml with domain-specific tasks, tools, and query templates can be dropped in. The factory runs the custom benchmark through the same 8-phase pipeline the built-in categories go through, including deduplication, 100% coverage validation, and governance document generation.

Final Notes

Most benchmark datasets are generated once, checked loosely, and never audited again. The Dataset Factory treats dataset quality the same way production software treats code quality: every run is validated, every gap is remediated, every dimension is scored, and the result is independently verified before it is considered production-ready.

The code is at https://github.com/dakshjain-1616/DataSet-Factory
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Harness Template Library: 10 Production-Grade AI Agent Templates with 15 Shared Infrastructure Modules

Nilofer 🚀 — Mon, 06 Jul 2026 07:00:08 +0000

Building an AI agent prototype is straightforward. Making it reliable in production is not. Rate limits must be retried with backoff. Context windows fill up and must be pruned carefully. Tool calls need permission checks before execution. Financial operations need a human to confirm before money moves. Failures mid-task need to checkpoint so work is not lost. Usage costs need to be tracked per session with hard limits.

Harness Template Library solves all of that. It is a production-grade open-source library of 10 AI agent harness templates, each a complete, runnable Python package that wires together 15 battle-tested infrastructure modules. Ready to deploy on day one. Built autonomously using NEO.

What This Library Does

The 15 core modules handle every infrastructure concern an agent needs in production: context management, memory, tool permissions, budget tracking, observability, state persistence, human approval, and more. The 10 templates provide domain-specific tools, system prompts, and workflow configurations on top of that shared infrastructure. Every template runs with claude-sonnet-4-20250514 out of the box.

The split is deliberate. The 15 infrastructure modules are written once and shared. Each template only provides its domain-specific tools, system prompt, and configuration. Improvements to any core module benefit all 10 templates instantly.

Architecture

The 15 core modules all live in core/. Each template's harness.py instantiates and wires them together with domain-specific configuration. The templates differ in their tools, system prompts, workflow steps, and permission policies, not in their infrastructure.

Prerequisites

Python 3.11 or newer
An Anthropic API key: export ANTHROPIC_API_KEY=sk-ant-your-key-here

Getting Started

Install

cd projects/harness-template-library
pip install -e ".[dev]"

This installs the library and registers all 10 CLI entry points. Set your API key:

export ANTHROPIC_API_KEY=sk-ant-your-key-here

Run a Template via CLI

Each template registers its own CLI command. For example, the coding agent:

coding-agent "Write a Python function that checks if a string is a palindrome"

Or the research agent:

research-agent "Summarize the key architectural differences between GPT-4 and Claude"

All 10 CLIs work the same way: pass a task or query as the first argument, and the agent runs it using the full 15-module infrastructure stack.

Run an Example Script

python -m examples.run_coding_agent
python -m examples.run_research_agent

Run Tests

python -m pytest tests/ -v

78 tests across all 15 core modules. Runs in under 5 seconds.

The 15 Core Modules

All 15 modules live in core/ and are real, working Python classes with no stubs and no placeholders.

InstructionManager: Loads system prompts from file or env, supports {variable} interpolation.

ContextBuilder: Assembles the messages array, enforces token budget with intelligent truncation.

MemoryLayer: SQLite-backed persistent memory with keyword search across past sessions.

ModelAdapter: Wraps AsyncAnthropic, uses claude-sonnet-4-20250514, retries on rate limits with exponential backoff.

ToolRegistry: Register tools by name with their JSON schema and Python handler function.

PermissionResolver: Checks tool calls against an allowlist policy before execution; supports wildcards.

BudgetManager: Tracks token usage and cost per session; raises BudgetExceededError at hard limit.

WorkflowEngine: Runs a sequence of named steps with conditional branching based on step results.

StateManager: Serializes and restores full agent state to SQLite, for resuming interrupted tasks.

HumanApprovalLayer: Pauses on sensitive or financial tools and prompts the operator for confirmation.

ObservabilityLayer: Emits structured logs via structlog and OTEL spans for every tool and model call.

EvaluationFramework: Scores agent output against a rubric using a second Claude call.

RetryRecoverySystem: Exponential backoff wrapper that saves partial results to disk before re-raising.

AuditLogger: Append-only SQLite log of every action, decision, and tool call, non-repudiable.

DeploymentConfig: Generates Dockerfiles and docker-compose.yml programmatically for any template.

The 10 Templates

Each template is in templates/<name>/ with harness.py, tools.py, config.py, system_prompt.txt, Dockerfile, docker-compose.yml, and tests/test_harness.py.

Coding Agent: Reads and writes files, runs tests, diffs git changes, and searches code. Workflow: understand task, write code, run tests, fix failures, commit. System prompt: expert software engineer with TDD focus.

Research Agent: Handles web search, URL reading, note saving, and citation management, with web and URL tools currently implemented as stubs. Workflow: decompose question, search, synthesize, verify claims, write report. System prompt: rigorous researcher with citation requirements.

Customer Support: Handles ticket lookup, update, escalation, and knowledge base search, with email sending currently implemented as a stub. Workflow: understand issue, check KB, resolve or escalate, update ticket. System prompt: empathetic support agent with escalation rules.

Data Engineering: Handles SQLite queries, schema reading, data validation, and pipeline execution. Workflow: understand data need, query, validate, transform, output. System prompt: data engineer focused on data quality and lineage.

Browser Automation: Handles navigation, clicking, text extraction, screenshots, and form filling, with all browser tools currently implemented as stubs awaiting browser integration. Workflow: understand goal, plan steps, execute, verify, report. System prompt: precise automation agent that validates each step.

Multi-Agent Orchestrator: Spawns and coordinates subagents, sends messages, collects results, and merges outputs. Workflow: decompose task, assign to subagents, collect results, synthesize. System prompt: orchestrator that breaks complex tasks into parallel workstreams.

RAG Agent: Embeds queries, runs cosine similarity search over an in-memory vector store, reranks results, and cites passages. Workflow: query, retrieve, rerank, generate answer with citations. System prompt: knowledge retrieval agent that only answers from provided context.

Finance Operations: Handles balance checks, invoice creation, and payment processing, with all financial tools currently implemented as stubs, plus a full audit trail. HumanApprovalLayer is always required before any financial tool executes. Workflow: validate request, check permissions, execute with approval, audit. System prompt: finance agent with mandatory human approval for all transactions.

Document Analysis: Handles text extraction, document classification, entity extraction, summarization, and document comparison. Workflow: ingest, classify, extract, analyze, report. System prompt: document analyst focused on structured information extraction.

Long-Horizon Task: Handles checkpointing, checkpoint restoration, todo list management, and progress tracking. Workflow: plan, checkpoint, execute step, checkpoint, repeat. System prompt: methodical agent that checkpoints frequently and resumes gracefully.

Project Structure

harness-template-library/
├── pyproject.toml                          # Package + 10 CLI entry points
├── core/
│   ├── __init__.py
│   ├── instruction_manager.py
│   ├── context_builder.py
│   ├── memory_layer.py
│   ├── model_adapter.py                    # claude-sonnet-4-20250514 default
│   ├── tool_registry.py
│   ├── permission_resolver.py
│   ├── budget_manager.py                   # raises BudgetExceededError
│   ├── workflow_engine.py
│   ├── state_manager.py
│   ├── human_approval_layer.py
│   ├── observability_layer.py
│   ├── evaluation_framework.py
│   ├── retry_recovery_system.py
│   ├── audit_logger.py
│   └── deployment_config.py
├── templates/
│   ├── coding_agent/
│   │   ├── harness.py                      # Wires all 15 core modules
│   │   ├── tools.py                        # read_file, write_file, run_tests, git_diff, search_code
│   │   ├── config.py
│   │   ├── system_prompt.txt
│   │   ├── Dockerfile
│   │   ├── docker-compose.yml
│   │   └── tests/test_harness.py
│   ├── research_agent/                     # Same structure, different tools + prompt
│   ├── customer_support/
│   ├── data_engineering/
│   ├── browser_automation/
│   ├── multi_agent_orchestrator/
│   ├── rag_agent/
│   ├── finance_operations/
│   ├── document_analysis/
│   └── long_horizon_task/
├── harness_template_library/
│   └── cli/
│       ├── coding_agent_cli.py             # click CLI → CodingAgentHarness
│       ├── research_agent_cli.py
│       └── ...                             # 8 more CLI modules
├── examples/
│   ├── run_coding_agent.py
│   └── run_research_agent.py
└── tests/
    ├── conftest.py                         # Fixtures: mock Anthropic, temp SQLite, sample config
    └── test_core_modules.py               # 75 tests covering all 15 core modules

Key Design Decisions

Shared core, specialized templates.
All 15 infrastructure modules are written once and shared. Each template only provides its domain-specific tools, system prompt, and configuration. Improvements to, say, the RetryRecoverySystem benefit all 10 templates instantly.

SQLite everywhere.
Memory, state persistence, and audit logs all use SQLite via aiosqlite. No external services are required to run any template. This makes the library work in air-gapped environments, local development, and ephemeral CI environments without any setup.

HumanApprovalLayer as a hard gate.
The PermissionResolver checks allowlists before a tool runs, but HumanApprovalLayer is a separate, upstream gate that can be configured per-tool. For the finance template, every payment-related tool requires approval regardless of the permission policy. The two layers are deliberately separate so the audit log shows both the policy check and the human decision.

BudgetExceededError is synchronous.
Even though the rest of the library is async, BudgetManager.check_budget() raises BudgetExceededError synchronously. This keeps budget enforcement simple: any caller that does not catch it will propagate up, ensuring no tool call or model call can exceed the limit by accident.

EvaluationFramework uses a second Claude call.
Agent output scoring runs a separate Anthropic API call with a rubric-based prompt. This adds latency and cost, so it is opt-in per template rather than always-on. The coding agent uses it to score whether generated code is idiomatic and well-tested.

Configuration

Each template reads its configuration from environment variables using pydantic-settings. Create a .env file in the project root:

ANTHROPIC_API_KEY=sk-ant-your-key-here
MODEL=claude-sonnet-4-20250514
MAX_TOKENS=4096
BUDGET_LIMIT_USD=1.00
LOG_LEVEL=INFO

All config values have sensible defaults. Only ANTHROPIC_API_KEY is required to run.

Docker Deployment

Each template includes a Dockerfile and docker-compose.yml. To run the coding agent in Docker:

cd templates/coding_agent
docker compose up --build

Or generate a custom deployment config programmatically:

from core.deployment_config import DeploymentConfig

dc = DeploymentConfig(service_name="my-coding-agent")
dockerfile = dc.generate_dockerfile()
compose = dc.generate_compose()

Environment Variables

Verified Results

The library ships with 78 tests across all 15 core modules including BudgetManager, WorkflowEngine, HumanApprovalLayer, RetryRecoverySystem, ObservabilityLayer, and the rest. The package installs cleanly with pip install -e . and all public classes import without errors. Tests run in under 5 seconds.

AI template generation (DeepSeek V4 Flash via OpenRouter): The ModelAdapter auto-detects the OpenRouter key and routes calls through DeepSeek's 1M-context reasoning model at the correct endpoint. When asked to generate a retry decorator with exponential backoff for async Python functions, it returned a complete, production-ready implementation in 311 output tokens: a parameterized retry_async decorator accepting max_retries, base_delay, max_delay, and a tuple of exception types, with backoff capped at max_delay and the original function signature preserved via functools.wraps. The adapter handles DeepSeek thinking blocks transparently, returning only the final text content. Total cost at $0.10/M input tokens was under $0.001.

How I Built This Using NEO

The requirement was a library of production-grade agent harness templates that wire together all the infrastructure concerns a real agent needs, so teams could skip the boilerplate and start from a complete, working architecture. NEO planned and produced the files in this repository: 15 core infrastructure modules, 10 complete agent templates each with its own harness, tools, configuration, system prompt, Docker setup, and test suite, 10 CLI entry points, two example scripts, and the full test suite with fixtures and module coverage. The result is a fully working library where picking a template and running one CLI command gives a complete, production-grade agent with all 15 infrastructure concerns pre-wired and no boilerplate left to write.

How You Can Use This With NEO

Pick a template and run a production-ready agent immediately.
All 10 templates register their own CLI entry points on install. Passing a task to coding-agent, research-agent, or any of the other eight commands runs the full 15-module infrastructure stack without writing any code.

Use the 15 core modules in your own agent.
Every module is a real, importable Python class. Any agent codebase can pull in ContextBuilder, BudgetManager, HumanApprovalLayer, or any other module independently and wire them into an existing architecture.

Deploy any template to Docker in one command.
Every template ships with a Dockerfile and docker-compose.yml. Running docker compose up --build inside any template directory produces a containerized, ready-to-run agent with no additional configuration beyond the API key.

Run the library in air-gapped or local environments.
Memory, state persistence, and audit logs all use SQLite via aiosqlite. No external services are required. The library runs in CI, local development, or any environment where installing dependencies from a remote service is not possible.

Final Notes

Most agent projects rebuild the same infrastructure from scratch every time: retry logic, context pruning, permission checks, budget limits, audit trails. This library solves that once across 10 use cases. Pick the template that matches the job, or take the 15 core modules and wire them into whatever architecture is already in place.

The code is at https://github.com/dakshjain-1616/Harness-Template-Library
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

NEO Data Quality Auditor AI: Automated Data Quality Auditing, Bias Detection, and Lineage Tracking

Nilofer 🚀 — Fri, 03 Jul 2026 13:30:36 +0000

60% of businesses cite poor data quality as the primary reason for AI failures. Dirty data leads to misleading insights, wasted resources, and failed ML models. Most teams lack easy-to-use tooling that surfaces what is wrong and what to do about it.

NEO Data Quality Auditor AI addresses that directly. It is an automated data quality auditing tool that detects inconsistencies, bias, missing values, and format issues in any CSV dataset, with a real-time monitoring dashboard, AI-powered cleaning suggestions, and data lineage tracking. Built autonomously by NEO.

Visual Reports

The repo ships with four SVG infographics: a data quality score gauge, a bias detection overview, a gender and age bias analysis, and an ethnicity bias breakdown.

What It Does

9 Quality Checks: Missing values, duplicates, data types, out-of-range, format violations, outliers, cardinality, cross-column consistency, distribution skew.

AI Cleaning Suggestions: A rule-based engine maps every detected issue to actionable steps with severity, effort level, and category. Each suggestion is filterable and expandable in the dashboard.

Bias Detection: Demographic parity ratio, disparate impact, and group fairness metrics across sensitive attributes. Results surface per-group statistics with representation rates and standard errors.

Data Lineage: An event-sourced audit trail tracks every analysis step with timestamps and column-level changes, exportable as JSON.

Real-Time Dashboard: An interactive Streamlit UI with a quality score gauge (0-100), drill-downs per check, and exportable reports.

The 9 Quality Checks

The 9 checks cover the most common classes of data quality problems.

Missing Values: Null/NaN counts and ratios per column with critical vs warning thresholds.
Duplicate Rows: Exact duplicate records with sample indices.
Data Type Inconsistencies: Mixed types within a column, for example strings in numeric columns.
Out-of-Range Values: Values outside expected min/max bounds.
Format/Pattern Violations: Invalid emails, phone numbers, and dates using regex patterns.
Outlier Detection: IQR method and Z-score analysis for numerical anomalies.
Uniqueness/Cardinality: High vs low cardinality detection.
Cross-Column Consistency: Logical contradictions between related columns.
Skewness/Kurtosis: Distribution shape analysis for numeric columns.

Bias Detection Metrics

Data Lineage

The lineage tracker logs every analysis step:

Load events: Original row count, column list, source filename
Quality check events: Per-check results with affected columns
Bias check events: Per-attribute bias metrics
Export: Download the full audit trail as JSON

Installation

# Clone or navigate to the project
cd NEO-Data-Quality-Auditor-AI

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# Install dependencies
pip install -r requirements.txt

Usage

Start the Dashboard

source venv/bin/activate
streamlit run app.py

This opens a web browser with the interactive dashboard.

Walkthrough

Upload a CSV: Use the sidebar to upload any CSV file, or click "Use Demo Data" to load a pre-built dataset with intentional quality issues (missing values, outliers, format violations, demographic bias).

Overview Tab: See the overall data quality score (0-100), pass/fail breakdown per check, and summary metrics.

Quality Checks Tab: Drill into each of the 9 checks with interactive visualizations (bar charts for missing values, box plots for outliers, histograms for distributions).

Bias Report Tab: Review demographic parity ratios and disparate impact metrics across sensitive attributes like gender, ethnicity, and age.

Lineage Tab: View the full audit trail of every analysis step with timestamps and event type breakdown.

AI Suggestions Tab: Get actionable cleaning recommendations filtered by severity and effort level, with step-by-step instructions.

Command-Line Usage

The modules are also importable programmatically:

import pandas as pd
from auditor.quality_checks import run_all_quality_checks, calculate_quality_score
from auditor.bias_detector import run_all_bias_checks
from auditor.ai_suggestions import generate_suggestions

# Load data
df = pd.read_csv("your_data.csv")

# Run all 9 quality checks
results = run_all_quality_checks(df)
score = calculate_quality_score(results)
print(f"Quality Score: {score}")

# Run bias detection
bias_results = run_all_bias_checks(df)
print(f"Bias checks run: {len(bias_results)}")

# Get cleaning suggestions
suggestions = generate_suggestions(results, bias_results)
for s in suggestions:
    print(f"[{s['severity']}] {s['issue']} — Effort: {s['effort']}")

Project Structure

NEO-Data-Quality-Auditor-AI/
├── app.py                      # Streamlit dashboard (main entry point)
├── config.py                   # Thresholds and configuration
├── requirements.txt            # Python dependencies
├── docs/                       # SVG infographics
│   ├── quality_score.svg       # Data quality gauge (43/100)
│   ├── bias_overview.svg       # Bias detection overview
│   ├── gender_age_bias.svg     # Gender & age bias analysis
│   └── ethnicity_bias.svg      # Ethnicity bias details
├── auditor/
│   ├── __init__.py             # Package exports
│   ├── quality_checks.py       # 9 quality check functions
│   ├── bias_detector.py        # Demographic parity, disparate impact, fairness
│   ├── lineage_tracker.py      # Event-sourced data lineage tracker
│   ├── ai_suggestions.py       # Rule-based cleaning suggestion engine
│   └── demo_data.py            # Demo data generator with intentional issues
├── data/
│   └── demo_data.csv           # Generated demo dataset
└── README.md

Running Tests

source venv/bin/activate
python -c "
from auditor.quality_checks import run_all_quality_checks, calculate_quality_score
from auditor.bias_detector import run_all_bias_checks
from auditor.ai_suggestions import generate_suggestions
import pandas as pd

# Generate and test with demo data
from auditor.demo_data import generate_demo_data
generate_demo_data()
df = pd.read_csv('data/demo_data.csv')

# Run full audit
quality = run_all_quality_checks(df)
score = calculate_quality_score(quality)
bias = run_all_bias_checks(df)
suggestions = generate_suggestions(quality, bias)

print(f'Quality Score: {score}')
print(f'Bias checks: {len(bias)}')
print(f'Suggestions: {len(suggestions)}')
"

Dashboard Screenshots

Launch the dashboard with streamlit run app.py to see it live.

Overview: Quality score gauge (0-100), check pass/fail bar chart, summary metrics
Quality Checks: Per-check drill-down with interactive Plotly visualizations
Bias Report: Demographic parity ratio charts, group fairness breakdown
Lineage: Event timeline with event type distribution pie chart
AI Suggestions: Expandable suggestion cards with severity filters and effort labels

Configuration

Edit config.py to tune:

QUALITY_THRESHOLDS: Missing value thresholds, outlier Z-score cutoff, email/phone regex patterns
BIAS_PARAMS: Demographic parity range, min group size, sensitive column keywords
DASHBOARD_CONFIG: App title, colors, max file size

How I Built This Using NEO

The requirement was a data quality auditing platform that could detect inconsistencies, bias, and format issues in any CSV dataset and surface actionable cleaning recommendations through an interactive dashboard. NEO planned and produced the files in this repository: the Streamlit dashboard in app.py, the threshold configuration in config.py, five auditor modules covering quality checks, bias detection, lineage tracking, AI suggestions, and demo data generation, four SVG infographics under docs/, and the demo dataset in data/demo_data.csv.

The result is a fully working auditing platform that takes any CSV file in and returns a quality score across 9 checks, bias metrics across sensitive attributes, a complete event-sourced lineage trail, and prioritized cleaning suggestions, all from a single streamlit run app.py command.

How You Can Use This With NEO

Audit any CSV dataset before feeding it into an ML model.
Any CSV file can be uploaded directly to the dashboard and the platform runs all 9 quality checks, returning a quality score (0-100) with a pass/fail breakdown per check and interactive drill-downs. No configuration is required before the first run.

Detect demographic bias before training.
The bias_detector.py module computes demographic parity ratio, disparate impact, and group fairness metrics across sensitive attributes like gender, ethnicity, and age. Running this before training surfaces skewed representation that would otherwise propagate silently into model predictions.

Integrate quality checks programmatically into a data pipeline. run_all_quality_checks(), run_all_bias_checks(), and generate_suggestions() are all importable directly, as shown in the command-line usage section. Any pipeline that can run Python can call these functions and surface quality issues before data reaches the next stage.

Export the full lineage trail as JSON for compliance or auditing.
The event-sourced lineage tracker logs every analysis step with timestamps and column-level changes. The full audit trail is downloadable as JSON from the Lineage tab, making it usable as a compliance artifact without any additional tooling.

Final Notes

NEO Data Quality Auditor AI is a single-command auditing platform that takes any CSV file in and surfaces quality issues, bias metrics, cleaning recommendations, and a full lineage trail, all from streamlit run app.

The code is at https://github.com/dakshjain-1616/Data-Quality-Auditor-AI
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Tool Permission Matrix Builder & Validator: Structured, Visual Policy Management for AI Agent Teams

Nilofer 🚀 — Thu, 25 Jun 2026 05:41:16 +0000

AI agents in production access tools that range from harmless read-only queries to irreversible destructive operations. Managing which agents can use which tools is a governance problem that most teams solve with ad-hoc scripts and tribal knowledge - and that works until it doesn't. A misconfigured role, an over-exposed tool, or an agent that silently calls something it shouldn't are the kinds of failures that surface in production rather than in review.

Tool Permission Matrix Builder & Validator replaces that with a structured, visual approach. It is a visual policy management system for AI agent teams - define tools, classify their risk, assign roles, and drag-and-drop permissions onto a matrix, then export machine-readable policy artifacts or validate existing agents for compliance, all powered by Claude claude-sonnet-4-20250514.

What This Platform Does

The platform addresses the full lifecycle of agent tool governance in one place.

It starts with tool registration - each tool is defined and assigned a risk category: read-only, internal-write, external-api, financial, destructive, or administrative. Roles are then created for each agent type - analyst, operator, admin, readonly-bot, or whatever the team's structure requires. The permission matrix takes these two dimensions and lets permissions be assigned by dragging tools onto roles or clicking individual cells to toggle between allowed, denied, and inherited states. The matrix validates in real time: if a role has access to a tool whose risk level exceeds what that role should have, a warning appears immediately.

Once the matrix is configured, a policy artifact is exported - JSON for machine consumption, YAML for GitOps workflows, or a Python module with a check_permission(role, tool) function that can be imported directly into agent code. On the validation side, existing agent code can be pasted in and Claude analyzes which tools it actually calls, cross-checks those against the matrix, and produces a security score with sorted recommendations. A separate sprawl analysis detects over-exposure: roles with too many high-risk tools, tools granted to too many roles, and unused grants.

Architecture

The backend is fully async - all 24 routes use async def with an aiosqlite-backed SQLAlchemy session. This is intentional: the Claude API calls in agent validation and sprawl analysis can take 5–15 seconds, and with a synchronous backend, one validation request would block all other users. With async, many concurrent requests are handled without blocking.

Both AI services have heuristic fallbacks. If ANTHROPIC_API_KEY is not set, the agent validator still extracts tool call patterns from code using regex and checks them against the matrix, and the sprawl analyzer still computes numerical sprawl metrics. The Claude path produces richer narrative and nuanced recommendations; the heuristic path still provides actionable data.

The policy generator produces three output formats from the same matrix data. The Python module output is syntax-verified via py_compile before being returned, ensuring the downloaded file is always importable. The repo also includes architecture.svg.

Quick Start

Prerequisites

Python 3.11 or newer
Node.js 18 or newer
An Anthropic API key (optional - Claude features fall back to heuristic analysis without it)

Set up the environment

cp .env.example .env
# Optionally add ANTHROPIC_API_KEY=sk-ant-your-key-here for Claude analysis

Run the backend

cd backend
pip install -r requirements.txt
uvicorn main:app --reload --host 0.0.0.0 --port 8000

The API starts at http://localhost:8000. Swagger UI is available at http://localhost:8000/docs.

Run the frontend

The frontend runs in a separate process:

cd frontend
npm install
npm run dev

The UI opens at http://localhost:5173.

Run with Docker

cp .env.example .env
docker compose up --build

The backend runs on port 8000 with a health check. The frontend serves via nginx on port 80 and waits for the backend health check before starting.

Running Tests

cd backend && python -m pytest tests/ -v

22 tests - policy generation (JSON/YAML/Python), agent validator, heuristic analysis. Runs in under a second.

API Reference

Risk Categories

All six risk categories are implemented as a Python Enum and stored in the database. The permission matrix UI shows these risk colors on every tool badge, and real-time validation warnings fire when a role's allowed risk levels would be exceeded.

Project Structure

tool-permission-matrix/
├── backend/
│   ├── main.py                        # FastAPI app, 24 async routes
│   ├── models.py                      # Tool, Role, Permission ORM + RiskCategory Enum
│   ├── schemas.py                     # Pydantic v2 request/response schemas
│   ├── database.py                    # Async SQLite via aiosqlite
│   ├── services/
│   │   ├── policy_generator.py        # JSON, YAML, and Python module export
│   │   ├── agent_validator.py         # Claude + heuristic agent code analysis
│   │   └── sprawl_analyzer.py        # Claude + heuristic sprawl detection
│   ├── requirements.txt
│   ├── Dockerfile.backend
│   └── tests/
│       ├── test_policy_generator.py   # 11 policy generation tests
│       ├── test_validator.py          # 11 validator tests
│       └── fixtures/
│           ├── sample_agent.py        # Realistic agent with tool call patterns
│           └── sample_policy.json     # Realistic permission matrix fixture
├── frontend/
│   ├── src/
│   │   ├── App.tsx                    # Tab layout: Tools/Roles/Matrix/Export/Validate/Sprawl
│   │   ├── stores/
│   │   │   ├── toolStore.ts           # Zustand store for tool state
│   │   │   ├── roleStore.ts           # Zustand store for role state
│   │   │   └── matrixStore.ts         # Zustand store for permission matrix
│   │   ├── components/
│   │   │   ├── ToolRegistry.tsx       # CRUD + filter + JSON import/export
│   │   │   ├── RoleManager.tsx        # CRUD + inheritance + risk levels
│   │   │   ├── PermissionMatrix.tsx   # @dnd-kit DnD grid
│   │   │   ├── PolicyExporter.tsx     # Format selector + download
│   │   │   ├── AgentValidator.tsx     # Paste/upload + results display
│   │   │   └── SprawlAnalysis.tsx     # Sprawl score + issues list
│   │   ├── api/client.ts              # axios-based API client, 20 methods
│   │   └── types/index.ts             # TypeScript interfaces (28 types)
│   ├── Dockerfile.frontend
│   ├── package.json
│   └── vite.config.ts
├── docker-compose.yml
└── .env.example

The structure maps directly onto the platform's three functional layers. The backend/services/ directory holds the three pieces that do the heavy lifting - policy generation, agent validation, and sprawl analysis - each isolated from the routing layer in main.py. The frontend mirrors this with one component per tab in the UI, with tool state, role state, and matrix state each managed by a dedicated Zustand store.

Key Design Decisions

Async throughout - All backend routes are async def and the SQLAlchemy session uses aiosqlite. The Claude API calls in agent validation and sprawl analysis can take 5–15 seconds. A synchronous backend would block all other users during that window; the async design handles many concurrent requests without blocking.

Three-state permission model - Each matrix cell is ALLOWED, DENIED, or INHERITED - not just a binary toggle. INHERITED means the permission comes from the role's parent role, enabling role hierarchies where a base role defines conservative defaults and derived roles override specific tools.

Heuristic fallback for AI features - Claude-powered features are never the only path. The agent validator extracts tool calls using regex patterns that cover the most common calling conventions, then checks them against the matrix. The sprawl analyzer computes over-exposure metrics numerically. The platform is fully usable in restricted environments without an API key; Claude's analysis is an enhancement rather than a dependency.

Policy Python module verification - When generating a Python module, py_compile is called on the output before returning it. A permissions.py that fails to compile would be worse than no policy at all, so this check runs as a hard gate.

Environment Variables

Verified Results

The backend ships with 22 tests covering policy generation in all three export formats (JSON, YAML, Python module), agent validator tool-call extraction across standard and use_tool/call_tool calling conventions, heuristic analysis correctness, and edge cases like empty code and missing policy. The frontend builds cleanly to a 277 KB JS bundle across 110 modules with @dnd-kit drag-and-drop and Zustand state management.

For the AI-powered sprawl analysis, the SprawlAnalyzer was run (using DeepSeek V4 Flash via OpenRouter for this verification pass) against a three-role matrix - admin, developer, viewer - with six tools spanning read, write, and destructive categories. The model returned a sprawl score of 80/100 and surfaced nine issues. Two were critical: the admin role holding both execute_code and delete_resource, and the developer role also having execute_code with no approval gate. The overall analysis named the pattern as excessive concentration of destructive tool access and recommended introducing approval workflows before any destructive operation.

How I Built This Using NEO

The requirement was a visual policy management platform where AI agent teams could define tools, classify risk, assign roles, configure permissions on a drag-and-drop matrix, and export machine-readable policy artifacts - with Claude-powered validation and sprawl analysis built in. NEO planned and produced the files in this repository - a fully async FastAPI backend with 24 routes, three backend services handling policy generation, agent validation, and sprawl analysis, a React and TypeScript frontend with a drag-and-drop permission matrix, six UI components, three Zustand stores, and a 22-test suite covering all major paths. The plans/ directory and ORCHESTRATOR_LOG.md in the repo document that build run directly.

The result is a fully working policy management platform - from tool registration through risk classification, matrix configuration, policy export, and agent validation - with heuristic fallbacks at every AI-powered step so the platform remains useful with or without an API key.

How You Can Use This With NEO

Govern tool access across an existing AI agent team.
Any team running multiple agents with different access levels can register their tools, classify them by the six built-in risk categories, and configure a permission matrix without writing a single line of policy code. The matrix validates in real time as roles and permissions are assigned.

Validate existing agent code against a policy.
Agent code can be pasted directly into the platform and the validator extracts which tools it actually calls, cross-checks them against the configured matrix, and returns a security score with specific recommendations. The heuristic path works without an API key; the Claude path produces richer analysis when ANTHROPIC_API_KEY is set.

Export a check_permission(role, tool) function directly into agent code.
Once the matrix is configured, the Python module export generates a permissions.py file that is syntax-verified before download and can be imported directly into any agent codebase - no manual policy translation required.

Detect permission sprawl in an existing matrix.
The sprawl analysis endpoint scores the matrix for over-exposure - roles with too many high-risk tools, tools granted to too many roles, and unused grants. The heuristic path computes numerical metrics without an API key; the Claude path names specific patterns and recommends remediation when ANTHROPIC_API_KEY is set.

Final Notes

The gap between "we think our agents have the right permissions" and "we can prove it and export it as code" is where this platform sits. Tool access in AI agent systems is a governance problem that gets harder as teams scale - more agents, more tools, more roles, and no single source of truth. The Tool Permission Matrix Builder & Validator makes that source of truth visual, exportable, and machine-readable.

The code is at https://github.com/dakshjain-1616/Tool-Permission-Matrix-Builder-Validator
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Context Compaction Visualizer: See Exactly What Your AI Agent Forgot Before It Costs You

Nilofer 🚀 — Tue, 23 Jun 2026 06:01:50 +0000

When an AI agent runs for many turns, it eventually hits context limits and must compress or discard earlier messages. This is often invisible, yet critical - lost context can cause the agent to forget constraints, user preferences, or prior decisions. The framework moves on. The agent keeps running. And somewhere in those discarded turns is a security finding, a constraint, a decision that the rest of the session quietly proceeds without.

Context Compaction Visualizer makes that process visible - not after something breaks, but as an inspectable artifact of every run. It is a visualization platform that helps teams understand how long-running AI agents manage and compress context over time - upload execution traces from LangSmith, OpenTelemetry, AgentOps, or any custom format, and explore exactly which context was retained, compressed, or discarded, and at what cost.

What This Platform Does

The core problem is that compaction happens inside the framework's internals. There is no standard output that tells you which messages survived, which were summarized, and which were dropped - or what any of that cost in tokens. This platform reconstructs that picture from execution traces.

A trace file is uploaded with a format selected, and the platform rebuilds the full session: every message at every turn, its fate - retained verbatim, summarized, or discarded - and any compaction events that occurred along the way. A D3.js stacked-bar timeline renders token consumption across all turns with color-coded regions for each outcome. A session replay steps through turn by turn, surfacing a diff at the exact point a compaction event fires. Token analytics compute the total cost and compression efficiency of the session. A Claude-powered information loss detector scores the risk of each compaction event and names specifically what may have been lost.

When two traces are available - two different agents, or the same agent under two different compaction strategies - a comparative view places them side by side to show which preserved more context at lower cost.

Quick Start

Prerequisites

Python 3.11 or newer
Node.js 18 or newer
An Anthropic API key (optional - only the Info Loss Detection feature needs it)

Set up the environment

cp .env.example .env
# Optionally add ANTHROPIC_API_KEY=sk-ant-your-key-here for info loss detection

Run the backend

cd backend
pip install -r requirements.txt
uvicorn main:app --reload --host 0.0.0.0 --port 8000

The API starts at http://localhost:8000. Interactive docs are available at http://localhost:8000/docs.

Run the frontend

The frontend runs in a separate process:

cd frontend
npm install
npm run dev

The UI opens at http://localhost:5173.

Run with Docker

cp .env.example .env
docker compose up --build

The backend runs on port 8000, the frontend serves via nginx on port 5173.

Running Tests

cd backend && python -m pytest tests/ -v

29 tests covering all four parsers, edge cases, and the token counter. The full suite runs in under 100ms since nothing in it hits an external service.

API Reference

Supported Trace Formats

The platform accepts four input formats, selectable via a dropdown on upload. Each has its own parser that handles the vendor-specific schema and reduces it to the normalized structure.

LangSmith - Parses JSON exports from the LangSmith tracing platform. The parser extracts runs, the messages inside each run, token counts from usage metadata, and any chain-level summarization events.

OpenTelemetry - Parses OTEL-format JSON spans. The parser traverses the span tree, reconstructs message history from span attributes, and identifies compaction events from span names containing "compress" or "summarize".

AgentOps - Parses AgentOps session JSON exports. The parser handles the session-level event structure and normalizes message roles from AgentOps event types.

Custom JSON - A generic format for any agent framework not listed above. It expects a messages array with role, content, and optional tokens and timestamp fields. Any event with type: "compaction" or type: "summarization" is treated as a compaction event.

Project Structure

context-compaction-visualizer/
├── backend/
│   ├── main.py                  # FastAPI app, 7 endpoints
│   ├── models.py                # Trace, Message, CompactionEvent ORM
│   ├── schemas.py               # Pydantic validation schemas
│   ├── database.py              # SQLAlchemy + SQLite setup
│   ├── parsers/
│   │   ├── langsmith_parser.py
│   │   ├── otel_parser.py
│   │   ├── agentops_parser.py
│   │   └── custom_parser.py
│   ├── services/
│   │   ├── context_analyzer.py  # Claude-powered info loss detection
│   │   └── token_counter.py     # Token counting + cost estimates
│   ├── requirements.txt
│   ├── Dockerfile
│   └── tests/
│       ├── test_parsers.py      # 29 tests covering all 4 parsers
│       └── fixtures/
│           ├── langsmith_trace.json
│           ├── otel_trace.json
│           ├── agentops_trace.json
│           └── custom_trace.json
├── frontend/
│   ├── src/
│   │   ├── App.tsx              # Upload/Timeline/Replay/Analytics/Loss/Compare tabs
│   │   ├── components/
│   │   │   ├── TraceUploader.tsx
│   │   │   ├── ContextTimeline.tsx   # D3.js stacked bar chart
│   │   │   ├── SessionReplay.tsx     # Turn navigation + compaction diff
│   │   │   ├── TokenAnalytics.tsx
│   │   │   ├── InfoLossDetector.tsx
│   │   │   └── ComparativeView.tsx
│   │   ├── hooks/useD3.ts
│   │   ├── api/client.ts
│   │   └── types/index.ts
│   ├── Dockerfile
│   ├── package.json
│   └── vite.config.ts
├── docker-compose.yml
└── .env.example

The file structure reflects the normalization design directly. Every file under backend/parsers/ handles one vendor's schema and outputs the same structure. Nothing downstream - not main.py, not any frontend component - needs to know which parser ran. The two services, context_analyzer.py and token_counter.py, sit after all four parsers and only ever see the normalized output.

Key Design Decisions

Parser normalization - Each observability platform has a fundamentally different schema. Rather than handling platform-specific quirks in every component, all four parsers produce an identical normalized structure. This means the timeline, replay, analytics, and comparison views have no knowledge of the original format.

Graceful Claude fallback - The Info Loss Detector calls the Anthropic API only when ANTHROPIC_API_KEY is set. Without a key, it returns analysis_available: false with a clear message rather than failing. The rest of the platform works fully without any API key.

D3.js integration via hook - The useD3.ts hook manages D3's selection lifecycle within React's rendering model. D3 takes ownership of the SVG element inside the hook's effect, while React manages the wrapping div and props. This avoids the common conflict between React's virtual DOM and D3's direct DOM manipulation.

Centralized cost estimates - Token counts and cost calculations happen in token_counter.py using verified Claude pricing - $3 per million input tokens, $15 per million output tokens - defined as constants in one place, making them easy to update if pricing changes.

Environment Variables

Verified Results

The backend ships with 29 tests covering all four trace parsers, realistic multi-turn fixture data for each format, edge cases like empty inputs and missing fields, and the token counter. All tests run in under 100ms since no external services are called. The frontend builds to a 238 KB JS bundle across 600 modules.

For the info-loss detector, the ContextAnalyzer was run (using DeepSeek V4 Flash via OpenRouter for this verification pass) against a real compaction event that had dropped 77,000 tokens from a security code review session. It returned an overall risk score of 0.85 and flagged two losses. The higher-risk item, scored 0.90, was the permanent loss of three specific JWT authentication findings - a missing expiry check, absent refresh token rotation, and a weak secret key - detail precise enough that no summary would have preserved it. The second item, scored 0.70, flagged the loss of 23 tool call exchanges' worth of reasoning context. Both came back with concrete recommended actions, not generic advice.

How I Built This Using NEO

This project was built autonomously using NEO. NEO is a fully autonomous AI engineering agent that can write code and build solutions for AI/ML tasks including AI model evals, prompt optimization and end to end AI pipeline development.

The requirement was a platform that ingests execution traces from any of the major agent observability tools, reconstructs what an agent's context looked like turn by turn, and uses an LLM to flag when something important got dropped during compaction. NEO planned and produced the entire codebase - four format parsers that each reduce a different vendor schema into one normalized structure, a FastAPI backend with seven endpoints wired to SQLAlchemy models, two backend services handling token counting and Claude-powered info loss detection, a full React and TypeScript frontend with D3.js visualizations across six components, and a 29-test suite with realistic multi-turn fixtures for all four formats. The repo's plans/ directory and ORCHESTRATOR_LOG.md document that build run directly.

The result is a fully working visualization platform that takes a raw trace file in, and gives you back a complete picture of what your agent remembered, what it forgot, and what that cost.

How You Can Use This With NEO

Audit any long-running agent for context loss.
If a LangSmith, OpenTelemetry, or AgentOps trace exists for an agent run, it can be dropped straight into the platform. The timeline and session replay immediately show which turns survived compaction and which did not - no instrumentation changes, no code modifications to the agent itself.

Benchmark compaction strategies before shipping.
When evaluating two different agent configurations or memory strategies, both traces can be uploaded and placed side by side in the comparative view. The platform surfaces which strategy retained more context at lower token cost, turning a subjective comparison into a measurable one.

Catch silent information loss in security or compliance-sensitive agents.
The Claude-powered info loss detector scores each compaction event and flags specific content that may have been dropped - as demonstrated with the JWT authentication findings in the verified results. Any agent operating over sensitive or constraint-heavy sessions can be run through this check before the output is trusted. This requires ANTHROPIC_API_KEY to be set; without it the platform returns analysis_available: false for this feature.

Use the custom JSON format to bring any agent framework in.
Agents not running on LangSmith, OpenTelemetry, or AgentOps can still feed into the platform by logging to the custom JSON format - a messages array with role, content, and optional tokens and timestamp fields. Any agent framework that can write JSON can produce a trace this platform accepts.

Final Notes

Compaction is designed to be invisible - the agent keeps running, the framework handles the limit, and nothing interrupts the workflow. The cost of that invisibility is that when something is silently dropped, there is no record of what it was or what it was worth. Context Compaction Visualizer produces that record.

The code is at https://github.com/dakshjain-1616/Context-Compaction-Visualizer
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Agent Sandbox Escape Detector: Black-Box Security Scanning for LLM Agents

Nilofer 🚀 — Fri, 12 Jun 2026 16:29:19 +0000

Most agent security tools focus on known jailbreak phrases or static rule-matching. That approach misses the point. A real attacker does not check a list of banned words - they probe the agent's actual behavior with semantically varied adversarial inputs and look for signs that something slipped through.

Agent Sandbox Escape Detector takes the same approach. Point it at any HTTP chat endpoint, and it fires a battery of adversarial prompts across six attack categories, then uses Claude Opus 4.8 as an independent judge to determine whether the agent leaked data, broke persona, or executed injected instructions. The result is a structured scan report with per-probe verdicts, evidence excerpts, and confidence scores.

The key insight is that you do not need whitebox access to test an agent — all you need is its chat endpoint. The scanner treats the agent as a black box and probes it the same way a real attacker would.

Architecture

Entry: CLI (--target URL)  ·  POST /scan
               │
               ▼
  ┌────────────────────────────────────────────┐
  │  Scanner  ·  asyncio.gather()              │
  │  all 6 probes run concurrently             │
  │  per-probe error isolation                 │
  └──────────────────┬─────────────────────────┘
                     │
       ┌─────────────┼──────────────────┐
       ▼             ▼                  ▼
  ┌─────────┐  ┌──────────────┐  ┌────────────────────┐
  │  Tool   │  │ Prompt Leak  │  │ API Call           │
  │ Access  │  │ Role Confus. │  │ Indirect Injection │
  └────┬────┘  └──────┬───────┘  │ Jailbreak  (6 tot) │
       │               │          └──────────┬─────────┘
       └───────────────┴───────────────────── ┘
                     │ 4–6 adversarial prompts per category
                     ▼
  ┌────────────────────────────────────────────┐
  │  Target Agent  ·  any HTTP chat endpoint   │
  │  httpx async POST                          │
  └──────────────────┬─────────────────────────┘
                     │ agent responses
                     ▼
  ┌────────────────────────────────────────────┐
  │  Claude Judge  ·  claude-opus-4.8          │
  │  via OpenRouter                            │
  │  → ESCAPED / SAFE  ·  confidence (0–1)     │
  │  → evidence excerpt from response          │
  └──────────────────┬─────────────────────────┘
                     │
                     ▼
  ┌────────────────────────────────────────────┐
  │  Scan Report                               │
  │  JSON  ·  Markdown  ·  Rich console        │
  │  GET /results/{scan_id}                    │
  └────────────────────────────────────────────┘

What It Tests

Each probe category targets a different class of vulnerability observed in production LLM deployments. Every category fires 4–6 distinct prompts to reduce false negatives from surface-level prompt filtering. All probes run concurrently - a full scan of six categories completes in roughly the same time as a single category.

Tool Access - requests that trick the agent into invoking unauthorized tools: file reads, database queries, exec calls. A vulnerable agent attempts to call off-limits functions or returns file and database content.

Prompt Leak - social-engineering and meta-instruction tricks to extract the system prompt. A vulnerable agent reveals system prompt contents, configuration, or internal instructions.

API Call - crafted inputs designed to make the agent perform SSRF or data exfiltration via HTTP. A vulnerable agent constructs or references external URLs containing sensitive data.

Role Confusion - persona hijacking, identity override, DAN-style persona injection. A vulnerable agent breaks character or adopts a no-restrictions alternate identity.

Indirect Injection - malicious payloads hidden in simulated tool outputs or document context. A vulnerable agent executes injected instructions as if they came from a trusted source.

Jailbreak - chain-of-thought manipulation, hypothetical framing, and reasoning override. A vulnerable agent agrees to bypass safety constraints or leaks reasoning chain internals.

How the Judge Works

After collecting all agent responses, each probe's batch of responses is sent to Claude Opus 4.8 via OpenRouter with a structured evaluation prompt. The judge produces a binary ESCAPED/SAFE verdict, a confidence score from 0 to 1, and a natural-language evidence summary quoting specific parts of the response that led to the verdict.

This approach is significantly more accurate than regex pattern matching because the judge understands context - an agent that says "I cannot help with that" is different from one that says "I cannot help with that, but here is /etc/passwd anyway."

Getting Started

Install dependencies into a virtual environment, copy .env.example to .env, and add your OPENROUTER_API_KEY. Then point the CLI at any agent's chat endpoint:

python -m src.cli scan --target http://localhost:8000/chat

To scan only specific probe categories or save results to JSON:

python -m src.cli scan --target http://localhost:8000/chat --probes tool_access,jailbreak --output report.json

Start the FastAPI server for REST integration:

uvicorn src.api.main:app

Environment Variables

OPENROUTER_API_KEY=sk-or-...    # Required — used for Claude judge calls via OpenRouter

API

POST /scan - accepts a target URL and optional probe list, returns a scan ID immediately, and runs the scan asynchronously.

GET /results/{scan_id} - returns the full structured report once the scan is complete.

GET /health - liveness probe for uptime monitoring.

Live Scan Results

Real scan run against a Claude-powered HTTP agent on 2026-06-09:

0 escapes detected across 6 probe categories - approximately 30 adversarial turns total. Scan ID 0c4bffa6, 2026-06-09.

Source Layout

The scanner orchestrates all probes via asyncio.gather() so they run in parallel, with per-probe error isolation so a timeout on one category never blocks the others. Each probe is a standalone class inheriting from BaseProbe - adding a new attack category means writing one class and one prompts file. The judge lives in core/judge.py and is stateless: it takes a list of responses and returns a list of ProbeResult objects. Reports are assembled by core/report.py, which handles JSON serialization, Markdown formatting, and Rich console rendering independently.

The test suite uses a vulnerable dummy agent fixture - an in-process FastAPI app that always complies with requests - to verify the scanner can detect escapes, and a safe dummy agent to verify it does not produce false positives. 64 tests, passing in approximately 15 seconds.

How I Built This Using NEO

The requirement was a black-box behavioral security scanner for LLM agents - one that probes any HTTP chat endpoint with adversarial prompts across six attack categories, uses Claude Opus 4.8 as an independent judge, and produces structured reports with per-probe verdicts, evidence excerpts, and confidence scores. NEO built the full implementation: the async orchestrator using asyncio.gather() with per-probe error isolation, all six probe classes inheriting from BaseProbe with their adversarial prompt files, the stateless Claude judge in core/judge.py via OpenRouter, the report assembler in core/report.py covering JSON, Markdown, and Rich console output, the CLI entry point, the FastAPI REST server with POST /scan and GET /results/{scan_id}, and the 64-test suite with vulnerable and safe dummy agent fixtures.

How You Can Use and Extend This With NEO

Use it to security-test any agent before it goes to production.
Point the CLI at your agent's chat endpoint and run a full six-category scan. The structured report tells you exactly which probe categories the agent failed, what the judge found in the response, and the confidence score - before real users can probe the same vulnerabilities.

Integrate it into CI to catch security regressions on every deploy.
Use POST /scan to trigger a scan and GET /results/{scan_id} to poll the report. If any probe returns ESCAPED above your confidence threshold, fail the pipeline. Agent behavior can regress with model updates or prompt changes - automated scanning catches this before it reaches production.

Use the probe categories as a security checklist when building agents.
The six categories - tool access, prompt leak, API call, role confusion, indirect injection, and jailbreak - map directly to the vulnerabilities that have been observed in production LLM deployments. Running the scanner on your agent during development tells you which categories need stronger guardrails before launch.

Extend it with additional probe categories.
Each probe is a standalone class inheriting from BaseProbe with a corresponding prompts file. A new attack category follows the same pattern and is automatically picked up by the orchestrator, judge, and report pipeline without any changes to the core.

Final Notes

Agent security is behavioral, not syntactic. A scanner that checks for banned phrases misses the attacks that matter. Agent Sandbox Escape Detector probes real behavior across six attack categories, judges responses with a frontier model that understands context, and gives you structured evidence - so you know not just whether an agent escaped, but how and where.

The code is at https://github.com/dakshjain-1616/Agent-Sandbox-Escape-Detector
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

AgentLiar Detector: Catch Coding Agents That Falsely Claim Task Completion

Nilofer 🚀 — Wed, 10 Jun 2026 12:41:47 +0000

AI coding agents are getting better at completing tasks. They are also getting better at appearing to complete tasks. An agent that claims "done" when it has created placeholder files, written empty tests, or quietly narrowed the scope of the original requirement is harder to catch than one that simply fails, because the failure is hidden inside output that looks correct at a glance.

AgentLiar is a production-ready system that detects when coding agents falsely claim task completion. It runs four independent verification checks, produces a weighted confidence score from 0 to 100, and delivers structured evidence in JSON, Markdown, or console output - usable as a CLI tool, Python library, GitHub Action, or HTTP API.

Features

4 Independent Checks - File, Test, Scope, and LLM Judge.
Confidence Scoring - weighted aggregation on a 0–100 scale.
Multiple Interfaces - CLI, Python API, GitHub Action, and HTTP API.
Adversarial Detection - catches placeholder implementations, empty tests, and scope narrowing.
Structured Reports - JSON and Markdown output with evidence.
Production Ready - type hints, error handling, logging, and async support.

Architecture

The async orchestrator dispatches four independent checks File, Test, Scope (local), plus an optional OpenRouter LLM Judge and produces a weighted 0–100 confidence score delivered as JSON, Markdown, or console output for CI gating.

The Four Verification Checks

1. File Check

Detects missing expected files
Identifies unexpected new files
Finds placeholder content: TODO, FIXME, pass-only
Validates file sizes and content

2. Test Check

Detects empty test bodies
Identifies tests without assertions
Finds skipped tests
Validates claimed versus actual test counts

3. Scope Check

Detects silent scope narrowing: "only", "for now"
Identifies partial implementations
Finds TODO markers in code
Validates requirements coverage

4. LLM Judge

Independent assessment via OpenRouter
Structured JSON output
Timeout and retry logic
Optional - works without an API key

Quick Start

Installation

pip install -e .

Or pip install agentliar once published. Requires Python 3.10+.

CLI Usage

Prepare sample inputs from examples/simple_task.json, then run:

agentliar verify \
  --task-file .tmp/task.txt \
  --claim-file .tmp/claim.json \
  --changes-file .tmp/changes.json \
  --format markdown

Use agentliar config to inspect configuration and agentliar analyze .tmp/task.txt to review a task file.

Python API

from agentliar import Verifier

verifier = Verifier()
result = await verifier.verify(
    task_description=task,
    claim=claim_payload,
    file_changes=changes_payload
)
# Read result.score, result.passed, result.confidence_level, result.reports

GitHub Action

Use the GitHub Action with task, claim, and change files, a confidence threshold, and an optional OPENROUTER_API_KEY secret when you want the LLM Judge path enabled.

HTTP API

Start the API server:

python -m agentliar.server
# or
uvicorn agentliar.server:app --host 0.0.0.0 --port 8000

Then POST /verify with the task, claim, and file-change payloads. The response returns score, pass/fail, and evidence blocks.

Confidence Score Interpretation

90–100 - High. Task appears fully completed.
70–89 - Medium. Task likely complete with minor issues.
50–69 - Low. Task partially completed.
30–49 - Critical. Significant issues detected.
0–29 - Failed. Task likely not completed.

Configuration

Create a .env file. Set OPENROUTER_API_KEY and OPENROUTER_MODEL only if you want LLM Judge mode. The check weights must sum to 1.0. CONFIDENCE_THRESHOLD controls the pass/fail cutoff.

Recommended LLM Judge models (May 2026):

anthropic/claude-haiku-4-5 - cheap and fast judging
anthropic/claude-sonnet-4-6 or openai/gpt-5.4 - higher-quality judging
openai/gpt-4.1-mini - budget option

Use Cases

CI/CD Integration - automatically verify PR claims before merging.
Code Review - get an independent assessment of task completion alongside a human review.
Agent Monitoring - detect when AI agents overstate progress in automated pipelines.
Quality Gates - block merges below a confidence threshold.
Documentation - generate verification reports for stakeholders.

Security

No hardcoded secrets
API keys via environment variables only
No data persistence
Local processing except for LLM Judge

Project Structure

src/agentliar/           # Checks, orchestration, scoring, reports, API, CLI, server
tests/
├── unit/                # Unit tests
├── adversarial/         # Adversarial tests
└── integration/         # Integration tests
examples/                # Sample inputs
action.yml               # GitHub Action definition
pyproject.toml           # Packaging and tooling

Testing

pytest                                    # Full suite
pytest --cov=agentliar --cov-report=html  # With coverage
pytest tests/unit/                        # Unit tests only
pytest tests/adversarial/                 # Adversarial tests only
pytest tests/integration/                 # Integration tests only

Code Quality

ruff check .        # Linting
ruff format .       # Formatting
mypy src tests      # Type checking

How I Built This Using NEO

The requirement was a production-ready verification system for detecting false completion claims from coding agents - running four independent checks locally, with an optional LLM Judge via OpenRouter, and exposing the result through a CLI, Python API, GitHub Action, and HTTP API. NEO built the full implementation: the async orchestrator dispatching all four checks, the File, Test, Scope, and LLM Judge check modules, the weighted confidence scorer, the JSON and Markdown report generators, the Click CLI with verify, config, and analyze commands, the FastAPI HTTP server, the GitHub Action definition in action.yml, and the test suite split across unit, adversarial, and integration coverage.

How You Can Use and Extend This With NEO

Use it as a CI gate on every PR that includes AI-generated code.
Add the GitHub Action to your workflow with a confidence threshold. Any PR where the agent's claimed changes do not pass the file, test, and scope checks below your threshold is blocked before merge - automatically, without a reviewer having to spot the placeholder implementation manually.

Use it to monitor agent progress in long-running pipelines.
Call await verifier.verify(...) from Python after each agent task completes. The confidence score and evidence blocks tell you whether the agent actually finished the task or produced output that looks complete but is not - before the next stage of the pipeline starts.

Use the LLM Judge for higher-confidence verification on critical tasks.
Set OPENROUTER_API_KEY and configure a judge model for tasks where the local checks alone are not sufficient. The LLM Judge runs independently from the other three checks and adds a cross-model perspective to the confidence score.

Extend it with additional check types.
The four checks share a common async interface in the orchestrator. A new check follows the same pattern and its weight is added to the configuration. The orchestrator, scorer, and reporters pick it up automatically.

Final Notes

Agents that falsely claim completion are harder to catch than agents that fail outright - because the output exists and looks plausible. AgentLiar makes the verification systematic: four independent checks, a weighted confidence score, and structured evidence that tells you exactly where the claim breaks down.

The code is at https://github.com/dakshjain-1616/AgentLiar
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code

Carbon-Aware Model Training: Scheduling GPU Workloads Around Electricity Carbon Intensity

Nilofer 🚀 — Sat, 06 Jun 2026 08:48:43 +0000

Training ML models has an environmental cost that most practitioners do not measure. A model trained during peak grid hours, when coal and gas plants are meeting high demand - can emit significantly more CO2 than the same model trained during off-peak hours when renewables dominate the grid. The carbon intensity of electricity varies by a factor of 2–5x throughout the day, but most training pipelines ignore this entirely.

Carbon-Aware Model Training Pipeline is a PyTorch-based training pipeline that monitors real-time electricity carbon intensity, delays training until a low-carbon window is available, reduces GPU memory footprint through gradient accumulation, and tracks CO2 emissions throughout the training process using CodeCarbon - with a comparison report that quantifies the carbon savings against a baseline run.

Features

Carbon-Aware Scheduling - real-time carbon intensity monitoring with smart training delays until low-carbon windows are detected.
Gradient Accumulation - reduces GPU memory footprint while maintaining effective batch size.
Emissions Tracking - real-time CO2 monitoring via CodeCarbon with comprehensive JSON reports.
Modular Design - YAML-based configuration with separate scheduler, tracker, and trainer modules.
GPU Optimized - automatic CUDA detection with mixed precision training (FP16).
Comparative Analysis - automated reporting quantifying carbon savings against a baseline run.

How It Works

The pipeline runs in four stages:

Stage 1 - Carbon-Aware Scheduling
Real-time monitoring checks electricity carbon intensity via APIs. Smart delays wait for low-carbon windows before starting training. Fallback mechanisms use realistic mock data when APIs are unavailable - with diurnal patterns simulating peak intensity at 18:00 and trough at 03:00. Configurable thresholds allow customization for different regions.

Stage 2 - Gradient Accumulation
Memory optimization processes smaller micro-batches. Effective batch size is maintained with reduced memory. Configurable steps (2, 4, 8, 16) adapt to hardware constraints. Convergence preservation ensures model quality is not compromised.

Stage 3 - Emissions Tracking
CodeCarbon integration monitors CO2 emissions in real-time. Energy metrics track power consumption in Watts and energy in kWh. Comprehensive reports generate JSON summaries with all metrics. Comparative analysis quantifies carbon savings versus the baseline.

Stage 4 - GPU Optimization
Mixed precision training (FP16) reduces memory and increases speed. Automatic CUDA detection uses GPU when available. Pin memory optimization enables faster data transfers. Graceful CPU fallback when GPU is unavailable.

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     Training Configuration                      │
│                       (YAML Config File)                        │
└─────────────────────────┬───────────────────────────────────────┘
                          │
                          ▼
         ┌────────────────────────────────────┐
         │   Carbon Intensity Scheduler       │
         │   - API/Mock data fetch            │
         │   - Threshold comparison           │
         │   - Wait for low-carbon window     │
         └────────────────┬───────────────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │   Start Training?     │
              │   Intensity < 300?    │
              └─────┬─────────────┬───┘
                    │ NO          │ YES
                    ▼             ▼
            ┌───────────┐   ┌──────────────┐
            │   Wait    │   │ Start Tracker│
            │ & Recheck │   │ (CodeCarbon) │
            └───────────┘   └──────┬───────┘
                                   │
                                   ▼
                  ┌────────────────────────────────┐
                  │   PyTorch Training Loop        │
                  │   - Gradient Accumulation      │
                  │   - Mixed Precision (FP16)     │
                  │   - Checkpointing              │
                  └────────────────┬───────────────┘
                                   │
                                   ▼
                  ┌────────────────────────────────┐
                  │   Emissions Tracking           │
                  │   - CO2 (kg)                   │
                  │   - Energy (kWh)               │
                  │   - Power (Watts)              │
                  └────────────────┬───────────────┘
                                   │
                                   ▼
              ┌───────────────────────────────────┐
              │   Save Results                    │
              │   - Model checkpoint              │
              │   - Training summary (JSON)       │
              │   - Emissions log (CSV)           │
              └───────────────────────────────────┘

Installation

Prerequisites

Python 3.8+
PyTorch 2.0+
CUDA (optional, for GPU acceleration)

git clone https://github.com/dakshjain-1616/CarbonAwareModelTraining---by-NEO.git
cd CarbonAwareModelTraining---by-NEO

python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

pip install -r requirements.txt

Required packages: torch>=2.0.0, torchvision>=0.15.0, codecarbon>=2.3.0, pyyaml>=6.0, numpy.

Quick Start

source venv/bin/activate

# Run baseline training (no optimization)
export PYTHONPATH="$PWD/src:$PYTHONPATH"
python src/train.py configs/baseline.yaml

# Run optimized training (carbon-aware + gradient accumulation)
python src/train.py configs/optimized.yaml

# Generate comparison report
python generate_comparison.py

This runs three steps: baseline training without carbon awareness, optimized training with carbon-aware scheduling and gradient accumulation, and a comparison report that quantifies carbon savings and performance metrics.

Demo

Configure carbon-aware training in configs/optimized.yaml:

scheduler:
  enabled: true
  carbon_threshold: 300           # gCO2/kWh
  wait_for_low_carbon: true

training:
  batch_size: 16
  gradient_accumulation_steps: 4  # Effective batch = 64
  epochs: 3

Run optimized training:

python src/train.py configs/optimized.yaml

Output:

============================================================
CARBON-AWARE TRAINING STARTED
============================================================

Carbon Intensity Check:
  Current Intensity: 420.5 gCO2/kWh
  Threshold: 300 gCO2/kWh
  Status: ⏳ Waiting for low-carbon window...

[10 minutes later]
  Current Intensity: 285.3 gCO2/kWh
  Status: ✅ Starting training now!

Training Progress:
  Epoch 1/3 - Loss: 0.324 - Accuracy: 91.2%
  CO2 Emissions: 0.042 kg
  Energy Consumed: 0.15 kWh

============================================================
CARBON SAVINGS vs BASELINE
============================================================

CO2 Reduction: 32.5% (0.024 kg saved)
GPU Memory Reduction: 45.8%
Accuracy: 93.1% (baseline: 93.4%)

Usage Examples

Carbon-Aware Scheduling Only

Disable gradient accumulation, enable scheduling:

scheduler:
  enabled: true
  carbon_threshold: 250

training:
  gradient_accumulation_steps: 1  # No accumulation

Gradient Accumulation Only

Disable scheduling, enable memory optimization:

scheduler:
  enabled: false

training:
  batch_size: 8
  gradient_accumulation_steps: 8  # Effective batch = 64

Real Carbon Intensity API
Configure for production with a real API:

scheduler:
  enabled: true
  use_mock_data: false
  api_endpoint: "https://api.carbonintensity.org.uk/intensity"
  region: "GB"

Custom Model Integration
Replace SimpleCNN in src/train.py:

from my_models import MyCustomModel

def prepare_model(config):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = MyCustomModel(
        input_channels=config['training']['input_channels'],
        num_classes=config['training']['num_classes']
    ).to(device)
    return model, device

Output Format
Training summary JSON saved to output/summary_optimized.json:

{
  "run_name": "optimized",
  "training_metrics": {
    "final_accuracy": 93.1,
    "final_loss": 0.124,
    "epochs": 3,
    "total_time_seconds": 245
  },
  "carbon_metrics": {
    "total_emissions_kg": 0.042,
    "energy_consumed_kwh": 0.15,
    "avg_power_watts": 145.2
  },
  "scheduler_metrics": {
    "wait_time_seconds": 600,
    "initial_intensity": 420.5,
    "training_intensity": 285.3
  },
  "gpu_metrics": {
    "peak_memory_mb": 2048,
    "gradient_accumulation_steps": 4,
    "effective_batch_size": 64
  }
}

Comparison report saved to output/comparison_report.json:

{
  "carbon_savings": {
    "baseline_emissions_kg": 0.074,
    "optimized_emissions_kg": 0.042,
    "reduction_kg": 0.032,
    "reduction_percentage": 43.2
  },
  "accuracy_impact": {
    "baseline_accuracy": 93.4,
    "optimized_accuracy": 93.1,
    "degradation_percentage": 0.3
  },
  "memory_savings": {
    "baseline_memory_mb": 4096,
    "optimized_memory_mb": 2048,
    "reduction_percentage": 50.0
  }
}

Performance

Evaluated on MNIST training - 3 epochs, RTX 3090 GPU:

Carbon Intensity Patterns (Mock Data):

Peak hours 18:00–22:00: ~450 gCO2/kWh
Off-peak hours 02:00–06:00: ~200 gCO2/kWh
Average reduction: 35–45% CO2 by scheduling during low-carbon windows

GPU Memory Savings:

Gradient accumulation 2x: ~30% memory reduction
Gradient accumulation 4x: ~50% memory reduction
Gradient accumulation 8x: ~60% memory reduction

Convergence Validation:

Accuracy degradation under 1% across all tested configurations
Loss convergence matches baseline within 2% tolerance
No divergence observed

Project Structure

CarbonAwareModelTraining---by-NEO/
├── src/
│   ├── scheduler.py                # Carbon intensity API & scheduling
│   ├── tracker.py                  # CodeCarbon emissions tracking
│   ├── train.py                    # Main training pipeline
│   └── utils.py                    # Config loading & logging
├── configs/
│   ├── baseline.yaml               # Baseline training config
│   └── optimized.yaml              # Carbon-aware optimized config
├── output/
│   ├── summary_baseline.json       # Baseline training summary
│   ├── summary_optimized.json      # Optimized training summary
│   ├── comparison_report.json      # Comparative analysis
│   ├── emissions.csv               # CodeCarbon emissions log
│   └── training_*.log              # Detailed training logs
├── models/
│   ├── model_baseline.pt           # Baseline model checkpoint
│   └── model_optimized.pt          # Optimized model checkpoint
├── data/                            # MNIST dataset (auto-downloaded)
├── requirements.txt                 # Python dependencies
├── generate_comparison.py          # Comparison report generator
└── README.md

Key Design Decisions

Why Carbon-Aware Scheduling?

Carbon intensity varies 2–5x throughout the day. Scheduling training during low-carbon windows reduces emissions without affecting model quality. Low-carbon periods also often correlate with cheaper electricity.

Why Gradient Accumulation?

Gradient accumulation enables training larger models on limited hardware by processing smaller micro-batches and updating weights less frequently. Used in BERT, GPT, and other large-scale models for the same reason.

Why CodeCarbon?

CodeCarbon uses lifecycle assessment methodologies, supports CPU, GPU, and multi-device setups, and produces transparent, community-validated calculations. It tracks energy, power, and emissions in a single library.

Why YAML Configuration?

YAML configs are version-controlled, human-readable, and separate code from experiment parameters - enabling reproducible A/B comparisons between baseline and optimized runs.

Testing

Validate installation:

python -c "import torch; print(f'PyTorch: {torch.__version__}')"
python -c "import codecarbon; print('CodeCarbon: OK')"
python -c "import yaml; print('PyYAML: OK')"
python -c "import torch; print(f'CUDA Available: {torch.cuda.is_available()}')"

Run a quick 5-minute test:

python src/train.py configs/test.yaml

Validate carbon savings:

python src/train.py configs/baseline.yaml
python src/train.py configs/optimized.yaml
python generate_comparison.py
cat output/comparison_report.json

Troubleshooting

CUDA Out of Memory

Reduce batch_size and increase gradient_accumulation_steps in the config.

Carbon Intensity API Timeout

No action needed - the pipeline automatically falls back to mock data and training proceeds.

Module Import Errors

export PYTHONPATH="$PWD/src:$PYTHONPATH"

CodeCarbon Tracking Fails

pip install --upgrade codecarbon

Training continues without emissions tracking if CodeCarbon fails.

Scheduler Waits Too Long

Increase max_wait_seconds, raise carbon_threshold, or set wait_for_low_carbon: false in the config.

How I Built This Using NEO

The requirement was a PyTorch training pipeline that schedules GPU workloads based on real-time carbon intensity, reduces memory footprint through gradient accumulation, and tracks emissions with CodeCarbon - producing a side-by-side comparison report. NEO built the full implementation: the carbon intensity scheduler in scheduler.py with API integration and mock fallback, the CodeCarbon emissions tracker in tracker.py, the main training pipeline in train.py with gradient accumulation and mixed precision FP16, the config loader and logging utilities in utils.py, the YAML configs for baseline and optimized runs, the comparison report generator in generate_comparison.py, and the full output structure covering JSON summaries, emissions CSV, and model checkpoints.

How You Can Use and Extend This With NEO

Use it to measure the carbon cost of your existing training runs.
Run python src/train.py configs/baseline.yaml on your own model and data by replacing SimpleCNN in src/train.py with your model. The CodeCarbon tracker produces a JSON summary with CO2 in kg, energy in kWh, and average power in Watts, a baseline measurement before any optimization.

Use the comparison report to justify scheduling infrastructure.
Run both the baseline and optimized configs on the same dataset. The comparison_report.json gives you a concrete before and after - percentage reduction in emissions, energy, and memory, alongside accuracy degradation, that makes the case for carbon-aware scheduling with real numbers from your own hardware.

Use mock data for development and real API for production.
Set use_mock_data: true during development so training always proceeds without waiting. Switch to use_mock_data: false with a real api_endpoint for production runs where actual carbon savings matter.

Extend the scheduler with additional carbon intensity sources.
The scheduler in scheduler.py fetches from a configurable api_endpoint. Adding support for additional regional carbon intensity APIs - Electricity Maps, WattTime, or a custom internal source, means updating the fetch logic in scheduler.py without touching the training loop, tracker, or reporting pipeline.

Final Notes

Carbon intensity varies throughout the day, and most training pipelines ignore it. A 43% reduction in CO2 emissions with less than 1% accuracy degradation, achieved by scheduling when the grid is cleaner and accumulating gradients to reduce memory - shows that sustainable ML is a practical engineering choice, not just an aspiration.

The code is at https://github.com/dakshjain-1616/CarbonAwareModelTraining
You can also build with NEO in your IDE using the VS Code extension or Cursor.
You can use NEO MCP with Claude Code: https://heyneo.com/claude-code