DEV Community

Cover image for Preventing the Agent Death Spiral: Why Your AI Agents Need a Circuit Breaker
Debashish Ghosal
Debashish Ghosal

Posted on

Preventing the Agent Death Spiral: Why Your AI Agents Need a Circuit Breaker

Preventing the Agent Death Spiral: Why Your AI Agents Need a Circuit Breaker

Repo: github.com/deghosal-2026/ai-loopguard · Install: pip install ai-loopguard · MIT · 436 tests · 97.3% coverage


TL;DR

AI agents get stuck in loops — repeating errors, oscillating on test failures, returning bad schema. Tokens burn. Nothing improves. A $0.001 call that loops 10× costs more than a $0.05 call that solves it once.

ai-loopguard is a drop-in circuit breaker for LLM agent loops. It detects stuck patterns and escalates to a stronger model before costs spiral. 4 lines of code. Works with LangGraph, CrewAI, or raw Python.

from ai_loopguard import Guard

guard = Guard(escalation_model=gpt4_model, workhorse_model_name="qwen2.5-coder")

@guard.protect
def agent_step(state):
    return cheap_model.generate(state)
Enter fullscreen mode Exit fullscreen mode

What you get:

  • 4 trigger types (test failures, repeated errors, schema violations, custom callbacks)
  • Auto-escalation to a stronger model OR human-in-the-loop interrupt via LangGraph's native interrupt()
  • Prompt injection sanitization + secret redaction before sending context to the cloud model
  • Cost per completed task (not per call), escalation rate, routing vs failover metrics
  • JSONL logs + OpenTelemetry spans + CLI analyzer
  • Fail-open by design — loopguard never crashes your agent

Results: 100% task success rate in field study (15 tasks, 3 repos, 0/15 baseline → 15/15 guarded). 2.75 µs detection overhead. 436 tests. 97.3% coverage. MIT.

Links: repo · PRD · SPEC · field study · CONTRIBUTING.md


👉 Visit the repo: github.com/deghosal-2026/ai-loopguard

If this solves a problem you've been hitting, give it a star — it helps other engineers find it. Open an issue, drop a comment, or contribute a trigger, integration, or CLI improvement. The best ideas will come from people running agents in production.


The production nightmare every AI engineer will face

It's 2 AM. Your phone buzzes. The on-call alert says your AI agent pipeline has been running for 6 hours on a task that should take 2 minutes.

You open the logs. The agent changed code. Tests failed. It changed something else. A different test failed. It reverted. The original test failed again. 500 iterations. Zero progress.

Your LLM API bill for this one task: $200. For a task that didn't complete.

This isn't a hypothetical. It's the rite of passage for every engineer building agentic systems. Non-deterministic systems (agents) interacting with deterministic environments (code, APIs, databases) fall into "stuck states." And when they do, they don't just waste time — they burn real money.

Here's the dirty secret of cheap AI models: they're cheap per call, but expensive per completed task. A $0.001 call that loops 10 times costs more than a $0.05 call that solves it once.

In traditional software, we have a pattern for this. Circuit breakers prevent cascading failures in distributed systems. When a service is failing, the circuit opens, traffic stops, and the system degrades gracefully instead of spiraling.

Why are we deploying AI agents into production without them?

I looked for a library that would detect stuck agent loops and escalate to a stronger model. Here's what I found:

What exists What it does What's missing
LangGraph, CrewAI, AutoGen Agent frameworks No failure detection. No escalation.
LangSmith, LangFuse Observability dashboards After-the-fact. No control loop. No intervention.
agentcost-sdk, compute-cfo Per-call cost tracking No stuck-loop detection. No escalation.
Overseer Full quality framework Replaces your entire stack. Not drop-in.

The missing middle: nobody sits between frameworks and cost trackers — a quality layer that drops into whatever you've already built.

So I built one. Full PRD. Full SPEC. Full WBS. 12 milestones. 436 tests. 97.3% coverage. Not vibe-coding — big-tech SDLC applied to an open-source library.


The core problem: detection vs. resolution

Building a circuit breaker for AI agents is two hard problems, not one.

Problem 1: How do you detect a loop in a non-deterministic system?

