简介基于DAG区块链的联邦学习框架以Python源码加项目说明的形式打包面向需要完成区块链与机器学习交叉方向课程设计、期末大作业或毕业设计的学生以及希望探索去中心化联邦学习实现方法的算法爱好者。框架包含完整代码与说明文档可直接下载运行适合在已有Python基础上做二次开发与调试。压缩包共77个文件以46个Python脚本为核心涵盖节点通信、DAG账本维护、tip选择策略、恶意节点检测、模型聚合等模块另含3个Jupyter Notebook便于复现实验和可视化分析以及环境配置文件与环境说明文档包体大小约1.17MB。目前已有221人学习使用可作为理解“DAG联邦学习”结合实现的直接参考资料。资源还附带数据集自动下载与参数配置入口运行结果统一输出到experiments目录方便对照论文或报告进行结果分析代码中涉及个性化、去中心化等关键机制有助于快速上手并延伸改进。1. 基于DAG区块链的联邦学习框架去中心化不是口号是个账本工程如果你做过联邦学习大概率遇到过同一个痛点模型参数要汇总到一个中心服务器聚合完再下发。这个服务器一旦挂了训练就停摆一旦被攻破所有客户端本地数据分布和更新轨迹都会被扒干净。基于DAG区块链的联邦学习框架解决的就是这件事——把「模型更新广播、验证、聚合」从中心服务器手里拆出来放进一张去中心化的有向无环图账本里。DAG 不像传统区块链那样把所有交易按时间戳排成一条链而是允许新更新的模型参数同时引用多个历史更新各自并行确认吞吐量高好几个量级。个性化则意味着每个客户端不只会拿到同一份全局模型而是在本地数据上微调再把微调痕迹写回 DAG形成一个既有全局共识、又保留个体差异的模型体系。拿到这样一份标题为「python源码项目说明」的压缩包我建议你先别急着跑 demo。先去理解它要替代的中心化范式长什么样再去检查它的 DAG 账本实现是不是真的可靠。这篇笔记会从数据结构、协议闭环、个性化策略、踩坑排查四个层面拆开这个框架并在每个关键节点给你可以直接抄走的最小实现。2. 从线性链到 DAG联邦学习场景为什么需要一张非线性账本2.1 线性区块链在模型更新场景下的三个结构性瓶颈传统区块链的核心结构是一条不断追加的线性链。每个区块包含若干交易矿工或验证节点把新区块链接到链尾靠共识机制保证全网络对链的状态达成一致。这个设计在金融转账、存证确权这类低频、高价值的场景里足够用但在联邦学习里会立刻撞墙。第一个瓶颈是吞吐量。模型更新的频率远高于转账频率一个 100 客户端的实验每轮通信就产生至少 100 条模型更新记录。如果让这些更新排队进区块网络很快会被未确认交易挤爆。第二个瓶颈是确认延迟。线性链的确认要等后续区块追加假设平均出块时间是 10 秒一笔更新等 3 个确认就要 30 秒而联邦学习单轮训练可能需要上千次参数交换训练总时长会膨胀到不可接受。第三个瓶颈是恢复成本——链上任何分叉都要通过重组解决重组期间所有状态不可用这在训练场景里意味着 GPU 集群空转等你。所以当标题里把「DAG 区块链」和「联邦学习」放在一起时本质诉求只有一个找一个既能记录历史更新、又能支撑高频并发的账本结构。DAG 恰好是这种结构——新交易不需要等别人只需要引用已经存在的旧交易多个分支可以同时长出来确认靠后续交易的累计引用权重而非挖矿算力。2.2 DAG 账本的数据结构tip 选择与累计权重理解 DAG 区块链的联邦学习先要过三关顶点是什么、边是什么、tip 是谁。顶点是一个模型更新交易里面装的不一定是全部权重张量而是模型哈希、客户端 ID、本轮编号、本地评估指标、数据分布摘要这些轻量字段。边表示验证关系——新交易创建时必须引用一到两个还没有被其他交易引用过的旧交易被引用的旧交易就完成了验证。那些还没被引用过的交易就是 tip也就是整个账本里等待确认的最前沿节点。在联邦学习场景里每个客户端在本地训练完成之后要做的事就是寻找当前账本上的 tip用自己新生成的更新交易去引用它。tip 选择的策略直接决定账本形态和模型融合路径。最常用的是 CMH 加权随机选择——按累计权重做加权采样。累计权重 该交易自身的初始权重 所有直接或间接引用它的后代交易的权重之和权重高的交易被选中的概率更大这样新交易更倾向于引用已经积累较多确认的历史更新账本会收敛得比较快。import random from dataclasses import dataclass, field from typing import Dict, List dataclass class Tx: tx_id: str client_id: str round_no: int model_hash: str parent_ids: List[str] field(default_factorylist) cumulative_weight: float 1.0 is_tip: bool True class DAGLedger: def __init__(self): self.txs: Dict[str, Tx] {} self.tips: set[str] set() def add_tx(self, tx: Tx): # 一个新交易引用旧tip旧tip就不再是tip for parent_id in tx.parent_ids: self.txs[parent_id].is_tip False self.txs[tx.tx_id] tx tx.is_tip True self.tips.discard(tx.tx_id) for parent_id in tx.parent_ids: self.tips.discard(parent_id) self.tips.add(tx.tx_id) def select_tip(self, decay: float 0.9) - str: # 按照累计权重做加权随机权重对log平滑避免大权重垄断 weights [] for tip_id in self.tips: w (self.txs[tip_id].cumulative_weight ** decay) weights.append(w) return random.choices(list(self.tips), weightsweights, k1)[0]这段代码里decay是核心参数。它控制选择策略对高权重 tip 的偏好程度一般取 0.8 到 0.9。取 1 会导致权重最高的 tip 被反复引用账本变成一个伪链失去并行性取 0 则退化成纯随机每个 tip 被平等对待账本分叉严重模型更新路径混乱。在实际实验里我通常先固定 0.85 跑 50 轮观察 tip 数量是否稳定在 3 到 8 之间——如果 tip 数量持续增长超过 20说明 decay 太大模型更新互相不理睬账本长成了一棵完全分叉的树。2.3 模型更新交易的字段设计轻量账本与重量参数分离既然要上链就得考虑链上不能直接放大模型。一个 ResNet 的权重序列化后可能几十到几百 MB所有节点都去存一遍内存和带宽立刻爆掉。所以成熟的 DAG 联邦学习框架都采用一个默认约定链上只放模型摘要链下走 P2P 传输具体权重。这里有个细节容易被忽略——「模型哈希」怎么算。直接用md5(pickle.dumps(model.state_dict())) 在跨 Python 版本、跨库版本的环境里会不稳定因为 PyTorch 的浮点序列化在不同版本之间存在微小差异。常见的做法是先把权重张量映射到低精度表示比如转成 float16 后做哈希或者只在整数化的量化权重上做校验。否则你会遇到一个很诡异的玄学问题同一个模型在两个环境里算出的哈希不同账本认为更新无效。import hashlib import torch import time dataclass class ModelUpdateTx: tx_id: str client_id: str round_no: int model_hash: str # 摘要哈希 ip: str # 链下权重下载地址 port: int data_size: int # 本地训练样本数 acc: float # 本地验证集准确率 timestamp: float field(default_factorytime.time) def make_update_tx(client_id: str, round_no: int, state_dict, ip: str, port: int, data_size: int, acc: float): # 关键把权重统一到 float16 再哈希避免版本差异导致的哈希漂移 hasher hashlib.sha256() for k, v in state_dict.items(): hasher.update(k.encode(utf-8)) hasher.update(v.detach().cpu().to(torch.float16).numpy().tobytes()) model_hash hasher.hexdigest() # 交易ID由哈希和轮次共同决定保证同一轮内同一模型的更新可以统一去重 tx_id hashlib.sha256(f{client_id}:{round_no}:{model_hash}.encode()).hexdigest() return ModelUpdateTx(tx_idtx_id, client_idclient_id, round_noround_no, model_hashmodel_hash, ipip, portport, data_sizedata_size, accacc)交易 ID 的生成规则值得单独说。很多初版实现直接把 model_hash 当 tx_id后果就是同一个客户端在相邻两轮产生了两笔哈希相同的交易被 DAG 判重丢弃训练直接卡死。我习惯把 client_id、round_no、model_hash 三者拼接后再哈希一遍这样每一笔更新都是唯一可追踪的同时链上天然保留了一条「同一个客户端从哪个模型演化到哪个模型」的血缘线索。3. 去中心化联邦最小闭环用 Python 写出从更新生成到确认的完整流程3.1 节点发现与交易池先解决“彼此怎么找到对方”的问题去中心化不等于没有基础设施。节点的互相发现方式决定了网络的规模和容错能力。最常见的做法是 bootstrap 列表加动态扩展——每个节点内置一个固定种子节点列表启动后向种子节点索要当前在线节点地址再主动连接其中一部分。如果你只是在自己电脑上做模拟实验那用 localhost 多端口直接模拟多个客户端就够了。每个节点本地需要维护两个数据结构一个是已知账本confirmed_txs另一个是待验证交易池pending_pool。收到新交易后不要立刻放进账本先放进 pending_pool等它引用的父交易都已确认才把它也确认。这个延迟确认机制是 DAG 体系里防止「引用不存在的交易」的关键。class Node: def __init__(self, node_id: str, ledger: DAGLedger): self.node_id node_id self.ledger ledger self.pending_pool {} self.confirmed_txs {} def receive_tx(self, tx: ModelUpdateTx): # 第一步验签和基本字段校验 if tx.round_no 0: raise ValueError(fround_no必须为正整数收到{tx.round_no}) if tx.tx_id in self.confirmed_txs: return # 重复交易直接忽略 # 第二步放进pending池等待父交易确认 self.pending_pool[tx.tx_id] tx def try_confirm(self): # 循环扫描pending池pua直到这一轮没有任何交易能确认 progressed True while progressed: progressed False for tx_id in list(self.pending_pool.keys()): tx self.pending_pool[tx_id] parents tx.parent_ids if all(pid in self.confirmed_txs for pid in parents): self.ledger.add_tx(tx) self.confirmed_txs[tx_id] tx del self.pending_pool[tx_id] progressed True这里有个参数值得反复检查pending_pool 的容量上限。如果网络中存在大量低质量更新pending_pool 会无限制膨胀内存被拖垮。我一般会给 pending_pool 设置 max_pending5000超过上限时按时间戳丢弃最早进入的交易并打一条 WARNING 日志——因为这种异常通常意味着网络里有节点在作恶刷交易而不是正常负载。3.2 本地聚合FedAvg 在 DAG 里的权重计算方式确认了一批父交易之后节点要做的是把这些模型更新聚合出「自己的全局视图」。注意这个词——在去中心化框架里每个节点的视角不完全一样因为不同节点确认的交易集合可能不同。所以这里本质上不再有「唯一全局模型」而是「每个节点收敛到同一批主流更新的聚合结果」最终靠 DAG 的累计权重让共识浮现。聚合权重最常见的选择是样本数加权。假设你确认了 k 笔更新每笔更新的本地样本数为 n_i聚合公式就是联邦学习里经典的 FedAvgdef federated_average(state_dicts, weights): # state_dicts: list of model.state_dict() # weights: list of int对应每个模型的本地样本数 total_weight sum(weights) avg_state {k: torch.zeros_like(v) for k, v in state_dicts[0].items()} for sd, w in zip(state_dicts, weights): for k in avg_state: avg_state[k] sd[k] * w / total_weight return avg_state这里容易翻车的地方是torch.zeros_like处理 BatchNorm 统计量。BatchNorm 的 running_mean 和 running_var 是全局统计量直接按样本数加权会有偏差。常见做法是在聚合时把这些 key 剔掉等模型在本地数据上重新 forward 一遍校准。对于图像分类任务这个偏差不明显但一旦切到语义分割或者带 batch norm 的生成模型你会发现聚合出来的模型在验证集上直接不收敛排查半天才发现是 BN 统计量的问题。3.3 交易传播与确认深度什么时候可以信任一笔更新DAG 账本没有「出块」这个概念一笔交易被确认靠的是后续交易不断引用它累计权重爬到一定阈值。在联邦学习场景里设置「确认深度」比单纯看累计权重更直观——确认深度 该交易的后代中距离它最远的一条引用路径的层数。def confirmation_depth(ledger: DAGLedger, tx_id: str, max_depth: int 50) - int: # BFS找最长引用链深度 depth 0 queue [(tx_id, 0)] visited set() while queue: current, d queue.pop(0) if current in visited: continue visited.add(current) depth max(depth, d) if d max_depth: break for observed_tx in ledger.txs.values(): if tx_id in observed_tx.parent_ids: queue.append((observed_tx.tx_id, d 1)) return depth确认深度在实验里通常设为 3 或 5。深度 3 意味着这笔更新已经被三轮后续更新间接验证在 100 节点规模下基本可以认为不容易被回滚。深度设太大比如 20会让聚合延迟高到失去实用价值很多实际框架采用动态阈值网络早期节点少深度 2 就确认节点多了以后自动提升到 5。你可以把这个逻辑做成可配置参数而不是在代码里写死。4. 个性化到底怎么做全局共识与本地模型的软分离4.1 数据异构带来的「一模型吃天下」困局中心化联邦学习有一个不常被摆到台面上的假设所有客户端的数据分布大致一致全局模型才可能对所有客户端都好用。真实场景里这条假设基本不成立。一个手机输入法模型面对的是不同打字习惯的用户一个医疗影像模型面对的是不同设备厂商和不同疾病的图像分布。全局模型在 A 设备上效果不错在 B 设备上可能直接掉点。个性化联邦学习就是在训练目标里把「全局收敛」和「本地适应」同时纳入考量。常见路线有三条第一条是本地微调用全局模型做初始化再在本地数据上多训练几个 epoch第二条是模型混合训练一个全局模型和一个本地模型推理时按权重融合第三条是多任务学习把每个客户端看作一个任务共享底层特征学习不同任务专属的分类层。标题里「个性化」这个词对应的通常是前两条路线的组合因为它最容易被集成进 DAG 更新流程——你已经有了全局聚合模型只要再加一步本地精调。4.2 基于 DAG 框架的个性化流程聚合之后做 Local Update在 DAG 联邦框架里个性化的实现位置很清晰每个节点在「确认父交易 → 聚合得到全局视图」之后不要立刻把聚合结果广播出去而是先在自己的本地数据上做几轮微调再生成新的更新交易。这样账本里流通的不再是同一个标准化全局模型而是每个节点带有本地偏好的「视图」。def local_finetune(global_state, local_loader, local_epochs, lr, devicecpu): model create_model() model.load_state_dict(global_state) optimizer torch.optim.SGD(model.parameters(), lrlr, momentum0.9) model.train() for epoch in range(local_epochs): for x, y in local_loader: x, y x.to(device), y.to(device) optimizer.zero_grad() loss torch.nn.functional.cross_entropy(model(x), y) loss.backward() optimizer.step() return model.state_dict()这里的local_epochs是全局模型与本地模型差异程度的控制阀。设 1 到 3 轮效果最好——你只是把全局模型「拉向」本地数据分布而不是完全重训。如果设到 10 以上每个节点都会得到一个严重过拟合本地的模型后续聚合时互相扯后腿全局收敛直接崩掉。另一个关键参数是lr本地微调时要比正常训练小一到两个数量级常见做法是全局学习率的 0.1 倍因为你在一个已经比较光滑的损失曲面附近做局部搜索步子太大会跳出好区域。4.3 混合权重在个性化与全局一致性之间做权衡只有本地微调往往不够因为纯微调之后模型参数偏离全局共识太远下一轮聚合时你没资格去指导别人。所以生产级框架都会在发布更新之前把本地微调模型和聚合全局模型做一次线性插值。这就是个性化联邦学习论文里最常见的 PFA 思路新模型 alpha * 全局模型 (1 - alpha) * 本地微调模型。def mix_models(global_state, local_state, alpha0.6): mixed {} for k in global_state.keys(): mixed[k] alpha * global_state[k] (1.0 - alpha) * local_state[k] return mixedalpha是极度敏感的参数。alpha 太高个性化形同虚设每个客户端广播的还是同一个模型的微小变体alpha 太低全局信息被丢弃DAG 账本里流通的模型群体分化成互不相关的孤岛聚合失去意义。我通常的做法是先离线跑一个小实验固定其他条件分别取 alpha 0.3 / 0.5 / 0.7比较每个客户端本地测试集准确率的均值与方差。如果你的业务场景要求「所有客户端平均表现最优」选均值最高的一组如果要求「不能有任何一个客户端掉点超过 5 个百分点」就比较各方差的 90 分位。4.4 血缘追踪让 DAG 记录每个个性化模型的历史路径个性化之后账本上的模型更新不再只有一个标准答案你必须让网络有能力判断「这笔更新的家谱是什么样的」。做法是在交易信息里增加一个字段列表记录该模型由哪些祖先交易聚合而来。这个字段可以直接放进 model_update 数据里不需要额外做密码学承诺因为每条交易都已经有了模型哈希血缘追踪只需要追溯父交易链即可。{ tx_id: 7e3f9a..., client_id: client_17, round_no: 12, model_hash: b3c8d1..., parent_ids: [a1b2c3..., d4e5f6...], mix_alpha: 0.6, local_epochs: 2, local_data_size: 240, local_acc: 0.913 }从这张结构化记录里你可以算出很多有用的东西哪些客户端持续在给其他人提供高质量模型、哪些交易虽然被引用很多但贡献度很低、某个客户端是不是从第 5 轮开始带偏了整个账本。把这些统计值接入 DAG 的 tip 选择权重计算里就是下一代去中心化联邦学习在攻防和贡献度量化方向上的核心玩法。5. DAG 联邦框架落地避坑五个我踩过的坑与排查记录5.1 孤儿交易堆积tip 数量指数级膨胀现象训练跑了几十轮之后节点内存持续走高账本里 tip 数量从 3 变成 30新交易的父引用散落在各个分支上谁也等不到确认。原因tip 选择的 decay 参数太大每条新交易都倾向去引用累计权重最高的旧 tip导致少数 tip 被反复引用而另外一些更新一直在账本边缘等不到第二个引用。这些边缘交易就是「孤儿交易」它们死不掉也活不了。解决把 decay 从 0.9 调到 0.75 附近同时给账本加一个最大引用等待轮次。超过 10 轮未被引用的 tip 直接标记为 abandonedtip 选择时不考虑它们。代码里只需要在 select_tip 函数里加一个is_active过滤字段即可。5.2 聚合模型震荡全局准确率上不去现象客户端准确率曲线不是稳步上升而是每轮大起大落幅度超过 10 个百分点。原因数据异构程度高时等权重的 Federated Average 会把所有客户端的梯度方向强行平均极端情况下两个分布差异大的客户端互相抵消聚合结果变成一个谁都不适用的中间模型。这类问题在中心化 FL 里也存在在 DAG 场景下更严重因为每个节点的聚合视图不完全一致震荡会沿着引用链递归放大。解决放弃纯样本数加权改用「样本数 相似度加权」。计算每个本地模型与聚合前一轮全局模型在验证集上的输出分布 KL 散度KL 过大就降低该更新在聚合时的权重。KL 过大的交易同时会被 DAG 网络标记为低置信更新confirm 速度变慢等于给账本引入了动态的信用体系。5.3 旧轮次交易被重复广播模型回退现象账本里出现两个来自同一客户端、轮次不同但模型哈希相同的历史更新后续交易同时引用了它们聚合结果被拉回到几个版本之前。原因客户端在断线重连后把本地存储的交易记录重新广播了一遍新节点没有做好去重——只按 tx_id 去重而 tx_id 里如果没包含 round_no 就无法区分新旧。解决tx_id 生成必须包含 round_no同时节点在 receive_tx 阶段增加单调契约检查同一个 client_id 下新交易引用的父交易 round_no 不能低于自己已经确认过的该客户端的最高轮次。违反这个契约的交易直接丢弃。这是所有基于 DAG 的联邦框架都必须要做的一道闸门否则回放攻击会变成实际风险。5.4 交易池无限增长内存被拖垮现象运行 4 小时后节点内存占用从 200MB 涨到 4GB最终 OOM 崩溃。原因pending_pool 收到了大量未被确认的交易而 try_confirm 扫描循环里对每笔交易都遍历全部账本儿子节点找引用关系复杂度是 O(N²)既占内存又占 CPU。解决两层手段并行。第一层限制 pool 容量满了丢弃最旧的未确认交易第二层给每笔待确认交易设置超时时间超过 30 秒父交易仍未到齐的直接从 pool 里摘除。同时给 DAG 账本加定时快照压缩——每 100 轮把确认过的交易合并成状态快照历史交易体只保留哈希索引不再保留完整字段。5.5 本地微调后模型冲突奇怪的是每个客户端单独评测都涨点聚合评测全面下降现象个性化阶段一切正常每个客户端本地准确率都有提升但下一轮聚合后全局模型在标准测试集上明显退化。原因本地微调的 epoch 过多、混合系数 alpha 过低导致每个客户端广播出去的模型严重偏向本地分布。聚合时这些偏斜模型互相抵消模型走向了所有客户端的「分布均值」而不是「任务共性」。解决把 local_epochs 控制在 1 到 2alpha 下限设为 0.5。另外在生成更新交易时记录本地微调前后在本地验证集上的准确率差 delta_acc如果 delta_acc 超过 15 个百分点就强制把这个模型不广播只保留本地使用。这个策略本质上就是给个性化加了一个保险丝个人可以走得快但不能把整个网络带偏。6. 进阶验证不给源码也能先跑通 DAG 联邦的关键技巧拿到一份只带项目说明的源码包最有价值的验证方式不是把它当黑匣子直接跑而是启动之前先在本地构造一个最小模拟环境搞清楚三件事交易是否能够在多个节点之间形成并行引用、累计权重能否稳定增长并完成确认、个性化参数对账本形态的影响是正向还是负向。这三件事全部可以通过 6 行以内的辅助代码完成验证。我常用的小技巧是「一个账本、三个客户端」的模拟实验。开三个 Python 进程分别模拟三个客户端共享一个本地 JSON 文件作为 DAG 账本存储。每个客户端做完本地训练后读取账本文件、选 tip、写新交易、用flock加文件锁防止并发写冲突。这个做法的好处在于你可以直观地打开 JSON看到每一笔交易的 parent_ids 是怎么交织的tip 集合长什么样哪个模型是网络上当前最受认可的——比任何可视化工具都直观。验证收敛性的一个核心指标是「聚合模型在固定测试集上的方差」。如果训练 30 轮后三个客户端各自聚合出来的全局视图差异收敛到 1% 以内说明 DAG 账本已经形成了有效共识如果方差始终在 5% 以上说明 tip 选择策略或确认深度参数还有问题不要急着往上加功能。另外一个经验是关于源码的拿到别人的 DAG 联邦实现我做的第一件事永远是去改两行参数——把网络规模调到最小、把通信延迟调到零先跑通数据通路再逐项加入网络噪声。这种从简到繁的推进方式能帮我快速判断问题出在算法层还是工程层。多年做分布式系统的血泪经验是大部分翻车现场都不是原理错了而是某个节点在某个时刻悄悄地把交易丢了或者某个内存缓存没有及时释放。DAG 联邦学习框架的复杂度是中心和分布式协议叠加出来的每加一个模块都要有对应的可观测性指标否则你会在数百个节点同时广播更新的那一刻陷入无从下手的状态。希望这篇笔记能帮你把 DAG 区块链和联邦学习的结合点看得更清晰也预祝你在本地跑通第一版模拟器的时候账本形态能像你预期的那样有条理地生长。本文还有配套的精品资源点击获取