feat(search): retrieval telemetry — log every semantic retrieval
CI & Build / Python lint (push) Successful in 2s
CI & Build / integration (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Successful in 44s
CI & Build / Build & push image (push) Successful in 58s

Add retrieval_logs (migration 0068) + services/retrieval_telemetry with a
fire-and-forget record_retrieval(), wired into the MCP search tool
(source=mcp_search) and the REST search route (source=rest_search). Captures
query, effective params, and the per-result score distribution so KB-injection
thresholds can be tuned from data rather than guessed.

Scribe: project 2, milestone 93, task 1032.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Xz4j1H7pjYSjKsEpgcNH5E
This commit is contained in:
2026-06-22 20:10:15 -04:00
parent 513019786e
commit 807f478cac
7 changed files with 384 additions and 2 deletions
+11 -1
View File
@@ -7,8 +7,11 @@ working. Differences from fable-mcp:
"""
from __future__ import annotations
import time
from scribe.mcp._context import current_user_id
from scribe.services.embeddings import semantic_search_notes
from scribe.services.embeddings import DEFAULT_SIMILARITY_THRESHOLD, semantic_search_notes
from scribe.services.retrieval_telemetry import record_retrieval
async def search(
@@ -43,10 +46,17 @@ async def search(
uid = current_user_id()
limit = max(1, min(limit, 50))
is_task = {"note": False, "task": True}.get(content_type) # None => any
t0 = time.perf_counter()
raw = await semantic_search_notes(
uid, q, limit=limit, is_task=is_task,
project_id=project_id or None,
)
record_retrieval(
user_id=uid, source="mcp_search", query=q,
threshold=DEFAULT_SIMILARITY_THRESHOLD, limit=limit,
project_id=project_id or None, is_task=is_task, results=raw,
duration_ms=(time.perf_counter() - t0) * 1000.0,
)
return {
"results": [
{
+1
View File
@@ -26,6 +26,7 @@ from scribe.models.app_log import AppLog # noqa: E402, F401
from scribe.models.password_reset import PasswordResetToken # noqa: E402, F401
from scribe.models.invitation import InvitationToken # noqa: E402, F401
from scribe.models.embedding import NoteEmbedding # noqa: E402, F401
from scribe.models.retrieval_log import RetrievalLog # noqa: E402, F401
from scribe.models.project import Project # noqa: E402, F401
from scribe.models.event import Event # noqa: E402, F401
from scribe.models.milestone import Milestone # noqa: E402, F401
+70
View File
@@ -0,0 +1,70 @@
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, Float, Index, Integer, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from scribe.models import Base
class RetrievalLog(Base):
"""One row per semantic-retrieval call, for KB-injection tuning.
Captures what a query asked for, what came back, and the score
distribution of the results — the empirical basis for tuning the
similarity threshold and top-k per surface. `result_ids` holds the ranked
hits (id + score + rank) so a later pass can correlate "what we surfaced"
against "what the agent then fetched/referenced".
Deliberately FK-free on user_id (mirrors AppLog): telemetry should outlive
the row it describes, and a deleted user shouldn't cascade away history.
"""
__tablename__ = "retrieval_logs"
id: Mapped[int] = mapped_column(primary_key=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
)
user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
# Retrieval surface: 'mcp_search' | 'rest_search' | 'auto_inject' | ...
source: Mapped[str] = mapped_column(Text, nullable=False)
query: Mapped[str | None] = mapped_column(Text, nullable=True)
# Effective parameters actually used for this call.
threshold: Mapped[float | None] = mapped_column(Float, nullable=True)
limit_n: Mapped[int | None] = mapped_column(Integer, nullable=True)
project_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
# The content-type filter as passed to semantic_search_notes: True=tasks,
# False=notes, NULL=any.
is_task: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
result_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
top_score: Mapped[float | None] = mapped_column(Float, nullable=True)
min_score: Mapped[float | None] = mapped_column(Float, nullable=True)
# [{"id": int, "score": float, "rank": int}, ...], highest-first.
result_ids: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
duration_ms: Mapped[float | None] = mapped_column(Float, nullable=True)
__table_args__ = (
Index("ix_retrieval_logs_created_at", "created_at"),
Index("ix_retrieval_logs_user_id", "user_id"),
Index("ix_retrieval_logs_source", "source"),
Index("ix_retrieval_logs_source_created_at", "source", created_at.desc()),
)
def to_dict(self) -> dict:
return {
"id": self.id,
"created_at": self.created_at.isoformat() if self.created_at else None,
"user_id": self.user_id,
"source": self.source,
"query": self.query,
"threshold": self.threshold,
"limit_n": self.limit_n,
"project_id": self.project_id,
"is_task": self.is_task,
"result_count": self.result_count,
"top_score": self.top_score,
"min_score": self.min_score,
"result_ids": self.result_ids,
"duration_ms": self.duration_ms,
}
+15 -1
View File
@@ -1,7 +1,14 @@
import time
from quart import Blueprint, jsonify, request
from scribe.auth import login_required, get_current_user_id
from scribe.services.embeddings import semantic_search_notes
from scribe.services.retrieval_telemetry import record_retrieval
# This route searches with a looser floor than the MCP tool default — it powers
# an interactive feed where loosely-related hits still have value.
_REST_SEARCH_THRESHOLD = 0.3
search_bp = Blueprint("search", __name__, url_prefix="/api/search")
@@ -27,8 +34,15 @@ async def search_route():
limit = min(request.args.get("limit", 10, type=int), 50)
is_task = _content_type_to_is_task(content_type)
t0 = time.perf_counter()
results = await semantic_search_notes(
uid, q, limit=limit, is_task=is_task, threshold=0.3
uid, q, limit=limit, is_task=is_task, threshold=_REST_SEARCH_THRESHOLD
)
record_retrieval(
user_id=uid, source="rest_search", query=q,
threshold=_REST_SEARCH_THRESHOLD, limit=limit,
project_id=None, is_task=is_task, results=results,
duration_ms=(time.perf_counter() - t0) * 1000.0,
)
return jsonify({
"results": [
+115
View File
@@ -0,0 +1,115 @@
"""Retrieval telemetry — one RetrievalLog row per semantic-retrieval call.
This is the empirical basis for KB-injection tuning: it records what each query
asked for, the score distribution of what came back, and the effective params,
so the similarity threshold and top-k can be tuned from data rather than guessed.
Design notes:
- Fire-and-forget, mirroring upsert_note_embedding: `record_retrieval` extracts
the primitives it needs SYNCHRONOUSLY (while the caller's Note objects are
still valid) and schedules the DB insert as a background task, so logging
never adds latency to — or can break — the search response.
- Result objects are reduced to {id, score, rank} before scheduling; the
background writer touches only plain data, never a possibly-detached ORM row.
- Every failure path is swallowed: telemetry must never take down retrieval.
"""
from __future__ import annotations
import asyncio
import logging
from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.retrieval_log import RetrievalLog
logger = logging.getLogger(__name__)
def _build_payload(
*,
user_id: int | None,
source: str,
query: str | None,
threshold: float | None,
limit: int | None,
project_id: int | None,
is_task: bool | None,
results: list[tuple[float, Note]],
duration_ms: float | None,
) -> dict:
"""Reduce a retrieval call to a flat, JSON-safe RetrievalLog payload.
Pure and synchronous (no DB, no event loop) so it is unit-testable and safe
to run inline before scheduling the write. `results` is the
`(score, Note)` list from semantic_search_notes, already highest-first.
"""
items = [
{"id": int(note.id), "score": round(float(score), 5), "rank": rank}
for rank, (score, note) in enumerate(results)
]
scores = [it["score"] for it in items]
return {
"user_id": user_id,
"source": source,
"query": query,
"threshold": threshold,
"limit_n": limit,
"project_id": project_id,
"is_task": is_task,
"result_count": len(items),
"top_score": (scores[0] if scores else None),
"min_score": (scores[-1] if scores else None),
"result_ids": items,
"duration_ms": (round(duration_ms, 2) if duration_ms is not None else None),
}
async def _insert_retrieval_log(payload: dict) -> None:
"""Persist one RetrievalLog row. Best-effort: all errors are swallowed."""
try:
async with async_session() as session:
session.add(RetrievalLog(**payload))
await session.commit()
except Exception:
logger.debug("retrieval telemetry write skipped", exc_info=True)
def record_retrieval(
*,
user_id: int | None,
source: str,
query: str | None,
threshold: float | None,
limit: int | None,
project_id: int | None,
is_task: bool | None,
results: list[tuple[float, Note]],
duration_ms: float | None = None,
) -> None:
"""Fire-and-forget: record one retrieval call.
Builds the payload inline (synchronously) then schedules the insert so the
caller returns immediately. Never raises — telemetry must not affect search.
"""
try:
payload = _build_payload(
user_id=user_id,
source=source,
query=query,
threshold=threshold,
limit=limit,
project_id=project_id,
is_task=is_task,
results=results,
duration_ms=duration_ms,
)
except Exception:
logger.debug("retrieval telemetry payload build failed", exc_info=True)
return
try:
asyncio.get_running_loop().create_task(_insert_retrieval_log(payload))
except RuntimeError:
# No running loop (e.g. called from sync context outside the app) —
# skip rather than block. The app paths always run on the loop.
logger.debug("retrieval telemetry skipped — no running event loop")