LlamaIndex Tutorial (Python): adding audit logs for beginners

This tutorial shows how to add audit logging around a basic LlamaIndex query flow in Python. You’ll capture who asked what, when they asked it, and what the system returned, which is the minimum you need for traceability in internal tools and regulated environments.

What You'll Need

• •Python 3.10+
• •A virtual environment
• •llama-index
• •An OpenAI API key set as OPENAI_API_KEY
• •Basic familiarity with VectorStoreIndex and QueryEngine
• •A writable local folder for log files

Install the package:

pip install llama-index

Step-by-Step

1. •Create a small LlamaIndex app and a place to store audit events.
   We’ll use a JSONL file because it’s easy to append to, easy to parse later, and works well for beginner-friendly audit trails.

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

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

AUDIT_LOG_PATH = Path("audit_logs.jsonl")
DATA_DIR = Path("data")

def write_audit_event(event: dict) -> None:
    event["ts"] = datetime.now(timezone.utc).isoformat()
    with AUDIT_LOG_PATH.open("a", encoding="utf-8") as f:
        f.write(json.dumps(event) + "\n")

documents = SimpleDirectoryReader(str(DATA_DIR)).load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

2. •Wrap your query call so every request gets logged before and after execution.
   In production, this is where you’d add request IDs, user IDs, tenant IDs, and policy metadata from your app layer.

def audited_query(user_id: str, question: str) -> str:
    write_audit_event({
        "event": "query_started",
        "user_id": user_id,
        "question": question,
    })

    response = query_engine.query(question)
    answer = str(response)

    write_audit_event({
        "event": "query_completed",
        "user_id": user_id,
        "question": question,
        "answer": answer,
    })

    return answer

3. •Call the wrapper from your application code.
   Keep the audit logic outside your business logic so you can reuse it across chat endpoints, background jobs, or internal admin tools.

if __name__ == "__main__":
    user_id = "user-123"
    question = "What does this document collection talk about?"

    result = audited_query(user_id=user_id, question=question)
    print(result)

4. •Add failure logging so you don’t lose evidence when queries break.
   This matters because failed requests are often more important than successful ones during incident review.

def audited_query_with_errors(user_id: str, question: str) -> str:
    write_audit_event({
        "event": "query_started",
        "user_id": user_id,
        "question": question,
    })

    try:
        response = query_engine.query(question)
        answer = str(response)

        write_audit_event({
            "event": "query_completed",
            "user_id": user_id,
            "question": question,
            "answer": answer,
            "status": "ok",
        })
        return answer

    except Exception as exc:
        write_audit_event({
            "event": "query_failed",
            "user_id": user_id,
            "question": question,
            "error_type": type(exc).__name__,
            "error_message": str(exc),
            "status": "error",
        })
        raise

5. •Make the audit trail more useful by adding a simple redaction rule.
   Don’t log raw secrets or sensitive payloads unless your compliance team explicitly wants that and your storage controls are in place.

def redact(text: str) -> str:
    if not text:
        return text
    return text.replace("password=", "password=[REDACTED]")

def audited_query_redacted(user_id: str, question: str) -> str:
    safe_question = redact(question)

    write_audit_event({
        "event": "query_started",
        "user_id": user_id,
        "question": safe_question,
    })

    response = query_engine.query(question)
    answer = str(response)

    write_audit_event({
        "event": "query_completed",
        "user_id": user_id,
        "question": safe_question,
        "answer_preview": answer[:500],
    })

    return answer

Testing It

Run the script against a folder with at least one text file in data/, then inspect audit_logs.jsonl. You should see one line for query_started and one for query_completed, each with a UTC timestamp.

If you trigger an error by pointing DATA_DIR at an empty folder or breaking the OpenAI key configuration, check that query_failed gets written before the exception bubbles up. That tells you your failure path is covered.

A quick sanity check is to open the log file and confirm it’s valid JSON Lines format: one JSON object per line. That format makes it easy to ship later into Elasticsearch, S3, Datadog, or a SIEM.

For a real system, also verify that:

• •User identity is coming from your auth layer, not hardcoded
• •Sensitive fields are redacted before logging
• •Log files are rotated or shipped off-host regularly

Next Steps

• •Add correlation IDs so one user request can be traced across multiple services.
• •Store logs in PostgreSQL or object storage instead of local files.
• •Hook this into FastAPI middleware so every LlamaIndex call is audited automatically.

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.