The naive approach is a counter: max_iterations=10. Kill the agent after 10 steps. But that's a blunt instrument. It doesn't tell you why or when the agent got stuck. It can't distinguish between an agent making progress on a hard task and an agent spinning in circles. And it certainly can't detect the most insidious pattern: oscillation.

Oscillation is when the agent bounces between two states. Test passes. Agent changes code. Test fails. Agent reverts. Test passes. Agent changes the same code. Test fails. The iteration counter goes up, but the agent is going nowhere.

loopguard uses pattern detection strategies — not counters:

Detection strategy What it catches Why it's better than a counter
Error clustering Same exception type + message across N consecutive steps Catches ValueError: rate limited repeated 3× — the agent isn't learning from the error
Test-failure cycles Same test fails N consecutive times, or oscillates pass→fail→pass→fail Catches the agent breaking test_parser while trying to fix test_lint — it's thrashing, not progressing
Schema validation failures Output fails schema validation N consecutive times Catches the agent returning malformed JSON when you need {"answer": str, "citations": list} — it's not understanding the contract
Custom callbacks Your callback returns True — fires instantly Wall-clock timeout, token budget exceeded, your own domain heuristics

Each trigger has a configurable threshold (max_retries), is evaluated on every step, and fires deterministically — no self-grading, no "let the model decide if it's stuck." Self-grading is unreliable. Triggers are deterministic or user-defined.

Priority order is fixed: customrepeated_errortest_failureschema_invalid. First trigger to fire wins. You can't accidentally reorder them by shuffling your config.

Problem 2: What do you do when a loop is detected?

This is where loopguard stands out. Most tools just crash the application or log a warning. loopguard handles it gracefully with an escalation layer:

Model upgrade — Instead of failing, loopguard packages the context (what was tried, what failed, error traces), sanitizes it, and escalates to a stronger model that can reason its way out of the trap. Your cheap workhorse model couldn't solve it. Your frontier model can.

Human-in-the-loop — For cost-sensitive workflows, loopguard can pause execution and ask: "Agent is stuck on test_parser (3 failures). Escalate to GPT-4? [y/n]" — you confirm before a single token is spent on escalation.

Fail-open — If loopguard itself fails (detection bug, escalation model down, network error), your agent keeps running. Three configurable modes: re-raise the original exception, return the last successful output, or return a sentinel value. The circuit breaker never takes down your system.


Show me the code

Developers skip text to look for code. Here's how "drop-in" it actually is:

from ai_loopguard import Guard

guard = Guard(
    escalation_model=gpt4_model,
    workhorse_model_name="qwen2.5-coder",
    triggers={
        "test_failure": {"max_retries": 3},
        "repeated_error": {"max_retries": 3},
        "schema_invalid": {"max_retries": 3},
    },
)

@guard.protect
def agent_step(state):
    output = cheap_model.generate(state)
    guard.record_test_results(run_tests(output))
    return output
Enter fullscreen mode Exit fullscreen mode

If agent_step fails 3 times, loopguard:

  1. 📦 Packages the context — what was tried, what failed, error traces
  2. 🛡️ Sanitizes against prompt injection (system/user separation, delimiters)
  3. 🔒 Redacts sensitive fields (API keys, tokens, PII)
  4. 🚀 Escalates to your configured cloud model
  5. 📊 Logs the event with full cost impact
  6. ✅ Returns the escalated output — the loop continues

4 lines of code to add. Zero refactoring. You don't rewrite your agent core logic. You wrap it.


LangGraph integration — wired into the graph lifecycle

If you're building agents with LangGraph, loopguard is designed to feel native. Not bolted on — integrated into the graph's node lifecycle.

pip install ai-loopguard[langgraph]
Enter fullscreen mode Exit fullscreen mode
from ai_loopguard import Guard
from ai_loopguard.integrations.langgraph import LangGraphHandler

guard = Guard(
    escalation_model=gpt4,
    workhorse_model_name="qwen2.5-coder",
    on_escalate="auto",
    max_escalations_per_run=2,
)
handler = LangGraphHandler(guard)

