# Project Aura — 调度引擎 (Scheduler) ## 架构概览 调度系统分为三层：**心跳层 → 决策调度层 → 动作执行层** ``` 用户状态变化 / 定时心跳 ↓ ┌──────────────┐ │ 心跳调度器 │ ← 自适应泊松过程 │ Heartbeat │ └──────┬───────┘ ↓ 触发唤醒 ┌──────────────┐ │ 双系统调度器 │ ← 快/慢双通道 │ DualSystem │ ├──────────────┤ │ 快系统(ms级) │ 规则引擎、阈值判断、状态匹配 │ 慢系统(s级) │ LLM评估、上下文检索、意图生成 └──────┬───────┘ ↓ 输出决策 ┌──────────────┐ │ 执行调度器 │ ← 频率控制 + 队列管理 │ Executor │ └──────┬───────┘ ↓ 发起动作 用户界面 ``` --- ## 1. 心跳调度器 (Heartbeat Scheduler) ### 数学模型 心跳间隔由自适应泊松过程决定： ``` λ(t) = λ₀ · (1 + α · Δt_idle / T_window) · (1 - β · interaction_density) ``` - λ₀: 基础心跳频率（默认 180秒） - Δt_idle: 自上次用户交互以来的空闲时间 - T_window: 观察窗口（默认 30分钟） - α: 空闲加速因子（0.1~0.5）— 用户越久没交互，心跳越快 - β: 交互密度衰减因子（0.3~0.8）— 用户越活跃，心跳越慢 ### 状态机 ``` WAITING → (心跳触发) → CHECKING → (价值>阈值) → DECIDING → (决策完成) → WAITING ↓ 价值<阈值 ↓ "等待更佳时机" WAITING DEFERRED → (N分钟后) → WAITING ``` ### 代码实现 ```python import time import random import threading from enum import Enum from dataclasses import dataclass from typing import Optional, Callable class HeartbeatState(Enum): WAITING = "waiting" # 等待下次心跳 CHECKING = "checking" # 正在检查是否需要唤醒 DECIDING = "deciding" # 正在做决策 DEFERRED = "deferred" # 延迟到更佳时机 ACTIVE = "active" # 正在与用户交互 @dataclass class HeartbeatConfig: base_interval: float = 180.0 # 基础间隔（秒） min_interval: float = 30.0 # 最小间隔 max_interval: float = 600.0 # 最大间隔（10分钟无交互就降到最低频） idle_accel_factor: float = 0.3 # 空闲加速 α interact_decay: float = 0.5 # 交互衰减 β observation_window: float = 1800.0 # 观察窗口（30分钟） defer_backoff: float = 120.0 # 延迟后重试间隔 @dataclass class UserContext: last_interaction_time: float = 0.0 # 上次用户交互时间戳 interaction_count_30m: int = 0 # 30分钟内交互次数 current_state: HeartbeatState = HeartbeatState.WAITING is_active_session: bool = False # 是否在活跃对话中 device_mode: str = "normal" # normal / meeting / sleep / driving class HeartbeatScheduler: """ 自适应心跳调度器：决定何时唤醒决策系统 """ def __init__(self, config: HeartbeatConfig = None): self.config = config or HeartbeatConfig() self.context = UserContext() self._timer: Optional[threading.Timer] = None self._on_tick: Optional[Callable] = None self._running = False def set_callback(self, callback: Callable): """设置心跳触发回调""" self._on_tick = callback def compute_interval(self) -> float: """计算下一次心跳间隔""" now = time.time() idle_seconds = now - self.context.last_interaction_time # 基础间隔 + 空闲加速 raw_interval = self.config.base_interval * ( 1 + self.config.idle_accel_factor * idle_seconds / self.config.observation_window ) # 交互密度衰减 density = self.context.interaction_count_30m / (self.config.observation_window / 60) raw_interval *= (1 - self.config.interact_decay * min(density, 1.0)) # 场景修正 if self.context.device_mode == "meeting": raw_interval *= 3 # 会议中降频 elif self.context.device_mode == "sleep": raw_interval *= 10 # 睡眠中极低频 elif self.context.device_mode == "driving": raw_interval *= 5 # 如果在活跃会话中，暂时停用心跳 if self.context.is_active_session: return self.config.max_interval # 夹紧到[min, max] return max(self.config.min_interval, min(raw_interval, self.config.max_interval)) def record_interaction(self): """记录用户交互""" now = time.time() self.context.last_interaction_time = now self.context.interaction_count_30m += 1 self.context.current_state = HeartbeatState.ACTIVE ``` --- ## 2. 双系统调度器 (DualSystem Scheduler) ### 核心思路 借鉴丹尼尔·卡尼曼的「思考，快与慢」理论： - **快系统（System 1）**：规则引擎 + 轻量分类器，毫秒级响应，处理80%的日常判断 - **慢系统（System 2）**：LLM驱动的深度评估，秒级响应，处理需要语境理解的复杂场景 ### 决策流程图 ``` 输入：用户上下文 + 时间 + 设备状态 │ ▼ ┌──────────────────────┐ │ 快系统（System 1） │ │ │ │ 1. 规则过滤 │ │ - 时间窗检查 │ │ - 频率控制检查 │ │ - 场景模式检查 │ │ │ │ 2. 状态模式匹配 │ │ - 深夜+长时间空闲 │ │ - 会话结束后未回复 │ │ - 日程事件触发 │ │ │ │ 3. 快速价值评分 │ │ Value = w₁·urgency │ │ + w₂·recency│ │ + w₃·affinity│ └──────────┬───────────┘ │ ┌──────┴──────┐ │ │ Value>0.7 Value<=0.7 │ │ ▼ ▼ 直接执行 ┌──────────────┐ │ 慢系统(System2)│ │ │ │ 1. LLM评估 │ │ 2. 上下文检索 │ │ 3. 意图生成 │ │ 4. 价值重估 │ └──────┬───────┘ │ ┌─────┴─────┐ │ │ 价值>0.85 价值<=0.85 │ │ ▼ ▼ 执行决策 推迟或放弃 ``` ### 代码实现 ```python from dataclasses import dataclass, field from typing import List, Optional, Tuple from enum import Enum import time class Decision(Enum): IMMEDIATE_CARE = "immediate_care" # 立即关心 GENTLE_REMIND = "gentle_remind" # 温和提醒 DEFER = "defer" # 延后 SILENCE = "silence" # 保持沉默 CONTENT_SUGGEST = "content_suggest" # 内容推荐 CHECK_IN = "check_in" # 问候关心 @dataclass class DecisionContext: user_id: str idle_minutes: float time_of_day: str # morning/afternoon/evening/night device_mode: str last_conversation_topic: Optional[str] = None user_mood: str = "unknown" # happy/sad/anxious/neutral/unknown upcoming_events: List[str] = field(default_factory=list) unread_count: int = 0 @dataclass class ValueScore: score: float # 0.0 - 1.0 urgency: float # 紧急程度 relevance: float # 相关性 interruption_cost: float # 打扰成本 confidence: float # 置信度 reasoning: str = "" # 决策理由（用于思维快照） class FastSystem: """ 快系统：规则引擎 + 轻量评估 响应时间 < 50ms """ # 规则集：优先级从高到低 RULES = [ # (条件函数, 决策结果, 价值分) ("深夜独处", lambda ctx: ctx.time_of_day == "night" and ctx.idle_minutes > 120, Decision.IMMEDIATE_CARE, 0.85), ("长时间沉默", lambda ctx: ctx.idle_minutes > 360 and ctx.time_of_day != "night", Decision.GENTLE_REMIND, 0.70), ("会话搁置", lambda ctx: ctx.last_conversation_topic and ctx.idle_minutes > 30, Decision.CHECK_IN, 0.65), ("会议中", lambda ctx: ctx.device_mode == "meeting", Decision.SILENCE, 0.0), ("睡眠中", lambda ctx: ctx.device_mode == "sleep", Decision.SILENCE, 0.0), ] @classmethod def evaluate(cls, ctx: DecisionContext) -> Tuple[Optional[Decision], ValueScore]: """ 快系统评估 返回 (决策, 价值分数) """ for rule_name, condition, decision, base_score in cls.RULES: if condition(ctx): # 简单价值计算 urgency = min(ctx.idle_minutes / 120, 1.0) # 2小时达到最大紧迫度 interruption = 0.3 if ctx.time_of_day == "night" else 0.1 score = ValueScore( score=base_score * (1 + urgency * 0.3 - interruption * 0.2), urgency=urgency, relevance=base_score, interruption_cost=interruption, confidence=0.8, # 规则引擎置信度高 reasoning=f"[Fast] 匹配规则: {rule_name}" ) return decision, score return None, ValueScore(score=0.0, urgency=0, relevance=0, interruption_cost=0, confidence=0) class SlowSystem: """ 慢系统：LLM驱动的深度评估 响应时间 500ms - 5s """ @staticmethod def evaluate(ctx: DecisionContext) -> Tuple[Optional[Decision], ValueScore]: """ 慢系统评估（调用LLM进行语义理解） 评估维度： 1. 用户情感状态推断（从历史消息中提取） 2. 当前场景的语义价值（是否有关联话题） 3. 打扰成本的精细化估计（用户当前可能在做什么） 4. 个性化偏好匹配（用户之前对类似行为的反馈） 返回决策和价值评分 """ # 这里调用LLM进行深度评估 # prompt示例： """ 你是一个AI主动交互决策引擎。请评估在当前场景下是否应该主动发起对话。 用户上下文： - 空闲时间：{idle_minutes}分钟 - 时段：{time_of_day} - 设备模式：{device_mode} - 上次话题：{last_conversation_topic} - 用户情绪：{user_mood} - 近期事件：{upcoming_events} 请从以下维度评分（0-1）： 1. 消息紧迫性：用户需要关注的事情 2. 话题相关性：当前是否有关联话题可以延续 3. 打扰风险：打扰用户的代价 4. 情感价值：这个互动能提供的情感价值 综合评分 > 0.85 时建议主动发起，0.7-0.85 时延后，< 0.7 时保持沉默。 """ # 模拟返回 return Decision.GENTLE_REMIND, ValueScore( score=0.78, urgency=0.6, relevance=0.7, interruption_cost=0.3, confidence=0.65, reasoning="[Slow] 用户已空闲90分钟，上次话题有延续价值" ) class DualSystemScheduler: """ 双系统调度器：协调快慢系统的协作 """ def __init__(self, fast_system=FastSystem, slow_system=SlowSystem): self.fast = fast_system self.slow = slow_system self.confidence_threshold = 0.7 # 快系统置信度阈值 self.deflection_count = 0 # 连续被跳过次数 def decide(self, ctx: DecisionContext) -> Tuple[Optional[Decision], ValueScore, str]: """ 主决策入口 返回：(决策, 价值分数, 使用系统标识) """ # 1. 先跑快系统 fast_decision, fast_score = self.fast.evaluate(ctx) # 2. 如果快系统有高置信度结果，直接采纳 if fast_decision and fast_score.confidence >= self.confidence_threshold: # 但连续被打断时逐渐降低信任 effective_confidence = fast_score.confidence * (0.9 ** self.deflection_count) if effective_confidence >= self.confidence_threshold: return fast_decision, fast_score, "fast" # 3. 否则调慢系统做深度评估 slow_decision, slow_score = self.slow.evaluate(ctx) # 4. 快慢系统加权融合 combined_score = fast_score.score * 0.3 + slow_score.score * 0.7 # 5. 最终决策 if combined_score >= 0.85: self.deflection_count = 0 return slow_decision or Decision.IMMEDIATE_CARE, ValueScore( score=combined_score, urgency=(fast_score.urgency + slow_score.urgency) / 2, relevance=(fast_score.relevance + slow_score.relevance) / 2, interruption_cost=(fast_score.interruption_cost + slow_score.interruption_cost) / 2, confidence=(fast_score.confidence + slow_score.confidence) / 2, reasoning=f"[Fusion] Fast({fast_score.score:.2f}) + Slow({slow_score.score:.2f}) = {combined_score:.2f}" ), "fusion" elif combined_score >= 0.70: self.deflection_count += 1 return Decision.DEFER, ValueScore( score=combined_score, urgency=0, relevance=0, interruption_cost=0, confidence=0, reasoning="Deferred: 价值不足，等待更佳时机" ), "defer" else: self.deflection_count += 1 return Decision.SILENCE, ValueScore( score=combined_score, urgency=0, relevance=0, interruption_cost=0, confidence=0, reasoning="Silence: 当前不值得打扰用户" ), "silence" ``` --- ## 3. 频率控制 & 反骚扰 (Frequency Control & Anti-Spam) ### 核心约束 ``` 全局约束: · 每小时主动发言 ≤ 3 次 · 每天主动发言 ≤ 15 次 · 连续两次主动发言间隔 ≥ 20 分钟 场景约束: · 用户标记"不喜欢"同一类内容后，7天内不再触发同类消息 · 用户明确说"别打扰"后，24小时内完全静默 · 深夜(23:00-07:00)仅允许紧急事件触发 记忆约束: · 同一话题主动发起 ≤ 2 次（避免唠叨） · 每次主动发言后进入"冷静期"，时长 = 反馈评分 × 基础冷静期 ``` ### 代码实现 ```python @dataclass class FrequencyConfig: max_per_hour: int = 3 max_per_day: int = 15 min_interval_seconds: int = 1200 # 20分钟 silence_after_reject_hours: int = 24 same_topic_max: int = 2 night_silence_start: int = 23 # 23:00 night_silence_end: int = 7 # 07:00 class FrequencyController: """ 频率控制 + 反骚扰 """ def __init__(self, config: FrequencyConfig = None): self.config = config or FrequencyConfig() self.action_log: List[float] = [] # 时间戳列表 self.reject_log: List[Tuple[str, float]] = [] # (topic, 时间戳) self.silence_until: float = 0 # 静默截止时间 self.topic_count: dict = {} def can_act(self, decision: Decision, topic: str = "") -> Tuple[bool, str]: """检查是否允许执行主动发言""" now = time.time() # 1. 静默期内 / 被用户拒绝 if now < self.silence_until: return False, "Silence period active" # 2. 每小时上限 recent_hour = [t for t in self.action_log if now - t < 3600] if len(recent_hour) >= self.config.max_per_hour: return False, "Hourly limit reached" # 3. 每天上限 recent_day = [t for t in self.action_log if now - t < 86400] if len(recent_day) >= self.config.max_per_day: return False, "Daily limit reached" # 4. 最小间隔 if self.action_log and (now - self.action_log[-1]) < self.config.min_interval_seconds: remaining = self.config.min_interval_seconds - (now - self.action_log[-1]) return False, f"Cooldown: {remaining:.0f}s remaining" # 5. 同一话题检查 if topic and self.topic_count.get(topic, 0) >= self.config.same_topic_max: return False, f"Topic '{topic}' already triggered {self.same_topic_max} times" # 6. 深夜限制 hour = time.localtime().tm_hour if self.config.night_silence_start <= hour or hour < self.config.night_silence_end: if decision not in (Decision.IMMEDIATE_CARE,): return False, "Night mode: only urgent allowed" return True, "OK" def record_action(self, decision: Decision, topic: str = "", feedback: float = 0.5): """记录一次主动发言""" now = time.time() self.action_log.append(now) if topic: self.topic_count[topic] = self.topic_count.get(topic, 0) + 1 # 根据反馈动态调整冷静期 cooldown = self.config.min_interval_seconds * (1 + feedback) def record_rejection(self, topic: str = ""): """记录用户拒绝""" now = time.time() self.reject_log.append((topic, now)) self.silence_until = now + self.config.silence_after_reject_hours * 3600 ``` --- ## 4. 完整调度链路示例 ```python # 初始化各组件 heartbeat = HeartbeatScheduler() fast = FastSystem() slow = SlowSystem() scheduler = DualSystemScheduler(fast, slow) freq = FrequencyController() # 心跳触发回调 def on_heartbeat_tick(): ctx = DecisionContext( user_id="user_001", idle_minutes=90, time_of_day="night", device_mode="normal", last_conversation_topic="工作压力", user_mood="anxious", ) decision, score, system = scheduler.decide(ctx) if decision == Decision.SILENCE or decision == Decision.DEFER: return # 不打扰 allowed, reason = freq.can_act(decision, topic=ctx.last_conversation_topic) if not allowed: return # 频率限制 freq.record_action(decision) # 执行主动发言（调用生成模块） execute_action(decision, score, ctx) # 启动心跳 heartbeat.set_callback(on_heartbeat_tick) heartbeat.start() ``` --- ## 5. 性能指标 | 组件 | 响应时间 | 内存 | 适用场景 | |------|---------|------|---------| | 心跳调度器 | <1ms | 2KB | 决定何时检查 | | 快系统规则引擎 | 5-50ms | 100KB | 80%的日常决策 | | 慢系统LLM评估 | 500ms-5s | 动态加载 | 20%需要深度理解的场景 | | 频率控制器 | <1ms | 10KB | 每次执行前检查 |