欢迎来到图解强化学习的世界博客主页卿云阁欢迎关注点赞收藏⭐️留言首发时间2026年4月12日✉️希望可以和大家一起完成进阶之路作者水平很有限如果发现错误请留言轰炸哦万分感谢目录初始化函数加载历史缓冲区保存在缓冲区增加一条新的样本微调模型模拟运行初始化函数def __init__(self, trainer, processor, buffer_file, model_file, max_days, update_every):self.trainer trainer self.processor processor将之前训练好的trainer和processor移交给在线更新器。self.buffer_file buffer_file self.model_file model_file指定缓冲区文件和模型文件在硬盘上的位置。self.update_every update_every # 默认为 30 self.buffer deque(maxlenmax_days) # 默认为 365设定了每积攒30 天的数据才进行一次微调。deque 是双端队列设置了 maxlen 后当存入第366 天的数据时它会自动把第 1 天的数据给“吐”出来。self._load()在对象创建的一瞬间立刻去硬盘里把之前保存的online_buffer.jsonl全部读进内存。加载历史缓冲区def _load(self): 加载历史缓冲区 if os.path.exists(self.buffer_file): with open(self.buffer_file, r, encodingutf-8) as f: for line in f: try: self.buffer.append(json.loads(line)) except: pass print(f✓ 加载 {len(self.buffer)} 条历史样本)假设你的buffer_file即online_buffer.jsonl在硬盘里长这个样子只有 3 行{date: 2025-03-01, reward: 0.5, action: 2} {date: 2025-03-02, reward: 0.8, action: 1} {date: 2025-03-03, reward: -0.2, action: 4}self.buffer deque(maxlen2) 假设我们为了演示记忆长度只设为 2读取第 1 行line {date: 2025-03-01, reward: 0.5, action: 2} 执行 append。当前内存 buffer[{date: 3-01, ...}] (长度 1)读取第 2 行line {date: 2025-03-02, reward: 0.8, action: 1} 执行 append。当前内存 buffer[{date: 3-01, ...}, {date: 3-02, ...}] (长度 2已满)读取第 3 行关键点line {date: 2025-03-03, reward: -0.2, action: 4}\n执行 append。内存变化因为 maxlen2队列会自动弹出最老的一行3-01。最终内存 buffer[{date: 3-02, ...}, {date: 3-03, ...}] (长度 2)最后输出 ✓ 加载 2 条历史样本self.buffer最后的样子如下deque([ {date: 2025-01-01, state: [...], action: 2, ...}, {date: 2025-01-02, state: [...], action: 1, ...}, ... ], maxlen365)保存在缓冲区def _save(self): 保存缓冲区 with open(self.buffer_file, w, encodingutf-8) as f: for item in self.buffer: f.write(json.dumps(item, ensure_asciiFalse) \n)执行前内存中的 self.buffer 状态条目 1{date: 2025-03-15, reward: 0.5}条目 2{date: 2025-03-16, reward: 0.9}硬盘上原本旧的文件内容{date: 2024-03-14, reward: 0.2} {date: 2025-03-15, reward: 0.5}执行后硬盘上的新文件内容{date: 2025-03-15, reward: 0.5} {date: 2025-03-16, reward: 0.9}增加一条新的样本def add_sample(self, date, state, pac_dose, eff_tp_next): 添加一条新样本 参数: date : 日期字符串 2025-03-16 state : 状态字典 {inf_tp, inf_ph, inf_temp, flow_rate_10kt_d, eff_tp} pac_dose : 今日执行的PAC量 (kg/h) eff_tp_next : 次日出水TP (mg/L) # 计算奖励 prior lookup_prior(self.processor.lookup, state[inf_tp], state[inf_temp]) reward compute_reward_causal(eff_tp_next, pac_dose, state[inf_tp], prior, self.processor.pac_mean) # 归一化状态 s self.processor.transform_state(state).tolist() s_next self.processor.transform_state({**state, eff_tp: eff_tp_next}).tolist() # 动作索引 action int(np.argmin(np.abs(PAC_ACTIONS - pac_dose))) prior_action int(np.argmin(np.abs(PAC_ACTIONS - prior))) # 保存样本 sample { date: date, state: s, action: action, reward: float(reward), next_state: s_next, done: 0, prior_action: prior_action, raw: {pac: pac_dose, eff_tp: eff_tp_next} } self.buffer.append(sample) self._save() # 判断是否微调 if len(self.buffer) % self.update_every 0 and len(self.buffer) self.update_every: print(f\n[{date}] 累计 {len(self.buffer)} 条样本触发微调...) self.finetune() return {微调: True, 奖励: round(reward, 3)} return {微调: False, 奖励: round(reward, 3)}date 2025-03-16state {inf_tp: 3.0, inf_ph: 7.0, inf_temp: 20, flow_rate_10kt_d: 10, eff_tp: 0.25}pac_dose 15.8eff_tp_next 0.22prior lookup_prior(self.processor.lookup, state[inf_tp], state[inf_temp]) reward compute_reward_causal(eff_tp_next, pac_dose, state[inf_tp], prior, self.processor.pac_mean)prior 14.0reward 0.85# 归一化状态 s self.processor.transform_state(state).tolist() s_next self.processor.transform_state({**state, eff_tp: eff_tp_next}).tolist()s [0.2, -0.5, 0.1, 0.8, -0.2]s_next [0.2, -0.5, 0.1, 0.8, -0.4]action int(np.argmin(np.abs(PAC_ACTIONS - pac_dose))) prior_action int(np.argmin(np.abs(PAC_ACTIONS - prior)))实际动作索引 actionaction 4先验动作索引 prior_action prior_action 3# 保存样本 sample { date: date, state: s, action: action, reward: float(reward), next_state: s_next, done: 0, prior_action: prior_action, raw: {pac: pac_dose, eff_tp: eff_tp_next} } self.buffer.append(sample) self._save()构造 sample 字典{ date: 2025-03-16, state: [0.2, -0.5, 0.1, 0.8, -0.2], action: 4, reward: 0.85, next_state: [0.2, -0.5, 0.1, 0.8, -0.4], prior_action: 3, ... }执行 self.buffer.append(sample)将这条数据塞进 365 天的记忆队列。执行 self._save()立刻把这条热乎的数据写进硬盘的 .jsonl 文件中防止停电丢失。len(self.buffer) 刚好达到了30条 调用 self.finetune()返回结果{微调: True, 奖励: 0.85}微调模型# 构建数据集 buf list(self.buffer) dataset { states: np.array([b[state] for b in buf], dtypenp.float32), actions: np.array([b[action] for b in buf], dtypenp.int64), rewards: np.array([b[reward] for b in buf], dtypenp.float32), next_states: np.array([b[next_state] for b in buf], dtypenp.float32), dones: np.array([b[done] for b in buf], dtypenp.float32), prior_actions: np.array([b[prior_action] for b in buf], dtypenp.int64), }假设 buffer 里只有 2 天的数据内存中的 buf{state: [0.1, 0.2], action: 2, reward: 0.8, ...}{state: [0.3, 0.4], action: 1, reward: 0.5, ...}执行这段代码后dataset 变成了dataset[states] [[0.1, 0.2], [0.3, 0.4]] 矩阵dataset[actions] [2, 1] 向量dataset[rewards] [0.8, 0.5] 向量模拟运行Row 0 (今天 i0): date04-01, inf_tp3.0, eff_tp0.25, inf_temp20Row 1 (明天 i1): date04-02, inf_tp3.2, eff_tp0.21today df_test.iloc[i] tomorrow df_test.iloc[i1] state {c: today[c] for c in CFG[state_cols]}state {inf_tp: 3.0, eff_tp: 0.25, ...}s_norm processor.transform_state(state) prior lookup_prior(processor.lookup, today[inf_tp], today[inf_temp]) rec trainer.recommend(s_norm, today[eff_tp], prior) pac rec[推荐PAC量]对当天的状态标准化处理得到s_norm查询当天状态的priorpac模型给出的推荐的加药量eff_tp_next tomorrow[eff_tp]# 添加样本 info updater.add_sample( datestr(today[date])[:10], statestate, pac_dosepac, eff_tp_nexteff_tp_next )