WeClaw 防骚扰冷却机制：拒绝不是失败，让陪伴更懂用户

yweng18

463人浏览 · 2026-03-22 11:31:14

yweng18 · 2026-03-22 11:31:14 发布

WeClaw 防骚扰冷却机制：拒绝不是失败，让陪伴更懂用户

系列文章第 22 篇 - 预算制+拒绝惩罚+连续忽略检测的三层防护与动态调整算法

📚 专栏信息

《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏

本文是模块六·主动陪伴系统的第 2 篇（共 3 篇），深入剖析 CooldownManager 的三层冷却机制——为什么简单的频率限制无法满足"有温度的陪伴"需求，以及如何用"信用体系"思维设计一套自适应的防骚扰系统。

模块定位：主动陪伴系统 · 第 2 篇（共 3 篇）
前置知识：了解 asyncio.Lock、SQLite 基础
关联文章：第 21 篇（陪伴引擎架构）、第 23 篇（渐进式建档）

在这里插入图片描述

👨‍💻 作者与项目

作者简介：翁勇刚 WENG YONGGANG
新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者
理念：“让 AI 不只是被动工具，更是主动关怀用户的生活伙伴”

💻 项目地址：https://github.com/wyg5208/weclaw.git
🌐 官网地址：https://weclaw.link
📝 作者 CSDN：https://blog.csdn.net/yweng18
⭐ 欢迎 Star⭐、Fork🍴、贡献代码🤝

📝 摘要

本文结构概览：
从"用户被骚扰"的真实场景出发，分析简单频率限制的局限性，引出三层冷却机制的设计理念；用"信用体系"比喻帮助理解预算制、拒绝惩罚、连续忽略检测的协同工作原理；逐行解析 CooldownManager 核心代码，展示如何用 asyncio.Lock 保证并发安全；最后通过一个真实 Bug 的诊断修复过程，总结防骚扰系统的最佳实践。

背景：
主动陪伴系统的核心价值在于"恰到好处"——在用户需要时出现，不需要时保持沉默。如何判断"恰到好处"的边界？这是 CooldownManager 要解决的核心问题。

核心问题：
为什么简单的"每天最多 N 次"配额制无法满足真正的防骚扰需求？用户拒绝和忽略有什么区别？如何让系统从用户反馈中学习？

解决方案：
设计三层防护机制：每日配额（第一层）→ 拒绝惩罚（第二层）→ 连续忽略检测（第三层），配合反馈驱动的动态预算调整算法，实现"越懂用户，越不骚扰"的自适应陪伴。

关键成果：

三层检查实现 100% 防骚扰覆盖
asyncio.Lock 保证并发场景下的状态一致性
动态预算调整：正面反馈 → 配额+1（最多8），负面反馈 → 配额-1（最少2）
SQLite 持久化，跨会话保留用户偏好

适合读者：有 Python 异步编程基础，对用户体验设计感兴趣，正在开发主动式 AI 应用的开发者

阅读时长：约 18 分钟

关键词：CooldownManager、asyncio.Lock、防骚扰、三层冷却、动态预算、用户反馈、SQLite 持久化

一、为什么"频率限制"还不够？——从三种"不耐烦"理解真正的需求

1.1 三种"不耐烦"场景

想象你开发了一个"贴心助手"，它会主动问候用户、提醒任务、关心情绪。上线第一天就收到这些反馈：

场景 A：一天被问候5次

“早上好！”“中午好！”“下午好！”“傍晚好！”“晚安！”——用户：我又不是幼儿园小朋友！

这种问题很好解决：加个每日配额，比如最多 3 次。✅ 问题解决！

场景 B：每次都拒绝但系统继续骚扰

10:00 "今天天气不错，要不要出去走走？"
用户: "不要，我很忙"

12:00 "午餐时间到了，记得吃饭哦~"
用户: "别烦我！"

14:00 "下午了，喝杯水吧？"
用户: 直接卸载 App

