ArgusWatcher
The main class for instrumenting LangGraph pipelines. Import from the top-level package:
from argus import ArgusWatchergraphStateGraphUncompiled StateGraph to monitor. Or omit and call attach() on a StateGraph or compiled app.
Default: None
max_field_sizeintMax characters per captured state field before truncation.
Default: 50_000
validatorsdict | NonePer-node semantic validators. Use "*" as key to run on every node. Each validator is a (bool, str) callable.
Default: None
strictboolEnable extra checks: nested error keys, rate-limit responses, empty lists, type mismatches. Recommended for CI/staging.
Default: False
investigatebool | "always"LLM root-cause investigation. True = on failure only, "always" = every node, False = off.
Default: True
redact_keysset[str] | NoneField names to redact from stored outputs (e.g. {"password", "api_key"}).
Default: None
persist_stateboolSave run records to .argus/runs/. Set False for ephemeral monitoring.
Default: True
record_httpboolRecord all external HTTP/API calls for deterministic replay.
Default: True
semantic_judgeboolLLM-powered quality judge on every node output. Requires a provider key (OpenAI, Anthropic, or Google).
Default: False
judge_modelstrModel for the semantic judge and investigation.
Default: "gpt-4o"
Methods
.attach()
One call for a StateGraph or an already-compiled app. Returns a compiled app with monitoring. invoke() / ainvoke() persist the run when they return. Prefer this for new code.
watcher = ArgusWatcher()
app = watcher.attach(graph) # StateGraph or compiled app
result = app.invoke(initial_state) # persisted automatically.watch()
Thin wrapper around attach() for an uncompiled StateGraph. Prefer attach() for new code. Not needed if you passed graph to the constructor.
watcher.watch(graph: StateGraph) -> None.watch_compiled()
Thin wrapper around attach() for an already-compiled graph. Prefer attach() for new code.
app = watcher.attach(compiled_app) # preferred
app = watcher.watch_compiled(app) # equivalent thin wrapper.finalize()
Optional idempotent flush. Not required after invoke() — including cyclic graphs. Safe to call; a second call is a no-op.
watcher.finalize() # optionalIf strict=True and detections fire, raises DetectionError after storing the run.
.get_trace()
Retrieve the trace after the run. Returns the same object as finalize().
trace = watcher.get_trace() -> Trace
# Trace properties
trace.id # str — unique trace identifier
trace.status # "ok" | "warning" | "failed"
trace.duration_ms # int — total execution time
trace.steps # list[TraceStep]
trace.detections # list[Detection]
trace.forensics # Forensics | None
trace.summary # str — human-readable summary.run_id
Access the run ID directly after execution.
print(watcher.run_id) # e.g. "run-abc12345"ArgusSession (without LangGraph)
For plain Python functions, Prefect, Temporal, or any non-LangGraph pipeline:
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 Python callable. ArgusWatcher requires LangGraph 0.2+; ArgusSession has no framework dependency.
Data Models
# TraceStep — one node execution
class TraceStep:
id: str
step_number: int
node_name: str
input_state: dict
output_state: dict
duration_ms: int
timestamp: datetime
# Detection — one detected issue
class Detection:
id: str
layer: str # "statistical" | "semantic" | "behavioral" | "structural"
severity: str # "info" | "warning" | "critical"
message: str
details: dict
step_id: str # which step triggered this
# Forensics — root cause analysis
class Forensics:
root_cause_step: str # step ID of the root cause
explanation: str # human-readable explanation
causal_chain: list # ordered list of steps from cause to symptom
detection_ids: list # which detections this explainsType hints