519f2e572a
Search - Retrieval was hybrid in name only: the keyword filter was applied to the SQL query, so results were the *intersection* of the two rankers. A question that matched the meaning but not the literal string could never be returned. It is now a union, fused with Reciprocal Rank Fusion (a text rank and a cosine distance are not on comparable scales, so RRF uses only their orderings). - Added a generated `search_vector` tsvector + GIN index, so the lexical half is ranked full text rather than ILIKE substring matching. - Chose Postgres + pgvector over OpenSearch/Elasticsearch: a search cluster would add a second datastore to keep in sync and a JVM on this host, to replace an index Postgres maintains inside the same transaction. - Removed the keyword-only mode. It looks precise but silently drops the question that asks the same thing in different words. Embeddings — measured on 500 real questions, using each question's own explanation as a paraphrase query (known answer, no hand labelling): bge-small (local CPU, 384d) R@1 0.840 R@5 0.953 186ms/query bge-m3 (LiteLLM proxy, 1024d) R@1 0.847 R@5 0.973 93ms/query BGE-M3 wins on both quality and latency and needs no extra credential, since llm.danvics.com already serves `openrouter-bge-m3`. Three gaps this exposed, all fixed: - Nothing recorded which model produced a stored vector, so changing models silently mixed incomparable spaces. `embedding_model` / `embedded_at` now stamp every vector, `GET /admin/embedding/health` reports current vs stale vs missing, and regeneration defaults to stale-only. - The generator read the model from env while the stamp read a Redis override, so a vector could be labelled with a model that did not produce it. Both now resolve through one function, with a regression test. - Embedding at creation is best effort, and a failure left a question invisible to semantic search forever. `retry_missing_embeddings` runs every 15 minutes via Celery beat and backfills missing or stale rows. - Query embeddings are cached in Redis per model, so typing is not a network round-trip per keystroke. `dimensions` is only sent to OpenAI's embedding-3 family; BGE-M3 rejects it. Tests: 8 new backend tests (union not intersection, fusion ordering, per-ranker failure degradation, provenance stamping, stale/missing accounting, generator and stamp agreement). Full suites green: 95 backend, 127 frontend, build clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014yhHB8Pc7oQqyqn2Vo9DXA
894 lines
40 KiB
Python
894 lines
40 KiB
Python
"""Async quiz extraction task with step-by-step progress reporting via Redis."""
|
||
import json
|
||
import logging
|
||
import time
|
||
import os
|
||
|
||
from sqlalchemy import text as sa_text
|
||
|
||
from app.tasks import celery_app
|
||
from app.database import SessionLocal
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
EXPIRE_SECONDS = 3600
|
||
CHUNK_PAGES = 50
|
||
|
||
|
||
def _redis():
|
||
import redis
|
||
from app.config import settings
|
||
return redis.from_url(settings.REDIS_URL, decode_responses=True)
|
||
|
||
|
||
def _push_step(r, job_id: str, step: str, message: str):
|
||
key = f"extraction:steps:{job_id}"
|
||
entry = json.dumps({"step": step, "message": message, "ts": time.time()})
|
||
r.rpush(key, entry)
|
||
r.expire(key, EXPIRE_SECONDS)
|
||
|
||
|
||
def _normalize_ocr(text: str) -> str:
|
||
"""Fix common OCR artifacts in PREP PDFs."""
|
||
return (text
|
||
.replace("Pref erred", "Preferred")
|
||
.replace("Pre ferred", "Preferred")
|
||
.replace("Prefer red", "Preferred")
|
||
.replace("ltem", "Item")
|
||
.replace("ltcm", "Item"))
|
||
|
||
|
||
@celery_app.task(name="extract_quiz", bind=True)
|
||
def extract_quiz(
|
||
self,
|
||
job_id: str,
|
||
user_id: int,
|
||
section_id: int,
|
||
title: str,
|
||
mode: str,
|
||
time_limit_minutes: int | None,
|
||
model_id: str | None,
|
||
question_category_id: int | None,
|
||
extraction_mode: str = "standard",
|
||
):
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.section import Section
|
||
from app.models.pdf_document import PDFDocument
|
||
from app.services import ai_service, vector_service, pdf_service, embedding_service
|
||
from app.models.quiz import Quiz
|
||
from app.models.question import Question
|
||
from app.models.quiz_question_link import QuizQuestionLink
|
||
from app.config import settings
|
||
|
||
_push_step(r, job_id, "start", "Starting extraction…")
|
||
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
if not section:
|
||
raise ValueError("Section not found")
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
if not document:
|
||
raise ValueError("Document not found")
|
||
|
||
total_pages = section.end_page - section.start_page + 1
|
||
_push_step(r, job_id, "text", f"Loading text from pages {section.start_page}–{section.end_page} ({total_pages} pages)…")
|
||
|
||
# Determine model
|
||
if model_id:
|
||
from app.models.ai_model_config import AIModelConfig
|
||
config = db.query(AIModelConfig).filter(AIModelConfig.model_id == model_id).first()
|
||
api_key = config.api_key if config and config.api_key else None
|
||
model_name = config.name if config else model_id
|
||
else:
|
||
model_id, api_key = ai_service.get_model_for_task(db, "extraction")
|
||
model_name = model_id
|
||
|
||
# Split into 50-page chunks
|
||
if total_pages <= CHUNK_PAGES:
|
||
chunks = [(section.start_page, section.end_page)]
|
||
else:
|
||
chunks = []
|
||
p = section.start_page
|
||
while p <= section.end_page:
|
||
end = min(p + CHUNK_PAGES - 1, section.end_page)
|
||
chunks.append((p, end))
|
||
p = end + 1
|
||
|
||
n_chunks = len(chunks)
|
||
if n_chunks > 1:
|
||
_push_step(r, job_id, "text", f"Large section: splitting into {n_chunks} chunks of up to {CHUNK_PAGES} pages each.")
|
||
|
||
all_valid_questions = []
|
||
all_skipped = []
|
||
|
||
# ── Non-standard extraction modes ─────────────────────────────────────
|
||
if extraction_mode != "standard":
|
||
from app.services.extraction_modes import (
|
||
extract_questions_only, extract_two_step,
|
||
extract_with_regex, ai_decide_strategy, generate_from_text,
|
||
ai_answer_questions,
|
||
)
|
||
|
||
resolved_mode = extraction_mode
|
||
|
||
if extraction_mode == "ai_decide":
|
||
_push_step(r, job_id, "ai", "AI is analysing the document to choose the best strategy…")
|
||
resolved_mode, reasoning = ai_decide_strategy(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
)
|
||
_push_step(r, job_id, "ai", f"AI chose: {resolved_mode} — {reasoning}")
|
||
|
||
if resolved_mode == "questions_only":
|
||
_push_step(r, job_id, "ai", "Mode: Questions Only — extracting questions without answers.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
qs = extract_questions_only(_normalize_ocr(chunk_content),
|
||
f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
elif resolved_mode == "ai_answer":
|
||
_push_step(r, job_id, "ai", "Mode: AI Answer — extracting questions and using AI to determine correct answers.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p} — extracting questions…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
normalized = _normalize_ocr(chunk_content)
|
||
qs = extract_questions_only(normalized, f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
if qs:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions found, AI determining answers…")
|
||
qs = ai_answer_questions(qs, normalized, f"{start_p}-{end_p}", model_id, api_key)
|
||
answered = sum(1 for q in qs if q.get("correct_answer") and q["correct_answer"] != "PENDING")
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {answered}/{len(qs)} answered. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
elif resolved_mode == "two_step":
|
||
_push_step(r, job_id, "ai", "Mode: Two-Step (separate answer key section).")
|
||
all_valid_questions, all_skipped = extract_two_step(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
push_step=lambda step, msg: _push_step(r, job_id, step, msg),
|
||
chunk_pages=CHUNK_PAGES,
|
||
)
|
||
|
||
elif resolved_mode == "regex":
|
||
_push_step(r, job_id, "ai", "Mode: AI+Regex — analysing format then applying regex.")
|
||
all_valid_questions, all_skipped = extract_with_regex(
|
||
section.document_id, section.start_page, section.end_page,
|
||
model_id, api_key,
|
||
push_step=lambda step, msg: _push_step(r, job_id, step, msg),
|
||
chunk_pages=CHUNK_PAGES,
|
||
)
|
||
|
||
elif resolved_mode == "generate":
|
||
_push_step(r, job_id, "ai", "Mode: Generate — AI creates questions from the text.")
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p)
|
||
if not chunk_content:
|
||
continue
|
||
try:
|
||
qs = generate_from_text(_normalize_ocr(chunk_content),
|
||
f"{start_p}-{end_p}", start_p, model_id, api_key)
|
||
all_valid_questions.extend(qs)
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p}: {len(qs)} questions generated. Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}")
|
||
|
||
else:
|
||
# ai_decide resolved to standard — fall through to standard loop below
|
||
extraction_mode = "standard"
|
||
|
||
if extraction_mode == "standard":
|
||
# ── STANDARD: existing working extraction (unchanged) ──────────────
|
||
for chunk_idx, (start_p, end_p) in enumerate(chunks, 1):
|
||
if r.get(f"extraction:status:{job_id}") == "cancelled":
|
||
_push_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
if n_chunks > 1:
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p} → {model_name}…")
|
||
else:
|
||
_push_step(r, job_id, "ai", f"Sending pages {start_p}–{end_p} to {model_name}…")
|
||
chunk_content = vector_service.get_pages_text(
|
||
document_id=section.document_id, start_page=start_p, end_page=end_p,
|
||
)
|
||
if not chunk_content:
|
||
_push_step(r, job_id, "ai", f" No text found for pages {start_p}–{end_p}, skipping.")
|
||
continue
|
||
try:
|
||
chunk_data = ai_service.extract_questions(
|
||
_normalize_ocr(chunk_content),
|
||
page_info=f"{start_p}-{end_p}",
|
||
page_ref=start_p,
|
||
model_id=model_id,
|
||
api_key=api_key,
|
||
)
|
||
chunk_skipped = chunk_data[0].pop("skipped", []) if chunk_data else []
|
||
chunk_valid = [q for q in chunk_data if q.get("correct_answer")]
|
||
all_valid_questions.extend(chunk_valid)
|
||
all_skipped.extend(chunk_skipped)
|
||
_push_step(r, job_id, "ai",
|
||
f" Pages {start_p}–{end_p}: {len(chunk_valid)} questions"
|
||
f"{f', {len(chunk_skipped)} skipped' if chunk_skipped else ''}. "
|
||
f"Total: {len(all_valid_questions)}.")
|
||
except Exception as e:
|
||
_push_step(r, job_id, "ai", f" Pages {start_p}–{end_p} failed: {e}. Continuing…")
|
||
|
||
valid_questions = all_valid_questions
|
||
skipped = all_skipped
|
||
|
||
_push_step(r, job_id, "ai", f"Extraction complete: {len(valid_questions)} valid questions{f', {len(skipped)} skipped' if skipped else ''}.")
|
||
|
||
if not valid_questions:
|
||
raise ValueError("No valid questions extracted. The AI could not find questions with correct answers in this page range.")
|
||
|
||
# Refresh DB connection — it may have gone stale during long LLM extraction
|
||
from sqlalchemy import text as _text
|
||
try:
|
||
db.execute(_text("SELECT 1"))
|
||
except Exception:
|
||
db.rollback()
|
||
db.close()
|
||
db = SessionLocal()
|
||
# Re-fetch objects that were bound to the old session
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
|
||
# Extract images
|
||
_push_step(r, job_id, "images", "Extracting question images…")
|
||
file_path = os.path.join(settings.UPLOAD_DIR, document.filename)
|
||
page_images = {}
|
||
if os.path.exists(file_path):
|
||
try:
|
||
page_images = pdf_service.extract_all_images(
|
||
file_path, document.id, section.start_page, section.end_page
|
||
)
|
||
except Exception as e:
|
||
_push_step(r, job_id, "images", f"Image extraction skipped: {e}")
|
||
|
||
# Create quiz
|
||
quiz = Quiz(
|
||
section_id=section_id,
|
||
user_id=user_id,
|
||
title=title,
|
||
questions_count=len(valid_questions),
|
||
mode=mode,
|
||
time_limit_minutes=time_limit_minutes,
|
||
skipped_questions=json.dumps(skipped) if skipped else None,
|
||
)
|
||
db.add(quiz)
|
||
db.flush()
|
||
|
||
_push_step(r, job_id, "save", f"Saving {len(valid_questions)} questions and generating embeddings…")
|
||
|
||
for pos, q in enumerate(valid_questions):
|
||
page_ref = q.get("page_reference")
|
||
image_path = None
|
||
# Only link an image if the AI flagged the question as having a figure
|
||
if q.get("has_figure") and page_ref and page_ref in page_images and page_images[page_ref]:
|
||
image_path = page_images[page_ref].pop(0)
|
||
if not page_images[page_ref]:
|
||
del page_images[page_ref]
|
||
|
||
question = Question(
|
||
source_quiz_id=quiz.id,
|
||
question_category_id=question_category_id,
|
||
question_text=q["question_text"],
|
||
question_type=q["question_type"],
|
||
options=q.get("options"),
|
||
correct_answer=q["correct_answer"],
|
||
explanation=q.get("explanation", ""),
|
||
page_reference=page_ref,
|
||
image_path=image_path,
|
||
)
|
||
db.add(question)
|
||
db.flush()
|
||
db.add(QuizQuestionLink(quiz_id=quiz.id, question_id=question.id, position=pos))
|
||
try:
|
||
embedding_service.embed_question(question)
|
||
except Exception as e:
|
||
logger.warning(f"Embedding failed for question {question.id}: {e}")
|
||
|
||
db.commit()
|
||
db.refresh(quiz)
|
||
|
||
_push_step(r, job_id, "done", f"Quiz ready! {len(valid_questions)} questions extracted and saved.")
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:quiz_id:{job_id}", str(quiz.id), ex=EXPIRE_SECONDS)
|
||
return quiz.id
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Quiz extraction failed for job {job_id}")
|
||
_push_step(r, job_id, "error", f"Extraction failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(e)[:500], ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
CLASSIFY_EXPIRE = 3600
|
||
|
||
|
||
def _push_classify_step(r, job_id: str, step: str, message: str):
|
||
key = f"classify:steps:{job_id}"
|
||
entry = json.dumps({"step": step, "message": message, "ts": time.time()})
|
||
r.rpush(key, entry)
|
||
r.expire(key, CLASSIFY_EXPIRE)
|
||
|
||
|
||
def _create_classification_snapshot(db, job_id: str, user_id: int) -> tuple[int, int, int]:
|
||
row = db.execute(sa_text("""
|
||
INSERT INTO question_classification_snapshots (job_id, created_by)
|
||
VALUES (:job_id, :user_id)
|
||
RETURNING id
|
||
"""), {"job_id": job_id, "user_id": user_id}).fetchone()
|
||
snapshot_id = row[0]
|
||
|
||
db.execute(sa_text("""
|
||
INSERT INTO question_classification_snapshot_links (snapshot_id, question_id, tag_name, tag_type)
|
||
SELECT :snapshot_id, tl.question_id, t.name, t.type
|
||
FROM question_tag_links tl
|
||
JOIN question_tags t ON t.id = tl.tag_id
|
||
ON CONFLICT DO NOTHING
|
||
"""), {"snapshot_id": snapshot_id})
|
||
|
||
stats = db.execute(sa_text("""
|
||
SELECT COUNT(DISTINCT question_id) AS question_count, COUNT(*) AS link_count
|
||
FROM question_classification_snapshot_links
|
||
WHERE snapshot_id = :snapshot_id
|
||
"""), {"snapshot_id": snapshot_id}).fetchone()
|
||
question_count = int(stats[0] or 0)
|
||
link_count = int(stats[1] or 0)
|
||
|
||
db.execute(sa_text("""
|
||
UPDATE question_classification_snapshots
|
||
SET question_count = :question_count, link_count = :link_count
|
||
WHERE id = :snapshot_id
|
||
"""), {
|
||
"snapshot_id": snapshot_id,
|
||
"question_count": question_count,
|
||
"link_count": link_count,
|
||
})
|
||
db.commit()
|
||
return snapshot_id, question_count, link_count
|
||
|
||
|
||
@celery_app.task(name="classify_questions", bind=True)
|
||
def classify_questions(self, job_id: str, user_id: int):
|
||
"""Classify untagged questions using AI — subjects, diseases, keywords."""
|
||
r = _redis()
|
||
r.set(f"classify:status:{job_id}", "running", ex=CLASSIFY_EXPIRE)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.question import Question
|
||
from app.services import ai_service
|
||
|
||
_push_classify_step(r, job_id, "snapshot", "Saving rollback snapshot for current classifications...")
|
||
snapshot_id, snapshot_questions, snapshot_links = _create_classification_snapshot(db, job_id, user_id)
|
||
_push_classify_step(
|
||
r,
|
||
job_id,
|
||
"snapshot",
|
||
f"Saved rollback snapshot #{snapshot_id} with {snapshot_questions} tagged questions and {snapshot_links} tag assignments.",
|
||
)
|
||
_push_classify_step(r, job_id, "start", "Finding untagged questions...")
|
||
|
||
# Get IDs of questions that already have tags
|
||
tagged_ids_rows = db.execute(sa_text(
|
||
"SELECT DISTINCT question_id FROM question_tag_links"
|
||
)).fetchall()
|
||
tagged_ids = {row[0] for row in tagged_ids_rows}
|
||
|
||
# Get all questions not yet tagged
|
||
all_questions = db.query(Question).all()
|
||
untagged = [q for q in all_questions if q.id not in tagged_ids]
|
||
|
||
if not untagged:
|
||
_push_classify_step(r, job_id, "done", "All questions are already tagged.")
|
||
r.set(f"classify:status:{job_id}", "completed", ex=CLASSIFY_EXPIRE)
|
||
return
|
||
|
||
total = len(untagged)
|
||
_push_classify_step(r, job_id, "start", f"Found {total} untagged questions. Starting classification...")
|
||
|
||
# Get AI model for keyword task
|
||
model_id, api_key = ai_service.get_model_for_task(db, "keyword")
|
||
|
||
batch_size = 10
|
||
classified = 0
|
||
|
||
for i in range(0, total, batch_size):
|
||
if r.get(f"classify:status:{job_id}") == "cancelled":
|
||
_push_classify_step(r, job_id, "cancelled", "Job cancelled.")
|
||
return
|
||
|
||
batch = untagged[i:i + batch_size]
|
||
batch_num = (i // batch_size) + 1
|
||
total_batches = (total + batch_size - 1) // batch_size
|
||
_push_classify_step(r, job_id, "progress",
|
||
f"Batch {batch_num}/{total_batches}: classifying {len(batch)} questions...")
|
||
|
||
# Build questions JSON for prompt
|
||
questions_json = json.dumps([
|
||
{"id": q.id, "question_text": q.question_text[:500]}
|
||
for q in batch
|
||
], indent=2)
|
||
|
||
prompt = f"""Classify each medical question below. For each question, provide:
|
||
- subjects: 1-3 medical subjects/specialties (e.g., "Cardiology", "Infectious Disease", "Neonatology")
|
||
- diseases: 1-3 specific diseases/conditions mentioned (e.g., "Kawasaki Disease", "Pneumonia", "Type 1 Diabetes")
|
||
- keywords: 2-4 key clinical concepts (e.g., "fever workup", "antibiotic resistance", "fluid management")
|
||
|
||
Return ONLY JSON:
|
||
{{"classifications": [
|
||
{{"id": <question_id>, "subjects": [...], "diseases": [...], "keywords": [...]}}
|
||
]}}
|
||
|
||
Questions:
|
||
{questions_json}"""
|
||
|
||
try:
|
||
raw = ai_service._call_model(prompt, model_id, api_key)
|
||
# Parse JSON from response
|
||
text = raw.strip()
|
||
if text.startswith("```"):
|
||
text = text.split("\n", 1)[1] if "\n" in text else text[3:]
|
||
if text.endswith("```"):
|
||
text = text[:-3]
|
||
text = text.strip()
|
||
data = json.loads(text)
|
||
classifications = data.get("classifications", [])
|
||
|
||
# Store tags
|
||
for cls in classifications:
|
||
q_id = cls.get("id")
|
||
if not q_id:
|
||
continue
|
||
|
||
for tag_type, tag_list in [("subject", cls.get("subjects", [])),
|
||
("disease", cls.get("diseases", [])),
|
||
("keyword", cls.get("keywords", []))]:
|
||
for tag_name in tag_list:
|
||
if not tag_name or not isinstance(tag_name, str):
|
||
continue
|
||
normalized = tag_name.strip().title()
|
||
if not normalized:
|
||
continue
|
||
|
||
# Insert tag (ON CONFLICT DO NOTHING for case-insensitive uniqueness)
|
||
db.execute(sa_text("""
|
||
INSERT INTO question_tags (name, type)
|
||
VALUES (:name, :type)
|
||
ON CONFLICT (LOWER(name), type) DO NOTHING
|
||
"""), {"name": normalized, "type": tag_type})
|
||
db.flush()
|
||
|
||
# Get the tag ID
|
||
tag_row = db.execute(sa_text("""
|
||
SELECT id FROM question_tags
|
||
WHERE LOWER(name) = LOWER(:name) AND type = :type
|
||
"""), {"name": normalized, "type": tag_type}).fetchone()
|
||
|
||
if tag_row:
|
||
db.execute(sa_text("""
|
||
INSERT INTO question_tag_links (question_id, tag_id)
|
||
VALUES (:qid, :tid)
|
||
ON CONFLICT DO NOTHING
|
||
"""), {"qid": q_id, "tid": tag_row[0]})
|
||
|
||
db.commit()
|
||
classified += len(batch)
|
||
_push_classify_step(r, job_id, "progress",
|
||
f"Batch {batch_num}/{total_batches} done. {classified}/{total} classified.")
|
||
|
||
except Exception as e:
|
||
logger.warning(f"Classification batch {batch_num} failed: {e}")
|
||
_push_classify_step(r, job_id, "progress",
|
||
f"Batch {batch_num} failed: {e}. Continuing...")
|
||
continue
|
||
|
||
_push_classify_step(r, job_id, "done", f"Classification complete. {classified}/{total} questions classified.")
|
||
r.set(f"classify:status:{job_id}", "completed", ex=CLASSIFY_EXPIRE)
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Classification failed for job {job_id}")
|
||
_push_classify_step(r, job_id, "error", f"Failed: {e}")
|
||
r.set(f"classify:status:{job_id}", "failed", ex=CLASSIFY_EXPIRE)
|
||
r.set(f"classify:error:{job_id}", str(e)[:500], ex=CLASSIFY_EXPIRE)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="retry_missing_embeddings")
|
||
def retry_missing_embeddings(batch: int = 200) -> dict:
|
||
"""Backfill questions that have no usable vector.
|
||
|
||
Embedding at creation time is best effort: if the encoder is briefly
|
||
unavailable the question is still saved, and without this it would stay
|
||
invisible to semantic search forever. Runs on a schedule and normally finds
|
||
nothing. Also catches rows left by an embedding-model change.
|
||
"""
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.question import Question
|
||
from app.services import embedding_service
|
||
|
||
active = embedding_service._get_embedding_model()
|
||
pending = (
|
||
db.query(Question)
|
||
.filter(
|
||
(Question.embedding.is_(None))
|
||
| (Question.embedding_model.is_(None))
|
||
| (Question.embedding_model != active)
|
||
)
|
||
.limit(batch)
|
||
.all()
|
||
)
|
||
embedded = 0
|
||
for question in pending:
|
||
try:
|
||
if embedding_service.embed_question(question):
|
||
embedded += 1
|
||
except Exception:
|
||
logger.warning("Retry embedding failed for question %s", question.id, exc_info=True)
|
||
if embedded:
|
||
db.commit()
|
||
logger.info("Backfilled %s embeddings (%s pending in this batch)", embedded, len(pending))
|
||
return {"pending": len(pending), "embedded": embedded, "model": active}
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="regenerate_embeddings", bind=True)
|
||
def regenerate_embeddings(self, job_id: str, user_id: int, stale_only: bool = True):
|
||
"""Re-embed questions with the current model.
|
||
|
||
`stale_only` (the default) covers exactly what breaks semantic search: rows
|
||
with no vector, and rows whose vector came from a different model and so sits
|
||
in an incomparable space. Pass False to rebuild the whole bank.
|
||
"""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:job_title:{job_id}", "Regenerate Embeddings", ex=EXPIRE_SECONDS)
|
||
r.lpush(f"extraction:user_jobs:{user_id}", job_id)
|
||
r.expire(f"extraction:user_jobs:{user_id}", 86400)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.question import Question
|
||
from app.services import embedding_service
|
||
|
||
query = db.query(Question)
|
||
if stale_only:
|
||
active = embedding_service._get_embedding_model()
|
||
query = query.filter(
|
||
(Question.embedding.is_(None))
|
||
| (Question.embedding_model.is_(None))
|
||
| (Question.embedding_model != active)
|
||
)
|
||
questions = query.all()
|
||
total = len(questions)
|
||
scope = "missing or stale" if stale_only else "all"
|
||
_push_step(r, job_id, "start", f"Regenerating embeddings for {total} {scope} questions…")
|
||
|
||
ok = 0
|
||
for i, q in enumerate(questions):
|
||
try:
|
||
if embedding_service.embed_question(q):
|
||
ok += 1
|
||
if (i + 1) % 50 == 0:
|
||
db.commit()
|
||
_push_step(r, job_id, "progress", f"{i + 1}/{total} processed ({ok} embedded)")
|
||
except Exception as e:
|
||
logger.warning(f"Embedding failed for question {q.id}: {e}")
|
||
|
||
db.commit()
|
||
_push_step(r, job_id, "done", f"Done — {ok}/{total} questions re-embedded.")
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
except Exception as e:
|
||
logger.exception(f"Embedding regeneration failed for job {job_id}")
|
||
_push_step(r, job_id, "error", f"Failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="generate_flashcard_deck", bind=True)
|
||
def generate_flashcard_deck(self, job_id: str, section_id: int, user_id: int,
|
||
title: str, model_id: str | None = None):
|
||
"""Generate flashcards from a document section using AI."""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.section import Section
|
||
from app.models.pdf_document import PDFDocument
|
||
from app.services import vector_service
|
||
from app.services import extraction_modes
|
||
|
||
section = db.query(Section).filter(Section.id == section_id).first()
|
||
if not section:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Section not found")
|
||
return
|
||
document = db.query(PDFDocument).filter(PDFDocument.id == section.document_id).first()
|
||
|
||
from app.services.ai_service import get_model_for_task
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "flashcard")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
|
||
total_pages = section.end_page - section.start_page + 1
|
||
_push_step(r, job_id, "start", f"Generating flashcards from {total_pages} pages…")
|
||
|
||
all_cards = []
|
||
|
||
if total_pages <= CHUNK_PAGES:
|
||
content = vector_service.get_pages_text(section.document_id, section.start_page, section.end_page)
|
||
if content:
|
||
_push_step(r, job_id, "ai", f"Generating flashcards from pages {section.start_page}–{section.end_page}…")
|
||
cards = extraction_modes.generate_flashcards(
|
||
content, f"{section.start_page}–{section.end_page}",
|
||
section.start_page, ai_model_id, ai_api_key,
|
||
)
|
||
all_cards.extend(cards)
|
||
_push_step(r, job_id, "ai", f"Generated {len(cards)} cards")
|
||
else:
|
||
n_chunks = (total_pages + CHUNK_PAGES - 1) // CHUNK_PAGES
|
||
_push_step(r, job_id, "ai", f"Large section: splitting into {n_chunks} chunks")
|
||
for chunk_idx in range(1, n_chunks + 1):
|
||
start_p = section.start_page + (chunk_idx - 1) * CHUNK_PAGES
|
||
end_p = min(start_p + CHUNK_PAGES - 1, section.end_page)
|
||
content = vector_service.get_pages_text(section.document_id, start_p, end_p)
|
||
if not content or len(content.strip()) < 100:
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: no text, skipping")
|
||
continue
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: pages {start_p}–{end_p}…")
|
||
cards = extraction_modes.generate_flashcards(
|
||
content, f"{start_p}–{end_p}", start_p, ai_model_id, ai_api_key,
|
||
)
|
||
all_cards.extend(cards)
|
||
_push_step(r, job_id, "ai", f"Chunk {chunk_idx}/{n_chunks}: {len(cards)} cards")
|
||
|
||
if not all_cards:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "No flashcards could be generated")
|
||
return
|
||
|
||
# Refresh DB connection for save phase
|
||
from sqlalchemy import text as _text
|
||
try:
|
||
db.execute(_text("SELECT 1"))
|
||
except Exception:
|
||
db.rollback()
|
||
db.close()
|
||
db = SessionLocal()
|
||
|
||
_push_step(r, job_id, "save", f"Saving {len(all_cards)} flashcards…")
|
||
|
||
from app.models.flashcard import FlashcardDeck, Flashcard
|
||
deck = FlashcardDeck(
|
||
title=title,
|
||
section_id=section_id,
|
||
user_id=user_id,
|
||
card_count=len(all_cards),
|
||
)
|
||
db.add(deck)
|
||
db.flush()
|
||
|
||
for c in all_cards:
|
||
card = Flashcard(
|
||
deck_id=deck.id,
|
||
front=c["front"],
|
||
back=c["back"],
|
||
page_reference=c.get("page_reference"),
|
||
)
|
||
db.add(card)
|
||
|
||
db.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:deck_id:{job_id}", str(deck.id), ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"Created deck '{title}' with {len(all_cards)} cards")
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Flashcard generation failed: {e}")
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(e)[:500], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", f"Failed: {str(e)[:200]}")
|
||
try:
|
||
db.rollback()
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
ARTICLE_DRAFT_PROMPT = """You write educational articles for a pediatric medical learning platform.
|
||
Topic: {topic}
|
||
{instructions}
|
||
{existing}Return ONLY strict JSON with this exact shape:
|
||
{{"title": "...", "slug": "lowercase-hyphenated", "summary": "1-2 sentences", "content": "introduction markdown", "sections": [{{"id": "32 lowercase hex chars", "slug": "lowercase-hyphenated", "title": "...", "content": "markdown"}}]}}
|
||
Rules: markdown formatting; headings, lists and tables welcome; no fabricated references, citations or clinical ranges; keep 2-6 sections with stable unique ids; do not mention these instructions."""
|
||
|
||
|
||
@celery_app.task(name="generate_article_draft", bind=True)
|
||
def generate_article_draft(self, job_id: str, user_id: int, topic: str,
|
||
instructions: str = "", article_id: int | None = None,
|
||
model_id: str | None = None):
|
||
"""Create or refine an educator article draft; never publishes."""
|
||
import re
|
||
import uuid
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.article import Article
|
||
from app.services.ai_service import get_model_for_task, _proxy_model
|
||
from app.config import settings
|
||
|
||
existing = db.get(Article, article_id) if article_id else None
|
||
if article_id and not existing:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Article not found")
|
||
return
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "article")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
_push_step(r, job_id, "ai", "Drafting article…")
|
||
existing_block = ""
|
||
if existing:
|
||
sections_text = "\n\n".join(
|
||
f"## {s.get('title', 'Section')}\n{s.get('content', '')}" for s in (existing.sections or []))
|
||
existing_block = (f"Existing draft to refine (preserve and improve its content):\n"
|
||
f"Summary: {existing.summary or ''}\nIntro: {existing.content or ''}\n"
|
||
f"{sections_text}\n\n")
|
||
prompt = ARTICLE_DRAFT_PROMPT.format(
|
||
topic=topic,
|
||
instructions=f"Refine this existing draft: {existing.title}\n{instructions}" if existing else instructions or "",
|
||
existing=existing_block,
|
||
)
|
||
import litellm
|
||
kwargs = {"model": _proxy_model(ai_model_id), "messages": [{"role": "user", "content": prompt}],
|
||
"max_tokens": 4000, "temperature": 0.4}
|
||
if ai_api_key:
|
||
kwargs["api_key"] = ai_api_key
|
||
elif settings.LITELLM_API_KEY:
|
||
kwargs["api_key"] = settings.LITELLM_API_KEY
|
||
if settings.LITELLM_API_BASE:
|
||
kwargs["api_base"] = settings.LITELLM_API_BASE
|
||
response = litellm.completion(**kwargs)
|
||
raw = response.choices[0].message.content.strip()
|
||
if raw.startswith("```"):
|
||
raw = raw.split("\n", 1)[1] if "\n" in raw else raw[3:]
|
||
if raw.endswith("```"):
|
||
raw = raw[:-3]
|
||
raw = raw.strip()
|
||
data = json.loads(raw)
|
||
title = str(data.get("title", topic)).strip()[:300]
|
||
slug = re.sub(r"[^a-z0-9]+", "-", str(data.get("slug", topic)).strip().lower()).strip("-")[:120] or "topic"
|
||
sections = []
|
||
for section in data.get("sections", []):
|
||
section_id = str(section.get("id") or "").strip().lower()
|
||
if not re.fullmatch(r"[0-9a-f]{32}", section_id):
|
||
section_id = uuid.uuid4().hex
|
||
sections.append({
|
||
"id": section_id,
|
||
"slug": re.sub(r"[^a-z0-9]+", "-", str(section.get("slug", "section")).strip().lower()).strip("-")[:120] or "section",
|
||
"title": str(section.get("title", "Section")).strip()[:300] or "Section",
|
||
"content": str(section.get("content", "")),
|
||
})
|
||
if existing:
|
||
existing.title, existing.slug, existing.summary = title, slug, str(data.get("summary", "") or "")[:2000]
|
||
existing.content, existing.sections = str(data.get("content", "") or ""), sections
|
||
else:
|
||
base_slug = slug
|
||
n = 2
|
||
while db.query(Article.id).filter(Article.slug == slug).first():
|
||
slug = f"{base_slug}-{n}"
|
||
n += 1
|
||
db.add(Article(slug=slug, title=title, summary=str(data.get("summary", "") or "")[:2000],
|
||
content=str(data.get("content", "") or ""), sections=sections,
|
||
user_id=user_id, status="draft"))
|
||
db.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"Draft saved: {title}")
|
||
except Exception as exc:
|
||
logger.warning("Article draft job %s failed: %s", job_id, exc)
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(exc)[:300], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Drafting failed; the model may need an 'article' configuration.")
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@celery_app.task(name="generate_article_cards", bind=True)
|
||
def generate_article_cards(self, job_id: str, user_id: int, article_id: int,
|
||
model_id: str | None = None):
|
||
"""Generate cards from an article into an unshared educator deck; links stay private until shared."""
|
||
r = _redis()
|
||
r.set(f"extraction:status:{job_id}", "running", ex=EXPIRE_SECONDS)
|
||
db = SessionLocal()
|
||
try:
|
||
from app.models.article import Article
|
||
from app.models.flashcard import Flashcard, FlashcardDeck, FlashcardArticleLink
|
||
from app.services import extraction_modes
|
||
from app.services.ai_service import get_model_for_task
|
||
|
||
article = db.get(Article, article_id)
|
||
if not article:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Article not found")
|
||
return
|
||
ai_model_id, ai_api_key = get_model_for_task(db, "flashcard")
|
||
if model_id:
|
||
ai_model_id = model_id
|
||
_push_step(r, job_id, "ai", f"Generating cards from {article.title}…")
|
||
content = "\n\n".join(filter(None, [
|
||
article.title, article.summary, article.content,
|
||
*[f"## {s['title']}\n{s['content']}" for s in (article.sections or [])],
|
||
]))
|
||
cards = extraction_modes.generate_flashcards(content, "article", None, ai_model_id, ai_api_key)
|
||
if not cards:
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "No cards could be generated")
|
||
return
|
||
deck = db.query(FlashcardDeck).filter(
|
||
FlashcardDeck.title == f"Cards: {article.title}",
|
||
FlashcardDeck.user_id == user_id,
|
||
FlashcardDeck.deleted_at.is_(None),
|
||
).first()
|
||
if not deck:
|
||
deck = FlashcardDeck(title=f"Cards: {article.title}", user_id=user_id, card_count=0, is_shared=0)
|
||
db.add(deck)
|
||
db.flush()
|
||
new_cards = []
|
||
for card in cards:
|
||
item = Flashcard(deck_id=deck.id, front=card["front"], back=card["back"],
|
||
page_reference=card.get("page_reference"))
|
||
db.add(item)
|
||
new_cards.append(item)
|
||
db.flush()
|
||
for item in new_cards:
|
||
db.add(FlashcardArticleLink(flashcard_id=item.id, article_id=article.id))
|
||
deck.card_count = db.query(Flashcard).filter(Flashcard.deck_id == deck.id).count()
|
||
db.commit()
|
||
r.set(f"extraction:status:{job_id}", "completed", ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "done", f"{len(cards)} cards saved to private deck {deck.title}")
|
||
except Exception as exc:
|
||
logger.warning("Article cards job %s failed: %s", job_id, exc)
|
||
db.rollback()
|
||
r.set(f"extraction:status:{job_id}", "failed", ex=EXPIRE_SECONDS)
|
||
r.set(f"extraction:error:{job_id}", str(exc)[:300], ex=EXPIRE_SECONDS)
|
||
_push_step(r, job_id, "error", "Card generation failed.")
|
||
finally:
|
||
db.close()
|