用户明确说了"别烦我"，但系统无动于衷。配额制只关心"次数"，不关心"用户态度"。❌ 配额制失败！

场景 C：心情好时接纳、抑郁时需暂停

同一个用户：

周一心情好：“谢谢关心！你说得对！” → 互动愉快
周三被裁员：“……”（既不回复也不拒绝，就是沉默）
周四继续触发：“要不要聊聊？” → 用户更烦躁

用户没有明确拒绝，但连续的沉默已经说明了态度。固定频率完全无法感知这种"软拒绝"。❌ 固定频率失败！

1.2 方案对比：哪种能解决全部问题？

方案	像什么？	能解决场景A	能解决场景B	能解决场景C
固定时间间隔	每 2 小时响一次的闹钟	❌ 可能一天响 8 次	❌ 不管用户态度	❌ 无法感知沉默
简单计数配额	每天限 3 张优惠券	✅ 最多 3 次	❌ 用完 3 次才停	❌ 无法感知沉默
三层冷却机制	信用体系	✅ 每日配额	✅ 拒绝→惩罚期	✅ 忽略→降级

1.3 核心挑战：从"配额制"到"智能节流"

现在我们面临三个必须同时解决的难题：

总量控制：一天不能交互太多次（但具体多少次应该因人而异）
态度响应：用户明确拒绝后，必须立即"冷静"一段时间
沉默感知：连续被忽略时，应该主动降低频率

如何让这三层防护协同工作，而不是互相冲突？

答案就在后面的"信用体系"设计中…

二、核心概念 —— 用"信用体系"理解三层冷却

2.1 什么是"三层冷却机制"？

官方定义：

CooldownManager 是一个多维度的交互频率控制器，通过每日配额、拒绝惩罚、连续忽略检测三个独立但协作的检查层，实现自适应的防骚扰保护。

大白话解释：
把系统当成一个"有情商的朋友"——它知道一天不能打扰你太多次（配额），被你明确拒绝后会"识趣地消失一会儿"（惩罚期），发现你连续沉默后会主动减少打扰（忽略检测）。

生活化比喻——信用体系：

配额 = 信用额度
├── 每日 5 次交互机会，用完就没了
├── 正面反馈 → 额度提升（最高 8）
└── 负面反馈 → 额度降低（最低 2）

拒绝 = 征信记录
├── 用户明确说"不要" → 留下不良记录
├── 进入 4 小时惩罚期 → 期间不能交互
└── 惩罚期结束 → 记录清除

忽略 = 信用降级
├── 用户沉默不回应 → 忽略计数 +1
├── 连续忽略 ≥2 次 → 触发降级
└── 任何响应 → 计数清零

2.2 工作原理：can_interact() 的三层检查

看图理解：

                        can_interact() 检查流程
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 第 1 层：每日配额检查                                         │
│                                                              │
│   daily_count >= daily_budget ?                              │
│   ├── YES → return False ❌ "今日配额已用完"                  │
│   └── NO  → 继续检查 ↓                                       │
└─────────────────────────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 第 2 层：拒绝惩罚检查                                         │
│                                                              │
│   now < last_rejection_time + 4h ?                           │
│   ├── YES → return False ❌ "仍在惩罚期内"                    │
│   └── NO  → 继续检查 ↓                                       │
└─────────────────────────────────────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 第 3 层：连续忽略检查                                         │
│                                                              │
│   consecutive_ignores >= 2 ?                                 │
│   ├── YES → return False ❌ "连续忽略过多"                    │
│   └── NO  → return True  ✅ "可以交互！"                      │
└─────────────────────────────────────────────────────────────┘

关键设计：三层检查是短路求值（Short-Circuit Evaluation）——一旦某层失败，立即返回 False，不再检查后续层。这保证了效率，也符合"严进宽出"的防骚扰原则。

2.3 对比：无冷却 vs 简单计数 vs 三层机制

