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

This tutorial shows how to add audit logging around a LlamaIndex pipeline in Python so you can trace every retrieval, query, and response path. You need this when your app handles regulated data, internal knowledge bases, or support workflows where you must answer: who asked what, what sources were used, and what the system returned.

What You'll Need

• •Python 3.10+
• •llama-index
• •openai API key set as OPENAI_API_KEY
• •A writable local directory for logs
• •Basic familiarity with VectorStoreIndex, SimpleDirectoryReader, and QueryEngine
• •Optional: pydantic if you want structured audit records

Install the packages:

pip install llama-index openai

Step-by-Step

1. •Start by creating a small audit logger that writes JSON lines to disk. JSONL is the right shape here because it is easy to grep, ship to SIEM tools, or load into analytics later.

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path

AUDIT_PATH = Path("audit.log")

def write_audit(event_type: str, payload: dict) -> None:
    record = {
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **payload,
    }
    with AUDIT_PATH.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

2. •Load your documents and build a basic index. This example uses a local data/ folder so you can test the full flow without extra infrastructure.

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

write_audit(
    "index_built",
    {
        "document_count": len(documents),
        "source_dir": "data",
    },
)

3. •Wrap the query path so every request gets logged before and after execution. In production, this is where you attach user identity, tenant ID, request ID, and any policy tags you need for compliance.

def audited_query(query_engine, user_id: str, question: str):
    request_id = str(uuid.uuid4())

    write_audit(
        "query_started",
        {
            "request_id": request_id,
            "user_id": user_id,
            "question": question,
        },
    )

    response = query_engine.query(question)

    write_audit(
        "query_completed",
        {
            "request_id": request_id,
            "user_id": user_id,
            "question": question,
            "response_text": str(response),
        },
    )

    return response

4. •If you need source-level traceability, log the retrieved nodes too. This is the part auditors usually care about most because it shows which chunks influenced the answer.

retriever = index.as_retriever(similarity_top_k=3)
nodes = retriever.retrieve("What does the policy say about refunds?")

write_audit(
    "retrieval_completed",
    {
        "query": "What does the policy say about refunds?",
        "top_k": 3,
        "nodes": [
            {
                "node_id": node.node.node_id,
                "score": node.score,
                "text_preview": node.node.get_content()[:200],
            }
            for node in nodes
        ],
    },
)

5. •Put it together in a small runnable script. This gives you a repeatable pattern: build index once, log retrievals and queries every time.

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex

def main():
    documents = SimpleDirectoryReader("data").load_data()
    index = VectorStoreIndex.from_documents(documents)
    query_engine = index.as_query_engine()

    question = "Summarize the refund policy."
    response = audited_query(query_engine, user_id="u-12345", question=question)

    print(response)

if __name__ == "__main__":
    main()

Testing It

Run the script against a small folder of text files in data/, then inspect audit.log. You should see separate entries for indexing, retrieval, query start, and query completion.

Check that each record has a unique event_id, an ISO-8601 UTC timestamp, and enough metadata to reconstruct the request later. If you want stronger guarantees, add tests that assert every audit record includes request_id, user_id, and event_type.

A good sanity check is to run the same question twice and confirm both runs produce distinct request IDs but identical source documents when the corpus has not changed.

Next Steps

• •Add redaction before logging questions or responses that may contain PII or account numbers.
• •Ship JSONL audit events to OpenTelemetry or your SIEM instead of keeping them on disk.
• •Move from wrapper-based logging to callback-based instrumentation if you need deeper visibility into LlamaIndex internals.

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.