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) — will need 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). This determines whether to enable record_http.

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)
print(watcher.run_id)

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

watcher = ArgusWatcher()
app = watcher.watch_compiled(app)

## STEP 4 — PICK THE RIGHT CONFIG

Choose parameters based on what you found in the audit:

watcher = ArgusWatcher(graph,
    # ALWAYS RECOMMENDED
    strict=True,              # catches empty lists, nested errors, type mismatches
    persist_state=True,       # saves runs to .argus/runs/

    # IF nodes make paid API calls (OpenAI, etc.) — enables free deterministic reruns
    record_http=True,

    # IF you want LLM-powered quality checks on outputs (needs OPENAI_API_KEY)
    semantic_judge=True,      # reviews every node's output for subtle issues
    judge_model="gpt-4o",     # or "gpt-4o-mini" for cheaper runs

    # IF you want automatic root cause analysis on failures
    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"),
    },
)

# IF the graph has cycles (loops / back-edges):
# watcher.finalize()    ← call after invoke

After running the pipeline, use "argus show last" to see what ARGUS caught.