"""Narrative interpretation of module outputs. Takes the raw, numeric outputs produced by the evidence modules (today: Layla's historical-comparator / PubTime module, protocol completeness, publication outlook — tomorrow: more) and turns them into a short, plain-language read for a trial PI. An LLM does the interpreting so the prose tracks whatever the modules actually returned instead of a fixed template; a deterministic writer is the fallback when no model is configured or a call fails. The LLM is given *only* the numbers the modules produced and is told to ground every claim in them — it interprets, it does not invent. """ from __future__ import annotations import json from pathlib import Path from typing import Any from llm import LlmError, build_client SYSTEM_PROMPT = ( "You are a clinical-trial methodologist helping an investigator read a " "pre-submission design check. You are given a JSON brief of numbers produced " "by analysis modules — the main one compares this planned trial against " "matched historical trials in the same disease area.\n\n" "Write a short interpretation for the investigator. Rules:\n" "- Ground every statement in the numbers in the brief. Never invent a number, " "rate, or fact that is not present.\n" "- The rates come from comparable past trials, not a validated predictive " "model. Convey this once, in natural language (e.g. 'trials like yours'), and " "do not repeat a disclaimer.\n" "- Lead with what matters: where the plan diverges from comparators and what " "that implies for publication and results reporting.\n" "- Be concrete and concise. Short declarative sentences, active voice. No " "filler, no 'it is important to note', no 'please consult', no restating the " "inputs back, no hedging padding.\n\n" "You also rewrite the flagged issues into clear, actionable recommendations. " "Use only the items in 'design_flags' and 'protocol_completeness.weakest_sections' " "— do not invent new issues. Each action is one short imperative sentence telling " "the investigator what to check or change and, briefly, why. Merge duplicates. " "Tag each action 'review' for a design/methodology issue or 'completeness' for a " "missing registry field.\n\n" "Return a JSON object with exactly three keys:\n" ' "takeaway": one sentence the investigator reads first (<= 30 words).\n' ' "summary": an array of 2 to 4 sentences expanding on the takeaway.\n' ' "actions": an array of objects, each {"severity": "review"|"completeness", ' '"text": "..."}, ordered most important first. Empty array if nothing is flagged.\n' ) def build_evidence_brief(profile: dict[str, Any], capabilities: dict[str, Any]) -> dict[str, Any]: comparator = capabilities.get("historical_comparator") or {} summary = comparator.get("summary", {}) comparison = comparator.get("comparison", {}) outlook = capabilities.get("publication_outlook") or {} completeness = capabilities.get("protocol_completeness") or {} return { "planned_trial": { "domain": profile.get("domain_label"), "phase": profile.get("phase"), "primary_purpose": profile.get("primary_purpose"), "allocation": profile.get("allocation"), "masking": profile.get("masking"), "enrollment": profile.get("enrollment"), "arms": profile.get("number_of_arms"), "facilities": profile.get("number_of_facilities"), "primary_outcomes": profile.get("number_of_primary_outcomes"), "secondary_outcomes": profile.get("number_of_secondary_outcomes"), "has_dmc": profile.get("has_dmc"), }, "comparator_cohort": { "matched_trials": comparator.get("used_rows"), "match_strategy": comparator.get("match_strategy"), "publication_rate": _as_percent(summary.get("publication_rate")), "results_reported_rate": _as_percent(summary.get("results_reported_rate")), "median_time_to_publication_days": summary.get("median_time_to_publication_days"), "median_enrollment": summary.get("median_enrollment"), "median_facilities": summary.get("median_facilities"), "median_arms": summary.get("median_arms"), "median_duration_months": summary.get("median_duration_months"), "median_primary_outcomes": summary.get("median_primary_outcomes"), "median_secondary_outcomes": summary.get("median_secondary_outcomes"), }, "publication_outlook": { "publication_likelihood": _as_percent(outlook.get("publication_likelihood")), "results_reporting_likelihood": _as_percent(outlook.get("results_reporting_likelihood")), "source": outlook.get("provenance_label"), "model_type": outlook.get("model_type"), }, "protocol_completeness": { "completion_percent": _as_percent(completeness.get("completion_ratio")), "weakest_sections": [ {"section": section.get("section"), "missing": section.get("missing")} for section in completeness.get("weakest_sections", []) ], }, "design_flags": comparison.get("flags", []), "review_priority": comparison.get("review_priority"), } def summarize_with_llm(profile: dict[str, Any], capabilities: dict[str, Any], project_root: Path) -> dict[str, Any]: """Interpret module outputs with the LLM; fall back to the template on any failure.""" client = build_client(Path(project_root) / ".env") if client is None: result = template_summary(profile, capabilities) result["warning"] = "LLM not configured (no OpenAI API key); used deterministic summary." return result brief = build_evidence_brief(profile, capabilities) user_prompt = "Evidence brief:\n" + json.dumps(brief, indent=2, default=str) try: raw = client.complete_json(system_prompt=SYSTEM_PROMPT, user_prompt=user_prompt) return _coerce_llm_output(raw) except (LlmError, ValueError, KeyError) as exc: result = template_summary(profile, capabilities) result["warning"] = f"LLM interpretation failed ({exc}); used deterministic summary." return result def _coerce_llm_output(raw: dict[str, Any]) -> dict[str, Any]: takeaway = str(raw.get("takeaway", "")).strip() summary_field = raw.get("summary", []) if isinstance(summary_field, str): summary = [summary_field.strip()] else: summary = [str(item).strip() for item in summary_field if str(item).strip()] if not takeaway and summary: takeaway = summary[0] if not takeaway: raise LlmError("Model returned an empty summary.") return { "takeaway": takeaway, "summary": summary or [takeaway], "actions": _coerce_actions(raw.get("actions")), "source": "llm", } def _coerce_actions(raw: Any) -> list[dict[str, str]]: if not isinstance(raw, list): return [] actions: list[dict[str, str]] = [] for item in raw: if isinstance(item, dict): text = str(item.get("text", "")).strip() severity = str(item.get("severity", "review")).strip().lower() else: text, severity = str(item).strip(), "review" if not text: continue if severity not in {"review", "completeness"}: severity = "review" actions.append({"severity": severity, "text": text}) return actions def template_summary(profile: dict[str, Any], capabilities: dict[str, Any]) -> dict[str, Any]: """Deterministic, concise fallback — grounded, no boilerplate disclaimers.""" comparator = capabilities.get("historical_comparator") or {} summary = comparator.get("summary", {}) outlook = capabilities.get("publication_outlook") or {} comparison = comparator.get("comparison", {}) flags = comparison.get("flags", []) pub = _pct(outlook.get("publication_likelihood")) results = _pct(outlook.get("results_reporting_likelihood")) matched = comparator.get("used_rows") domain = profile.get("domain_label") phase = profile.get("phase") takeaway = ( f"Among {matched} comparable {domain} trials, {pub} reached publication and " f"{results} posted results; " + (f"{len(flags)} design issue(s) warrant review." if flags else "no design issues were flagged.") ) lines = [ f"This {domain} {phase} plan was matched to {matched} historical trials with similar design.", f"In that group, publication ran at {pub} and results reporting at {results}.", ] if flags: lines.append("Key divergence: " + flags[0]) return {"takeaway": takeaway, "summary": lines, "source": "template"} def _pct(value: float | None) -> str: if value is None: return "an unknown share of" return f"{round(value * 100)}%" def _as_percent(value: float | None) -> str | None: if value is None: return None return f"{round(value * 100)}%"