维度	无冷却	简单计数	三层机制
实现复杂度	O(1)	O(1)	O(1)
配额控制	❌ 无	✅ 有	✅ 有
拒绝响应	❌ 无	❌ 无	✅ 4h 惩罚期
沉默感知	❌ 无	❌ 无	✅ ≥2 次降级
动态调整	❌ 无	❌ 固定	✅ 反馈驱动
持久化	❌ 无	⚠️ 可选	✅ SQLite
用户体验	😡 骚扰	😐 勉强	😊 贴心

为什么选择三层机制？
因为它是唯一能同时解决"配额控制"“态度响应”"沉默感知"三个需求的方案！

三、实战代码详解 —— 手把手教你实现防骚扰系统

3.1 数据结构设计

首先看 CooldownManager 的核心属性：

# src/core/companion_engine.py
class CooldownManager:
    """防骚扰冷却机制。
    
    控制主动交互的频率，防止对用户造成骚扰。
    """
    
    def __init__(self, db_path: Path):
        """初始化冷却管理器。"""
        # === 配额参数 ===
        self.daily_budget: int = 5       # 每日配额（初始值）
        self.min_budget: int = 2         # 最小配额下限
        self.max_budget: int = 8         # 最大配额上限
        
        # === 冷却参数 ===
        self.consecutive_limit: int = 2           # 连续忽略上限
        self.rejection_penalty_hours: int = 4     # 拒绝惩罚时长（小时）
        
        # === 并发控制 ===
        self._interaction_lock: asyncio.Lock = asyncio.Lock()  # 交互锁
        self._current_topic_id: str | None = None              # 当前主题
        
        # === 持久化 ===
        self._db_path = db_path
        
        # 从数据库恢复状态
        self._load_state()

字段说明：

daily_budget: 当前每日配额，可动态调整（2-8）
consecutive_limit: 连续被忽略多少次后停止交互
rejection_penalty_hours: 被拒绝后"闭嘴"多少小时
_interaction_lock: 保证同一时刻只有一个主动交互

SQLite 表结构（由 CompanionEngine._init_db() 创建）：

-- 状态表：存储键值对形式的状态
CREATE TABLE IF NOT EXISTS companion_state (
    key TEXT PRIMARY KEY,           -- 状态键
    value TEXT NOT NULL,            -- 状态值
    updated_at TEXT NOT NULL        -- 更新时间
);

关键状态键：

key	含义	示例值
`daily_count`	今日已交互次数	`"3"`
`daily_budget`	当前每日配额	`"5"`
`last_reset_date`	上次重置日期	`"2026-03-22"`
`last_rejection_time`	上次被拒绝时间	`"2026-03-22T10:30:00"`
`consecutive_ignores`	连续忽略次数	`"1"`
`last_interaction_{topic_id}`	某主题上次交互时间	`"2026-03-22T09:00:00"`

3.2 核心方法：can_interact() —— 三层检查

这是整个冷却机制的心脏：

# src/core/companion_engine.py - CooldownManager.can_interact()
def can_interact(self) -> bool:
    """检查是否可以进行主动交互。
    
    Returns:
        True 如果可以交互，False 如果受限
    """
    # ========== 第 1 层：每日配额检查 ==========
    daily_count = self.get_daily_count()
    if daily_count >= self.daily_budget:
        # ✅ 关键：配额用完，今天不再打扰
        logger.debug("每日配额已用完: %d/%d", daily_count, self.daily_budget)
        return False
    
    # ========== 第 2 层：拒绝惩罚检查 ==========
    last_rejection = self._get_state("last_rejection_time")
    if last_rejection:
        try:
            rejection_time = datetime.fromisoformat(last_rejection)
            # ✅ 关键：计算惩罚期结束时间
            penalty_end = rejection_time + timedelta(hours=self.rejection_penalty_hours)
            if datetime.now() < penalty_end:
                # 还在惩罚期内，不能交互
                logger.debug("仍在拒绝惩罚期内，需等待至 %s", penalty_end)
                return False
        except ValueError:
            pass  # 时间格式错误，忽略
    
    # ========== 第 3 层：连续忽略检查 ==========
    consecutive_ignores = int(self._get_state("consecutive_ignores", "0"))
    if consecutive_ignores >= self.consecutive_limit:
        # ✅ 关键：用户沉默太多次，主动退让
        logger.debug("连续忽略次数过多: %d", consecutive_ignores)
        return False
    
    # ✅ 全部通过，可以交互！
    return True

