Show HN: LoopGain — 用控制理论而非 max_iterations 停止智能体循环

一个面向 AI 智能体循环的开源成本控制器。

AI 智能体循环在不知何时停止时会浪费时间和金钱。LoopGain 会实时测量循环状况，在实际收敛时立即停止——并在质量下降前回滚，而不是一直运行到固定的 max_iterations 上限。

基准测试——10 个负载单元上的 2,000 次配对试验（自己运行）:

• 节省 92.8% 的 API 开销，相比 max_iter=20——总基准花费从 $27.05 降至 $1.94
• 约快 15 倍——每次试验的中位挂钟时间从 30.9s 降至 2.1s
• 质量得以保持，而非用速度换质量——在自然分布负载（W1–W4）上，评判胜率 0.50–0.63（多数单元 CI 排除零值），在人为失败负载（W5）上为 0.92–0.95；在 1,800 次评判比较中加权偏好为 0.678
• 六个终止准则均未触发（所有六个准则在运行前已预先注册阈值）

诚实的使用限制，开门见山： LoopGain 检测的是收敛，而非正确性——它知道何时再迭代也无济于事，而不是答案是否正确，且其效果仅取决于你提供的误差信号背后的验证器。完整的不适用场景列表 →

主页: loopgain.ai

适用于任何具有可度量误差信号的迭代 AI 工作流——验证-修订循环、精炼轮次、工具使用重试链、带自纠错的 RAG、带 linter 反馈的代码生成、多步推理循环。预置适配器支持 LangGraph、CrewAI、AutoGen、LangChain、OpenAI Agents SDK 和 Claude Agent SDK；可通过原始 API 直接嵌入任意自定义技术栈。纯 Python，无运行时依赖。

---

生产级智能体循环普遍使用 max_iterations=N 作为终止策略。这是智能体 AI 的尴尬默认设置：要么浪费计算资源（循环停得太晚），要么产出糟糕输出（循环停得太早）。LoopGain 用基于控制论的停止并回滚策略取代它，该策略立足于 Barkhausen 准则——来自电气工程反馈振荡器分析（1921 年）的基础成果。

---

纯 Python，无依赖，支持 Python 3.10+。

正在使用 Claude Code？ loopgain-plugin 会扫描整个仓库寻找可封装的循环——字面循环、递归循环、图循环和语义循环——并逐个文件提出经过审查的差异（绝不自动应用）：

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

---

三行代码即可封装任何具有可度量误差信号的迭代循环：

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)          # 误差最低的那个迭代的输出
print(result.iterations_used)
print(result.savings_vs_fixed_cap)

observe() 可接受数值误差量级或任意序列（其长度作为量级）。传入 output=... 可启用迄今为止最佳结果的缓冲。

---

定义你的误差信号

你唯一需要提供的是误差信号：每次迭代一个非负数，表示当前输出有多错误。越小越好；零表示完成。 LoopGain 不知道你的循环做什么——它只观察这个数字的轨迹，并决定是继续、停止还是回滚。

你的循环早已有某种方式知道输出还不够好（否则它不会继续修订）。把这个转成数字：

循环类型	误差信号 =
智能体编程（编写代码 → 运行测试）	不通过的测试数量（10 → 3 → 0）
JSON / 结构化提取	模式违反数
带自纠错的 RAG	仍缺失的必需事实数
用 LLM 评判的自精炼	评判结果的距离目标的差距（例如 10 − quality_score）
Lint / 格式循环	lint 错误数

唯一规则：非负，且随着输出变好而变小。直接返回原始问题列表即可——observe() 会使用其长度作为量级（例如，把不通过的测试列表传给它）。

如果你的质量标准模糊且没有自然的“零值”，请以 target_error=None 运行：此时 LoopGain 会在数值停止改进时停止，无论平台在哪里，而不是等待一个确切目标。

每一次停止/继续决策都基于这一个数字，因此 LoopGain 的效果仅取决于你给出的误差信号——选择一个真正能跟踪输出质量的信号。

---

