Show HN: LoopGain - Stop agent loops with control theory, not max_iterations

Show HN: LoopGain - Stop agent loops with control theory, not max_iterations

An open-source cost controller for AI agent loops. AI agent loops waste time and money when they don't know when to stop. LoopGain measures the loop in real time and stops it the moment it has actually converged - and rolls back before it degrades - instead of running to a fixed max_iterations cap.

Benchmark

2,000 paired trials across 10 workload cells (run it yourself):

• 92.8% less API spend than max_iter=20 - $27.05 → $1.94 in total benchmark spend
• ~15× faster - median wall-clock per trial 30.9s → 2.1s
• Quality preserved, not traded for speed - judge win-rate 0.50-0.63 on natural-distribution workloads (W1-W4, CI excluding null on most cells), 0.92-0.95 on engineered-failure workloads (W5); 0.678 weighted preference across 1,800 judge comparisons
• Zero of six kill criteria fired (all six pre-registered with thresholds before the run)

Honest limits, up front

LoopGain detects convergence, not correctness - it knows when more iterations won't help, not whether the answer is right, and it's only as good as the verifier behind your error signal. The full list of what it can't do → Home: loopgain.ai

Works for any iterative AI workflow with a measurable error signal - verify-revise loops, refinement passes, tool-use retry chains, RAG with self-correction, code-gen with linter feedback, multi-step reasoning loops. Pre-built adapters for LangGraph, CrewAI, AutoGen, LangChain, OpenAI Agents SDK, and Claude Agent SDK; drop-in via the raw API for any custom stack. Pure Python, no runtime dependencies.

Production agent loops universally use max_iterations=N as their termination policy. It's the embarrassing default of agentic AI: you either waste compute (loop stops too late) or ship bad output (loop stops too early). LoopGain replaces it with a control-theoretic stop-and-rollback policy grounded in the Barkhausen criterion - a foundational result from electrical-engineering feedback-oscillator analysis (1921).

pip install loopgain

Pure Python, no dependencies, supports Python 3.10+.

Using Claude Code?

The loopgain-plugin scans your whole repo for wrappable loops - literal, recursive, graph-cycle, and semantic - and proposes reviewed diffs one file at a time (never auto-applied):

/plugin marketplace add loopgain-ai/loopgain-plugin
/plugin install loopgain

Three lines of code

Wrap any iterative loop with a measurable error signal:

from loopgain import LoopGain

lg = LoopGain(target_error=0.1)

while lg.should_continue():
    errors = verifier.verify(output)
    lg.observe(errors, output=output)
    output = reviser.revise(output, errors)

result = lg.result
print(result.outcome)            # "converged" | "oscillating" | "diverged" | "stalled" | "max_iterations"
print(result.best_output)        # the lowest-error iteration's output
print(result.iterations_used)
print(result.savings_vs_fixed_cap)

observe() accepts either a numeric error magnitude or any sequence (whose length becomes the magnitude). Pass output=... to enable the best-so-far buffer.

The one thing you provide is the error signal

A single non-negative number, every iteration, that says how wrong the current output is. Lower is better; zero means done. LoopGain doesn't know what your loop does - it just watches that number's trajectory and decides whether to keep going, stop, or roll back.

Your loop already has some way of knowing the output isn't good yet (or it wouldn't keep revising). Turn that into a number:

The only rules: non-negative, and smaller as the output gets better. Returning the raw list of problems works directly - observe() uses its length as the magnitude (e.g. hand it the list of failing tests).

If your quality is fuzzy and has no natural "zero," run with target_error=None: LoopGain then stops when the number stops improving, wherever that plateau is, instead of waiting for an exact target.

Every stop/continue decision is made from this one number, so LoopGain is only as good as the error signal you give it - pick one that genuinely tracks output quality.

How it works

LoopGain measures empirical loop gain (Aβ = E(n) / E(n-1)) at every iteration and exposes it as a smoothed time series for visualization. The decision engine, however, classifies the full error trajectory using four features:

E_ratio = E_current / E_first          # cumulative reduction
slope_log = OLS slope of log10(E)      # geometric trend direction
slope_p = t-test p-value of slope      # statistical significance
osc_std = std of detrended log10(E)    # oscillation magnitude

It routes the trajectory into one of five named states:

Plus a short-circuit: if observed error drops at or below target_error, the loop stops immediately with state TARGET_MET. The default target_error=0.0 short-circuits on exactly zero error - the natural completion signal for verifier-driven loops. Pass target_error=None to disable the short-circuit and rely on stability detection alone.

