"""Inference: prompt revisions (REST), Call 1 vision classification, Call 2 long description, test-inference, inference-off server description, inference_log. Cites METADATA-35..115, SETTINGS-30/45/90. Transport is the OpenAI-compatible chat/completions endpoint via httpx; tests mock it. Groupers (species/wcat) are resolved BY SLUG only (QUALITY-105). """ from __future__ import annotations import json import time from datetime import datetime, timezone from typing import Any from fastapi import APIRouter from fastapi.responses import JSONResponse from .db import get_connection from .prompts import CALL1_SEED, CALL2_SEED, active_prompt, seed_prompts from .inference_core import ( InferenceClient, InferenceConnectionError, InferenceError, _enabled, _gate, _inference_creds, _log_call, _parse_json, _redact_request_for_log, _slab_photos, _taxonomy_lists, fit_long_edge_png, ) router = APIRouter(prefix="/api/v1") def _error(code: str, message: str, detail: Any = None) -> dict: return {"error": {"code": code, "message": message, "detail": detail}} def _now() -> str: return datetime.now(timezone.utc).replace(microsecond=0, tzinfo=None).isoformat() # ---------- Inference client (OpenAI-compatible, httpx) ---------- @router.post("/slabs/{slab_id}/inference/call1") async def call1(slab_id: str) -> dict: conn = get_connection() row = conn.execute("SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (slab_id,)).fetchone() if row is None: return JSONResponse(status_code=404, content=_error("not_found", "Slab not found")) if not _enabled(conn, "inference_enabled"): return JSONResponse(status_code=409, content=_error("inference_disabled", "Inference is off")) base, key, model = _inference_creds(conn) if not base or not key: return JSONResponse(status_code=409, content=_error("inference_unconfigured", "Inference is not configured")) client = InferenceClient(base, key, model) seed_prompts(conn) conn.commit() act = active_prompt(conn, "call1") prompt_text = act[1] if act else CALL1_SEED image_parts, photo_meta = _slab_photos(conn, slab_id) tax = _taxonomy_lists(conn) text = ( f"{prompt_text}\n\nLISTS:\n{json.dumps(tax, ensure_ascii=False)}\n\n" f"SKU={row['sku']} length_in={row['length_in']} thickness_in={row['thickness_in']}" ) messages = [{"role": "user", "content": image_parts + [{"type": "text", "text": text}]}] t0 = time.time() try: raw = client.chat(messages) except InferenceConnectionError as e: return JSONResponse(status_code=503, content=_error("inference_unreachable", str(e), {"recoverable": True})) except InferenceError as e: return JSONResponse(status_code=502, content=_error("inference_failed", str(e), {"recoverable": True})) latency = int((time.time() - t0) * 1000) parsed = _parse_json(raw) _log_call(conn, slab_id, 1, act[0] if act else None, model, messages, raw, latency, photo_meta) return {"ok": True, "suggestions": _gate(parsed)} @router.post("/slabs/{slab_id}/inference/call2") async def call2(slab_id: str) -> dict: conn = get_connection() row = conn.execute("SELECT * FROM slabs WHERE id=? AND deleted_at IS NULL", (slab_id,)).fetchone() if row is None: return JSONResponse(status_code=404, content=_error("not_found", "Slab not found")) if not _enabled(conn, "content_llm_enabled"): return JSONResponse(status_code=409, content=_error("content_disabled", "Content generation is off")) base, key, model = _inference_creds(conn) if not base or not key: return JSONResponse(status_code=409, content=_error("inference_unconfigured", "Inference is not configured")) client = InferenceClient(base, key, model) act = active_prompt(conn, "call2") prompt_text = act[1] if act else CALL2_SEED image_parts, photo_meta = _slab_photos(conn, slab_id) brand = conn.execute("SELECT value FROM settings WHERE key='brand_voice'").fetchone() geo = conn.execute("SELECT value FROM settings WHERE key='geo_context'").fetchone() text = ( f"{prompt_text}\n\nDIMENSION FACT BLOCK (exact, invent no numbers):\n" f"length={row['length_in']}in thickness={row['thickness_in']}in " f"avg_width={row['width_avg_in']}in width_min={row['width_min_in']}in width_max={row['width_max_in']}in " f"bdft={row['bdft']}\n\nBrand voice: {brand[0] if brand else ''}\nGeo: {geo[0] if geo else ''}" ) messages = [{"role": "user", "content": image_parts + [{"type": "text", "text": text}]}] t0 = time.time() try: raw = client.chat(messages) except InferenceConnectionError as e: return JSONResponse(status_code=503, content=_error("inference_unreachable", str(e), {"recoverable": True})) except InferenceError as e: return JSONResponse(status_code=502, content=_error("inference_failed", str(e), {"recoverable": True})) latency = int((time.time() - t0) * 1000) _log_call(conn, slab_id, 2, act[0] if act else None, model, messages, raw, latency, photo_meta) conn.execute("UPDATE slabs SET description=?, content_source='llm', updated_at=? WHERE id=?", (raw.strip(), _now(), slab_id)) conn.commit() return {"ok": True, "description": raw.strip()} def server_description(slab: dict) -> str: """Inference-off long description (METADATA-70).""" return ( f"A {slab.get('title', 'slab')} from the mill, measured at " f"{slab.get('length_in')}in long, {slab.get('width_avg_in')}in average width, and " f"{slab.get('thickness_in')}in thick ({slab.get('bdft')} bdft). " "Ready to ship." ) @router.post("/admin/test-inference") async def test_inference() -> dict: conn = get_connection() base, key, model = _inference_creds(conn) if not base or not key: return {"ok": False, "mode": "fail", "vision_capable": False, "detail": "Inference is not configured"} client = InferenceClient(base, key, model, timeout=20.0) t0 = time.time() try: client.chat([{"role": "user", "content": "Reply with the single word: ok"}]) except InferenceConnectionError as e: return {"ok": False, "mode": "fail", "vision_capable": False, "detail": str(e)} except InferenceError as e: return {"ok": False, "mode": "fail", "vision_capable": False, "detail": str(e)} return {"ok": True, "mode": "stateless", "vision_capable": True, "detail": f"reachable in {int((time.time()-t0)*1000)}ms"}