Spaces:
Running
Running
| """ | |
| app.py — NeuroLens: End-to-End Cognitive Health Screening & Coaching Pipeline | |
| A three-stage Streamlit app: | |
| 1. Conversational cognitive assessment (typed language samples, enforced timers) | |
| 2. Linguistic biomarker extraction + dashboard (Stage 2) | |
| 3. Grounded, citation-backed prevention coaching via RAG (Stage 3) | |
| IMPORTANT: This is a research/portfolio demonstration prototype inspired by | |
| published cognitive assessment paradigms. It is NOT a validated clinical or | |
| diagnostic instrument. See the "About this project" section in the sidebar | |
| and README.md for full limitations and citations. | |
| """ | |
| import os | |
| import time | |
| import streamlit as st | |
| from streamlit_autorefresh import st_autorefresh | |
| from biomarkers import build_profile | |
| from rag_engine import LiteratureRetriever, build_retrieved_context, generate_coaching, check_faithfulness | |
| st.set_page_config(page_title="NeuroLens", page_icon="🧠", layout="wide") | |
| RECALL_WORDS = ["apple", "river", "hammer", "candle", "bicycle", "lantern"] | |
| FLUENCY_TIME_LIMIT = 60 # seconds, enforced — matches the standard 60s fluency-task paradigm | |
| DISTRACTOR_TIME_LIMIT = 30 # seconds, enforced counting-backward distractor | |
| TASK_FLOW = ["semantic", "phonemic", "narrative", "distractor", "recall"] | |
| # --------------------------------------------------------------------------- | |
| # Timer helper — enforces a hard time limit on a task using periodic reruns. | |
| # The text typed so far is preserved (tied to session_state by widget key); | |
| # once time is up the input is disabled and the user must continue. | |
| # --------------------------------------------------------------------------- | |
| def enforced_timer(task_key: str, limit_seconds: int): | |
| start_key = f"{task_key}_start_time" | |
| if start_key not in st.session_state: | |
| st.session_state[start_key] = time.time() | |
| elapsed = time.time() - st.session_state[start_key] | |
| remaining = max(0, limit_seconds - elapsed) | |
| timed_out = remaining <= 0 | |
| if not timed_out: | |
| # Re-run this script once per second until the limit is hit, so the | |
| # countdown visibly ticks down and the widget gets disabled the | |
| # moment time expires — without needing any JS component. | |
| st_autorefresh(interval=1000, limit=limit_seconds + 2, key=f"refresh_{task_key}") | |
| mins, secs = divmod(int(remaining), 60) | |
| bar_col, time_col = st.columns([4, 1]) | |
| with bar_col: | |
| st.progress(min(1.0, elapsed / limit_seconds) if limit_seconds else 1.0) | |
| with time_col: | |
| st.markdown(f"**⏱️ 0:{secs:02d}**" if mins == 0 else f"**⏱️ {mins}:{secs:02d}**") | |
| if timed_out: | |
| st.warning("⏰ Time's up — your response has been locked in.", icon="⏰") | |
| return timed_out | |
| def reset_task_timer(task_key: str): | |
| st.session_state.pop(f"{task_key}_start_time", None) | |
| # --------------------------------------------------------------------------- | |
| # Sidebar — framing & disclosure (lead with the limitation, not the capability) | |
| # --------------------------------------------------------------------------- | |
| with st.sidebar: | |
| st.markdown("## 🧠 About NeuroLens") | |
| st.warning( | |
| "**This is a research / portfolio demonstration prototype**, not a " | |
| "validated diagnostic or clinical tool. It is inspired by published " | |
| "cognitive-linguistic assessment paradigms but has not been clinically " | |
| "validated. Any real-world deployment would require IRB-reviewed " | |
| "studies and licensed clinical oversight.", | |
| icon="⚠️", | |
| ) | |
| st.markdown( | |
| "**What it actually demonstrates:**\n" | |
| "- NLP feature extraction grounded in real cognitive-aging research\n" | |
| "- Enforced, timed task administration (matching the standard 60s " | |
| "fluency-task paradigm used in the literature)\n" | |
| "- A retrieval-augmented generation (RAG) pipeline with a faithfulness check\n" | |
| "- End-to-end product thinking applied to a health-adjacent domain" | |
| ) | |
| st.markdown("---") | |
| st.markdown( | |
| "Built as a proof-of-concept exploring the kind of AI engineering + " | |
| "domain literacy relevant to evidence-informed brain-health products. " | |
| "See `README.md` for full citations and the linguistic-biomarker literature." | |
| ) | |
| api_key_present = bool(os.environ.get("GROQ_API_KEY")) | |
| if not api_key_present: | |
| st.info("Set the `GROQ_API_KEY` environment variable to enable Stage 3 coaching generation.", icon="🔑") | |
| # --------------------------------------------------------------------------- | |
| # Session state | |
| # --------------------------------------------------------------------------- | |
| if "stage" not in st.session_state: | |
| st.session_state.stage = 1 | |
| if "task_idx" not in st.session_state: | |
| st.session_state.task_idx = 0 | |
| if "responses" not in st.session_state: | |
| st.session_state.responses = {} | |
| if "profile" not in st.session_state: | |
| st.session_state.profile = None | |
| if "coaching" not in st.session_state: | |
| st.session_state.coaching = None | |
| st.title("🧠 NeuroLens") | |
| st.caption("Research-prototype cognitive-linguistic screening → biomarker dashboard → grounded prevention coaching") | |
| stage_labels = ["1. Assessment", "2. Biomarker Dashboard", "3. Prevention Coaching"] | |
| st.progress((st.session_state.stage - 1) / 2, text=stage_labels[st.session_state.stage - 1]) | |
| def goto_next_task(): | |
| st.session_state.task_idx += 1 | |
| if st.session_state.task_idx >= len(TASK_FLOW): | |
| st.session_state.profile = build_profile(st.session_state.responses) | |
| st.session_state.stage = 2 | |
| st.rerun() | |
| # --------------------------------------------------------------------------- | |
| # STAGE 1 — Conversational Cognitive Assessment (each task individually timed) | |
| # --------------------------------------------------------------------------- | |
| if st.session_state.stage == 1: | |
| st.header("Stage 1 · Short Language Assessment") | |
| st.caption( | |
| "Simplified, non-clinical versions of standard cognitive-linguistics research tasks. " | |
| "The fluency tasks are timed (60s) to match the standard paradigm; just type freely — " | |
| "there are no right or wrong answers." | |
| ) | |
| current_task = TASK_FLOW[st.session_state.task_idx] | |
| st.subheader(f"Task {st.session_state.task_idx + 1} of {len(TASK_FLOW)}") | |
| # --- Semantic (category) fluency: 60s enforced --- | |
| if current_task == "semantic": | |
| st.markdown("### 🐾 Category Fluency") | |
| st.write("List as many **animals** as you can before the timer runs out. Separate with commas or spaces.") | |
| timed_out = enforced_timer("semantic", FLUENCY_TIME_LIMIT) | |
| text = st.text_area("Animals", key="semantic_input", height=110, | |
| placeholder="e.g. dog, cat, elephant, ...", disabled=timed_out, | |
| label_visibility="collapsed") | |
| c1, c2 = st.columns([1, 1]) | |
| with c1: | |
| early = st.button("I'm done →", disabled=timed_out) | |
| if timed_out or early: | |
| st.session_state.responses["semantic_fluency_text"] = st.session_state.get("semantic_input", "") | |
| reset_task_timer("semantic") | |
| goto_next_task() | |
| # --- Phonemic (letter) fluency: 60s enforced --- | |
| elif current_task == "phonemic": | |
| st.markdown("### 🔤 Letter Fluency") | |
| st.write("List as many words as you can that start with the letter **F** (no names of people/places).") | |
| timed_out = enforced_timer("phonemic", FLUENCY_TIME_LIMIT) | |
| text = st.text_area("F words", key="phonemic_input", height=110, | |
| placeholder="e.g. fish, fast, forest, ...", disabled=timed_out, | |
| label_visibility="collapsed") | |
| early = st.button("I'm done →", disabled=timed_out) | |
| if timed_out or early: | |
| st.session_state.responses["phonemic_fluency_text"] = st.session_state.get("phonemic_input", "") | |
| st.session_state.responses["phonemic_letter"] = "F" | |
| reset_task_timer("phonemic") | |
| goto_next_task() | |
| # --- Narrative description: open-ended, no enforced timer (per paradigm) --- | |
| elif current_task == "narrative": | |
| st.markdown("### 📝 Narrative Description") | |
| st.write("Describe your **morning routine** in as much detail as you can. Take your time — this task isn't timed.") | |
| text = st.text_area("Your morning routine", key="narrative_input", height=160, | |
| placeholder="I usually wake up around...", label_visibility="collapsed") | |
| if st.button("Continue →"): | |
| st.session_state.responses["narrative_text"] = st.session_state.get("narrative_input", "") | |
| goto_next_task() | |
| # --- Distractor task: 30s enforced counting-backward, before recall --- | |
| elif current_task == "distractor": | |
| st.markdown("### 🔢 Quick Mental Task") | |
| st.write("Try to keep the six words below in mind, then count backward from 100 by 7s " | |
| "(100, 93, 86 ...) until the timer ends.") | |
| st.markdown("#### " + " • ".join(w.upper() for w in RECALL_WORDS)) | |
| timed_out = enforced_timer("distractor", DISTRACTOR_TIME_LIMIT) | |
| if timed_out: | |
| if st.button("Continue to recall →"): | |
| reset_task_timer("distractor") | |
| goto_next_task() | |
| else: | |
| st.caption("The recall task will unlock automatically when the timer ends.") | |
| # --- Delayed recall: open-ended, no enforced timer --- | |
| elif current_task == "recall": | |
| st.markdown("### 🧩 Delayed Recall") | |
| st.write("Now list as many of the six words from a moment ago as you can remember:") | |
| text = st.text_area("Recalled words", key="recall_input", height=90, label_visibility="collapsed") | |
| if st.button("Submit assessment →", type="primary"): | |
| st.session_state.responses["recall_response_text"] = st.session_state.get("recall_input", "") | |
| st.session_state.responses["recall_target_words"] = RECALL_WORDS | |
| goto_next_task() | |
| # --------------------------------------------------------------------------- | |
| # STAGE 2 — Linguistic Biomarker Dashboard | |
| # --------------------------------------------------------------------------- | |
| elif st.session_state.stage == 2: | |
| st.header("Stage 2 · Linguistic Biomarker Dashboard") | |
| st.caption( | |
| "Computed from your timed/typed responses using standard NLP feature-extraction techniques " | |
| "(spaCy/NLTK), compared against rough, literature-informed reference ranges — " | |
| "**not a clinically derived normative dataset.** See README for the limitation." | |
| ) | |
| profile = st.session_state.profile | |
| bands = profile["bands"] | |
| band_icon = {"below_typical_range": "🔻", "within_typical_range": "✅", "above_typical_range": "🔺"} | |
| band_label = {"below_typical_range": "Below typical range", "within_typical_range": "Within typical range", | |
| "above_typical_range": "Above typical range"} | |
| cols = st.columns(3) | |
| metrics = [ | |
| ("Semantic fluency", profile["semantic_fluency"]["unique_valid_count"], "unique animals named", | |
| bands["semantic_fluency"]), | |
| ("Phonemic fluency", profile["phonemic_fluency"]["unique_valid_count"], "unique F-words named", | |
| bands["phonemic_fluency"]), | |
| ("Lexical diversity (MATTR)", profile["lexical_diversity"]["mattr"], "vocabulary variety score", | |
| bands["lexical_diversity"]), | |
| ("Idea density (approx.)", profile["idea_density"]["approx_idea_density_per_10_words"], "propositions per 10 words", | |
| bands["idea_density"]), | |
| ("Syntactic complexity", profile["syntactic_complexity"]["subordination_index"], "subordinate clauses / sentence", | |
| bands["syntactic_complexity"]), | |
| ("Delayed recall", f"{profile['delayed_recall']['correct_count']}/{profile['delayed_recall']['total_targets']}", | |
| "words correctly recalled", None), | |
| ] | |
| for i, (label, value, sublabel, band) in enumerate(metrics): | |
| with cols[i % 3]: | |
| badge = f"{band_icon.get(band,'')} {band_label.get(band,'')}" if band else "" | |
| st.metric(label, value, help=sublabel) | |
| if badge: | |
| st.caption(badge) | |
| with st.expander("📋 See full extracted profile (raw output)"): | |
| st.json(profile) | |
| with st.expander("ℹ️ What do these markers mean? (with citations)"): | |
| st.markdown(""" | |
| - **Semantic fluency** (animal naming): one of the most-replicated early markers in the | |
| cognitive-aging literature; semantic fluency has been shown to decline faster than letter | |
| fluency specifically in those at elevated Alzheimer's risk (PMC7403823). | |
| - **Lexical diversity (MATTR)**: a length-corrected measure of vocabulary variety; reduced | |
| lexical diversity is associated with reduced vocabulary access in spontaneous speech. | |
| - **Idea density**: a simplified, POS-based approximation of the propositional idea-density | |
| measure made famous by the Nun Study, where idea density in essays written in early | |
| adulthood predicted Alzheimer's pathology roughly 60 years later (Snowdon et al., 1996; | |
| PMC11852352). | |
| - **Syntactic complexity**: simplified sentence structure can reflect increased cognitive load | |
| during language production. | |
| - **Delayed recall**: a classic episodic memory measure, included here in simplified form. | |
| Full citation list is in `README.md`. | |
| """) | |
| st.warning( | |
| "Reference ranges shown above are **rough, literature-informed approximations**, " | |
| "not a validated clinical normative dataset. A single below-range score on a short, " | |
| "typed task says very little on its own — performance varies widely with mood, " | |
| "fatigue, typing speed, and unfamiliarity with the task.", | |
| icon="⚠️", | |
| ) | |
| col_a, col_b = st.columns(2) | |
| with col_a: | |
| if st.button("← Retake assessment"): | |
| st.session_state.stage = 1 | |
| st.session_state.task_idx = 0 | |
| st.session_state.responses = {} | |
| st.session_state.profile = None | |
| for t in TASK_FLOW: | |
| reset_task_timer(t) | |
| st.rerun() | |
| with col_b: | |
| if st.button("Continue to personalized coaching →", type="primary"): | |
| st.session_state.stage = 3 | |
| st.rerun() | |
| # --------------------------------------------------------------------------- | |
| # STAGE 3 — Grounded Prevention Coaching (RAG) | |
| # --------------------------------------------------------------------------- | |
| elif st.session_state.stage == 3: | |
| st.header("Stage 3 · Personalized Prevention Coaching") | |
| st.caption( | |
| "Retrieves relevant brain-health prevention literature based on your Stage 2 profile, " | |
| "then generates a grounded summary — with a faithfulness check against the retrieved sources." | |
| ) | |
| api_key = os.environ.get("GROQ_API_KEY") | |
| if not api_key: | |
| st.error( | |
| "No `GROQ_API_KEY` found in the environment. Set it (e.g. as a HuggingFace " | |
| "Spaces secret) to generate coaching text — see README.md for setup.", | |
| icon="🔑", | |
| ) | |
| else: | |
| if st.session_state.coaching is None: | |
| with st.spinner("Retrieving relevant literature and generating your summary..."): | |
| retriever = LiteratureRetriever() | |
| retrieved = build_retrieved_context(retriever, st.session_state.profile) | |
| coaching = generate_coaching(st.session_state.profile, retrieved) | |
| faithfulness = check_faithfulness(coaching["text"], coaching["sources_used"]) | |
| st.session_state.coaching = {"coaching": coaching, "faithfulness": faithfulness, | |
| "retriever_backend": retriever.backend} | |
| result = st.session_state.coaching | |
| st.markdown("### Your personalized summary") | |
| st.markdown(result["coaching"]["text"]) | |
| if result["coaching"]["sources_used"]: | |
| with st.expander(f"📚 Sources used ({len(result['coaching']['sources_used'])})"): | |
| for s in result["coaching"]["sources_used"]: | |
| st.markdown(f"**{s['title']}** \n*{s['source']}* \n{s['summary']} \n[Link]({s['url']})") | |
| st.markdown("---") | |
| with st.expander("✅ Faithfulness check (RAGAS-style evaluation)"): | |
| f = result["faithfulness"] | |
| score = f.get("faithfulness_score") | |
| if score is not None: | |
| st.metric("Faithfulness score", f"{score:.0%}", | |
| help="Fraction of generated claims directly supported by retrieved sources") | |
| for c in f.get("claims", []): | |
| icon = {"SUPPORTED": "✅", "PARTIAL": "🟡", "UNSUPPORTED": "❌"}.get(c["verdict"], "❓") | |
| st.write(f"{icon} **{c['verdict']}** — {c['claim']}") | |
| st.caption(c.get("reason", "")) | |
| st.caption(f"Retrieval backend used: `{result['retriever_backend']}`") | |
| st.markdown("---") | |
| col_a, col_b = st.columns(2) | |
| with col_a: | |
| if st.button("← Back to dashboard"): | |
| st.session_state.stage = 2 | |
| st.rerun() | |
| with col_b: | |
| if st.button("Start over"): | |
| st.session_state.stage = 1 | |
| st.session_state.task_idx = 0 | |
| st.session_state.responses = {} | |
| st.session_state.profile = None | |
| st.session_state.coaching = None | |
| for t in TASK_FLOW: | |
| reset_task_timer(t) | |
| st.rerun() | |