Forem: Nilofer 🚀

Agent Failure Classifier: Post-Hoc Root Cause Analysis for Failed LLM Agent Runs

Nilofer 🚀 — Wed, 29 Apr 2026 07:56:24 +0000

When an LLM agent fails, the trace is right there, the user turns, the tool calls, the responses, the final result. But knowing what happened and knowing why it failed are two different things. Most teams read traces manually, form a guess, and move on.

Agent Failure Classifier is a CLI tool and Python library for post-hoc root cause analysis of failed or low-quality LLM agent runs. Feed it any agent trace and it classifies the failure into one of eight named failure modes, identifies the first turn where things went wrong, and produces a structured report with actionable fixes.

The classifier combines eight fast rule-based detectors with an optional LLM-as-judge pass via OpenRouter. The rule-based layer is free, deterministic, and requires no network access. The LLM pass breaks ties and classifies traces the rules cannot resolve alone.

The Eight Failure Modes

The classifier recognises exactly eight failure modes, each with a precise definition:

HALLUCINATION: Agent stated facts or called tools that do not exist
TOOL_MISUSE: Agent called a real tool with wrong parameters or at the wrong time
CONTEXT_LOSS: Agent forgot earlier decisions or repeated already-completed steps
CIRCULAR_REASONING: Agent looped between the same 2-3 steps without making progress
GOAL_DRIFT: Agent started pursuing a sub-goal and forgot the original task
OVER_REFUSAL: Agent refused an action it was capable of and should have taken
SCHEMA_ERROR: Agent generated malformed JSON for a tool call or structured output
TIMEOUT_CASCADE: One slow tool call caused the agent to rush or skip subsequent steps

These are not fuzzy categories. Each one maps to a specific detector with specific signals. A hallucination is flagged when the agent asserts a factual claim without invoking any retrieval tool. A timeout cascade is flagged when a tool call exceeds a latency threshold and the subsequent agent turn is unusually short relative to the tool output.

How It Works

The classification pipeline runs in two layers.

The rule-based layer runs eight deterministic detectors over the trace. Each detector looks for specific structural signals repeated tool calls with identical inputs, cycles in agent turn content, latency spikes followed by short responses, malformed JSON in tool call outputs. This layer runs offline, requires no API key, and classifies all eight failure modes.

The LLM-as-judge layer is optional. When enabled, it receives traces the rule-based layer couldn't resolve with high confidence and breaks ties. The judge runs via OpenRouter and can be pointed at any OpenRouter model or a local OpenAI-compatible server (Ollama, vLLM, llama.cpp).
Every classification produces a structured report with the classified failure mode, a confidence score, the first turn where the failure was detected, a root cause summary, and a list of actionable fixes.

Getting Started

Install

git clone https://github.com/dakshjain-1616/agent-failure-classifier
cd agent-failure-classifier
python3 -m venv venv
source venv/bin/activate
pip install -e .

Requires Python 3.8+. The only dependencies are pydantic, rich, click, and requests. The rule-based layer runs with no additional setup.

LLM Judge Setup (Optional)
To enable the LLM-judge pass, copy .env.example to .env and set your OpenRouter key:

cp .env.example .env
# edit .env and set OPENROUTER_API_KEY=sk-or-...

Without any key, pass --no-llm to every classify or batch call. The rule-based layer alone classifies all eight failure modes.

CLI

The CLI is exposed as both a console script (agent-failure-classifier) and an importable module (python -m agent_failure_classifier.cli).

Classify a single trace
The core command takes a trace JSON file and returns a structured report. --no-llm keeps it offline, rule-based only, no API call.

agent-failure-classifier classify --trace traces/hallucination_example.json --no-llm

Key flags:

Validate a trace
Before classifying, validate parses the trace and prints its structure: trace ID, goal, turn count, and a preview of each turn. Useful for confirming the trace loaded correctly before running classification.

agent-failure-classifier validate --trace traces/hallucination_example.json

Batch classification
batch runs classification over every *.json file in a directory and produces a failure-mode distribution table plus a per-trace summary.

agent-failure-classifier batch --traces-dir ./traces/ --no-llm

Worked Examples

Example 1 - Hallucination
The trace has a user asking for WWII death statistics. The agent responds directly with a factual claim, no tool call, no retrieval.

{
  "trace_id": "hallucination-001",
  "original_goal": "Get population statistics",
  "final_result": "70 million people died in WWII.",
  "is_successful": false,
  "turns": [
    {"turn_number": 0, "role": "user",  "content": "How many people died in WWII?"},
    {"turn_number": 1, "role": "agent", "content": "70 million people died in WWII."}
  ]
}

Classification: HALLUCINATION, confidence 75%, first failure at turn 1. The detector flags that the agent asserted a factual claim without invoking any retrieval tool. Recommended fixes include adding a fact-checking step, requiring tool verification for factual claims, and implementing retrieval-augmented generation.

Example 2 - Circular Reasoning
Four turns alternating between "Let me analyze this step by step." and "I need more information." The agent makes no progress across the entire trace.
Classification: CIRCULAR_REASONING, confidence 80%. The rule-based detector identifies a 2-step cycle repeating across agent turns and recommends a maximum-iteration limit plus state-change detection.

Example 3 - Timeout Cascade
A slow_api tool call with latency_ms: 6000 followed by a one-word agent response "OK".
Classification: TIMEOUT_CASCADE, confidence 70%. The detector flags the latency breach and notes that the subsequent agent turn is a one-word response, less than half the length of the tool output, indicating the agent rushed through the remaining steps.

Python API

Classify a trace programmatically

import json
from agent_failure_classifier.classifier import FailureClassifier
from agent_failure_classifier.models import AgentTrace

trace = AgentTrace(**json.load(open("traces/hallucination_example.json")))
report = FailureClassifier(use_llm=False).classify(trace)

print(report.classified_failure_mode, report.confidence)
print(report.root_cause_summary)

Record a trace live with TraceRecorder
Rather than constructing trace JSON by hand, TraceRecorder is a context manager that captures an agent run as it executes and writes a trace file to disk on exit. The output is immediately compatible with the CLI and with FailureClassifier.

from agent_failure_classifier.recorder import TraceRecorder

with TraceRecorder(goal="Find Italian restaurants", output_dir="./traces") as r:
    r.add_turn(role="user", content="Find Italian restaurants near me")
    r.add_turn(role="agent", content="Searching...")
    r.add_turn(
        role="tool",
        content="results",
        tool_name="search",
        tool_input={"query": "italian restaurants"},
        tool_output='{"hits": ["Luigi Bistro", "Pasta Palace"]}',
        latency_ms=120,
    )
    r.add_turn(role="agent", content="I found Luigi Bistro and Pasta Palace.")
    r.set_final_result("Found Luigi Bistro and Pasta Palace", is_successful=True)

On exit the trace is saved to ./traces/trace_<id>_<timestamp>.json.

Parse traces from other frameworks
AutoParser auto-detects and normalises three input formats into the canonical AgentTrace model. No manual conversion needed regardless of where the trace came from.

from agent_failure_classifier.formats import AutoParser

trace = AutoParser().parse_file("path/to/trace.json")

The three supported formats are:

Native / generic: a dict with trace_id, original_goal, is_successful, and a turns list. This is the format emitted by TraceRecorder.
LangSmith run export: a dict with run_type, inputs, outputs, and optional child_runs. Tool child runs become TOOL turns; chain and LLM child runs become AGENT turns.
LangGraph state dict: a dict with thread_id and a state.messages list whose entries use type values human, ai, and tool. A minimal list-of-dicts ([{"role": "...", "content": "..."}, ...]) is also accepted by the generic parser.