def generate_node(state):
    try:
        output = workhorse.invoke(state["messages"]).content
    except Exception as exc:
        return handler.on_step_end(state["step"], state, None, error=exc)
    return handler.on_step_end(state["step"], state, output)

def tests_node(state):
    guard.record_test_results(run_tests(state["output"]))
    return handler.on_step_end(state["step"] + 1, state, state["output"])
Enter fullscreen mode Exit fullscreen mode

Why this is different from LangGraph's built-in retry

LangGraph gives you DAGs, conditional edges, parallel stages, and approval gates. It does NOT give you:

  • Stuck-loop detection — LangGraph will happily run your node 100 times if your edge condition keeps routing back
  • Test-failure cycle detection — pass → fail → pass → fail oscillation on the same test
  • Automatic escalation — switching to a stronger model when the cheap one is stuck
  • Cost-per-completed-task tracking — the real economic metric, not per-call

loopguard adds all four. And it uses LangGraph's own interrupt() for human-in-the-loop escalation — so graph state is preserved in the checkpointer, not lost:

guard = Guard(escalation_model=gpt4, on_escalate="interrupt")
Enter fullscreen mode Exit fullscreen mode

When a trigger fires, loopguard calls LangGraph's native interrupt():

"Agent stuck (test_failure: test_parser failed 3 consecutive times).
 Escalate to gpt-4o? [y/n]"
Enter fullscreen mode Exit fullscreen mode

You confirm → Command(resume=escalated) → the graph continues with the escalated output. You decline → original output returned. The graph's edge condition routes normally from there.

Full LangGraph example

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from ai_loopguard import Guard
from ai_loopguard.integrations.langgraph import LangGraphHandler

workhorse = ChatOpenAI(model="qwen2.5-coder", temperature=0)
gpt4 = ChatOpenAI(model="gpt-4o", temperature=0)

guard = Guard(
    escalation_model=gpt4,
    workhorse_model_name="qwen2.5-coder",
    on_escalate="auto",
    max_escalations_per_run=2,
)
handler = LangGraphHandler(guard)

class State(TypedDict):
    step: int
    messages: list[str]
    output: str

def generate_node(state: State) -> dict:
    try:
        output = workhorse.invoke(state["messages"]).content
    except Exception as exc:
        return handler.on_step_end(state["step"], state, None, error=exc)
    return handler.on_step_end(state["step"], state, output)

def tests_node(state: State) -> dict:
    results = run_tests(state["output"])
    guard.record_test_results(results)
    return handler.on_step_end(state["step"] + 1, state, state["output"])

builder = StateGraph(State)
builder.add_node("generate", generate_node)
builder.add_node("tests", tests_node)
builder.set_entry_point("generate")
builder.add_edge("generate", "tests")
builder.add_conditional_edges("tests", lambda s: "generate" if not all(s.get("output", "")) else "__end__")
graph = builder.compile()

result = graph.invoke({"step": 0, "messages": [...], "output": ""})
Enter fullscreen mode Exit fullscreen mode

The handler holds no state — all step history lives on the guard's shared GuardState. Call guard.reset() before invoking the graph for an independent task so trigger history doesn't bleed across runs.


CrewAI integration — per-agent isolation for multi-agent crews

If you're running CrewAI crews with multiple agents, loopguard monitors each agent independently. Different agents can hit different triggers simultaneously — and escalate separately.

pip install ai-loopguard[crewai]
Enter fullscreen mode Exit fullscreen mode
from ai_loopguard import Guard
from ai_loopguard.integrations.crewai import CrewAIWrapper

guard = Guard(escalation_model=gpt4, workhorse_model_name="qwen")
wrapper = CrewAIWrapper(guard)
crew = wrapper.wrap(crew)
result = crew.kickoff()
Enter fullscreen mode Exit fullscreen mode

Why per-agent state matters

Without per-agent isolation, steps from different agents accumulate in a single history. A repeated_error trigger would see Agent A's ValueError, then Agent B's different KeyError, and fire on a false pattern.

loopguard's CrewAIWrapper gives each agent its own GuardState, keyed by role. The detector is shared (it only holds config), but it evaluates each agent's history independently. The Researcher can loop on a retrieval error while the Writer loops on malformed JSON — both trigger, both escalate, neither interferes with the other.