代码解析：

第 8-11 行：最基础的检查——今天还有没有"额度"
第 14-23 行：检查是否在"惩罚期"内——被拒绝后 4 小时内不能交互
第 26-30 行：检查"软拒绝"——连续被忽略 2 次后停止

为什么这个顺序？
按照"检查成本从低到高"排序：

配额检查：只需比较两个整数
惩罚期检查：需要解析时间字符串
忽略检查：需要读取数据库状态

这种排序让大多数"不可交互"的情况在第 1 层就被拦截，提升性能。

3.3 并发控制：acquire_interaction_lock()

# src/core/companion_engine.py - CooldownManager
async def acquire_interaction_lock(self, topic_id: str) -> bool:
    """尝试获取交互锁。
    
    同一时间只允许一个主动交互进行。
    
    Args:
        topic_id: 要执行的主题 ID
        
    Returns:
        True 如果成功获取锁，False 如果锁已被占用
    """
    # ✅ 关键：非阻塞检查，不等待
    if self._interaction_lock.locked():
        logger.debug("交互锁已被占用，topic_id: %s", self._current_topic_id)
        return False
    
    # 获取锁
    await self._interaction_lock.acquire()
    self._current_topic_id = topic_id
    return True

def release_interaction_lock(self) -> None:
    """释放交互锁。"""
    self._current_topic_id = None
    if self._interaction_lock.locked():
        self._interaction_lock.release()

设计亮点：

非阻塞设计：locked() 检查不会等待，立即返回。这避免了多个主题同时触发时互相阻塞。
记录当前主题：_current_topic_id 便于调试时知道"谁占着锁"。
安全释放：先检查 locked() 再释放，避免"重复释放"异常。

为什么不用 async with lock: ？

# ❌ 不适合的场景
async def bad_design():
    async with self._interaction_lock:  # 会无限等待！
        await do_interaction()

# ✅ 正确做法：非阻塞检查
async def good_design():
    if self._interaction_lock.locked():
        return False  # 直接放弃，不等待
    await self._interaction_lock.acquire()
    try:
        await do_interaction()
    finally:
        self._interaction_lock.release()

主动关怀的场景是"宁可不触发，也不要排队等待"——用户不会希望系统在后台"憋着"一堆待发的关怀消息。

3.4 反馈驱动：adjust_budget()

这是实现"越懂用户，越不骚扰"的关键算法：

# src/core/companion_engine.py - CooldownManager.adjust_budget()
def adjust_budget(self, user_response: str, consecutive_ignores: int) -> None:
    """根据用户反馈调整每日配额。
    
    Args:
        user_response: 用户反馈类型 (positive/negative/neutral)
        consecutive_ignores: 连续忽略次数
    """
    old_budget = self.daily_budget
    
    # ========== 正面反馈：可能增加配额 ==========
    if user_response == "positive":
        # ✅ 关键：只有连续忽略为 0 时才增加
        # 这防止了"刚回复一次就大幅提升"的问题
        if consecutive_ignores == 0:
            self.daily_budget = min(self.daily_budget + 1, self.max_budget)
    
    # ========== 负面反馈或多次忽略：减少配额 ==========
    elif user_response == "negative" or consecutive_ignores >= 2:
        # ✅ 关键：负面反馈直接减配额
        # 连续忽略 ≥2 也视为"软负面"
        self.daily_budget = max(self.daily_budget - 1, self.min_budget)
    
    # ========== 持久化变更 ==========
    if self.daily_budget != old_budget:
        self._save_state("daily_budget", str(self.daily_budget))
        logger.info("每日配额调整: %d → %d", old_budget, self.daily_budget)

