Logging LlamaIndex prompts and responses is the difference between guessing why your RAG pipeline returned garbage and knowing exactly which instruction or retrieved chunk poisoned the context. Before you wire up evals or production traffic, you need a durable record of every LLM call LlamaIndex makes on your behalf. This tutorial builds a custom callback handler that writes those pairs to structured logs you can grep, ship to S3, or load into a dataframe.
Prerequisites
- Python 3.10 or newer
llama-indexandllama-index-llms-openai(pip install)- An OpenAI API key in
OPENAI_API_KEY, or a compatible endpoint. If you’re routing through n4n.ai, point the OpenAI client at its OpenAI-compatible endpoint to get 240+ models with automatic fallback; the logging code below is identical. - A scratch directory:
mkdir llama_log_demo && cd llama_log_demo
python -m venv .venv && source .venv/bin/activate
pip install llama-index llama-index-llms-openai
echo "The quick brown fox jumps over the lazy dog." > data.txtStep 1: Understand the callback surface
LlamaIndex emits typed events through a CallbackManager. The event you care about is CBEventType.LLM, fired for every chat/completion call. Each event carries a payload with messages or prompt on start, and response on end. By subclassing BaseCallbackHandler and filtering on that type, you capture exactly the data needed for logging LlamaIndex prompts and responses without noise from embedding or retriever events.
Step 2: Write the custom handler
Create logger_handler.py. We use stdlib logging and emit one line per prompt message and one per response. For chat models the payload uses EventPayload.MESSAGES; for legacy completion it uses EventPayload.PROMPT.
import logging
from llama_index.core.callbacks import (
BaseCallbackHandler,
CBEventType,
EventPayload,
)
class PromptResponseLogger(BaseCallbackHandler):
def __init__(self, level=logging.INFO):
# Only subscribe to LLM events
super().__init__(event_types=[CBEventType.LLM])
self.logger = logging.getLogger("llama_index.llm")
self.logger.setLevel(level)
def on_event_start(self, event_type, payload=None, event_id="", parent_id="", **kwargs):
if payload is None:
return
if EventPayload.MESSAGES in payload:
for msg in payload[EventPayload.MESSAGES]:
self.logger.info(f"PROMPT [{msg.role}]: {msg.content}")
elif EventPayload.PROMPT in payload:
self.logger.info(f"PROMPT: {payload[EventPayload.PROMPT]}")
def on_event_end(self, event_type, payload=None, event_id="", parent_id="", **kwargs):
if payload is None:
return
resp = payload.get(EventPayload.RESPONSE)
if resp is None:
return
# ChatResponse has .message.content, CompletionResponse has .text
text = getattr(resp, "message", None)
if text is not None:
text = text.content
else:
text = getattr(resp, "text", str(resp))
self.logger.info(f"RESPONSE: {text}")Attach a basic stream handler in your entry script so logs print to stderr.
Step 3: Wire the handler into Settings
LlamaIndex uses a global Settings object for LLM and callback configuration. Set both before building any index.
import logging
import sys
import os
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.core.callbacks import CallbackManager
from logger_handler import PromptResponseLogger
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
llm = OpenAI(model="gpt-3.5-turbo", temperature=0)
handler = PromptResponseLogger()
Settings.llm = llm
Settings.callback_manager = CallbackManager([handler])If you use a gateway, instantiate OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o") — the handler stays the same.
Step 4: Build index and run a query
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
docs = SimpleDirectoryReader(input_files=["data.txt"]).load_data()
index = VectorStoreIndex.from_documents(docs)
query_engine = index.as_query_engine()
response = query_engine.query("What does the fox do?")
print("ANSWER:", response.response)Run it:
python main.pyExpected output at checkpoint
You should see interleaved log lines from the handler, then the printed answer:
INFO:llama_index.llm:PROMPT [system]: You are a helpful assistant.
INFO:llama_index.llm:PROMPT [user]: Context: The quick brown fox jumps over the lazy dog.\nQuestion: What does the fox do?
INFO:llama_index.llm:RESPONSE: The fox jumps over the lazy dog.
ANSWER: The fox jumps over the lazy dog.The system prompt is auto-generated by LlamaIndex; your user question is wrapped with retrieved context. Logging LlamaIndex prompts and responses this way makes it trivial to spot when the context is empty or the instruction is malformed.
Step 5: Emit JSON for downstream analysis
Plain text is fine for eyeballing, but structured logs survive aggregation. Extend the handler to dump a dict per event:
import json
import time
class JsonPromptResponseLogger(BaseCallbackHandler):
def __init__(self, sink=print):
super().__init__(event_types=[CBEventType.LLM])
self.sink = sink
def on_event_start(self, event_type, payload=None, event_id="", parent_id="", **kwargs):
if not payload:
return
if EventPayload.MESSAGES in payload:
for m in payload[EventPayload.MESSAGES]:
self.sink(json.dumps({
"ts": time.time(),
"phase": "prompt",
"role": m.role,
"content": m.content,
}))
elif EventPayload.PROMPT in payload:
self.sink(json.dumps({
"ts": time.time(),
"phase": "prompt",
"content": payload[EventPayload.PROMPT],
}))
def on_event_end(self, event_type, payload=None, event_id="", parent_id="", **kwargs):
if not payload:
return
resp = payload.get(EventPayload.RESPONSE)
if resp is None:
return
text = getattr(resp, "message", None)
text = text.content if text else getattr(resp, "text", str(resp))
self.sink(json.dumps({
"ts": time.time(),
"phase": "response",
"content": text,
}))Pipe python main.py 2>&1 | jq to verify each line parses.
Step 6: Capture token usage
The response object often carries usage metadata. Append it to the response log:
usage = resp.additional_kwargs.get("usage") if hasattr(resp, "additional_kwargs") else None
if usage:
self.sink(json.dumps({"phase": "usage", "usage": usage}))This closes the loop on cost debugging—pair the prompt text with the billed token count.
Handling streaming responses
When streaming=True on the LLM, on_event_end still fires with the full aggregated response, so the same handler works. If you need token-by-token logs, override on_event_start to note the stream has begun and accumulate chunks in on_event_end via resp.delta if present; most LlamaIndex response objects expose the full text after completion, so the basic version is sufficient for audit trails.
What to do with the logs
Store them. A 100k-call sample of logged pairs is the cheapest eval set you will ever build. Diff prompts across model swaps, replay responses against a scoring function, or just grep for “PROMPT [system]” to confirm your instruction template didn’t drift after a library upgrade. Logging LlamaIndex prompts and responses is not observability theater—it’s the substrate for every later improvement.
That’s it. You now have a drop-in callback that captures every LLM interaction with zero changes to your query code.