aer1-llamaindex

AER-1 verifiable workflow receipts for LlamaIndex.

Attach one collector to a CallbackManager and every run emits a hash-chained, offline-verifiable verifiable workflow receipt (AER-1, Section 8): what the agent did, in what order, with per-step hashes and a Merkle root over the whole run. No network calls, no behavior changes, the collector only observes.

The AER-1 framework collector family

The LlamaIndex 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-llamaindex, LlamaIndex agents via callback handler
• 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

Install

pip install aer1-llamaindex

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

# pip install aer1-llamaindex
from aer1_llamaindex import AER1ReceiptCollector
from llama_index.core.callbacks import CallbackManager
from llama_index.core.callbacks.schema import CBEventType

def get_weather(location: str) -> str:
    """Local mock tool: executed on this machine, no network."""
    return f"Sunny, 22C in {location}"

collector = AER1ReceiptCollector(goal="check the Paris weather")
manager = CallbackManager([collector])  # line 1: register the collector

# A two-step flow: LLM reasoning, then a tool call. In production these
# events come from your real agent, query engine, or workflow, which gets
# `manager` as its callback_manager.
eid = manager.on_event_start(
    CBEventType.LLM, payload={"messages": ["What is the weather in Paris?"]})
manager.on_event_end(
    CBEventType.LLM, payload={"response": "I will call get_weather."}, event_id=eid)

eid = manager.on_event_start(
    CBEventType.FUNCTION_CALL,
    payload={"tool_name": "get_weather", "tool_kwargs": {"location": "Paris"}})
observation = get_weather("Paris")
manager.on_event_end(
    CBEventType.FUNCTION_CALL, payload={"observation": observation}, event_id=eid)

receipt = collector.finalize(final_answer=observation)  # line 2
assert collector.verify(receipt) == []  # VALID
collector.save("receipt.json", workflow=receipt)

That is the whole integration: attach the collector, run, finalize. The receipt is a plain JSON object you can store, ship to an auditor, or render in a UI.

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 event, seq 1..n in order
• merkle_root: Section 8.1 root over the ordered step receipt ids
• output_hash: SHA-256 of the final answer
• verify_url: where the verification procedure is documented

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

• seq, receipt_id, tool, receipt_hash, started_at, ended_at, status

Each step tool is the callback event type (for example llm, function_call, retrieval, synthesis). arguments is the event-start payload and observations is the event-end payload; each step receipt_hash is SHA-256 over the canonical JSON of those three, so the hash commits to what actually happened. The receipt stays compact while the hash commits to the content.

Verification

collector.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 collector.verify(receipt) != []  # fails, as it should

Notes

• Pass the CallbackManager as the callback_manager of your agent, query engine, or workflow; the collector records every event as a step, in event-end order.
• Pass goal= to the collector; LlamaIndex does not forward a task description through callback events, so the constructor is the reliable place for it.
• Exception events (event-end payload carrying an exception) are recorded with status: "error", which marks the whole receipt error.
• Use event_starts_to_ignore / event_ends_to_ignore (standard BaseCallbackHandler arguments) to skip noisy event types.
• session_id defaults to a fresh UUID per collector; 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 is an IETF Internet-Draft, draft-zambo-aer1, https://datatracker.ietf.org/doc/draft-zambo-aer1/

See it live

Your receipt is offline-verifiable, but you can also check it on the live verifier:

1. Copy the receipt JSON your code produced
2. Paste it at https://zambo.dev/verify
3. See the verification result with the Merkle root and step hashes

Or mint a live receipt directly: run any call at https://zambo.dev/demo and get a shareable receipt URL like https://zambo.dev/run/.

License

Apache-2.0