How I Built This Using NEO

This project was built using NEO, a fully autonomous AI engineering agent that writes code and builds solutions for AI/ML tasks including model evals, prompt optimisation, and end-to-end pipeline development.

The problem was defined at a high level: a tool that takes any agent trace, runs deterministic detectors over it, and classifies the failure into a named category with a structured report and actionable fixes. NEO generated the full implementation: the eight rule-based detectors, the FailureClassifier orchestration layer, the optional LLM-as-judge pass via OpenRouter, the TraceRecorder context manager, the AutoParser with support for native, LangSmith, and LangGraph formats, and the Click-based CLI with classify, validate, and batch commands.

How You Can Build Further With NEO

Use it as a CI/CD quality gate for your agent.
If you're shipping an LLM agent, you can integrate the classifier directly into your deployment pipeline. Record traces from your test suite with TraceRecorder, run batch classification on every pull request, and fail the build if a new failure mode appears or if the rate of a known one spikes. You get a systematic regression check on agent behaviour, not just on code.

Use it to understand where your agent breaks most.
Run batch classification across a directory of historical traces and look at the failure mode distribution. If CONTEXT_LOSS shows up in 40% of your traces, that's a signal about your agent's memory design, not a one-off bug. This turns debugging from reactive to diagnostic, you're looking at patterns across runs, not reading individual traces one by one.

Use it as a live monitoring layer in a multi-agent system.
The classifier runs as an A2A agent, which means it can sit as a node in a multi-agent pipeline. Any agent in the system can send its trace to the classifier after each run and get a structured failure report back. An orchestrator can use that signal to decide whether to retry, reroute, or escalate without any human in the loop.

Use it during agent development to catch regressions early.
Wrap TraceRecorder around your agent during development. Every run produces a trace. Feed those traces into the classifier after each session and you'll know immediately if a change introduced a new failure mode. It's the difference between finding out something broke in production versus finding out in your local environment.

Final Notes

Agent Failure Classifier turns trace debugging from a manual read-and-guess process into a systematic one. Eight named failure modes, a deterministic rule-based layer that runs offline, an optional LLM judge for ambiguous cases, and support for traces from native formats, LangSmith, and LangGraph, all producing a structured report with the first failure turn and actionable fixes.

The code is at https://github.com/dakshjain-1616/agent-failure-classifier
You can also build with NEO in your IDE using the VS Code extension or Cursor.

Synthetic Data Flywheel: A Closed-Loop Pipeline for Instruction-Tuning Data

Nilofer 🚀 — Tue, 28 Apr 2026 10:44:54 +0000

Fine-tuning a model requires data. Good data requires human labeling. Human labeling doesn't scale. And most synthetic generation pipelines stop at generation, they produce candidate pairs but have no mechanism to filter them, measure quality, or feed failure cases back into the next round.
Synthetic Data Flywheel is a closed-loop pipeline that handles the full cycle: generate candidate instruction-output pairs, validate them deterministically, score them with an LLM-as-judge, calibrate that judge against human labels, export clean training data, and feed the failure cases from one cycle as seeds into the next. It ships as a CLI, a Python library, and an A2A-protocol agent surface for multi-agent orchestration.

Everything except the optional fine-tuning step runs on CPU.

The Problem

Synthetic data generation without a quality gate produces noise at scale. And quality gates without calibration produce a judge whose scores you can't trust. The flywheel addresses both: every candidate pair is scored, every score can be validated against human labels, and every failure becomes signal for the next generation cycle rather than a dead end.

How It Works

A dataset moves through a series of additive stages, each producing artifacts keyed by the dataset name. Every stage is idempotent and re-runnable.

Generation: Candidate pairs are produced from seed prompts via OpenRouter, using one of four prompt templates: QA, INSTRUCTION, REASONING, or CREATIVE.

Validation: Deterministic checks run over each pair: schema, length, dedup, PII, language, profanity. Results are written as a JSON report with severity levels (error, warning, never). A cleaned copy of the dataset can be written at this stage.

Judging: An LLM-as-judge scores each pair against a rubric. The judge supports three backends: Ollama, OpenRouter, and Anthropic. Judgments are cached on disk keyed by (backend, model, pair.id, rubric.name@version), repeated judge passes on unchanged pairs are free.

Labeling: Three modes: Interactive (human reviews pairs one by one), bulk (apply a status to a filtered subset), and auto-from-judge (derive labels from judgment scores above a threshold). Labels are stored append-only so sessions can be interrupted and resumed.

Calibration: Treats human labels (status == approved) as ground truth and measures the judge's precision, recall, F1, and accuracy.

Compare: Two or more judgment runs on the same dataset are compared: pass-agreement, Cohen's kappa, and Pearson correlation on the overall score.

Export: Pairs that clear the judge filter are written to a train/val split. The filter expression uses a safe evaluator, only arithmetic, comparisons, and subscript access into the context dict. Attribute access and function calls are rejected.

Cycle feedback: Failure instructions from one cycle are extracted and fed as additional seeds into cycle N+1. The autonomous loop stops when the pass rate drops below min_pass_rate (default 0.5) or max_cycles is reached.

Getting Started

Install

git clone https://github.com/dakshjain-1616/synthetic-data-flywheel
cd synthetic-data-flywheel
pip install -e .

Requires Python 3.11+. Generation requires OPENROUTER_API_KEY. The local judge path requires Ollama, verified against gemma4:latest. Fine-tuning requires Unsloth and a GPU; the repo was verified on a free Colab T4.

Initialize
init creates the directory structure the rest of the pipeline writes into.

flywheel init

Synthetic Data Flywheel Initialized
Data Directory: ./data
Checkpoint Directory: ./data/checkpoints
Report Directory: ./reports
Directories created successfully

Ingest
ingest normalises an existing dataset into the flywheel's internal JSONL format. It supports jsonl, csv, and HuggingFace datasets, and accepts a field mapping flag when the source uses different column names.

flywheel ingest -i demo.jsonl -n demo --tag demo1

Ingested 8 pairs -> data/user/demo.jsonl

Other ingest forms:

flywheel ingest -i data.csv              -n my_dataset -f csv
flywheel ingest -i hf://tatsu-lab/alpaca -n alpaca --limit 500 --hf-split train
flywheel ingest -i data.jsonl -n aliased --map "instruction=prompt,output=completion"
flywheel ingest -i data.jsonl -n x --dry-run

Each successful ingest writes data/user/<name>.jsonl and data/user/<name>.meta.json.

Validate
Before any judging happens, the validator runs deterministic checks over the dataset. This catches structural problems, duplicate pairs, PII, malformed schema, before spending LLM calls on them.

flywheel validate -d demo --checks schema,length,dedup,pii --write-clean data/user/demo.clean.jsonl

Validation: demo
  Total pairs       8
  pii               1
  severity:warning  1
Report: data/validation/demo.report.json
Clean dataset written (8 pairs): data/user/demo.clean.jsonl

The --fail-on error|warning|never flag lets you gate CI on validation issues.

Judge
With a clean dataset, the judge scores each pair against a rubric. The default rubric is built-in; custom rubrics can be passed with --rubric. Results are cached, so re-running after adding new pairs only scores the new ones.

flywheel judge -d demo --backend ollama --model gemma4:latest --tag v1 --max-pairs 3