LoopGain 在每次迭代时测量经验环路增益（Aβ = E(n) / E(n-1)），并将其作为平滑时间序列暴露出来以供可视化。然而，决策引擎会使用四个特征对完整误差轨迹进行分类：

E_ratio   = E_current / E_first      # 累计减少比例
slope_log = OLS slope of log10(E)    # 几何趋势方向
slope_p   = t-test p-value of slope  # 斜率的统计显著性
osc_std   = std of detrended log10(E) # 振荡幅度

它把轨迹归类为五种命名状态之一：

状态	条件	动作
FAST_CONVERGE	累计降至 E_first 的 ≤ 10%	继续
CONVERGING	负斜率且 p < 0.05，或累计 ≤ 50%	继续，留意上升漂移
STALLING	无显著斜率，无检测到振荡	连续两次读数后停止——返回迄今为止最佳结果
OSCILLATING	残差方差高，趋势平坦	停止——返回迄今为止最佳结果
DIVERGING	正斜率且 p < 0.05 且累计 > 110%	中止——回滚至迄今为止最佳结果

外加一个捷径：如果观测到的误差降到 target_error 或更低，循环立即停止，状态为 TARGET_MET。默认 target_error=0.0 在误差恰好为零时截断——这是验证器驱动循环的自然完成信号。传入 target_error=None 可禁用捷径，完全依赖稳定性检测。

决策设计上偏向保守：在终止前要求同时满足统计显著性和有意义的累积变化，以避免在嘈杂的真实 LLM 误差序列上发生误中止。在 5 种模式上以 N=1000 确定性模拟轨迹验证，宏平均准确率达 98.8%（参见 RESULTS_v2_classifier.md）。约 94% 的 STALLING 上限是 t 检验无法消除的 5% I 类错误率，而非分类器的弱点。

建议至少 6 次迭代才能保证可靠的趋势显著性。当 n≤4 时 t 检验功效严重不足（df=2 要求 |t|>4.3 才能 p<0.05）——证据不足时分类器会保守地退回到 STALLING。阈值是通过分析推导（控制论 + 统计惯例），而非拟合得到；一旦你有了生产轨迹，可通过 TrajectoryThresholds 参数按领域调整。

旧版单特征分类器： 最初的 v0.1 单 Aβ 带分类器（阈值 0.3 / 0.85 / 0.95 / 1.05）仍可通过 LoopGain(classifier='legacy_bands') 使用，供已经针对特定负载经验调整过带范围的使用者使用。

---

LoopGain 保留所有已观测输出及其对应误差分数的缓冲区。在终止时它返回 argmin(error)，而非最后迭代：

终止状态	返回的输出
TARGET_MET	当前输出（按定义为最佳）
OSCILLATING	缓冲区中误差最低的迭代
DIVERGING	误差最低的迭代（不是最后一个）

这将发散检测从“中止并出产垃圾”转变为“中止，返回迄今为止最佳”——一个免费的质量底线。

---

在集群中查看（可选仪表板）

该库本身就是完整产品，运行在本地——遥测是可选的且可自托管。如果你想在团队内纵览每个循环的稳定性、成本和回滚情况，有一个托管仪表板，数据由遥测接收器提供：

** 打开实时演示 → **——无需注册，真实基准数据。

接收器和仪表板都是开源的——自托管，以完全掌控遥测数据。

仓库	说明
loopgain	本库——Apache-2.0 控制循环（你在这里）
telemetry-receiver	摄取匿名化循环遥测的 Cloudflare Worker
dashboard	集群仪表板——可自托管
loopgain-bench	上述数字背后可重现的 2,000 次试验基准
loopgain-plugin	Claude Code 插件——扫描仓库寻找可封装循环，提出审查后的差异

---

LoopGain 能保证什么，不能保证什么

LoopGain 在循环停止改进时停止，从而节省金钱——更少迭代、更少 token。在我们的公开基准中，相比 max_iterations=20，总 API 开销减少 92.8%，且输出质量得以保持。两个诚实限制：

