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)
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: custom → repeated_error → test_failure → schema_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
If agent_step fails 3 times, loopguard:
- 📦 Packages the context — what was tried, what failed, error traces
- 🛡️ Sanitizes against prompt injection (system/user separation, delimiters)
- 🔒 Redacts sensitive fields (API keys, tokens, PII)
- 🚀 Escalates to your configured cloud model
- 📊 Logs the event with full cost impact
- ✅ 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]
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"])
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")
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]"
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": ""})
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]
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()
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"})
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.0and 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
),
)
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
==================================================
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
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())
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│ │
│ └──────────┘ │
└──────────────────────────────────────────────┘
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)
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]
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]"
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
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)
-
Add a
TriggerConfiginconfig.pywith Pydantic-validated constraints -
Implement the check in
detectors.py— a method that inspectsGuardStatestep history and returns aTriggerResult -
Add unit tests — threshold boundary, below threshold, reset after
guard.reset(), empty history edge cases - Add an e2e test — full pipeline: trigger fires → context packaged → model invoked → event logged
-
Update docs —
docs/triggers.mdwith what it detects, config, defaults, example
How to add a new framework integration
-
Create
ai_loopguard/integrations/<name>.py— a handler/wrapper that accepts aGuardand exposes framework-appropriate hooks -
Add an optional dependency extra in
pyproject.toml -
Follow the existing patterns in
langgraph.py,crewai.py,otel.py— acceptGuardin constructor, callguard.record_*at framework lifecycle points, fail open by default - Add integration + e2e tests — normal step, trigger firing, fail-open behavior
-
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) ─
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
from ai_loopguard import Guard
guard = Guard(escalation_model=gpt4_model)
@guard.protect
def agent_step(state):
return cheap_model.generate(state)
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_cycletrigger 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.pyorcrewai.py, and open a PR. -
Want to improve the CLI?
loopguard watchfor 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
- Repository: github.com/deghosal-2026/ai-loopguard
- Issues: github.com/deghosal-2026/ai-loopguard/issues
- Contributing: CONTRIBUTING.md
- PRD (886 lines): docs/PRD.md
- SPEC (1,511 lines): docs/SPEC.md
- WBS (1,247 lines): docs/WBS.md
- Field study: docs/field-study/report.md
- Trigger reference: docs/triggers.md
- LangGraph integration: docs/integrations/langgraph.md
- CrewAI integration: docs/integrations/crewai.md
- Observability guide: docs/observability.md
- Release notes: docs/release-notes.md
- Performance benchmarks: benchmarks/performance-benchmarks.md
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)