Full CrewAI example

from langchain_openai import ChatOpenAI
from crewai import Agent, Crew, Task
from ai_loopguard import Guard
from ai_loopguard.integrations.crewai import CrewAIWrapper

workhorse = ChatOpenAI(model="qwen2.5-coder", temperature=0)
gpt4 = ChatOpenAI(model="gpt-4o", temperature=0)

guard = Guard(
    escalation_model=gpt4,
    workhorse_model_name="qwen2.5-coder",
    on_escalate="auto",
    max_escalations_per_run=2,
)
wrapper = CrewAIWrapper(guard)

researcher = Agent(
    role="Researcher",
    goal="Find sources for the topic",
    backstory="A meticulous research analyst.",
    llm=workhorse,
)
writer = Agent(
    role="Writer",
    goal="Draft the article from the research",
    backstory="A concise technical writer.",
    llm=workhorse,
)

research_task = Task(
    description="Find 3 authoritative sources on {topic}.",
    expected_output="A list of 3 sources with URLs.",
    agent=researcher,
)
write_task = Task(
    description="Write a 500-word article using the research.",
    expected_output="A 500-word markdown article.",
    agent=writer,
)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
crew = wrapper.wrap(crew)
result = crew.kickoff(inputs={"topic": "vector databases"})
Enter fullscreen mode Exit fullscreen mode

If the Researcher loops on a retrieval error 3 times, the repeated_error trigger fires for the Researcher only. The Writer's history is untouched. The escalation output replaces that agent's step — the crew continues.

Note: CrewAI ships frequent breaking changes. The wrapper targets crewai>=0.50.0,<0.60.0 and is marked experimental. Pin your version and re-test on upgrade.


Security — because you're sending failed agent output to a cloud model

This is the part most teams skip. When your agent gets stuck, its state might contain:

  • API keys from environment variables
  • User PII from the task it was processing
  • Malicious instructions from a web page it scraped (prompt injection)

loopguard treats all agent output as untrusted and does three things before sending anything to the escalation model:

1. Context compression

Not a full dump. First + last + summary of steps, bounded by max_context_tokens (default 4000). You control the token budget.

2. Sanitization (prompt injection defense)

Clear system vs user separation. Delimiters around agent-produced content. The escalation model knows what's instruction vs. what's data.

3. Redaction (secret/PII leakage defense)

Built-in patterns for OpenAI keys, AWS keys, GitHub tokens. Plus your own regex and field names:

guard = Guard(
    escalation_model=gpt4_model,
    config=GuardConfig(
        redact_patterns=[r"sk-[a-zA-Z0-9]{48}"],
        redact_fields=["api_key", "password", "ssn"],
        sanitize_context=True,  # default
    ),
)
Enter fullscreen mode Exit fullscreen mode

Every EscalationEvent log includes sanitized: bool and redacted_fields: list[str] — you can audit what was stripped.

Full threat model (6 threats, all mitigated)

ID Threat Impact Mitigation
T1 Prompt injection via escalation context High — escalation model hijacked Sanitize: system/user separation, delimiters
T2 Secret/PII leakage to cloud model High — data breach, credential exposure Configurable redaction (regex + field names)
T3 Unbounded failure history → memory exhaustion Medium — DoS, crash max_history_steps hard cap (default 100)
T4 Escalation model down or compromised Medium — escalation fails Fail-open: return last output, log the event
T5 Supply chain compromise Medium — code execution Pinned deps, pip-audit in CI
T6 Economic DoS — runaway escalations Medium — unexpected billing max_escalations_per_run cap (default 1)

Threat T6 is the one nobody thinks about. If your agent loop is buggy and loopguard escalates on every step, you could rack up $50 before you notice. The cap prevents that — one escalation per guarded execution by default, configurable up to 10.


Architectural benefits for engineering teams

Cost predictability — hard financial boundaries on agent exploration

Most cost trackers tell you cost per call. That's the wrong metric.

A task that loops 10 times at $0.001/call costs $0.01. A task that succeeds once at $0.05/call costs $0.05. The "cheap" model is 5× more expensive per completed task.

