3a00ff4d33
- Complete Fastify gateway with 8-stage pipeline - Circuit breaker (opossum) per model tier - Rate limiting per caller - Ban list validation (EN/DE/auto-detected) - TIP validator (SFF-8024, part numbers, wavelengths) - Prometheus metrics - pg-boss async queue - PostgreSQL audit log + review queue - 9 prompt templates (TIP, LinkedIn, ShieldX) - Learning engine scaffolding - Auto-learning: ban-list, few-shot, routing, prompt optimizer
307 lines
9.3 KiB
Python
307 lines
9.3 KiB
Python
"""
|
|
scheduler.py - Cron-based trigger logic for fine-tuning runs.
|
|
|
|
Evaluates whether each fine-tuning strategy (task-specific LoRA,
|
|
general SFT, DPO) should fire based on corpus size and recency of
|
|
previous runs. All DB access is read-only; no mutations happen here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
import psycopg2
|
|
import psycopg2.extras
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Thresholds (immutable constants — never mutate at runtime)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
FINE_TUNING_THRESHOLDS: dict = {
|
|
"task_specific": {
|
|
"min_positive_examples": 100,
|
|
"min_confidence": 7.5,
|
|
"min_days_since_last_run": 7,
|
|
},
|
|
"general": {
|
|
"min_positive_examples": 500,
|
|
"min_confidence": 7.0,
|
|
"min_days_since_last_run": 14,
|
|
},
|
|
"dpo": {
|
|
"min_preference_pairs": 50,
|
|
"min_days_since_last_run": 7,
|
|
},
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _days_since_last_run(
|
|
conn: psycopg2.extensions.connection,
|
|
run_type: str,
|
|
task_type: Optional[str] = None,
|
|
) -> float:
|
|
"""
|
|
Return days elapsed since the most recent *completed* run of the given type.
|
|
Returns a very large number when no prior run exists (always trigger).
|
|
"""
|
|
params: dict = {"run_type": run_type}
|
|
|
|
if task_type is not None:
|
|
sql = """
|
|
SELECT MAX(completed_at) AS last_run
|
|
FROM fine_tuning_runs
|
|
WHERE
|
|
run_type = %(run_type)s
|
|
AND task_type = %(task_type)s
|
|
AND status = 'completed'
|
|
"""
|
|
params["task_type"] = task_type
|
|
else:
|
|
sql = """
|
|
SELECT MAX(completed_at) AS last_run
|
|
FROM fine_tuning_runs
|
|
WHERE run_type = %(run_type)s AND status = 'completed'
|
|
"""
|
|
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(sql, params)
|
|
row = cur.fetchone()
|
|
|
|
if row is None or row["last_run"] is None:
|
|
return float("inf")
|
|
|
|
last_run: datetime = row["last_run"]
|
|
if last_run.tzinfo is None:
|
|
last_run = last_run.replace(tzinfo=timezone.utc)
|
|
|
|
elapsed = (datetime.now(timezone.utc) - last_run).total_seconds() / 86400
|
|
return elapsed
|
|
|
|
|
|
def _count_available_positive(
|
|
conn: psycopg2.extensions.connection,
|
|
task_type: Optional[str],
|
|
min_confidence: float,
|
|
) -> int:
|
|
"""Count unused positive examples meeting the confidence bar."""
|
|
params: dict = {"min_confidence": min_confidence}
|
|
sql_base = """
|
|
SELECT COUNT(*) AS cnt
|
|
FROM learning_corpus
|
|
WHERE
|
|
status = 'approved'
|
|
AND confidence_score >= %(min_confidence)s
|
|
AND used_in_training IS NULL
|
|
AND input_text IS NOT NULL
|
|
AND output_text IS NOT NULL
|
|
AND system_prompt IS NOT NULL
|
|
"""
|
|
if task_type is not None:
|
|
sql_base += " AND task_type = %(task_type)s"
|
|
params["task_type"] = task_type
|
|
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(sql_base, params)
|
|
row = cur.fetchone()
|
|
|
|
return int(row["cnt"]) if row else 0
|
|
|
|
|
|
def _count_available_dpo_pairs(conn: psycopg2.extensions.connection) -> int:
|
|
"""Count unused preference pairs for DPO training."""
|
|
sql = """
|
|
SELECT COUNT(*) AS cnt
|
|
FROM learning_corpus
|
|
WHERE
|
|
human_edited = TRUE
|
|
AND edited_output IS NOT NULL
|
|
AND edited_output <> output_text
|
|
AND used_in_dpo_training IS NULL
|
|
AND input_text IS NOT NULL
|
|
AND output_text IS NOT NULL
|
|
"""
|
|
with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
|
|
cur.execute(sql)
|
|
row = cur.fetchone()
|
|
|
|
return int(row["cnt"]) if row else 0
|
|
|
|
|
|
def _is_run_in_progress(
|
|
conn: psycopg2.extensions.connection,
|
|
run_type: str,
|
|
task_type: Optional[str] = None,
|
|
) -> bool:
|
|
"""Return True if a run of this type is currently running or queued."""
|
|
params: dict = {"run_type": run_type}
|
|
sql_base = """
|
|
SELECT 1 FROM fine_tuning_runs
|
|
WHERE run_type = %(run_type)s AND status IN ('queued', 'running')
|
|
"""
|
|
if task_type is not None:
|
|
sql_base += " AND task_type = %(task_type)s"
|
|
params["task_type"] = task_type
|
|
|
|
sql_base += " LIMIT 1"
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql_base, params)
|
|
return cur.fetchone() is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public trigger functions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def should_trigger_task_specific(
|
|
conn: psycopg2.extensions.connection,
|
|
task_type: str,
|
|
) -> bool:
|
|
"""
|
|
Return True if a task-specific LoRA run should be started for task_type.
|
|
|
|
Conditions (all must be true):
|
|
1. Enough unused positive examples (>= 100 with confidence >= 7.5)
|
|
2. No run of this type is currently in progress for this task_type
|
|
3. At least 7 days since the last completed run for this task_type
|
|
"""
|
|
thresholds = FINE_TUNING_THRESHOLDS["task_specific"]
|
|
|
|
if _is_run_in_progress(conn, "task_specific", task_type):
|
|
logger.debug("should_trigger_task_specific(%s): run already in progress", task_type)
|
|
return False
|
|
|
|
count = _count_available_positive(
|
|
conn, task_type, thresholds["min_confidence"]
|
|
)
|
|
if count < thresholds["min_positive_examples"]:
|
|
logger.debug(
|
|
"should_trigger_task_specific(%s): only %d examples (need %d)",
|
|
task_type,
|
|
count,
|
|
thresholds["min_positive_examples"],
|
|
)
|
|
return False
|
|
|
|
days = _days_since_last_run(conn, "task_specific", task_type)
|
|
if days < thresholds["min_days_since_last_run"]:
|
|
logger.debug(
|
|
"should_trigger_task_specific(%s): last run %.1f days ago (need %d)",
|
|
task_type,
|
|
days,
|
|
thresholds["min_days_since_last_run"],
|
|
)
|
|
return False
|
|
|
|
logger.info(
|
|
"should_trigger_task_specific(%s): TRIGGER — %d examples, %.1f days since last run",
|
|
task_type,
|
|
count,
|
|
days,
|
|
)
|
|
return True
|
|
|
|
|
|
def should_trigger_general(conn: psycopg2.extensions.connection) -> bool:
|
|
"""
|
|
Return True if a general (cross-task) SFT run should be started.
|
|
|
|
Conditions (all must be true):
|
|
1. Total unused positive examples across all tasks >= 500
|
|
2. No general run currently in progress
|
|
3. At least 14 days since the last completed general run
|
|
"""
|
|
thresholds = FINE_TUNING_THRESHOLDS["general"]
|
|
|
|
if _is_run_in_progress(conn, "general"):
|
|
logger.debug("should_trigger_general: run already in progress")
|
|
return False
|
|
|
|
count = _count_available_positive(conn, None, thresholds["min_confidence"])
|
|
if count < thresholds["min_positive_examples"]:
|
|
logger.debug(
|
|
"should_trigger_general: only %d examples (need %d)",
|
|
count,
|
|
thresholds["min_positive_examples"],
|
|
)
|
|
return False
|
|
|
|
days = _days_since_last_run(conn, "general")
|
|
if days < thresholds["min_days_since_last_run"]:
|
|
logger.debug(
|
|
"should_trigger_general: last run %.1f days ago (need %d)",
|
|
days,
|
|
thresholds["min_days_since_last_run"],
|
|
)
|
|
return False
|
|
|
|
logger.info(
|
|
"should_trigger_general: TRIGGER — %d examples, %.1f days since last run",
|
|
count,
|
|
days,
|
|
)
|
|
return True
|
|
|
|
|
|
def should_trigger_dpo(conn: psycopg2.extensions.connection) -> bool:
|
|
"""
|
|
Return True if a DPO preference-learning run should be started.
|
|
|
|
Conditions (all must be true):
|
|
1. At least 50 unused human-edited preference pairs
|
|
2. No DPO run currently in progress
|
|
3. At least 7 days since the last completed DPO run
|
|
"""
|
|
thresholds = FINE_TUNING_THRESHOLDS["dpo"]
|
|
|
|
if _is_run_in_progress(conn, "dpo"):
|
|
logger.debug("should_trigger_dpo: run already in progress")
|
|
return False
|
|
|
|
pairs = _count_available_dpo_pairs(conn)
|
|
if pairs < thresholds["min_preference_pairs"]:
|
|
logger.debug(
|
|
"should_trigger_dpo: only %d pairs (need %d)",
|
|
pairs,
|
|
thresholds["min_preference_pairs"],
|
|
)
|
|
return False
|
|
|
|
days = _days_since_last_run(conn, "dpo")
|
|
if days < thresholds["min_days_since_last_run"]:
|
|
logger.debug(
|
|
"should_trigger_dpo: last run %.1f days ago (need %d)",
|
|
days,
|
|
thresholds["min_days_since_last_run"],
|
|
)
|
|
return False
|
|
|
|
logger.info(
|
|
"should_trigger_dpo: TRIGGER — %d preference pairs, %.1f days since last run",
|
|
pairs,
|
|
days,
|
|
)
|
|
return True
|
|
|
|
|
|
def list_active_task_types(conn: psycopg2.extensions.connection) -> list[str]:
|
|
"""Return all distinct task_types present in the learning corpus."""
|
|
sql = """
|
|
SELECT DISTINCT task_type
|
|
FROM learning_corpus
|
|
WHERE task_type IS NOT NULL
|
|
ORDER BY task_type
|
|
"""
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql)
|
|
return [row[0] for row in cur.fetchall()]
|