From c4576394d4d4bbabd4a90ffe7706d0cdbf041e8a Mon Sep 17 00:00:00 2001 From: starburtMr <1977995433@qq.com> Date: Thu, 14 May 2026 11:09:31 +0800 Subject: [PATCH] Add wellness recommender domain --- README.md | 25 ++++- arch__WellnessRecommender.py | 15 +++ tests/test_wellness_domain.py | 69 ++++++++++++++ wellness_domain.py | 151 ++++++++++++++++++++++++++++++ wellness_recommender.py | 167 ++++++++++++++++++++++++++++++++++ 5 files changed, 426 insertions(+), 1 deletion(-) create mode 100644 arch__WellnessRecommender.py create mode 100644 tests/test_wellness_domain.py create mode 100644 wellness_domain.py create mode 100644 wellness_recommender.py diff --git a/README.md b/README.md index 26cb9cc..c0b383d 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,28 @@ If you plan to run the app in a conda or virtual environment, make sure to set u 4. Once running, the app will be accessible at `localhost:8501`. +### Wellness Micro-Habit Domain Demo + +This repository also includes a wellness micro-habit domain adaptation that recommends small everyday habits from a local catalog. It can use AO packages when installed, and it includes a deterministic fallback so reviewers can run it without private packages or paid APIs. + +Run the wellness demo: + +```bash +streamlit run wellness_recommender.py +``` + +Run the fallback CLI: + +```bash +python wellness_recommender.py +``` + +Run the wellness domain tests: + +```bash +python -m unittest tests/test_wellness_domain.py +``` + ### Docker Installation @@ -55,10 +77,11 @@ You're done! Access the app at `localhost:8501` in your browser. The recommender system works by loading a set of random video links. Once the user hits the Run button, a video will be shown, and the system will suggest whether it recommends the video or not. The user can then provide feedback using "pain" or "pleasure" signals to guide the recommendation process. Based on this feedback, the system adjusts its responses and suggests another video. This cycle continues, allowing for more accurate and personalized recommendations over time. +The wellness micro-habit demo follows the same continuous-feedback pattern with a different domain. It encodes each habit into the same eight-bit AO-compatible input shape using focus area, time bucket, social context, and the user's current goal. User feedback updates fallback rankings immediately and trains an AO Agent when optional AO packages are available. + ## Contributing Fork the repository, make your changes, and submit a pull request for review. - diff --git a/arch__WellnessRecommender.py b/arch__WellnessRecommender.py new file mode 100644 index 0000000..12da572 --- /dev/null +++ b/arch__WellnessRecommender.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +"""AO architecture for the wellness micro-habit recommender domain.""" + +import ao_arch as ar + + +description = "Wellness Micro-Habit Recommender System" + +# focus area, time bucket, social flag, current wellness goal +arch_i = [3, 2, 1, 2] +arch_z = [10] +arch_c = [] +connector_function = "full_conn" + +arch = ar.Arch(arch_i, arch_z, arch_c, connector_function, description) diff --git a/tests/test_wellness_domain.py b/tests/test_wellness_domain.py new file mode 100644 index 0000000..536ebd2 --- /dev/null +++ b/tests/test_wellness_domain.py @@ -0,0 +1,69 @@ +import unittest + +from wellness_domain import ( + WELLNESS_HABITS, + apply_feedback, + encode_wellness_habit, + recommend_wellness_habits, +) + + +class WellnessDomainTests(unittest.TestCase): + def test_encoding_matches_ao_input_shape(self): + habit = WELLNESS_HABITS[0] + + encoded = encode_wellness_habit(habit, goal="calm") + + self.assertEqual(len(encoded), 8) + self.assertTrue(all(bit in (0, 1) for bit in encoded)) + + def test_sleep_goal_prioritizes_sleep_or_reflection(self): + ranked = recommend_wellness_habits( + goal="better-sleep", + minutes_available=20, + limit=3, + ) + + self.assertTrue( + any(habit.focus in {"sleep", "reflection", "stress"} for habit, _ in ranked) + ) + + def test_deep_work_goal_prioritizes_focus(self): + ranked = recommend_wellness_habits(goal="deep-work", minutes_available=30, limit=3) + + self.assertTrue(any(habit.focus == "focus" for habit, _ in ranked)) + + def test_social_filter_penalizes_social_habits(self): + ranked = recommend_wellness_habits( + goal="reset", + minutes_available=20, + social_ok=False, + limit=5, + ) + + self.assertTrue(all(not habit.social for habit, _ in ranked[:3])) + + def test_feedback_changes_ranking(self): + baseline = recommend_wellness_habits(goal="calm", minutes_available=15, limit=1)[0][0] + target = "Long-form weekly reset" + feedback = {} + for _ in range(10): + feedback = apply_feedback(feedback, target, liked=True) + + updated = recommend_wellness_habits( + goal="calm", + minutes_available=15, + feedback=feedback, + limit=1, + )[0][0] + + self.assertNotEqual(baseline.name, updated.name) + self.assertEqual(updated.name, target) + + def test_unknown_goal_is_rejected(self): + with self.assertRaises(ValueError): + recommend_wellness_habits(goal="unknown") + + +if __name__ == "__main__": + unittest.main() diff --git a/wellness_domain.py b/wellness_domain.py new file mode 100644 index 0000000..b160dd2 --- /dev/null +++ b/wellness_domain.py @@ -0,0 +1,151 @@ +"""Wellness micro-habit recommendation domain for the AO recommender demo.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable + + +FOCUS_BITS = { + "stress": [0, 0, 0], + "sleep": [0, 0, 1], + "focus": [0, 1, 0], + "energy": [0, 1, 1], + "connection": [1, 0, 0], + "reflection": [1, 0, 1], +} + +TIME_BITS = { + "short": [0, 0], + "medium": [0, 1], + "long": [1, 1], +} + +GOAL_BITS = { + "calm": [0, 0], + "better-sleep": [0, 1], + "deep-work": [1, 0], + "reset": [1, 1], +} + + +@dataclass(frozen=True) +class WellnessHabit: + """A small wellness habit that can be recommended.""" + + name: str + focus: str + time_bucket: str + social: bool + energy_level: str + minutes: int + contexts: tuple[str, ...] + description: str + + +WELLNESS_HABITS: tuple[WellnessHabit, ...] = ( + WellnessHabit("Two-minute box breathing", "stress", "short", False, "low", 2, ("calm", "reset"), "Use a simple breathing cadence before a meeting or after context switching."), + WellnessHabit("Write tomorrow's first task", "focus", "short", False, "low", 5, ("deep-work", "better-sleep"), "Close the day by choosing one concrete next action for tomorrow."), + WellnessHabit("Screen-off wind-down block", "sleep", "medium", False, "low", 20, ("better-sleep", "calm"), "Create a low-stimulation buffer before bedtime."), + WellnessHabit("Sunlight and water reset", "energy", "short", False, "medium", 10, ("reset", "deep-work"), "Step outside, hydrate, and give your attention a clean restart."), + WellnessHabit("Focus sprint with phone away", "focus", "medium", False, "medium", 25, ("deep-work", "reset"), "Work on one task with notifications out of reach."), + WellnessHabit("Send one appreciation note", "connection", "short", True, "low", 5, ("calm", "reset"), "Strengthen a relationship with a quick specific message."), + WellnessHabit("Midday walk without audio", "reflection", "medium", False, "medium", 15, ("calm", "reset"), "Let your mind settle while moving at an easy pace."), + WellnessHabit("Tidy one visible surface", "energy", "short", False, "medium", 8, ("reset", "deep-work"), "Remove one small source of visual friction from your workspace."), + WellnessHabit("Guided body scan", "stress", "medium", False, "low", 12, ("calm", "better-sleep"), "Notice tension and relax one area at a time."), + WellnessHabit("Plan a low-effort social check-in", "connection", "medium", True, "medium", 15, ("calm", "reset"), "Pick one person and one easy way to reconnect this week."), + WellnessHabit("Evening reflection note", "reflection", "short", False, "low", 7, ("better-sleep", "calm"), "Write what worked today and what can be lighter tomorrow."), + WellnessHabit("Long-form weekly reset", "reflection", "long", False, "medium", 35, ("reset", "deep-work"), "Review commitments, remove stale tasks, and choose the next week's focus."), +) + + +def encode_wellness_habit(habit: WellnessHabit, goal: str = "calm") -> list[int]: + """Encode a wellness habit plus user goal into the AO eight-bit input shape.""" + + if habit.focus not in FOCUS_BITS: + raise ValueError(f"Unknown focus: {habit.focus}") + if habit.time_bucket not in TIME_BITS: + raise ValueError(f"Unknown time bucket: {habit.time_bucket}") + if goal not in GOAL_BITS: + raise ValueError(f"Unknown goal: {goal}") + + social_bit = [1 if habit.social else 0] + return FOCUS_BITS[habit.focus] + TIME_BITS[habit.time_bucket] + social_bit + GOAL_BITS[goal] + + +def score_wellness_habit( + habit: WellnessHabit, + goal: str, + minutes_available: int, + preferred_energy: str = "low", + social_ok: bool = True, + feedback: dict[str, int] | None = None, +) -> int: + """Score a habit using deterministic context preferences plus feedback.""" + + score = 0 + if goal in habit.contexts: + score += 35 + if habit.minutes <= minutes_available: + score += 20 + else: + score -= (habit.minutes - minutes_available) // 5 * 5 + if habit.energy_level == preferred_energy: + score += 14 + if not social_ok and habit.social: + score -= 18 + if goal == "better-sleep" and habit.focus in {"sleep", "reflection", "stress"}: + score += 12 + if goal == "deep-work" and habit.focus in {"focus", "energy"}: + score += 12 + if goal == "calm" and habit.energy_level == "low": + score += 8 + if feedback: + score += feedback.get(habit.name, 0) * 12 + return score + + +def recommend_wellness_habits( + goal: str = "calm", + minutes_available: int = 15, + preferred_energy: str = "low", + social_ok: bool = True, + feedback: dict[str, int] | None = None, + habits: Iterable[WellnessHabit] = WELLNESS_HABITS, + limit: int = 5, +) -> list[tuple[WellnessHabit, int]]: + """Return wellness recommendations sorted from strongest to weakest match.""" + + if goal not in GOAL_BITS: + raise ValueError(f"Unknown goal: {goal}") + if preferred_energy not in {"low", "medium"}: + raise ValueError(f"Unknown preferred energy: {preferred_energy}") + + ranked = [ + ( + habit, + score_wellness_habit( + habit, + goal=goal, + minutes_available=minutes_available, + preferred_energy=preferred_energy, + social_ok=social_ok, + feedback=feedback, + ), + ) + for habit in habits + ] + ranked.sort(key=lambda item: (item[1], -item[0].minutes, item[0].name), reverse=True) + return ranked[:limit] + + +def apply_feedback( + feedback: dict[str, int] | None, + habit_name: str, + liked: bool, +) -> dict[str, int]: + """Return updated feedback weights for a wellness habit.""" + + updated = dict(feedback or {}) + updated[habit_name] = updated.get(habit_name, 0) + (1 if liked else -1) + return updated diff --git a/wellness_recommender.py b/wellness_recommender.py new file mode 100644 index 0000000..575edbb --- /dev/null +++ b/wellness_recommender.py @@ -0,0 +1,167 @@ +"""Streamlit and CLI demo for a wellness micro-habit recommender domain.""" + +from __future__ import annotations + +from wellness_domain import ( + WELLNESS_HABITS, + apply_feedback, + encode_wellness_habit, + recommend_wellness_habits, +) + + +def create_agent(): + """Create an AO Agent when optional AO packages are installed.""" + + try: + import ao_core as ao + from arch__WellnessRecommender import arch + except Exception: + return None + + agent = ao.Agent(arch, notes="Wellness Micro-Habit Agent") + for _ in range(4): + agent.reset_state() + agent.reset_state(training=True) + return agent + + +def ao_percentage(agent, binary_input: list[int]) -> int | None: + """Return an AO recommendation percentage, or None when AO is unavailable.""" + + if agent is None: + return None + + agent.reset_state() + response = None + for _ in range(5): + response = agent.next_state(INPUT=binary_input, print_result=False) + + if response is None: + return None + + return round(sum(1 for value in response if value == 1) / len(response) * 100) + + +def train_agent(agent, binary_input: list[int], liked: bool) -> None: + """Train the optional AO Agent on user feedback.""" + + if agent is None: + return + + import numpy as np + + label_value = 1 if liked else 0 + label = np.full(agent.arch.Z__flat.shape, label_value, dtype=np.int8) + for _ in range(5 if liked else 10): + agent.reset_state() + agent.next_state(INPUT=binary_input, LABEL=label, print_result=False, unsequenced=True) + + +def run_cli() -> None: + """Print fallback recommendations without requiring Streamlit or AO packages.""" + + print("Top wellness micro-habit recommendations:") + for habit, score in recommend_wellness_habits(goal="calm", minutes_available=15): + print(f"- {habit.name} ({score})") + print(f" input={encode_wellness_habit(habit, 'calm')}") + print(f" {habit.description}") + + +def run_streamlit() -> None: + """Run the interactive Streamlit demo.""" + + import streamlit as st + + st.set_page_config( + page_title="Wellness Recommender by AO Labs", + page_icon="misc/ao_favicon.png", + layout="wide", + initial_sidebar_state="expanded", + ) + + if "wellness_feedback" not in st.session_state: + st.session_state.wellness_feedback = {} + if "wellness_agent" not in st.session_state: + st.session_state.wellness_agent = create_agent() + + st.title("Wellness Micro-Habit Recommender") + st.write("A domain adaptation of the AO recommender for everyday wellness habits.") + + with st.sidebar: + goal = st.selectbox( + "Current goal", + ("calm", "better-sleep", "deep-work", "reset"), + index=0, + format_func=lambda value: value.replace("-", " ").title(), + ) + preferred_energy = st.selectbox( + "Preferred effort level", + ("low", "medium"), + index=0, + format_func=str.title, + ) + minutes_available = st.slider("Minutes available", 2, 40, 15, 1) + social_ok = st.checkbox("Include social habits", value=True) + ao_status = "available" if st.session_state.wellness_agent is not None else "fallback mode" + st.write(f"AO Agent: {ao_status}") + + ranked = recommend_wellness_habits( + goal=goal, + minutes_available=minutes_available, + preferred_energy=preferred_energy, + social_ok=social_ok, + feedback=st.session_state.wellness_feedback, + limit=len(WELLNESS_HABITS), + ) + + for habit, fallback_score in ranked[:5]: + binary_input = encode_wellness_habit(habit, goal) + ao_score = ao_percentage(st.session_state.wellness_agent, binary_input) + display_score = ao_score if ao_score is not None else fallback_score + + st.subheader(habit.name) + st.write(habit.description) + st.write( + { + "focus": habit.focus, + "time_bucket": habit.time_bucket, + "social": habit.social, + "energy_level": habit.energy_level, + "minutes": habit.minutes, + "encoded_input": binary_input, + "score": display_score, + } + ) + + left, right = st.columns(2) + if left.button("Recommend more like this", key=f"like-{habit.name}"): + st.session_state.wellness_feedback = apply_feedback( + st.session_state.wellness_feedback, habit.name, liked=True + ) + train_agent(st.session_state.wellness_agent, binary_input, liked=True) + st.rerun() + if right.button("Recommend less like this", key=f"less-{habit.name}"): + st.session_state.wellness_feedback = apply_feedback( + st.session_state.wellness_feedback, habit.name, liked=False + ) + train_agent(st.session_state.wellness_agent, binary_input, liked=False) + st.rerun() + + +def is_streamlit_runtime() -> bool: + """Detect whether this script is being executed by Streamlit.""" + + try: + from streamlit.runtime.scriptrunner import get_script_run_ctx + except Exception: + return False + + return get_script_run_ctx() is not None + + +if __name__ == "__main__": + if is_streamlit_runtime(): + run_streamlit() + else: + run_cli()