""" 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()