简介《智慧城市低成本方案DeepSeek-V3微调实现交通流量预测》是一份面向智慧城市、智能交通及大模型应用开发者的PDF技术文档。资源包为1个PDF文件约1.9MB共27页目录清晰、内容完整已有74人学习浏览。文档围绕低成本交通流量预测需求系统梳理数据收集、清洗、特征工程与数据划分以及DeepSeek-V3环境准备、模型加载、微调参数设置、训练循环与评估的完整流程并进一步展开预测模型架构设计、模型融合、训练优化、部署与实时预测等环节介绍超参数调优、模型结构调整和数据增强方法。针对成本控制还给出硬件资源优化、开源数据利用、模型压缩、迁移学习与自动化运维等策略并结合实际案例进行预测精度、应用效果和成本效益分析。适合希望以较低成本落地交通流量预测、掌握大模型微调实战路径的读者参考。1. 一个区级项目的预算常常在第一版方案里就被算力吃光去年帮一个区里做交通流量预测的验证最先被否掉的不是算法而是账单。按传统思路自建检测器加训练专用时序网络光把两百多个路口的地磁线圈补齐硬件加运维就能吃掉整个项目预算后面模型迭代还得再往里填人力和机器。后来换了个思路感知设备维持原样预测模型换成 DeepSeek-V3 做底座用 LoRA 低秩适配把需要更新的参数量压到原来的极小比例在几张消费级卡上就能跑完一轮实验。这份 PDF 讲的正是这条路线它把交通流量预测拆成四段——数据管道、DeepSeek-V3 微调、预测头接入、实时推理服务每一段都按低成本约束来设计。它适合两类人手里有城市交通数据但算力有限、想试大模型的工程同学以及做时序预测、想看清楚微调到底改了哪些参数、改了之后精度从哪来的从业者。下面的内容按能复现的顺序展开参数给到具体值坑也标在踩过的地方。2. 交通流量数据管道MQTT 采集、异常清洗与滑动窗口样本交通流量预测的精度天花板在数据侧就定死了模型只是在逼近这个天花板。这一章把采集、清洗、特征化到样本构造整条链路拆开每一段都给出可以直接抄的代码和参数重点解释为什么这样取值而不是取另一组。2.1 采样粒度先定后面所有参数都跟着它走一个常见误区是把采样频率拉到 1 分钟觉得数据越密越好。实测下来 1 分钟粒度的噪声占比极高同一个路口相邻两分钟的流量能差出三倍模型学到的大部分是随机波动而不是规律。行业里比较稳的做法是 5 分钟或 15 分钟聚合5 分钟粒度能反映信号灯周期带来的起伏15 分钟粒度对传感器缺数更宽容样本量也只有 1 分钟的三分之一。对低成本方案我一般先定 15 分钟聚合。理由是它能直接复用大部分城市已经开放的公开数据集采集和筛选的边际成本最低如果后面精度不够再往 5 分钟调此时只需要重跑聚合脚本采集侧的改动很小。这个顺序很重要反过来做等于把最贵的环节放在最不确定的阶段。表 2-1 是我实际会拉的几类数据以及各自在缺失情况下的降级策略。数据源典型字段更新频率获取方式是否必需路口检测器flow、speed、occupancy1 分钟原始聚合到 15 分钟MQTT 订阅或数据库直读必需气象数据降水、温度、能见度小时级HTTP 接口定时拉取强烈建议节假日与事件节假日类型、大型活动天级本地日历表 人工维护必需路网拓扑上下游路口、车道数静态路网基础表可选浮动车 GPS路段平均速度分钟级第三方或自有平台可选2.2 MQTT 订阅与气象、日历数据的对齐检测器数据走 MQTT 是最省事的方案轻量、断线重连逻辑成熟边缘设备跑得动。订阅端不需要写得复杂把消息落成带时间戳的宽表就行真正的清洗放在离线批处理里做避免采集端出问题导致丢数。import json import sqlite3 import paho.mqtt.client as mqtt conn sqlite3.connect(traffic_raw.db) conn.execute(CREATE TABLE IF NOT EXISTS raw_flow( ts TEXT, road_id TEXT, flow REAL, speed REAL, occupancy REAL)) def on_connect(client, userdata, flags, rc): # 订阅所有路口的主题用通配符避免每加一个路口就改一次配置 client.subscribe(city/traffic//flow, qos1) def on_message(client, userdata, msg): # 主题格式 city/traffic/{road_id}/flow从主题里取路口编号 road_id msg.topic.split(/)[2] payload json.loads(msg.payload) conn.execute(INSERT INTO raw_flow VALUES(?,?,?,?,?), ( payload[ts], road_id, payload.get(flow), payload.get(speed), payload.get(occupancy))) conn.commit() client mqtt.Client(client_idflow-collector-01) client.on_connect on_connect client.on_message on_message client.connect(mqtt.internal.host, 1883, 60) client.loop_forever()这段代码里有两个值值得留意。qos1表示至少投递一次代价是可能收到重复消息所以下游聚合时要用drop_duplicates按(ts, road_id)去重而不是假设数据唯一。client_id固定下来是为了让服务端保留会话状态断线期间的消息在重连后能补发如果每次启动生成随机 id这部分数据就永久丢了。气象和日历数据用 HTTP 拉取就够了注意两点气象接口通常按小时返回需要前向填充到 15 分钟粒度代表「这一小时内天气没变」节假日字段不要只填「是/否」要区分工作日、周末、法定节假日、调休工作日四类因为调休工作日的流量形态和普通工作日差别很大。2.3 缺失值与异常值全局 Z-score 会把高峰判成异常清洗环节最容易写错的是异常检测。教科书上的 Z-score 用全局均值和全局标准差放到交通流量上会直接把早高峰识别成异常点因为高峰流量本身就是偏离均值的。我一般改用滚动窗口的中位数和 MADimport numpy as np import pandas as pd def flag_outliers(series, window96, k3.0): window96 表示 15 分钟粒度下的一天用同期的中位数做参照 med series.rolling(window, min_periodswindow // 2).median() mad (series - med).abs().rolling(window, min_periodswindow // 2).median() # 1.4826 是把 MAD 折算成标准差尺度的一致性系数 return (series - med).abs() k * 1.4826 * mad df pd.read_csv(traffic_15min.csv, parse_dates[time]) bad flag_outliers(df[flow]) df.loc[bad, flow] np.nan # 先置空再统一走填充流程滚动中位数描述的是「这个路口这个时段最近一天左右的正常水平」MAD 描述的是波动幅度两者都不受个别尖峰影响。k3.0是经验值传感器精度差的点位可以放宽到 4.0否则会把真实的突发流量误删。填充要分长短缺口处理短缺口插值长缺口用同星期同一时段的中位数兜底pivot df.pivot_table(indextime, columnsroad_id, valuesflow) full_index pd.date_range(pivot.index.min(), pivot.index.max(), freq15min) pivot pivot.reindex(full_index) # 补出整段缺失的时间点 pivot pivot.interpolate(limit4, limit_directionboth) # 连续 4 个点以内插值 # 一天 96 个 15 分钟槽位用「星期几 槽位」作为兜底分组键 slot pivot.index.dayofweek * 96 pivot.index.hour * 4 pivot.index.minute // 15 pivot pivot.fillna(pivot.groupby(slot).transform(median))limit4对应一小时以内的缺口插值结果可信超过一小时的缺口插值会拉出一条直线模型会把它当成真实趋势学进去所以必须切到分组中位数。注意填充之后要在样本里额外带一个「是否填充」的布尔标记让模型知道哪些点不可信。很多团队省掉这一步结果模型在填充密集的路段误差明显偏高。2.4 滑动窗口样本与特征标准化大模型做时序预测输入仍然要切成固定长度的窗口。15 分钟粒度下in_len96正好是过去 24 小时out_len4是未来 1 小时这个组合能覆盖绝大多数信号配时和诱导屏的决策周期。def make_windows(arr, in_len96, out_len4): X, y [], [] for i in range(len(arr) - in_len - out_len 1): X.append(arr[i:i in_len]) y.append(arr[i in_len:i in_len out_len]) return np.asarray(X, dtypefloat32), np.asarray(y, dtypefloat32) # 标准化参数只能从训练段统计避免测试集信息泄漏 train_end int(len(pivot) * 0.7) mu, sigma pivot.iloc[:train_end].stack().mean(), pivot.iloc[:train_end].stack().std() normed (pivot - mu) / (sigma 1e-6)mu和sigma只从训练段统计这一条是硬要求。见过把全量数据一起标准化再划分数据集的写法测试集指标会好看一截上线后立刻打回原形。数据划分按时间顺序 70/15/15 切三段不做随机打乱因为随机打乱会让未来的数据进到训练集这在时序任务里是最隐蔽的泄漏。3. DeepSeek-V3 微调方式选型LoRA 低秩适配的参数表与显存账把预测任务挂到大模型上第一步不是写训练循环而是算清楚显存和可训练参数。这一章先给出选型对比再落到具体配置和训练脚本。3.1 全量微调、LoRA、QLoRA 的取舍三种方式的差别不在精度上限而在你愿意付出多少显存换多少适配自由度。表 3-1 是我按实际压测整理的对照显存数值是泛指区间具体取决于底座规模和序列长度。微调方式可训练参数占比单卡显存需求训练速度适用场景全量微调100%需要多卡并行慢数据量大、要做长期迭代LoRA通常低于 1%单卡可跑快单任务适配本方案首选QLoRA与 LoRA 同量级比 LoRA 再降一档略慢显存紧张、验证阶段低成本方案里 LoRA 是默认选项原因是它不动底座权重只训练旁路矩阵产出的适配器文件只有几十到几百 MB可以按路口群或按城市分别保存多份切换时只换适配器不换底座。这一点对智慧城市场景特别有用不同区域的流量形态差异大用同一个底座挂多个适配器比维护多个完整模型便宜得多。提示适配器是加在注意力层和前馈层的线性投影上的底座本身的权重在合并前保持只读。上线部署时可以选择在服务启动阶段把适配器合并回底座减少一次前向的额外开销。3.2 LoRA 秩、alpha 与 target_modules 怎么定这三个参数决定了适配器的容量。秩r太小模型没有足够自由度学新任务的偏移太大可训练参数回到全量微调的量级低成本的优势就没了。表 3-2 是我在交通流量任务上的经验取值。参数常用取值调大后的影响调小后的影响lora_rank16 或 32容量提升易过拟合显存上升欠拟合高峰段学不动lora_alpha通常为 rank 的 2 倍缩放后等效学习率变大更新幅度偏小收敛慢lora_dropout0.05 至 0.1正则更强收敛变慢小数据集上容易过拟合target_modules注意力四个投影层覆盖面广参数变多只调 q、v 时容量偏紧lora_alpha与lora_rank的比值决定了实际更新强度所以常见组合是r16, alpha32等价于把 LoRA 分支的输出放大一倍。如果换了秩但没改 alpha会觉得「调整没效果」其实是被这个比值抵消掉了。target_modules建议至少覆盖q_proj, k_proj, v_proj, o_proj四个注意力投影。只调q_proj、v_proj是早期论文的省钱配置放到数值回归任务上容量偏紧高峰时段的误差会明显偏高。3.3 用 LLaMA-Factory 或原生 PEFT 跑通第一轮如果只是要快速验证可行性用 LLaMA-Factory 这类一站式微调平台最省时间配置写成 YAML 就能跑### 示例LLaMA-Factory 的 LoRA 微调配置片段 stage: sft do_train: true model_name_or_path: /models/DeepSeek-V3 # 本地权重目录 dataset: traffic_flow_sft # 已注册的数据集名 template: default finetuning_type: lora lora_rank: 16 lora_alpha: 32 lora_dropout: 0.05 lora_target: q_proj,k_proj,v_proj,o_proj cutoff_len: 1024 # 96 个历史点 提示模板足够 per_device_train_batch_size: 2 gradient_accumulation_steps: 16 # 等效 batch 32 learning_rate: 1.0e-4 num_train_epochs: 3 lr_scheduler_type: cosine warmup_ratio: 0.03 bf16: true gradient_checkpointing: true output_dir: /out/traffic_loraper_device_train_batch_size设 2 配合gradient_accumulation_steps16等效批次是 32但峰值显存只按 2 条样本算这是单卡跑起来的关键。gradient_checkpointing用时间换显存大概多花两成训练时间换来的是能塞下更长的序列。cutoff_len1024不用开太大交通流量窗口加上提示模板一般不超过 512 个 token留一倍余量避免截断。要精细控制结构走原生 PEFT 更清楚from peft import LoraConfig, get_peft_model from transformers import AutoModelForCausalLM model AutoModelForCausalLM.from_pretrained( /models/DeepSeek-V3, torch_dtypebfloat16, device_mapauto, load_in_4bitTrue, # 显存紧张时启用即 QLoRA ) cfg LoraConfig( r16, lora_alpha32, lora_dropout0.05, target_modules[q_proj, k_proj, v_proj, o_proj], biasnone, task_typeCAUSAL_LM, ) model get_peft_model(model, cfg) model.print_trainable_parameters() # 先确认可训练参数占比符合预期load_in_4bitTrue打开后底座权重以 4 位存储显存大约降到原来的一半以下代价是反量化带来的一点速度损失。print_trainable_parameters()这一步别省它直接告诉你这次微调到底在训练多少参数如果打出来接近全量多半是target_modules写成了通配符或者把整个模块名带进去了。3.4 第一轮训练最容易踩的三个坑第一个是损失不下降但也不报错。多数情况下是学习率太小1e-5那套是全量微调的常用值LoRA 因为只更新旁路矩阵通常要放大到1e-4附近才能动起来。第二个是 loss 直接变成 NaN检查输入里有没有标准化之后没处理的极端值交通流量数据在设备重启那一瞬可能冒出异常大的数。第三个是显存越跑越大一般是把验证集的张量堆在显存里没释放验证循环里记得加torch.no_grad()。注意每轮结束把验证损失和验证集的分时段 MAE 一起记到日志。只看总损失会漏掉「平峰拟合很好、高峰全错」的情况这种模型在交通场景基本没法用。4. 预测头接入与实时推理服务从隐状态到流量标量底座微调完成只解决了「表征」把表征变成未来一小时的四个流量值还需要一个回归头以及一套能扛住实时请求的服务。4.1 回归头取哪一层、取哪个 token语言模型输出的是每个位置的隐状态做流量回归要把它压成一个定长向量。常见做法有两种对所有非填充位置做平均池化或者只取最后一个有效 token。交通流量这种「用历史推未来」的任务把窗口末尾的信息放在序列最后更自然所以我一般取最后一个有效位置import torch import torch.nn as nn class TrafficHead(nn.Module): def __init__(self, hidden_size, horizon4, dropout0.1): super().__init__() self.norm nn.LayerNorm(hidden_size) self.drop nn.Dropout(dropout) self.proj nn.Linear(hidden_size, horizon) def forward(self, hidden_states, attention_mask): # 用 mask 求和减一得到每条样本最后一个非 padding 的位置 last_idx attention_mask.sum(dim1) - 1 h hidden_states[torch.arange(hidden_states.size(0)), last_idx] return self.proj(self.drop(self.norm(h)))attention_mask.sum(dim1) - 1这个写法比固定取最后一列安全因为批内样本长度可能不一致填充位置的隐状态是无意义的。LayerNorm放在投影前是为了让不同批次之间的尺度稳定训练时收敛更快。horizon4对应未来一小时要预测更长时直接改这个数不用动其他结构。4.2 数值怎么进模型模板拼装与投影层的取舍大模型的输入是 token交通流量是浮点数这里有两种主流做法。一种是把数值离散化后按文本拼进提示模板训练和推理都简单缺点是精度受分箱粒度限制另一种是给数值单独接一个线性投影层把连续值直接映射到隐空间精度高但要改动输入结构。低成本方案我倾向第一种理由是它完全复用现成的微调流程数据格式就是「历史序列 目标描述」的文本对不需要改模型前向。分箱粒度取到 0.1 的归一化精度就够再细对 MAE 的贡献很小反而拉长序列。4.3 评估指标MAPE 在低流量时段会骗人交通流量预测常用三个指标各自有适用面表 4-1 是我在项目里同时记录它们的原因。指标含义对高峰的敏感度使用注意MAE平均绝对误差中单位直观便于和业务阈值对齐RMSE均方根误差高放大大误差适合盯高峰MAPE平均绝对百分比误差低夜间低流量时会被放大需分时段看MAPE 在凌晨时段特别容易失真真实流量是每小时 30 辆预测差 10 辆误差率就超过 30%但这个绝对值对交通管理几乎没有影响。所以评估一定要按平峰、高峰、夜间分段统计并且把 MAPE 的权重按流量占比加权否则会为了压夜间误差牺牲高峰精度。4.4 推理服务批处理与半精度上线服务用 FastAPI 加半精度推理是最短的路径重点是别让每个请求单独跑一次前向from fastapi import FastAPI import torch app FastAPI() device cuda app.post(/predict) def predict(payload: dict): # payload[history] 是最近 96 个已归一化的流量点 seq torch.tensor(payload[history], dtypetorch.float32, devicedevice).unsqueeze(0) with torch.no_grad(), torch.autocast(cuda, dtypetorch.bfloat16): out model(seq) # 输出未来 4 个点的归一化值 return {horizon: out[0].float().cpu().tolist()}torch.autocast让矩阵乘走 bfloat16显存占用和延迟都会下降数值回归任务对这点精度损失不敏感。生产环境要再包一层批量聚合把 50 毫秒内到达的请求攒成一个批次一起前向吞吐能提升数倍代价是增加几十毫秒延迟对分钟级的交通预测完全可接受。5. 低成本验证与排错把误差拆到分时段再决定改哪里模型跑起来之后钱主要花在两个地方反复重训和无效的扩容。这一章给出一套低成本定位问题的方法避免在错误的方向上加机器。5.1 先量显存和吞吐再决定是否加卡训练卡顿的绝大多数原因不是算力不够而是显存碎片或数据加载成为瓶颈。用 PyTorch 自带的 profiler 跑一百步就能看出来import torch from torch.profiler import profile, ProfilerActivity with profile(activities[ProfilerActivity.CUDA], record_shapesTrue) as prof: for step, batch in enumerate(train_loader): train_step(batch) if step 100: break print(prof.key_averages().table(sort_bycuda_time_total, row_limit10))如果表格里排在前面的算子时间很短、但总时间很长问题在数据加载把DataLoader的num_workers调大并加pin_memoryTrue就行如果某个矩阵乘算子占了绝大多数时间才说明是计算量真的顶到了硬件上限。这两种情况的处理方向完全相反先做这一步能省下大量试错时间。5.2 把误差按小时和按路口交叉拆开总指标只能告诉你「不够好」说明不了改哪里。我一般会输出一张交叉表行是小时段列是路口等级主干道、次干道、支路格子里填 MAE。经验规律是主干道高峰段的误差最难压因为它的流量波动本身方差就大而支路夜间误差偏高往往是填充标记没进特征模型把插值出的直线当成了真实数据。如果误差集中在少数几个路口优先检查这几处的传感器是不是长期偏移别急着调模型。数据侧修一个传感器效果通常比调一周超参更明显。5.3 用分位数损失同时拿到点预测和区间做到最后一步与其反复调结构去压 MAE不如换个输出形式让模型输出分位数用 pinball loss 训练。这样一次前向就能同时得到点预测和置信区间区间宽度还能当成信号配时的安全裕度用。import torch def pinball_loss(pred, target, quantiles(0.1, 0.5, 0.9)): pred 形状为 (batch, horizon, len(quantiles))target 形状为 (batch, horizon) losses [] for i, q in enumerate(quantiles): e target - pred[..., i] # 分位数损失低估和高估分别按 q 和 1-q 加权 losses.append(torch.maximum(q * e, (q - 1) * e).mean()) return sum(losses) / len(losses)训练时把回归头输出维度从horizon改成horizon * len(quantiles)推理时取第 0.5 分位作为点预测上报给诱导屏取 0.9 分位作为高峰时段信号配时的上限参考取 0.1 分位作为下界用于判断是否需要提前切换配时方案。三个分位数用同一次前向算出来推理成本不变这也是低成本方案里性价比最高的一处改动。损失权重上如果业务更在意高峰不堵可以把 0.9 分位的权重单独乘 1.5让模型在偏保守的一侧更用力。本文还有配套的精品资源点击获取
DeepSeek-V3+LoRA：低成本交通流量预测与实时推理
简介《智慧城市低成本方案DeepSeek-V3微调实现交通流量预测》是一份面向智慧城市、智能交通及大模型应用开发者的PDF技术文档。资源包为1个PDF文件约1.9MB共27页目录清晰、内容完整已有74人学习浏览。文档围绕低成本交通流量预测需求系统梳理数据收集、清洗、特征工程与数据划分以及DeepSeek-V3环境准备、模型加载、微调参数设置、训练循环与评估的完整流程并进一步展开预测模型架构设计、模型融合、训练优化、部署与实时预测等环节介绍超参数调优、模型结构调整和数据增强方法。针对成本控制还给出硬件资源优化、开源数据利用、模型压缩、迁移学习与自动化运维等策略并结合实际案例进行预测精度、应用效果和成本效益分析。适合希望以较低成本落地交通流量预测、掌握大模型微调实战路径的读者参考。1. 一个区级项目的预算常常在第一版方案里就被算力吃光去年帮一个区里做交通流量预测的验证最先被否掉的不是算法而是账单。按传统思路自建检测器加训练专用时序网络光把两百多个路口的地磁线圈补齐硬件加运维就能吃掉整个项目预算后面模型迭代还得再往里填人力和机器。后来换了个思路感知设备维持原样预测模型换成 DeepSeek-V3 做底座用 LoRA 低秩适配把需要更新的参数量压到原来的极小比例在几张消费级卡上就能跑完一轮实验。这份 PDF 讲的正是这条路线它把交通流量预测拆成四段——数据管道、DeepSeek-V3 微调、预测头接入、实时推理服务每一段都按低成本约束来设计。它适合两类人手里有城市交通数据但算力有限、想试大模型的工程同学以及做时序预测、想看清楚微调到底改了哪些参数、改了之后精度从哪来的从业者。下面的内容按能复现的顺序展开参数给到具体值坑也标在踩过的地方。2. 交通流量数据管道MQTT 采集、异常清洗与滑动窗口样本交通流量预测的精度天花板在数据侧就定死了模型只是在逼近这个天花板。这一章把采集、清洗、特征化到样本构造整条链路拆开每一段都给出可以直接抄的代码和参数重点解释为什么这样取值而不是取另一组。2.1 采样粒度先定后面所有参数都跟着它走一个常见误区是把采样频率拉到 1 分钟觉得数据越密越好。实测下来 1 分钟粒度的噪声占比极高同一个路口相邻两分钟的流量能差出三倍模型学到的大部分是随机波动而不是规律。行业里比较稳的做法是 5 分钟或 15 分钟聚合5 分钟粒度能反映信号灯周期带来的起伏15 分钟粒度对传感器缺数更宽容样本量也只有 1 分钟的三分之一。对低成本方案我一般先定 15 分钟聚合。理由是它能直接复用大部分城市已经开放的公开数据集采集和筛选的边际成本最低如果后面精度不够再往 5 分钟调此时只需要重跑聚合脚本采集侧的改动很小。这个顺序很重要反过来做等于把最贵的环节放在最不确定的阶段。表 2-1 是我实际会拉的几类数据以及各自在缺失情况下的降级策略。数据源典型字段更新频率获取方式是否必需路口检测器flow、speed、occupancy1 分钟原始聚合到 15 分钟MQTT 订阅或数据库直读必需气象数据降水、温度、能见度小时级HTTP 接口定时拉取强烈建议节假日与事件节假日类型、大型活动天级本地日历表 人工维护必需路网拓扑上下游路口、车道数静态路网基础表可选浮动车 GPS路段平均速度分钟级第三方或自有平台可选2.2 MQTT 订阅与气象、日历数据的对齐检测器数据走 MQTT 是最省事的方案轻量、断线重连逻辑成熟边缘设备跑得动。订阅端不需要写得复杂把消息落成带时间戳的宽表就行真正的清洗放在离线批处理里做避免采集端出问题导致丢数。import json import sqlite3 import paho.mqtt.client as mqtt conn sqlite3.connect(traffic_raw.db) conn.execute(CREATE TABLE IF NOT EXISTS raw_flow( ts TEXT, road_id TEXT, flow REAL, speed REAL, occupancy REAL)) def on_connect(client, userdata, flags, rc): # 订阅所有路口的主题用通配符避免每加一个路口就改一次配置 client.subscribe(city/traffic//flow, qos1) def on_message(client, userdata, msg): # 主题格式 city/traffic/{road_id}/flow从主题里取路口编号 road_id msg.topic.split(/)[2] payload json.loads(msg.payload) conn.execute(INSERT INTO raw_flow VALUES(?,?,?,?,?), ( payload[ts], road_id, payload.get(flow), payload.get(speed), payload.get(occupancy))) conn.commit() client mqtt.Client(client_idflow-collector-01) client.on_connect on_connect client.on_message on_message client.connect(mqtt.internal.host, 1883, 60) client.loop_forever()这段代码里有两个值值得留意。qos1表示至少投递一次代价是可能收到重复消息所以下游聚合时要用drop_duplicates按(ts, road_id)去重而不是假设数据唯一。client_id固定下来是为了让服务端保留会话状态断线期间的消息在重连后能补发如果每次启动生成随机 id这部分数据就永久丢了。气象和日历数据用 HTTP 拉取就够了注意两点气象接口通常按小时返回需要前向填充到 15 分钟粒度代表「这一小时内天气没变」节假日字段不要只填「是/否」要区分工作日、周末、法定节假日、调休工作日四类因为调休工作日的流量形态和普通工作日差别很大。2.3 缺失值与异常值全局 Z-score 会把高峰判成异常清洗环节最容易写错的是异常检测。教科书上的 Z-score 用全局均值和全局标准差放到交通流量上会直接把早高峰识别成异常点因为高峰流量本身就是偏离均值的。我一般改用滚动窗口的中位数和 MADimport numpy as np import pandas as pd def flag_outliers(series, window96, k3.0): window96 表示 15 分钟粒度下的一天用同期的中位数做参照 med series.rolling(window, min_periodswindow // 2).median() mad (series - med).abs().rolling(window, min_periodswindow // 2).median() # 1.4826 是把 MAD 折算成标准差尺度的一致性系数 return (series - med).abs() k * 1.4826 * mad df pd.read_csv(traffic_15min.csv, parse_dates[time]) bad flag_outliers(df[flow]) df.loc[bad, flow] np.nan # 先置空再统一走填充流程滚动中位数描述的是「这个路口这个时段最近一天左右的正常水平」MAD 描述的是波动幅度两者都不受个别尖峰影响。k3.0是经验值传感器精度差的点位可以放宽到 4.0否则会把真实的突发流量误删。填充要分长短缺口处理短缺口插值长缺口用同星期同一时段的中位数兜底pivot df.pivot_table(indextime, columnsroad_id, valuesflow) full_index pd.date_range(pivot.index.min(), pivot.index.max(), freq15min) pivot pivot.reindex(full_index) # 补出整段缺失的时间点 pivot pivot.interpolate(limit4, limit_directionboth) # 连续 4 个点以内插值 # 一天 96 个 15 分钟槽位用「星期几 槽位」作为兜底分组键 slot pivot.index.dayofweek * 96 pivot.index.hour * 4 pivot.index.minute // 15 pivot pivot.fillna(pivot.groupby(slot).transform(median))limit4对应一小时以内的缺口插值结果可信超过一小时的缺口插值会拉出一条直线模型会把它当成真实趋势学进去所以必须切到分组中位数。注意填充之后要在样本里额外带一个「是否填充」的布尔标记让模型知道哪些点不可信。很多团队省掉这一步结果模型在填充密集的路段误差明显偏高。2.4 滑动窗口样本与特征标准化大模型做时序预测输入仍然要切成固定长度的窗口。15 分钟粒度下in_len96正好是过去 24 小时out_len4是未来 1 小时这个组合能覆盖绝大多数信号配时和诱导屏的决策周期。def make_windows(arr, in_len96, out_len4): X, y [], [] for i in range(len(arr) - in_len - out_len 1): X.append(arr[i:i in_len]) y.append(arr[i in_len:i in_len out_len]) return np.asarray(X, dtypefloat32), np.asarray(y, dtypefloat32) # 标准化参数只能从训练段统计避免测试集信息泄漏 train_end int(len(pivot) * 0.7) mu, sigma pivot.iloc[:train_end].stack().mean(), pivot.iloc[:train_end].stack().std() normed (pivot - mu) / (sigma 1e-6)mu和sigma只从训练段统计这一条是硬要求。见过把全量数据一起标准化再划分数据集的写法测试集指标会好看一截上线后立刻打回原形。数据划分按时间顺序 70/15/15 切三段不做随机打乱因为随机打乱会让未来的数据进到训练集这在时序任务里是最隐蔽的泄漏。3. DeepSeek-V3 微调方式选型LoRA 低秩适配的参数表与显存账把预测任务挂到大模型上第一步不是写训练循环而是算清楚显存和可训练参数。这一章先给出选型对比再落到具体配置和训练脚本。3.1 全量微调、LoRA、QLoRA 的取舍三种方式的差别不在精度上限而在你愿意付出多少显存换多少适配自由度。表 3-1 是我按实际压测整理的对照显存数值是泛指区间具体取决于底座规模和序列长度。微调方式可训练参数占比单卡显存需求训练速度适用场景全量微调100%需要多卡并行慢数据量大、要做长期迭代LoRA通常低于 1%单卡可跑快单任务适配本方案首选QLoRA与 LoRA 同量级比 LoRA 再降一档略慢显存紧张、验证阶段低成本方案里 LoRA 是默认选项原因是它不动底座权重只训练旁路矩阵产出的适配器文件只有几十到几百 MB可以按路口群或按城市分别保存多份切换时只换适配器不换底座。这一点对智慧城市场景特别有用不同区域的流量形态差异大用同一个底座挂多个适配器比维护多个完整模型便宜得多。提示适配器是加在注意力层和前馈层的线性投影上的底座本身的权重在合并前保持只读。上线部署时可以选择在服务启动阶段把适配器合并回底座减少一次前向的额外开销。3.2 LoRA 秩、alpha 与 target_modules 怎么定这三个参数决定了适配器的容量。秩r太小模型没有足够自由度学新任务的偏移太大可训练参数回到全量微调的量级低成本的优势就没了。表 3-2 是我在交通流量任务上的经验取值。参数常用取值调大后的影响调小后的影响lora_rank16 或 32容量提升易过拟合显存上升欠拟合高峰段学不动lora_alpha通常为 rank 的 2 倍缩放后等效学习率变大更新幅度偏小收敛慢lora_dropout0.05 至 0.1正则更强收敛变慢小数据集上容易过拟合target_modules注意力四个投影层覆盖面广参数变多只调 q、v 时容量偏紧lora_alpha与lora_rank的比值决定了实际更新强度所以常见组合是r16, alpha32等价于把 LoRA 分支的输出放大一倍。如果换了秩但没改 alpha会觉得「调整没效果」其实是被这个比值抵消掉了。target_modules建议至少覆盖q_proj, k_proj, v_proj, o_proj四个注意力投影。只调q_proj、v_proj是早期论文的省钱配置放到数值回归任务上容量偏紧高峰时段的误差会明显偏高。3.3 用 LLaMA-Factory 或原生 PEFT 跑通第一轮如果只是要快速验证可行性用 LLaMA-Factory 这类一站式微调平台最省时间配置写成 YAML 就能跑### 示例LLaMA-Factory 的 LoRA 微调配置片段 stage: sft do_train: true model_name_or_path: /models/DeepSeek-V3 # 本地权重目录 dataset: traffic_flow_sft # 已注册的数据集名 template: default finetuning_type: lora lora_rank: 16 lora_alpha: 32 lora_dropout: 0.05 lora_target: q_proj,k_proj,v_proj,o_proj cutoff_len: 1024 # 96 个历史点 提示模板足够 per_device_train_batch_size: 2 gradient_accumulation_steps: 16 # 等效 batch 32 learning_rate: 1.0e-4 num_train_epochs: 3 lr_scheduler_type: cosine warmup_ratio: 0.03 bf16: true gradient_checkpointing: true output_dir: /out/traffic_loraper_device_train_batch_size设 2 配合gradient_accumulation_steps16等效批次是 32但峰值显存只按 2 条样本算这是单卡跑起来的关键。gradient_checkpointing用时间换显存大概多花两成训练时间换来的是能塞下更长的序列。cutoff_len1024不用开太大交通流量窗口加上提示模板一般不超过 512 个 token留一倍余量避免截断。要精细控制结构走原生 PEFT 更清楚from peft import LoraConfig, get_peft_model from transformers import AutoModelForCausalLM model AutoModelForCausalLM.from_pretrained( /models/DeepSeek-V3, torch_dtypebfloat16, device_mapauto, load_in_4bitTrue, # 显存紧张时启用即 QLoRA ) cfg LoraConfig( r16, lora_alpha32, lora_dropout0.05, target_modules[q_proj, k_proj, v_proj, o_proj], biasnone, task_typeCAUSAL_LM, ) model get_peft_model(model, cfg) model.print_trainable_parameters() # 先确认可训练参数占比符合预期load_in_4bitTrue打开后底座权重以 4 位存储显存大约降到原来的一半以下代价是反量化带来的一点速度损失。print_trainable_parameters()这一步别省它直接告诉你这次微调到底在训练多少参数如果打出来接近全量多半是target_modules写成了通配符或者把整个模块名带进去了。3.4 第一轮训练最容易踩的三个坑第一个是损失不下降但也不报错。多数情况下是学习率太小1e-5那套是全量微调的常用值LoRA 因为只更新旁路矩阵通常要放大到1e-4附近才能动起来。第二个是 loss 直接变成 NaN检查输入里有没有标准化之后没处理的极端值交通流量数据在设备重启那一瞬可能冒出异常大的数。第三个是显存越跑越大一般是把验证集的张量堆在显存里没释放验证循环里记得加torch.no_grad()。注意每轮结束把验证损失和验证集的分时段 MAE 一起记到日志。只看总损失会漏掉「平峰拟合很好、高峰全错」的情况这种模型在交通场景基本没法用。4. 预测头接入与实时推理服务从隐状态到流量标量底座微调完成只解决了「表征」把表征变成未来一小时的四个流量值还需要一个回归头以及一套能扛住实时请求的服务。4.1 回归头取哪一层、取哪个 token语言模型输出的是每个位置的隐状态做流量回归要把它压成一个定长向量。常见做法有两种对所有非填充位置做平均池化或者只取最后一个有效 token。交通流量这种「用历史推未来」的任务把窗口末尾的信息放在序列最后更自然所以我一般取最后一个有效位置import torch import torch.nn as nn class TrafficHead(nn.Module): def __init__(self, hidden_size, horizon4, dropout0.1): super().__init__() self.norm nn.LayerNorm(hidden_size) self.drop nn.Dropout(dropout) self.proj nn.Linear(hidden_size, horizon) def forward(self, hidden_states, attention_mask): # 用 mask 求和减一得到每条样本最后一个非 padding 的位置 last_idx attention_mask.sum(dim1) - 1 h hidden_states[torch.arange(hidden_states.size(0)), last_idx] return self.proj(self.drop(self.norm(h)))attention_mask.sum(dim1) - 1这个写法比固定取最后一列安全因为批内样本长度可能不一致填充位置的隐状态是无意义的。LayerNorm放在投影前是为了让不同批次之间的尺度稳定训练时收敛更快。horizon4对应未来一小时要预测更长时直接改这个数不用动其他结构。4.2 数值怎么进模型模板拼装与投影层的取舍大模型的输入是 token交通流量是浮点数这里有两种主流做法。一种是把数值离散化后按文本拼进提示模板训练和推理都简单缺点是精度受分箱粒度限制另一种是给数值单独接一个线性投影层把连续值直接映射到隐空间精度高但要改动输入结构。低成本方案我倾向第一种理由是它完全复用现成的微调流程数据格式就是「历史序列 目标描述」的文本对不需要改模型前向。分箱粒度取到 0.1 的归一化精度就够再细对 MAE 的贡献很小反而拉长序列。4.3 评估指标MAPE 在低流量时段会骗人交通流量预测常用三个指标各自有适用面表 4-1 是我在项目里同时记录它们的原因。指标含义对高峰的敏感度使用注意MAE平均绝对误差中单位直观便于和业务阈值对齐RMSE均方根误差高放大大误差适合盯高峰MAPE平均绝对百分比误差低夜间低流量时会被放大需分时段看MAPE 在凌晨时段特别容易失真真实流量是每小时 30 辆预测差 10 辆误差率就超过 30%但这个绝对值对交通管理几乎没有影响。所以评估一定要按平峰、高峰、夜间分段统计并且把 MAPE 的权重按流量占比加权否则会为了压夜间误差牺牲高峰精度。4.4 推理服务批处理与半精度上线服务用 FastAPI 加半精度推理是最短的路径重点是别让每个请求单独跑一次前向from fastapi import FastAPI import torch app FastAPI() device cuda app.post(/predict) def predict(payload: dict): # payload[history] 是最近 96 个已归一化的流量点 seq torch.tensor(payload[history], dtypetorch.float32, devicedevice).unsqueeze(0) with torch.no_grad(), torch.autocast(cuda, dtypetorch.bfloat16): out model(seq) # 输出未来 4 个点的归一化值 return {horizon: out[0].float().cpu().tolist()}torch.autocast让矩阵乘走 bfloat16显存占用和延迟都会下降数值回归任务对这点精度损失不敏感。生产环境要再包一层批量聚合把 50 毫秒内到达的请求攒成一个批次一起前向吞吐能提升数倍代价是增加几十毫秒延迟对分钟级的交通预测完全可接受。5. 低成本验证与排错把误差拆到分时段再决定改哪里模型跑起来之后钱主要花在两个地方反复重训和无效的扩容。这一章给出一套低成本定位问题的方法避免在错误的方向上加机器。5.1 先量显存和吞吐再决定是否加卡训练卡顿的绝大多数原因不是算力不够而是显存碎片或数据加载成为瓶颈。用 PyTorch 自带的 profiler 跑一百步就能看出来import torch from torch.profiler import profile, ProfilerActivity with profile(activities[ProfilerActivity.CUDA], record_shapesTrue) as prof: for step, batch in enumerate(train_loader): train_step(batch) if step 100: break print(prof.key_averages().table(sort_bycuda_time_total, row_limit10))如果表格里排在前面的算子时间很短、但总时间很长问题在数据加载把DataLoader的num_workers调大并加pin_memoryTrue就行如果某个矩阵乘算子占了绝大多数时间才说明是计算量真的顶到了硬件上限。这两种情况的处理方向完全相反先做这一步能省下大量试错时间。5.2 把误差按小时和按路口交叉拆开总指标只能告诉你「不够好」说明不了改哪里。我一般会输出一张交叉表行是小时段列是路口等级主干道、次干道、支路格子里填 MAE。经验规律是主干道高峰段的误差最难压因为它的流量波动本身方差就大而支路夜间误差偏高往往是填充标记没进特征模型把插值出的直线当成了真实数据。如果误差集中在少数几个路口优先检查这几处的传感器是不是长期偏移别急着调模型。数据侧修一个传感器效果通常比调一周超参更明显。5.3 用分位数损失同时拿到点预测和区间做到最后一步与其反复调结构去压 MAE不如换个输出形式让模型输出分位数用 pinball loss 训练。这样一次前向就能同时得到点预测和置信区间区间宽度还能当成信号配时的安全裕度用。import torch def pinball_loss(pred, target, quantiles(0.1, 0.5, 0.9)): pred 形状为 (batch, horizon, len(quantiles))target 形状为 (batch, horizon) losses [] for i, q in enumerate(quantiles): e target - pred[..., i] # 分位数损失低估和高估分别按 q 和 1-q 加权 losses.append(torch.maximum(q * e, (q - 1) * e).mean()) return sum(losses) / len(losses)训练时把回归头输出维度从horizon改成horizon * len(quantiles)推理时取第 0.5 分位作为点预测上报给诱导屏取 0.9 分位作为高峰时段信号配时的上限参考取 0.1 分位作为下界用于判断是否需要提前切换配时方案。三个分位数用同一次前向算出来推理成本不变这也是低成本方案里性价比最高的一处改动。损失权重上如果业务更在意高峰不堵可以把 0.9 分位的权重单独乘 1.5让模型在偏保守的一侧更用力。本文还有配套的精品资源点击获取