loopguard tracks three metrics that actually matter:

Metric What it tells you Why it matters
Escalation rate escalations / total_guarded_calls If 80% escalate, you haven't built a router — you've built a system that pays for a cheap attempt before every expensive call
Cost per completed task retries + failed loops + escalation calls The real number. Not per-call — per-task-that-actually-got-done
Routing vs failover Intentional escalation vs availability failure Routing = model couldn't do it. Failover = model was down. Different problems, different fixes

CI/CD safety — automated agent pipelines don't hang and block deploys

If you're running AI agents in your CI/CD pipeline — automated PR reviews, code generation, test fixing — a stuck agent doesn't just waste tokens. It blocks the deployment pipeline. Your entire release waits for an agent that's spinning in circles.

loopguard detects the stuck pattern in milliseconds, escalates, and returns. The pipeline keeps moving. And the max_escalations_per_run cap ensures that even in the worst case, the cost is bounded.

Telemetry and observability — know why your agent broke down

Every escalation event is logged as structured JSONL with: trigger type, retry count, models involved, cost impact, routing vs failover category, sanitized status, redacted fields. You get the full forensic picture of why the agent got stuck.

loopguard analyze logs.jsonl --summary
Enter fullscreen mode Exit fullscreen mode
==================================================
loopguard escalation summary
==================================================
Total escalations:       42
Total cost (USD):        $1.2340

Cost breakdown by trigger:
  repeated_error            20  $0.5600
  test_failure              15  $0.4500
  schema_invalid             7  $0.2240
Enter fullscreen mode Exit fullscreen mode

OpenTelemetry — wire it to your existing dashboards

from ai_loopguard.integrations.otel import OTelEventHook

guard = Guard(escalation_model=gpt4_model)
guard.logger.register_hook(OTelEventHook())
Enter fullscreen mode Exit fullscreen mode

Every escalation emits a span with GenAI semantic conventions. Pick it up in Grafana, Datadog, Jaeger — wherever your OTel collector sends spans. Set an alert: "Escalation rate > 30% → investigate routing config."


Fail-open — the circuit breaker never takes down your system

If loopguard itself fails — detection bug, escalation model down, network error — your agent keeps running.

Mode What happens
raise_original (default) Re-raise the original exception. Agent behaves as if loopguard isn't there.
return_last_output Return the last successful output. Agent continues with stale but valid data.
return_sentinel Return a configurable sentinel. Agent knows something went wrong.

Every fail-open event is logged. You'll know loopguard failed even if your agent didn't.


Field study — does it actually work?

15 tasks across 3 external open-source repos: SWE-agent (~14k★), Aider (~46k★), LangGraph (~100k★). Same tasks with and without loopguard.

Metric Without loopguard With loopguard Target
Task success rate 0% (0/15) 100% (15/15) >70% ✅
Total time 923ms 458ms (50% faster) flat or better ✅
Escalation success rate 100% (15/15) >70% ✅
Lines of code to add 4.0 avg per task <10 ✅

Every task that failed without loopguard succeeded with it. Every escalation produced a valid result. Average integration: 4 lines of code.

Repo Tasks Baseline Guarded Triggers fired
SWE-agent (~14k★) 5 0/5 ❌ 5/5 ✅ test_failure, repeated_error, schema_invalid, oscillation
Aider (~46k★) 5 0/5 ❌ 5/5 ✅ test_failure, repeated_error, schema_invalid, oscillation
LangGraph (~100k★) 5 0/5 ❌ 5/5 ✅ repeated_error, schema_invalid, custom (timeout), test_failure

Full report: docs/field-study/report.md


Performance benchmarks — you won't notice it's there

7 benchmarks. All passing. All with massive headroom.

Benchmark Median Target Headroom
Detection overhead 2.75 µs 1 ms 363× under target
Context packaging 10.25 µs 100 ms 9,756× under target
Import time 54.8 ms 200 ms 3.6× under target
Escalation full flow 2.6 µs 50 ms 19,048× under target
Redaction throughput 0.87 ms 10 ms 11.5× under target
Sanitization throughput 21.6 µs 5 ms 231× under target
Memory footprint 39.9 KB 50 MB 1,253× under target

