meida_backend/packages/llm/prompts.py

"""Versioned prompt templates + prompt-cache prefix discipline.

架构文档 §8.1 Prompt 版本: Prompt 存文件（随代码版本化）+ 场景绑定版本号.
架构文档 §7.2 缓存前缀纪律 / TD-04:

    系统提示词与稳定知识片段必须放在 prompt 头部且**不含时间戳、随机 ID、
    用户昵称等变量**，否则缓存永远不命中，输入成本翻数倍。

That rule is unenforceable by review alone — every prompt edit is a chance to
break it, and the damage (a silent 5-10x input-cost increase) is invisible in
functional tests. So it is enforced in code: :func:`validate_cache_prefix` runs
on every gateway call.

File layout::

    prompts/<prompt_id>/<version>.txt

    [system]
    ...fixed, cacheable prefix — no variables at all...

    [user]
    ...template rendered with str.format(**vars)...
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from packages.llm.errors import CachePrefixViolationError, PromptNotFoundError

CachePrefixMode = Literal["strict", "warn", "off"]

_SECTION_RE = re.compile(r"^\[(system|user)\]\s*$", re.MULTILINE)

#: Patterns whose presence in a cached prefix destroys the provider prompt cache.
_VOLATILE_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("iso-date", re.compile(r"\b\d{4}-\d{2}-\d{2}\b")),
    ("iso-time", re.compile(r"\b\d{2}:\d{2}:\d{2}\b")),
    ("cn-date", re.compile(r"\b\d{4}\s*年\s*\d{1,2}\s*月")),
    ("epoch-seconds", re.compile(r"\b1[6-9]\d{8}\b")),
    (
        "uuid",
        re.compile(
            r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
            r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b"
        ),
    ),
    ("long-hex-id", re.compile(r"\b[0-9a-f]{24,}\b")),
    (
        "volatile-placeholder",
        re.compile(
            r"\{[^{}]*(?:nick|user|name|now|today|date|time|stamp|uuid|rand|"
            r"session|request|trace|token|id)[^{}]*\}",
            re.IGNORECASE,
        ),
    ),
)


@dataclass(frozen=True, slots=True)
class CachePrefixViolation:
    kind: str
    sample: str

    def __str__(self) -> str:
        return f"{self.kind}({self.sample!r})"


def find_volatile_tokens(prefix: str) -> list[CachePrefixViolation]:
    """Return every volatile token found in a would-be cached prefix."""
    violations: list[CachePrefixViolation] = []
    for kind, pattern in _VOLATILE_PATTERNS:
        for match in pattern.finditer(prefix):
            violations.append(CachePrefixViolation(kind=kind, sample=match.group(0)))
    return violations


def validate_cache_prefix(
    prefix: str,
    *,
    mode: CachePrefixMode = "strict",
    ref: str = "<inline>",
    on_warning: object | None = None,
) -> list[CachePrefixViolation]:
    """Enforce the cache-prefix discipline.

    ``strict`` raises :class:`CachePrefixViolationError`; ``warn`` returns the
    violations for the caller to log; ``off`` skips the check entirely.
    """
    if mode == "off":
        return []
    violations = find_volatile_tokens(prefix)
    if violations and mode == "strict":
        raise CachePrefixViolationError(ref, [str(v) for v in violations])
    return violations


@dataclass(frozen=True, slots=True)
class PromptTemplate:
    prompt_id: str
    version: str
    #: Fixed, cacheable prefix. Placed first in the message list.
    system_prefix: str
    #: Optional variable part, rendered per call.
    user_template: str = ""

    @property
    def ref(self) -> str:
        return f"{self.prompt_id}@{self.version}"

    def render_user(self, variables: dict[str, str] | None = None) -> str:
        if not self.user_template:
            return ""
        try:
            return self.user_template.format(**(variables or {}))
        except KeyError as exc:
            msg = f"{self.ref}: missing prompt variable {exc.args[0]!r}"
            raise PromptNotFoundError(msg) from exc


def parse_prompt_file(text: str, *, prompt_id: str, version: str) -> PromptTemplate:
    sections: dict[str, str] = {}
    parts = _SECTION_RE.split(text)
    # parts == [preamble, name, body, name, body, ...]
    for name, body in zip(parts[1::2], parts[2::2], strict=True):
        sections[name] = body.strip()
    if "system" not in sections:
        msg = f"{prompt_id}@{version}: prompt file must contain a [system] section"
        raise PromptNotFoundError(msg)
    return PromptTemplate(
        prompt_id=prompt_id,
        version=version,
        system_prefix=sections["system"],
        user_template=sections.get("user", ""),
    )


class PromptRegistry:
    """Loads and caches versioned prompt templates from disk."""

    def __init__(
        self,
        root: Path | str,
        *,
        cache_prefix_mode: CachePrefixMode = "strict",
        suffixes: tuple[str, ...] = (".txt", ".md"),
    ) -> None:
        self._root = Path(root)
        self._mode: CachePrefixMode = cache_prefix_mode
        self._suffixes = suffixes
        self._cache: dict[tuple[str, str], PromptTemplate] = {}

    @property
    def cache_prefix_mode(self) -> CachePrefixMode:
        return self._mode

    def get(self, prompt_id: str, version: str) -> PromptTemplate:
        key = (prompt_id, version)
        cached = self._cache.get(key)
        if cached is not None:
            return cached
        template = self._load(prompt_id, version)
        # Fail fast at load time as well as call time.
        validate_cache_prefix(template.system_prefix, mode=self._mode, ref=template.ref)
        self._cache[key] = template
        return template

    def _load(self, prompt_id: str, version: str) -> PromptTemplate:
        for suffix in self._suffixes:
            path = self._root / prompt_id / f"{version}{suffix}"
            if path.is_file():
                return parse_prompt_file(
                    path.read_text(encoding="utf-8"), prompt_id=prompt_id, version=version
                )
        msg = f"prompt {prompt_id}@{version} not found under {self._root}"
        raise PromptNotFoundError(msg)

    def available_versions(self, prompt_id: str) -> list[str]:
        directory = self._root / prompt_id
        if not directory.is_dir():
            return []
        return sorted(p.stem for p in directory.iterdir() if p.suffix in self._suffixes)