Spaces:
Runtime error
Runtime error
File size: 5,077 Bytes
732b14f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """OpenAI prompt-cache helpers: stable system prefix, cache keys, usage metrics."""
from __future__ import annotations
import hashlib
from typing import Any
import structlog
import tiktoken
from app.config import settings
_log = structlog.get_logger(__name__)
_ENCODING = tiktoken.get_encoding("cl100k_base")
# Static block appended when the system prompt is below OpenAI automatic cache threshold (~1024 tokens).
_CACHE_PAD_HEADER = (
"\n\n--- RICS REPORT SCHEMA REFERENCE (STATIC — DO NOT EDIT) ---\n"
"This block exists only to stabilise the cached system prefix for inference. "
"Section codes follow RICS Home Survey Standard layout (A–L, D1–D9, E1–E9, etc.). "
"Condition ratings: NI, 1, 2, 3. Survey levels: 1 Condition Report, 2 Home Survey, 3 Building Survey. "
"British English only. Non-invention: property-specific facts originate from inspector notes only. "
)
def prompt_caching_active() -> bool:
return bool(settings.enable_async_pipeline and settings.enable_prompt_caching)
def count_prompt_tokens(text: str) -> int:
return len(_ENCODING.encode(text or ""))
def ensure_cacheable_system_prefix(system: str, *, min_tokens: int | None = None) -> str:
"""Pad the system prompt so identical prefixes exceed OpenAI automatic cache minimum."""
floor = int(min_tokens or settings.prompt_cache_min_system_tokens)
body = (system or "").strip()
if count_prompt_tokens(body) >= floor:
return body
pad = _CACHE_PAD_HEADER
# Grow pad deterministically until token floor is met.
while count_prompt_tokens(body + pad) < floor:
pad += (
" Standard paragraphs map notes to HB-BS clauses. "
"Retrieval tiers: document, section, paragraph. "
"Evidence must be traceable to notes or approved standard wording. "
)
return body + pad
def normalize_messages_for_caching(messages: list[Any]) -> list[Any]:
"""Pad the first system message so multi-turn and raw message lists cache reliably."""
if not prompt_caching_active() or not messages:
return messages
out: list[Any] = []
padded_system = False
for msg in messages:
if not isinstance(msg, dict):
out.append(msg)
continue
role = msg.get("role")
content = msg.get("content")
if (
not padded_system
and role == "system"
and isinstance(content, str)
):
out.append({**msg, "content": ensure_cacheable_system_prefix(content)})
padded_system = True
else:
out.append(dict(msg))
return out
def build_chat_messages(*, system: str, user: str) -> list[dict[str, str]]:
"""System (static) first, user (dynamic) second — maximises prefix cache hits."""
sys = ensure_cacheable_system_prefix(system) if prompt_caching_active() else system
return [
{"role": "system", "content": sys},
{"role": "user", "content": user},
]
def prompt_cache_key(
*,
phase: str,
model: str,
survey_level: int | None = None,
interference_level: str | None = None,
tenant_id: str | None = None,
) -> str:
il = (interference_level or "").strip().lower() or "legacy"
lvl = int(survey_level or 0)
tenant = (tenant_id or "").strip() or "_"
raw = f"rics|{tenant}|{model}|{phase}|L{lvl}|{il}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
def openai_extra_kwargs(
*,
phase: str,
model: str,
survey_level: int | None = None,
interference_level: str | None = None,
tenant_id: str | None = None,
) -> dict[str, Any]:
if not prompt_caching_active():
return {}
return {"prompt_cache_key": prompt_cache_key(
phase=phase,
model=model,
survey_level=survey_level,
interference_level=interference_level,
tenant_id=tenant_id,
)}
def log_openai_cache_usage(
response: Any,
*,
phase: str,
section_id: str | None,
) -> bool | None:
"""Log cache_read_input_tokens / input_tokens; return True if cache was used."""
usage = getattr(response, "usage", None)
if usage is None:
return None
input_tokens = int(getattr(usage, "prompt_tokens", 0) or getattr(usage, "input_tokens", 0) or 0)
cached = int(getattr(usage, "prompt_tokens_details", None) and getattr(
usage.prompt_tokens_details, "cached_tokens", 0
) or getattr(usage, "cache_read_input_tokens", 0) or 0)
# OpenAI SDK: cached_tokens may live on prompt_tokens_details
details = getattr(usage, "prompt_tokens_details", None)
if details is not None:
cached = int(getattr(details, "cached_tokens", 0) or 0)
rate = (cached / input_tokens) if input_tokens > 0 else 0.0
cache_hit = cached > 0
_log.info(
event="openai_prompt_cache",
phase=phase,
section_id=section_id,
cache_hit=cache_hit,
cache_read_input_tokens=cached,
input_tokens=input_tokens,
cache_hit_rate=round(rate, 4),
)
return cache_hit
|