算法图解：

                    用户反馈处理流程
                          │
            ┌─────────────┼─────────────┐
            │             │             │
            ▼             ▼             ▼
       "positive"    "neutral"    "negative"
            │             │             │
            ▼             │             ▼
  consecutive_ignores     │      budget -= 1
       == 0 ?             │      (最少 2)
     ┌────┴────┐          │
     │         │          │
    YES       NO          │
     │         │          │
     ▼         ▼          ▼
budget += 1   保持     保持/降低
(最多 8)    不变

配额变化规则总结：

反馈类型	连续忽略	配额变化	说明
positive	0	+1	用户喜欢，增加互动
positive	≥1	不变	刚恢复互动，观察
neutral	任意	不变	保守不动
negative	任意	-1	用户不喜欢，减少
任意	≥2	-1	软拒绝，减少

3.5 交互记录：record_interaction()

# src/core/companion_engine.py - CooldownManager.record_interaction()
def record_interaction(self, topic_id: str, outcome: str) -> None:
    """记录一次交互结果。
    
    Args:
        topic_id: 主题 ID
        outcome: 结果类型 (completed/ignored/rejected/triggered)
    """
    now = datetime.now().isoformat()
    
    # 更新主题的最后交互时间
    self._save_state(f"last_interaction_{topic_id}", now)
    
    # 更新每日计数
    daily_count = self.get_daily_count()
    self._save_state("daily_count", str(daily_count + 1))
    
    # ========== 处理不同的结果 ==========
    if outcome == "rejected":
        # ✅ 被明确拒绝 → 进入惩罚期
        self._save_state("last_rejection_time", now)
        self._save_state("consecutive_ignores", "0")  # 重置忽略计数
        logger.info("用户拒绝了主动关怀，进入 %d 小时冷却期", 
                   self.rejection_penalty_hours)
    
    elif outcome == "ignored":
        # ✅ 被忽略（沉默）→ 累加忽略计数
        consecutive = int(self._get_state("consecutive_ignores", "0"))
        self._save_state("consecutive_ignores", str(consecutive + 1))
        logger.debug("用户忽略了主动关怀，连续忽略: %d", consecutive + 1)
    
    elif outcome in ("completed", "triggered"):
        # ✅ 成功完成或正常触发 → 重置忽略计数
        self._save_state("consecutive_ignores", "0")

outcome 状态转换图：

┌──────────────────────────────────────────────────────────┐
│                     outcome 处理逻辑                      │
└──────────────────────────────────────────────────────────┘

 "rejected"              "ignored"           "completed/triggered"
     │                       │                       │
     ▼                       ▼                       ▼
┌─────────────┐      ┌─────────────┐         ┌─────────────┐
│ 记录惩罚时间 │      │ 忽略计数+1  │         │ 忽略计数=0  │
│ 忽略计数=0  │      │             │         │             │
└─────────────┘      └─────────────┘         └─────────────┘
     │                       │                       │
     ▼                       ▼                       ▼
  4小时内            连续≥2次                   状态健康
  禁止交互           禁止交互                   可继续交互

3.6 易错点：拒绝 vs 忽略的区别

这是初学者最容易混淆的地方：

行为	判定条件	系统响应	恢复方式
拒绝	用户说"不要""别烦我"等	4 小时惩罚期	时间到自动恢复
忽略	60 秒内无任何响应	忽略计数 +1	任何响应即清零
完成	用户正常回复/互动	忽略计数清零	N/A

代码中如何判定？

拒绝检测由上层的 InteractionOrchestrator 在接收用户响应时判定：

# 伪代码：响应分析
def analyze_response(user_text: str) -> str:
    rejection_keywords = ["不要", "别烦", "停止", "不想", "闭嘴", "安静"]
    if any(kw in user_text for kw in rejection_keywords):
        return "rejected"
    return "completed"