Judging 3 pairs with ollama:gemma4:latest
  Judged                3
  Passed                0 (0.0%)
  Avg overall (scored)  5.00
  Output                data/judgments/demo.v1.jsonl
  Cache                 hits=0 misses=3 writes=3

Judgments land at data/judgments/<dataset>.<tag>.jsonl. The --tag flag is how multiple judgment runs on the same dataset are tracked separately.

Label
Labeling bridges human judgment and automated scoring. auto-from-judge derives labels directly from the judgment scores, pairs above the threshold are approved, pairs below are rejected.

flywheel label -d demo --mode auto-from-judge --judgments data/judgments/demo.v1.jsonl --reject-below 3.5

For manual review, --mode interactive walks through pairs one by one. For bulk operations, --mode bulk applies a status to a filtered subset. All labels are stored append-only at data/labels/<dataset>.jsonl.

Compare
When you have two judgment runs, say from two different models, compare measures how much they agree. Cohen's kappa close to 1.0 means the two judges are making the same pass/fail decisions.

flywheel compare -d demo --tags judge_a,judge_b

Judge comparison: judge_a vs judge_b
  Common pairs          8
  judge_a passed / mean 6 / 7.44
  judge_b passed / mean 6 / 7.19
  Pass agreement        100.0%
  Cohen's kappa (p/f)   1.000  (near-perfect)
  Score Pearson r       0.965
  Output                reports/demo/compare.json

Calibrate
Calibration answers the question you need to answer before trusting your judge: does its passed decision align with human labels? Precision of 1.0 means every pair the judge passed, a human also approved. Recall of 0.75 means the judge missed 25% of the pairs humans would have kept.

flywheel calibrate -d demo --tag judge_a --approved-is approved

Evaluated pairs  8
  Precision        1.000
  Recall           0.750
  F1               0.857
  Accuracy         0.750
  TP/FP/TN/FN      6/0/0/2

Visualize
visualize renders a suite of PNG charts and an index.html for a dataset — covering label distribution, score distributions, pass/fail breakdown, pair lengths, categories, judge agreement matrix, and validation results.

flywheel visualize -d demo

categories      reports/demo/categories.png
  lengths         reports/demo/lengths.png
  validation      reports/demo/validation.png
  pass_fail       reports/demo/pass_fail.png
  scores          reports/demo/scores.png
  criteria        reports/demo/criteria.png
  labels          reports/demo/labels.png
  judge_agreement reports/demo/judge_agreement.png
  index.html      reports/demo/index.html

Dataset inspection and export
Before exporting, dataset ls and dataset info show what artifacts exist for each dataset.

flywheel dataset ls

name   pairs  source  tags
  demo   8      jsonl   demo1

flywheel dataset info demo

pairs       data/user/demo.jsonl               present
  meta        data/user/demo.meta.json           present
  validation  data/validation/demo.report.json   present
  labels      data/labels/demo.jsonl             present
  judgments   data/judgments                     5 set(s)

Export filters pairs using a safe expression, only pairs with an overall score of 7 or above are written, split 80/20 into train and val.

flywheel dataset export demo \
  --to data/exports/demo.jsonl \
  --format jsonl \
  --judgments data/judgments/demo.judge_a.jsonl \
  --filter "scores['overall'] >= 7" \
  --split train=0.8,val=0.2

Wrote 4 pairs -> data/exports/demo.train.jsonl
Wrote 2 pairs -> data/exports/demo.val.jsonl

Run the autonomous loop
flywheel run ties everything together into a seeds-to-checkpoint cycle. Generation goes through OpenRouter; judging goes through Ollama. If Ollama isn't running, generation still succeeds and pairs are saved in the checkpoint, every judgment falls back to passed=false. The standalone flywheel judge --backend openrouter works fully without Ollama.

export OPENROUTER_API_KEY=sk-or-...
export OPENROUTER_MODEL=meta-llama/llama-3.2-3b-instruct

flywheel run -s "benefits of green tea,history of python language" --max-cycles 1

╭───── Configuration ─────╮
│ Synthetic Data Flywheel │
│ Seeds: 2                │
│ Max Cycles: 1           │
╰─────────────────────────╯
Starting Flywheel with max_cycles=1
============================================================
Starting Cycle 1
============================================================
Using 2 seeds
Generating synthetic data...
Generated 2 pairs
Judging quality...
Passed: 0, Failed: 2
Cycle 1 complete. Pass rate: 0.00%
Flywheel complete. Ran 1 cycles.
       Flywheel Summary
┏━━━━━━━━━━━━━━━━━━━━┳━━━━━━━┓
┃ Metric             ┃ Value ┃
┡━━━━━━━━━━━━━━━━━━━━╇━━━━━━━┩
│ Total Cycles       │ 1     │
│ Total Passed Pairs │ 0     │
│ Avg Pass Rate      │ 0.00% │
└────────────────────┴───────┘

Each cycle writes a checkpoint. The generated pair is saved verbatim inside data/checkpoints/checkpoint_001.json:

{
  "instruction": "benefits of green tea",
  "output": "Here is an example of an instruction-following training data in JSON format:\n\n{\n  \"instruction\": \"What are some of the benefits of drinking green tea?\",\n  \"output\": \"Green tea has numerous benefits, including: - High antioxidant content - Anti-inflammatory properties - May help with weight loss ...\",\n  \"category\": \"instruction\"\n}",
  "source_seed": "benefits of green tea"
}

Status and report

flywheel status
flywheel report

status summarises checkpoint state. report produces an HTML report across cycles written to reports/flywheel_report_<timestamp>.html.

CLI Reference

flywheel --help lists the command groups. Every command has --help with full flag docs.

$ flywheel --help
Usage: flywheel [OPTIONS] COMMAND [ARGS]...

  Synthetic Data Flywheel - Autonomous data generation pipeline.

