LangGraph Tutorial (Python): adding audit logs for advanced developers

This tutorial shows how to add durable audit logs to a LangGraph app in Python without polluting your agent logic. You’ll end up with a graph that records each node transition, the state diff, and the final outcome in a structured format you can ship to a database, SIEM, or object store.

What You'll Need

• •Python 3.10+
• •langgraph
• •langchain-core
• •Optional: python-dotenv if you want to load env vars from a .env file
• •A working understanding of:
  • •StateGraph
  • •nodes and edges
  • •invoke() / stream()
• •No LLM API key is required for this tutorial because we’ll use deterministic nodes

Step-by-Step

1. •Start with a typed state that carries both business data and an audit trail.
   The key pattern here is to keep audit metadata inside the graph state so every node can append to it without side effects.

from __future__ import annotations

from typing import Annotated, TypedDict
import operator
from datetime import datetime, timezone

from langgraph.graph import StateGraph, START, END


class AuditEvent(TypedDict):
    ts: str
    node: str
    action: str
    details: dict


class GraphState(TypedDict):
    input_text: str
    classification: str
    approved: bool
    audit_log: Annotated[list[AuditEvent], operator.add]

2. •Add a tiny helper for creating audit entries and keep it pure.
   In production, this helper is where you normalize fields for downstream systems like Splunk, Elastic, or PostgreSQL.

def make_event(node: str, action: str, **details) -> AuditEvent:
    return {
        "ts": datetime.now(timezone.utc).isoformat(),
        "node": node,
        "action": action,
        "details": details,
    }


def classify_node(state: GraphState) -> dict:
    text = state["input_text"].lower()
    classification = "high_risk" if any(word in text for word in ["refund", "chargeback", "fraud"]) else "low_risk"
    return {
        "classification": classification,
        "audit_log": [make_event("classify_node", "classified", classification=classification)],
    }

3. •Build the decision node and append an explicit audit record for the branch taken.
   This is the part most teams miss: logging only the input/output is not enough when you need to explain why a branch executed.

def decision_node(state: GraphState) -> dict:
    approved = state["classification"] == "low_risk"
    action = "approved" if approved else "escalated"

    return {
        "approved": approved,
        "audit_log": [
            make_event(
                "decision_node",
                action,
                classification=state["classification"],
                approved=approved,
            )
        ],
    }

4. •Add terminal nodes that write their own audit events and keep them separate from business logic.
   In regulated workflows, terminal nodes are where you usually emit final status for case management or compliance storage.

def approve_node(state: GraphState) -> dict:
    return {
        "audit_log": [
            make_event(
                "approve_node",
                "finalized",
                result="approved",
                input_text=state["input_text"],
            )
        ]
    }


def escalate_node(state: GraphState) -> dict:
    return {
        "audit_log": [
            make_event(
                "escalate_node",
                "finalized",
                result="escalated",
                input_text=state["input_text"],
            )
        ]
    }

5. •Wire the graph together and compile it.
   Use conditional routing so the audit trail captures both the branch decision and the final path taken.

def route_after_decision(state: GraphState) -> str:
    return "approve" if state["approved"] else "escalate"


builder = StateGraph(GraphState)
builder.add_node("classify", classify_node)
builder.add_node("decide", decision_node)
builder.add_node("approve", approve_node)
builder.add_node("escalate", escalate_node)

builder.add_edge(START, "classify")
builder.add_edge("classify", "decide")
builder.add_conditional_edges(
    "decide",
    route_after_decision,
    {
        "approve": "approve",
        "escalate": "escalate",
    },
)
builder.add_edge("approve", END)
builder.add_edge("escalate", END)

graph = builder.compile()

6. •Run it and inspect the accumulated log entries.
   Because audit_log uses list concatenation via operator.add, each node contributes an immutable event batch instead of mutating shared state.

result = graph.invoke(
    {
        "input_text": "Customer requests refund after suspected fraud",
        "classification": "",
        "approved": False,
        "audit_log": [],
    }
)

print("Approved:", result["approved"])
print("Classification:", result["classification"])
print("Audit events:")
for event in result["audit_log"]:
    print(event)

Testing It

Run the script twice with different inputs and confirm that low-risk text routes to approve while refund/fraud language routes to escalate. Check that each run produces multiple audit records in order: classification, decision, then terminal action.

If you want stronger verification, assert on both state and log contents in a unit test. For example, verify that result["audit_log"][-1]["node"] matches the expected terminal node and that details["classification"] matches your branching rule.

For production validation, serialize the returned audit_log as JSON and confirm it can be stored without transformation. That matters because audit pipelines break when logs contain non-serializable objects like datetimes or custom classes.

Next Steps

• •Move audit emission out of state and into a custom checkpointer or external sink for long-running graphs.
• •Add correlation IDs and user IDs to every event so logs can be joined across services.
• •Learn LangGraph interrupts and human-in-the-loop patterns so escalation paths can capture reviewer actions too.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.