How to log vector search queries for debugging

Logging vector search queries is the difference between guessing why a semantic search returned junk and tracing the exact embedding, filter, and latency that produced it. This tutorial builds a minimal instrumentation layer around a RAG retrieval path using Python, pgvector, and an OpenAI-compatible embeddings endpoint, so you can reproduce any bad search result from logs alone.

Prerequisites

• Python 3.11 or newer
• A PostgreSQL instance with the pgvector extension installed
• psycopg (v3) and openai Python packages (pip install psycopg openai)
• An embedding API key (OpenAI, or any OpenAI-compatible gateway)
• A documents table defined as:

CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE documents (
    id uuid PRIMARY KEY,
    content text,
    topic text,
    embedding vector(1536)
);
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

You should already have rows loaded. We focus exclusively on making the read path observable.

Step 1: Structured logging with context

Use the standard logging module with a JSON formatter. Structured lines let you grep by trace_id or event in production and pipe them to any log agent without regex gymnastics.

import logging
import json
import sys
from contextvars import ContextVar

trace_id_ctx = ContextVar("trace_id", default="none")

class JsonFormatter(logging.Formatter):
    def format(self, record):
        base = {
            "ts": self.formatTime(record),
            "level": record.levelname,
            "event": record.getMessage(),
            "trace_id": trace_id_ctx.get(),
        }
        # merge any extra dict passed via logger.info("msg", extra={...})
        for key, val in record.__dict__.items():
            if key not in ("ts", "level", "event", "trace_id", "message", "msg", "args", "exc_info", "stack_info", "levelno", "levelname", "pathname", "filename", "module", "lineno", "funcName", "created", "msecs", "relativeCreated", "thread", "threadName", "processName", "process", "getMessage"):
                base[key] = val
        if record.exc_info:
            base["exc"] = self.formatException(record.exc_info)
        return json.dumps(base)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("vectordb")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

Step 2: Wrap the embedding call

The query vector is the root cause of most relevance bugs. Log the model, a hash of the input, the vector norm, and the upstream request ID. If you route embeddings through n4n.ai, the same OpenAI-compatible client works and the response headers include a per-request ID you can record for metering reconciliation.

from openai import OpenAI
import hashlib

client = OpenAI(api_key="sk-...")  # or base_url="https://api.n4n.ai/v1"

def embed_query(text: str) -> tuple[list[float], dict]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
        encoding_format="float",
    )
    vec = resp.data[0].embedding
    norm = sum(x*x for x in vec) ** 0.5
    meta = {
        "model": resp.model,
        "input_hash": hashlib.sha256(text.encode()).hexdigest()[:16],
        "vec_norm": round(norm, 4),
        "upstream_id": resp.headers.get("x-request-id", "n/a"),
    }
    logger.info("embed_query", extra=meta)
    return vec, meta

Expected log line:

{"ts": "2024-05-12 10:22:01", "level": "INFO", "event": "embed_query", "trace_id": "none", "model": "text-embedding-3-small", "input_hash": "a1b2c3d4e5f6a7b8", "vec_norm": 1.0002, "upstream_id": "req_123"}

Step 3: Instrument the pgvector search

Write a function that runs cosine similarity with optional metadata filters. When logging vector search queries, never log the raw 1536-float vector—log its norm and dimension instead to keep log volume sane. Capture the SQL, bound parameters, k, and elapsed milliseconds.

import time
import psycopg

conn = psycopg.connect("postgres://user:pass@localhost/db")

def search_docs(vec: list[float], k: int = 5, filter_topic: str | None = None):
    start = time.monotonic()
    sql = """
        SELECT id, content, 1 - (embedding <=> %s) AS score
        FROM documents
        WHERE (%s IS NULL OR topic = %s)
        ORDER BY embedding <=> %s
        LIMIT %s
    """
    params = (vec, filter_topic, filter_topic, vec, k)
    with conn.cursor() as cur:
        cur.execute(sql, params)
        rows = cur.fetchall()
    elapsed_ms = round((time.monotonic() - start) * 1000, 2)
    logger.info("vector_search", extra={
        "sql": sql.strip(),
        "k": k,
        "filter": filter_topic,
        "result_ids": [str(r[0]) for r in rows],
        "top_score": rows[0][2] if rows else None,
        "elapsed_ms": elapsed_ms,
        "vec_dim": len(vec),
    })
    return rows