Commands:
  calibrate  Measure judge 'passed' against human labels (precision/recall/F1).
  compare    Compare two+ judgment runs (Cohen's kappa, agreement, ...).
  dataset    Dataset management: ls | info | export.
  ingest     Ingest a user dataset into the flywheel's JSONL format.
  init       Initialize flywheel configuration.
  judge      Judge a dataset with an LLM-as-judge backend.
  label      Label a dataset: interactive/bulk/auto-from-judge.
  pipeline   Run declarative YAML pipelines.
  report     Generate HTML report from checkpoints.
  run        Run the synthetic data flywheel.
  status     Show current flywheel status.
  validate   Validate a dataset and write a ValidationReport.
  visualize  Render a suite of PNG charts + index.html for a dataset.

Pipeline Runner

Individual commands can be composed into a declarative YAML pipeline and run as a single step. This is useful for repeatable workflows, the pipeline dispatches through the same Click commands as manual runs, so behaviour is identical.

# pipeline_demo.yaml
dataset: demo
steps:
  - validate:
      checks: [schema, length, dedup]
  - export:
      to: data/user/demo_pipeline.jsonl
      format: jsonl

flywheel pipeline run pipeline_demo.yaml

[1/2] flywheel validate -d demo --checks schema,length,dedup
[2/2] flywheel dataset export demo --to data/user/demo_pipeline.jsonl --format jsonl
   Pipeline: demo
  1  validate  ok  0
  2  export    ok  0

Python API

The full pipeline is available as a library. The minimal end-to-end call scores a dataset with an async judge backed by Ollama:

import asyncio
from pathlib import Path
from synthetic_data_flywheel.ingest import load_dataset_jsonl
from synthetic_data_flywheel.rubrics import default_rubric
from synthetic_data_flywheel.judge import AsyncQualityJudge
from synthetic_data_flywheel.judge_backends import get_backend
from synthetic_data_flywheel.judge_cache import JudgmentCache

pairs = load_dataset_jsonl("data/user/demo.jsonl")
backend = get_backend("ollama", model="gemma4:latest")
judge = AsyncQualityJudge(
    backend=backend,
    rubric=default_rubric(),
    cache=JudgmentCache(root=Path(".cache/judge")),
    backend_name="ollama",
)
judgments = asyncio.run(judge.judge_batch(pairs, concurrency=2))
print(sum(j.passed for j in judgments), "/", len(judgments), "passed")

The statistical functions used internally by calibrate and compare are also directly callable:

from synthetic_data_flywheel.stats import cohens_kappa, pearson, prf

cohens_kappa([True, False, True, False], [True, True, True, False])
# 0.5

pearson([1,2,3,4], [1,3,2,5])
# 0.8315...

prf([True,True,False,False], [True,False,True,False])
# {'precision': 0.5, 'recall': 0.5, 'f1': 0.5, 'accuracy': 0.5,
#  'tp': 1, 'fp': 1, 'tn': 1, 'fn': 1}

A2A Agent

The flywheel exposes a FastAPI application implementing the A2A protocol surface, /a2a/capabilities, /a2a/tasks/send, /a2a/tasks/get, /a2a/tasks/cancel so it can be orchestrated as a node in a multi-agent ML pipeline.

python -m synthetic_data_flywheel.a2a_agent
# or
uvicorn synthetic_data_flywheel.a2a_agent:app --host 0.0.0.0 --port 8080

Three capabilities are exposed: generate_synthetic_data, get_status, generate_report. Querying /a2a/capabilities returns the agent's identity and the full capability list:

from fastapi.testclient import TestClient
from synthetic_data_flywheel.a2a_agent import app

client = TestClient(app)
print(client.get("/a2a/capabilities").json())
# {'agent_name': 'synthetic_data_flywheel', 'version': '0.1.0',
#  'capabilities': [{'name': 'generate_synthetic_data', ...},
#                   {'name': 'get_status', ...},
#                   {'name': 'generate_report', ...}]}

r = client.post("/a2a/tasks/send", json={
    "capability": "get_status",
    "inputs": [],
    "parameters": {},
})
print(r.json())
# {'task_id': '...', 'status': {'state': 'completed'},
#  'result': {'type': 'status_result',
#             'content': {'checkpoints_found': 1,
#                         'checkpoint_dir': 'data/checkpoints'}}}

Configuration

All settings are read from environment variables or a .env file:

OPENROUTER_API_KEY=sk-or-...
OPENROUTER_MODEL=qwen/qwen3-8b:free
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=gemma4:latest
DEFAULT_JUDGE_BACKEND=ollama        # ollama | openrouter | anthropic
JUDGE_CONCURRENCY=4
JUDGE_TIMEOUT=600
QUALITY_MIN_SCORE=7.0
MAX_CYCLES=10
PII_POLICY=warn                     # strict | warn | off
A2A_HOST=0.0.0.0
A2A_PORT=8080

JUDGE_TIMEOUT defaults to 600 seconds, large local models can take over two minutes on first call.

Limitations

Fine-tuning requires a GPU: Trainer.prepare_training_artifacts writes a Colab-ready Unsloth notebook under notebooks/training_cycle_NNN.ipynb. Running the training step locally on CPU is not supported by Unsloth.

Autonomous generation requires OpenRouter: flywheel run requires OPENROUTER_API_KEY. The in-loop judge is hardcoded to Ollama (engine.create_judge constructs a sync QualityJudge over OllamaClient); if Ollama isn't available, pairs are persisted but every judgment falls back to passed=false.

Large local judges are slow to cold-start: Gemma 4 (9 GB) takes about 130 seconds the first time it loads into VRAM/RAM. The default JUDGE_TIMEOUT is 600 seconds to cover this.

HuggingFace ingest requires datasets: already a dependency, but gated datasets additionally require HUGGINGFACE_TOKEN.

Anthropic judge backend requires ANTHROPIC_API_KEY: no offline fallback.

How I Built This Using NEO

The problem was defined at a high level: a closed-loop pipeline that generates synthetic instruction-tuning pairs, filters them with a calibrated LLM judge, and feeds failure cases back as seeds for the next cycle. NEO generated the full implementation, the FlywheelEngine cycle loop with checkpointing, the AsyncQualityJudge with three pluggable backends and disk-backed cache, the deterministic Validator with six check types, the LabelStore with append-only storage, the statistical calibration layer (cohens_kappa, pearson, prf), the safe-eval export filter, the declarative YAML pipeline runner, the Matplotlib visualisation suite, and the A2A FastAPI agent surface. 100 tests pass.

How You Can Build Further With NEO

Additional judge backends: the three existing backends share a common interface via get_backend. Any OpenAI-compatible endpoint can be wired in as a new backend, and the judge cache, calibration, and compare logic all work with it immediately without any changes.

Additional generation templates: the generator ships with four templates: QA, INSTRUCTION, REASONING, CREATIVE. New domain-specific templates would let the flywheel produce specialised training data, code generation, structured extraction, tool-use, while the cycle loop, judge, and export pipeline stay entirely unchanged.

Additional validation checks: the Validator already supports six check types plugged into the same --checks flag and report format. New checks for domain-specific quality signals would run in the same validation pass and appear in the same JSON report and visualisation output.

Multi-judge ensembling: compare already computes agreement metrics across judgment runs. Taking the average or majority vote across two or more judge scores before the pass/fail decision would reduce the noise that small local models introduce, without touching the labeling, calibration, or export logic downstream.

Final Notes

Synthetic Data Flywheel closes the loop that most synthetic data pipelines leave open. It generates, validates, judges, calibrates, and exports, and feeds what failed back into the next cycle. The result is a data pipeline that improves with each run rather than producing a static batch.
The code is at https://github.com/dakshjain-1616/synthetic-data-flywheel

You can also build with NEO in your IDE using the VS Code extension or Cursor.

Token Budget Negotiator

Nilofer 🚀 — Mon, 27 Apr 2026 21:55:15 +0000

Everyone knows long prompts cost money. Almost nobody knows which parts of their prompt actually matter.

Prompts accumulate over time, a system message, a style guide, a few-shot example or two, some background context. Each addition made sense when it was added. Over hundreds of API calls, the overhead compounds. And the honest answer to "which of these sections can I remove?" is: you don't know until you test.

Token Budget Negotiator makes that test systematic. It takes a prompt split into named, prioritised sections, runs a greedy ablation loop that drops one section at a time, scores the remaining prompt against a rubric using a local or remote LLM judge, and stops when savings hit the target without falling below the quality threshold. The result is the smallest prompt that still behaves like the original.

It ships as a CLI, a Python library, and an MCP server.

The Problem

Prompt sections are not equal in value, but there's no principled way to know which ones matter for a given task without testing. Manual trimming is guesswork. Token Budget Negotiator answers the question empirically per section, per task, against a rubric that defines what quality means for that use case.

How It Works

A prompt is defined as a YAML file with named sections. Each section carries a type (system, few_shot, context, instruction), a content block, and a priority integer. Priority determines the order in which sections are considered for removal low-priority sections are evaluated first, high-priority sections last.

Before any removal happens, the full prompt is scored by the judge LLM against the rubric. This establishes a baseline. The quality target for the run is baseline_score × threshold.

The ablation loop then works through sections in ascending priority order. For each candidate, a test prompt is built without that section and rescored. If the score still meets the target, the section is dropped permanently and the loop continues with the updated prompt. If not, the section is kept and the next candidate is evaluated.

Two conditions stop the loop:

Token savings reach min_token_savings,the target has been hit.
A removal would push savings above max_token_savings, the ceiling is enforced.

Every accepted removal is verified to actually reduce the token count. The loop cannot produce a larger prompt than it started with.
The output is a NegotiationResult containing the original and optimised token counts, the list of sections removed, per-step scores, quality retention percentage, elapsed time, scoring call count, rubric name, and a full ablation log. This can be written to JSON or YAML.

Getting Started

Install

cd token-budget-negotiator
pip install -e .

Requires Python 3.11+. The local judge path requires Ollama with a model pulled, verified end-to-end against gemma4:latest. The OpenRouter path requires OPENROUTER_API_KEY.

Analyze token distribution
Before negotiating, analyze prints how many tokens each section holds and its share of the total budget:

$ token-budget analyze examples/prompt.yaml

Token Distribution Analysis:
Section              Type              Tokens        % Priority
-----------------------------------------------------------------
system               system                22    18.6%       30
style_guide          system                26    22.0%       10
few_shot_1           few_shot              26    22.0%       20
few_shot_2           few_shot              20    16.9%       25
context              context               12    10.2%       40
instruction          instruction           12    10.2%      100
-----------------------------------------------------------------
TOTAL                                     118   100.0%

Check the local judge:

$ token-budget check-ollama --model gemma4:latest
Ollama is connected
  Host: http://localhost:11434
  Model requested: gemma4:latest
  Model available: Yes

Run the negotiator:

$ token-budget negotiate examples/prompt.yaml \
    --scorer ollama --model gemma4:latest \
    --threshold 0.80 --min-savings 0.20 --max-savings 0.80 \
    --output result.json --format json

Negotiation Result:
  Original: 118 tokens, score=0.600
  Optimized: 92 tokens, score=0.700
  Savings: 22.0%
  Quality Retention: 116.7%
  Success: Yes
  Sections removed: style_guide

Results saved to result.json

result.json contains the full ablation log, the final optimized prompt, per-step scores, and metadata (elapsed time, scoring call count, rubric name).

Run the negotiator - OpenRouter path

$ export OPENROUTER_API_KEY=sk-or-...
$ token-budget check-openrouter
OpenRouter is connected
  Base URL: https://openrouter.ai/api/v1
  Model requested: qwen/qwen3-8b

$ token-budget -v negotiate examples/prompt.yaml \
    --scorer openrouter --model meta-llama/llama-3.2-3b-instruct \
    --rubric rubrics/qa.yaml \
    --threshold 0.7 --min-savings 0.1 --max-savings 0.6 --no-cache
Connected to openrouter

Negotiation Result:
  Original: 118 tokens, score=1.000
  Optimized: 92 tokens, score=0.900
  Savings: 22.0%
  Quality Retention: 90.0%
  Success: Yes
  Sections removed: style_guide

With a looser threshold (-t 0.7 --min-savings 0.1 --max-savings 0.5) and caching left on, the same model drops two sections for 44.1% savings at 100% quality retention.

CLI Reference

Key negotiate flags:

-r, --rubric PATH: YAML rubric. Defaults to a built-in accuracy+relevance rubric.
-s, --scorer {ollama,openrouter}: which judge to use. Default ollama.
-m, --model TEXT: model name (gemma4:latest for Ollama, qwen/qwen3-8b for OpenRouter, etc.).
-t, --threshold FLOAT: minimum fraction of the baseline score to keep. Default 0.95.
--min-savings FLOAT: stop once savings reach this fraction. Default 0.40.
--max-savings FLOAT: never drop sections if it would save more than this. Default 0.60.
-o, --output PATH / -f, --format {json,yaml}: write a machine-readable report.
--no-cache: disable the in-memory scoring cache.

Python API

from token_budget_negotiator import (
    Negotiator, OllamaScorer, PromptSection, Rubric,
)
from token_budget_negotiator.models import RubricCriterion, SectionType

sections = [
    PromptSection(name="sys", content="You are helpful.",
                  section_type=SectionType.SYSTEM, priority=20),
    PromptSection(name="q", content="What is 2+2?",
                  section_type=SectionType.INSTRUCTION, priority=100),
]

rubric = Rubric(
    name="qa", description="qa rubric",
    criteria=[RubricCriterion(name="accuracy",
                              description="factually correct",
                              weight=1.0)],
)

scorer = OllamaScorer(model="gemma4:latest")
negotiator = Negotiator(scorer=scorer, quality_threshold=0.9,
                        min_token_savings=0.1, max_token_savings=0.9)

result = negotiator.negotiate(sections=sections, rubric=rubric, target_task="qa")
print(result.original_token_count, "->", result.optimized_token_count)
print("removed:", result.sections_removed)

Rubric Format

The rubric defines what quality means for the task. The judge scores each test prompt against it. Three rubrics ship in rubrics/: qa.yaml, coding.yaml, summarization.yaml.

name: qa
description: General question-answer rubric
version: "1.0"
criteria:
  - name: accuracy
    description: Is the response factually correct?
    weight: 1.0
  - name: relevance
    description: Does it answer what was asked?
    weight: 1.0
scoring_instructions: |
  Score 0-1. 1 = perfect, 0 = wrong or irrelevant.
output_format: json

MCP Server

The library also runs as an MCP server over stdio transport, exposing two tools, analyze and negotiate, so Claude Code or any MCP-compatible agent can call it directly during a session.

python -m token_budget_negotiator.mcp_server --scorer ollama --model gemma4:latest

analyze takes a sections list and returns token distribution as JSON. negotiate takes sections, rubric, task, thresholds, and scorer config and returns the full negotiation result as JSON.

Limitations

Ablation is greedy one-at-a-time in priority order, not exhaustive subset search.
The judge is asked for strict JSON; free-text replies fall back to regex score extraction with reduced confidence.
Small local judges like gemma4 are noisy, prefer thresholds in the 0.80-0.90 range and expect multi-minute wall clock even for short prompts.
check-openrouter and the OpenRouter scorer require OPENROUTER_API_KEY; there is no offline stub.
Only the remove compression strategy is wired up. CompressionStrategy and sections_compressed exist on the model but are not yet produced by the negotiator.

How I Built This Using NEO

The problem was defined at a high level: a tool that takes a structured prompt, scores it with a local or remote LLM judge, and finds the minimum set of sections needed to hit a quality threshold. NEO generated the full implementation, the greedy ablation loop in Negotiator, the OllamaScorer and OpenRouterScorer with their shared interface, the ScoreCache with TTL-based invalidation, the SectionTokenizer backed by tiktoken, the YAML rubric format, the MCP server with its two exposed tools, and the CLI built on Click. All 49 tests pass.

Final Notes

Token Budget Negotiator turns prompt compression from guesswork into an empirical process. It scores every section against a rubric, drops only what demonstrably doesn't matter, and produces a report showing exactly what changed and why.
The code is at https://github.com/dakshjain-1616/token-budget-negotiator
You can also build with NEO in your IDE using the VS Code extension or Cursor.

Agent Memory Compressor: Intelligent Memory Compression for Long-Running LLM Agents

Nilofer 🚀 — Mon, 27 Apr 2026 13:02:47 +0000

A 10-turn agent session can easily accumulate 20,000+ tokens of raw history, leaving almost no room for the current task. Naive truncation drops older turns wholesale, including the decisions and discovered facts the agent needs to avoid repeating work. Developers need a principled way to compress history rather than discard it.

Agent Memory Compressor is a Python library that implements an intelligent memory compression pipeline for long-running LLM agents. It combines importance-based scoring, LLM-driven summarization, a forgetting curve trigger, and a token-budgeted context builder so agents can run indefinitely without exhausting their context windows, while preserving the facts and decisions that matter.

The Problem: Context Window Exhaustion

The problem has three dimensions, and agent-memory-compressor addresses each one directly:

What to keep: A multi-signal importance scorer ranks every memory entry.
How to shrink: Three pluggable compression strategies replace low-value entries with compact equivalents using any OpenAI-compatible LLM.
When to act: A forgetting curve fires compression automatically when either a turn interval or a token threshold is crossed.

How It Works

Importance Scoring
Every memory entry is scored by the ImportanceScorer, which combines three signals:

Compression Strategies
Given a scored store, the CompressionEngine exposes three strategies:

summarize(entry): Asks the LLM for a short summary that preserves all decisions and facts.
extract_facts(entry): Asks the LLM for a bullet list of facts and decisions, stored as high-importance compressed entries.
archive(entry): Replaces the entry with a minimal reference; the original content is retained in the entry's compression_history for audit.

The MemoryCompressor orchestrates the pipeline: score, pick the lowest-scoring non-protected entries, apply the least-destructive strategy first, and iterate until the store is under token_budget. Every successful replacement is verified to actually reduce the token count, so compression can never make the context larger.

The Forgetting Curve
The ForgettingCurve decides when to compress. It combines two triggers:

Turn-based: fires once the number of turns since the last compression reaches compression_interval_turns (default: 10)
Token-based: fires once MemoryStore.token_total() exceeds compression_threshold_tokens (default: 6000), with hysteresis to prevent thrashing.

should_compress(store) returns True as soon as either condition is met. get_compression_priority(store) returns entries sorted by importance, so the orchestrator always attacks the least-valuable history first.

Installation

pip install -e .
# optional, for live LLM calls
pip install openai

The package depends on pydantic, tiktoken (for cl100k_base token counts), click, and rich.

Usage Example

from agent_memory_compressor import MemoryEntry, MemoryStore, MemoryCompressor
from agent_memory_compressor.triggers import ForgettingCurve
from agent_memory_compressor.context import ContextBuilder, ContextConfig
from agent_memory_compressor.strategies import LLMClient, CompressionEngine

store = MemoryStore()
for turn, (role, content) in enumerate(conversation, start=1):
    store.add_entry(MemoryEntry(content=content, role=role, turn_number=turn))

llm = LLMClient(api_key="sk-...", model="gpt-4o-mini")
compressor = MemoryCompressor(
    token_budget=4000,
    protected_recent=3,
    engine=CompressionEngine(llm_client=llm),
)

curve = ForgettingCurve(compression_interval_turns=10,
                       compression_threshold_tokens=6000)

if curve.should_compress(store):
    report = compressor.compress(store)
    curve.mark_compressed(store)
    print(f"Saved {report.tokens_saved} tokens "
          f"({report.compression_ratio:.0%} reduction)")

context = ContextBuilder(ContextConfig(max_tokens=4000)).build_context(
    store, system_message="You are a helpful assistant."
)

Without an API key, LLMClient falls back to a deterministic short stub so pipelines remain runnable in tests and offline demos. A full end-to-end demo lives at demos/long_run_demo.py.

API Reference

A memory-cli entrypoint (click-based) is installed for quick inspection, compression, and demo runs.

Integration with the Session Manager

The adapters module wires the compressor directly into the Stateful Agent Session Manager:

from agent_memory_compressor.adapters import compress_session

compressed_messages, report = compress_session(
    session,              # anything exposing get_messages() / get_metadata()
    token_budget=4000,
    protected_recent=3,
)

SessionAdapter.session_to_store projects session messages into a MemoryStore, compressor.compress(...) runs the pipeline, and store_to_session projects the compressed entries back into the session's message format, preserving original roles and retaining the compression history on each compacted entry.

How I build This Using NEO

This project was built using NEO. A fully autonomous AI engineering agent that writes code end-to-end for AI/ML tasks including model evals, prompt optimization, and pipeline development.
I described the problem at a high level: an intelligent memory pipeline for long-running agents that scores history by importance, compresses the least valuable entries, and assembles a token-bounded context.

NEO generated the full implementation, the multi-signal ImportanceScorer, the three compression strategies in CompressionEngine, the turn- and token-based ForgettingCurve triggers, the token-budgeted ContextBuilder, and the SessionAdapter that wires everything into an existing agent session, all as a coherent, installable Python library.

How You Can Build Further With NEO

Semantic similarity scoring: straightforward, just call an embeddings API and add the score to the existing pipeline. Done all the time in RAG systems.
Pluggable tokenizers: purely an engineering task, just abstract the tiktoken call. No research needed.
More agent framework adapters: LangChain/LlamaIndex all expose message lists. The session_to_store pattern already exists, just repeat it for each framework.
Streaming compression: the trigger logic already exists, moving it per-turn is a refactor not a research problem.

Final Notes

Agent Memory Compressor is a principled answer to context window exhaustion for long-running LLM agents.

Instead of truncating history blindly, it scores every piece of memory, applies the least-destructive compression strategy first, and assembles a token-bounded context that preserves what the agent actually needs, the decisions, discovered facts, and recent turns that matter most.

The code is at https://github.com/dakshjain-1616/Agent-Memory-Compressor
You can also build with NEO in your IDE using the VS Code extension or Cursor.

Cache-Augmented Generation (CAG): A RAG-less Approach to Document QA

Nilofer 🚀 — Sat, 25 Apr 2026 10:29:22 +0000

Most document QA systems today rely on Retrieval-Augmented Generation (RAG). The standard pipeline is familiar: chunk the document, generate embeddings, store them in a vector database, and retrieve relevant chunks at query time.

This works, but it comes with trade-offs. The model only sees fragments of the document, retrieval adds latency, and the system becomes more complex with multiple moving parts.

Cache-Augmented Generation (CAG) explores a different approach,where the document is processed once and reused across queries instead of being retrieved repeatedly.

What is Cache-Augmented Generation

Cache-Augmented Generation (CAG) approaches document QA by reusing the model’s internal state instead of retrieving context for every query.

During ingestion, the entire document is processed in a single pass. In this step, the model builds its KV (key-value) cache, which represents the document’s context.

This KV cache is then saved to disk.

When a query is made, the cache is restored and the query is appended, allowing the model to generate responses using the previously processed document.

How It Works

Step 1: Ingest (done once per document)
The document is wrapped in a structured prompt and sent to llama-server. The model runs a full prefill pass, loading every token into the KV cache. This takes time, proportional to document size, but only happens once. The KV cache is then saved to a .bin file on disk.

Step 2: Query (instant, repeatable)
Before each query, the saved .bin file is restored into llama-server's KV cache in ~1 second. The user's question is appended and the model generates an answer with full document context active. No re-reading, no re-embedding.

Step 3: Persistence
KV slots survive server restarts. Kill the server, restart it, and your next query restores the cache from disk just as fast. The 24-minute prefill for War and Peace only needs to happen once ever.

Validated Results

All 11 GPU tests were run on an NVIDIA RTX A6000 (48 GB VRAM) with Qwen3.5-35B-A3B Q3_K_M at 1,048,576 token context.

Performance

Example Output
“Who is Pierre Bezukhov?” → Correct, detailed answer
“What happened at the Battle of Borodino?” → Correct, detailed answer

Quick Start

Prerequisites: Linux, NVIDIA GPU (8 GB+ VRAM), Python 3.8+

# 1. Build llama.cpp + download model (one-time, ~35 min)
./setup.sh

# 2. Start the LLM server
./start_server.sh

# 3. Start the API server
python3 src/api_server.py

# 4. Ingest a document
python3 src/ingest.py my_document.txt --corpus-id my_doc

# 5. Query it
python3 src/query.py my_doc "What is this document about?"

That's it. After step 4, the KV cache is saved to kv_slots/my_doc.bin. Every future query restores it instantly, and it survives server restarts.

Model Selection

setup.sh auto-detects GPU VRAM and picks the right model:

The 24 GB+ path uses unsloth/Qwen3.5-35B-A3B-GGUF on HuggingFace and requires a free HF account + access token.

REST API

Start the API server with python3 src/api_server.py --port 8000 (optionally set CAG_API_KEY env var to enable key auth).

Full API docs available at http://localhost:8000/docs when the server is running.

Directory Structure

.
├── setup.sh              # Builds llama.cpp, downloads model
├── start_server.sh       # Launches llama-server with CAG flags
├── requirements.txt
├── src/
│   ├── api_server.py     # FastAPI REST API
│   ├── ingest.py         # CLI: ingest a document
│   ├── query.py          # CLI: query a corpus
│   └── demo.py           # End-to-end demo
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── docs/
│   ├── REPORT.md         # Full GPU validation report with all 11 test results
│   └── GPU_TESTING.md    # GPU test checklist
├── models/               # GGUF weights (not committed)
├── kv_slots/             # Saved KV cache .bin files (not committed)
└── logs/                 # Runtime logs (not committed)

Limitations

Linux + NVIDIA only: TurboQuant CUDA kernels require Linux and NVIDIA GPUs (no Windows, macOS, or AMD).

Long initial prefill: ~900K tokens can take ~24 minutes on an A6000. This is a one-time cost; subsequent queries restore in ~1 second.

VRAM gating: Systems with lower VRAM use smaller models with shorter context.

Single active corpus: Uses a single llama.cpp slot (slot 0). Switching corpora requires restoring a different KV cache (~1 second).

Long-context limitations: YaRN extrapolation biases attention toward the start and end of documents, so mid-document content can be missed at very large context sizes.

Build time: Initial setup (./setup.sh) can take ~35 minutes to compile CUDA kernels.

Model access requirements: Large models (e.g., Qwen3.5-35B) require a Hugging Face account and access token.

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 system was defined at a high level, describing a document QA workflow that avoids RAG by loading full documents into an LLM, saving the KV cache, and restoring it for repeated queries.

Based on this, NEO generated the implementation, handled debugging across CUDA, Python, and shell components, and validated the system through a series of GPU tests.

This included fixing multiple issues during development and running end-to-end validation to ensure ingestion, cache restoration, and query flows worked reliably.

How to Extend This Further with NEO

The system can be extended in several ways:

supporting multiple KV cache slots
improving handling of long-context attention limitations
optimizing cache storage and compression
exploring hybrid approaches combining CAG with retrieval
extending API capabilities

These extensions would require changes to the current implementation and can be explored based on system requirements.

Final Notes

Cache-Augmented Generation is an alternative way to approach document QA.

Instead of retrieving context at query time, it shifts the cost to a one-time preprocessing step and reuses the model’s KV cache.

This makes repeated queries faster and makes the document context available to the model through the KV cache, while introducing trade-offs in setup time and hardware requirements.

The code is at https://github.com/dakshjain-1616/Cache-Augmented-Generation-CAG-System
You can also build with NEO in your IDE using the VS Code extension or Cursor.

Loop Anti-Pattern Linter: Finding Hidden Performance Issues in Python

Nilofer 🚀 — Sat, 25 Apr 2026 04:40:58 +0000

When writing Python code, loop-heavy logic often looks correct but hides performance issues that only show up at scale.

Patterns like repeated membership checks, string concatenation, or nested iteration over the same data can silently increase time complexity. These are easy to miss during development because they do not break functionality, but they can significantly affect runtime as input size grows.

Why This Matters

Most of these patterns are not syntax errors. The code runs correctly, but the performance cost grows with input size.

Because they are subtle, they are rarely caught during code review unless someone is actively looking for them. This is especially relevant in data processing, backend services, and any code that operates on large collections.

What is Loop Anti-Pattern Linter?

This is a static analyzer that scans Python code to detect common loop anti-patterns.

It does not execute the code. Instead, it parses the source using Python’s ast module and identifies inefficient patterns using rule-based detectors.

Each finding includes an estimated slowdown percentage based on Big-O heuristics and a suggestion. Results are ranked so that higher-impact issues appear first.

How It Works

At a high level, the tool processes code in the following steps:

Accepts a Python file or directory as input
Parses the code into an Abstract Syntax Tree (AST)
Runs detectors implemented as NodeVisitor subclasses
Each detector targets a specific anti-pattern
Assigns a predefined estimated_slowdown_pct value
Sorts findings by this value
Outputs results as a table or JSON

Because the analysis is static and AST-based, it introduces zero runtime overhead.

How the Scoring Works

The scoring is based on Big-O heuristics.

Each pattern is associated with a predefined slowdown estimate based on typical complexity impact. For example, patterns that introduce nested iteration or repeated linear scans are assigned higher percentages.

These values are not derived from runtime benchmarking. They are directional signals to help prioritize fixes.

Findings are sorted by the estimated_slowdown_pct field so that the most impactful issues appear first.

Detected Loop Anti-Patterns

The tool detects a set of common loop inefficiencies and assigns a heuristic slowdown estimate to each:

Example Output

By default, the output includes the fields shown below. When using the --explain flag, an additional explanation field is included.

{
  "file": "src/processor.py",
  "line": 42,
  "pattern": "ListAppendInLoop",
  "estimated_slowdown_pct": 30,
  "suggestion": "Replace with list comprehension"
}

CLI Usage

# Analyze a directory
python loop_antipattern_lint.py ./src

# Analyze a single file
python loop_antipattern_lint.py mymodule.py

# JSON output
python loop_antipattern_lint.py ./src --json

# Filter high-impact issues
python loop_antipattern_lint.py ./src --min-slowdown 30

AI-Powered Explanations

The --explain flag adds natural language explanations to each finding using OpenRouter.

OPENROUTER_API_KEY=... python loop_antipattern_lint.py ./src --explain

This helps explain why a pattern is inefficient and how to improve it.

How I Built This Using NEO

This project started from a simple requirement: identify inefficient loop patterns in Python code and highlight the ones that are likely to impact performance.

Instead of building everything manually, I used NEO to generate an initial version of the tool from a high-level description.

FYI: 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 prompt focused on the expected behavior:

Build a Python CLI tool that analyzes code using AST, detects loop anti-patterns,assigns slowdown estimates based on Big-O heuristics, and outputs ranked results with suggestions.

This produced a working baseline that aligned with the intended functionality of the tool.

From there, the focus was on validating the output and making small adjustments. This included checking that the detected patterns matched expectations, refining the heuristic slowdown values, and ensuring the CLI usage behaved as intended.

This approach made it possible to move from a high-level idea to a functional tool quickly, without manually implementing each part from scratch.

How to Extend This Further with NEO

The current system can be extended in several ways:

adding new detectors for additional patterns
refining slowdown estimates
improving suggestions
integrating with CI pipelines
extending analysis to other inefficiencies

These extensions can be approached by describing the required behavior and iterating on the existing implementation using NEO.

Running the Project

pip install -r requirements.txt
python loop_antipattern_lint.py ./src

Final Notes

This tool focuses on identifying inefficiencies that are easy to miss during development but can have a measurable impact as data size grows.

By combining static analysis with cost estimation based on Big-O heuristics, it helps prioritize optimizations that are worth addressing.

The code is at https://github.com/dakshjain-1616/Loop-Anti-Pattern-Linter
You can also build with NEO in your IDE using the VS Code extension or Cursor.

Low-Latency Model Router: Automatic LLM Selection Across OpenRouter

Nilofer 🚀 — Wed, 22 Apr 2026 14:41:20 +0000

When calling an LLM API directly, the model selection is typically fixed ahead of time. In practice, this creates several limitations across different workloads.
Latency varies depending on the model and request type. Lower-cost models may not maintain quality under certain conditions. External API failures require fallback handling. Repeated identical requests increase cost if caching is not applied.
These constraints require a routing layer that can dynamically select models based on latency, cost, and quality, while also handling caching, fallback, and observability.

What This Project Does

This project implements a low-latency LLM router that dynamically selects the best model for each request based on latency, cost, and quality.

Instead of sending every request to a fixed model, the router evaluates multiple candidates at runtime and routes each request to the most suitable option.

How the Scoring Engine Works

The scoring engine is the component responsible for evaluating available models and selecting the most suitable one for each request.

It assigns a score to each model based on latency, cost, and quality, and then selects the model with the highest score according to the defined priority.

Every model in the catalogue is scored on three dimensions. A routing decision is a single pass over the catalogue to find the highest-scoring candidate given your weight preferences:

Score = w_latency * (1 - norm_latency)
      + w_cost    * (1 - norm_cost)
      + w_quality * quality_score

You control the weights via the priority field:

Model Catalogue

If the selected model fails, the router automatically retries with the next-best candidate. Identical requests are served from cache. Redis, or in-memory if Redis is unavailable.

Project Structure

ml_project_0652/
├── src/
│   ├── models.py              # Pydantic schemas
│   ├── router/
│   │   ├── core.py            # Weighted scoring engine + model catalogue
│   │   ├── metrics.py         # Rolling-window metrics tracker
│   │   ├── openrouter.py      # Async OpenRouter API client
│   │   └── cache.py           # Redis cache + MockCache fallback
│   ├── api/
│   │   ├── main.py            # FastAPI app
│   │   └── routes.py          # Route definitions
│   └── cli/
│       └── commands.py        # Typer CLI
├── tests/                     # 29 unit + integration tests
├── start_router.py            # Server entry point
├── config.yaml                # Server, Redis, and routing config
├── .env.example               # Environment variable template
└── requirements.txt

REST API

curl -X POST http://localhost:8000/route \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "What is the capital of France?"}],
    "priority": "balanced"
  }'