• 节省幅度取决于你的负载。 通常快速成功的循环节省最多（约 96%）；对抗性、易失败的循环节省较少（约 78–84%）。总体数字是混合结果——在引用数字前，请用你自己的循环运行基准。
• LoopGain 检测的是收敛，而非正确性。 它在你的误差信号停止改进时停止——这意味着再迭代也无济于事，而不是循环成功。在基准上这保持了质量（它很少在更差的输出上提前停止；误停止率 ≤4.5%），但循环可能在误差仍高于零时停滞——例如，还剩下 2 个不通过的测试。因此在信任输出之前，请检查 result.best_error（或你自己的通过/失败标准）：如果它在达到你的目标前平台化，那是 LoopGain 无法看到的质量差距，而迫使重新运行的错误停止是唯一会侵蚀节省的情况。LoopGain 决定何时停止；你决定答案是否足够好。
• LoopGain 的准确性仅取决于你的验证器。 它依据你提供的误差信号行动。如果你的验证器报告零错误，LoopGain 会信任并停止——因此一个存在盲区的验证器可能在一个仍然错误的答案上报告成功，LoopGain 就会自信地停在那里。这并非上述平台化情况：误差读数为零，循环看起来是干净的成功，因此无论是 LoopGain 还是其收敛信号都无法标记。停止的质量受限于你误差信号背后检查的质量。我们在基准的代码生成负载上测量过：4.5% 的收敛运行（16/355）通过了循环运行的每一项检查，却未能通过完整的保留测试套件——而那是一个下限，而非上限，因为循环内的验证器很强；更弱的验证器会暴露更多。（与上述 ≤4.5% 的误停止率不同——数字巧合，失败模式不同。）为 LoopGain 配备你能在停止时负担得起的、最强的验证器——对样本子集的可执行测试、模式或类型检查而非主观感受、循环未优化过的保留检查。如何设计强验证器的现场指南 正应对此问题。

---

LoopGain(target_error=0.0, max_iterations=50, thresholds=None, trajectory_thresholds=None, classifier='trajectory', smoothing_window=3, assumed_fixed_cap=10)

构建监控器。

• target_error — 当观测误差降至该值或以下时停止。默认 0.0 在误差恰好为零时截断（验证器驱动循环的自然完成信号）。传入 None 完全禁用截断。
• max_iterations — 硬安全后盾。默认 50，这样循环永不会无限运行；稳定性判决通常远在此之前终止。传入 None 可完全解除限制（仅当你的循环保证能达到 target_error 或停止状态时才安全），或传入更小的整数以更紧约束。
• thresholds — 为旧版单 Aβ 带分类器自定义 ThresholdBands。当 classifier='trajectory' 时忽略。
• trajectory_thresholds — 为多特征分类器（默认）自定义 TrajectoryThresholds。仅在有负载特定证据时覆盖。
• classifier — 要使用的分类器。可选 'trajectory'（默认）或 'legacy_bands'。
• smoothing_window — 用于环路增益计算的移动平均窗口大小。默认 3。
• assumed_fixed_cap — 用于节省计算的对比固定迭代次数。默认 10。显示 savings_vs_fixed_cap 时使用。

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 forLangGraph, 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).

---

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.

---

Defining your error signal

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:

Loop	Error signal =
Agentic coding (write code → run tests)	number of failing tests (10 → 3 → 0)
JSON / structured extraction	number of schema violations
RAG with self-correction	number of required facts still missing
Self-refinement with an LLM judge	judge's gap to target (e.g. 10 − quality_score)
Lint / format loop	lint error count

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.

---

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:

State	Condition	Action
FAST_CONVERGE	cumulative reduction to ≤ 10% of E_first	Continue
CONVERGING	negative slope with p < 0.05, OR cumulative ≤ 50%	Continue, watch for upward drift
STALLING	no significant slope, no detectable oscillation	Stop after 2 consecutive readings — return best-so-far
OSCILLATING	high residual variance with flat trend	Stop — return best-so-far
DIVERGING	positive slope with p < 0.05 AND cumulative > 110%	Abort — roll back to best-so-far

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) — the classifier conservatively falls back to STALLING when evidence is thin. The thresholds are derived analytically (control theory + statistical convention), not fitted; tune them per domain via the TrajectoryThresholds argument once you have production traces.

