用户刷题行为序列建模：从时间序列中发现学习模式的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, dtype=float) 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, dtype=float) 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, dtype=float) 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], dtype=float) 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], dtype=float ) 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, dtype=float) 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), dtype=float) 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, dtype=float) 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( sequence=seq, label=1.0 if next_record.is_accepted else 0.0, user_id=user_id, next_problem_id=next_record.problem_id, next_problem_difficulty=next_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_size=5) # 模拟一个用户的行为序列 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( timestamp=base_time + timedelta(hours=i * 8 + np.random.randint(0, 24)), problem_id=1000 + i, difficulty_score=3.0 + np.random.random() * 5, # 3-8分 tags=tags_pool[i % len(tags_pool)], is_accepted=np.random.random() > 0.35, # ~65% 通过率 submission_count=1 + np.random.randint(0, 3), time_spent_min=5 + 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_length=5 ) print(f" 训练样本数: {len(samples)}") # 训练模型 model = BehaviorLSTM( feature_dim=encoder.FEATURE_DIM, hidden_dim=32 ) # 模拟一个样本的前向传播 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()