Example response:

{
  "id": "gen-abc123",
  "model": "google/gemini-flash-1.5",
  "choices": [
    {
      "message": {
        "role": "assistant",
        "content": "Paris."
      }
    }
  ],
  "usage": {
    "prompt_tokens": 14,
    "completion_tokens": 3,
    "total_tokens": 17
  },
  "routing_decision": {
    "selected_model": "google/gemini-flash-1.5",
    "reason": "Best composite score based on latency, cost, and quality"
  },
  "latency_ms": 312.4,
  "cached": false
}

Route with speed priority and latency cap:

curl -X POST http://localhost:8000/route \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "Translate: hello"}],
    "priority": "speed",
    "max_latency_ms": 700
  }'

All Endpoints

Caching

Identical requests are cached using a hash of the request.
Redis configuration:

redis:
  host: "localhost"
  port: 6379
  ttl_seconds: 3600

If Redis is unavailable, the system automatically falls back to in-memory caching.

Fallback

Fallback models are defined in config.yaml:

routing:
  fallback_models:
    - "openai/gpt-4o-mini"
    - "anthropic/claude-3-haiku"
    - "google/gemini-flash-1.5"

Metrics

The system tracks:

average latency
p95 latency
p99 latency
per-model usage
cache hit rate

Available via /metrics.

