EzVibe/docs/v2_design/HEALTH_TRACKER_SPEC.md

HealthTracker 模块规格

本文档定义 HealthTracker 模块的设计规格，用于结构化健康数据的采集与存储。

---

1. 背景与目标

1.1 为什么需要 HealthTracker

当前 EzVibe 的健康数据混在 RAG 记忆系统（VectorMemory）中，存在以下问题：

问题	影响
RAG 擅长模糊检索，不擅长精准时序统计	无法快速回答"今天喝了几次水？"
语义相似度 ≠ 精确时间	"上次起身时间"需要扫描所有记忆条目
记忆条目无结构化字段	LLM 无法直接注入数值型健康数据

1.2 HealthTracker 的定位

结构化健康数据库（SQLite）：

• 记录精确时间戳的事件（喝水、起身、久坐）
• 提供快速查询接口（今日统计、连续时长）
• 作为 LLM System Prompt 的固定上下文注入

与 VectorMemory 的分工：

• HealthTracker：结构化时序数据（健康事件）
• VectorMemory：语义数据（偏好、习惯、对话）

---

2. 数据模型

2.1 健康事件表

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 每日统计表

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 久坐警报表

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 类

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 的集成

# 当用户完成健康动作时，触发情绪更新
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 的集成

# 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 的集成

# 久坐检测逻辑
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 实现

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。

# 初始化时检查数据库是否存在
def _ensure_db(self):
    if not Path(self._db_path).exists():
        self._init_schema()

未来如果需要合并，可以提供迁移脚本。