欢迎来到图解强化学习的世界博客主页卿云阁欢迎关注点赞收藏⭐️留言首发时间2026年3月17日✉️希望可以和大家一起完成进阶之路作者水平很有限如果发现错误请留言轰炸哦万分感谢目录初始配置模块一数据与环境重构DataProcessortransform_state其它函数【1】分层历史最优查询表安全先验【2】查询当前工况的历史安全先验PAC量【3】因果增强奖励函数Dueling Q网络CausalCQLTrainer[1]第一部分【2】实战部分初始配置因果参数 (CAUSAL)causal不管历史数据怎么骗你我确切地告诉你每多加1公斤PAC出水总磷TP就能下降大约0.00609。今天加的药明天才能看到效果。CAUSAL { pac_to_tp: 0.00609, # 每1 kg/h PAC → eff_tp变化(mg/L) lag_days: 1, # PAC效应在次日体现 n_natural_exp: 253, # 自然实验样本数 }配置字典CFG cfg这个是一份全局配置文件 (CFG)里面存放着相关的参数下面我们具体看看有那些参数。状态与动作 (State Action)state_cols: [inf_tp, inf_ph, inf_temp, flow_rate_10kt_d, eff_tp],强化学习中的“状态State”。这是 AI 每次决定加药前允许它看的数据仪表盘。代表今天水质有多恶劣这是客观存在的。eff_tp是当天的出水。pac_min: 2.0, pac_max: 30.0, action_bins: 20,强化学习中的“动作空间Action Space”。如果让 AI 直接输出连续数字比如 15.342kg/h模型极难收敛。所以代码在最底部用了 np.linspace把 2.0 到 30.0均匀切成了 20 个“档位”2.0, 3.47, 4.94 ... 30.0。AI 只需要做一个 20 选 1 的选择题今天挂第几档。奖励函数 (Reward Function)环保局规定的出水 TP 上限假设是 0.30。一旦预测到超过这个值直接给 AI 极其严厉的惩罚扣 100 分。tp_target: 0.30, # 超过此值惩罚-100 r_exceed: -100.0,安全黄线缓冲区。如果出水在 0.25 到 0.30 之间说明虽然没超标但在危险边缘疯狂试探。扣 10 分。tp_alert: 0.25, # 超过此值惩罚-10 r_alert: -10.0,每多加 1 档或 1kgPAC 药剂扣 0.5 分。这让 AI 在确保水质安全不被扣100分的前提下变得极其“抠门”拼命寻找刚好达标且用药最少的平衡点。这就实现了“节能降耗”。drug_penalty: 0.5, # 药耗惩罚系数CQL 超参数gamma: 0.95, # 折扣因子0.95 意味着 AI 在做今天的决定时不仅看今天的得分还会考虑今天加的药对明后天水质的长远影响因为水处理有大滞后性。如果是 0AI 就只管今天不管明天死活。alpha_cql: 1.0, # CQL保守正则强度alpha_cql 就是用来打压这种盲目自信的。设为 1.0代表模型极度保守。它告诉 AI“历史数据里没试过的操作一律按最坏的结果扣分绝对不许在真实水厂里瞎尝试”神经网络超参数lr: 3e-4, # 学习率 batch_size: 128, # 批次大小 epochs: 200, # 训练轮数 hidden_dim: 128, # 隐藏层维度在线更新 (Online MLOps)update_freq: 30, # 每30天微调一次真实的水厂会随着季节变化水泵叶轮也会老化。如果你用 2022 年的数据训出一个模型用到 2025 年肯定不准了这叫模型漂移。模型上线后每积累30 天的新鲜真实运行数据就在后台自动启动“微调模式”让 AI 适应最新的水厂工况。动作空间PAC_ACTIONS np.linspace(CFG[pac_min], CFG[pac_max], CFG[action_bins])CFG[pac_min] 2.0 起点CFG[pac_max] 30.0 终点CFG[action_bins] 20 一共要切成多少个点[ 2. 3.47368421 4.94736842 6.42105263 7.89473684 9.36842105 10.84210526 12.31578947 13.78947368 15.26315789 16.73684211 18.21052632 19.68421053 21.15789474 22.63157895 24.10526316 25.57894737 27.05263158 28.52631579 30. ]模块一数据与环境重构强化学习的本质是智能体AI观察当前环境状态State做出一个动作Action然后环境反馈一个奖励Reward并进入下一个状态Next State。DataProcessor这个类就是专门用来把 water_final.csv 翻译成 AI 能听懂的“强化学习语言”的。加载与时序清洗加载与清洗数据把表格里的“date”列从普通的文字字符串转换成 Python 能看懂的“时间对象”严格按照时间先后顺序把所有数据重新排一次队检查我们最关心的那几个特征列和加药量pac_dose。# ── 加载数据 ────────────────────────────────────────────────── df pd.read_csv(data/water_final.csv, encodingutf-8-sig) df[date] pd.to_datetime(df[date], formatmixed) df df.sort_values(date).reset_index(dropTrue) df df.dropna(subsetCFG[state_cols] [pac_dose]).reset_index(dropTrue) print(f数据{len(df)} 行 ({df[date].min().date()} ~ {df[date].max().date()}))在原始数据上划分训练集和测试集n_train int(len(df)*0.8) df_train df.iloc[:n_train].copy() df_test df.iloc[n_train:].copy()processor DataProcessor()调用这个类DataProcessorprocessor 是类对象。dataset processor.build_dataset(df_train)调用对象 (Instance)processor即 DataProcessor 类的实例化对象。调用方法 (Method)执行 DataProcessor 类内部的 build_dataset(self, df: pd.DataFrame) 方法。传入参数 (Input)df_train一个包含过去 80% 连续时间序列数据的 Pandas DataFrame。DataProcessor这是一个Python类名字是DataProcessor。有4个类属性和两个类方法。4个类属性分别是self.scaler None # 归一化器负责统一数据量纲 self.pac_mean None # 历史PAC平均用量 self.inf_tp_mean None # 历史进水TP平均浓度 self.lookup None # 历史先验查询表老厂长经验库还有两个类方法build_dataset排队与提取“环境状态” (State)函数的输入是pd.DataFrame对象sort_values:强行按日期把数据排好队sc:拿出我们关心的5个环境指标进水TP、流量、出水TP等S_raw:把这 3 天的这 5 个指标抠出来变成一个 3 *5 的纯数字矩阵。现在S_raw的第一行就是 [3.0, ..., 0.28]代表第1天的原始环境。df df.sort_values(date).reset_index(dropTrue) sc CFG[state_cols] S_raw df[sc].values.astype(np.float32)归一化统一量纲防崩溃进水TP [3.0, 4.0, 2.5] 经过 transform 后可能变成了类似 [0.1, 1.2, -0.9] 的数字。self.scaler StandardScaler().fit(S_raw) S_norm self.scaler.transform(S_raw)第三步计算全局统计与“老厂长经验”出历史平均加药量比如这三天平均是15.0。然后生成那张我们之前讲过的lookup历史既达标又省药的 P25 经验表。self.scaler StandardScaler().fit(S_raw) self.pac_mean float(df[pac_dose].mean()) self.inf_tp_mean float(df[inf_tp].mean()) self.lookup build_lookup_table(df)第四步把连续药量变成“选择题”动作离散化 Actionactions np.array([np.argmin(np.abs(PAC_ACTIONS - p)) for p in df[pac_dose].values])PAC_ACTIONS里面的数据如下所示假设此时的p是1列表中的每一个数据都会和1做差然后取差值的最小值的索引。比如这里的索引显然是0号位置。假设你的水处理系统有以下设定标准动作池 (PAC_ACTIONS): [0, 10, 20, 30] (单位mg/L)实际记录数据 (df[pac_dose]): [11.2, 28.5, 2.1]执行逻辑如下第一步处理第一个数据 11.2计算差值| [0, 10, 20, 30] - 11.2 | $\rightarrow$ [11.2, 1.2, 8.8, 18.8]找最小值最小的是 1.2。找索引1.2 在列表中的位置是 1对应标准值 10。第二步处理第二个数据 28.5计算差值| [0, 10, 20, 30] - 28.5 | $\rightarrow$ [28.5, 18.5, 8.5, 1.5]找最小值最小的是 1.5。找索引1.5 在列表中的位置是 3对应标准值 30。第三步处理第三个数据 2.1计算差值| [0, 10, 20, 30] - 2.1 | [2.1, 7.9, 17.9, 27.9]找最小值最小的是 2.1。找索引2.1 在列表中的位置是 0对应标准值 0。最终结果actions 将会是 np.array([1, 3, 0])。[ 2. 3.47368421 4.94736842 6.42105263 7.89473684 9.36842105 10.84210526 12.31578947 13.78947368 15.26315789 16.73684211 18.21052632 19.68421053 21.15789474 22.63157895 24.10526316 25.57894737 27.05263158 28.52631579 30. ]prior_doses扫描结束后会得到一个长长的列表里面装满了老厂长给出的具体加药公斤数连续值。比如[12.5, 14.0, 9.54, 15.2, ...]。prior_doses df.apply( lambda r: lookup_prior(self.lookup, r[inf_tp], r[inf_temp]), axis1)把“公斤数”翻译成“机器档位”离散化映射prior_actions np.array([np.argmin(np.abs(PAC_ACTIONS - p)) for p in prior_doses])输出结果 prior_actions刚才那个装满公斤数的列表瞬间变成了一串纯整数构成的索引数组[7, 8, 4, 9, ...]。# 因果增强奖励 rewards np.array([ compute_reward_causal(row[eff_tp], row[pac_dose], row[inf_tp], prior_doses.iloc[i], self.pac_mean) for i, (_, row) in enumerate(df.iterrows()) ], dtypenp.float32)它最终的样子大概是这样的[ 0.34, -2.50, 1.20, -10.0, 0.55, ... ]# 下一状态 df_s2 df.copy() for c in sc: df_s2[c] df[c].shift(-1) df_s2 df_s2.ffill() S2_norm self.scaler.transform(df_s2[sc].values.astype(np.float32)) dones np.zeros(len(df), dtypenp.float32) dones[-1] 1.0 return {states: S_norm.astype(np.float32), actions: actions.astype(np.int64), rewards: rewards, next_states: S2_norm.astype(np.float32), dones: dones, prior_actions: prior_actions.astype(np.int64)}假设我们的历史数据表 df 总共有3000天的记录行数 N 3000并且我们用来描述水质状态的特征比如进水总磷、水温、流量等总共有5个特征数 F 5。字典里的名字NumPy 数组的形状 (Shape)相当于 Excel 里的什么样子states(3000, 5)一个二维表格 (矩阵)。包含 3000 行5 列。记录了每天的 5 项水质指标。⚙️actions(3000,)一个一维数组 (单列向量)。只有一列包含 3000 个整数每天的加药档位。rewards(3000,)一个一维数组。只有一列包含 3000 个小数每天操作的得分。⏩next_states(3000, 5)一个二维表格。大小和states完全一样装的是“明天”的 5 项水质指标。dones(3000,)一个一维数组。只有一列前 2999 个是0.0最后一个是1.0。prior_actions(3000,)一个一维数组。只有一列包含 3000 个整数老厂长建议的档位。transform_statedef transform_state(self, d: dict) - np.ndarray: s np.array([d[c] for c in CFG[state_cols]], dtypenp.float32) return self.scaler.transform(s.reshape(1,-1))[0]假设系统配置 CFG[state_cols]规定 AI 只需要看两个指标[inf_tp, inf_temp]进水总磷、水温。而今天传感器传来的实时原始数据 d 是一大坨信息{inf_tp: 4.0, inf_temp: 20.0,operator: 张三, pump_status: ON}。 代码根据配置表精准地从字典 d 中只把 inf_tp (4.0)和 inf_temp (20.0) 抽了出来过滤掉无关的“张三”和水泵状态。结果 得到了一个一维数组 [4.0,20.0]。[4.0, 20.0] 变成了 [[4.0, 20.0]]。[[4.0, 20.0]] 可能被转换成了 [[0.5, -1.2]]。最终输出 [0.5,-1.2]其它函数【1】分层历史最优查询表安全先验输入历史数据整理成20个工况的记录表build_lookup_table(df)首先看一下这个函数的输入是df是一个表格文件。dfdf.copy()这是一个良好的编程习惯。我们在函数里要做很多切割、修改操作为了不把外面原本的表格弄乱先复印copy一份在里面随便折腾。df df.copy()数据分箱 Binningpd.cut 的功能是把连续的小数切分成几个固定的“区间”。水质和水温每天都在变如果精确到小数点AI 就找不到历史上完全一模一样的一天了。所以我们要“模糊化”处理。进水TP 3.0会被划入箱子(2.5, 3.5]表示正常偏高浓度。进水温度 18℃ 会被划入箱子(15, 20]表示春秋季常温。现在这 10 天的数据都被贴上了一个统一的标签组合“TP在2.5-3.5之间 温度在15-20之间”。df[bin_tp] pd.cut(df[inf_tp], bins[0,2.5,3.5,5,8,20]) df[bin_temp] pd.cut(df[inf_temp], bins[0,15,20,25,50])此时 df 的输出模样在末尾多了两列标签第 1~7 天的进水TP都在 2.5~3.5 之间水温都在 15~20 之间。所以它们都被贴上了 (2.5, 3.5] 和(15, 20] 的标签。第 8 天被贴上了 (5.0, 8.0] 和 (0, 15] 的标签。此时定义一个空列表rows []日期bin_tp (进水TP箱)bin_temp (水温箱)eff_tp (出水)pac_dose (加药量)达标情况工况A(2.5, 3.5](15, 20]0.2010✅ 达标工况A(2.5, 3.5](15, 20]0.2212✅ 达标工况A(2.5, 3.5](15, 20]0.2514✅ 达标工况A(2.5, 3.5](15, 20]0.2616✅ 达标工况A(2.5, 3.5](15, 20]0.2718✅ 达标工况A(2.5, 3.5](15, 20]0.358❌ 超标工况A(2.5, 3.5](15, 20]0.406❌ 超标工况A(2.5, 3.5](15, 20]0.329❌ 超标------------------工况B(3.5, 5.0](15, 20]0.2125✅ 达标工况B(3.5, 5.0](15, 20]0.2528✅ 达标工况B(3.5, 5.0](15, 20]0.2630✅ 达标假设我们的输入是上面这种 第 1 轮循环打开【工况A 文件夹】此时bt (2.5, 3.5]btemp (15, 20]grp包含上面的前 8 行数据。第 1 步剔除坏数据系统检查这 8 行数据把出水大于 0.28 的那 3 行加药 8、6、9 的三天全部扔掉。ok表格里现在只剩下 5 行达标数据对应的加药量是[10, 12, 14, 16, 18]。检查样本数量(每个组的样本数量大于5数学计算与打包记录算出每个工况算 pac_median (中位数) [10, 12, 14, 16, 18] 最中间的那个数是14.0。算 pac_prior (P25 分位数) 排在前面 25% 位置的数字。在这个数组中计算结果是12.0。rows里面只有一条记录来自工况Apd.DataFrame把这堆字典变成了一个漂亮的表格这也就是self.lookup最后保存的内容inf_tp_loinf_tp_hitemp_lotemp_hipac_priorpac_mediann_okok_rate2.53.515.020.012.014.050.625【2】查询当前工况的历史安全先验PAC量输入工况表当前的进水TP进水温度输出一个节药的加药量lookup_prior(table: pd.DataFrame,inf_tp: float, inf_temp: float) - float: 场景设定我们手里的那本秘籍假设传入的table即self.lookup长这样行号inf_tp_lo (TP下限)inf_tp_hi (TP上限)temp_lo (温度下限)temp_hi (温度上限)pac_prior (老厂长药量)n_ok (成功天数)第0行2.53.515.020.012.05第1行3.55.015.020.020.010假设今天水厂的真实进水是inf_tp 3.0inf_temp 18.0扫描第 0 行2.5 3.0 3.5 (真) 且 15.0 18.0 20.0 (真)。这行匹配成功扫描第 1 行3.5 3.0 5.0 (假)。匹配失败。最终 mask 是一串结果[True, False]。mask ((table[inf_tp_lo] inf_tp) (table[inf_tp_hi] inf_tp) (table[temp_lo] inf_temp) (table[temp_hi] inf_temp))把刚才 mask 里结果是 True 的那几行数据真正地从大表里抽出来存进叫 hits 的小表格里。例子体现 此时 hits 里面就只有第 0 行的数据了老厂长建议加药 12.0背后有 5 天成功经验支撑。hits table[mask]安全兜底防线找不到怎么办if len(hits) 0: return 9.54 # 回退到全局均值现实工业环境极其复杂。万一今天下暴雨进水 TP 突增到了 12.0爆表了而我们之前编纂的秘籍里根本没有这么恶劣工况的记录查不到任何匹配的行hits 为空。怎么办程序不能报错崩溃解决办法 一旦查不到就返回一个写死的“全局安全均值”9.54。 避坑提示 这个 9.54 是作者针对他们水厂写死的代码。在你实际应用时最好把它改成动态的全局均值比如传进来的 self.pac_mean或者直接改成一个对你水厂相对安全的常量比如30.0防止没查到数据时加药太少导致翻车。加权平均大招处理多个命中项理论上我们的区间是不重叠的hits 应该永远只有 1 行。但作为工业级代码必须防御性编程。万一以后修改了分箱逻辑导致两个区间的边缘有重叠今天的数据同时命中了 2 条历史规律怎么办按成功经验投票加权 假设命中了规律 A有 10 天成功经验建议加 15kg和规律 B有 40 天成功经验建议加 20kg。权重 w 怎么算规律 A 的话语权是 10 / (1040) 20规律 B 的话语权是 40 / 50 80\%。最终返回的药量就是15*20 20*80\% 19.0 kg。在我们的例子中hits 只有第 0 行。所以权重 w 5 / 5 1.0100%听它的。最终返回的数值就是 12.0*1.0 12.0。# 按样本数加权平均 w hits[n_ok] / hits[n_ok].sum() return float((hits[pac_prior] * w).sum())【3】因果增强奖励函数compute_reward_causal 输入 (Inputs)函数接收 5 个参数代表了当时的工况环境和具体操作 eff_tp (出水总磷)处理后的水质指标代表最终结果。 pac_dose (实际加药量)这一回合实际投入的药剂数量。 inf_tp (进水总磷)处理前的原水水质代表这一天的“初始任务难度”。 pac_prior (历史先验经验)老厂长在同样工况下建议的安全药量。 pac_mean (全局平均加药量)一个常数默认为 9.54主要用来作为计算超量比例的分母。 输出 (Output)float函数最后返回一个带有正负号的小数浮点数。这个数字就是这一回合 AI 拿到的最终奖励总分Reward。最终得分 基础达标奖励 (base) 因果预期奖励 (expected_benefit) - 超量惩罚 (drug_penalty)Dueling Q网络​ # ══════════════════════════════════════════════════════════════════ # Dueling Q网络同前 # ══════════════════════════════════════════════════════════════════ class DuelingQNetwork(nn.Module): def __init__(self, state_dim: int, action_dim: int, hidden: int 128): super().__init__() self.backbone nn.Sequential( nn.Linear(state_dim, hidden), nn.LayerNorm(hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.LayerNorm(hidden), nn.ReLU(), ) self.value_stream nn.Sequential( nn.Linear(hidden, hidden//2), nn.ReLU(), nn.Linear(hidden//2, 1)) self.advantage_stream nn.Sequential( nn.Linear(hidden, hidden//2), nn.ReLU(), nn.Linear(hidden//2, action_dim)) def forward(self, x): f self.backbone(x) V self.value_stream(f) A self.advantage_stream(f) return V (A - A.mean(dim1, keepdimTrue)) ​假设今天传感器传来的数据张量是 x [[4.0, 20.0]]形状是1行 x 2列。让我们跟着它穿过网络的代码层输出动作 (Action):只有 2 个加药档位比如[加5kg, 加10kg]。隐藏层 (Hidden):只有 3 个神经元方便我们看具体的数字。1️⃣ 骨干网络 self.backbone (特征提取)它负责把原始的“物理指标”变成大脑能理解的“抽象特征”。nn.Linear(2, 3) (线性层): * 输入: x [[4.0, 20.0]] (形状 1x2)作用: 通过内部的权重矩阵相乘并加上偏置像解多元一次方程。输出: 假设计算后变成了 [[1.5, -2.0, 3.0]] (形状 1x3)。nn.LayerNorm(3) (归一化层):输入: [[1.5, -2.0, 3.0]]作用: 把这组数字按比例压缩拉伸防止有的数字太大、有的太小。输出: 假设变成了 [[0.2, -1.2, 1.0]]。nn.ReLU() (激活函数):输入: [[0.2, -1.2, 1.0]]作用: 规则极度简单——把所有负数变成 0正数保持不变。输出: 变成了 [[0.2, 0.0, 1.0]]。这就是最终提取出的特征 f2️⃣价值流 self.value_stream (环境评分 V它拿到了特征 f [[0.2, 0.0, 1.0]]继续往下算经过它自己的 Linear 和 ReLU 后最后一层 nn.Linear(..., 1) 强制输出 1 个数字。输出: 假设V [[8.0]]。意思是“今天这个水质底子不错保底能拿 8 分”。3️⃣ 优势流 self.advantage_stream (动作加分 A)它也拿到了同样的特征 f [[0.2, 0.0, 1.0]]经过最后一层 nn.Linear(..., 2) 强制输出 2 个数字对应 2 个动作。输出: 假设A [[2.0, -1.0]]。意思是“相比平均水平加 5kg 能多赚 2 分加 10kg 会倒扣 1 分”。4️⃣ 最终融合 (forward 的最后一行)计算公式Q V (A - A的平均值)A 的平均值 (2.0 (-1.0)) / 2 0.5调整后的优势 [2.0 - 0.5, -1.0 - 0.5] [[1.5, -1.5]]最终 Q 值 [[8.0]] [[1.5, -1.5]] [[9.5, 6.5]]CausalCQLTrainer[1]第一部分实例化主 Q 网络 (self.q_net)它负责接收当前的水质状态进水 TP、流量等并计算出每个加药动作对应的预期收益Q 值。我们在上一段代码中讨论的 Dueling 架构就在这里运行。创建目标网络 (self.q_target)q_target保持在一个旧的版本提供稳定的预测参考主网络每隔一段时间才去对齐它。设置优化器 (self.optimizer)和学习率。class CausalCQLTrainer: def __init__(self, state_dim, action_dim): self.q_net DuelingQNetwork(state_dim, action_dim, CFG[hidden_dim]) self.q_target copy.deepcopy(self.q_net) self.optimizer optim.Adam(self.q_net.parameters(), lrCFG[lr], weight_decay1e-4) self.scheduler optim.lr_scheduler.CosineAnnealingLR( self.optimizer, T_maxCFG[epochs])假设污水厂在 2026年3月22日 有这样一条真实记录今日水质 (S)进水总磷 3.2 mg/L水温 22°C。今日加药 (A)人工实际加了 10kg/h 的药对应模型里的第 8 号档位。明日水质 (S2)加药后的第二天出水总磷降到了 0.21 mg/L达标。奖励分数 (R)因为降磷效果好且没浪费系统给这次操作打了 0.8 分。老专家经验 (PA)查历史表发现类似这种水质历史上最省药的成功案例是加药 9.5 kg/h对应第 7 号档位。相当于把 64 天的记录打包成一个箱子epochs epochs or CFG[epochs] S torch.FloatTensor(dataset[states]) A torch.LongTensor(dataset[actions]) R torch.FloatTensor(dataset[rewards]) S2 torch.FloatTensor(dataset[next_states]) Done torch.FloatTensor(dataset[dones]) PA torch.LongTensor(dataset[prior_actions]) # 历史先验动作TD目标Double QS2是明天的状态送进网络中会得到20个动作的打分。取最大动作的索引输入S2输出动作2。self.q_target(s2明天的预测工况 s2一个矩阵包含所有 20 档加药动作的分数。[[0.1, 0.4,0.9, 0.3, ...]]假设第 3 个数最高它是药量档位。best_a.unsqueeze(1)主大脑刚才选出的最好动作编号。比如 2。把标量 2 变成 [[2]]。.gather(1, ...)(精准提取)从第一步那 20 个分数里把第 best_a 个分数“抠”出来。从 [0.1, 0.4, 0.9, 0.3] 中提取第 3 个数得到 [[0.9]].squeeze(1) (压扁去皮)纯粹的数字0.9。td_tgt动作价值函数“td_tgt 是根据贝尔曼方程算出来的、动作价值函数在当前状态下理应达到的目标值。”q_all self.q_net(s)当前污水厂的工况数据进水 TP、流量、pH、水温等。假设是一个经过标准化的向量如 [0.5,-0.2, 1.1, ...]。如果你把加药量分成了 20 个档位2kg 到 30kg那么q_all 就是一个长度为 20的列表。[0.12, 0.25, 0.55, ..., 0.88, ..., 0.31]。a.unsqueeze(1)把 2 包装成 [[2]] .gather(1, ...)在 q_all 的第一维度横向搜索根据索引 [[2]]抓取数值。[[0.8]] .squeeze(1)把 [[0.8]] 这种嵌套的括号去掉。纯粹的数字 0.8。q_taken给那天那组【特定入水工况】下人工选择的那个【特定投药量】打个分with torch.no_grad(): # 1. 问主大脑明天(s2)该怎么办 best_a self.q_net(s2).argmax(dim1) # 输入 s2 - 输出 动作11 # 2. 问影子大脑如果明天做动作11能得几分 q2 self.q_target(s2).gather(1, best_a.unsqueeze(1)) # 输入 s211 - 输出 0.9分 # 3. 计算总目标分 (今天得的 明天预期的) td_tgt r CFG[gamma] * q2 * (1 - done) # 输入 0.8 0.9*0.9 - 输出 1.61 q_all self.q_net(s) # 输入 s - 输出 20个动作的分数矩阵 [0.2, ..., 0.7(动作10), ...] q_taken q_all.gather(1, a.unsqueeze(1)) # 输入 动作10 - 输出 0.7分 td_loss nn.HuberLoss()(q_taken, td_tgt) # 输入 0.7 vs 1.61 - 输出 误差 0.91结合当天的真实奖惩r与对明天的期待q2算出一个复盘标杆分td_tgt然后将其与 AI 当时的主观预判分q_taken进行对比计算出误差td_loss从而驱动 AI 学习“如何更准地评估加药价值”。CQL保守项对未历史的动作压低Q值log_sum torch.logsumexp(q_all, dim1).mean()q_all [0.5, 1.2, 1.5]torch.logsumexp(q_all, dim1)2.25模型对当前环境下所有可能路径的“野心”或“总潜力”的平滑最大值。q_taken [0.8, 1.2, 0.7] 0.9根据模型目前的记忆它觉得历史上那些真实发生过的、老工人们干过的活儿平均能值多少分。当前工况进水总磷 3.0。历史经验 (q_hist)历史记录里这种情况下加 10kg 药表现不错AI 给它打 1.0 分。AI 的野心 (log_sum)AI 突然产生幻觉觉得如果只加 2kg 药历史没试过可能会省大钱于是给 2kg 偷偷打了个 2.5 高分。这导致 log_sum 被拉高到了 2.8。保守系数 (alpha_cql)代码里设为 0.3。先验引导项q_prior q_all.gather(1, pa.unsqueeze(1)).squeeze(1)q_prior 代表了AI 目前对“专家经验”的认可程度。q_prior.detach() 0.5我们取 AI 现在的评分强行加上0.5【2】实战部分self.q_net.eval() with torch.no_grad(): q self.q_net(torch.FloatTensor(state_norm).unsqueeze(0)).squeeze(0).numpy() best_idx int(np.argmax(q)) best_dose float(PAC_ACTIONS[best_idx])【3】 recommendstate_norm [[0.5, -1.2, 0.8, 1.5, -0.3]] 1 条样本 5 个特征q[[-2.5, 4.1, 9.8, 1.2, -5.0]]best_idx 2best_dose 16.0current_eff_tp CFG[tp_alert]0.295 0.280.295 - 0.28 0.015urgency 0.75代表紧急程度min_dosenp.percentile(PAC_ACTIONS, 62.5) 19.5从这个百分位去做best_dose max(best_dose, min_dose)max(5.0, 19.5)19.5 AI和之前经验的最大值best_dose20 explore_dose20.0 * (1 - 0.05) 19.0 kg/h实战中根据当前的状态inf_tp (进水总磷) inf_ph (进水酸碱度) inf_temp (进水温度) flow_rate_10kt_deff_tp (当前的出水总磷)。通过一个网络给20个动作打分得到一个得分最高的动作。如果当前的出水TP比较大就增大药剂的投放。如果当前的出水TP非常小就减少药剂的投放。