ArgusWatcher
The main class for instrumenting LangGraph pipelines. Import from the top-level package:
from argus import ArgusWatchergraphStateGraphLangGraph graph to monitor. If passed, watch() is called automatically.
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 OPENAI_API_KEY.
Default: False
judge_modelstrModel for the semantic judge and investigation.
Default: "gpt-4o"
Methods
.watch()
Instrument a graph for monitoring. Call before graph.compile(). Not needed if you passed graph to the constructor.
watcher.watch(graph: StateGraph) -> Nonewatch() before graph.compile(). If you compile first, ARGUS can't instrument the nodes..watch_compiled()
Instrument an already-compiled graph. Use when you can't call watch() before compilation (e.g. when using a checkpointer).
app = graph.compile(checkpointer=memory)
app = watcher.watch_compiled(app) -> CompiledGraph.finalize()
Run all detection layers, execute forensic analysis, and persist results. Only needed for cyclic graphs — linear and fan-out graphs auto-save.
watcher.finalize() -> TraceReturns the completed Trace object. If strict=True and detections fire, raises DetectionError after storing the trace.
.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