Files
oO/ml/serving/prompts.py
alvis 430804e9a5 feat(ml): prompt registry + per-request variant selection
Replaces the hardcoded "v1" label with a real prompt registry:

  ml/serving/prompts.py       — keyed by version: v1 (baseline),
                                v2-mentor (calm/specific persona),
                                v3-few-shot (v1 persona + curated examples)
  ml/serving/main.py          — POST /generate accepts optional prompt_version,
                                422 on unknown, echoes the version actually used
                                back in the response
  services/api/src/config.ts  — TIP_PROMPT_VERSION: empty / single / comma-list
                                (uniform random per request)
  services/api/src/routes/recommender.ts
                              — pickPromptVersion() drives selection; the
                                response's prompt_version (not a stale TS
                                constant) is what lands in tip_scores so the
                                #92 reward-analytics dashboard shows real
                                per-variant reaction rates

Closes #84.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-24 15:44:04 +00:00

106 lines
4.1 KiB
Python

"""Prompt registry for tip generation (#84).
Each entry is an immutable (system, build_user) pair keyed by a stable version
string. Adding a new version here makes it selectable via the ``prompt_version``
field on ``POST /generate`` — the selected version flows back in the response
and is persisted to ``tip_scores.prompt_version`` so the admin reward-analytics
dashboard can bucket reactions per variant.
Versions:
v1 — neutral "productivity coach" baseline (unchanged from ffdf707).
v2-mentor — calm/specific mentor persona; same structural prompt as v1.
v3-few-shot — v1 persona plus two curated example tips inside the system prompt.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Callable, Protocol
class _Ctx(Protocol):
tasks: list[dict]
hour_of_day: int
day_of_week: int
extra: dict
@dataclass(frozen=True)
class Prompt:
version: str
system: str
build_user: Callable[["_Ctx", int], str]
def _base_user_lines(ctx: "_Ctx") -> list[str]:
lines = [f"Time: {ctx.hour_of_day:02d}:00, day_of_week={ctx.day_of_week}"]
if ctx.tasks:
overdue = [t for t in ctx.tasks if t.get("is_overdue")]
lines.append(f"Tasks: {len(ctx.tasks)} total, {len(overdue)} overdue")
for t in ctx.tasks[:5]:
due = t.get("due_date", "no due date")
lines.append(f" - [{t.get('priority','?')}] {t.get('content','?')} (due: {due})")
for k, v in ctx.extra.items():
lines.append(f"{k}: {v}")
return lines
def _build_user_v1(ctx: "_Ctx", n: int) -> str:
return "\n".join([*_base_user_lines(ctx), f"\nGenerate {n} tips as a JSON array."])
_SYS_V1 = (
"You are a personal productivity coach. "
"Given the user's current context, generate actionable, specific tips. "
"Respond ONLY with a JSON array of objects, each with keys: "
'"id" (short slug), "content" (the tip, ≤2 sentences), "rationale" (why now, ≤1 sentence). '
"No markdown, no prose outside the JSON array."
)
_SYS_V2_MENTOR = (
"You are a calm, wise mentor — the kind who has seen a thousand people get stuck on "
"the same thing and knows when a small, concrete step unblocks them. Your tips are "
"earned, never generic; they reference the user's specific context and respect that "
"their time is short. Speak plainly. Prefer one precise action over vague encouragement. "
"Respond ONLY with a JSON array of objects, each with keys: "
'"id" (short slug), "content" (the tip, ≤2 sentences), "rationale" (why now, ≤1 sentence). '
"No markdown, no prose outside the JSON array."
)
# Two curated examples illustrate the shape we want: (1) a precise micro-action
# for an overdue item, and (2) a time-aware tip that trades tiny effort now for
# reduced friction later. Kept inside the system prompt so token cost is paid
# once per conversation and not per user turn.
_SYS_V3_FEW_SHOT = _SYS_V1 + (
"\n\nExamples of the shape and tone to aim for:\n"
'[{"id":"overdue-anchor",'
'"content":"Spend the next 12 minutes on \\"Call dentist\\" — set a timer and stop '
'when it rings, done or not.",'
'"rationale":"Overdue 6 days; a fixed micro-session breaks the avoidance loop."},'
'{"id":"evening-wind-down",'
'"content":"Pick one task from tomorrow\'s list and write its first line now while '
'context is fresh.",'
'"rationale":"It is 21:00; tomorrow-you will thank present-you for not starting cold."}]'
)
PROMPTS: dict[str, Prompt] = {
"v1": Prompt("v1", _SYS_V1, _build_user_v1),
"v2-mentor": Prompt("v2-mentor", _SYS_V2_MENTOR, _build_user_v1),
"v3-few-shot": Prompt("v3-few-shot", _SYS_V3_FEW_SHOT, _build_user_v1),
}
def default_version() -> str:
return os.getenv("DEFAULT_PROMPT_VERSION", "v1")
def get_prompt(version: str | None) -> Prompt:
"""Look up a prompt by version. Falls back to ``DEFAULT_PROMPT_VERSION`` when
``version`` is ``None``; raises :class:`KeyError` for unknown versions so
callers can surface a 422 to clients."""
v = version or default_version()
if v not in PROMPTS:
raise KeyError(v)
return PROMPTS[v]