Sample output after a query for “postgres indexing”:

{"ts": "2024-05-12 10:22:02", "level": "INFO", "event": "vector_search", "trace_id": "none", "k": 5, "filter": null, "result_ids": ["uuid1","uuid2"], "top_score": 0.81, "elapsed_ms": 12.4, "vec_dim": 1536}

Step 4: Propagate a trace ID

In a web service, set trace_id_ctx from the incoming request header. For a script, generate one per call so embedding and search logs share a key.

import uuid

def handle_search(query: str, topic: str | None = None):
    trace_id_ctx.set(uuid.uuid4().hex[:12])
    vec, _ = embed_query(query)
    rows = search_docs(vec, k=5, filter_topic=topic)
    logger.info("search_complete", extra={"num_results": len(rows)})
    return rows

Now both the embedding and search logs share trace_id, so you can reconstruct the full path:

grep '"trace_id":"9f3a1b2c4d5e"' app.log

Step 5: Capture empty or degraded results

Add a warning when the top score is suspiciously low or zero rows return. This is where logging vector search queries pays off—you see whether the embedding diverged or the filter excluded everything.

def search_docs_with_guard(vec, k=5, filter_topic=None):
    rows = search_docs(vec, k, filter_topic)
    if not rows:
        logger.warning("empty_result", extra={"filter": filter_topic})
    elif rows[0][2] < 0.65:
        logger.warning("low_relevance", extra={"top_score": rows[0][2]})
    return rows

Step 6: Sampling for high throughput

At scale, logging every vector query is noisy. Sample 5% via a deterministic hash of the trace ID. Keep errors and warnings at 100%.

def should_log(trace_id: str, rate: float = 0.05) -> bool:
    return (int(trace_id, 16) % 100) < (rate * 100)

def handle_search_sampled(query: str, topic: str | None = None):
    tid = uuid.uuid4().hex[:12]
    trace_id_ctx.set(tid)
    vec, _ = embed_query(query)  # embedding call logged inside
    if should_log(tid):
        return search_docs_with_guard(vec, k=5, filter_topic=topic)
    # non-logging fast path
    with conn.cursor() as cur:
        cur.execute(
            "SELECT id, content, 1 - (embedding <=> %s) AS score FROM documents WHERE (%s IS NULL OR topic = %s) ORDER BY embedding <=> %s LIMIT %s",
            (vec, topic, topic, vec, 5),
        )
        return cur.fetchall()

Local verification

Run a complete cycle to confirm the wiring:

if __name__ == "__main__":
    rows = handle_search("how to tune pgvector index", topic="postgres")
    print(f"returned {len(rows)} rows")

Console shows two JSON lines (embed + search) sharing "trace_id", then the printed row count. If you see empty_result or low_relevance warnings, the guard works.

Debugging patterns from logs

When a user reports “search returned irrelevant docs”, pull the trace:

1. Check embed_query input_hash—confirm the text sent matches what the user typed (truncation or middleware bugs are common).
2. Check vec_norm—a norm far from 1.0 indicates a broken pooling or normalization step upstream.
3. Check vector_search elapsed_ms—if consistently high, your IVFFlat lists parameter or HNSW graph needs tuning, or you are missing an index.
4. Check filter—a stale topic filter silently drops candidates before similarity scoring.

If you see low_relevance with a healthy norm and low latency, the issue is data coverage, not your query code.

Instrumentation checklist

• Log embedding model and upstream request ID for every query path.
• Never log raw vectors; log dimension and norm.
• Correlate embedding and DB calls with a contextvar trace ID.
• Warn on empty or low-score results; sample verbose logs in production.
• Keep error logs unsampled.

That is the minimal surface for logging vector search queries that survives contact with real traffic.