"""Write a topic article from what the clinical library returns. The division of labour is deliberate and matches AI Mode: **retrieval supplies the facts and the provenance; the model supplies the prose.** References are built from the metadata of the passages that were actually retrieved, never from the model, so a reference cannot be invented — the same property that makes an AI Mode citation trustworthy. What is generated is an original piece of writing grounded in the library, not an extract from it. A textbook's sentences belong to its publisher; its facts do not, and an article that reproduced the sentences would be redistribution whatever produced the copy. The prompt says this plainly, and the length limits make a long verbatim passage impossible to hide. Three views come out of one call rather than three, because they are three readings of one topic and generating them separately lets them contradict each other on the numbers. """ import json import logging import re import uuid from sqlalchemy.orm import Session from app.models.article import Article from app.services import article_service, clinical_library from app.services.ai_service import _call_model, get_model_for_task logger = logging.getLogger(__name__) PASSAGES = 14 # Enough retrieved prose to write from. Below this the model is being asked to # write a medical article out of fragments, and it will either refuse or invent. MIN_SOURCE_CHARS = 3000 # Generous for a long article, short enough that a stalled call is noticed in # minutes rather than discovered hours later with nothing written since. WRITE_TIMEOUT = 150 # Three views of one topic, the long one about 550 words. Four thousand tokens is # comfortable for that and keeps each request small enough to be affordable. WRITE_MAX_TOKENS = 4000 # Long enough to be worth reading, short enough that nobody skims past the point. LONG_WORDS = 550 SHELF = "Pediatrics" def slugify(name: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", (name or "").lower()).strip("-") return slug[:120] or "topic" def _prompt(topic: str, passages: list[dict]) -> str: sources = "\n\n".join( f"[{i + 1}] {p['source'].get('title', 'source')}" f"{', p. ' + str(p['source']['page']) if p['source'].get('page') else ''}\n{p['text']}" for i, p in enumerate(passages) ) return f"""You are writing a study article about **{topic}** for a paediatric exam-revision platform. Write it from the reference passages below. They are extracts from standard textbooks and guidelines. Use them for the facts, the numbers and the structure — then write the article in your own words. Do not reproduce sentences from the passages: this is an article about the topic, not an extract from a book. If the passages do not cover something, leave it out rather than filling the gap from memory. A gap a reader can see is safer than a confident sentence nobody can check. Produce three views of the same topic: 1. "long" — the full article, about {LONG_WORDS} words, in 4 to 7 sections. Typical headings: Definition, Epidemiology, Etiology, Pathophysiology, Clinical features, Diagnostics, Differential diagnosis, Treatment, Complications, Prevention. Use only the ones the passages support. 2. "short" — one section titled "In short", 6 to 10 tight bullets a learner could revise from the night before an exam. Every bullet must carry a fact: no bullet that only says a topic is important. 3. "clinical" — one or two sections covering management at the bedside: what to do, in what order, with drug doses and routes where the passages give them. Include units and per-kilogram dosing exactly as stated. If a dose is not in the passages, do not state one. The summary is two sentences, 30 to 60 words, that make claims — never a list of the topics the article covers. The first says what the condition is, who gets it and when; the second says what changes management, or how it ends. "Peanut allergy prevention and management: early introduction guidelines, risk stratification, and anaphylaxis treatment" is a table of contents with the colons filed off: every phrase sounds informative and together they assert nothing. Write the two sentences a reader finishes knowing something. Spell everything in American English — pediatric, hemorrhage, edema, anemia, diarrhea, esophageal, epinephrine — throughout the summary and every section. The rest of the corpus is American and so is the board this prepares people for. Return ONLY valid JSON, no markdown fence: {{ "summary": "two sentences making claims, no heading", "long": [{{"title": "Definition", "content": "markdown"}}], "short": [{{"title": "In short", "content": "- bullet\\n- bullet"}}], "clinical": [{{"title": "Management", "content": "markdown"}}] }} Markdown may use lists, bold and tables. Do not write citation markers, footnote numbers or URLs anywhere: the reference list is attached separately. REFERENCE PASSAGES {sources}""" def _sections(blocks: list[dict], variant: str) -> list[dict]: """Turn the model's blocks into stable sections of one view.""" out = [] for block in blocks or []: # The model occasionally returns a bare string where a block was asked # for. Skipping it costs one section; letting it through ends the run. if not isinstance(block, dict): continue title = str(block.get("title") or "").strip() content = str(block.get("content") or "").strip() if not title or not content: continue out.append({ # A fresh id per section, because links are made against these and a # regenerated article must not silently inherit another's links. "id": uuid.uuid4().hex, "slug": slugify(title)[:60] or f"section-{len(out) + 1}", "title": title[:300], "content": content, "parent_id": None, "variant": variant, }) return out def _unique_slugs(sections: list[dict]) -> list[dict]: """Slugs are unique within an article, and three views repeat titles.""" seen: set[str] = set() for section in sections: base = section["slug"] candidate, n = base, 2 while candidate in seen: candidate = f"{base}-{n}" n += 1 section["slug"] = candidate seen.add(candidate) return sections def write_article(db: Session, topic: str, category_id: int | None = None, user_id: int | None = None) -> dict: """Retrieve, write, and store one article as a draft. Returns a small report rather than the article, because the caller is a background run over hundreds of topics and wants to know what happened. """ passages = clinical_library.search( f"{topic} in children: definition, causes, clinical features, diagnosis and management", limit=PASSAGES, folder_contains=SHELF) source_chars = sum(len(p["text"]) for p in passages) if len(passages) < 3 or source_chars < MIN_SOURCE_CHARS: # Too little to ground an article. Writing one anyway would produce # exactly the confident unverifiable prose this is designed to avoid, # and a topic named after a shelf rather than a condition lands here. return {"topic": topic, "status": "skipped", "reason": f"only {source_chars} chars from {len(passages)} passages", "passages": len(passages)} model_id, api_key = get_model_for_task(db, "extraction") # One topic must not be able to stall a run of five hundred. raw = _call_model(_prompt(topic, passages), model_id, api_key, timeout=WRITE_TIMEOUT, max_tokens=WRITE_MAX_TOKENS) text = raw.strip() if text.startswith("```"): text = re.sub(r"^```[a-z]*\n?|```$", "", text).strip() try: data = json.loads(text) except ValueError: logger.warning("Article writer returned unparseable JSON for %s", topic) return {"topic": topic, "status": "failed", "reason": "model did not return JSON"} if not isinstance(data, dict): return {"topic": topic, "status": "failed", "reason": "model returned the wrong shape"} sections = _unique_slugs([ *_sections(data.get("long"), "long"), *_sections(data.get("short"), "short"), *_sections(data.get("clinical"), "clinical"), ]) if not sections: return {"topic": topic, "status": "failed", "reason": "no usable sections"} slug = slugify(topic) existing = db.query(Article).filter(Article.slug == slug).first() if existing: return {"topic": topic, "status": "exists", "article_id": existing.id} article = Article( slug=slug, title=topic[:300], summary=(data.get("summary") or "").strip()[:2000] or None, content=None, sections=sections, category_id=category_id, user_id=None, # Never published by generation. A person decides that. status="draft", # References come from the retrieved metadata, not from the model, so a # source cannot be invented. references_json=clinical_library.references_from(passages), generated_by=f"clinical-library:{model_id}", ) from datetime import datetime article.generated_at = datetime.utcnow() db.add(article) db.flush() article_service.record_slug(db, article) db.commit() db.refresh(article) # The section index is what makes the body searchable at all. This writer # produced 323 articles without it, so the library's whole depth was # invisible to search while every one of them looked fine on the page. article_service.reindex(db, article) return {"topic": topic, "status": "written", "article_id": article.id, "sections": len(sections), "references": len(article.references_json or []), "passages": len(passages)}