From bb23b8a120a64bcac49c21b463eaf98163792319 Mon Sep 17 00:00:00 2001 From: Chaim Date: Sat, 25 Apr 2026 20:14:00 +0000 Subject: [PATCH] fix(kb): classifier reliability + pending-batches review endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two production-blocker fixes for the v0.8.0 batch upload UI, plus a small endpoint to recover review state after a hard refresh. classifier.py: - timeout 45s → 90s and concurrency 5 → 2. ai-gateway proxies a single Claude OAuth session so requests effectively serialize at the gateway; with concurrency=5, the 4th and 5th files in a batch of 5 timed out from queue waits alone (observed 2026-04-25 on user upload of 5 circulars: jobs 10 and 13 returned empty). - labels schema/prompt: arrays of strings via ai-gateway came back as [{}, {}, {}] (proxy mangled items.type). Switched the param to a comma-separated string with explicit format guidance, then split in _normalize. _normalize is defensive — accepts both string and list[str|dict] shapes so older callers don't break. - Sharpened the prompt distinction between subject (free text on the source) and labels (navigation tags), since identical examples were causing the model to emit one and skip the other. ingest_jobs.list_pending_batches + admin_kb GET /admin/kb/batches: the review screen lives in JS RAM (_activeBatch). A hard refresh wipes it, leaving uploaded jobs orphaned in awaiting_review with no UI path back. The new endpoint groups jobs by batch_id with status counters; the EspoCRM 'ממתינים לאישור' panel uses it to let users resume any pending review. Co-Authored-By: Claude Opus 4.7 (1M context) --- api/routes/admin_kb.py | 12 +++++ api/services/kb/classifier.py | 81 ++++++++++++++++++++++------------ api/services/kb/ingest_jobs.py | 55 +++++++++++++++++++++++ 3 files changed, 120 insertions(+), 28 deletions(-) diff --git a/api/routes/admin_kb.py b/api/routes/admin_kb.py index 2331579..1ab8574 100644 --- a/api/routes/admin_kb.py +++ b/api/routes/admin_kb.py @@ -548,6 +548,18 @@ async def get_batch(request: Request, batch_id: str): return {"batch_id": batch_id, "items": items, "count": len(items)} +@router.get("/batches") +async def list_batches(request: Request, limit: int = 50): + """Batches that still have awaiting_review (or in-progress) jobs. + + Powers the 'ממתינים לאישור' panel in the management UI: a user can + resume a review session even after a hard refresh wiped JS state. + """ + _verify_admin(request) + items = await kb_jobs.list_pending_batches(limit=limit) + return {"items": items, "count": len(items)} + + class JobCommit(BaseModel): """User-confirmed metadata from the review screen.""" kind: Literal["law", "regulation", "circular", "caselaw", "tool"] diff --git a/api/services/kb/classifier.py b/api/services/kb/classifier.py index d462345..4cfb49a 100644 --- a/api/services/kb/classifier.py +++ b/api/services/kb/classifier.py @@ -34,14 +34,18 @@ from openai import AsyncOpenAI logger = logging.getLogger("shira.kb.classifier") -# Bound concurrency so a 10–30 file batch doesn't fan out 30 parallel -# Claude calls. Five at a time gives reasonable wall-clock latency -# (~6–10s for 30 files) without overwhelming ai-gateway. -_CONCURRENCY = int(os.environ.get("KB_CLASSIFIER_CONCURRENCY", "5")) +# ai-gateway proxies a single Claude OAuth session, so requests +# effectively serialize at the gateway. Setting concurrency higher +# than 2 only piles requests in the gateway's queue, where they eat +# our timeout budget. Two in flight gives a tiny pipelining win if +# the gateway can overlap socket I/O with Claude latency, otherwise +# it's no worse than 1. +_CONCURRENCY = int(os.environ.get("KB_CLASSIFIER_CONCURRENCY", "2")) # Empirically the ai-gateway → Claude tool-call round-trip with a ~3KB -# Hebrew system prompt + 5KB user content runs 18-25s. Pad to 45s so -# we don't fail-soft on the long-tail latency. -_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "45")) +# Hebrew system prompt + 5KB user content runs 18-25s. With 20 files +# queued through the gateway one-at-a-time, the tail can hit ~80s. +# Padded to 90s so we don't fail-soft on serialized queue waits. +_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "90")) _MAX_INPUT_CHARS = 20_000 # ~3 pages of Hebrew legal text _semaphore = asyncio.Semaphore(_CONCURRENCY) @@ -71,17 +75,23 @@ _SYSTEM_PROMPT = """אתה מסווג מסמכים משפטיים בעברית • tool: שם הגוף המוציא ("CanChild — McMaster University") אם אין זיהוי ברור — null. אל תמציא. -‏4. **subject** — תיאור תת-נושא קצר (≤80 תווים). למשל: "ילד נכה / אוטיזם", +‏4. **subject** — תיאור תת-נושא קצר וחופשי, כטקסט אחד (≤80 תווים). זהו תיאור + אינפורמטיבי שמופיע בכרטיסיית המקור בחיפוש. למשל: "ילד נכה / אוטיזם", "ילד נכה / שיתוק מוחין", "נפגעי עבודה / מחלת מקצוע". -‏5. **labels** — עד 3 תוויות לסיווג מעמיק. כל תווית בפורמט "קטגוריה / תת-קטגוריה" אם ניתן. - דוגמאות: - • "ילד נכה / אוטיזם" - • "ילד נכה / שיתוק מוחין" - • "ילד נכה / תלות בזולת" - • "ילד נכה / עיכוב התפתחותי" - • "ילד נכה / מומים בגפיים" - • "נפגעי עבודה / שמיעה" +‏5. **labels** — **שדה חובה**: 1 עד 3 תוויות לניווט וסינון במאגר, מופרדות בפסיק. + החזר מחרוזת אחת (string) — לא רשימה ולא אובייקט. דוגמאות בהמשך מראות בדיוק את הפורמט. + זהו שדה נפרד מ-subject ואסור להשמיט אותו, גם אם הוא חופף לחלוטין ל-subject. + הכלל: התווית הראשונה היא תמיד התת-קטגוריה הספציפית (זהה ל-subject ברוב המקרים). + התוויות הבאות, אם יש, הן רחבות יותר או הקשרים אחרים — שם כלי הערכה, סוג ועדה, + מספר חוזר וכו' — שיעזרו למצוא את המסמך מזוויות שונות. + כל תווית בפורמט "קטגוריה / תת-קטגוריה" כשרלוונטי. + דוגמאות לערך הסופי של השדה (string מופרד בפסיקים): + • "ילד נכה / אוטיזם, ילד נכה, DSM-5" + • "ילד נכה / שיתוק מוחין, GMFCS, כלי הערכה" + • "ילד נכה / תלות בזולת, ועדה רפואית, ילד נכה" + • "ילד נכה / עיכוב התפתחותי, DQ, פעוטות" + • "נפגעי עבודה / שמיעה, אודיומטריה" ‏6. **published_at / effective_at** — בפורמט YYYY-MM-DD בלבד. אם רק שנה ידועה — YYYY-01-01. אם לא ידוע — null. @@ -101,7 +111,7 @@ _TOOL_SCHEMA = { "description": "Return extracted metadata for the document.", "parameters": { "type": "object", - "required": ["kind", "title", "summary"], + "required": ["kind", "title", "summary", "labels"], "properties": { "kind": { "type": "string", @@ -111,9 +121,14 @@ _TOOL_SCHEMA = { "identifier": {"type": ["string", "null"], "maxLength": 200}, "subject": {"type": ["string", "null"], "maxLength": 80}, "labels": { - "type": "array", - "maxItems": 3, - "items": {"type": "string", "maxLength": 80}, + "type": "string", + "minLength": 1, + "maxLength": 250, + "description": ( + "1 to 3 navigation labels separated by commas. " + "Plain string only — NOT an array, NOT an object. " + "Example: 'ילד נכה / אוטיזם, ילד נכה, DSM-5'" + ), }, "published_at": { "type": ["string", "null"], @@ -220,12 +235,22 @@ def _normalize(d: dict) -> dict: v = d.get(key) if isinstance(v, str) and len(v) == 10 and v[4] == "-" and v[7] == "-": out[key] = v - labels = d.get("labels") or [] - if isinstance(labels, list): - clean: list[str] = [] - for raw in labels[:3]: - if isinstance(raw, str) and raw.strip(): - clean.append(raw.strip()) - if clean: - out["labels"] = clean + # labels comes as a comma-separated string from the model. Older + # callers may still hand us a list (defensive — we used to ask for + # an array and Claude proxies sometimes returned [{}] instead of + # strings, see the 2026-04-25 incident). Accept both shapes. + raw_labels = d.get("labels") + items: list[str] = [] + if isinstance(raw_labels, str): + items = [s.strip() for s in raw_labels.split(",")] + elif isinstance(raw_labels, list): + for raw in raw_labels: + if isinstance(raw, str): + items.append(raw.strip()) + elif isinstance(raw, dict): + # Empty {} from broken proxy round-trip — drop it. + continue + clean = [s for s in items if s][:3] + if clean: + out["labels"] = clean return out diff --git a/api/services/kb/ingest_jobs.py b/api/services/kb/ingest_jobs.py index 1beb33f..10b7167 100644 --- a/api/services/kb/ingest_jobs.py +++ b/api/services/kb/ingest_jobs.py @@ -126,6 +126,61 @@ async def list_jobs_by_batch(batch_id: str) -> list[dict]: return [_row_to_dict(r) for r in rows] +async def list_pending_batches(limit: int = 50) -> list[dict]: + """Batches that have at least one `awaiting_review` job. + + The review UI lives in `_activeBatch` JS state, which is RAM-only. + A user who uploads, hard-refreshes, then comes back cannot otherwise + find their pending review. This list lets them resume from the + 'ממתינים לאישור' panel. + + Sorted oldest-first so long-pending batches surface — the user is + less likely to forget the freshly-uploaded ones. + """ + pool = await get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + SELECT + batch_id, + COUNT(*) AS total, + COUNT(*) FILTER (WHERE status = 'awaiting_review') AS awaiting, + COUNT(*) FILTER (WHERE status = 'queued') AS queued, + COUNT(*) FILTER (WHERE status = 'processing') AS processing, + COUNT(*) FILTER (WHERE status = 'failed') AS failed, + COUNT(*) FILTER (WHERE status = 'done') AS done, + MIN(created_at) AS created_at, + MIN(topic_id) AS topic_id, + ARRAY_AGG(DISTINCT kind) AS kinds, + (ARRAY_AGG(original_filename ORDER BY id))[1] AS first_filename + FROM kb_ingest_job + WHERE batch_id IS NOT NULL + GROUP BY batch_id + HAVING COUNT(*) FILTER (WHERE status = 'awaiting_review') > 0 + OR COUNT(*) FILTER (WHERE status IN ('queued','processing')) > 0 + ORDER BY MIN(created_at) ASC + LIMIT $1 + """, + max(1, min(200, limit)), + ) + out: list[dict] = [] + for r in rows: + out.append({ + "batch_id": r["batch_id"], + "total": r["total"], + "awaiting_review": r["awaiting"], + "queued": r["queued"], + "processing": r["processing"], + "failed": r["failed"], + "done": r["done"], + "topic_id": r["topic_id"], + "kinds": list(r["kinds"]) if r["kinds"] else [], + "first_filename": r["first_filename"], + "created_at": r["created_at"].isoformat() if r["created_at"] else None, + }) + return out + + async def list_jobs( *, topic_id: int | None = None,