The decision is conservative by design: requiring both statistical significance and meaningful cumulative motion before terminating prevents false-positive aborts on noisy real-LLM error series. Validated at 98.8% macro-averaged accuracy across 5 regimes on N=1000 deterministic-mock trajectories (see RESULTS_v2_classifier.md). The STALLING ceiling of ~94% is the t-test's irreducible 5% type-I error rate, not a classifier weakness.

Recommended minimum: 6 iterations for reliable trend significance. At n≤4 the t-test is severely underpowered (df=2 requires |t|>4.3 for p<0.05).

Telemetry

send_telemetry() - your dashboard

lg.send_telemetry(
    workload_id="my-rag-pipeline",
    actual_dollars_spent=1.94,
    actual_dollars_saved=25.11,
)

Signature:

def send_telemetry(
    self,
    workload_id: str = "",
    include_per_iteration: bool = True,
    endpoint: str = "",
    token: str = "",
    actual_dollars_spent: float | None = None,
    actual_dollars_saved: float | None = None,
) -> bool

Opt-in. Send a single anonymized telemetry POST after the loop terminates. Best-effort - never raises, returns True on 2xx, False otherwise. Adapters auto-stamp framework; workload_id and team are free-form labels that surface as filters in the dashboard. Pass include_per_iteration=False to send aggregate summary only.

