# HealthTracker 模块规格 > 本文档定义 HealthTracker 模块的设计规格,用于结构化健康数据的采集与存储。 --- ## 1. 背景与目标 ### 1.1 为什么需要 HealthTracker 当前 EzVibe 的健康数据混在 RAG 记忆系统(VectorMemory)中,存在以下问题: | 问题 | 影响 | |------|------| | RAG 擅长模糊检索,不擅长精准时序统计 | 无法快速回答"今天喝了几次水?" | | 语义相似度 ≠ 精确时间 | "上次起身时间"需要扫描所有记忆条目 | | 记忆条目无结构化字段 | LLM 无法直接注入数值型健康数据 | ### 1.2 HealthTracker 的定位 **结构化健康数据库**(SQLite): - 记录精确时间戳的事件(喝水、起身、久坐) - 提供快速查询接口(今日统计、连续时长) - 作为 LLM System Prompt 的固定上下文注入 **与 VectorMemory 的分工**: - `HealthTracker`:结构化时序数据(健康事件) - `VectorMemory`:语义数据(偏好、习惯、对话) --- ## 2. 数据模型 ### 2.1 健康事件表 ```sql CREATE TABLE health_events ( id TEXT PRIMARY KEY, event_type TEXT NOT NULL, -- 'water' | 'stand' | 'stretch' | 'screen_break' timestamp REAL NOT NULL, -- Unix timestamp source TEXT NOT NULL, -- 'user_action' | 'reminder_confirmed' | 'implicit_detected' metadata TEXT NOT NULL, -- JSON: {"count": 1, "note": ""} created_at REAL NOT NULL ); CREATE INDEX idx_event_type ON health_events(event_type); CREATE INDEX idx_timestamp ON health_events(timestamp DESC); ``` **事件类型**: - `water`:喝水 - `stand`:起身/站立 - `stretch`:伸展 - `screen_break`:离开屏幕 **来源**: - `user_action`:用户主动点击"已喝/已做" - `reminder_confirmed`:提醒被确认(点击"吨吨吨") - `implicit_detected`:隐式检测(起身超过 2 分钟无键鼠) ### 2.2 每日统计表 ```sql CREATE TABLE daily_stats ( date TEXT PRIMARY KEY, -- 'YYYY-MM-DD' water_count INTEGER DEFAULT 0, stand_count INTEGER DEFAULT 0, stretch_count INTEGER DEFAULT 0, sedentary_minutes INTEGER DEFAULT 0, -- 连续静坐分钟数 screen_time_minutes INTEGER DEFAULT 0, -- 当日总屏幕时间 last_stand_at REAL, -- 上次起身时间戳 last_water_at REAL, -- 上次喝水时间戳 updated_at REAL NOT NULL ); ``` ### 2.3 久坐警报表 ```sql CREATE TABLE sedentary_alerts ( id TEXT PRIMARY KEY, triggered_at REAL NOT NULL, -- 触发时间戳 acknowledged INTEGER DEFAULT 0, -- 0=未确认, 1=已确认 acknowledged_at REAL, -- 确认时间戳 message TEXT -- 提醒文案 ); ``` --- ## 3. 接口设计 ### 3.1 HealthTracker 类 ```python class HealthTracker: """结构化健康数据追踪器""" def __init__(self, db_path: str = "data/health.db") -> None: ... # ── 事件记录 ────────────────────────────────────────────── async def record_water(self, count: int = 1, source: str = "user_action") -> str: """记录喝水事件""" async def record_stand(self, source: str = "implicit_detected") -> str: """记录起身事件""" async def record_stretch(self, source: str = "reminder_confirmed") -> str: """记录伸展事件""" # ── 查询接口 ────────────────────────────────────────────── def get_today_stats(self) -> dict: """ 返回今日统计: { "water_count": 3, "stand_count": 2, "sedentary_minutes": 120, "last_water_at": 1716098400.0, "last_stand_at": 1716096000.0, } """ def get_consecutive_sedentary_minutes(self) -> int: """返回当前连续静坐分钟数""" def get_last_event_time(self, event_type: str) -> float | None: """返回指定事件类型的最近一次时间戳""" async def get_health_context_for_llm(self) -> str: """ 生成注入 LLM System Prompt 的健康上下文: 【用户健康状态】 - 今日喝水:3 次(上次 14:30) - 连续静坐:120 分钟 ⚠️ - 上次起身:16:00 """ # ── 警报管理 ────────────────────────────────────────────── async def trigger_sedentary_alert(self, message: str) -> str: """触发久坐警报""" def get_pending_alerts(self) -> list[dict]: """返回未确认的久坐警报""" async def acknowledge_alert(self, alert_id: str) -> bool: """确认警报""" # ── 统计聚合 ────────────────────────────────────────────── def get_weekly_report(self) -> dict: """返回本周健康报告""" def get_streak(self, event_type: str) -> int: """返回连续完成某事件的日数(用于成就系统)""" ``` --- ## 4. 与其他模块的集成 ### 4.1 与 EmotionEngine 的集成 ```python # 当用户完成健康动作时,触发情绪更新 async def on_healthy_action(action_type: str): tracker = HealthTracker() await tracker.record_water() # 或 record_stand() 等 emotion = EmotionEngine() emotion.update("user_healthy_action") # → happy +2.0 ``` ### 4.2 与 AgentBrain 的集成 ```python # think() 调用前,注入健康上下文 class AgentBrain: async def think(self, user_input: str, ...): # 获取健康上下文 health_context = await self._health_tracker.get_health_context_for_llm() # 注入 System Prompt system_prompt = self._system_prompt + f"\n\n{health_context}" ``` ### 4.3 与 BehaviorScheduler 的集成 ```python # 久坐检测逻辑 class BehaviorScheduler: async def check_and_trigger(self, user_activity_level: float): consecutive = self._health_tracker.get_consecutive_sedentary_minutes() # 超过 60 分钟且用户极度专注 → 触发 P0 提醒(安静模式) if consecutive > 60 and user_activity_level < 0.15: return self._create_quiet_reminder("久坐提醒", "💧") # 超过 30 分钟 → 普通 P0 提醒 elif consecutive > 30: return self._create_normal_reminder("起来活动一下吧!") ``` --- ## 5. 隐式确认机制 ### 5.1 设计逻辑 当桌宠提醒起身后,系统等待 2 分钟,然后检测: - `ActivityDetector.get_activity() == 0`(键鼠完全无动作) - `ScreenCapture` 无明显变化(仍在工位) 如果两者同时满足,判定用户"去休息了",自动记录 `stand` 事件并触发 `happy` 情绪。 ### 5.2 实现 ```python class ImplicitConfirmation: """ 隐式确认检测器 工作流程: 1. 提醒触发 → 启动 2 分钟计时器 2. 计时结束 → 检查 ActivityDetector 3. 如果无活动 → 记录 stand 事件 + 触发 happy """ def __init__(self, monitor: KeyboardMouseMonitor, tracker: HealthTracker): self._monitor = monitor self._tracker = tracker async def wait_and_confirm(self, reminder_id: str): await asyncio.sleep(120) # 等待 2 分钟 activity = self._monitor.get_activity() if activity == 0: # 用户去休息了 await self._tracker.record_stand(source="implicit_detected") emotion.update("user_healthy_action") return True return False ``` --- ## 6. 实现优先级 | 优先级 | 功能 | 说明 | |--------|------|------| | P0 | `HealthTracker` 基础结构 + 事件记录 | SQLite 表 + CRUD | | P0 | `get_today_stats()` 查询接口 | 核心统计功能 | | P0 | `get_health_context_for_llm()` | LLM 上下文注入 | | P1 | 隐式确认机制 | 增强用户体验 | | P1 | 每日统计聚合 | `daily_stats` 表维护 | | P2 | `get_weekly_report()` 周报 | 数据可视化 | | P2 | 连续成就系统 | `get_streak()` | --- ## 7. 数据库迁移 如果已有 `data/MEMORY.db`,HealthTracker 使用独立的 `data/health.db`。 ```python # 初始化时检查数据库是否存在 def _ensure_db(self): if not Path(self._db_path).exists(): self._init_schema() ``` 未来如果需要合并,可以提供迁移脚本。