aer1-langgraph

AER-1 verifiable workflow receipts for LangGraph multi-agent swarms.

Attach one callback handler to your graph and every run emits a hash-chained, offline-verifiable verifiable workflow receipt (AER-1, Section 8): which agent did what, in what order, with per-step hashes and one Merkle root over the whole swarm. Parallel Send branches are recorded with their true branch structure, not flattened. No network calls, no behavior changes, the handler only observes.

The AER-1 framework collector family

The LangGraph collector in the AER-1 framework collector family. Any agent running on these frameworks can emit verifiable AER-1 execution receipts: every step recorded, hash-chained, one Merkle root over the whole run.

• aer1-smolagents ��� SmolAgents agents via step callbacks
• aer1-openai-agents ��� the OpenAI Agents SDK via its TracingProcessor
• aer1-crewai ��� CrewAI crews via the event bus
• aer1-langgraph ��� LangGraph swarms via callback handler
• aer1-autogen ��� AutoGen multi-agent chats

Plus the aer1 metapackage: pip install aer1, then import aer1; aer1.instrument(). One line, zero config, auto-detects your framework.

Install

pip install aer1-langgraph

Use it (copy, paste, run ��� no API keys needed)

# pip install aer1-langgraph
from langgraph.graph import StateGraph, START, END
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from aer1_langgraph import AER1LangGraphHandler

def researcher(state: dict):
    model = GenericFakeChatModel(messages=iter(["found 3 papers"]))
    return {"notes": model.invoke("research").content}

def writer(state: dict):
    model = GenericFakeChatModel(messages=iter(["draft complete"]))
    return {"draft": model.invoke("write").content}

graph = StateGraph(dict)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", END)
app = graph.compile()

handler = AER1LangGraphHandler(goal="Write the report")
result = app.invoke({}, config={"callbacks": [handler]})
receipt = handler.finalize(final_answer=result)
assert handler.verify(receipt) == []  # VALID
handler.save("receipt.json", workflow=receipt)

Swap the fake models for your real chat models and the same three handler lines receipt a live swarm. The receipt is a plain JSON object you can store, ship to an auditor, or render in a UI.

Swarm receipts

Each node execution becomes one receipt step carrying the node name as agent identity (tool and agent on every step). A three-agent swarm with two parallel workers produces steps like:

seq=1 tool=router      agent=router
seq=2 tool=worker      agent=worker   branch=worker:9f3a...
seq=3 tool=worker      agent=worker   branch=worker:41bc...
seq=4 tool=aggregator  agent=aggregator

Steps carry run_id, parent_run_id, and the LangGraph checkpoint namespace, so the true execution graph (who ran in parallel under whom) is reconstructible from the receipt. handler.swarm_summary() returns the agent roster, step counts, and branch groups in one call.

Tool calls and model calls inside a node are folded into that node's step: the step's receipt_hash is SHA-256 over the canonical JSON of the node name, inputs, outputs, tool calls (name, arguments, output), model-call summaries, branch metadata, and any error. The hash commits to what the agent actually did; the receipt stays compact.

What the receipt contains

Workflow level (AER-1 Section 8, Table 2):

• type, version, workflow_id, receipt_id, session_id
• goal, status
• steps: one record per node execution, seq 1..n in completion order
• merkle_root: Section 8.1 root over the ordered step receipt ids
• output_hash: SHA-256 of the final graph output
• verify_url: where the verification procedure is documented

Step level (AER-1 Section 8, Table 3, plus swarm fields):

• seq, receipt_id, tool (= node name), receipt_hash
• started_at, ended_at, status
• agent, run_id, parent_run_id, checkpoint_ns
• tool_calls, llm_calls (folded records)

Verification

handler.verify(receipt) runs the full offline check and returns a list of failure reasons, empty when valid:

• all Table 2 / Table 3 members present and well-formed
• seq values exactly 1..n in order, no gaps
• no two steps share a receipt_id (MM-1)
• merkle_root matches the recomputed Section 8.1 root
• strict RFC 3339 timestamps, lowercase UUIDs, 64-char hex digests

Tamper with any field and verification fails. Try it:

receipt["steps"][0]["tool"] = ""
assert handler.verify(receipt) != []  # fails, as it should

Notes

• One handler observes one run at a time. Call handler.reset() or construct a fresh handler before the next run. If a new root graph run starts on a handler that already finished one, it auto-resets so two runs never mix into one receipt.
• The __start__ / __end__ pseudo-nodes are machinery, not agent work, and are not recorded as steps. Conditional-edge router functions are user code and are recorded.
• Steps still in flight when finalize() runs (interrupted runs) are flushed as status: "incomplete" rather than dropped.
• session_id defaults to a fresh UUID per handler; pass your own to correlate receipts across runs.
• verify_url defaults to the AER-1 specification page; point it at your own verifier in production.

Spec

AER-1: Agent Execution Receipts, draft-zambo-aer1, https://datatracker.ietf.org/doc/draft-zambo-aer1/

License

Apache-2.0