忽略检测通过超时机制：

# 伪代码：超时检测
async def wait_for_response(timeout: int = 60) -> str:
    try:
        response = await asyncio.wait_for(get_user_input(), timeout=timeout)
        return "completed"
    except asyncio.TimeoutError:
        return "ignored"

四、问题诊断与修复 —— 从"我不想被打扰"到完美解决

4.1 问题现象：拒绝后仍被骚扰

用户报告：

“我明明说了’我不想被打扰’，为什么系统 30 分钟后又来问候我？”

服务器日志：

2026-03-22 10:30:15 | companion | INFO | 触发早间问候
2026-03-22 10:30:20 | companion | DEBUG | 用户响应: "我不想被打扰"
2026-03-22 10:30:20 | companion | INFO | 记录交互: outcome=completed  ← 问题！
2026-03-22 11:00:00 | companion | INFO | 触发午间关怀

奇怪：用户说"我不想被打扰"，为什么 outcome 是 completed 而不是 rejected？

4.2 根因分析：拒绝检测漏网

排查步骤：

1️⃣ 检查 record_interaction 的调用链：

# InteractionOrchestrator.initiate_care()
result = await self._orchestrator.initiate_care(topic, context)

# 发现：outcome 是由调用方传入的，不是 CooldownManager 判定的
self._cooldown.record_interaction(topic_id, "triggered")  # 总是传 "triggered"

2️⃣ 检查用户响应处理：

# 发现问题所在！
async def on_user_response(response_text: str):
    # ❌ 错误：直接当作完成
    cooldown.record_interaction(topic_id, "completed")
    
    # 没有检查 response_text 是否包含拒绝意图！

3️⃣ 问题链路图：

用户说"我不想被打扰"
          │
          ▼
   on_user_response()
          │
          ▼
   直接标记为 "completed" ← 问题根源！
          │
          ▼
   忽略计数清零
          │
          ▼
   30 分钟后触发下一次

根本原因：响应处理逻辑没有分析用户文本内容，一律视为"完成"！

4.3 修复方案：增加拒绝关键词检测

修复代码：

# ✅ 修复后：增加拒绝检测
REJECTION_KEYWORDS = [
    "不要", "别烦", "停止", "不想", "闭嘴", "安静",
    "别打扰", "烦死了", "够了", "不需要",
]

async def on_user_response(response_text: str, topic_id: str):
    # 检查是否为拒绝
    text_lower = response_text.lower()
    is_rejection = any(kw in text_lower for kw in REJECTION_KEYWORDS)
    
    if is_rejection:
        # ✅ 识别为拒绝
        cooldown.record_interaction(topic_id, "rejected")
        logger.info("检测到拒绝意图: %s", response_text[:50])
    else:
        # 正常完成
        cooldown.record_interaction(topic_id, "completed")

验证结果：

✅ 步骤 1：用户说"我不想被打扰"
   → outcome = "rejected"
   → last_rejection_time = "2026-03-22T10:30:20"
   
✅ 步骤 2：10:30 + 4h = 14:30 之前
   → can_interact() 返回 False
   → 系统保持沉默

✅ 步骤 3：14:30 之后
   → can_interact() 返回 True
   → 可以恢复交互

4.4 经验教训：防止"永久沉默"

修复后出现新问题：如果用户说了"停止"，然后 4 小时内又想互动怎么办？

解决方案：用户主动发起的交互不受惩罚期限制：

async def on_user_initiated_chat(user_message: str):
    """用户主动发起对话时调用。"""
    # ✅ 重置惩罚期（用户主动说话表示愿意互动）
    cooldown._save_state("last_rejection_time", "")
    cooldown._save_state("consecutive_ignores", "0")
    
    # 正常处理对话...

Checklist：

拒绝检测：是否覆盖了常见的拒绝表达？
惩罚恢复：用户主动发起时是否重置惩罚期？
每日重置：新的一天是否清除所有限制？
日志记录：outcome 是否正确记录？

