argus-agents 0.8.1

Production readiness platform for AI agent pipelines — detects silent failures, captures full state, enables step-level replay.

---

Catch silent failures in AI agent pipelines before production.

Your LangGraph pipeline runs fine — no exception. But three nodes later, something crashes with a KeyError. The real cause? A node upstream silently dropped a field. ARGUS catches this.

---

Install

pip install argus-agents

Quick Start

from argus import ArgusWatcher

watcher = ArgusWatcher(graph)       # attach to your StateGraph
app = graph.compile()
result = app.invoke(initial_state)  # run auto-saves on completion

That's it. ARGUS monitors every node, detects failures, and saves the run. No changes to your node functions.

---

What It Catches

Problem	Example
Silent failures	Node returns {} or drops a required field — no exception, pipeline keeps running broken
Semantic failures	Output structure is fine but values are wrong (placeholders, refusals, degraded text)
Loop stalls	Agent retries 5 times producing identical output — stuck loop burning tokens
Unnecessary retries	Loop produces correct answer on attempt 2, but validator forces 3 more iterations
Crash root cause	Traces KeyError at node 5 back to the upstream node that actually dropped the field
Contract violations	Output types don't match the next node's expected input schema

---

Detection Layers

Runs in order, each more expensive — only fires when needed:

1. Heuristics — 150+ failure signatures (placeholders, empty results, error keys, semantic degradation). Zero cost.
2. Anomaly detector — statistical checks for output size anomalies, timing outliers. Deterministic.
3. Correlator — traces failure propagation across nodes. Points at the origin, not the crash site.
4. LLM investigator — root cause explanations and debugging suggestions. Only on ambiguous failures.
5. Loop analyzer — mandatory LLM analysis for looped nodes: summarizes iterations, detects stalls, flags wasted retries.

---

Loop-Aware Inspection (new in v0.7.7)

Pipelines with loops (LLM → compiler → if fail, retry) get special treatment:

• Earlier iterations that self-corrected are marked retried (not counted as failures)
• Only the final iteration determines pass/fail
• LLM automatically analyzes every loop: what went wrong, what changed between attempts, whether retries were necessary
• Dashboard shows iteration badges, collapse/expand, and natural-language loop summaries

---

Replay

Fix a bug, re-run from the failing node. Skip upstream nodes entirely:

argus replay <run-id> node_7          # re-run from node_7 onward
argus replay <run-id> node_7 --only   # just that one node
argus diff <rerun-id>                 # compare vs original

External API calls (OpenAI, etc.) are recorded by default — replays are free and deterministic.

---

Semantic Judge

For subtle quality issues that pattern matching can't catch:

watcher = ArgusWatcher(graph, semantic_judge=True)

LLM evaluates output quality on every passing node. Catches wrong tone, unhelpful responses, outdated info.

---

Custom Validators

watcher = ArgusWatcher(graph, validators={
    "classify": lambda o: (o.get("label") in ["yes", "no"], "unexpected label"),
    "*":        lambda o: ("error" not in o, "error key present"),  # runs on every node
})

---

CLI

argus show last                      # most recent run
argus replay <id> <node>             # re-run from a node
argus diff <id-a> <id-b>             # compare two runs
argus ui                             # web dashboard
argus doctor                         # check setup health
argus login                          # sign in for cloud sync

---

Web Dashboard

argus ui    # opens at localhost:7842

Shows all runs, node-level detail, AI analysis, replay diffs, loop iteration badges, and comparison views. No account needed for local use.

---

Without LangGraph

from argus import ArgusSession

session = ArgusSession()
session.set_edges({"fetch": ["classify"], "classify": ["process"]})

fetch    = session.wrap("fetch",    fetch_fn)
classify = session.wrap("classify", classify_fn)
process  = session.wrap("process",  process_fn)

state = fetch(initial_state)
state = classify(state)
state = process(state)
session.finalize()

Works with any framework — Prefect, Temporal, plain Python.

---

Requirements

• Python 3.9+
• LangGraph 0.2+ (only for ArgusWatcher)
• OPENAI_API_KEY in env for semantic features (optional — all heuristic detection works without it)

---

v0.7.7 — changelog