When a chatbot returns a bad answer, guessing at the prompt won’t fix it. To capture request response pairs debugging effectively, you need the exact inbound payload and the raw model completion stored together with a correlation ID. This guide shows how to build a minimal capture layer for any OpenAI-compatible endpoint, including streaming, so you can replay sessions later.
Step 1: Pick the capture boundary
You can capture request response pairs debugging either inside the client process or at a network gateway. Client-side wrappers are easiest to drop into an existing Python service; a reverse proxy works better if you have multiple languages or want central collection.
If you already route traffic through a gateway such as n4n.ai—an OpenAI-compatible endpoint that addresses 240+ models with automatic fallback when a provider is degraded—you can log at the gateway and skip per-service code. Otherwise, wrap the client.
Key requirements for either approach:
- Record the full request JSON (model, messages, params, headers).
- Record the full response JSON (choices, usage, finish_reason).
- Attach a session ID and turn number for replay.
Decide before writing code. A client wrapper gives you request context (local variables, user ID) for free. A gateway gives you a single choke point. Both satisfy the goal to capture request response pairs debugging; the code below assumes the client wrapper because it is the most portable.
Step 2: Wrap the OpenAI client in Python
The openai SDK accepts a custom httpx.Client via http_client. We’ll use an event hook to snapshot the request body and the response body. This captures both sides without modifying your call sites.
import json
import time
import httpx
import openai
LOG_FILE = "llm_pairs.jsonl"
def log_pair(session_id, turn_id, request_json, response_json):
with open(LOG_FILE, "a") as f:
f.write(json.dumps({
"ts": time.time(),
"session_id": session_id,
"turn_id": turn_id,
"request": request_json,
"response": response_json,
}) + "\n")
class CaptureTransport(httpx.HTTPTransport):
def handle_request(self, request):
# Read request body (JSON) before send
req_body = json.loads(request.content.decode()) if request.content else None
response = super().handle_request(request)
# Clone response to read body
resp_body = response.read()
response = response.clone() # ensure stream not consumed
# We need session/turn; pass via request headers
session_id = request.headers.get("x-session-id", "unknown")
turn_id = request.headers.get("x-turn-id", "unknown")
log_pair(session_id, turn_id, req_body, json.loads(resp_body.decode()))
return response
client = openai.OpenAI(
api_key="sk-...",
base_url="https://api.openai.com/v1", # or your gateway
http_client=httpx.Client(transport=CaptureTransport()),
)This transport logs every call. Note that response.read() consumes the stream; we call clone() so the SDK still gets the bytes. For non-streaming calls this is fine. The request body is parsed from request.content; if you use compressed bodies, decompress first.
To capture request response pairs debugging with full fidelity, also copy request.headers into the logged request object. Headers carry routing and cache hints that explain provider behavior.
Step 3: Capture streaming responses correctly
Streaming breaks the simple transport above because the body is delivered in chunks and not fully present at handle_request. You must wrap the streaming iterator and accumulate deltas.
import json
class StreamingCapture:
def __init__(self, original_stream, session_id, turn_id):
self.original = original_stream
self.session_id = session_id
self.turn_id = turn_id
self.accumulated = []
def __iter__(self):
for chunk in self.original:
self.accumulated.append(chunk.model_dump())
yield chunk
# After stream ends, log the aggregated pair
log_pair(self.session_id, self.turn_id,
{"stream": True}, {"chunks": self.accumulated})
def create_with_capture(client, session_id, turn_id, **kwargs):
kwargs.setdefault("stream", False)
if kwargs["stream"]:
resp = client.chat.completions.create(**kwargs)
return StreamingCapture(resp, session_id, turn_id)
else:
return client.chat.completions.create(**kwargs)Use model_dump() (OpenAI SDK v1+) to serialize chunk objects. This gives you a faithful replay artifact. For async code, implement an __aiter__ variant with async for. The accumulated chunks contain delta.content fragments; reconstruct the final text by concatenation during replay.
Step 4: Persist with session replay in mind
A flat JSONL file works for early development. For production, store in a table with indexes on session_id.
CREATE TABLE llm_pairs (
id INTEGER PRIMARY KEY,
session_id TEXT,
turn_id TEXT,
ts REAL,
request_json TEXT,
response_json TEXT
);In Python, replace log_pair with an insert:
import sqlite3
conn = sqlite3.connect("debug.db")
def log_pair(session_id, turn_id, request_json, response_json):
conn.execute(
"INSERT INTO llm_pairs (session_id, turn_id, ts, request_json, response_json) VALUES (?,?,?,?,?)",
(session_id, turn_id, time.time(), json.dumps(request_json), json.dumps(response_json))
)
conn.commit()Now you can query all turns for a session and reconstruct the exact conversation.
Why correlation IDs matter
A session ID groups turns; a turn ID orders them. Without these, you have orphaned prompts and completions that can’t be stitched into a replay. Inject them as headers (x-session-id, x-turn-id) from your application layer so the transport can read them. When you later debug, filter by session_id to see the full trajectory that led to the bad output.
Step 5: Propagate routing and cache hints
When you send requests to a gateway, you may include routing directives or cache-control in headers. Capture those too. For example, n4n.ai honors client routing directives and forwards provider cache-control hints; log extra_headers so you know whether a response was served from cache.
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Hello"}],
extra_headers={
"x-routing-key": "primary",
"x-cache-control": "max-age=3600"
}
)In your transport, copy request.headers into the logged request object. That makes capture request response pairs debugging complete: you see not just what was asked, but how it was routed. If a fallback occurred, the response may carry a different provider label; log the response headers as well.
Step 6: Verify the pipeline end to end
Write a small script that exercises both streaming and non-streaming paths and asserts the logs exist.
import os, json
os.remove("llm_pairs.jsonl") if os.path.exists("llm_pairs.jsonl") else None
# Non-streaming
client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "ping"}],
extra_headers={"x-session-id": "s1", "x-turn-id": "t1"}
)
# Streaming
stream = create_with_capture(client, "s1", "t2", model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "stream"}], stream=True)
for _ in stream:
pass
with open("llm_pairs.jsonl") as f:
lines = f.readlines()
assert len(lines) >= 2, "expected at least two captured pairs"
for line in lines:
obj = json.loads(line)
assert "request" in obj and "response" in obj
print("OK: captured", len(lines), "pairs")Run it with python verify_capture.py. If it prints OK and your DB has rows, the capture layer works. For CI, wrap the assertions in pytest; fail the build if pairs are missing.
Production caveats
- Redaction: LLM requests often contain PII. Hash or drop sensitive fields before writing to disk.
- Volume: Token-level logging for every call gets expensive. Sample 10% of sessions in prod, capture all in staging.
- Concurrency: Use a thread-safe logger (e.g.,
loggingmodule with aQueueHandler) instead of raw file appends. - Streaming memory: Accumulating all chunks is fine for chat, but for long generations consider writing chunks incrementally.
- Retention: Store pairs for a bounded window (e.g., 14 days) to limit compliance exposure.
Capturing request response pairs debugging is not glamorous, but it turns “the model hallucinated” into a reproducible test case. Do it once, correctly, and your debugging cycles drop from hours to minutes.