Detection overhead is 2.75 microseconds. That's less than a CPU branch misprediction. You will never measure the cost of loopguard watching your agent — until it saves you.


Quality bar — because trust is earned

Metric Value
Tests passing 436 (1 skipped)
Test coverage 97.3%
Ruff errors 0
Mypy strict errors 0
Pip-audit vulnerabilities 0
CI matrix Python 3.10/3.11/3.12 × ubuntu/macos/windows
Framework integrations 3 (LangGraph, CrewAI, OTel)
Trigger types 4
Fail-open modes 3
Performance benchmarks 7 (with regression detection in CI)

4-layer test suite: unit → integration → e2e → field study. Mock LLM models in CI (no real API calls). Snapshot testing for JSONL log format. Doc examples runnable via pytest --doctest-modules.


How is this different from...?

...LangGraph or CrewAI?

Those are frameworks — they define how you build agents. loopguard is a library — it wraps whatever you've already built. LangGraph doesn't detect stuck loops. CrewAI doesn't escalate. loopguard does both, and drops into either.

...cost trackers?

Cost trackers tell you what happened after the fact. loopguard stops the bleeding in real time. Dashboard vs. circuit breaker.

...a model router?

Routers decide upfront which model to use. loopguard handles the failure side: when the chosen model gets stuck. Different problem, complementary solutions.

...Overseer?

Overseer replaces your entire stack. loopguard is a drop-in. 4 lines of code. Keep your framework.


Architecture — how it's built

         ┌────────────────────────────────────────────────┐
         │                  Guard                          │
         │  (wraps your function via decorator)            │
         │                                                 │
  user ─▶│  ┌────────────┐    ┌──────────────────┐        │
  calls  │  │ FailureDet │───▶│ EscalationTrigger│        │
         │  └─────┬──────┘    └────────┬─────────┘        │
         │        │                     │ trigger hit?     │
         │        │              ┌──────┴──────┐          │
         │        │              │ No    Yes   │          │
         │        │              ▼      ▼     │          │
         │        │     return orig  ┌──────────┐         │
         │        │     output       │ Context  │         │
         │        │                  │ + Sanitize│         │
         │        │                  │ + Redact  │         │
         │        │                  └─────┬─────┘         │
         │        │                        ▼               │
         │        │                  ┌──────────────┐      │
         │        │                  │ Escalation   │      │
         │        │                  │ Model.invoke │      │
         │        │                  └──────┬───────┘      │
         │        │                         ▼              │
         │        │                  ┌──────────────┐      │
         │        │                  │ Logger       │      │
         │        │                  │ (JSONL+OTel) │      │
         │        │                  └──────┬───────┘      │
         │        │                         ▼              │
         │        │              return escalated output   │
         │        ▼                                      │
         │  ┌──────────┐                                │
         │  │CostTracker│                               │
         │  └──────────┘                                │
         └──────────────────────────────────────────────┘
Enter fullscreen mode Exit fullscreen mode

18 modules. Core has two dependencies: langchain-core (model binding) and pydantic (configuration). Everything else is an optional extra.

Key data models

GuardConfig — Pydantic-validated, every field constrained:

class GuardConfig(BaseModel):
    escalation_model: BaseChatModel
    on_escalate: Literal["auto", "interrupt"] = "auto"
    max_escalations_per_run: int = Field(default=1, ge=1, le=10)
    triggers: dict[str, TriggerConfig] = ...
    max_context_tokens: int = Field(default=4000, ge=500, le=32000)
    compress_context: bool = True
    redact_patterns: list[str] = []
    redact_fields: list[str] = []
    sanitize_context: bool = True
    on_guard_error: Literal["raise_original", "return_last_output", "return_sentinel"] = "raise_original"
    max_history_steps: int = Field(default=100, ge=10, le=1000)
Enter fullscreen mode Exit fullscreen mode

EscalationEvent — structured log record for every escalation:

class EscalationEvent(BaseModel):
    timestamp: float
    trigger_type: str          # "repeated_error" | "test_failure" | ...
    trigger_detail: str        # "ImportError on line 42, 3 consecutive"
    retry_count: int
    workhorse_model: str
    escalation_model: str
    escalation_category: Literal["routing", "failover"]
    context_tokens: int
    escalation_tokens: int
    escalation_cost_usd: float
    total_task_cost_usd: float # retries + escalation
    success: bool
    sanitized: bool
    redacted_fields: list[str]
Enter fullscreen mode Exit fullscreen mode

Contribute — this is where the community comes in

This is an open-source project built with big-tech engineering practices. The repo has a full CONTRIBUTING.md. Here's the short version:

Get started in 3 commands

git clone https://github.com/<your-username>/ai-loopguard.git
cd ai-loopguard
pip install -e ".[dev]"
Enter fullscreen mode Exit fullscreen mode

That installs the full dev toolchain: pytest, pytest-asyncio, pytest-cov, pytest-snapshot, ruff, mypy, pip-audit, pip-licenses.

Verify your setup

pytest          # 436 tests, 97.3% coverage
ruff check      # 0 errors
mypy --strict ai_loopguard/ tests/  # 0 errors
Enter fullscreen mode Exit fullscreen mode

Good first issues — where to start

Area What you can build Difficulty
New trigger hallucination_cycle detection — agent claims to fix something it already attempted. Add a TriggerConfig, implement _check_hallucination_cycle in detectors.py, add unit + e2e tests Medium
LangGraph integration Add support for LangGraph's Send API for parallel agent fan-out. Currently on_step_end handles sequential nodes — extend it for parallel sub-graphs Medium
CrewAI integration Thread-safe per-agent state for concurrent CrewAI execution. The current state-swap is sequential — make it work with CrewAI's planned parallel mode Hard
New integration Add an integration for AutoGen, Semantic Kernel, or your favorite agent framework. Follow the pattern in langgraph.py / crewai.py Medium
CLI improvements Add JSON output mode, loopguard watch for tailing live logs, or --model filter Easy
Docs More integration examples, troubleshooting guide, migration guide for v0.2.0 Easy
Performance Add benchmarks for large step histories (1000+ steps), concurrent guard instances Medium

How to add a new trigger (the most common contribution)

  1. Add a TriggerConfig in config.py with Pydantic-validated constraints
  2. Implement the check in detectors.py — a method that inspects GuardState step history and returns a TriggerResult
  3. Add unit tests — threshold boundary, below threshold, reset after guard.reset(), empty history edge cases
  4. Add an e2e test — full pipeline: trigger fires → context packaged → model invoked → event logged
  5. Update docsdocs/triggers.md with what it detects, config, defaults, example

How to add a new framework integration

  1. Create ai_loopguard/integrations/<name>.py — a handler/wrapper that accepts a Guard and exposes framework-appropriate hooks
  2. Add an optional dependency extra in pyproject.toml
  3. Follow the existing patterns in langgraph.py, crewai.py, otel.py — accept Guard in constructor, call guard.record_* at framework lifecycle points, fail open by default
  4. Add integration + e2e tests — normal step, trigger firing, fail-open behavior
  5. Write an integration guide in docs/integrations/<name>.md

Code standards

Standard Tool Requirement
Linting ruff check Zero errors. Line length 100
Types mypy --strict Zero errors. Python 3.10+
Coverage pytest --cov ≥ 90% enforced
Commits Conventional Commits feat:, fix:, docs:, test:, refactor:
Branches Prefix naming feature/, fix/, docs/, test/

