danvics/pdf-quiz-generator
1
0
Fork 0
Code Issues Pull requests Projects Releases 3 Packages Wiki Activity Actions

feat: a session prepared for you, and a model that can see when the one on the job cannot

Browse source 
**Prepared sessions.** Most of this existed: unanswered first, weakest topic
next, wrong-before-right after that, all scaled by what share of the real paper
each topic carries. What it could not do was change with time, say anything
about itself, or be reached without filling in a form.

Evidence now decays on a thirty-day half-life. Exponential rather than a fixed
window because memory has a slope, not a cliff — under a window, 29 days counts
fully and 31 counts for nothing — and because it is memoryless, so an answer's
weight does not shift when unrelated questions are answered, which is what lets
the preview stay a valid forecast. Spring is worth an eighth of last week. Two
things decay: a question's recall probability, drifting towards even rather
than past it, so an old right answer becomes eligible rather than wrong; and a
topic's accuracy, against a prior of two "no idea" answers, which fixes "right
once, known forever".

Strict unanswered-first meant that on a bank of 2,900 nothing was ever
recycled — spaced repetition existed and was unreachable. Review now takes up
to two fifths of a session. And the damping that spread the picks across topics
was applied only to seen material, so a learner with no history was handed the
heaviest domain entire instead of a spread; that was live.

The plan is the product. It is computed, shown, and then the session is built
from that plan's own ids and the plan returned with it, so the two cannot
differ; every figure in it is a tally over the chosen questions rather than a
forecast. No model touches the ranking — a learner asking "why these twenty"
has to get the same answer twice.

**Vision.** The proxy's own `/model/info` says which models can see, so nothing
is hard-coded: 77 report yes, 11 no, and 328 say nothing at all, which means
absent rather than incapable — so those are asked once with an 8px PNG and the
refusal cached. The deployment's main model turns out not to see, and questions
carry figures the learner is looking at, so the tutor was answering about an
image it had never been shown. It routes to a configured tool model now, folds
the description back in as text saying plainly where it came from, and caches
on the bytes because the same figure is re-sent every turn.

Also fixed on the way: `article` was missing from the admin's task list, so
article drafting always ran on the fallback model whatever an administrator
chose; and `.jpx` stem images were sent as JPEG because `mimetypes` guesses
that from the name, so the provider rejected them two hops later.

An administrator must pick a tool model in Settings → AI models. Until then the
tutor says a figure exists that nothing could read, rather than describing one
it cannot see.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
This commit is contained in:
Daniel 2026-09-12 15:46:04 +02:00
parent 0a331cf4eb
commit ce1c0775ab
26 changed files with 3689 additions and 248 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
backend/app/routers/admin.py
View file