避坑指南：

永久沉默：不要让惩罚期无限延长，设置合理的上限（如 4 小时）
关键词穷举：拒绝关键词要覆盖常见表达，但也要避免误判（如"不要紧"不是拒绝）
每日重置：新的一天应该给用户"重新开始"的机会

五、性能优化与最佳实践

5.1 性能瓶颈分析

Profiling 数据：

can_interact():           0.5ms  (3次数据库读取)
record_interaction():     1.2ms  (2-3次数据库写入)
acquire_interaction_lock(): 0.01ms (纯内存操作)
adjust_budget():          0.3ms  (1次数据库写入)

结论：主要瓶颈在 SQLite I/O。每次 can_interact() 需要 3 次数据库读取。

5.2 优化策略

策略 1：状态缓存

# ❌ 优化前（每次都查数据库）
def can_interact(self) -> bool:
    daily_count = int(self._get_state("daily_count", "0"))
    consecutive_ignores = int(self._get_state("consecutive_ignores", "0"))
    last_rejection = self._get_state("last_rejection_time")
    ...

# ✅ 优化后（内存缓存 + 按需刷新）
class CooldownManager:
    def __init__(self, ...):
        # 内存缓存
        self._cache = {
            "daily_count": 0,
            "consecutive_ignores": 0,
            "last_rejection_time": "",
        }
        self._cache_loaded = False
    
    def _ensure_cache(self):
        """按需加载缓存。"""
        if not self._cache_loaded:
            self._cache["daily_count"] = int(self._get_state("daily_count", "0"))
            self._cache["consecutive_ignores"] = int(self._get_state("consecutive_ignores", "0"))
            self._cache["last_rejection_time"] = self._get_state("last_rejection_time")
            self._cache_loaded = True
    
    def can_interact(self) -> bool:
        self._ensure_cache()
        # 使用缓存数据，避免重复查询
        if self._cache["daily_count"] >= self.daily_budget:
            return False
        ...

代价：增加了 ~50 行代码
收益：can_interact() 从 0.5ms 降到 0.05ms（10x 提升）

策略 2：批量写入

# ❌ 优化前（每个状态单独写入）
def record_interaction(self, topic_id: str, outcome: str):
    self._save_state(f"last_interaction_{topic_id}", now)
    self._save_state("daily_count", str(daily_count + 1))
    if outcome == "rejected":
        self._save_state("last_rejection_time", now)
        self._save_state("consecutive_ignores", "0")

# ✅ 优化后（批量写入）
def record_interaction(self, topic_id: str, outcome: str):
    now = datetime.now().isoformat()
    updates = [
        (f"last_interaction_{topic_id}", now),
        ("daily_count", str(self.get_daily_count() + 1)),
    ]
    if outcome == "rejected":
        updates.extend([
            ("last_rejection_time", now),
            ("consecutive_ignores", "0"),
        ])
    
    # 一次事务完成所有写入
    self._batch_save(updates)

5.3 最佳实践总结

Do’s（推荐做法）：

✅ 使用 asyncio.Lock 保证并发安全
✅ 状态持久化到 SQLite，支持跨会话
✅ 三层检查使用短路求值，提升效率
✅ 提供"用户主动发起"的绕过机制
✅ 每日自动重置，给用户"重新开始"的机会

Don’ts（避免做法）：

❌ 不要在 __init__ 中创建 asyncio.Lock（Qt 场景可能报错）
❌ 不要让惩罚期无限延长
❌ 不要忽略用户的"软拒绝"（沉默）
❌ 不要在主线程执行数据库操作（可能阻塞 UI）
❌ 不要硬编码配额上下限（应该可配置）

黄金法则：

宁可少打扰一次，也不要多打扰一次。用户的信任是最宝贵的资源。

六、总结与展望

6.1 核心要点回顾

本文讲解了 CooldownManager 的三层冷却机制：

3 个关键点：