CLI

# List models
python -m src.cli.commands models

# Preview routing decision
python -m src.cli.commands route "What is 2+2?" --dry-run

# Route with quality priority
python -m src.cli.commands route "Summarize this article" --priority quality --dry-run

# Route with latency cap
python -m src.cli.commands route "Hello" --priority speed --max-latency 600 --dry-run

# Live call
python -m src.cli.commands route "What is 2+2?" --priority balanced

# Benchmark
python -m src.cli.commands benchmark --iterations 10

Configuration

server:
  host: "0.0.0.0"
  port: 8000

redis:
  host: "localhost"
  port: 6379
  ttl_seconds: 3600

routing:
  default_weights:
    latency: 0.4
    cost: 0.3
    quality: 0.3
  fallback_models:
    - "openai/gpt-4o-mini"
    - "anthropic/claude-3-haiku"
    - "google/gemini-flash-1.5"

How I Built This Using NEO

I used NEO AI Engineer to build this project by starting with a high-level description of the system requirements.

FYI: 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 goal was to create a routing layer that can dynamically select LLMs based on latency, cost, and quality, while also supporting caching, fallback handling, and metrics tracking.

I began by giving this task prompt to Neo:

Build a FastAPI LLM router that selects models from OpenRouter based on latency, cost, and quality. Include weighted scoring, fallback handling, caching, and metrics tracking.