endpoint and token are optional (v0.6.3+): with LOOPGAIN_TELEMETRY_ENDPOINT and LOOPGAIN_TELEMETRY_TOKEN exported, a bare lg.send_telemetry() is fully configured - the endpoint may be the receiver base URL (https://telemetry.loopgain.ai) or the full /v1/aggregate path. Nothing configured → returns False, sends nothing.

actual_dollars_spent and actual_dollars_saved are optional real-cost fields (v0.6.1+). Populate them only when you have a genuinely measured dollar figure - summed real API usage × list price, or an actually-executed paired-baseline comparison run. Never a formula-derived estimate. When populated, the dashboard displays your real number instead of its iter-count × $/iter extrapolation; passing an estimate through this field would present it as ground truth to every consumer of your tenant's data, not just you.

from loopgain import LoopGain

lg = LoopGain(target_error=0.1)
# ... run the loop ...
lg.send_telemetry(workload_id="my-rag-pipeline")  # endpoint/token from env (v0.6.3+)

Verify the pipeline before wiring a real loop

loopgain doctor runs a tiny in-process loop (no model calls, $0) and sends one test event:

export LOOPGAIN_TELEMETRY_ENDPOINT="https://telemetry.loopgain.ai"
export LOOPGAIN_TELEMETRY_TOKEN="lgk_..."
loopgain doctor
# -> event accepted by the receiver -> appears in your dashboard as 'loopgain-doctor'

Recommended setup

Store the token outside source. Two clean options:

# Option A: environment variable (simplest)
export LOOPGAIN_TELEMETRY_ENDPOINT="https://telemetry.loopgain.ai/v1/aggregate"
export LOOPGAIN_TELEMETRY_TOKEN="lgk_..."
# add to ~/.zshrc or ~/.bashrc

# Option B: macOS Keychain (more secure)
pip install keyring
python3 -c "import keyring; keyring.set_password('loopgain', 'telemetry', input('Token: '))"
# Then in code:
keyring.get_password('loopgain', 'telemetry')

What is sent

State transitions, Aβ summary (min/max/median), rollback flag, iterations used, savings, library version, optional opaque workload_id, threshold config, hour-bucketed timestamp - and, unless you pass include_per_iteration=False, a length-capped per-iteration trajectory (smoothed Aβ values and numeric error magnitudes; this is what drives the dashboard's convergence-profile scrubbing).

What is NEVER sent

Prompts, completions, error contents, the output buffer, or any customer identity beyond the bearer token. Numeric error magnitudes are sent (they're the loop-gain signal); error contents never are. Privacy contract is enforced by the payload-shape unit tests in tests/test_telemetry.py.

The hosted endpoint at telemetry.loopgain.ai is one acceptable destination. The receiver and dashboard are both open-source - self-host to keep telemetry fully under your control.

This is not the same as anonymous usage telemetry. send_telemetry sends your loop data to your dashboard, and only when you call it. There's a separate, opt-in funnel telemetry described below. The two never share data or code.

Anonymous funnel telemetry (opt-in)

LoopGain can report anonymous usage counts so a solo maintainer can tell whether the library is actually being used - install → first observe() → recurring use. It is opt-in and default-decline: nothing is sent unless you explicitly turn it on.

loopgain telemetry --show      # status + exactly what would be sent
loopgain telemetry --enable    # opt in (or: export LOOPGAIN_TELEMETRY=1)
loopgain telemetry --disable   # opt out (or: export LOOPGAIN_TELEMETRY=0)

DO_NOT_TRACK=1 is honored as a hard opt-out, and CI environments are auto-detected and declined silently.

When enabled, payloads carry only a locally-generated random id (not derived from your machine), hour-bucketed timestamps, library/Python/OS versions, the adapter in use, and a coarse outcome count. Prompts, outputs, error contents, keys, paths, and IPs are never collected. Delivery is batched, async, https-only, and fail-silent - it can never break your loop.

Full details and the privacy contract: TELEMETRY.md.

If LoopGain is useful to you, opting in is the cheapest way to support the project - these counts are the only signal a solo-maintained library has that it's working for anyone.

CLI

loopgain --version
# or:
loopgain version

loopgain telemetry --show      # inspect / control anonymous funnel telemetry
python -m loopgain telemetry --show   # equivalent, without the console script

Adapters

Thin wrappers under loopgain.integrations drive each major agent framework's iteration with a LoopGain monitor and auto-stamp framework=" " on telemetry. The frameworks themselves are optional dependencies - install the extra you need:

pip install 'loopgain[langgraph]'          # LangGraph
pip install 'loopgain[crewai]'            # CrewAI
pip install 'loopgain[autogen]'           # AutoGen v0.4+
pip install 'loopgain[langchain]'         # LangChain (create_agent / AgentExecutor)
pip install 'loopgain[openai-agents]'     # OpenAI Agents SDK
pip install 'loopgain[claude-agent-sdk]'  # Anthropic Claude Agent SDK
pip install 'loopgain[all]'               # all six

All adapters take a LoopGain instance plus an error_fn you provide - the framework doesn't know what your error signal is, so the adapter doesn't either. error_fn returns a non-negative number (or None to skip an iteration).

LangGraph

Drives graph.stream(input, stream_mode="updates"). Each update is one iteration.

from loopgain import LoopGain
from loopgain.integrations import LangGraphAdapter

graph = build_my_verify_revise_graph().compile()
lg = LoopGain(target_error=0.1, max_iterations=20)

adapter = LangGraphAdapter(
    lg=lg,
    error_fn=lambda update: len(update.get("verifier", {}).get("errors", [])),
)

final_state = adapter.run(graph, {"draft": initial})

lg.send_telemetry(
    endpoint=os.environ["LOOPGAIN_TELEMETRY_ENDPOINT"],
    token=os.environ["LOOPGAIN_TELEMETRY_TOKEN"],
    workload_id="rag-rewrite",
    framework=adapter.framework_name,  # "langgraph", auto-stamped
)

adapter.stream(...) yields each item if you want the full trace; adapter.arun(...) / adapter.astream(...) are the async counterparts and accept an async error_fn.

CrewAI

Installs step_callback and/or task_callback on a Crew. Pick whichever granularity matches your loop - step_error_fn for refinement within a Task, task_error_fn for refinement across Tasks.

from crewai import Crew
from loopgain import LoopGain
from loopgain.integrations import CrewAIAdapter

lg = LoopGain(target_error=0.1, max_iterations=20)

adapter = CrewAIAdapter(
    lg=lg,
    task_error_fn=lambda task_output: count_failed_checks(task_output.raw),
)

crew = Crew(agents=[...], tasks=[...])
adapter.install(crew)
result = crew.kickoff()
adapter.uninstall()
# or use `with CrewAIAdapter(...) as a:` context

lg.send_telemetry(
    endpoint=...,
    token=...,
    framework=adapter.framework_name,  # "crewai"
)

The adapter chains with any callback you already had installed - your existing instrumentation isn't overwritten.

AutoGen (v0.4+)

Wraps team.run_stream(task=...). In a verify-revise rotation, filter to the verifier's messages with observe_sources={"verifier"} so only it drives observe().

from autogen_agentchat.teams import RoundRobinGroupChat
from loopgain import LoopGain
from loopgain.integrations import AutoGenAdapter

team = RoundRobinGroupChat(participants=[generator, verifier])
lg = LoopGain(target_error=0.1, max_iterations=20)

adapter = AutoGenAdapter(
    lg=lg,
    error_fn=lambda msg: parse_verifier_score(msg.content),
    observe_sources={"verifier"},
)

result = await adapter.run(team, task="...")

lg.send_telemetry(
    endpoint=...,
    token=...,
    framework=adapter.framework_name,  # "autogen"
)

Pass a cancellation_token to adapter.run(...) and the adapter will cancel it when LoopGain reaches a terminal state (target met, oscillation, divergence). The legacy v0.2 ConversableAgent.initiate_chat API is not supported - use the v0.4 event-driven runtime.