用户刷题行为序列建模：从时间序列中发现学习模式的AI方法

用户刷题行为序列建模从时间序列中发现学习模式的AI方法一、刷题日志里的故事每个用户刷题都会留下一条行为序列(时间, 题目ID, 结果, 耗时) 的元组串。单看一条记录无趣但把 300 条记录串在一起就能发现极具价值的模式用户 A 每次遇到 DP 题就耗时暴增 3 倍用户 B 在凌晨 1 点刷题的通过率比下午低 40%用户 C 连续错了同类题后第 4 题的通过率显著提升——说明他学会了这些模式无法通过简单的统计如总通过率 65%发现。它们隐藏在时间维度和题目维度的交互中需要序列模型来捕捉。行为序列建模的价值在于把用户现在会什么从静态标签升级为动态状态。一个用户今天不会 DP 不代表明天也不会——序列模型可以捕捉这种学习中的状态转移。二、行为序列的特征化在输入序列模型之前原始行为数据需要被转换为固定维度的特征向量。这里的关键是时间窗口和上下文特征的选择。import numpy as np from typing import List, Tuple, Dict from dataclasses import dataclass from datetime import datetime, timedelta dataclass class SubmitRecord: 单次提交记录 timestamp: datetime problem_id: int difficulty_score: float # 题目难度分 (0-10) tags: List[str] # 题目标签 [dp, array] is_accepted: bool # 是否通过 submission_count: int # 这道题的第几次提交 time_spent_min: float # 解题耗时分钟 dataclass class SequenceSample: 一个训练样本固定长度的行为序列 标签 sequence: np.ndarray # shape: (seq_len, feature_dim) label: float # 下一次提交的通过概率 user_id: str next_problem_id: int next_problem_difficulty: float class BehaviorSequenceEncoder: 将用户行为序列编码为模型可用的特征序列 每个时间步的特征向量包含以下维度 - 题目特征难度分、是否是特定标签one-hot × K个标签 - 行为特征是否通过、第几次提交、耗时 - 时间特征距上次刷题的间隔、当前时段 - 累积特征最近N题的通过率、同类题的通过率 # 关注的算法标签one-hot 编码 TRACKED_TAGS [ dp, greedy, binary_search, bfs, dfs, backtracking, two_pointers, sliding_window, hashmap, stack, queue, tree, graph ] FEATURE_DIM ( 1 # difficulty_score len(TRACKED_TAGS) # 标签 one-hot 3 # is_accepted, submission_count, time_spent 3 # 时间特征 3 # 累积特征 ) def __init__(self, window_size: int 5): 初始化编码器 Args: window_size: 用于计算累积特征的时间窗口最近N题 self.window_size window_size def _extract_problem_features( self, record: SubmitRecord ) - np.ndarray: 提取单条提交记录的题目特征 feats [] # 难度分归一化到 [0, 1] feats.append(record.difficulty_score / 10.0) # 标签 one-hot 编码 tag_set set(record.tags) for tag in self.TRACKED_TAGS: feats.append(1.0 if tag in tag_set else 0.0) return np.array(feats, dtypefloat) def _extract_behavior_features( self, record: SubmitRecord ) - np.ndarray: 提取行为特征 feats [ 1.0 if record.is_accepted else 0.0, # 通过标志 min(record.submission_count / 10.0, 1.0), # 提交次数截断 min(record.time_spent_min / 60.0, 1.0) # 耗时/60分钟截断 ] return np.array(feats, dtypefloat) def _extract_temporal_features( self, current: SubmitRecord, previous: SubmitRecord None ) - np.ndarray: 提取时间特征 feats [] if previous is not None: # 距上次刷题的时间间隔小时log 后归一化 interval_hours ( current.timestamp - previous.timestamp ).total_seconds() / 3600 feats.append(min(np.log1p(interval_hours) / 10.0, 1.0)) else: feats.append(0.0) # 刷题时段0-23点归一化 feats.append(current.timestamp.hour / 24.0) # 是否周末周末刷题行为模式不同 feats.append(1.0 if current.timestamp.weekday() 5 else 0.0) return np.array(feats, dtypefloat) def _extract_cumulative_features( self, recent_records: List[SubmitRecord] ) - np.ndarray: 提取累积统计特征 Args: recent_records: 最近 window_size 个提交记录 累积特征反映用户的当前状态 - 最近通过率最近N题做对了几道 - 同类题频率最近N题中有多少是相同标签 - 平均耗时趋势耗时在上升还是下降 if not recent_records: return np.array([0.0, 0.0, 0.0], dtypefloat) n len(recent_records) # 最近通过率 recent_pass_rate sum( 1 for r in recent_records if r.is_accepted ) / n # 平均耗时min归一化 avg_time sum(r.time_spent_min for r in recent_records) / n avg_time_norm min(avg_time / 30.0, 1.0) # 平均难度 avg_difficulty sum( r.difficulty_score for r in recent_records ) / n / 10.0 return np.array( [recent_pass_rate, avg_time_norm, avg_difficulty], dtypefloat ) def encode_single( self, record: SubmitRecord, previous: SubmitRecord None, recent: List[SubmitRecord] None ) - np.ndarray: 将单条提交记录编码为特征向量 feats [] feats.extend(self._extract_problem_features(record)) feats.extend(self._extract_behavior_features(record)) feats.extend(self._extract_temporal_features(record, previous)) feats.extend(self._extract_cumulative_features(recent or [])) return np.array(feats, dtypefloat) def encode_sequence( self, records: List[SubmitRecord] ) - np.ndarray: 将整个行为序列编码为特征矩阵 shape: (seq_len, feature_dim) Args: records: 按时间排序的行为记录列表 Returns: 特征矩阵每行是一个时间步的特征向量 n len(records) if n 0: return np.zeros((0, self.FEATURE_DIM), dtypefloat) sequence [] for i, record in enumerate(records): previous records[i - 1] if i 0 else None # 最近 N 条记录不包含当前 start max(0, i - self.window_size) recent records[start:i] feats self.encode_single(record, previous, recent) sequence.append(feats) return np.array(sequence, dtypefloat) def build_training_samples( self, user_records: Dict[str, List[SubmitRecord]], seq_length: int 20 ) - List[SequenceSample]: 从多个用户的历史记录构建训练样本 滑动窗口策略对于每个用户的记录序列 用前 seq_length 个时间步的特征预测第 seq_length 1 步的通过概率。 这是序列模型的标准训练样本构造方法 [t0, t1, ..., t19] → 预测 t20 的 is_accepted [t1, t2, ..., t20] → 预测 t21 的 is_accepted ... samples [] for user_id, records in user_records.items(): if len(records) seq_length 1: continue # 先对整个序列做特征编码 full_seq self.encode_sequence(records) # 滑动窗口构造训练样本 for i in range(len(records) - seq_length): seq full_seq[i:i seq_length] next_record records[i seq_length] sample SequenceSample( sequenceseq, label1.0 if next_record.is_accepted else 0.0, user_iduser_id, next_problem_idnext_record.problem_id, next_problem_difficultynext_record.difficulty_score ) samples.append(sample) return samples特征工程的核心是将时间维度融入每一个时间步。不只是用户这道题对了还是错了更是在对上道题一个月没碰的情况下这次做对了吗——这是行为序列建模和简单分类器在信号层面的本质区别。三、LSTM 序列模型捕捉长期依赖LSTM 天然适合行为序列建模用户今天的通过概率不仅取决于今天刷了哪道题还取决于过去一周的行为模式。class BehaviorLSTM: 基于 LSTM 的用户行为预测模型 输入shape (batch_size, seq_length, feature_dim) 输出shape (batch_size, 1) → 下一次提交的通过概率 架构 1. LSTM 层捕获时序依赖 2. Dropout防止过拟合行为序列噪声大 3. 全连接层输出标量预测 def __init__(self, feature_dim: int, hidden_dim: int 64): self.feature_dim feature_dim self.hidden_dim hidden_dim # LSTM 参数手动实现以展示原理 self._init_params(hidden_dim, feature_dim) def _init_params(self, hidden_dim: int, input_dim: int): Xavier 初始化 LSTM 权重 实际项目请使用 PyTorch nn.LSTM。 这里手动实现是为了展示 LSTM 门控机制的内部计算 - 遗忘门决定丢弃哪些历史信息 - 输入门决定写入哪些新信息 - 输出门决定输出哪些信息到隐藏状态 rng np.random.RandomState(42) scale np.sqrt(2.0 / (input_dim hidden_dim)) # 输入到隐藏的权重四个门拼接在一起 self.W_ih rng.randn(4 * hidden_dim, input_dim) * scale # 隐藏到隐藏的权重 self.W_hh rng.randn(4 * hidden_dim, hidden_dim) * scale # 偏置 self.b_ih np.zeros(4 * hidden_dim) self.b_hh np.zeros(4 * hidden_dim) # 全连接输出层 self.W_out rng.randn(1, hidden_dim) * scale self.b_out np.zeros(1) def _lstm_cell( self, x_t: np.ndarray, h_prev: np.ndarray, c_prev: np.ndarray ) - Tuple[np.ndarray, np.ndarray]: 单个 LSTM 时间步的前向传播 # 门控计算gate W_ih x W_hh h_prev bias gates ( self.W_ih x_t self.W_hh h_prev self.b_ih self.b_hh ) # 拆分四个门 i_gate gates[:self.hidden_dim] # 输入门 f_gate gates[self.hidden_dim:2 * self.hidden_dim] # 遗忘门 g_gate gates[2 * self.hidden_dim:3 * self.hidden_dim] # 候选记忆 o_gate gates[3 * self.hidden_dim:] # 输出门 # Sigmoid 激活用于门控输出 0~1 之间的门控信号 i 1.0 / (1.0 np.exp(-i_gate)) f 1.0 / (1.0 np.exp(-f_gate)) o 1.0 / (1.0 np.exp(-o_gate)) # tanh 激活用于候选记忆输出 -1~1 之间 g np.tanh(g_gate) # 记忆更新遗忘门 × 旧记忆 输入门 × 候选记忆 c_t f * c_prev i * g # 隐藏状态输出门 × tanh(新记忆) h_t o * np.tanh(c_t) return h_t, c_t def forward(self, sequence: np.ndarray) - np.ndarray: 前向传播整个序列 Args: sequence: shape (seq_length, feature_dim) Returns: scalar: 预测的通过概率 (0-1) seq_len sequence.shape[0] h np.zeros(self.hidden_dim) c np.zeros(self.hidden_dim) # 逐时间步处理序列 for t in range(seq_len): h, c self._lstm_cell(sequence[t], h, c) # 取最后一个时间步的隐藏状态做预测 logit self.W_out h self.b_out prob 1.0 / (1.0 np.exp(-logit[0])) return prob def predict_batch( self, sequences: np.ndarray ) - np.ndarray: 批量预测 Args: sequences: shape (batch_size, seq_length, feature_dim) Returns: shape (batch_size,): 每个样本的通过概率 probs [] for seq in sequences: prob self.forward(seq) probs.append(prob) return np.array(probs) def demo_training_pipeline(): 演示完整的训练流程 # 模拟用户行为数据实际应用中从数据库读取 encoder BehaviorSequenceEncoder(window_size5) # 模拟一个用户的行为序列 base_time datetime(2026, 7, 1, 14, 0, 0) mock_records [] tags_pool [ [dp], [greedy], [binary_search, array], [dp, array], [bfs, graph], [dfs, tree], [two_pointers], [hashmap], [stack], [queue] ] np.random.seed(42) for i in range(25): record SubmitRecord( timestampbase_time timedelta(hoursi * 8 np.random.randint(0, 24)), problem_id1000 i, difficulty_score3.0 np.random.random() * 5, # 3-8分 tagstags_pool[i % len(tags_pool)], is_acceptednp.random.random() 0.35, # ~65% 通过率 submission_count1 np.random.randint(0, 3), time_spent_min5 np.random.exponential(15) ) mock_records.append(record) # 编码序列 seq encoder.encode_sequence(mock_records) print(f行为序列编码完成) print(f 序列长度: {len(mock_records)}) print(f 特征维度: {seq.shape[1]}) print(f 特征矩阵: {seq.shape}) # 构建训练样本 user_records {user_001: mock_records} samples encoder.build_training_samples( user_records, seq_length5 ) print(f 训练样本数: {len(samples)}) # 训练模型 model BehaviorLSTM( feature_dimencoder.FEATURE_DIM, hidden_dim32 ) # 模拟一个样本的前向传播 if samples: prob model.forward(samples[0].sequence) print(f 示例预测: 通过概率 {prob:.3f}) print(f 实际结果: {通过 if samples[0].label 0.5 else 未通过}) if __name__ __main__: demo_training_pipeline()四、序列建模的边界与局限行为序列建模在当前阶段有三个核心局限局限1冷启动用户新用户没有足够的历史序列。此时必须退化为基于总体统计的默认预测全局平均通过率 题目难度调整同时在推荐系统中对新用户优先推荐低变异性题目。局限2行为噪声用户的通过/失败不完全取决于能力——可能刚好状态好、刚好状态差、碰巧遇到同一道题、中途被打断。序列模型需要通过足够长的窗口≥ 20 题来平滑这些噪声。局限3概念漂移用户的能力会随时间变化提升或遗忘。序列模型如果只取最近时间步的隐藏状态做预测会过度关注近期模式如果对所有时间步等权聚合又会忽视用户最近在进步这个关键信号。建议加上时间衰减注意力——越近的时间步权重越高。五、总结序列提供了静态统计无法提供的东西行为序列建模把一个标签问题这个用户会 DP 吗变成了一个时序预测问题给定过去 20 题的行为序列下一道 DP 题他能过吗。这种视角转换的核心价值在于捕捉学习曲线连续做同类题后通过率提升 → 序列模型能学到他正在掌握的信号发现遗忘模式对某个标签长时间未曾触碰 突然出现 → 预测通过率下降识别卡壳时机连续失败 3 次同类题 耗时增加 → 触发干预建议最终目标是让推荐系统不停留在这道题和上次做的题难度差不多而是基于你最近的行为模式这道题的挑战恰好匹配你的当前状态。