Legacy single-feature classifier: the original v0.1 single-Aβ-band classifier (thresholds 0.3 / 0.85 / 0.95 / 1.05) is still available via LoopGain(classifier='legacy_bands') for callers that have empirically tuned the bands to a specific workload.

---

LoopGain keeps a buffer of all observed outputs paired with their error scores. On termination it returns argmin(error), not the last iteration:

Terminal state	Returned output
TARGET_MET	Current output (by definition, the best)
OSCILLATING	Lowest-error iteration in the buffer
DIVERGING	Lowest-error iteration (which is not the last one)

This transforms divergence detection from "abort with garbage" into "abort with the best you've seen so far" — a free quality floor.

---

See it across a fleet (optional dashboard)

The library is the whole product locally — telemetry is opt-in and self-hostable. If you want a fleet view of every loop's stability, cost, and rollbacks across a team, there's a hosted dashboard fed by the telemetry receiver:

Open the live demo → — no signup, real benchmark data.

The receiver and dashboard are both open-source — self-host to keep telemetry entirely under your control.

Repo	What it is
loopgain	This library — the Apache-2.0 control loop (you are here)
telemetry-receiver	Cloudflare Worker that ingests anonymized loop telemetry
dashboard	The fleet dashboard — self-hostable
loopgain-bench	The reproducible 2,000-trial benchmark behind the numbers above
loopgain-plugin	Claude Code plugin — scans a repo for wrappable loops, proposes reviewed diffs

---

What LoopGain does and doesn't guarantee

LoopGain saves money by stopping a loop once it stops improving — fewer iterations, fewer tokens. In our public benchmark, that was a 92.8% cut in total API spend vs max_iterations=20, with output quality preserved. Two honest limits:

