Guide

I want to add ARGUS monitoring to my LangGraph pipeline. Before writing any code, audit my codebase and then integrate it properly.

## STEP 1 — AUDIT MY PIPELINE

Find the file where my StateGraph is defined and check these things:

1. STATE TYPE: Find my state class. ARGUS works best when state is a TypedDict (or Pydantic model / dataclass). If my state is just a plain dict, convert it to a TypedDict with proper field annotations. Example:

   # BAD — plain dict, ARGUS can't check field contracts
   app = graph.compile()
   result = app.invoke({"query": "...", "results": []})

   # GOOD — TypedDict lets ARGUS verify fields between nodes
   class AgentState(TypedDict):
       query: str
       results: list[str]
       summary: str

2. NODE RETURN TYPES: Check every node function. Each should:
   - Accept the state type as its first parameter (type-annotated)
   - Return a dict with only the fields it's responsible for
   - NOT return the entire state — just the fields it modifies

   # BAD — no type hint, returns everything
   def search(state):
       return {**state, "results": [...]}

   # GOOD — typed, returns only what it produces
   def search(state: AgentState) -> dict:
       return {"results": [...]}

3. GRAPH STRUCTURE: Check if the graph is:
   - Linear (A → B → C) — auto-finalize works, no extra code needed
   - Fan-out/fan-in (DAG) — auto-finalize works
   - Cyclic (has loops / back-edges) — MUST call watcher.finalize() after invoke

4. ASYNC CHECK: If node functions are async (async def), ARGUS handles both — just make sure you're using await app.ainvoke() not app.invoke().

5. EXTERNAL CALLS: List which nodes make external API calls (OpenAI, search APIs, databases).

Print a summary of what you found and any fixes needed before proceeding.

## STEP 2 — FIX COMPATIBILITY ISSUES

If you found issues in Step 1, fix them now:
- Convert plain dict state to TypedDict
- Add type annotations to node function parameters
- Make nodes return only their output fields (not the full state)
- Ensure all fields referenced by downstream nodes exist in the TypedDict

## STEP 3 — INTEGRATE ARGUS

Install: pip install argus-agents

Add ArgusWatcher to the file where the graph is built:

from argus import ArgusWatcher

watcher = ArgusWatcher(graph)          # pass StateGraph before compile()
app = graph.compile()
result = app.invoke(initial_state)
watcher.finalize()                     # ALWAYS call this — required for cyclic graphs, safe for all
print(watcher.run_id)

If the graph is already compiled elsewhere, use watch_compiled():

watcher = ArgusWatcher()
app = watcher.watch_compiled(app)
result = app.invoke(initial_state)
watcher.finalize()

## STEP 4 — PICK THE RIGHT CONFIG

Choose parameters based on what you found in the audit.
Note: record_http, semantic_judge, investigate, and persist_state are all enabled by default.

watcher = ArgusWatcher(graph,
    # DETECTION STRICTNESS — catches empty lists, nested errors, type mismatches
    strict=True,

    # IF you want automatic root cause analysis on failures (default: True)
    investigate=True,         # or "always" to investigate every run

    # IF any fields contain secrets or tokens
    redact_keys={"token", "api_key", "password"},

    # ADD validators for nodes that produce critical output:
    validators={
        # example: ensure summaries aren't empty stubs
        "summarize": lambda o: (len(o.get("summary", "")) > 10, "Summary too short"),
        # wildcard: runs on every node
        "*": lambda o: ("error" not in o, "error key present"),
    },
)

app = graph.compile()
result = app.invoke(initial_state)
watcher.finalize()                     # persists the run to .argus/runs/

After running the pipeline:
  argus list              # see all recorded runs
  argus show last         # inspect the most recent run
  argus show <id>         # inspect a specific run by ID
  argus ui                # open the web dashboard