RAGRetrieval-Augmented Generation检索增强生成在自然语言处理领域已经火了很久。最近我在调研时间序列预测方案时发现不少团队开始把这套“先检索、再生成”的思路迁移到时序预测任务上TS-RAG 就是其中比较有代表性的方向之一。传统的时间序列预测模型通常只依赖当前输入窗口内的历史数据一旦遇到训练集中很少出现的模式预测效果就很容易下滑。而 RAG 的思路恰好可以补上这个短板先从一个大型历史片段库中检索出与当前输入最相似的若干历史模式再把这些模式及其后续走势作为额外上下文交给预测模型。这样模型在做预测时就相当于参考了“过去遇到类似情况时结果是怎么演化的”。这篇文章我会从 RAG 的原始概念讲起拆解 TS-RAG 的整体架构然后给出一个可以直接运行的 Python 实战案例覆盖片段切分、向量化、相似检索、增强特征构造、预测生成和效果评估全流程。最后还会整理常见问题和工程落地建议适合刚接触时序预测的读者也适合想在业务中引入检索增强机制的开发者。1. 背景与核心概念1.1 从 RAG 说起RAG 最初是为大语言模型服务的。标准的 RAG 流程大致是先把外部知识文档切分成块做向量化后存入索引用户提问时系统先把问题向量化去索引中检索出最相关的若干文档片段最后把原始问题和检索到的文档片段一起拼进 Prompt交给大模型生成回答。这个流程的核心价值有三个缓解幻觉模型回答时有了事实依据而不是凭记忆编造。知识可更新不需要重新训练模型只要替换知识库就能引入新信息。支持可解释引用回答可以追溯到具体检索到的文档片段。把 RAG 从文本领域挪到时间序列领域思想是一样的把“文档片段”替换成“历史时间序列片段”把“用户问题”替换成“当前输入窗口”把“生成回答”替换成“预测未来走势”。1.2 时间序列预测的三大痛点时间序列预测已经在电力负荷、销量、流量、气象、工业监控等场景中广泛使用。传统方法 ARIMA、Prophet、LSTM、Transformer 都有各自的优势但在落地时普遍会遇到几个问题第一个痛点是长尾模式预测困难。很多真实业务数据里高频出现的常规模式模型学得很好但低频出现的异常模式、突发模式样本太少模型很难从训练集中充分学习。第二个痛点是概念漂移。真实系统的数据分布不是一成不变的用户习惯变化、市场环境变化、设备老化都会导致历史规律失效。离线训练好的模型上线一段时间后准确率往往明显下降。第三个痛点是结果难以解释。深度模型的预测结果很难说清楚“为什么这么预测”业务方在决策时往往不放心。而如果模型能告诉用户“本次预测参考了历史上几次非常相似的模式它们的后续走势分别是多少”信任度会高很多。1.3 TS-RAG 的核心思想TS-RAG 的思路可以概括成一句话在预测未来之前先检索历史中与当前模式最相似的一批“案例”把这些案例的后续走势作为增强上下文交给生成器。整个流程和文本 RAG 高度对应历史时序数据 - 片段切分 - 向量化 - 构建索引 当前输入窗口 - 向量化 - 检索 Top-K 相似片段 候选片段 后续走势 - 构造增强特征 增强特征 - 预测生成器 - 输出未来预测本质上TS-RAG 是一种案例推理Case-Based Reasoning和现代向量检索的结合。它的优势在于即使模型训练集里没有覆盖某个罕见模式只要历史片段库足够大仍然有可能检索到形状相似的参考案例。1.4 典型应用场景电力负荷预测不同季节、不同天气、不同工作日的负荷曲线形态不同通过检索相似日期的历史负荷曲线辅助预测。电商销量预测大促、节假日、突发事件造成销量模式突变时检索相似促销周期的历史销量走势。工业设备监测设备故障前的传感器时间序列往往有相似的前兆模式检索历史故障样本辅助剩余寿命预测。气象预报检索历史上相似的气压、温度、湿度演变过程辅助短期气象要素预测。金融时间序列研究在合规前提下检索相似行情形态作为量价分析的参考注意控制风险不能作为投资建议。2. TS-RAG 整体架构设计2.1 总体架构为了便于理解我先把 TS-RAG 的架构画出来。这里不使用复杂的绘图工具直接用文字图表示┌──────────────────────────────────────┐ │ 历史时间序列库 │ └──────────────────────────────────────┘ │ 滑动窗口切分 ▼ ┌──────────────────────────────────────┐ │ 历史片段 后续走势 │ └──────────────────────────────────────┘ │ 向量化 ▼ ┌──────────────────────────────────────┐ │ 向量索引检索库 │ └──────────────────────────────────────┘ ▲ │ 相似度检索Top-K ┌────────────┐ ┌─────────┴─────────┐ │ 当前输入窗口 │─────▶│ 查询向量化 │ └────────────┘ └───────────────────┘ │ ▼ ┌──────────────────────────────────────┐ │ 增强上下文构造候选片段后续走势 距离 │ └──────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────┐ │ 预测生成器RF/LSTM/LLM等 │ └──────────────────────────────────────┘ │ ▼ 预测结果2.2 五个核心模块片段切分模块负责把长序列切成固定长度的窗口片段。和文本分块类似窗口大小、滑动步长会直接影响索引规模和检索效果。向量化模块负责把时间序列片段映射成向量。最简单的方式是直接使用归一化后的序列值加统计特征更复杂的方式是使用自编码器或时序基础模型提取表示。检索模块负责从历史片段库中找到与查询片段最相似的 Top-K 候选。这里的核心是相似度度量和向量索引。小规模场景可以直接暴力计算大规模场景需要引入近似最近邻索引ANN。增强上下文构造模块负责把检索到的候选片段、候选片段的后续走势、相似度距离等信息拼接成模型可用的特征。这是 TS-RAG 和普通预测模型最大的区别。预测生成模块负责根据增强特征生成未来预测。生成器可以是随机森林、XGBoost、LSTM、Transformer也可以是时序基础模型或通用大语言模型。2.3 与传统预测流程的区别对比维度传统预测流程TS-RAG 预测流程输入信息仅当前输入窗口当前输入窗口 检索到的历史相似案例模式利用靠模型隐式学习训练集中的模式预测时显式检索相似模式冷启动能力训练数据不足时效果差有历史片段库即可检索参考可解释性较弱可以说明参考了哪些历史案例工程复杂度相对简单需要额外维护片段库和索引从这张表能看出TS-RAG 并不是要完全替代传统模型而是在传统模型基础上增加一个“外部记忆”通道让模型有机会参考更丰富的历史案例。3. 环境准备与数据说明3.1 环境依赖本文的实战案例使用 Python 实现主要依赖以下库库名作用numpy数值计算与数组操作pandas数据读取与处理scikit-learn检索器、预测模型、评估指标关于版本建议使用当前常用的稳定版本即可。Python 使用 3.9 及以上版本重点代码不依赖最新语法旧版本也能兼容。安装依赖可以直接执行pip install numpy pandas scikit-learn3.2 项目结构为了演示方便我把所有代码组织成一个脚本加若干辅助函数项目结构如下ts_rag_demo/ ├── ts_rag_demo.py # 完整可运行脚本 └── README.md # 说明文档实际业务项目中建议拆分成多个模块ts_rag/ ├── data.py # 数据加载与构造 ├── segment.py # 时间序列片段切分 ├── embedding.py # 片段向量化 ├── retriever.py # 检索器封装 ├── features.py # 增强特征构造 ├── generator.py # 预测生成器 └── pipeline.py # 完整流水线3.3 数据构造说明为了实验可复现我不使用外部数据集而是构造一份模拟用电负荷数据。数据包含三个成分日周期性每天的用电高峰和低谷。季节趋势长时间尺度上的缓慢变化。随机噪声模拟真实采集中的不确定性。生成函数如下import numpy as np def generate_load_data(n_days400, points_per_day24, seed42): 生成模拟电力负荷数据 - n_days: 模拟天数 - points_per_day: 每天采样点数这里取24即每小时一个点 rng np.random.default_rng(seed) total_points n_days * points_per_day t np.arange(total_points) # 每日周期性高峰在白天低谷在凌晨 daily 50 20 * np.sin(2 * np.pi * t / points_per_day - np.pi / 2) # 季节趋势用低频正弦模拟 seasonal 10 * np.sin(2 * np.pi * t / (total_points * 0.2)) # 周效应用7天周期模拟工作日/周末差异 weekly 5 * np.sin(2 * np.pi * t / (7 * points_per_day)) # 随机噪声 noise rng.normal(0, 2, sizetotal_points) return daily seasonal weekly noise这里的参数不是固定的你可以根据实际业务调整周期项。重点是理解数据包含多种周期性成分这样 TS-RAG 才有机会通过检索相似周期片段发挥作用。4. 核心模块代码实现下面我按模块拆开讲解最后在完整实战中把代码串起来。4.1 时间序列片段切分时间序列片段切分是 TS-RAG 的基础。我们定义两个概念输入片段长度为window_size的历史窗口。后续走势输入片段之后长度为horizon的真实未来值也是模型要预测的目标。切分时采用滑动窗口方式步长为 1这样能最大化利用历史数据def make_samples(series, window_size24, horizon6): 构造时间序列样本 - series: 一维时间序列 - window_size: 输入窗口长度 - horizon: 预测步长 返回: - segments: 输入片段列表 - continuations: 片段后续走势列表 - starts: 每个片段后续走势的起始位置 segments [] continuations [] starts [] for start in range(0, len(series) - window_size - horizon 1): seg series[start:start window_size] cont series[start window_size:start window_size horizon] segments.append(seg) continuations.append(cont) starts.append(start window_size) # 后续走势的起始下标 return np.array(segments), np.array(continuations), np.array(starts)这里的关键点是starts[i]记录了第 i 个候选片段的“未来起点”。后面做时间安全过滤时会用到当查询片段的起始位置是query_start时我们只允许检索那些“未来起点 query_start”的候选片段避免未来信息泄漏。4.2 片段向量化时间序列片段不能直接放进最近邻索引需要先转成向量。最简单的向量化方式是把“归一化后的序列值”和“统计特征”拼接起来。归一化后的序列值描述片段的形状统计特征描述片段的整体水平def build_query_features(segment): 将时间序列片段转成特征向量 特征 统计特征 归一化序列值 seg np.asarray(segment, dtypenp.float64) mean_val np.mean(seg) std_val np.std(seg) 1e-8 # 加极小值防止除零 # 归一化消除幅度差异保留形状信息 normalized (seg - mean_val) / std_val # 统计特征幅度信息 stats np.array([ mean_val, std_val, np.min(seg), np.max(seg), np.median(seg) ]) return np.concatenate([stats, normalized])为什么要同时保留统计特征和归一化序列值如果只用归一化序列值那么“整体偏高”的片段和“整体偏低”但形状相同的片段会非常相似这会混淆检索结果。加入均值、标准差、最大值、最小值等统计特征后检索既能关注形状相似也能关注幅度相似。如果你的场景里更关注形状而不是幅度也可以去掉统计特征只保留归一化后的序列值。这个取舍取决于业务。4.3 相似片段检索含时间安全过滤检索模块是 TS-RAG 的核心。本文使用 scikit-learn 的NearestNeighbors距离度量采用余弦相似度实际是余弦距离。余弦距离在时间序列场景中的表现通常不错因为归一化后的序列向量已经消除了尺度差异余弦距离关注的是方向而不是绝对数值。直接使用kneighbors会有一个严重问题如果查询片段本身来自历史序列检索库中可能包含和它高度重叠甚至完全相同的候选片段。这样模型在训练和评估时都会“偷看未来”导致效果被高估。解决办法是加入时间安全过滤候选片段的后续走势起始位置必须早于查询片段的起始位置即候选片段完全发生在查询片段之前。from sklearn.neighbors import NearestNeighbors def safe_retrieve(retriever, query_vec, candidate_starts, query_start, candidate_k50, top_k5): 安全检索候选片段必须早于查询片段防止未来信息泄漏 - retriever: 已拟合的 NearestNeighbors - query_vec: 查询片段向量 - candidate_starts: 候选片段后续走势的起始位置 - query_start: 查询片段的起始位置 - candidate_k: 先取多少个近邻再过滤 - top_k: 最终返回的候选数量 distances, indices retriever.kneighbors([query_vec], n_neighborscandidate_k) results [] for idx, dist in zip(indices[0], distances[0]): if candidate_starts[idx] query_start: results.append((idx, candidate_starts[idx], dist)) if len(results) top_k: break return results这里先取candidate_k50个近邻再按时间条件过滤是工程上常用的折中方案。如果过滤后仍然凑不够top_k个候选说明候选片段库中的可用历史太短此时特征填充部分会用 0 值补齐。4.4 增强特征构造检索到候选片段之后需要把候选信息转成增强特征。每个候选片段可以贡献两个信息候选片段的后续走势也就是历史上“类似模式发生之后的结果”。候选片段与查询片段之间的相似度距离用于让生成器知道参考案例的可靠程度。具体代码如下def build_rag_features(query_segment, query_start, retriever, candidate_starts, continuations, window_size24, horizon6, top_k5): 构造检索增强特征 - query_segment: 当前输入窗口 - query_start: 当前输入窗口的起始下标 - retriever: 已拟合的 NearestNeighbors - candidate_starts: 候选片段后续走势的起始位置 - continuations: 候选片段的后续走势 - window_size: 输入窗口长度 - horizon: 预测步长 - top_k: 检索候选数 base_vec build_query_features(query_segment) features list(base_vec) query_vec build_query_features(query_segment) hits safe_retrieve(retriever, query_vec, candidate_starts, query_start, top_ktop_k) used 0 for idx, _, dist in hits: features.extend(continuations[idx]) # 后续走势 features.append(dist) # 相似度距离 used 1 # 如果不足 top_k 个候选用 0 填充保持维度一致 if used top_k: for _ in range(top_k - used): features.extend([0.0] * horizon) features.append(1.0) return np.array(features)这段代码需要注意一个细节初始化时构建了两次build_query_features(query_segment)其实可以优化成一次。在完整脚本中我会把过程整理得更精简这里为了让每一步逻辑清晰所以重复一次。特征维度的计算是这样的基础特征维度为5 window_size每个检索候选额外贡献horizon 1维所以 RAG 特征总维度为5 window_size top_k * (horizon 1)。如果window_size24horizon6top_k5最终维度是5 24 5 * 7 64维。4.5 预测生成器生成器负责把增强特征映射成未来预测值。本文使用随机森林回归器from sklearn.ensemble import RandomForestRegressor model RandomForestRegressor( n_estimators200, max_depth10, min_samples_leaf2, random_state42 )随机森林的优势是不需要特征标准化能处理中等规模表格数据训练速度快适合作为 TS-RAG 的基线生成器。在更复杂的场景中生成器可以替换成LSTM / GRU把检索到的候选片段作为序列输入的一部分适合捕捉时间依赖。Transformer通过注意力机制让模型自动关注不同候选片段的参考价值。时序基础模型如 Chronos、TimesFM 等预训练模型把检索结果作为额外上下文输入。通用 LLM将相似历史模式描述成文本拼接 Prompt 后让大模型生成预测值适合需要自然语言解释的场景。下面的内容中我们先把核心重点放在“检索增强”本身用随机森林做生成器验证 TS-RAG 的完整流程。5. 完整实战用电负荷预测5.1 完整脚本下面给出一个完整可运行的脚本。这个脚本做了四件事生成模拟用电负荷数据。切分训练集和测试集。构造“基线特征”和“RAG增强特征”。分别训练两个随机森林模型并对比效果。# ts_rag_demo.py TS-RAG面向时间序列预测的检索增强生成 Demo 依赖numpy, pandas, scikit-learn import numpy as np from sklearn.neighbors import NearestNeighbors from sklearn.ensemble import RandomForestRegressor from sklearn.metrics import mean_absolute_error, mean_squared_error # ---------- 1. 数据生成 ---------- def generate_load_data(n_days400, points_per_day24, seed42): rng np.random.default_rng(seed) total_points n_days * points_per_day t np.arange(total_points) daily 50 20 * np.sin(2 * np.pi * t / points_per_day - np.pi / 2) seasonal 10 * np.sin(2 * np.pi * t / (total_points * 0.2)) weekly 5 * np.sin(2 * np.pi * t / (7 * points_per_day)) noise rng.normal(0, 2, sizetotal_points) return daily seasonal weekly noise # ---------- 2. 样本与特征构造 ---------- def make_samples(series, window_size24, horizon6): segments [] continuations [] starts [] for start in range(0, len(series) - window_size - horizon 1): seg series[start:start window_size] cont series[start window_size:start window_size horizon] segments.append(seg) continuations.append(cont) starts.append(start window_size) return np.array(segments), np.array(continuations), np.array(starts) def build_query_features(segment): seg np.asarray(segment, dtypenp.float64) mean_val np.mean(seg) std_val np.std(seg) 1e-8 normalized (seg - mean_val) / std_val stats np.array([ mean_val, std_val, np.min(seg), np.max(seg), np.median(seg) ]) return np.concatenate([stats, normalized]) def safe_retrieve(retriever, query_vec, candidate_starts, query_start, candidate_k50, top_k5): distances, indices retriever.kneighbors([query_vec], n_neighborscandidate_k) results [] for idx, dist in zip(indices[0], distances[0]): if candidate_starts[idx] query_start: results.append((idx, candidate_starts[idx], dist)) if len(results) top_k: break return results def build_rag_features(query_segment, query_start, retriever, candidate_starts, continuations, window_size24, horizon6, top_k5): base_vec build_query_features(query_segment) features list(base_vec) query_vec build_query_features(query_segment) hits safe_retrieve(retriever, query_vec, candidate_starts, query_start, top_ktop_k) used 0 for idx, _, dist in hits: features.extend(continuations[idx]) features.append(dist) used 1 if used top_k: for _ in range(top_k - used): features.extend([0.0] * horizon) features.append(1.0) return np.array(features) # ---------- 3. 主流程 ---------- def main(): # 参数设置 window_size 24 horizon 6 top_k 5 # 生成数据 series generate_load_data(seed42) split int(len(series) * 0.8) train_series series[:split] test_series series[split:] # 构造检索库只用训练集避免测试信息泄漏 train_segments, train_continuations, train_starts make_samples( train_series, window_size, horizon ) # 向量化检索库 train_vectors np.array([ build_query_features(s) for s in train_segments ]) retriever NearestNeighbors(n_neighbors50, metriccosine) retriever.fit(train_vectors) # 构造训练特征 X_base_train [] X_rag_train [] Y_train [] for i in range(window_size, len(train_series) - horizon 1): q_start i - window_size q train_series[q_start:i] y train_series[i:i horizon] X_base_train.append(build_query_features(q)) X_rag_train.append( build_rag_features( q, q_start, retriever, train_starts, train_continuations, window_size, horizon, top_k ) ) Y_train.append(y) X_base_train np.array(X_base_train) X_rag_train np.array(X_rag_train) Y_train np.array(Y_train) # 训练两个模型基线 vs TS-RAG base_model RandomForestRegressor( n_estimators200, max_depth10, min_samples_leaf2, random_state42 ) base_model.fit(X_base_train, Y_train) rag_model RandomForestRegressor( n_estimators200, max_depth10, min_samples_leaf2, random_state42 ) rag_model.fit(X_rag_train, Y_train) # 构造测试集特征 X_base_test [] X_rag_test [] Y_test [] for i in range(window_size, len(test_series) - horizon 1): q_start len(train_series) i - window_size q test_series[i - window_size:i] y test_series[i:i horizon] X_base_test.append(build_query_features(q)) X_rag_test.append( build_rag_features( q, q_start, retriever, train_starts, train_continuations, window_size, horizon, top_k ) ) Y_test.append(y) X_base_test np.array(X_base_test) X_rag_test np.array(X_rag_test) Y_test np.array(Y_test) # 预测 pred_base base_model.predict(X_base_test) pred_rag rag_model.predict(X_rag_test) # 评估 mae_base mean_absolute_error(Y_test, pred_base) rmse_base mean_squared_error(Y_test, pred_base) ** 0.5 mae_rag mean_absolute_error(Y_test, pred_rag) rmse_rag mean_squared_error(Y_test, pred_rag) ** 0.5 print(基线模型无检索增强 MAE {:.4f}, RMSE {:.4f}.format(mae_base, rmse_base)) print(TS-RAG 模型检索增强 MAE {:.4f}, RMSE {:.4f}.format(mae_rag, rmse_rag)) print(MAE 相对变化率 {:.2f}%.format((mae_base - mae_rag) / mae_base * 100)) if __name__ __main__: main()5.2 运行方式把上面的代码保存为ts_rag_demo.py在命令行执行python ts_rag_demo.py运行过程中不需要联网也不依赖 GPU普通的笔记本就可以完成。执行后会输出两个模型的评估指标对比。5.3 结果说明由于随机种子和数据生成逻辑固定在你的机器上运行会得到相对稳定的结果。我这里给出一次运行的输出示例实际数值可能会因为 scikit-learn 版本或机器环境略有浮动基线模型无检索增强 MAE 1.4321, RMSE 1.8547 TS-RAG 模型检索增强 MAE 1.2186, RMSE 1.6033 MAE 相对变化率 14.91%从这个输出能看出加入检索增强后 MAE 和 RMSE 都有所下降。这说明随机森林在做出预测时参考了历史上相似模式片段的后续走势确实带来了信息增益。如果你在自己的数据上跑出的提升不明显可能有几个原因历史片段库中缺少足够多的相似模式。片段长度window_size或预测步长horizon设置不合理。生成器太弱或太强掩盖了检索增强的作用。基线模型本身已经非常强。因此TS-RAG 在实际项目中的收益需要结合具体数据验证不要期待所有场景都提升。6. 常见问题与排查思路在实际使用 TS-RAG 的过程中我整理了下面几个高频问题方便你对照排查。问题现象常见原因解决思路特征维度不一致模型训练报错build_rag_features中候选数量不够时没有正确填充检查填充逻辑保证每个样本特征长度相同检索到的候选片段全是最近一段时间的检索库没有做时间多样性限制加入滑动窗口的时间衰减权重或者限制检索时间范围模型在验证集效果很好上线后效果下降检索库使用了未来信息造成数据泄漏严格执行时间安全过滤确保候选片段全部早于查询片段检索太慢无法满足线上延迟使用了暴力最近邻搜索片段库过大切换到近似最近邻索引例如 faiss、annoy、hnswlib检索结果同质化严重Top-K 都长得差不多相似度度量只关注形状忽略了多样性在检索时去重或使用 MMR 多样性重排预测结果波动大不稳定生成器对检索候选过于敏感把相似度距离作为特征传入模型或对候选预测做加权平均冷启动阶段检索库太小没有足够历史数据构建片段库先用传统模型过渡积累一定数据后再启用检索增强关于数据泄漏我再多说一句。这是 TS-RAG 最容易踩的坑。构建检索库时只能使用“查询时刻之前”的历史数据。如果测试集数据被提前放进了索引模型在测试时会检索到“未来的答案”效果虚高上线后大概率翻车。一个安全的做法是在训练和评估时都显式传入query_start并且只允许candidate_starts query_start的候选进入结果集。我在上面的safe_retrieve函数中已经做了这件事你可以直接把这段逻辑复制到自己的项目中。7. 最佳实践与工程建议如果要把 TS-RAG 真正落地到业务系统下面这些建议值得参考。7.1 时间序列切分策略窗口大小window_size要和业务预测周期匹配。如果预测未来 24 小时输入窗口至少应该包含 24 小时甚至更长的上下文。滑动步长也不一定固定为 1大规模场景下步长太大会丢失样本步长太小会增加索引体积。实际操作中可以从下面几个维度交叉验证输入窗口覆盖至少 2 到 3 个业务周期。预测步长等于实际业务需要的提前期。片段库更新时间定期追加最新数据删除过时数据。7.2 向量化与相似度度量我在示例中使用了统计特征加归一化序列值的简单向量化方式优点是稳定、可解释、无需训练。如果序列形态复杂可以换成