• Savings depend on your workload. Loops that usually succeed fast save the most (~96%); adversarial, failure-prone loops save less (~78–84%). The headline is a blend — run the benchmark on your own loops before quoting a number.
• LoopGain detects convergence, not correctness. It stops when your error signal stops improving — which means more iterations won't help, not that the loop succeeded. On the benchmark this preserved quality (it rarely stopped early on a worse output; false-stop rate ≤4.5%), but a loop can stall with the error still above zero — a plateau at, say, 2 failing tests. So check result.best_error (or your own pass/fail) before you trust the output: if it plateaued short of your target, that's a quality gap LoopGain can't see, and a false stop that forces a rerun is the one way it eats into the savings. LoopGain decides when to stop ; you decide whether the answer is good enough.
• LoopGain is only as right as your verifier. It acts on the error signal you give it. If your verifier reports zero errors, LoopGain trusts that and stops — so a verifier with blind spots can report success on an answer that is still wrong, and LoopGain will confidently stop there. This is not the plateau case above: the error reads zero and the loop looks like a clean success, so neither LoopGain nor its convergence signal can flag it. The quality of the stop is bounded by the quality of the check behind your error signal. We measured this on the benchmark's code-gen workload: 4.5% of converged runs (16/355) passed every check the loop ran but failed the full held-out test suite — and that's a floor, not a ceiling, because the in-loop verifier there was strong; a weaker verifier exposes more. (Distinct from the ≤4.5% false-stop rate above — the numbers coincide, the failure modes don't.) Pair LoopGain with the strongest verifier you can afford at the stop — executable tests over a sampled subset, a schema or type check over a vibe, a held-out check the loop didn't optimize against. How to design a strong verifier is a field guide to exactly this.

---

LoopGain(target_error=0.0, max_iterations=50, thresholds=None, trajectory_thresholds=None, classifier='trajectory', smoothing_window=3, assumed_fixed_cap=10)

Construct the monitor.

• target_error — Stop when an observed error drops at or below this. Default 0.0 short-circuits on exactly zero error (the natural completion signal for verifier-driven loops). Pass None to disable the short-circuit entirely.
• max_iterations — Hard safety backstop. Default 50 so the loop can never run unbounded; a stability verdict normally terminates it well before this. Pass None to opt into a fully unbounded loop (only safe if your loop is guaranteed to reach target_error or a stop-state), or a smaller integer to cap tighter.
• thresholds — Custom ThresholdBands for the legacy single-Aβ-band classifier. Ignored when classifier='trajectory'.
• trajectory_thresholds — Custom TrajectoryThresholds for the multi-feature classifier (the default). Override only with workload-specific evidence.
• classifier — 'trajectory' (default, v0.2 multi-feature classifier) or 'legacy_bands' (v0.1 single-Aβ-band classifier).
• smoothing_window — EMA window for the smoothed Aβ series (always maintained for visualization, regardless of classifier choice). Default 3.
• assumed_fixed_cap — Used to compute savings_vs_fixed_cap. Default 10.

lg.observe(errors, output=None) -> str

Record this iteration's errors and optional output. Returns the current state name. errors accepts a number (used directly) or any sequence (length used as magnitude).

lg.should_continue() -> bool

Returns False once a terminal state fires.

Current state name. One of INIT, FAST_CONVERGE, CONVERGING, STALLING, OSCILLATING, DIVERGING, TARGET_MET, MAX_ITERATIONS. The corresponding terminal result.outcome values are converged, oscillating, diverged, stalled (v0.2 trajectory mode only — STALLING terminating after 2 consecutive readings), max_iterations, or in_progress.

lg.result -> LoopGainResult

Terminal result with outcome, iterations_used, best_index, best_output, best_error, convergence_profile, error_history, savings_vs_fixed_cap. Safe to call mid-loop.

lg.send_telemetry(endpoint=None, token=None, workload_id=None, timeout=2.0, allow_insecure=False, framework=None, loop_type=None, team=None, include_per_iteration=True, retries=2, retry_backoff=0.25, actual_dollars_spent=None, actual_dollars_saved=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; loop_type 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 x 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 x $/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, off by default)

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.

---

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

---

Thin wrappers under loopgain.integrations drive each major agent framework's iteration with a LoopGain monitor and auto-stamp framework="<name>" 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).

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.

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.

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.

Duck-types against any LangChain agent that exposes .stream(input, **kwargs) / .astream(input, **kwargs) — both the current langchain.agents.create_agent() (v1+) and the legacy AgentExecutor. The adapter forwards **stream_kwargs verbatim, so the chunk shape your error_fn sees is the one your agent emits.

from langchain.agents import create_agent
from loopgain import LoopGain
from loopgain.integrations import LangChainAdapter

agent = create_agent(model="gpt-5-nano", tools=[get_weather])
lg = LoopGain(target_error=0.0, max_iterations=20)

def error_fn(chunk):
    if chunk.get("type") != "updates":
        return None
    # Count unresolved tool calls; drops to 0 once the agent stops calling tools.
    return sum(
        1 for _, update in chunk["data"].items()
        if getattr(update.get("messages", [None])[-1], "tool_calls", None)
    )

adapter = LangChainAdapter(lg=lg, error_fn=error_fn)
final = adapter.run(
    agent,
    {"messages": [{"role": "user", "content": "What's the weather?"}]},
    stream_mode="updates",
    version="v2",
)

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

For legacy AgentExecutor: just drop the stream_mode / version kwargs; each yielded chunk is an AddableDict per step (parse intermediate_steps or the terminal output key in your error_fn).

Wraps Runner.run_streamed(agent, input).stream_events(). The SDK is async-first; the adapter mirrors that. A run_sync helper wraps the async path with asyncio.run for synchronous callers.

from agents import Agent, function_tool
from loopgain import LoopGain
from loopgain.integrations import OpenAIAgentsAdapter

agent = Agent(name="Reviser", instructions="...", tools=[...])

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

def error_fn(event):
    # Default observes only run_item_stream_event; pull the verifier's
    # reported failure count off tool outputs.
    if event.item.type == "tool_call_output_item":
        return float(event.item.output.get("failures", 0))
    return None

adapter = OpenAIAgentsAdapter(lg=lg, error_fn=error_fn)
result = await adapter.run(agent, input="Fix the bug.")
print(result.final_output)

lg.send_telemetry(
    endpoint=...,
    token=...,
    framework=adapter.framework_name,        # "openai-agents"
)

By default the adapter only forwards run_item_stream_event to error_fn — pass observe_event_types=None to see every event (including raw token deltas and agent-handoff notifications). When LoopGain reaches a terminal state, the adapter best-effort calls .cancel() on the underlying RunResultStreaming.

Wraps Anthropic's claude_agent_sdk.query(prompt=..., options=...) async iterator. By default observes only AssistantMessage (skips UserMessage / SystemMessage / ResultMessage); override with observe_message_types=None or a custom tuple.

from claude_agent_sdk import ClaudeAgentOptions, TextBlock
from loopgain import LoopGain
from loopgain.integrations import ClaudeAgentSDKAdapter

def error_fn(message):
    # Count `FAIL:` markers a self-verifying persona emits.
    for block in getattr(message, "content", []):
        if isinstance(block, TextBlock):
            return float(block.text.count("FAIL:"))
    return None

lg = LoopGain(target_error=0.0, max_iterations=20)
adapter = ClaudeAgentSDKAdapter(lg=lg, error_fn=error_fn)

options = ClaudeAgentOptions(system_prompt="Self-verify each draft.")
result = await adapter.run(
    prompt="Write a haiku about feedback loops.",
    options=options,
)

lg.send_telemetry(
    endpoint=...,
    token=...,
    framework=adapter.framework_name,        # "claude-agent-sdk"
)

For the bidirectional ClaudeSDKClient use case, pass message_iterator=client.receive_messages() instead of prompt=....

For frameworks without an adapter, the raw LoopGain.observe() API works against any iterable. The adapters are 100-200 lines each — copy one of loopgain/integrations/{langgraph,crewai,autogen,langchain,openai_agents,claude_agent_sdk}.py as a starting point.

---

Initial public release. Core library shipped (current version: see the PyPI badge at the top). Framework adapters (LangGraph, CrewAI, AutoGen, LangChain, OpenAI Agents SDK, Claude Agent SDK) are installable as optional extras. The cloud-aggregator telemetry receiver and dashboard are live as separate open-source repos. The math and the API surface are stable.

This is alpha software. The API may break before 1.0 if production usage surfaces design issues; pin the version.

---

Apache-2.0.

---

LoopGain applies the Barkhausen stability criterion (Heinrich Barkhausen, 1921 — the foundational result on when feedback amplifiers oscillate) to AI agent feedback loops. The criterion was originally a way to predict whether an electronic oscillator would sustain oscillation; it turns out to map cleanly onto any feedback loop you can attach an error signal to.

The cleanest summary: an iterative AI loop with a measurable error signal is a feedback system. The ratio E(n) / E(n-1) is its empirical loop gain. The Barkhausen result tells you that loop gain less than 1 converges, equal to 1 oscillates, greater than 1 diverges. LoopGain operationalizes this: classifies the loop's current band, and decides what to do — stop, continue, or roll back to the best output seen so far.

Loop types this applies to in practice:

• Verify-revise loops (GVR pattern) — generator produces, verifier finds issues, reviser fixes. Error = issue count or severity-weighted score.
• Refinement loops — initial output, iterate to improve. Error = distance from target spec / rubric score.
• Tool-use retry chains — agent calls tool, gets back error/success, retries. Error = consecutive failure count or aggregate score.
• RAG with self-correction — retrieve, generate, critique, re-retrieve. Error = critique severity or hallucination score.
• Code generation with linter/test feedback — generate, run tests/linter, fix, repeat. Error = failing test count or linter violation count.
• Multi-step reasoning loops — ReAct-style think/act/observe iterations. Error = whatever the agent's quality assessor returns.
• Custom feedback loops — anything where you can produce a number that should drop toward zero as the loop succeeds.