From this prompt, NEO generated the initial project structure, including the API layer and routing logic.

It then produced the core components required for the system, such as:

Model selection based on weighted scoring
Request handling through the API layer
Caching support with Redis and in-memory fallback
Fallback handling for failed model calls
Metrics tracking for latency and usage

These pieces came together as a working router that could process requests, select models based on defined priorities, and return responses with routing decisions and metrics.

This made it possible to move from a high-level idea to a functioning system without manually implementing each part of the pipeline.

How to Extend This Further with NEO

Once the base system is in place, NEO can also be used to iterate on specific components.

You can extend this project with more functionality such as:

Adjusting scoring weights for different workloads
Refining model selection strategies
Modifying cache policies and TTL behavior
Adding constraints such as latency limits or budget caps
Integrating additional model providers

Running the Project

git clone https://github.com/dakshjain-1616/low-Latency-Model-Router
cd low-Latency-Model-Router
pip install -r requirements.txt
cp .env.example .env
python start_router.py

Ensure that you specify your OpenRouter API Key in .env while running the low latency model router.

Final Notes

This router implements a routing layer that dynamically selects models based on latency, cost, and quality while handling caching, fallback, and observability.

It separates routing logic from model usage, allowing systems to adapt across different workloads without changing application logic.

The code is at https://github.com/dakshjain-1616/low-Latency-Model-Router
You can also build with NEO in your IDE using the VS Code extension or Cursor.

[Boost]

Nilofer 🚀 — Tue, 14 Apr 2026 15:36:01 +0000

Gaurav Vij

Apr 14

A CLI tool to score fine-tuning dataset quality before training starts

#ai #finetuning #llm #datascience

Comments

3 min read