138 lines
4.5 KiB
Python
138 lines
4.5 KiB
Python
"""Runtime-editable LLM system prompts, persisted in app_settings.
|
|
|
|
Three resolution levels with fallback (more specific → less specific):
|
|
1. `llm_system_prompt:<department_id>:<vertical>:<section_slug>` — section override
|
|
2. `llm_system_prompt:<department_id>:<vertical>` — department vertical override
|
|
3. built-in DEFAULT_RE_SYSTEM_PROMPT / DEFAULT_HR_SYSTEM_PROMPT
|
|
|
|
The prompt is read on every classification call but cached for a short
|
|
window so the DB isn't hit per-message. Edits via the API invalidate the
|
|
cache for that level, so a save in the UI takes effect within seconds.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Literal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
from parser_bot.db.models import AppSetting
|
|
from parser_bot.db.session import session_scope
|
|
|
|
Vertical = Literal["real_estate", "hr"]
|
|
|
|
_KEY_PREFIX = "llm_system_prompt:"
|
|
_CACHE_TTL_S = 5.0
|
|
_cache: dict[str, tuple[float, str | None]] = {}
|
|
|
|
|
|
def _key(
|
|
vertical: Vertical,
|
|
department_id: str | None,
|
|
section_slug: str | None = None,
|
|
) -> str:
|
|
dept = department_id or "global"
|
|
if section_slug:
|
|
return f"{_KEY_PREFIX}{dept}:{vertical}:{section_slug}"
|
|
return f"{_KEY_PREFIX}{dept}:{vertical}"
|
|
|
|
|
|
async def _load(key: str) -> str | None:
|
|
"""Read a stored prompt by exact key. None if missing or empty."""
|
|
now = time.monotonic()
|
|
cached_at, cached_value = _cache.get(key, (0.0, None))
|
|
if now - cached_at < _CACHE_TTL_S:
|
|
return cached_value
|
|
|
|
async with session_scope() as session:
|
|
row = await session.execute(
|
|
select(AppSetting.value).where(AppSetting.key == key)
|
|
)
|
|
value = row.scalar_one_or_none()
|
|
|
|
text = value if isinstance(value, str) and value.strip() else None
|
|
_cache[key] = (now, text)
|
|
return text
|
|
|
|
|
|
async def resolve(
|
|
vertical: Vertical, department_id: str | None, section_slug: str | None, default: str
|
|
) -> str:
|
|
"""Pick the most specific prompt available, falling back to `default`.
|
|
|
|
Always consults section-level → vertical-level → default. This is what
|
|
the classifier uses for every message.
|
|
"""
|
|
if section_slug:
|
|
text = await _load(_key(vertical, department_id, section_slug))
|
|
if text is not None:
|
|
return text
|
|
text = await _load(_key(vertical, department_id))
|
|
if text is not None:
|
|
return text
|
|
return default
|
|
|
|
|
|
async def get(
|
|
vertical: Vertical, department_id: str | None, section_slug: str | None, default: str
|
|
) -> tuple[str, str]:
|
|
"""For the settings UI: return (text, source) where source is one of
|
|
'section' | 'vertical' | 'default'. Lets the editor show which override
|
|
is currently active without a second round-trip.
|
|
"""
|
|
if section_slug:
|
|
text = await _load(_key(vertical, department_id, section_slug))
|
|
if text is not None:
|
|
return text, "section"
|
|
text = await _load(_key(vertical, department_id))
|
|
if text is not None:
|
|
return text, "vertical"
|
|
return default, "default"
|
|
|
|
|
|
async def set_prompt(
|
|
vertical: Vertical, department_id: str | None, section_slug: str | None, text: str
|
|
) -> None:
|
|
"""Save a new prompt at the given level (section or vertical)."""
|
|
if not isinstance(text, str) or not text.strip():
|
|
raise ValueError("prompt must be a non-empty string")
|
|
key = _key(vertical, department_id, section_slug)
|
|
async with session_scope() as session:
|
|
stmt = (
|
|
pg_insert(AppSetting)
|
|
.values(key=key, value=text)
|
|
.on_conflict_do_update(
|
|
index_elements=["key"], set_={"value": text}
|
|
)
|
|
)
|
|
await session.execute(stmt)
|
|
invalidate(key)
|
|
|
|
|
|
async def reset(
|
|
vertical: Vertical, department_id: str | None, section_slug: str | None
|
|
) -> None:
|
|
"""Drop the override at the given level."""
|
|
key = _key(vertical, department_id, section_slug)
|
|
async with session_scope() as session:
|
|
await session.execute(
|
|
AppSetting.__table__.delete().where(AppSetting.key == key)
|
|
)
|
|
invalidate(key)
|
|
|
|
|
|
def invalidate(key: str | None = None) -> None:
|
|
if key is None:
|
|
_cache.clear()
|
|
else:
|
|
_cache.pop(key, None)
|
|
|
|
|
|
async def is_overridden(
|
|
vertical: Vertical, department_id: str | None, section_slug: str | None = None
|
|
) -> bool:
|
|
"""True iff a custom prompt is stored at this exact level."""
|
|
text = await _load(_key(vertical, department_id, section_slug))
|
|
return text is not None
|