15 心跳机制

第六层：心跳机制

📌 核心文件：nanobot/heartbeat/service.py

概述

心跳机制（Heartbeat）是一个更灵活、更智能的定时提醒系统，结合了 Cron 的周期性和 Agent 的理解能力。

与 Cron 的区别

对比维度	Cron	Heartbeat
触发方式	固定时间/间隔	智能判断是否需要
消息内容	固定文本	Agent 动态生成
上下文感知	无	有（读取记忆、历史）
适用场景	简单定时提醒	智能助手式唤醒

HeartbeatService 实现

class HeartbeatService:
    """心跳服务 - 主动唤醒 Agent"""
    
    def __init__(self, workspace: Path, bus: MessageBus, interval: int = 3600):
        self.workspace = workspace
        self.bus = bus
        self.interval = interval  # 检查间隔（秒）
        self._running = False
    
    async def start(self):
        """启动心跳服务"""
        self._running = True
        
        while self._running:
            await asyncio.sleep(self.interval)
            
            # 触发心跳
            await self._beat()
    
    async def _beat(self):
        """执行一次心跳"""
        # 读取上下文信息
        now = datetime.now()
        hour = now.hour
        
        # 检查是否需要主动提醒
        should_remind = await self._should_remind(now)
        
        if should_remind:
            # 构建智能提醒消息
            message = await self._build_reminder_message(now)
            
            # 发布到消息总线
            await self.bus.publish_inbound(InboundMessage(
                channel="heartbeat",
                sender_id="system",
                chat_id="default",
                content=message,
                session_key="heartbeat:default"
            ))
    
    async def _should_remind(self, now: datetime) -> bool:
        """判断是否应该提醒"""
        hour = now.hour
        
        # 示例逻辑：
        # - 工作时间（9-18点）每 2 小时提醒一次
        # - 其他时间不提醒
        
        if 9 <= hour < 18:
            # 检查上次提醒时间
            last_beat_file = self.workspace / ".last_heartbeat"
            if last_beat_file.exists():
                last_beat = datetime.fromisoformat(last_beat_file.read_text())
                diff = (now - last_beat).total_seconds()
                
                if diff < 7200:  # 2 小时
                    return False
            
            # 记录本次心跳
            last_beat_file.write_text(now.isoformat())
            return True
        
        return False
    
    async def _build_reminder_message(self, now: datetime) -> str:
        """构建智能提醒消息"""
        hour = now.hour
        
        # 根据时间生成不同的提醒
        if hour == 9:
            return "早上好！有什么我可以帮忙的吗？"
        elif hour == 12:
            return "中午了，要休息一下吗？"
        elif hour == 18:
            return "快下班了，今天有什么收获吗？"
        else:
            return "我在这里，有需要帮忙的吗？"

高级用法：上下文感知

async def _build_reminder_message(self, now: datetime) -> str:
    """结合上下文生成智能提醒"""
    
    # 1. 读取今日笔记
    today = now.strftime("%Y-%m-%d")
    today_file = self.workspace / "memory" / f"{today}.md"
    
    tasks_done = 0
    if today_file.exists():
        content = today_file.read_text()
        tasks_done = content.count("✓")
    
    # 2. 读取长期记忆
    memory_file = self.workspace / "memory" / "MEMORY.md"
    has_deadline = False
    if memory_file.exists():
        content = memory_file.read_text()
        has_deadline = "deadline" in content.lower()
    
    # 3. 生成智能提醒
    if now.hour == 18:
        if tasks_done > 0:
            return f"今天完成了 {tasks_done} 个任务，干得好！明天继续加油！"
        else:
            return "今天似乎没有记录任务，要不要回顾一下今天做了什么？"
    
    if has_deadline and now.hour == 9:
        return "早上好！记得检查一下有截止日期的任务。"
    
    return "我在这里，有什么需要帮忙的吗？"

配置示例

{
  "heartbeat": {
    "enabled": true,
    "interval": 3600,  // 检查间隔（秒）
    "work_hours": {
      "start": 9,
      "end": 18
    },
    "remind_interval": 7200  // 提醒间隔（秒）
  }
}

在 Gateway 中使用

# 初始化
heartbeat = HeartbeatService(
    workspace=workspace,
    bus=bus,
    interval=config.heartbeat.interval
)

# 并发运行
await asyncio.gather(
    agent.run(),
    bus.dispatch_outbound(),
    channel_manager.start(),
    cron_service.start(),
    heartbeat.start(),  # 心跳服务
)

使用场景

1. 工作助手

9:00  - "早上好！今天计划做什么？"
12:00 - "中午了，要不要休息一下？"
15:00 - "下午茶时间，进度如何？"
18:00 - "今天完成了 5 个任务，干得好！"

2. 健康提醒

每 2 小时 - "该站起来活动一下了"
每 1 小时 - "记得喝水"

3. 项目追踪

每天检查：
- 是否有未完成的紧急任务
- 是否有临近deadline的项目
- 今天是否有重要会议

最佳实践

1. 避免过度打扰

# ❌ 太频繁
interval = 300  # 每 5 分钟

# ✅ 合理频率
interval = 3600  # 每小时检查，按需提醒

2. 智能判断

# ✅ 只在需要时提醒
if has_urgent_task and not_reminded_recently:
    remind()
else:
    skip()

3. 可配置性

# 允许用户关闭
if not config.heartbeat.enabled:
    return

# 允许自定义时间
work_hours = config.heartbeat.work_hours

与其他系统的结合

Heartbeat + Memory

# 读取记忆，生成个性化提醒
memory = read_memory()
if "deadline: project-x" in memory:
    remind("project-x 的 deadline 快到了！")

Heartbeat + Subagent

# 定期检查子代理状态
running_subagents = manager.get_running_count()
if running_subagents > 0:
    remind(f"有 {running_subagents} 个后台任务在运行")

小结

• ✅ 比 Cron 更智能的提醒机制
• ✅ 上下文感知（记忆、历史）
• ✅ 动态生成提醒内容
• ✅ 适合个人助手场景

下一步：16-开发自定义工具.md