每日配额（第一层）：控制总量，默认 5 次，可动态调整（2-8）
拒绝惩罚（第二层）：用户明确拒绝后，4 小时内禁止交互
忽略检测（第三层）：连续被忽略 ≥2 次，主动退让

1 个核心公式：

能否交互 = 配额未用完 AND 不在惩罚期 AND 连续忽略 < 2

配额调整 = positive + 忽略=0 → +1
         | negative 或 忽略≥2 → -1
         | 其他 → 不变

1 张架构图：

┌─────────────────────────────────────────────────────┐
│                 CooldownManager                      │
├─────────────────────────────────────────────────────┤
│  配置参数                                            │
│  ├── daily_budget: 5 (2-8)                          │
│  ├── consecutive_limit: 2                           │
│  └── rejection_penalty_hours: 4                     │
├─────────────────────────────────────────────────────┤
│  核心方法                                            │
│  ├── can_interact() → 三层检查                      │
│  ├── record_interaction() → 记录结果                │
│  ├── adjust_budget() → 反馈驱动调整                 │
│  └── acquire/release_interaction_lock() → 并发控制  │
├─────────────────────────────────────────────────────┤
│  持久化                                              │
│  └── SQLite companion_state 表                      │
└─────────────────────────────────────────────────────┘

6.2 下一步学习方向

前置知识：

✅ Python asyncio 基础（Lock、Task）
✅ SQLite 基本操作
✅ 状态机设计模式

后续主题：

📖 下一篇：《第 23 篇：渐进式建档——让 AI 在对话中悄悄了解你》
🔜 下下一篇：《第 24 篇：情绪感知陪伴——从文本中读懂用户心情》

扩展阅读：

6.3 互动环节

思考题：

如果用户连续 3 天每天都拒绝一次，配额会变成多少？（提示：每天重置）
如何区分"用户真的忙"和"用户不喜欢这类关怀"？

讨论话题：

你认为主动关怀的"边界"在哪里？系统应该多主动地关心用户，还是应该更被动地等待召唤？欢迎在评论区分享你的看法！

下期预告：《第 23 篇：渐进式建档——让 AI 在对话中悄悄了解你》

🔐 隐私优先：用户档案的分层存储设计
💬 对话挖掘：从聊天中提取用户偏好
🎯 渐进收集：不让用户感到"被审问"
📊 档案可视化：让用户看到 AI 了解了什么

敬请期待！

附录 A：完整代码清单

文件路径	变更类型	说明
`src/core/companion_engine.py`	核心文件	CooldownManager 类完整实现

关键方法：

CooldownManager.__init__() - 初始化配置和加载状态
CooldownManager.can_interact() - 三层检查核心逻辑
CooldownManager.acquire_interaction_lock() - 非阻塞锁获取
CooldownManager.release_interaction_lock() - 安全释放锁
CooldownManager.record_interaction() - 记录交互结果
CooldownManager.adjust_budget() - 反馈驱动预算调整
CooldownManager.reset_daily_count() - 每日重置

总代码量：约 280 行（CooldownManager 相关）
关键方法：7 个
数据库表：1 个（companion_state）

附录 B：参考资料

Python asyncio.Lock 官方文档
SQLite WAL 模式性能优化
用户体验：主动式 AI 的设计原则
上一篇：《第 21 篇：主动陪伴引擎架构》
下一篇：《第 23 篇：渐进式建档》

原文链接：https://blog.csdn.net/yweng18/article/details/（待发布后更新）

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

Poll/Select机制在驱动中的实现

本文探讨了Linux驱动中poll/select机制的实现要点。通过一个传感器驱动调试案例，分析了用户空间与内核的交互流程，详细讲解了驱动侧需要实现的三个关键动作：定义等待队列、编写poll函数和在适当时机唤醒队列。文章提供了完整的代码示例，并总结了常见陷阱（如忘记调用poll_wait、唤醒时序错误等）和调试技巧。作者结合多年经验建议，驱动开发需考虑系统整体性能，正确处理边界条件，使用原子操作避