Watchers | ARGUS Docs

Overview

ArgusWatcheris what you interact with. It hooks into your graph, records every node's execution, runs detectors, and produces traces. One watcher per run.

Basic Usage

One call covers StateGraph and already-compiled apps:

python

from argus import ArgusWatcher

watcher = ArgusWatcher()
app = watcher.attach(graph)         # StateGraph or compiled app
result = app.invoke(initial_state)  # persisted automatically

# Constructor form still works if you compile yourself:
watcher = ArgusWatcher(graph)       # uncompiled StateGraph
app = graph.compile()
result = app.invoke(initial_state)

Runs persist when invoke() returns, including cyclic graphs. finalize() is an optional idempotent flush, not required.

Parameters

Everything is optional. Pass to the constructor to override config file and environment variable values.

Core

graphStateGraph

Uncompiled StateGraph to monitor. Or omit and call attach() on a StateGraph or compiled app.

Default: None

max_field_sizeint

Max characters per captured state field. Fields exceeding this get truncated.

Default: 50_000

strictbool

Raise an exception if any detector fires. Use in CI/CD to fail builds on quality regressions.

Default: False

investigatebool | "always"

LLM root-cause investigation. True = on failure only, "always" = every node, False = off.

Default: True

Security

redact_keysset[str]

Field names to redact from stored outputs (e.g. {"password", "api_key"}).

Default: None

validatorsdict

Per-node semantic validators. Use "*" as key to run on every node. Each validator is a (bool, str) callable.

Default: None

python

# Validators — catch semantic failures
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.

Replay & Eval

persist_statebool

Save run records to .argus/runs/. Set False for ephemeral monitoring.

Default: True

record_httpbool

Record all external HTTP/API calls for deterministic replay.

Default: True

semantic_judgebool

LLM-powered quality judge on every node output. Requires a provider key (OpenAI, Anthropic, or Google).

Default: False

judge_modelstr

Model for the semantic judge and investigation.

Default: "gpt-4o"

python

# Full example with multiple options
watcher = ArgusWatcher(
    graph,
    semantic_judge=True,
    judge_model="gpt-4o-mini",
    strict=True,
    record_http=True,
    redact_keys={"api_key", "token"},
    validators={
        "summarize": lambda o: (len(o.get("summary", "")) > 10, "Summary too short"),
    },
)

Cost

semantic_judge sends node outputs to an LLM. This adds API cost proportional to the number of nodes. Use it in staging/CI, not every production run.

Lifecycle

A Watcher goes through four phases:

Created — constructor called, parameters loaded
Watching — graph instrumented, ready for execution
Recording — pipeline running, capturing node data
Finalized — detectors run, forensics generated, trace stored

One run per Watcher

A Watcher instance tracks one execution run. Create a new Watcher for the next run.