PR checklist

  • Link the related issue (Closes #N)
  • CI green: pytest, ruff check, mypy --strict
  • Coverage doesn't drop below 90%
  • Features require tests
  • Features require docs
  • One feature/fix per PR — split mixed changes

The PRD, SPEC, and WBS — full engineering docs in the repo

This wasn't built by vibe-coding. The repo includes:

Document Lines What it covers
PRD 886 Problem statement, competitive analysis, goals, non-goals, threat model, functional requirements, success metrics, release scope, roadmap
SPEC 1,511 Architecture, module structure, data models, trigger evaluation, escalation pipeline, security implementation, CLI spec
WBS 1,247 12 milestones (S1-S12), all complete, with checkpoints and LLM routing strategy per task

Competitive positioning (PRD §4.3)

                 Pre-call               In-flight              Post-hoc
                 ────────               ────────               ────────
Budget/routing:  ASP, TierForge, l6e    ─                      ─
Failure detect:  ─                      loopguard              ─
Cost tracking:   ─                      ─                      agentcost-sdk, LangSmith
Full framework:  ─                      Overseer (all-in-one)  ─
Enter fullscreen mode Exit fullscreen mode

loopguard owns the in-flight failure detection + escalation column. Nobody else does.


Roadmap

Version What ships
v0.1.0 (current) 4 triggers, sync+async decorators, LangGraph + CrewAI + OTel, CLI, JSONL logging, sanitization + redaction, fail-open, 436 tests
v0.2.0 Hallucination cycle detection, prompt injection detector, multi-agent coordination, env + YAML config, CLI filters
v0.3.0 Custom trigger SDK, escalation policy engine, cost budget enforcement
Post-launch Web dashboard, LangSmith integration, eval harness integration

Over to you

If you're running AI agents in production, you will hit a stuck loop. It's not a question of if — it's when. The question is whether you'll have a circuit breaker when it happens.

pip install ai-loopguard
Enter fullscreen mode Exit fullscreen mode
from ai_loopguard import Guard

guard = Guard(escalation_model=gpt4_model)

@guard.protect
def agent_step(state):
    return cheap_model.generate(state)
Enter fullscreen mode Exit fullscreen mode

4 lines. MIT licensed.

Visit the repo

👉 github.com/deghosal-2026/ai-loopguard

If this solves a problem you've been hitting, give it a star — it helps other engineers find it. Stars are the currency of open source discoverability, and this is a tool that every agentic team needs but nobody has built yet.

Contribute

This is a community project. The best contributions are the ones I can't anticipate:

  • Found a bug? Open an issue with a repro — I'll triage within 24 hours.
  • Have an idea for a trigger? The hallucination_cycle trigger is the most requested feature. If you've seen hallucination patterns in your agents, your field experience would shape the detection logic. Open an issue describing the pattern.
  • Using a framework I don't support yet? AutoGen, Semantic Kernel, LlamaIndex — the integration pattern is proven with LangGraph and CrewAI. Pick your framework, follow the pattern in langgraph.py or crewai.py, and open a PR.
  • Want to improve the CLI? loopguard watch for live log tailing, JSON output mode, Grafana dashboard export — all welcome.
  • Good at docs? More real-world examples, troubleshooting guides, and migration guides are always needed.

See CONTRIBUTING.md for the full guide. 3 commands to get started, full dev toolchain included.

Tell me how to make this better

I want your honest feedback. Comment below or open a GitHub discussion:

  • How are you handling agent loops today? Rigid iteration limits? Manual monitoring? Just hoping it doesn't happen? I want to know what your current stack looks like so I can build the right thing.
  • What triggers are you missing? The 4 current triggers (test_failure, repeated_error, schema_invalid, custom) came from my own experience. Your agents might get stuck in ways I haven't seen. Tell me about it.
  • Is auto-escalation the right default? Or should interrupt mode (ask before spending tokens) be the default for most teams? I chose auto for production throughput, but I might be wrong.
  • What's your escalation cost threshold? At what point does model escalation become not worth it? I'm gathering data to build a cost-aware escalation policy engine for v0.3.0.
  • Are you running multi-agent systems? The CrewAI integration gives each agent its own state. Does that match your mental model? Would you want cross-agent escalation (Agent A gets stuck → Agent B picks up the task)?

Your input shapes the v0.2.0 and v0.3.0 roadmap. The best ideas will come from the people running agents in production — not from me sitting in a vacuum.

Star the repo. Open an issue. Drop a comment. Share it with your team. Every contribution makes agent reliability better for everyone.


Links


This is the library I wish existed last month. So I built it. Now it's yours.

#ai #llm #agents #python #opensource #langgraph #crewai #devops #llmops #circuitbreaker #costoptimization #modelrouting #production #reliability

Top comments (0)