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

This tutorial shows how to add audit logs to a LangGraph workflow in Python without polluting your business logic. You’ll end up with a graph that records each node’s input, output, and timestamp so you can trace decisions later for debugging, compliance, or incident review.

What You'll Need

• •Python 3.10+
• •langgraph
• •langchain-core
• •langchain-openai
• •An OpenAI API key set as OPENAI_API_KEY
• •Basic familiarity with StateGraph, nodes, and edges in LangGraph

Install the packages:

pip install langgraph langchain-core langchain-openai

Step-by-Step

1. •Start by defining a state that carries both your application data and an audit trail. Keep the audit log inside the graph state so every node can append to it deterministically.

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    user_input: str
    draft: str
    final_answer: str
    audit_log: Annotated[list[dict], add]

2. •Create a small helper that writes one audit record per node execution. This keeps logging consistent and avoids repeating the same dictionary shape across nodes.

from datetime import datetime, timezone

def audit_event(node_name: str, input_data: dict, output_data: dict) -> dict:
    return {
        "node": node_name,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "input": input_data,
        "output": output_data,
    }

3. •Build your graph nodes so each one returns both its business output and a new audit entry. The key detail is that audit_log is merged using Annotated[..., add], so each node can return a one-item list and LangGraph will append it.

from langchain_core.runnables import RunnableLambda

def draft_node(state: AgentState) -> dict:
    output = {"draft": f"Draft response for: {state['user_input']}"}
    return {
        **output,
        "audit_log": [audit_event("draft_node", {"user_input": state["user_input"]}, output)],
    }

def final_node(state: AgentState) -> dict:
    output = {"final_answer": state["draft"] + " | approved"}
    return {
        **output,
        "audit_log": [audit_event("final_node", {"draft": state["draft"]}, output)],
    }

4. •Wire the nodes into a simple graph with an entry point and an end point. This example uses two processing steps, but the same pattern works for larger graphs with branching and conditional edges.

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)
builder.add_node("draft_node", RunnableLambda(draft_node))
builder.add_node("final_node", RunnableLambda(final_node))

builder.add_edge(START, "draft_node")
builder.add_edge("draft_node", "final_node")
builder.add_edge("final_node", END)

graph = builder.compile()

5. •Run the graph and inspect the resulting audit trail. In production, you would send these records to your logging system or database, but keeping them in state first makes validation easy.

result = graph.invoke({"user_input": "Summarize policy changes", "draft": "", "final_answer": "", "audit_log": []})

print("Final answer:", result["final_answer"])
print("\nAudit log:")
for item in result["audit_log"]:
    print(item)

6. •If you want to connect this to an LLM-backed node later, keep the same audit pattern around the model call. The node should capture prompt inputs before invocation and model outputs after invocation, then append both to the log.

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

def llm_node(state: AgentState) -> dict:
    prompt = f"Write a concise answer for: {state['user_input']}"
    response = llm.invoke(prompt)
    output = {"draft": response.content}
    return {
        **output,
        "audit_log": [audit_event("llm_node", {"prompt": prompt}, {"content": response.content})],
    }

Testing It

Run the script and confirm that result["audit_log"] contains one entry per node execution in order. Each record should include the node name, UTC timestamp, input payload, and output payload.

Then change user_input and rerun it to make sure the audit entries reflect the new request. If you swap in an LLM node, verify that prompt text is captured before the call and model content is captured after it.

For a stronger test, assert on the number of audit entries and on specific fields like node and timestamp. That catches regressions when someone changes node behavior later.

Next Steps

• •Add a persistence layer for audit events using PostgreSQL or DynamoDB.
• •Use conditional edges to log branch decisions separately from node outputs.
• •Add redaction for PII before writing prompts or raw payloads into logs

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.