@ -158,6 +158,10 @@ class LiteLLMSearchRequest(BaseModel):
#: Used to make a voice speak and a transcriber listen.
TEST_PHRASE = "Inspiratory stridor at rest."
#: Printed into a test image for the tool model. One word, upper case, and not
#: one a model could guess from the instruction it is given.
TEST_WORD = "STRIDOR"
log = logging.getLogger(__name__)
@ -235,7 +239,7 @@ def create_model(
db: Session = Depends(get_db),
admin: User = Depends(require_admin),
):
valid_tasks = ("extraction", "tts", "stt", "teach", "keyword", "flashcard")
valid_tasks = ("extraction", "tts", "stt", "teach", "keyword", "flashcard", "article", "tool")
if data.task not in valid_tasks:
raise HTTPException(status_code=400, detail=f"Task must be one of: {', '.join(valid_tasks)}")
@ -316,8 +320,11 @@ def test_model(
db: Session = Depends(get_db),
admin: User = Depends(require_admin),
):
"""Send a simple test completion to verify an LLM model is reachable.
TTS models are previewed via /tts/speak instead."""
"""Exercise a model the way its job will.
A chat model answers a prompt, a voice speaks, a transcriber listens, and
the tool model reads a word off a picture. A test that only proved the id
was spelled correctly taught administrators to distrust the button."""
model = db.query(AIModelConfig).filter(AIModelConfig.id == model_id).first()
if not model:
raise HTTPException(status_code=404, detail="Model config not found")
@ -375,6 +382,27 @@ def test_model(
raise HTTPException(status_code=502, detail=f"{model.model_id} heard nothing")
return {"message": f"✓ {model.model_id} heard “{heard.strip()}”"}
if model.task == "tool":
# The tool model exists to look at pictures for models that cannot, so
# a text prompt would test nothing about it. It is given a picture of a
# word and asked what it shows: only an eye gets that back.
from app.services import vision_service
image = vision_service.word_image(TEST_WORD)
try:
seen = vision_service.describe(
[image], model.model_id, model.api_key or None,
context="A test image containing a single printed word.",
use_cache=False)[0]
except Exception as e:
raise HTTPException(status_code=502, detail=str(e)[:300])
if TEST_WORD.lower() not in seen.lower():
raise HTTPException(
status_code=502,
detail=f"{model.model_id} answered, but did not read the word in the "
f"test image: “{seen[:160]}”")
return {"message": f"✓ {model.model_id} read “{TEST_WORD}” off the test image"}
try:
from app.services.ai_service import get_client
response = get_client(model.api_key).chat.completions.create(

62
backend/app/routers/articles.py
View file

@ -6,10 +6,9 @@ from pydantic import BaseModel, Field, field_validator
from sqlalchemy.orm import Session
from app.database import get_db
from app.services.search_service import hybrid_ids
from app.services import embedding_service
from app.services.search_service import article_ids_with_sections
from app.models.article import (
Article, ArticleRevision, ArticleSectionIndex, ArticleSlug, ArticleView,
Article, ArticleRevision, ArticleSlug, ArticleView,
QuestionArticleLink,
)
from app.services import article_service
@ -251,52 +250,6 @@ def _record_view(db, user, article) -> None:
log.warning("Could not record article view", exc_info=True)
def _reembed(db, article) -> None:
"""Embed the article and reproject its sections. Failures wait for the retry task."""
try:
embedding_service.embed_record(article, "article")
_rebuild_section_index(db, article)
db.commit()
except Exception:
db.rollback()
log.warning("Could not embed article %s; leaving it for the retry task", article.id, exc_info=True)
def _rebuild_section_index(db, article) -> None:
"""Mirror the article's JSON sections into their own searchable rows.
Sections live in a JSON column, so they cannot carry a vector or a full-text
index themselves. Projecting them lets a citation point at the right section
rather than the whole article. Rows are keyed by section id, so editing a
section updates it and removing one deletes it.
"""
sections = article.sections or []
keep = set()
for section in sections:
section_id = section.get("id")
if not section_id:
continue
keep.add(section_id)
row = db.query(ArticleSectionIndex).filter_by(
article_id=article.id, section_id=section_id).first()
text_changed = True
if row is None:
row = ArticleSectionIndex(article_id=article.id, section_id=section_id)
db.add(row)
else:
text_changed = (row.title != section.get("title")) or (row.content != section.get("content"))
row.title = section.get("title")
row.content = section.get("content")
# Only pay for an embedding when the text actually changed.
if text_changed or row.embedding is None:
embedding_service.embed_record(row, "article_section")
stale = db.query(ArticleSectionIndex).filter(ArticleSectionIndex.article_id == article.id)
if keep:
stale = stale.filter(~ArticleSectionIndex.section_id.in_(keep))
stale.delete(synchronize_session=False)
@router.get("/")
def list_articles(
category_id: int | None = Query(None),
@ -311,7 +264,10 @@ def list_articles(
if q and q.strip():
# Hybrid retrieval, same as the question bank: a title substring match
# could not find an article that says the same thing in other words.
ranked, _ = hybrid_ids(db, q.strip(), "article", limit=200)
# Sections are searched alongside the article row, because the body is
# in them — searching only the article row found nothing for a term that
# appears once, in one section, which is most of what anyone looks up.
ranked, _ = article_ids_with_sections(db, q.strip(), limit=200)
if not ranked:
return []
query = query.filter(Article.id.in_(ranked))
@ -347,7 +303,7 @@ def create_article(
article_service.record_slug(db, article)
db.commit()
db.refresh(article)
_reembed(db, article)
article_service.reindex(db, article)
return _article_json(article)
@ -558,7 +514,7 @@ def update_article(
).delete(synchronize_session=False)
db.commit()
db.refresh(article)
_reembed(db, article)
article_service.reindex(db, article)
# Told at the moment of saving, which is the last point the person who wrote
# the link is still looking at it.
return {**_article_json(article), "broken_links": article_service.broken_markers(db, article)}
@ -796,7 +752,7 @@ def restore_revision(article_id: int, revision_id: int, db: Session = Depends(ge
article.references_json = revision.references_json
db.commit()
db.refresh(article)
_reembed(db, article)
article_service.reindex(db, article)
return _article_json(article)

52
backend/app/routers/questions.py
View file

@ -28,9 +28,10 @@ from app.models.favorite import Favorite
from app.services import article_service
from app.services.search_service import hybrid_ids, hybrid_question_ids
from app.services.question_figures import figures_for as _figures_for
from app.services.prepared_session import prepare_session
from app.services.quiz_builder import (bank_query, bank_question_predicate, category_descendants,
filtered_bank_query, CreateFromBankRequest, GenerateTestRequest, create_saved_test,
exam_scope_predicate, generate_test)
filtered_bank_query, CreateFromBankRequest, GenerateTestRequest, TestOptions,
create_saved_test, exam_scope_predicate, generate_test)
from app.utils.auth import get_current_user, require_moderator
from app.utils.category_grants import (assert_can_manage_category, assert_user_can_manage,
is_question_manager, manageable_categories, question_in_scope, require_question_manager)
@ -600,6 +601,53 @@ def create_builder_quiz(data: GenerateTestRequest, db: Session = Depends(get_db)
return generate_test(db, current_user, data)
class PreparedSessionRequest(BaseModel):
"""Accepting a prepared session, optionally at a different length."""
count: int | None = Field(default=None, ge=1, le=200)
mode: Literal["timed", "learning"] = "learning"
title: str | None = Field(default=None, max_length=200)
@router.get("/builder/prepared")
def preview_prepared_session(
count: int | None = Query(default=None, ge=1, le=200),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""The session this learner would be given, before anything is written.
Returned without the question ids: the plan is for reading, and handing the
stems' identifiers to a page that only draws a summary is how a preview
becomes a way to enumerate the bank.
"""
plan = prepare_session(db, current_user, count)
return {key: value for key, value in plan.items() if key != "question_ids"}
@router.post("/builder/prepared")
def start_prepared_session(
data: PreparedSessionRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Commit a prepared session, and return the plan it was actually built from.
The plan is recomputed here rather than carried from the preview, and the
test is built from that plan's own question ids — so the account handed
back describes this session by construction, not by the two calls happening
to agree. A learner who changed the length gets the plan for the length
they chose, which is the honest answer and not the one they were shown.
"""
plan = prepare_session(db, current_user, data.count)
created = create_saved_test(
db, current_user,
TestOptions(title=(data.title or "Prepared session").strip() or "Prepared session",
mode=data.mode, time_limit_minutes=None),
plan["question_ids"])
return {**created, "plan": {key: value for key, value in plan.items() if key != "question_ids"}}
class DescribeRequest(BaseModel):
"""Free-text description of what the learner wants to study."""

14
backend/app/routers/search.py
View file

@ -22,14 +22,14 @@ from sqlalchemy import text as sa_text
from sqlalchemy.orm import Session
from app.database import get_db
from app.models.article import Article, ArticleSectionIndex
from app.models.article import Article
from app.models.flashcard import Flashcard, FlashcardDeck
from app.models.media import MediaAsset
from app.models.question import Question
from app.models.user import User
from app.routers.media import readable_libraries
from app.services.quiz_builder import bank_query, exam_scope_predicate
from app.services.search_service import hybrid_ids
from app.services.search_service import article_ids_with_sections, hybrid_ids
from app.utils.auth import get_current_user
router = APIRouter()
@ -75,16 +75,8 @@ def _ordered(rows, ranked: list[int]):
def _articles(db, user, q, limit):
ranked, _ = hybrid_ids(db, q, "article", limit=POOL)
# A section match belongs to its article, so both rankers feed one result.
section_ranked, _ = hybrid_ids(db, q, "article_section", limit=POOL)
sections = db.query(ArticleSectionIndex).filter(
ArticleSectionIndex.id.in_(section_ranked)).all() if section_ranked else []
by_article: dict[int, list] = {}
for section in sections:
by_article.setdefault(section.article_id, []).append(section)
wanted = list(dict.fromkeys([*ranked, *by_article.keys()]))
wanted, by_article = article_ids_with_sections(db, q, limit=POOL)
if not wanted:
return []
rows = db.query(Article).filter(Article.id.in_(wanted)).all()

58
backend/app/routers/teach.py
View file

@ -1,7 +1,9 @@
"""Teach chat endpoint — AI tutor for study mode questions."""
import logging
from datetime import datetime, time, timedelta, timezone
from fastapi import APIRouter, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from sqlalchemy.orm import Session
@ -11,11 +13,12 @@ from app.models.ai_model_config import AIModelConfig
from app.models.attempt import QuizAttempt
from app.models.user import User
from app.utils.quiz_access import require_question_access
from app.services import site_settings
from app.services import site_settings, vision_service
from app.services.quiz_builder import bank_question_predicate
from app.utils.auth import check_rate_limit, get_current_user, require_moderator
router = APIRouter()
log = logging.getLogger(__name__)
def _daily_teach_limit() -> tuple[str, int]:
@ -77,6 +80,24 @@ def _find_similar_questions(db: Session, question: Question, user: User, limit:
return []
def _figures(question: Question) -> list[vision_service.Image]:
"""The pictures printed with this question, loaded from storage.
The learner is looking at them. Until now the tutor was not: it was handed
the stem, the options and the answer key as text and left to teach around a
radiograph it had never seen, which produces confident sentences about a
finding nobody described. Captions say which is which, because "the figure
in the explanation" and "the figure in the stem" mean different things to a
tutor that has been told the answer.
"""
wanted = [
(question.image_path, "the figure printed with the question stem"),
(question.explanation_image_path, "the figure printed with the explanation"),
]
loaded = [vision_service.load_image(path, caption) for path, caption in wanted if path]
return [image for image in loaded if image]
def _option_label(index: int) -> str:
"""A, B, C … matching what the learner has on screen.
@ -274,6 +295,31 @@ async def chat(
system_prompt = _build_system_prompt(question, similar)
messages = [{"role": "system", "content": system_prompt}]
# Blocking work — object storage, and possibly a second model — inside an
# async endpoint, so it goes to a worker thread rather than stopping every
# other request on this uvicorn worker.
handoff = None
figures = await run_in_threadpool(_figures, question)
if figures:
try:
parts, handoff = await run_in_threadpool(
vision_service.image_context, db, figures,
model_id=model_id, api_key=api_key,
context=(question.question_text or "")[:1500])
except vision_service.VisionUnavailable as e:
# The figure is context this endpoint volunteered, not the learner's
# question, so nothing being able to read it costs the turn its
# picture rather than ending it. The tutor is told in the prompt
# all the same: a tutor that does not know it is blind writes a
# paragraph about a radiograph nobody looked at.
log.warning("Tutor figure went unread for question %s: %s", question.id, e)
parts = [{"type": "text", "text": (
"This question is printed with a figure. Nothing available can read it "
"for you — an administrator has not configured a Tool model. Teach from "
"the text, and neither describe nor infer what the figure shows.")}]
messages.append({"role": "user", "content": parts})
for msg in req.messages:
if msg.role not in ("user", "assistant"):
continue
@ -299,8 +345,12 @@ async def chat(
reply_lines.pop()
reply = "\n".join(reply_lines).strip()
return {"reply": reply, "suggestions": suggestions[:3]}
answer = {"reply": reply, "suggestions": suggestions[:3]}
if handoff and handoff.delegated:
# Two models ran. Said in the response as well as the log, so that
# "why was that turn slow" has an answer on the page it happened on.
answer["vision"] = handoff.as_dict()
return answer
except Exception as e:
import logging
logging.getLogger(__name__).error(f"TeachChat error for user {current_user.id} model {model_id}: {e}")
log.error(f"TeachChat error for user {current_user.id} model {model_id}: {e}")
raise HTTPException(status_code=502, detail="The AI tutor is temporarily unavailable. Please try again in a moment.")

2
backend/app/schemas/admin.py
View file

@ -7,7 +7,7 @@ class AIModelConfigCreate(BaseModel):
model_config = {"protected_namespaces": ()}
name: str
model_id: str
task: str # extraction, tts, stt, teach, keyword, flashcard
task: str # extraction, tts, stt, teach, keyword, flashcard, article, tool
api_key: str | None = None
is_active: bool = True
is_default: bool = False

19
backend/app/services/ai_service.py
View file

@ -128,8 +128,14 @@ Content from page(s) {page_info}:
{content}"""
def get_model_for_task(db, task: str = "extraction") -> tuple[str, str | None]:
"""Get the configured model for a specific task from DB, or fall back to settings."""
def get_configured_model(db, task: str) -> tuple[str, str | None] | None:
"""The model an administrator chose for this job, or None if they chose none.
The distinction matters for work that only exists because somebody set it
up: a vision handoff to whatever LITELLM_MODEL happens to be is exactly the
silent degradation the handoff was built to prevent, so that caller needs to
hear "nothing is configured" rather than be given the site default.
"""
try:
from app.models.ai_model_config import AIModelConfig
config = db.query(AIModelConfig).filter(
@ -140,8 +146,13 @@ def get_model_for_task(db, task: str = "extraction") -> tuple[str, str | None]:
if config:
return config.model_id, config.api_key
except Exception as e:
logger.warning(f"Failed to load AI model config for task '{task}', using fallback: {e}")
return settings.LITELLM_MODEL, settings.LITELLM_API_KEY or None
logger.warning(f"Failed to load AI model config for task '{task}': {e}")
return None
def get_model_for_task(db, task: str = "extraction") -> tuple[str, str | None]:
"""Get the configured model for a specific task from DB, or fall back to settings."""
return get_configured_model(db, task) or (settings.LITELLM_MODEL, settings.LITELLM_API_KEY or None)
def _truncate_content(content: str, max_chars: int = 100000) -> str:

64
backend/app/services/article_service.py
View file

@ -24,7 +24,10 @@ from datetime import datetime
from sqlalchemy.orm import Session
from app.models.article import Article, ArticleRevision, ArticleSlug
from app.models.article import (
Article, ArticleRevision, ArticleSectionIndex, ArticleSlug,
)
from app.services import embedding_service
logger = logging.getLogger(__name__)
@ -201,3 +204,62 @@ def readable_articles(db, user):
if getattr(user, "is_moderator", False):
return query
return query.filter((Article.status == "published") | (Article.user_id == user.id))
def rebuild_section_index(db: Session, article: Article) -> int:
"""Mirror the article's JSON sections into their own searchable rows.
Sections live in a JSON column, so they cannot carry a vector or a full-text
index themselves. Projecting them lets a citation point at the right section
rather than the whole article, and it is the only place the body is embedded
in full — the article's own vector has room for excerpts, not for 8k
characters of prose.
Rows are keyed by section id, so editing a section updates it and removing
one deletes it. Returns how many vectors were generated, which is what a
bulk caller needs in order to report cost.
Lives here rather than beside the route that first needed it because
anything that writes `Article.sections` has to call it; a copy that skipped
it left 323 generated articles with no section rows at all.
"""
pending, keep = [], set()
for section in article.sections or []:
if not isinstance(section, dict):
continue
section_id = section.get("id")
if not section_id:
continue
keep.add(section_id)
row = db.query(ArticleSectionIndex).filter_by(
article_id=article.id, section_id=section_id).first()
text_changed = True
if row is None:
row = ArticleSectionIndex(article_id=article.id, section_id=section_id)
db.add(row)
else:
text_changed = (row.title != section.get("title")) or (row.content != section.get("content"))
row.title = section.get("title")
row.content = section.get("content")
# Only pay for an embedding when the text actually changed.
if text_changed or row.embedding is None:
pending.append(row)
embedded = embedding_service.embed_records(pending, "article_section") if pending else 0
stale = db.query(ArticleSectionIndex).filter(ArticleSectionIndex.article_id == article.id)
if keep:
stale = stale.filter(~ArticleSectionIndex.section_id.in_(keep))
stale.delete(synchronize_session=False)
return embedded
def reindex(db: Session, article: Article) -> None:
"""Embed the article and reproject its sections. Failures wait for the retry task."""
try:
embedding_service.embed_record(article, "article")
rebuild_section_index(db, article)
db.commit()
except Exception:
db.rollback()
logger.warning("Could not embed article %s; leaving it for the retry task",
article.id, exc_info=True)

120
backend/app/services/embedding_service.py
View file

@ -105,11 +105,51 @@ def generate_embedding(text: str) -> list[float] | None:
return None
# One vector cannot hold a whole article: the body averages 8k characters of
# section text and runs past 13k, while `_join` and `generate_embedding` clamp
# at 4000. Budget below that so the clamp never has the last word — a silent
# truncation is how an index comes to be confidently wrong.
ARTICLE_EMBED_BUDGET = 3600
# Below this an excerpt is too short to say anything, so a very long article
# spends a little over budget rather than reducing every section to a phrase.
ARTICLE_MIN_EXCERPT = 90
def article_embedding_text(article) -> str:
"""Compose what a whole article contributes to its vector.
`content` holds the body for only eight hand-seeded articles; every
generated one puts its prose in the `sections` JSON, so title and summary
alone described 98% of the library and the article vector could not tell
two respiratory topics apart.
Every section gets an even slice rather than the head being kept: head
truncation of a twelve-section article stops somewhere in the
pathophysiology, so treatment, management and complications — the half a
learner actually searches — contribute nothing at all. Section titles go in
whole, being the densest signal per character available.
This stays a topical signal by design. Depth belongs to `article_section`
rows, where the longest section in the corpus still fits under the clamp
intact, so no sentence of the body goes unembedded anywhere.
"""
sections = [s for s in (article.sections or []) if isinstance(s, dict)]
parts = [article.title, article.summary, article.content]
parts.extend(s.get("title") for s in sections if s.get("title"))
spent = sum(len(part) + 1 for part in parts if part)
bodies = [body for body in ((s.get("content") or "").strip() for s in sections) if body]
if bodies:
excerpt = max(ARTICLE_MIN_EXCERPT, (ARTICLE_EMBED_BUDGET - spent) // len(bodies))
parts.extend(body[:excerpt] for body in bodies)
return _join(*parts)
# What each embeddable type contributes to its vector. Adding a type here is
# all that is needed for the retry task and the health report to cover it.
EMBEDDABLE = {
"question": lambda row: _join(row.question_text, *(row.options or [])),
"article": lambda row: _join(row.title, row.summary, row.content),
"article": article_embedding_text,
"flashcard": lambda row: _join(row.front, row.back),
"article_section": lambda row: _join(row.title, row.content),
# Text today; a vision model can embed the image itself later without
@ -122,6 +162,84 @@ def _join(*parts) -> str:
return " ".join(part for part in parts if part)[:4000]
# One request per row costs a round trip each; an article with fourteen sections
# spent fourteen of them on a single save. The proxy takes a list, so a save is
# one call and a corpus sweep is hundreds rather than thousands.
EMBED_BATCH = 32
def generate_embeddings(texts: list[str]) -> list[list[float] | None]:
"""Embed several texts in one round trip, aligned to the input list.
A batch is all-or-nothing at the transport level, so any failure falls back
to embedding the texts one at a time rather than dropping the lot: the
fallback chain in `generate_embedding` (Bedrock, dimension checks) is the
only place that logic should live, and a partial batch must not skip it.
"""
clean = [" ".join(text.split())[:4000] if text else "" for text in texts]
wanted = [index for index, text in enumerate(clean) if text]
out: list[list[float] | None] = [None] * len(clean)
if not wanted:
return out
embedding_model = _get_embedding_model()
api_base = (settings.LITELLM_API_BASE or "").rstrip("/").removesuffix("/v1")
if embedding_model and settings.LITELLM_API_KEY and api_base:
try:
import httpx, json as _json
body: dict = {"model": embedding_model, "input": [clean[i] for i in wanted]}
if "embedding-3" in embedding_model:
body["dimensions"] = settings.EMBEDDING_DIMENSIONS
resp = httpx.post(
f"{api_base}/v1/embeddings",
headers={"Authorization": f"Bearer {settings.LITELLM_API_KEY}", "Content-Type": "application/json"},
content=_json.dumps(body),
timeout=120,
)
resp.raise_for_status()
# The proxy is not required to preserve order, but it does return
# the index it was given; trusting position alone would silently
# attach one document's vector to another.
data = sorted(resp.json()["data"], key=lambda item: item.get("index", 0))
if len(data) == len(wanted):
for slot, item in zip(wanted, data):
vector = item["embedding"]
if len(vector) == settings.EMBEDDING_DIMENSIONS:
out[slot] = vector
if all(out[i] is not None for i in wanted):
return out
logger.warning("Batch embedding returned %d vectors for %d inputs", len(data), len(wanted))
except Exception as e:
logger.warning(f"Batch embedding failed, falling back to one at a time: {e}")
for index in wanted:
if out[index] is None:
out[index] = generate_embedding(clean[index])
return out
def embed_records(rows: list, kind: str) -> int:
"""Embed a list of rows of one kind in batches; returns how many got a vector.
Rows that fail keep whatever they had, including nothing — the retry task
exists for exactly that, and a half-written vector is worse than none.
"""
build = EMBEDDABLE.get(kind)
if build is None:
raise ValueError(f"Unknown embeddable kind: {kind}")
model, now, done = _get_embedding_model(), datetime.utcnow(), 0
for start in range(0, len(rows), EMBED_BATCH):
chunk = rows[start:start + EMBED_BATCH]
for row, embedding in zip(chunk, generate_embeddings([build(row) for row in chunk])):
if not embedding:
continue
row.embedding = embedding
row.embedding_model = model
row.embedded_at = now
done += 1
return done
def embed_record(row, kind: str) -> bool:
"""Embed any supported row, stamping the model that produced the vector.

276
backend/app/services/prepared_session.py Normal file
View file

@ -0,0 +1,276 @@
"""The prepared session: one action, and a plain statement of why.
The manual builder asks a learner to choose topics, a count, a difficulty and a
state before it will give them anything. Most of those are decisions nobody
opening the app at seven in the morning has the information to make, and the
answers are already in their own record of answers. This turns that record into
a session.
Two things are produced together and must never disagree: the questions, and
the account of why those questions. The account is the product — an adaptive
session a learner cannot see the reasoning behind is one they abandon for a
manual session they can — so a plan is built, stated, and only then committed,
and the same `CandidateRanking` produces both. There is no language model
anywhere in here. This is arithmetic over the learner's own answers, and it has
to be reproducible and legible; a model that could not show its working would
make the explanation impossible, which is the whole of what is being built.
Selection itself lives in `quiz_builder.CandidateRanking` — this module decides
how long the session should be, groups what was chosen, and says why.
"""
import random
import statistics
from datetime import datetime
from fastapi import HTTPException
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from app.models.attempt import AttemptAnswer, QuizAttempt
from app.models.exam import Exam
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.services import exam_blueprint
from app.services.knowledge_groups import category_ancestry
from app.services.quiz_builder import (CandidateRanking, bank_question_predicate,
general_question_predicate, require_objective)
#: Where a session starts before the learner has finished enough of them for
#: their own habit to be a measurement rather than an anecdote.
DEFAULT_SESSION_LENGTH = 20
#: How many finished sessions it takes before length is set from the learner's
#: own median. Three, because one session is not a habit and two cannot have a
#: median that is not simply the mean of the pair.
SESSIONS_BEFORE_LENGTH_IS_PERSONAL = 3
#: Bounds on an offered length. Not on what the learner may then ask for — the
#: builder's own 1–200 still governs that — only on what is offered unasked.
#: Below five a session says nothing about a topic; above sixty it stops being
#: something anyone finishes in a sitting, and length is set here from what
#: this learner actually finishes.
MIN_SESSION_LENGTH = 5
MAX_SESSION_LENGTH = 60
#: Accuracy below which a topic is described to the learner as a weak area.
WEAK_ACCURACY = 0.6
#: Decayed answers a topic needs behind it before that word is used. Below
#: this the prior in `combined_accuracy` is most of the number, and a topic
#: last answered — correctly — in the spring would come out under the
#: threshold and be reported as a weakness on no evidence at all. Three is the
#: point at which the learner's own answers outweigh the prior.
WEAK_EVIDENCE_ANSWERS = 3.0
def habitual_length(db: Session, user) -> tuple[int | None, int]:
"""The learner's own median finished session, and how many they have finished.
The median rather than the mean: one abandoned three-question session and
one marathon both pull a mean somewhere neither of them is, and a learner
who reliably does thirty should be offered thirty.
Counted as answers recorded, not as questions the session held, because
what is wanted is the length they *finish*. Expired and course attempts are
out for the same reason they are out of every other figure here.
"""
rows = db.query(func.count(AttemptAnswer.id)).select_from(QuizAttempt).join(
AttemptAnswer, AttemptAnswer.attempt_id == QuizAttempt.id).join(
Quiz, Quiz.id == QuizAttempt.quiz_id).filter(
QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)),
Quiz.course_id.is_(None),
).group_by(QuizAttempt.id).all()
lengths = sorted(row[0] for row in rows if row[0])
if len(lengths) < SESSIONS_BEFORE_LENGTH_IS_PERSONAL:
return None, len(lengths)
median = int(round(statistics.median(lengths)))
return max(MIN_SESSION_LENGTH, min(MAX_SESSION_LENGTH, median)), len(lengths)
def _ago(days: float) -> str:
"""How long ago, at the resolution a learner thinks in."""
if days < 7:
return "this week"
weeks = int(days // 7)
if weeks < 9:
return f"{weeks} week{'s' if weeks != 1 else ''} ago"
months = max(2, int(round(days / 30.44)))
return f"{months} months ago"
def _percent(weight: float | None) -> str | None:
return None if weight is None else f"{round(weight, 1):g}"
def _blueprint_paper(db: Session, user, count: int, seed: str) -> list[int]:
"""A paper shaped like the published exam, for a learner with no record yet.
Seeded, and seeded on the day rather than the moment, because the plan and
the session it commits are two requests: an unseeded draw would show one
paper and start another, and the explanation would then be a fiction.
"""
exam_id = getattr(user, "active_exam_id", None)
if not exam_id or not exam_blueprint.domains(db, exam_id):
return []
ids, _ = exam_blueprint.sample(
db, exam_id, count, predicate=bank_question_predicate(user) & general_question_predicate(),
rng=random.Random(seed))
return ids
def _topic_rows(db: Session, ranking: CandidateRanking, question_ids: list[int],
cold_start: bool) -> list[dict]:
"""Group the chosen questions by discipline, and say why each is there.
Counted from the questions actually chosen rather than predicted from the
ranking, so the plan cannot drift from the session: every count below is a
tally of `question_ids`.
Grouped by the top-level category — the discipline — because that is the
vocabulary the Analysis page already reports a learner's weaknesses in, and
a plan phrased in leaf topics would be twenty rows of one question each.
"""
categories = db.query(QuestionCategory).all()
names = {cat.id: cat.name for cat in categories}
ancestry = category_ancestry(categories)
discipline_of = {cid: chain[-1] for cid, chain in ancestry.items() if chain}
groups: dict[int | None, dict] = {}
for question_id in question_ids:
category = ranking.category_of.get(question_id)
key = discipline_of.get(category) if category is not None else None
row = groups.setdefault(key, {
"category_id": key, "name": names.get(key, "Unfiled") if key else "Unfiled",
"count": 0, "new_count": 0, "review_count": 0,
"_leaves": set(), "_ages": [],
})
row["count"] += 1
if question_id in ranking.recall:
row["review_count"] += 1
row["_ages"].append(ranking.latest[question_id][2])
else:
row["new_count"] += 1
if category is not None:
row["_leaves"].add(category)
rows = []
for row in groups.values():
leaves = row.pop("_leaves")
ages = row.pop("_ages")
# A discipline's own numbers, not the average of its leaves': what the
# learner is told about "Cardiology" has to be about cardiology.
scored = leaves | ({row["category_id"]} if row["category_id"] else set())
accuracy = ranking.combined_accuracy(scored)
evidence = ranking.evidence_weight(scored)
weight = ranking.weights.get(row["category_id"]) if ranking.weights else None
row["accuracy"] = round(100 * accuracy) if accuracy is not None else None
row["weight"] = round(weight, 1) if weight is not None else None
row["reason"] = _reason(row, accuracy, evidence, weight, ages, cold_start)
rows.append(row)
# Biggest share of the session first: the plan should read in the order the
# learner's attention is being spent.
rows.sort(key=lambda row: (-row["count"], row["name"]))
return rows
def _reason(row: dict, accuracy: float | None, evidence: float, weight: float | None,
ages: list[float], cold_start: bool) -> str:
"""One line saying why this topic is in the session.
Strongest reason first, so the sentence a learner reads is the reason and
not merely a true statement about the topic. A weak topic is usually also
due for review — being told the accuracy is the more useful of the two.
"""
share = _percent(weight)
if cold_start:
return f"Worth {share}% of the exam" if share else "Broad coverage — no history to go on yet"
if accuracy is not None and accuracy < WEAK_ACCURACY and evidence >= WEAK_EVIDENCE_ANSWERS:
return f"Weak area — {round(100 * accuracy)}% correct so far"
if ages and row["review_count"] >= row["new_count"]:
return f"Due for review — last answered {_ago(min(ages))}"
if accuracy is None:
return f"Not attempted yet, worth {share}% of the exam" if share else "Not attempted yet"
return f"Worth {share}% of the exam" if share else "Keeping your coverage even"
def prepare_session(db: Session, user, count: int | None = None, now: datetime | None = None) -> dict:
"""Build a session for this learner and the account of why it is that one.
Nothing is written. The same call is made again to commit, which is what
makes the plan honest: the questions named here are the questions started.
"""
require_objective(db, user)
now = now or datetime.utcnow()
ranking = CandidateRanking(db, user, now=now)
available = len(ranking.rows)
if not available:
raise HTTPException(400, "There are no questions in your bank yet")
habitual, finished = habitual_length(db, user)
asked = count is not None
if asked:
length = max(1, min(200, count))
else:
length = habitual or DEFAULT_SESSION_LENGTH
length = min(length, available)
# No finished session is the honest test for cold start, not "no answers":
# a learner who abandoned one question has told us nothing to personalize
# from, and being handed a "weak area" on the strength of it is the fake
# personalization this is meant to avoid.
cold_start = finished == 0
seed = f"{user.id}:{length}:{now.date().isoformat()}"
question_ids = _blueprint_paper(db, user, length, seed) if cold_start else []
blueprint_led = bool(question_ids)
if not question_ids:
question_ids = ranking.select(length)
if not question_ids:
raise HTTPException(400, "There are no questions in your bank yet")
topics = _topic_rows(db, ranking, question_ids, cold_start)
review_count = sum(row["review_count"] for row in topics)
new_count = sum(row["new_count"] for row in topics)
exam_name = None
if getattr(user, "active_exam_id", None):
exam = db.get(Exam, user.active_exam_id)
exam_name = exam.name if exam else None
return {
"count": len(question_ids),
"question_ids": question_ids,
"basis": "cold_start" if cold_start else "personalized",
"new_count": new_count,
"review_count": review_count,
"available": available,
"exam_name": exam_name,
"summary": _summary(cold_start, blueprint_led, new_count, review_count),
"length_reason": _length_reason(asked, habitual, finished, len(question_ids), available),
"topics": topics,
}
def _summary(cold_start: bool, blueprint_led: bool, new_count: int, review_count: int) -> str:
if cold_start:
spread = "the exam blueprint" if blueprint_led else "your whole question bank"
return ("You haven't finished a session yet, so this is an even spread across "
f"{spread} rather than a personalized one.")
if review_count:
return (f"{new_count} new and {review_count} due for review, ranked by what would "
"move your score the most.")
return f"{new_count} new questions, ranked by what would move your score the most."
def _length_reason(asked: bool, habitual: int | None, finished: int,
length: int, available: int) -> str:
if asked:
reason = f"{length} questions, your choice."
elif habitual:
reason = f"You usually finish {habitual} questions, so that's the length."
else:
remaining = SESSIONS_BEFORE_LENGTH_IS_PERSONAL - finished
reason = (f"{DEFAULT_SESSION_LENGTH} to start with — after {remaining} more finished "
f"session{'s' if remaining != 1 else ''} this matches your own median.")
if length >= available:
reason += " That is everything left in your bank."
return reason

340
backend/app/services/quiz_builder.py
View file

@ -1,6 +1,8 @@
"""Permission-safe, saved general-bank tests and category selection."""
import random
import statistics
from collections import defaultdict
from datetime import datetime
from typing import Literal
from fastapi import HTTPException
@ -107,7 +109,18 @@ def bank_query(db, user):
def filtered_bank_query(db, user, category_ids=(), state="all", difficulty=None, article_ids=(), tag_ids=(), system_ids=()):
"""The bank a learner can draw a session from, narrowed by their filters.
Scoped to their exam, like browsing and searching already were. It was not,
and the two disagreed in the worst possible direction: a learner studying
for an exam with no content linked to it saw an empty question bank and was
then handed a full session built from every question in it. Whatever the
right pool is, it has to be the same pool in both places.
"""
query = bank_query(db, user)
scope = exam_scope_predicate(db, user)
if scope is not None:
query = query.filter(scope)
if difficulty:
query = query.filter(Question.difficulty == difficulty)
if article_ids:
@ -285,88 +298,257 @@ def blueprint_weights(db, user) -> dict[int, float]:
return weights
def adaptive_select(db, user, count, category_ids, state, difficulty):
#: How long one answer keeps half its weight as evidence about the learner.
#:
#: The curve is exponential, `0.5 ** (age / half_life)`. A fixed window was the
#: obvious first thing and is wrong in a way that shows: it makes an answer
#: twenty-nine days old count in full and one thirty-one days old count for
#: nothing, so a topic crosses a cliff overnight and the ranking lurches
#: without the learner having done anything. Exponential also has the property
#: that matters for an order recomputed on every visit — it is memoryless, so
#: an answer's weight depends only on its own age and not on what has been
#: answered since, which is what keeps two consecutive sessions consistent with
#: each other. A power law fits very long retention slightly better, but it
#: needs an arbitrary offset to avoid a singularity at age zero and a second
#: parameter nothing here could justify; one named half-life describes the
#: whole of this curve.
#:
#: Thirty days because that is about the turn of a revision cycle. It puts a
#: ninety-day-old answer at an eighth of the weight of a fresh one — so what
#: was missed last week clearly outranks what was missed in spring — while not
#: writing off a topic revised last month as forgotten.
EVIDENCE_HALF_LIFE_DAYS = 30.0
#: What the last outcome says about whether a question is still known, at the
#: moment it was answered. Not 1 and 0: one answer is one observation, and a
#: right answer can be a guess as easily as a wrong one can be a slip. These
#: are the numbers the recycling order always used, named here because time
#: now moves them.
RECALL_AFTER_CORRECT = 0.85
RECALL_AFTER_WRONG = 0.25
#: Recall of a question there is no useful evidence about either way — and the
#: accuracy assumed for a topic never answered in, so it sorts between the
#: learner's strong and weak areas rather than jumping the queue.
NEUTRAL_RECALL = 0.5
#: Recall below which a question is due to come round again. A correct answer
#: decays past this at about three and a half weeks, which is the review
#: interval this is meant to express; anything ever answered wrongly is below
#: it from the moment it was answered.
DUE_RECALL = 0.7
#: The most of one session that may be spent on questions already seen. Review
#: is not what you do once the new material runs out — that rule meant a
#: learner with three thousand unseen questions never saw a repeat, which is
#: no spaced repetition at all. But somebody who opens the app and is handed
#: twenty questions they have already answered does not open it again, so the
#: majority of any session is still new.
MAX_REVIEW_SHARE = 0.4
#: Answers' worth of "no idea" mixed into every topic's accuracy. Without it a
#: single correct answer made a topic 100% known and it never came back, which
#: is the one thing a ranking that claims to decay must not do.
PRIOR_ANSWERS = 2.0
#: How fast a topic's priority falls as the session keeps drawing from it.
CATEGORY_DAMPING = 0.5
def recency_weight(age_days: float) -> float:
"""How much evidence that old still counts for."""
return 0.5 ** (max(0.0, age_days) / EVIDENCE_HALF_LIFE_DAYS)
def recall_probability(was_correct: bool, age_days: float) -> float:
"""Chance a question is still known, given how it last went and how long ago.
Decays towards a coin flip rather than towards zero. Forgetting a right
answer does not turn it into a wrong one, and time does not turn a wrong
answer into a right one either; both outcomes end up saying nothing, which
is exactly the state in which the question is worth asking again.
"""
settled = RECALL_AFTER_CORRECT if was_correct else RECALL_AFTER_WRONG
return NEUTRAL_RECALL + (settled - NEUTRAL_RECALL) * recency_weight(age_days)
class CandidateRanking:
"""One learner, one filtered bank, and everything needed to order it.
Built once and read many times. Selection and the plan that describes it
are two readings of this one object rather than two calculations that would
have to be kept in step — a plan that does not describe the session it
starts is worse than no plan.
The rules, in the order they apply:
**Unseen material is most of the session.** A question never met teaches
more than one already answered, so it takes every slot review is not
holding.
**Review takes the rest, up to `MAX_REVIEW_SHARE`, and only what is due.**
Due means recall has decayed below `DUE_RECALL` — everything answered
wrongly, and everything answered correctly long enough ago to be worth
checking.
**Within either, highest value first, damped per topic.** Value for unseen
material is the topic's impact, `(1 − accuracy) × blueprint weight`; for
review it is `(1 − recall) × blueprint weight`. Each pick halves its
topic's priority, which stops a session of twenty becoming twenty
questions from one subject — and gives a learner with no history at all a
spread across the paper instead of the heaviest domain entire.
Accuracy and recall both fade with time; see `EVIDENCE_HALF_LIFE_DAYS`.
"""
def __init__(self, db, user, category_ids=(), state="all", difficulty=None, now=None):
self.now = now or datetime.utcnow()
query = filtered_bank_query(db, user, category_ids, state, difficulty)
# No cap. This used to take the first 2,000 rows, which on a
# 2,948-question bank meant adaptive selection could not see about a
# third of it, and which third depended on database order. Two integer
# columns per question is not a size worth protecting against.
#
# Sorted by id so that every scan below, and so every tie, resolves the
# same way twice running: the plan and the session it commits are
# separate calls, and a ranking that reshuffles between them would make
# the plan a guess.
self.rows = sorted(((row[0], row[1]) for row in query.with_entities(
Question.id, Question.question_category_id).all()), key=lambda row: row[0])
self.category_of = dict(self.rows)
answered = db.query(AttemptAnswer.question_id, AttemptAnswer.is_correct, QuizAttempt.completed_at).join(
QuizAttempt, QuizAttempt.id == AttemptAnswer.attempt_id).join(Quiz, Quiz.id == QuizAttempt.quiz_id).filter(
QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)), Quiz.course_id.is_(None),
AttemptAnswer.question_id.in_([row[0] for row in self.rows]),
).order_by(QuizAttempt.completed_at.desc(), QuizAttempt.id.desc()).all()
#: question id → (was correct, when, age in days) of the latest answer.
self.latest: dict[int, tuple[bool, object, float]] = {}
evidence: dict = defaultdict(lambda: [0.0, 0.0])
for question_id, was_correct, when in answered:
age = max(0.0, (self.now - when).total_seconds() / 86400.0)
self.latest.setdefault(question_id, (bool(was_correct), when, age))
# Looked up, not scanned. This was a linear search through every
# candidate for every answer — the slowest part of building a
# session.
category = self.category_of.get(question_id)
if category is None:
continue
counts = evidence[category]
counts[0] += recency_weight(age)
if was_correct:
counts[1] += recency_weight(age)
self._evidence = evidence
self.recall = {question_id: recall_probability(was_correct, age)
for question_id, (was_correct, _, age) in self.latest.items()}
self.weights = blueprint_weights(db, user)
# The middle of what the board publishes, for a topic it does not
# mention. A zero would make unmapped material unreachable; the highest
# would make it the priority. Neither is a claim the blueprint supports.
self._neutral_weight = statistics.median(self.weights.values()) if self.weights else 1.0
self.unseen = [row for row in self.rows if row[0] not in self.latest]
self.seen = [row for row in self.rows if row[0] in self.latest]
self.due = [row for row in self.seen if self.recall[row[0]] < DUE_RECALL]
def accuracy(self, category) -> float:
"""Share of this topic answered correctly, recent answers counting most.
Pulled towards `NEUTRAL_RECALL` by `PRIOR_ANSWERS`, so one lucky answer
does not settle a topic and a topic left alone drifts back to unknown.
"""
total, correct = self._evidence.get(category, (0.0, 0.0))
return (correct + PRIOR_ANSWERS * NEUTRAL_RECALL) / (total + PRIOR_ANSWERS)
def evidence_weight(self, categories) -> float:
"""Answers' worth of evidence behind these topics, after decay.
How much the accuracy beside it is worth. Two answers from the spring
come to a fifth of one answer from yesterday, and a claim about the
learner should not be made on the strength of them.
"""
return sum(self._evidence.get(category, (0.0, 0.0))[0] for category in categories)
def combined_accuracy(self, categories) -> float | None:
"""Accuracy over several topics at once, or None with nothing to go on.
Evidence is pooled before the ratio is taken, rather than the topics'
accuracies being averaged: a discipline whose two hundred cardiology
answers went badly and whose one rheumatology answer went well is not
halfway between the two.
"""
total = self.evidence_weight(categories)
if not total:
return None
correct = sum(self._evidence.get(category, (0.0, 0.0))[1] for category in categories)
return (correct + PRIOR_ANSWERS * NEUTRAL_RECALL) / (total + PRIOR_ANSWERS)
def weight(self, category) -> float:
"""The topic's share of the real paper, or a neutral stand-in."""
return self.weights.get(category, self._neutral_weight) if self.weights else 1.0
def impact(self, category) -> float:
"""How much a question here could move the score."""
return (1 - self.accuracy(category)) * self.weight(category)
def _unseen_value(self, row) -> float:
return self.impact(row[1])
def _review_value(self, row) -> float:
return (1 - self.recall[row[0]]) * self.weight(row[1])
def value(self, row) -> float:
"""What one candidate is worth, whichever pool it came from."""
return self._review_value(row) if row[0] in self.recall else self._unseen_value(row)
def review_budget(self, count: int) -> int:
"""Slots this session gives to questions already seen."""
return min(len(self.due), round(MAX_REVIEW_SHARE * count)) if count > 0 else 0
def select(self, count: int) -> list[int]:
"""The questions, in the order they will be asked."""
if count <= 0:
return []
damping: dict = defaultdict(lambda: 1.0)
taken: list[int] = []
spent: set[int] = set()
def draw(pool, budget):
candidates = [row for row in pool if row[0] not in spent]
picked = 0
while candidates and picked < budget:
best, best_score = None, None
for row in candidates:
score = self.value(row) * damping[row[1]]
if best_score is None or score > best_score:
best, best_score = row, score
candidates.remove(best)
taken.append(best[0])
spent.add(best[0])
picked += 1
damping[best[1]] *= CATEGORY_DAMPING
review = self.review_budget(count)
draw(self.unseen, count - review)
draw(self.due, review)
# Whatever the two budgets could not fill. A bank with nothing unseen
# left, or nothing due, still owes the learner the length they asked
# for.
draw(self.rows, count - len(taken))
return taken
def adaptive_select(db, user, count, category_ids, state, difficulty, now=None):
"""Adaptive selection: the questions most likely to raise the learner's score.
Three rules, in order. Unanswered before answered, because a question never
seen teaches more than one already met. Among those, weakest topic first.
When the unanswered run out, recycle — wrong ones before right ones, oldest
first — damping each category as it is drawn from so a session of twenty
does not become twenty questions from the single worst subject.
Weakness is then multiplied by the topic's share of the real paper, which
the examining board publishes and `exam_blueprints.weight` holds. Being weak
at something worth 5% of the exam is worth more study than being equally
weak at something worth 1%, and until this was added the two ranked the
same. A topic the blueprint does not cover takes the median weight, so it
is neither promoted nor made unreachable.
The rules live on `CandidateRanking`, which the prepared session reads to
explain itself. This is the same selection reached from the Adaptive toggle
on the manual builder.
"""
from collections import defaultdict
query = filtered_bank_query(db, user, category_ids, state)
if difficulty:
query = query.filter(Question.difficulty == difficulty)
# No cap. This used to take the first 2,000 rows, which on a 2,948-question
# bank meant adaptive selection could not see about a third of it, and which
# third depended on database order. Two integer columns per question is not
# a size worth protecting against.
rows = query.with_entities(Question.id, Question.question_category_id).all()
if not rows:
return []
category_of = dict(rows)
answered = db.query(AttemptAnswer.question_id, AttemptAnswer.is_correct, QuizAttempt.completed_at).join(
QuizAttempt, QuizAttempt.id == AttemptAnswer.attempt_id).join(Quiz, Quiz.id == QuizAttempt.quiz_id).filter(
QuizAttempt.user_id == user.id, QuizAttempt.completed_at.isnot(None),
or_(QuizAttempt.expired == 0, QuizAttempt.expired.is_(None)), Quiz.course_id.is_(None),
AttemptAnswer.question_id.in_([r[0] for r in rows]),
).order_by(QuizAttempt.completed_at.desc(), QuizAttempt.id.desc()).all()
latest: dict[int, tuple] = {}
for qid, ok, when in answered:
latest.setdefault(qid, (ok, when))
cat_correct: dict = defaultdict(lambda: [0, 0])
for qid, ok, _ in answered:
# Looked up, not scanned. This was a linear search through every
# candidate for every answer — the slowest part of building a session.
category = category_of.get(qid)
if category is None:
continue
cat_correct[category][0] += 1
if ok:
cat_correct[category][1] += 1
def accuracy(category):
total, correct = cat_correct[category]
return correct / total if total else 0.5
weights = blueprint_weights(db, user)
# The middle of what the board publishes, for a topic it does not mention.
# A zero would make unmapped material unreachable; the highest would make
# it the priority. Neither is a claim the blueprint supports.
neutral = statistics.median(weights.values()) if weights else 1.0
def weight(category):
return weights.get(category, neutral) if weights else 1.0
def impact(category):
"""How much a question here could move the score."""
return (1 - accuracy(category)) * weight(category)
unanswered = [r for r in rows if r[0] not in latest]
recycled = [r for r in rows if r[0] in latest]
# Older incorrect questions resurface first among recycled.
recycled.sort(key=lambda r: (latest[r[0]][0], latest[r[0]][1]))
# Step 2: prefer unanswered, most impact first.
unanswered.sort(key=lambda r: -impact(r[1]))
selected = [r[0] for r in unanswered[:count]]
if len(selected) < count:
damping = defaultdict(lambda: 1.0)
while recycled and len(selected) < count:
# Wrong last time is worth about five times right last time, and
# the topic's share of the paper scales both.
best = max(recycled, key=lambda r: (
(1 - (0.85 if latest[r[0]][0] else 0.25)) * damping[r[1]] * weight(r[1])))
recycled.remove(best)
selected.append(best[0])
damping[best[1]] *= 0.5
return selected
return CandidateRanking(db, user, category_ids, state, difficulty, now).select(count)
def require_objective(db, user) -> None:

28
backend/app/services/search_service.py
View file

@ -206,3 +206,31 @@ def hybrid_ids(db: Session, query_text: str, kind: str = "question",
def hybrid_question_ids(db: Session, query_text: str, limit: int = 200) -> tuple[list[int], set[int]]:
"""Questions, for callers that predate the multi-corpus signature."""
return hybrid_ids(db, query_text, "question", limit)
def article_ids_with_sections(db: Session, query_text: str,
limit: int = 200) -> tuple[list[int], dict[int, list]]:
"""Article ids for a query, plus the section rows that matched, grouped by article.
An article's body lives in its sections, and the article's own row carries
only a topical vector and a weighted summary of that body. So a term that
appears in one section and nowhere else — a drug, a procedure, an eponym —
is found by searching the section corpus, not the article corpus.
Section hits are reported under their article rather than beside it: ten
matching sections of one article are one result with ten places to start
reading, not ten results that bury everything else.
"""
from app.models.article import ArticleSectionIndex
ranked, _ = hybrid_ids(db, query_text, "article", limit=limit)
section_ranked, _ = hybrid_ids(db, query_text, "article_section", limit=limit)
by_article: dict[int, list] = {}
if section_ranked:
position = {row_id: index for index, row_id in enumerate(section_ranked)}
rows = db.query(ArticleSectionIndex).filter(
ArticleSectionIndex.id.in_(section_ranked)).all()
for row in sorted(rows, key=lambda r: position.get(r.id, len(position))):
by_article.setdefault(row.article_id, []).append(row)
ordered = list(dict.fromkeys([*ranked, *by_article.keys()]))[:limit]
return ordered, by_article

459
backend/app/services/vision_service.py Normal file
View file

@ -0,0 +1,459 @@
"""Which models can see, and what happens when the one doing the job cannot.
A job's model is chosen for the job, not for the pictures that turn up in it.
The extraction and tutor models configured on this deployment report
`supports_vision: false`, and handing one an `image_url` part does not produce a
worse answer — it ends the request with a proxy error. So the picture goes to a
model that can see, the one an administrator sets for the `tool` job, and its
description is folded into the prompt as text, which a model that cannot see
reads perfectly well.
Capability is asked of the proxy rather than inferred from the model id: the
catalogue here runs to several hundred entries and changes without us.
`/model/info` carries `supports_vision` per deployment and settles most of it
outright — every OpenAI and Anthropic route, and a flat `false` on the DeepSeek
one. Where the field is absent the proxy genuinely does not know, and guessing
"no" would send work to the tool model that never needed to go there, so the
model itself is asked with an eight-pixel image, once.
Both answers are cached, because the question is about the model and not about
the request: the catalogue in-process for a few minutes, the probe in Redis for
a week. Descriptions are cached too — the tutor re-sends the same figure on
every turn of a conversation, and paying a second model call for each of them
was the first thing that made this feature look slow.
Nothing here degrades quietly. If the job's model cannot see and no tool model
is configured, the caller gets an error naming the setting to change; a
text-only answer about an image nobody looked at is worse than no answer.
"""
import base64
import hashlib
import io
import logging
import mimetypes
import re
import time
from dataclasses import dataclass, field
from app.config import settings
from app.services.ai_service import get_client, get_configured_model
logger = logging.getLogger(__name__)
#: How long the proxy's catalogue is trusted. Models are added to it by hand,
#: so minutes are plenty and a stale "cannot see" is only ever a slower answer.
CATALOGUE_TTL = 600
#: A probe verdict is a fact about the model, not about today, so it is kept
#: long enough to be worth having and short enough to survive a proxy rewiring.
PROBE_TTL = 7 * 86400
DESCRIPTION_TTL = 30 * 86400
#: Beyond this the base64 payload costs more than the detail is worth — a stem
#: image here is a 2–4 MB scanned radiograph, and 1600px of it answers the same
#: question at a tenth the size.
MAX_IMAGE_BYTES = 1_500_000
DOWNSCALE_WIDTH = 1600
#: What the tool model is asked. American, like the rest of the clinical copy.
DESCRIBE_PROMPT = (
"Describe this image for a colleague who cannot see it and must reason from "
"your words alone.\n"
"- Report what is visible: modality, body part, structures, colors, "
"measurements, axes, arrows, and any text printed on the image, quoted exactly.\n"
"- Describe abnormal findings in clinical terms. Do not name a diagnosis "
"unless the image itself is labeled with one.\n"
"- Say plainly if the image is a logo, page header, blank, or otherwise "
"carries no clinical content.\n"
"- Description only: no preamble, no answer to any question."
)
_catalogue: dict[str, bool | None] = {}
_catalogue_at: float = 0.0
_probes: dict[str, bool] = {}
class VisionUnavailable(RuntimeError):
"""No model available to look at an image the request depends on.
Raised rather than returned so that no caller can carry on without noticing
that the picture went unread. The message names the setting to change.
"""
@dataclass(frozen=True)
class Image:
data: bytes
media_type: str = "image/jpeg"
caption: str = ""
@dataclass
class Handoff:
"""What actually ran, so nobody has to infer a second model call from a
latency graph."""
primary_model: str
images: int = 0
delegated: bool = False
tool_model: str | None = None
reason: str = ""
cached: bool = False
elapsed_ms: int = 0
descriptions: list[str] = field(default_factory=list)
def as_dict(self) -> dict:
return {
"delegated": self.delegated,
"primary_model": self.primary_model,
"tool_model": self.tool_model,
"images": self.images,
"reason": self.reason,
"cached": self.cached,
"elapsed_ms": self.elapsed_ms,
}
def _redis():
try:
import redis as redis_lib
return redis_lib.from_url(settings.REDIS_URL, decode_responses=True,
socket_connect_timeout=1)
except Exception:
return None
def _cached(key: str) -> str | None:
client = _redis()
if client is None:
return None
try:
return client.get(key)
except Exception:
return None
def _remember(key: str, value: str, ttl: int) -> None:
client = _redis()
if client is None:
return
try:
client.set(key, value, ex=ttl)
except Exception:
# A cache that is down is a cost, never a failure.
logger.debug("Could not cache %s", key, exc_info=True)
def catalogue() -> dict[str, bool | None]:
"""model name -> what the proxy says about vision, refreshed on a timer.
A name can appear several times, once per deployment behind it: `best-chat`
fans out to six. A single `false` among them decides the alias, because a
request can land on any of them and an alias that fails one time in six is
worse than one that always takes the slower path.
"""
global _catalogue, _catalogue_at
if _catalogue and time.monotonic() - _catalogue_at < CATALOGUE_TTL:
return _catalogue
import httpx
base = (settings.LITELLM_API_BASE or "").rstrip("/").removesuffix("/v1")
if not base:
return {}
headers = {"Authorization": f"Bearer {settings.LITELLM_API_KEY}"} if settings.LITELLM_API_KEY else {}
try:
response = httpx.get(f"{base}/model/info", headers=headers, timeout=10)
response.raise_for_status()
rows = response.json().get("data", [])
except Exception:
# Leave whatever is already known in place: an unreachable catalogue is
# a reason to fall back to probing, not to forget what it said before.
logger.warning("Could not read the model catalogue; vision capability "
"falls back to probing", exc_info=True)
return _catalogue
verdicts: dict[str, bool | None] = {}
for row in rows:
name = row.get("model_name")
if not name:
continue
says = (row.get("model_info") or {}).get("supports_vision")
known = verdicts.get(name, "absent")
if says is False or known is False:
verdicts[name] = False
elif says is True:
verdicts[name] = True
elif known == "absent":
verdicts[name] = None
_catalogue, _catalogue_at = verdicts, time.monotonic()
return _catalogue
def _probe_image() -> str:
"""An eight-pixel PNG as a data URL. Deliberately not one pixel: some
providers reject an image below a minimum dimension, and a refusal of the
probe would read as a refusal of images."""
from PIL import Image as PILImage
buffer = io.BytesIO()
PILImage.new("RGB", (8, 8), (255, 255, 255)).save(buffer, format="PNG")
return data_url(buffer.getvalue(), "image/png")
def _probe(model_id: str, api_key: str | None) -> bool:
"""Ask the model itself, for the models the catalogue has no opinion on.
A 4xx is the proxy or the provider refusing the shape of the request, which
for a one-token call carrying nothing but a white square means it will not
take images; that verdict is worth keeping. A timeout or a 5xx says nothing
about the model, so it is not cached and the caller takes the safe path
this once.
"""
cache_key = f"vision:probe:{model_id}"
if model_id in _probes:
return _probes[model_id]
remembered = _cached(cache_key)
if remembered is not None:
_probes[model_id] = remembered == "1"
return _probes[model_id]
try:
get_client(api_key, timeout=30).chat.completions.create(
model=model_id,
max_tokens=1,
messages=[{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": _probe_image()}},
{"type": "text", "text": "Reply with: ok"},
]}],
)
verdict = True
except Exception as error:
status = getattr(error, "status_code", None)
if not (isinstance(status, int) and 400 <= status < 500):
logger.info("Vision probe of %s was inconclusive: %s", model_id, error)
return False
logger.info("Vision probe: %s refused an image (%s)", model_id, error)
verdict = False
_probes[model_id] = verdict
_remember(cache_key, "1" if verdict else "0", PROBE_TTL)
return verdict
def can_see(model_id: str, api_key: str | None = None) -> bool:
"""Whether this model may be handed an image. Cached both ways."""
if not model_id:
return False
verdict = catalogue().get(model_id, None)
if verdict is not None:
return verdict
return _probe(model_id, api_key)
def data_url(data: bytes, media_type: str) -> str:
return f"data:{media_type};base64,{base64.b64encode(data).decode()}"
#: How a JPEG 2000 file starts — the JP2 container, then a bare codestream.
#: Sniffed rather than taken from the name because the name lies: the stem
#: images extracted from these PDFs are `.jpx` files that `mimetypes` reports
#: as `image/jpeg`, and the provider answers "the image data you provided does
#: not represent a valid image" some seconds and two fallback hops later.
JPEG_2000_SIGNATURES = (b"\x00\x00\x00\x0cjP ", b"\xff\x4f\xff\x51")
def prepare(data: bytes, media_type: str | None = None, caption: str = "") -> Image | None:
"""Bytes as a vision API will take them, or None if they cannot be.
Two conversions, both learned the hard way: JPEG 2000 comes out of the PDF
pipeline and no vision API accepts it, and a full-resolution scan spends
more on base64 than the extra pixels are worth.
"""
if not data:
return None
media_type = media_type or "image/jpeg"
if data.startswith(JPEG_2000_SIGNATURES):
media_type = "image/jp2"
if media_type in ("image/jp2", "image/jpx"):
try:
import fitz # already a dependency of the PDF pipeline
data, media_type = fitz.Pixmap(data).tobytes("png"), "image/png"
except Exception:
logger.info("Could not re-encode a JPEG 2000 image for a vision model",
exc_info=True)
return None
if len(data) > MAX_IMAGE_BYTES:
from app.services import thumbnails
smaller = thumbnails.render(data, DOWNSCALE_WIDTH)
if smaller:
data, media_type = smaller, "image/webp"
return Image(data=data, media_type=media_type, caption=caption)
def load_image(key: str, caption: str = "") -> Image | None:
"""A stored upload, ready for a model. Never raises."""
if not key:
return None
try:
if re.match(r"^https?://", key, re.I):
import httpx
response = httpx.get(key, timeout=10, follow_redirects=True)
response.raise_for_status()
return prepare(response.content,
response.headers.get("content-type", "").split(";")[0] or None,
caption)
from app.services import storage_service
# Some rows hold the URL the browser uses rather than the storage key;
# `/uploads/` is that route's prefix and nothing in the bucket has it.
key = key.removeprefix("/uploads/")
data = storage_service.s3_object(key) or storage_service.load(key)
return prepare(data, mimetypes.guess_type(key)[0], caption) if data else None
except Exception:
logger.info("Could not load %s for a vision model", key, exc_info=True)
return None
def image_part(image: Image) -> dict:
return {"type": "image_url", "image_url": {"url": data_url(image.data, image.media_type)}}
def word_image(word: str) -> Image:
"""A picture of a word, for testing that a model can read a picture.
Presence in the catalogue is not proof, the same way a transcription model
listed by the proxy is not proof that anything can be transcribed. The test
makes the tool model read something only an eye could have read.
"""
from PIL import Image as PILImage, ImageDraw, ImageFont
canvas = PILImage.new("RGB", (480, 160), (255, 255, 255))
draw = ImageDraw.Draw(canvas)
try:
font = ImageFont.load_default(size=64)
except TypeError: # Pillow before 9.2 cannot size the default font
font = ImageFont.load_default()
draw.text((40, 45), word, fill=(0, 0, 0), font=font)
buffer = io.BytesIO()
canvas.save(buffer, format="PNG")
return Image(data=buffer.getvalue(), media_type="image/png", caption="test image")
def describe(images: list[Image], model_id: str, api_key: str | None = None,
context: str = "", use_cache: bool = True) -> list[str]:
"""What the tool model sees, one description per image.
Cached on the bytes, the surrounding context and the model, because the
same figure comes back on every turn of a tutor conversation and none of
those turns changes what the picture shows.
"""
return _describe(images, model_id, api_key, context, use_cache)[0]
def _describe(images: list[Image], model_id: str, api_key: str | None,
context: str, use_cache: bool) -> tuple[list[str], int]:
"""The descriptions, and how many of them came from the cache."""
out: list[str] = []
hits = 0
for image in images:
digest = hashlib.sha256(
image.data + context.encode() + model_id.encode()).hexdigest()[:24]
cache_key = f"vision:description:{digest}"
if use_cache:
remembered = _cached(cache_key)
if remembered:
out.append(remembered)
hits += 1
continue
prompt = DESCRIBE_PROMPT
if image.caption:
prompt += f"\n\nWhat this image is: {image.caption}"
if context:
prompt += f"\n\nWhere it appears:\n{context[:2000]}"
completion = get_client(api_key, timeout=120).chat.completions.create(
model=model_id,
temperature=0,
max_tokens=700,
messages=[{"role": "user", "content": [
image_part(image), {"type": "text", "text": prompt},
]}],
)
text = (completion.choices[0].message.content or "").strip()
if not text:
raise VisionUnavailable(
f"{model_id} is configured as the tool model but returned no "
"description of the image.")
out.append(text)
if use_cache:
_remember(cache_key, text, DESCRIPTION_TTL)
return out, hits
def image_context(db, images: list[Image], *, model_id: str, api_key: str | None = None,
context: str = "", tool: tuple[str, str | None] | None = None
) -> tuple[list[dict], Handoff]:
"""Message content parts carrying these images, whatever the model can do.
The caller splices the parts into a user message and does not need to know
which of the two things happened: either the images themselves, or a
description of each written by the tool model. The Handoff says which, for
the log line and for the response.
`tool` names the tool model outright, for callers running in a thread pool:
a SQLAlchemy session belongs to one thread, so those resolve it once on the
way in and pass it down rather than handing `db` to every worker.
"""
handoff = Handoff(primary_model=model_id, images=len(images))
if not images:
return [], handoff
started = time.monotonic()
if can_see(model_id, api_key):
handoff.reason = "the model reads images itself"
handoff.elapsed_ms = int((time.monotonic() - started) * 1000)
return [image_part(image) for image in images], handoff
tool = tool or get_configured_model(db, "tool")
if not tool:
raise VisionUnavailable(
f"{model_id} cannot read images and no tool model is configured. "
"In Settings → AI models, allow a model that can see for the Tool "
"job and select it.")
tool_model, tool_key = tool
if catalogue().get(tool_model) is False:
raise VisionUnavailable(
f"{model_id} cannot read images, and the tool model set to cover it "
f"({tool_model}) cannot either. Choose a model that can see for the "
"Tool job in Settings → AI models.")
handoff.delegated = True
handoff.tool_model = tool_model
handoff.reason = f"{model_id} cannot read images"
try:
handoff.descriptions, hits = _describe(images, tool_model, tool_key, context, True)
except VisionUnavailable:
raise
except Exception as error:
raise VisionUnavailable(
f"The tool model {tool_model} could not read the image: {error}") from error
handoff.elapsed_ms = int((time.monotonic() - started) * 1000)
handoff.cached = hits == len(images)
parts = []
for image, text in zip(images, handoff.descriptions):
what = image.caption or "an image accompanying this request"
parts.append({"type": "text", "text": (
f"[Description of {what}. You cannot see images, so {tool_model} "
f"looked at it and wrote this. Treat it as what the image shows.]\n{text}")})
logger.info("Vision handoff: %s cannot see, %s described %d image(s) in %d ms",
model_id, tool_model, len(images), handoff.elapsed_ms)
return parts, handoff

13
backend/app/tasks/quiz_tasks.py
View file

@ -580,6 +580,7 @@ def generate_article_draft(self, job_id: str, user_id: int, topic: str,
db = SessionLocal()
try:
from app.models.article import Article
from app.services import article_service
from app.services.ai_service import get_model_for_task, get_client
from app.config import settings
@ -630,16 +631,22 @@ def generate_article_draft(self, job_id: str, user_id: int, topic: str,
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
article = existing
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"))
article = 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.add(article)
db.commit()
# A draft that is not indexed is a draft nobody can find. Every writer of
# `Article.sections` has to do this; the ones that did not left 323
# articles with no section rows and a vector built from the title alone.
article_service.reindex(db, article)
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:

3
backend/scripts/article_pipeline.py
View file

@ -157,6 +157,9 @@ def cmd_import(args):
db.flush()
article_service.record_slug(db, article)
db.commit()
# Indexed on the way in, so an imported article is searchable by what it
# says rather than waiting for somebody to notice and run a backfill.
article_service.reindex(db, article)
print(f" {topic}: imported as draft #{article.id} "
f"({len(sections)} sections, {len(article.references_json)} references)")
finally:

120
backend/scripts/reindex_article_search.py Normal file
View file

@ -0,0 +1,120 @@
"""Make the semantic half of article search look at what an article says.
Two holes, one cause. Everything the generator writes goes into the `sections`
JSON column and `articles.content` stays NULL, so:
* the article's vector was built from title and summary alone for 323 of the
331 articles — enough to tell "Asthma" from "Migraine" and nothing finer;
* `article_section_index`, the projection that carries section-level retrieval,
held 24 rows covering the 8 hand-seeded samples. The generated articles had
no rows in it at all, so a term appearing once in one section — a drug, a
procedure, an eponym — was unreachable by meaning and reachable lexically
only after the `search_vector` fix in migration d6e7f8091a2b.
This re-embeds articles under the composition rule in
`embedding_service.article_embedding_text` and projects every section into the
index, embedding each one whole.
Resumable and idempotent by construction. `embed_records` stamps `embedded_at`,
so an article finished by a previous run is skipped by the same freshness test
that skips one nobody has edited; sections carry their own text, so an
unchanged section is compared and left alone rather than re-embedded. A run
that dies at article 200 costs nothing but the articles it had not reached.
docker compose exec backend python -m scripts.reindex_article_search
docker compose exec backend python -m scripts.reindex_article_search --apply
# after a change to how the article vector is composed, not just to the text:
docker compose exec backend python -m scripts.reindex_article_search --apply --all
Back up first; the writes touch every article row.
docker compose exec -T postgres sh -lc \\
'pg_dump -U "$POSTGRES_USER" "$POSTGRES_DB" -t articles -t article_section_index --data-only' \\
> backups/articles-$(date +%Y%m%d%H%M).sql
"""
import sys
import time
from datetime import datetime
from app.database import SessionLocal
from app.models.article import Article, ArticleSectionIndex
from app.services import article_service, embedding_service
# Commit boundary. Small enough that a crash loses seconds of work, large
# enough that the commit is not what the run spends its time on.
CHUNK = 20
def _needs_article_vector(article, active_model: str, started: datetime, force: bool) -> bool:
"""Whether this article's own vector is worth paying for again.
`embedded_at < updated_at` catches an edit. `--all` catches a change to the
composition rule, which no column records — bounded by the run's start time
so that resuming does not begin again from the top.
"""
if article.embedding is None or article.embedding_model != active_model:
return True
if article.embedded_at is None or article.embedded_at < (article.updated_at or datetime.min):
return True
return force and article.embedded_at < started
def main() -> int:
apply_changes = "--apply" in sys.argv
force = "--all" in sys.argv
started = datetime.utcnow()
db = SessionLocal()
try:
active = embedding_service._get_embedding_model()
articles = db.query(Article).order_by(Article.id).all()
indexed = {row[0] for row in db.query(ArticleSectionIndex.article_id).distinct()}
sections = sum(len(a.sections or []) for a in articles)
todo = [a for a in articles if _needs_article_vector(a, active, started, force)]
print(f" model : {active}")
print(f" articles : {len(articles)}")
print(f" sections in JSON : {sections}")
print(f" articles indexed : {len(indexed)} ({len(articles) - len(indexed)} with no section rows)")
print(f" article vectors : {len(todo)} to (re)generate")
if not apply_changes:
print("\n Re-run with --apply to write. Add --all to recompose every article vector.")
return 0
began = time.monotonic()
article_vectors, section_vectors, failed = 0, 0, []
for start in range(0, len(articles), CHUNK):
chunk = articles[start:start + CHUNK]
try:
wanted = [a for a in chunk if _needs_article_vector(a, active, started, force)]
article_vectors += embedding_service.embed_records(wanted, "article")
for article in chunk:
section_vectors += article_service.rebuild_section_index(db, article)
db.commit()
except Exception as exc:
# One bad article must not cost the run; it is reported and the
# next chunk carries on, because a sweep that has to be restarted
# from zero is a sweep nobody runs.
db.rollback()
failed.extend(a.id for a in chunk)
print(f" chunk at {start} failed: {exc}", flush=True)
done = min(start + CHUNK, len(articles))
print(f" …{done}/{len(articles)} "
f"{article_vectors} article + {section_vectors} section vectors", flush=True)
elapsed = time.monotonic() - began
rows = db.query(ArticleSectionIndex).count()
print(f"\n article vectors : {article_vectors}")
print(f" section vectors : {section_vectors}")
print(f" index rows now : {rows}")
print(f" elapsed : {elapsed:.0f}s")
if failed:
print(f" failed articles : {sorted(set(failed))}")
print(" Re-run to pick them up; finished work is skipped.")
finally:
db.close()
return 0
if __name__ == "__main__":
sys.exit(main())

1159
backend/scripts/rewrite_catalogue_summaries.py
View file

File diff suppressed because it is too large Load diff

4
backend/scripts/seed_sample_content.py
View file

@ -8,6 +8,7 @@ import sys
from app.database import SessionLocal
from app.models.article import Article, QuestionArticleLink
from app.services import article_service
from app.models.flashcard import Flashcard, FlashcardDeck, FlashcardArticleLink, FlashcardQuestionLink
from app.models.lab_reference import LabReference, LabReferenceCardLink
from app.models.question import Question
@ -55,6 +56,9 @@ def main():
articles.append(article)
print(f"Created sample article: {title}")
db.commit()
# Seeded content has to be findable by what it says, like anything else.
for article in articles:
article_service.reindex(db, article)
# ── Link bank questions ─────────────────────────────────────────
from app.services.quiz_builder import shareable_question_predicate

7
backend/scripts/seed_showcase_articles.py
View file

@ -14,6 +14,7 @@ import sys
from app.database import SessionLocal
from app.models.article import Article, QuestionArticleLink
from app.services import article_service
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.user import User
@ -167,7 +168,7 @@ def main():
moderator = moderator[0] if moderator else None
categories = {c.name: c.id for c in db.query(QuestionCategory).filter(QuestionCategory.parent_id.is_(None)).all()}
created = 0
created, made = 0, []
for slug, title, category_name, summary, intro, sections in SHOWCASE:
if db.query(Article.id).filter(Article.slug == slug).first():
print(f"skip (exists): {slug}")
@ -189,9 +190,13 @@ def main():
status="published",
)
db.add(article)
made.append(article)
created += 1
print(f"created: {title} ({category_name})")
db.commit()
# Seeded content has to be findable by what it says, like anything else.
for article in made:
article_service.reindex(db, article)
# Link up to three bank questions from the same category to each article.
linked = 0

58
backend/scripts/triage_question_images.py
View file

@ -27,16 +27,16 @@ edit UI uses, so any of this can be restored from the question's version history
--no-vision only apply the deterministic rule, ask no model
--limit N classify at most N unknowns (for a spot check)
--model ID override the model (default: the configured extraction model)
--model ID override the model (default: the configured extraction model,
which on this deployment cannot see — the tool model then
describes each picture for it)
--report PATH where the human-review list is written
Anything the model is not confident about is left exactly as it is and listed in
the report, with the question id and the image path, for a person to decide.
"""
import argparse
import base64
import json
import mimetypes
import re
import sys
from concurrent.futures import ThreadPoolExecutor
@ -44,8 +44,8 @@ from concurrent.futures import ThreadPoolExecutor
from app.database import SessionLocal
from app.models.question import Question
from app.routers.questions import _snapshot_question
from app.services import storage_service
from app.services.ai_service import get_client, get_model_for_task
from app.services import vision_service
from app.services.ai_service import get_client, get_configured_model, get_model_for_task
SCRIPT_EDITOR_ID = None
@ -102,39 +102,29 @@ Answer with ONLY this JSON, no markdown fence and no preamble:
"reason": "<one short sentence>"}}"""
def image_bytes(key: str) -> tuple[bytes, str] | None:
"""The image and the media type a vision model will accept, or None."""
data = storage_service.s3_object(key) or storage_service.load(key)
if not data:
return None
media_type = mimetypes.guess_type(key)[0] or "image/jpeg"
if key.lower().endswith((".jpx", ".jp2")) or media_type in ("image/jp2", "image/jpx"):
# JPEG 2000 is not accepted by the vision APIs; PyMuPDF is already a
# dependency of the PDF pipeline and re-encodes it without a new one.
try:
import fitz
return fitz.Pixmap(data).tobytes("png"), "image/png"
except Exception:
return None
return data, media_type
def classify(stem: str, explanation: str, key: str, model: str,
api_key: str | None) -> dict:
"""Ask the vision model where the image belongs. Never raises."""
loaded = image_bytes(key)
if not loaded:
api_key: str | None, tool: tuple[str, str | None] | None) -> dict:
"""Ask where the image belongs. Never raises.
The model that judges is the configured one, which on this deployment
cannot see: `image_context` then has the tool model describe the picture
and hands the judge that description instead, so the answer comes back in
the same shape either way. The tool model is resolved by the caller because
this runs in a thread pool and the session does not.
"""
image = vision_service.load_image(key)
if not image:
return {"belongs": "unclear", "confidence": 0.0, "reason": "image could not be read"}
data, media_type = loaded
try:
parts, _ = vision_service.image_context(
None, [image], model_id=model, api_key=api_key, tool=tool,
context=f"A figure printed with this board-exam question:\n{(stem or '')[:1000]}")
completion = get_client(api_key).chat.completions.create(
model=model,
temperature=0,
messages=[{"role": "user", "content": [
{"type": "image_url", "image_url": {
"url": f"data:{media_type};base64,{base64.b64encode(data).decode()}"}},
*parts,
{"type": "text", "text": PROMPT.format(
stem=(stem or "")[:4000] or "(empty)",
explanation=(explanation or "")[:4000] or "(empty)")},
@ -201,12 +191,18 @@ def main() -> int:
if unknown and not args.no_vision:
model = args.model or get_model_for_task(db, "extraction")[0]
api_key = None if args.model else get_model_for_task(db, "extraction")[1]
# Resolved here, on the thread that owns the session.
tool = get_configured_model(db, "tool")
batch = unknown[: args.limit] if args.limit else unknown
print(f"\n asking {model} about {len(batch)} image(s)…")
if not vision_service.can_see(model, api_key):
print(f" {model} cannot read images; "
+ (f"{tool[0]} will describe them" if tool
else "no tool model is configured — set one in Settings → AI models"))
payload = [(q.id, q.question_text, q.explanation, q.image_path) for q in batch]
with ThreadPoolExecutor(max_workers=args.workers) as pool:
results = list(pool.map(
lambda row: (row[0], classify(row[1], row[2], row[3], model, api_key)),
lambda row: (row[0], classify(row[1], row[2], row[3], model, api_key, tool)),
payload))
judged = dict(results)

170
backend/tests/test_article_search_index.py Normal file
View file

@ -0,0 +1,170 @@
"""What an article's vector is made of, and whether the section index keeps up.
Both halves of article search read from a projection rather than from the
article row: `articles.content` is NULL for all but the eight hand-seeded
samples, so anything that only looked at title, summary and `content` was
describing an empty body. These fix that in place, on disposable SQLite, with
no network — the embedder returns nothing without credentials, which is exactly
the "leave it for the retry task" path the routes are meant to survive.
"""
import unittest
import unittest.mock
from unittest.mock import patch
import test_quiz_builder as fixtures
from app.models.article import Article, ArticleSectionIndex
from app.routers import articles
from app.services.embedding_service import article_embedding_text
def _article(section_count=12, body_chars=1500):
sections = [{
"id": f"{index:032d}", "slug": f"section-{index}", "variant": "long",
"title": f"Heading {index}",
"content": f"OPENING{index} " + ("filler " * (body_chars // 7)) + f" CLOSING{index}",
} for index in range(section_count)]
return Article(title="Laryngomalacia", summary="Inspiratory stridor in an infant",
content=None, sections=sections)
class ArticleEmbeddingTextTests(unittest.TestCase):
def test_body_reaches_the_vector_at_all(self):
article = _article()
text = article_embedding_text(article)
self.assertIn("Laryngomalacia", text)
self.assertIn("Inspiratory stridor", text)
# The whole point: something from the body, which used to contribute
# nothing because it is in `sections` and not in `content`.
self.assertIn("OPENING0", text)
def test_every_section_is_represented_not_just_the_head(self):
"""Head truncation would stop in the pathophysiology and drop treatment."""
article = _article()
text = article_embedding_text(article)
for index in range(12):
self.assertIn(f"Heading {index}", text, "section outline is missing an entry")
self.assertIn(f"OPENING{index}", text, "a later section contributed nothing")
def test_stays_inside_the_embedder_clamp(self):
# `generate_embedding` truncates at 4000 characters without saying so,
# which is how an index comes to be silently wrong.
self.assertLessEqual(len(article_embedding_text(_article(14, 4000))), 4000)
def test_short_article_is_carried_whole(self):
article = _article(section_count=2, body_chars=100)
text = article_embedding_text(article)
for index in range(2):
self.assertIn(f"CLOSING{index}", text)
def test_missing_and_malformed_sections_do_not_raise(self):
empty = Article(title="Stub", summary=None, content=None, sections=None)
self.assertEqual(article_embedding_text(empty), "Stub")
odd = Article(title="Stub", summary=None, content=None,
sections=["not a dict", {"title": "Only a heading"}])
self.assertIn("Only a heading", article_embedding_text(odd))
class BatchEmbeddingTests(unittest.TestCase):
"""A batch attaching one document's vector to another is silent and permanent."""
def _proxy(self, payload):
from app.config import settings
from app.services import embedding_service
response = unittest.mock.Mock()
response.raise_for_status.return_value = None
response.json.return_value = payload
return patch.multiple(settings, LITELLM_API_KEY="k", LITELLM_API_BASE="https://proxy.test"), \
patch.object(embedding_service, "_get_embedding_model", return_value="m"), \
patch("httpx.post", return_value=response)
def _vectors(self, count, dim):
return {"data": [{"index": index, "embedding": [float(index)] * dim}
for index in range(count)]}
def test_vectors_land_on_the_row_they_were_made_from(self):
from app.config import settings
from app.services import embedding_service
dim = settings.EMBEDDING_DIMENSIONS
# Returned out of order, as a proxy is entitled to do.
payload = {"data": list(reversed(self._vectors(3, dim)["data"]))}
for context in self._proxy(payload):
context.start()
self.addCleanup(context.stop)
out = embedding_service.generate_embeddings(["one", "two", "three"])
self.assertEqual([vector[0] for vector in out], [0.0, 1.0, 2.0])
def test_blank_inputs_keep_their_place_in_the_result(self):
from app.config import settings
from app.services import embedding_service
dim = settings.EMBEDDING_DIMENSIONS
for context in self._proxy(self._vectors(2, dim)):
context.start()
self.addCleanup(context.stop)
out = embedding_service.generate_embeddings(["one", "", " ", "two"])
self.assertIsNone(out[1])
self.assertIsNone(out[2])
self.assertEqual([out[0][0], out[3][0]], [0.0, 1.0])
class SectionIndexInStepTests(unittest.TestCase):
def setUp(self):
self.bank = fixtures.BuilderTests()
self.bank.setUp()
self.client = self.bank.client
self.client.app.include_router(articles.router, prefix='/articles')
self.db = self.bank.db
self.bank.user = self.bank.mod
def tearDown(self):
self.bank.tearDown()
def rows(self, article_id):
return {row.section_id: (row.title, row.content) for row in
self.db.query(ArticleSectionIndex).filter_by(article_id=article_id).all()}
def payload(self, sections, **overrides):
return {"title": "Laryngomalacia", "slug": "laryngomalacia",
"summary": "Inspiratory stridor", "sections": sections, **overrides}
def test_index_follows_create_edit_and_delete_of_a_section(self):
first = {"id": "a" * 32, "slug": "definition", "title": "Definition", "content": "Dynamic collapse"}
second = {"id": "b" * 32, "slug": "treatment", "title": "Treatment", "content": "Supraglottoplasty"}
article = self.client.post('/articles/', json=self.payload([first, second])).json()
self.assertEqual(set(self.rows(article['id'])), {"a" * 32, "b" * 32})
self.assertEqual(self.rows(article['id'])["b" * 32][1], "Supraglottoplasty")
# Edit one, drop the other, add a third.
edited = {**second, "content": "Supraglottoplasty for severe cases"}
third = {"id": "c" * 32, "slug": "prognosis", "title": "Prognosis", "content": "Resolves by two years"}
self.client.patch(f"/articles/{article['id']}", json=self.payload([edited, third]))
rows = self.rows(article['id'])
self.assertEqual(set(rows), {"b" * 32, "c" * 32}, "a removed section left its row behind")
self.assertEqual(rows["b" * 32][1], "Supraglottoplasty for severe cases")
def test_a_term_only_in_a_section_body_finds_its_article(self):
"""The article row says nothing about it; the section row is the only hit."""
sections = [{"id": "a" * 32, "slug": "treatment", "title": "Treatment",
"content": "Oral griseofulvin for six to eight weeks"}]
article = self.client.post('/articles/', json=self.payload(
sections, title="Tinea capitis", slug="tinea-capitis",
summary="Scalp ringworm")).json()
self.client.post(f"/articles/{article['id']}/publish", json={'published': True})
found = self.client.get('/articles/', params={'q': 'griseofulvin'}).json()
self.assertEqual([a['id'] for a in found], [article['id']])
def test_an_orphaned_index_row_cannot_resurrect_a_deleted_article(self):
sections = [{"id": "a" * 32, "slug": "treatment", "title": "Treatment", "content": "Griseofulvin"}]
article = self.client.post('/articles/', json=self.payload(sections)).json()
self.client.patch(f"/articles/{article['id']}", json=self.payload([]))
self.assertEqual(self.rows(article['id']), {})
self.assertEqual(self.client.get('/articles/', params={'q': 'griseofulvin'}).json(), [])
if __name__ == '__main__':
unittest.main()

367
backend/tests/test_prepared_session.py Normal file
View file

@ -0,0 +1,367 @@
"""The prepared session: the arithmetic, and whether the plan tells the truth.
The test that matters here is the last one. A plan that does not describe the
session it starts is worse than no plan, so the account handed to the learner
is checked against the questions actually put in the quiz — not against the
ranking that produced them, which would only prove the code agrees with itself.
"""
import os
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
import sys
import unittest
from datetime import datetime, timedelta
from decimal import Decimal
from types import ModuleType
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
import test_quiz_builder # noqa: F401 — imports every model, so the metadata resolves
from app.database import Base, get_db
from app.models.attempt import AttemptAnswer, QuizAttempt
from app.models.exam import BlueprintCategoryLink, Exam, ExamBlueprint, QuestionExamLink
from app.models.question import Question
from app.models.question_category import QuestionCategory
from app.models.quiz import Quiz
from app.models.quiz_question_link import QuizQuestionLink
from app.models.user import User
from app.services import prepared_session
from app.services.quiz_builder import (DUE_RECALL, EVIDENCE_HALF_LIFE_DAYS, MAX_REVIEW_SHARE,
NEUTRAL_RECALL, CandidateRanking, recall_probability,
recency_weight)
from app.utils.auth import get_current_user
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
from app.routers import questions
#: Cardiology is filed a level down, which is where the plan's grouping has to
#: reach up from; the other two disciplines hold their questions directly.
CARDIOLOGY, CONGENITAL, RHEUMATOLOGY, DERMATOLOGY = 1, 2, 3, 4
class DecayTests(unittest.TestCase):
"""No database: the curve on its own."""
def test_evidence_halves_once_a_half_life(self):
self.assertAlmostEqual(recency_weight(0), 1.0)
self.assertAlmostEqual(recency_weight(EVIDENCE_HALF_LIFE_DAYS), 0.5)
self.assertAlmostEqual(recency_weight(3 * EVIDENCE_HALF_LIFE_DAYS), 0.125)
# An answer dated in the future is a clock disagreement, not evidence
# worth more than a fresh one.
self.assertAlmostEqual(recency_weight(-40), 1.0)
def test_both_outcomes_decay_towards_a_coin_flip_rather_than_past_it(self):
fresh_hit, old_hit = recall_probability(True, 0), recall_probability(True, 365)
fresh_miss, old_miss = recall_probability(False, 0), recall_probability(False, 365)
self.assertGreater(fresh_hit, old_hit)
self.assertLess(fresh_miss, old_miss)
# Time turns a right answer into "no idea", never into a wrong one.
self.assertAlmostEqual(old_hit, NEUTRAL_RECALL, places=2)
self.assertAlmostEqual(old_miss, NEUTRAL_RECALL, places=2)
self.assertGreater(old_hit, NEUTRAL_RECALL)
self.assertLess(old_miss, NEUTRAL_RECALL)
def test_a_recent_miss_outranks_an_older_one(self):
# The ranking is worth (1 − recall), so lower recall is picked first.
self.assertLess(recall_probability(False, 7), recall_probability(False, 90))
def test_a_correct_answer_comes_back_round_once_it_is_old_enough(self):
self.assertGreater(recall_probability(True, 3), DUE_RECALL)
self.assertLess(recall_probability(True, 90), DUE_RECALL)
# And a wrong answer is due from the moment it is given.
self.assertLess(recall_probability(False, 0), DUE_RECALL)
class Bank(unittest.TestCase):
"""Thirty questions over three disciplines, weighted 60 / 10 / 30."""
NOW = datetime(2026, 6, 1)
def setUp(self):
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
poolclass=StaticPool)
Base.metadata.create_all(self.engine)
self.db = Session(self.engine)
self.db.add(Exam(id=1, slug="boards", name="Pediatrics Boards", is_active=1))
self.user = User(id=1, name="Learner", email="l@example.test",
hashed_password="unused", active_exam_id=1)
self.db.add(self.user)
self.db.add_all([
QuestionCategory(id=CARDIOLOGY, name="Cardiology", user_id=1),
QuestionCategory(id=CONGENITAL, name="Congenital", parent_id=CARDIOLOGY, user_id=1),
QuestionCategory(id=RHEUMATOLOGY, name="Rheumatology", user_id=1),
QuestionCategory(id=DERMATOLOGY, name="Dermatology", user_id=1),
])
self.db.add(Quiz(id=1, title="Origin", user_id=1, is_published=1))
self.db.flush()
self.of_category = {}
for qid in range(1, 31):
category = CONGENITAL if qid <= 12 else RHEUMATOLOGY if qid <= 21 else DERMATOLOGY
self.of_category[qid] = category
self.db.add(Question(id=qid, question_category_id=category, user_id=1,
question_text=f"Q{qid}", question_type="mcq",
options=["a", "b"], correct_answer="a"))
self.db.add(QuestionExamLink(question_id=qid, exam_id=1))
self.db.add_all([
ExamBlueprint(id=1, exam_id=1, code="1", title="Cardiology", weight=Decimal("60")),
ExamBlueprint(id=2, exam_id=1, code="2", title="Rheumatology", weight=Decimal("10")),
ExamBlueprint(id=3, exam_id=1, code="3", title="Dermatology", weight=Decimal("30")),
])
self.db.flush()
self.db.add_all([BlueprintCategoryLink(blueprint_id=1, category_id=CARDIOLOGY),
BlueprintCategoryLink(blueprint_id=2, category_id=RHEUMATOLOGY),
BlueprintCategoryLink(blueprint_id=3, category_id=DERMATOLOGY)])
self.db.commit()
app = FastAPI()
app.include_router(questions.router, prefix="/questions")
app.dependency_overrides[get_db] = lambda: self.db
app.dependency_overrides[get_current_user] = lambda: self.user
self.client = TestClient(app)
def tearDown(self):
self.client.close()
self.db.close()
self.engine.dispose()
def sitting(self, marks, days_ago=1, when=None):
"""One finished attempt. `marks` is {question id: was it right}."""
finished = (when or self.NOW) - timedelta(days=days_ago)
attempt = QuizAttempt(user_id=1, quiz_id=1, completed_at=finished, expired=0,
total_questions=len(marks), score=sum(1 for ok in marks.values() if ok))
self.db.add(attempt)
self.db.flush()
for question_id, was_correct in marks.items():
self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=question_id,
is_correct=was_correct, user_answer="a" if was_correct else "b"))
self.db.commit()
return attempt
def ranking(self, now=None):
return CandidateRanking(self.db, self.user, now=now or self.NOW)
def plan(self, count=None, now=None):
return prepared_session.prepare_session(self.db, self.user, count, now or self.NOW)
class SelectionTests(Bank):
def test_a_miss_last_week_is_recycled_before_a_miss_in_the_spring(self):
# Same discipline, so the topic damping and the blueprint weight are
# identical and only the dates can separate them.
self.sitting({1: False}, days_ago=120)
self.sitting({2: False}, days_ago=7)
# Three questions leaves one review slot; the recent miss should take it.
picked = self.ranking().select(3)
self.assertIn(2, picked)
self.assertNotIn(1, picked)
def test_a_topic_answered_right_once_does_not_count_as_known_forever(self):
self.sitting({qid: True for qid in range(1, 13)}, days_ago=1)
fresh = self.ranking()
# Answered, recently, and correctly: nothing to review here yet.
self.assertEqual([row for row in fresh.due if row[0] <= 12], [])
stale = self.ranking(now=self.NOW + timedelta(days=180))
self.assertEqual(len(stale.due), 12)
# And the topic's accuracy has drifted back towards unknown rather than
# staying at the hundred percent one good day bought it.
self.assertGreater(fresh.accuracy(CONGENITAL), 0.8)
self.assertLess(stale.accuracy(CONGENITAL), 0.6)
def test_review_takes_its_share_and_no_more(self):
self.sitting({qid: False for qid in range(1, 13)}, days_ago=3)
picked = self.ranking().select(10)
seen = [qid for qid in picked if qid <= 12]
self.assertEqual(len(seen), round(MAX_REVIEW_SHARE * 10))
self.assertEqual(len(picked), 10)
def test_nothing_left_unseen_still_fills_the_session(self):
self.sitting({qid: qid % 2 == 0 for qid in range(1, 31)}, days_ago=3)
picked = self.ranking().select(12)
self.assertEqual(len(picked), 12)
self.assertEqual(len(set(picked)), 12)
def test_selection_is_reproducible(self):
self.sitting({1: False, 13: True, 25: False}, days_ago=10)
self.assertEqual(self.ranking().select(15), self.ranking().select(15))
class LengthTests(Bank):
def test_the_default_holds_until_enough_sessions_are_finished(self):
self.sitting({qid: True for qid in range(1, 6)}, days_ago=20)
self.sitting({qid: True for qid in range(6, 11)}, days_ago=15)
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 2))
self.assertEqual(self.plan()["count"], prepared_session.DEFAULT_SESSION_LENGTH)
def test_length_becomes_the_learner_s_own_median_finished_session(self):
# Eight, eight, thirty: the median is eight and the mean is fifteen,
# which is a length this learner has never once sat.
self.sitting({qid: True for qid in range(1, 9)}, days_ago=30)
self.sitting({qid: True for qid in range(9, 17)}, days_ago=20)
self.sitting({qid: True for qid in range(1, 31)}, days_ago=10)
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (8, 3))
plan = self.plan()
self.assertEqual(plan["count"], 8)
self.assertIn("You usually finish 8 questions", plan["length_reason"])
def test_an_unfinished_or_expired_sitting_is_not_a_length(self):
attempt = self.sitting({qid: True for qid in range(1, 4)}, days_ago=5)
attempt.expired = 1
self.db.commit()
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 0))
def test_the_offered_length_never_exceeds_what_is_left(self):
for qid in range(6, 31):
self.db.delete(self.db.get(Question, qid))
self.db.commit()
plan = self.plan()
self.assertEqual(plan["count"], 5)
self.assertIn("everything left in your bank", plan["length_reason"])
def test_a_length_the_learner_asks_for_is_honored_and_said_to_be_theirs(self):
plan = self.plan(count=7)
self.assertEqual(plan["count"], 7)
self.assertEqual(plan["length_reason"], "7 questions, your choice.")
class PlanTests(Bank):
def rows(self, plan):
return {row["name"]: row for row in plan["topics"]}
def test_a_learner_with_no_history_is_told_so_and_gets_the_blueprint_spread(self):
plan = self.plan(count=10)
self.assertEqual(plan["basis"], "cold_start")
self.assertIn("haven't finished a session yet", plan["summary"])
self.assertIn("exam blueprint", plan["summary"])
self.assertEqual(plan["review_count"], 0)
rows = self.rows(plan)
# 60 / 10 / 30 of ten questions, which is what the board publishes.
self.assertEqual(rows["Cardiology"]["count"], 6)
self.assertEqual(rows["Rheumatology"]["count"], 1)
self.assertEqual(rows["Dermatology"]["count"], 3)
# And the reason says the blueprint, not a weakness nobody has measured.
self.assertEqual(rows["Cardiology"]["reason"], "Worth 60% of the exam")
for row in plan["topics"]:
self.assertIsNone(row["accuracy"])
def test_a_learner_with_one_topic_answered_is_told_which_and_how_badly(self):
self.sitting({qid: False for qid in range(1, 9)}, days_ago=4)
plan = self.plan(count=10)
self.assertEqual(plan["basis"], "personalized")
rows = self.rows(plan)
self.assertRegex(rows["Cardiology"]["reason"], r"^Weak area — \d+% correct so far$")
self.assertLess(rows["Cardiology"]["accuracy"], 30)
# Cardiology is both the weakest and the heaviest, so it leads — but the
# damping keeps the other two disciplines in the session.
self.assertEqual(plan["topics"][0]["name"], "Cardiology")
self.assertGreater(len(plan["topics"]), 1)
# Nothing has been measured about dermatology, and the plan says so
# rather than inventing a figure for it.
self.assertIsNone(rows["Dermatology"]["accuracy"])
self.assertEqual(rows["Dermatology"]["reason"], "Not attempted yet, worth 30% of the exam")
def test_a_topic_last_seen_months_ago_is_named_as_due_not_as_a_weakness(self):
# Everything right, but long enough ago that it is worth checking. The
# prior would drag the accuracy under the weak threshold; the evidence
# gate is what stops "you got all of these right" reading as "weak".
self.sitting({qid: True for qid in range(13, 22)}, days_ago=150)
self.sitting({qid: True for qid in range(22, 25)}, days_ago=2)
self.sitting({qid: True for qid in range(25, 28)}, days_ago=1)
rows = self.rows(self.plan(count=12))
self.assertIn("Rheumatology", rows)
self.assertRegex(rows["Rheumatology"]["reason"], r"^Due for review — last answered \d+ months ago$")
def test_the_counts_in_the_plan_add_up_to_the_session(self):
self.sitting({qid: qid % 3 == 0 for qid in range(1, 16)}, days_ago=45)
plan = self.plan(count=14)
self.assertEqual(sum(row["count"] for row in plan["topics"]), plan["count"])
self.assertEqual(plan["new_count"] + plan["review_count"], plan["count"])
for row in plan["topics"]:
self.assertEqual(row["new_count"] + row["review_count"], row["count"])
class CommittedPlanTests(Bank):
"""The plan has to describe the session it starts."""
def start(self, **body):
response = self.client.post("/questions/builder/prepared", json=body)
self.assertEqual(response.status_code, 200, response.text)
return response.json()
def questions_in(self, quiz_id):
return [row[0] for row in self.db.query(QuizQuestionLink.question_id).filter(
QuizQuestionLink.quiz_id == quiz_id).order_by(QuizQuestionLink.position).all()]
def test_the_plan_describes_the_questions_actually_put_in_the_session(self):
self.sitting({qid: qid % 4 == 0 for qid in range(1, 19)}, days_ago=6)
self.sitting({qid: True for qid in range(19, 25)}, days_ago=200)
self.sitting({25: False, 26: False}, days_ago=2)
created = self.start(count=16, title="Prepared")
plan = created["plan"]
asked = self.questions_in(created["id"])
self.assertEqual(len(asked), 16)
self.assertEqual(created["questions_count"], 16)
self.assertEqual(plan["count"], 16)
# Every claim the plan makes, recounted from the questions themselves.
by_discipline = {}
for question_id in asked:
leaf = self.of_category[question_id]
top = CARDIOLOGY if leaf == CONGENITAL else leaf
by_discipline[top] = by_discipline.get(top, 0) + 1
named = {CARDIOLOGY: "Cardiology", RHEUMATOLOGY: "Rheumatology", DERMATOLOGY: "Dermatology"}
self.assertEqual({row["name"]: row["count"] for row in plan["topics"]},
{named[key]: value for key, value in by_discipline.items()})
seen_before = {row[0] for row in self.db.query(AttemptAnswer.question_id).all()}
self.assertEqual(plan["review_count"], len([q for q in asked if q in seen_before]))
self.assertEqual(plan["new_count"], len([q for q in asked if q not in seen_before]))
for row in plan["topics"]:
top = next(key for key, name in named.items() if name == row["name"])
here = [q for q in asked
if (CARDIOLOGY if self.of_category[q] == CONGENITAL else self.of_category[q]) == top]
self.assertEqual(row["review_count"], len([q for q in here if q in seen_before]))
self.assertEqual(row["new_count"], len([q for q in here if q not in seen_before]))
# And every topic named gives its one reason.
for row in plan["topics"]:
self.assertTrue(row["reason"])
def test_the_preview_says_what_starting_it_will_give(self):
self.sitting({qid: qid % 2 == 0 for qid in range(1, 20)}, days_ago=9)
preview = self.client.get("/questions/builder/prepared", params={"count": 12})
self.assertEqual(preview.status_code, 200, preview.text)
forecast = preview.json()
# The preview is for reading; it must not hand out the stems' ids.
self.assertNotIn("question_ids", forecast)
started = self.start(count=12)["plan"]
self.assertEqual(forecast, started)
def test_the_preview_writes_nothing(self):
before = self.db.query(Quiz).count()
self.client.get("/questions/builder/prepared")
self.assertEqual(self.db.query(Quiz).count(), before)
def test_changing_the_length_replans_rather_than_padding_the_old_plan(self):
self.sitting({qid: False for qid in range(1, 10)}, days_ago=5)
short, long = self.start(count=6)["plan"], self.start(count=24)["plan"]
self.assertEqual(short["count"], 6)
self.assertEqual(long["count"], 24)
self.assertEqual(sum(row["count"] for row in long["topics"]), 24)
def test_without_a_study_objective_it_refuses_rather_than_guessing(self):
self.user.active_exam_id = None
self.db.commit()
for call in (self.client.get("/questions/builder/prepared"),
self.client.post("/questions/builder/prepared", json={})):
self.assertEqual(call.status_code, 400, call.text)
self.assertIn("studying for", call.json()["detail"])
if __name__ == "__main__":
unittest.main()

76
backend/tests/test_quiz_builder.py
View file

@ -409,17 +409,21 @@ class AdaptiveSelectionTests(unittest.TestCase):
def tearDown(self):
self.bank.tearDown()
def test_unanswered_comes_first_and_the_weakest_category_leads(self):
from app.services.quiz_builder import adaptive_select
# Answer one question in category 2 wrongly, one in category 3 rightly,
# so category 2 is the weaker of the two.
def test_unseen_material_leads_and_review_takes_only_its_share(self):
from app.services.quiz_builder import MAX_REVIEW_SHARE, adaptive_select
# Answer one question in category 1 wrongly, one in category 2 rightly.
self.bank.answer(1, correct=False)
self.bank.answer(5, correct=True)
picked = adaptive_select(self.db, self.user, 2, [], "all", None)
self.assertEqual(len(picked), 2)
# Neither already-answered question is recycled while unseen ones remain.
self.assertNotIn(1, picked)
self.assertNotIn(5, picked)
# Unseen material still leads, but "unseen first, always" is not what
# this does any more: it meant a learner with three thousand unanswered
# questions never saw a repeat, which is no spaced repetition at all.
# Review gets its share of the session and no more — one slot in two.
seen = [qid for qid in picked if qid in (1, 5)]
self.assertEqual(len(seen), round(MAX_REVIEW_SHARE * 2))
# And the slot goes to the miss, not to the one that went well.
self.assertEqual(seen, [1])
def test_it_can_see_the_whole_bank_not_the_first_page_of_it(self):
from app.services.quiz_builder import adaptive_select, bank_query
@ -512,3 +516,61 @@ class ObjectiveRequiredTests(unittest.TestCase):
# A rule that locks an empty deployment is not a rule, it is a fault:
# the first administrator has nothing to choose from yet.
self.assertEqual(self.bank.generate(is_shared=True, category_ids=[1]).status_code, 200)
class ExamScopeTests(unittest.TestCase):
"""The pool a session is drawn from is the pool the bank shows.
These disagreed, and in the worst direction: a learner whose exam had no
content linked to it browsed an empty bank and was then handed a session
built from every question in it.
"""
def setUp(self):
self.bank = BuilderTests()
self.bank.setUp()
self.client = self.bank.client
self.db = self.bank.db
self.bank.user = self.bank.owner
def tearDown(self):
self.bank.tearDown()
def link(self, exam_id, *question_ids):
from app.models.exam import Exam, QuestionExamLink
if not self.db.get(Exam, exam_id):
self.db.add(Exam(id=exam_id, name=f"Exam {exam_id}", slug=f"exam-{exam_id}"))
self.db.flush()
for qid in question_ids:
self.db.add(QuestionExamLink(exam_id=exam_id, question_id=qid))
self.db.commit()
def test_an_exam_with_no_content_yields_no_session_rather_than_all_of_it(self):
# Every question filed against exam 1, as the live bank is, so nothing
# is unclassified and nothing falls through to exam 2 that way.
self.link(1, 1, 2, 3, 4, 5, 6)
self.link(2) # exists, nothing linked to it
self.bank.owner.active_exam_id = 2
self.db.commit()
self.assertEqual(self.bank.count(), 0)
self.assertEqual(self.bank.generate(count=2).status_code, 400)
def test_the_count_and_the_session_agree_on_the_pool(self):
self.link(1, 1, 2)
self.link(2, 3, 4, 6)
self.bank.owner.active_exam_id = 1
self.db.commit()
available = self.bank.count()
self.assertEqual(available, 2)
result = self.bank.generate(count=available, expected_count=available)
self.assertEqual(result.status_code, 200, result.text)
quiz = self.db.get(Quiz, result.json()["id"])
self.assertEqual({q.id for q in quiz.questions}, {1, 2})
def test_unclassified_content_still_reaches_everyone(self):
# A question linked to no exam is unclassified, not excluded — that is
# the rule that keeps new content visible before anybody files it.
self.link(1, 1)
self.bank.owner.active_exam_id = 1
self.db.commit()
self.assertGreater(self.bank.count(), 1)

4
backend/tests/test_related_privacy.py
View file

@ -40,6 +40,10 @@ class PrivacyTests(unittest.TestCase):
self.login(self.owner)
self.quota = patch.object(teach, 'check_rate_limit').start()
self.model = patch.object(teach, '_get_teach_model', return_value=('synthetic', None)).start()
# Every question here has a figure, and the tutor now hands those to
# the model. This file promises no network, so the capability lookup
# is answered here rather than by asking the proxy.
self.can_see = patch.object(teach.vision_service, 'can_see', return_value=True).start()
self.find_similar = teach._find_similar_questions
self.similar = patch.object(teach, '_find_similar_questions', return_value=[]).start()
self.ai = AsyncMock(return_value=SimpleNamespace(

257
backend/tests/test_vision_fallback.py Normal file
View file

@ -0,0 +1,257 @@
"""What happens to an image when the model doing the job cannot see one.
No network and no models: the proxy's catalogue and the completions client are
both stubbed, because what is worth testing is not what a model says about a
picture but which model is asked, how often, and what the caller is told when
nobody can be.
"""
import os
import unittest
from unittest.mock import Mock, patch
os.environ.setdefault("DATABASE_URL", "sqlite://")
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.config import settings
from app.models.ai_model_config import AIModelConfig
from app.services import vision_service
from app.services.vision_service import Image, VisionUnavailable
def catalogue_rows(*pairs):
"""A /model/info body, in the shape the live proxy answers with."""
return {"data": [{"model_name": name, "model_info": {"supports_vision": says}}
for name, says in pairs]}
def stub_client(reply="A chest radiograph with a right lower lobe opacity."):
create = Mock(return_value=Mock(choices=[Mock(message=Mock(content=reply))]))
return Mock(**{"chat.completions.create": create}), create
class VisionFallbackTests(unittest.TestCase):
def setUp(self):
engine = create_engine("sqlite://")
AIModelConfig.__table__.create(engine)
self.db = sessionmaker(bind=engine)()
self.image = Image(data=b"\x89PNG-not-really", media_type="image/png",
caption="the figure printed with the question stem")
# Redis is not part of what is being tested, and a cache that answers
# would hide the calls these tests are counting.
patch.object(vision_service, "_redis", return_value=None).start()
self.reset_caches()
def tearDown(self):
patch.stopall()
self.db.close()
self.reset_caches()
def reset_caches(self):
vision_service._catalogue = {}
vision_service._catalogue_at = 0.0
vision_service._probes.clear()
def catalogue(self, *pairs):
vision_service._catalogue = dict(pairs)
vision_service._catalogue_at = vision_service.time.monotonic()
def tool_model(self, model_id="tool-vision"):
self.db.add(AIModelConfig(name=model_id, model_id=model_id, task="tool",
is_active=True, is_default=True))
self.db.commit()
def test_a_model_that_can_see_is_handed_the_image_itself(self):
self.catalogue(("seeing-model", True))
client, create = stub_client()
with patch.object(vision_service, "get_client", return_value=client):
parts, handoff = vision_service.image_context(
self.db, [self.image], model_id="seeing-model")
self.assertEqual([part["type"] for part in parts], ["image_url"])
self.assertTrue(parts[0]["image_url"]["url"].startswith("data:image/png;base64,"))
self.assertFalse(handoff.delegated)
self.assertIsNone(handoff.tool_model)
create.assert_not_called()
def test_a_model_that_cannot_see_gets_the_tool_model_s_description(self):
self.catalogue(("blind-model", False), ("tool-vision", True))
self.tool_model()
client, create = stub_client()
with patch.object(vision_service, "get_client", return_value=client):
parts, handoff = vision_service.image_context(
self.db, [self.image], model_id="blind-model", context="A 4-year-old…")
self.assertTrue(handoff.delegated)
self.assertEqual(handoff.tool_model, "tool-vision")
self.assertEqual(handoff.as_dict()["primary_model"], "blind-model")
# The tool model saw the picture; the primary is given words.
create.assert_called_once()
sent = create.call_args.kwargs
self.assertEqual(sent["model"], "tool-vision")
self.assertEqual([part["type"] for part in sent["messages"][0]["content"]],
["image_url", "text"])
self.assertEqual([part["type"] for part in parts], ["text"])
self.assertIn("right lower lobe opacity", parts[0]["text"])
self.assertIn("tool-vision", parts[0]["text"])
def test_no_tool_model_is_an_error_naming_the_fix(self):
self.catalogue(("blind-model", False))
with self.assertRaises(VisionUnavailable) as raised:
vision_service.image_context(self.db, [self.image], model_id="blind-model")
message = str(raised.exception)
self.assertIn("cannot read images", message)
self.assertIn("Settings → AI models", message)
def test_a_tool_model_that_cannot_see_either_is_refused_before_it_is_called(self):
self.catalogue(("blind-model", False), ("also-blind", False))
self.tool_model("also-blind")
client, create = stub_client()
with patch.object(vision_service, "get_client", return_value=client):
with self.assertRaises(VisionUnavailable) as raised:
vision_service.image_context(self.db, [self.image], model_id="blind-model")
self.assertIn("also-blind", str(raised.exception))
create.assert_not_called()
def test_a_failing_tool_model_is_an_error_rather_than_a_text_only_answer(self):
self.catalogue(("blind-model", False), ("tool-vision", True))
self.tool_model()
client = Mock()
client.chat.completions.create.side_effect = RuntimeError("upstream 500")
with patch.object(vision_service, "get_client", return_value=client):
with self.assertRaises(VisionUnavailable) as raised:
vision_service.image_context(self.db, [self.image], model_id="blind-model")
self.assertIn("upstream 500", str(raised.exception))
class CapabilityLookupTests(unittest.TestCase):
def setUp(self):
patch.object(vision_service, "_redis", return_value=None).start()
self.base = settings.LITELLM_API_BASE
settings.LITELLM_API_BASE = "http://proxy.invalid"
vision_service._catalogue = {}
vision_service._catalogue_at = 0.0
vision_service._probes.clear()
def tearDown(self):
patch.stopall()
settings.LITELLM_API_BASE = self.base
vision_service._catalogue = {}
vision_service._catalogue_at = 0.0
vision_service._probes.clear()
def test_the_catalogue_is_read_once_for_many_questions(self):
response = Mock(json=Mock(return_value=catalogue_rows(
("sees", True), ("blind", False), ("unlisted-in-the-cost-map", None))))
with patch("httpx.get", return_value=response) as fetch:
for _ in range(5):
self.assertTrue(vision_service.can_see("sees"))
self.assertFalse(vision_service.can_see("blind"))
fetch.assert_called_once()
def test_one_deployment_that_cannot_see_decides_the_alias(self):
# `best-chat` fans out to six deployments and a request lands on any of
# them, so a single false is the answer for the name.
response = Mock(json=Mock(return_value=catalogue_rows(
("best-chat", True), ("best-chat", None), ("best-chat", False))))
with patch("httpx.get", return_value=response):
self.assertFalse(vision_service.can_see("best-chat"))
def test_a_model_the_catalogue_says_nothing_about_is_probed_once(self):
response = Mock(json=Mock(return_value=catalogue_rows(("quiet", None))))
client, create = stub_client("ok")
with patch("httpx.get", return_value=response), \
patch.object(vision_service, "get_client", return_value=client):
self.assertTrue(vision_service.can_see("quiet"))
self.assertTrue(vision_service.can_see("quiet"))
create.assert_called_once()
self.assertEqual(create.call_args.kwargs["max_tokens"], 1)
def test_a_refused_probe_is_remembered_and_a_broken_one_is_not(self):
class Refused(Exception):
status_code = 400
response = Mock(json=Mock(return_value=catalogue_rows(("quiet", None))))
client = Mock()
client.chat.completions.create.side_effect = Refused("image input not supported")
with patch("httpx.get", return_value=response), \
patch.object(vision_service, "get_client", return_value=client):
self.assertFalse(vision_service.can_see("quiet"))
self.assertFalse(vision_service.can_see("quiet"))
client.chat.completions.create.assert_called_once()
# A timeout says nothing about the model, so nothing is kept: asking
# again asks the model again.
vision_service._probes.clear()
client.chat.completions.create.reset_mock()
client.chat.completions.create.side_effect = TimeoutError("proxy is down")
with patch("httpx.get", return_value=response), \
patch.object(vision_service, "get_client", return_value=client):
self.assertFalse(vision_service.can_see("quiet"))
self.assertFalse(vision_service.can_see("quiet"))
self.assertEqual(client.chat.completions.create.call_count, 2)
class TutorFigureTests(unittest.TestCase):
"""The tutor is the first job in the app where a picture reaches a model.
Borrowed fixtures: every question in them is printed with a stem figure and
an explanation figure, which is exactly the case being tested.
"""
def setUp(self):
import test_related_privacy as fixtures
from app.routers import teach
self.teach = teach
self.privacy = fixtures.PrivacyTests()
self.privacy.setUp()
self.privacy.can_see.return_value = False
patch.object(vision_service, "_redis", return_value=None).start()
# The tool model is checked against the catalogue before it is trusted,
# and this suite reaches no proxy: an empty catalogue is "nothing known
# against it", which is the case being tested.
patch.object(vision_service, "catalogue", return_value={}).start()
vision_service._catalogue, vision_service._catalogue_at = {}, 0.0
def tearDown(self):
self.privacy.tearDown()
vision_service._catalogue, vision_service._catalogue_at = {}, 0.0
def chat(self):
return self.privacy.client.post("/teach/chat", json={
"question_id": 1, "messages": [{"role": "user", "content": "Explain"}]})
def sent_to_the_tutor(self):
return self.privacy.ai.call_args.kwargs["messages"]
def test_a_blind_tutor_is_given_the_tool_model_s_description(self):
self.privacy.db.add(AIModelConfig(name="tool-vision", model_id="tool-vision",
task="tool", is_active=True, is_default=True))
self.privacy.db.commit()
client, create = stub_client("A frontal chest radiograph.")
with patch.object(vision_service, "get_client", return_value=client):
response = self.chat()
self.assertEqual(response.status_code, 200, response.text)
self.assertTrue(response.json()["vision"]["delegated"])
self.assertEqual(response.json()["vision"]["tool_model"], "tool-vision")
# Both figures described, and the descriptions are what the tutor read.
self.assertEqual(create.call_count, 2)
figures = self.sent_to_the_tutor()[1]["content"]
self.assertEqual([part["type"] for part in figures], ["text", "text"])
self.assertIn("A frontal chest radiograph.", figures[0]["text"])
def test_with_nothing_able_to_read_it_the_tutor_is_told_so(self):
response = self.chat()
self.assertEqual(response.status_code, 200, response.text)
self.assertNotIn("vision", response.json())
told = self.sent_to_the_tutor()[1]["content"][0]["text"]
self.assertIn("neither describe nor infer", told)
if __name__ == "__main__":
unittest.main()

171
docs/adaptive-sessions.md
View file

@ -1,48 +1,142 @@
# Adaptive sessions — what "prioritised by impact" means here
**Code:** `backend/app/services/quiz_builder.py`, function `adaptive_select`
(the `algorithm: "adaptive"` branch of `generate_test`).
**Reached from:** the Adaptive toggle on the custom-session builder
(`frontend/src/pages/CustomQuizPage.jsx`), and the *Next step: adaptive
session* card on Analysis (`frontend/src/pages/AnalysisPage.jsx`).
**Tests:** `backend/tests/test_quiz_builder.py`,
`test_adaptive_selection_prefers_unanswered_then_recycles_weakest`.
**Code:** `backend/app/services/quiz_builder.py`, class `CandidateRanking`
(reached by `adaptive_select`, the `algorithm: "adaptive"` branch of
`generate_test`), and `backend/app/services/prepared_session.py`.
**Reached from:** the *Ready for you* card on the dashboard
(`frontend/src/components/PreparedSession.jsx`), the Adaptive toggle on the
custom-session builder (`frontend/src/pages/CustomQuizPage.jsx`), and the
*Next step: adaptive session* card on Analysis
(`frontend/src/pages/AnalysisPage.jsx`).
**Tests:** `backend/tests/test_prepared_session.py`,
`backend/tests/test_quiz_builder.py` (`AdaptiveSelectionTests`).
## What it does today
There is no language model anywhere in this. It is arithmetic over the
learner's own answers, and it has to stay that way: the whole product here is
that the session can be explained before it is sat, and a model that cannot
show its working could not do that.
Three rules, then one multiplier.
## What decides which questions
**1. Unanswered first.** A question you have never seen teaches more than one
you have. Everything you have not answered is taken before anything you have.
Four rules, then one multiplier.
**2. Among the unanswered, weakest topic first.** Each candidate is ordered by
your accuracy in its primary category. A category you have never answered in
scores 0.5 — treated as neither known nor unknown, so it sorts between your
strong and weak areas rather than jumping the queue.
**1. Unseen material is most of the session.** A question never met teaches
more than one already answered, so it takes every slot review is not holding.
**3. When the unanswered run out, recycle — wrong ones first, oldest first.**
Each recycled candidate is scored `(1 − p) × damping`, where `p` is 0.85 if you
got it right last time and 0.25 if you got it wrong. So a question you missed
is worth roughly five times one you got right. After each pick, that category's
damping halves, which stops a session of twenty becoming twenty cardiology
questions because cardiology happens to be your worst subject.
**2. Review holds up to 40% of the session, and only what is due.**
`MAX_REVIEW_SHARE`. Due means recall has decayed below `DUE_RECALL` — which is
everything ever answered wrongly, and everything answered correctly more than
about three and a half weeks ago. This used to be "recycle when the unanswered
run out", which on a bank of nearly three thousand questions meant a learner
never saw a repeat: no spaced repetition at all. The cap is the other half of
it — somebody handed twenty questions they have already answered does not come
back.
**4. All of it is scaled by the topic's share of the real paper.** Weakness
**3. Within either pool, highest value first.** For unseen material that is the
topic's **impact**, `(1 − accuracy) × blueprint weight`. For review it is
`(1 − recall) × blueprint weight`.
**4. Each pick halves its topic's priority.** `CATEGORY_DAMPING`. Without it a
session of twenty becomes twenty questions from the single worst subject — and
a learner with no history at all, whose topics are all equally unknown, gets
handed the heaviest domain entire rather than a spread.
**And all of it is scaled by the topic's share of the real paper.** Weakness
alone said that being weak at something worth 5% of the exam and something
worth 1% were the same problem. They are not. Every score above is multiplied
by the weight the examining board publishes for that topic's domain, which
`exam_blueprints.weight` holds and `blueprint_weights()` loads.
A topic the blueprint does not cover takes the **median** of the published
weights. A zero would make unmapped material unreachable; the highest would
make it the priority. Neither is a claim the blueprint supports.
worth 1% were the same problem. They are not. `exam_blueprints.weight` holds
what the examining board publishes and `blueprint_weights()` loads it. A topic
the blueprint does not cover takes the **median** of the published weights: a
zero would make unmapped material unreachable, the highest would make it the
priority, and neither is a claim the blueprint supports.
With no study objective set, or an objective carrying no weights, the
multiplier is absent and selection is about weakness alone — which is what it
was before this existed.
multiplier is absent and selection is about weakness alone.
Only completed, non-expired, non-course attempts count, and only your most
recent answer to each question.
Only completed, non-expired, non-course attempts count. The pool is scoped to
the learner's active exam, the same scope the bank and search use.
## How time enters it
`EVIDENCE_HALF_LIFE_DAYS = 30`. Every answer's weight is
`0.5 ** (age_days / 30)`.
**Why exponential.** A fixed window was the obvious thing and is wrong in a way
that shows: it makes an answer twenty-nine days old count in full and one
thirty-one days old count for nothing, so a topic crosses a cliff overnight and
the ranking lurches without the learner having done anything. Exponential decay
is also memoryless — an answer's weight depends only on its own age, not on
what has been answered since — which is what keeps two consecutive sessions
consistent with each other. A power law fits very long retention slightly
better, but it needs an arbitrary offset to avoid a singularity at age zero and
a second parameter nothing here could justify. One named half-life describes
the whole curve.
**Why thirty days.** About the turn of a revision cycle. A ninety-day-old
answer keeps an eighth of the weight of a fresh one, so what was missed last
week clearly outranks what was missed in spring, while a topic revised last
month is not written off as forgotten.
Two things decay:
- **A question's recall.** `NEUTRAL_RECALL + (settled − NEUTRAL_RECALL) × decay`,
where `settled` is 0.85 after a correct answer and 0.25 after a wrong one.
Both decay *towards a coin flip*, not towards zero: forgetting a right answer
does not turn it into a wrong one, and time does not turn a wrong answer into
a right one. Both end up saying nothing, which is exactly when the question
is worth asking again.
- **A topic's accuracy.** Each answer counts for `decay(age)` of an answer, and
`PRIOR_ANSWERS = 2` answers' worth of "no idea" is mixed in. Without that
prior a single correct answer made a topic 100% known and it never came back,
which is the one thing a ranking claiming to decay must not do.
## The prepared session
`prepared_session.prepare_session` — one action that produces a session tuned
to this learner and a plain statement of why, before anything is written.
- `GET /questions/builder/prepared?count=` returns the plan. Nothing is
written, and the question ids are withheld: the plan is for reading, and
handing out stem identifiers is how a preview becomes a way to enumerate the
bank.
- `POST /questions/builder/prepared` recomputes the plan, builds the test from
**that plan's own question ids**, and returns the plan alongside the created
quiz. So the account handed back describes the session by construction, not
because two calls happened to agree. A learner who changed the length gets
the plan for the length they chose.
The preview is a forecast, and it is accurate because the ranking is
deterministic: candidates are scanned in question-id order and every tie
resolves to the lowest id. The cold-start blueprint draw, which is random, is
seeded on `(user, length, date)` for the same reason.
**What the plan says.** A one-line summary, a length and why that length, and
one row per discipline — count, how much of it is new versus review, the
learner's accuracy there, the topic's share of the paper, and one line of
reasoning. Rows are grouped by top-level category because that is the
vocabulary Analysis already reports weaknesses in; leaf topics would be twenty
rows of one question each. **Every count in the plan is tallied from the
questions actually chosen**, never predicted from the ranking — see
`test_the_plan_describes_the_questions_actually_put_in_the_session`.
The reason is the strongest true fact, in this order: the blueprint share (cold
start), a weak area, due for review, not attempted yet, the blueprint share. A
weak topic is usually also due, and the accuracy is the more useful of the two.
"Weak area" needs `WEAK_EVIDENCE_ANSWERS = 3` decayed answers behind it — below
that the prior is most of the number, and a topic last answered *correctly* in
the spring would otherwise be reported as a weakness on no evidence.
**Length** is the learner's own median finished session, clamped to 5–60,
counted as answers recorded rather than questions offered. The median rather
than the mean: one abandoned session and one marathon pull a mean somewhere
neither of them is. Until `SESSIONS_BEFORE_LENGTH_IS_PERSONAL = 3` sessions are
finished it is 20, and the plan says so.
**No history degrades honestly.** A learner who has finished nothing gets a
paper drawn to the exam blueprint — `exam_blueprint.sample`, the same code
behind `algorithm: "blueprint"` — and is told in as many words that this is a
spread across the exam rather than a personalisation. No exam or no blueprint
falls back to the damped impact spread over the whole bank, and says that
instead.
## What it is still *not*
@ -58,9 +152,16 @@ Dealing a paper shaped like the real exam is a different algorithm —
In the order they are worth doing:
- **Adapt the difficulty, not only the count and the mix.** `Question.difficulty`
is on every row and nothing in the ranking reads it. A learner at 40% on a
topic and one at 80% get the same questions from it.
- **Use every category a question is filed under, not only its primary.**
Step 2 reads `question_category_id` alone, so a question's extra links do not
influence which topic it counts as.
The ranking reads `question_category_id` alone, so a question's extra links
do not influence which topic it counts as. The plan's grouping has the same
limit.
- **Let time answer.** A question answered correctly in ten seconds is not the
same as one answered correctly in four minutes, and `seconds_spent` is
recorded on every answer already.
- **Let the half-life be per learner.** Thirty days is a defensible constant and
a worse fit than a rate fitted to somebody's own forgetting, which their
answer history could estimate once there is enough of it.