1. 联邦学习到底在解决什么问题联邦学习Federated Learning这个词听起来挺唬人但核心直觉只有一句话数据不动模型动。多个客户端各自在本地用自己的数据训练模型只把模型更新梯度或权重上传到中心服务器服务器聚合后再把新模型发回去。整个过程原始数据始终留在本地谁也不用把数据交出去。它能做什么最典型的场景是几家医院想联合训练一个诊断模型但谁也不能把病人数据给出去几家银行想联合做风控但客户信息属于核心资产。联邦学习让这些机构在不共享原始数据的前提下共同得到一个比单打独斗更好的全局模型。适合谁看这篇如果你刚接触联邦学习看了一堆论文里的公式还是不知道从哪下手那这篇就是写给你的。我不打算堆数学推导而是带你在单机环境里模拟多个客户端跑通一次最小可运行的联邦学习实验。你会拿到可复制的settings.json和config.toml骨架会看到怎么用 TaoToken 的统一 Key 和 API 通道来接入模型服务最后还会做一次检查动作确认各客户端的更新确实被正确聚合了。为什么要在单机模拟因为真实的多机联邦环境搭建成本高网络、权限、依赖问题一堆。单机模拟能让你先把流程跑通客户端怎么初始化、本地训练怎么触发、更新怎么上传、服务器怎么聚合。流程清楚了再扩展到多机就是换配置的事。这里有个容易混淆的点先讲清楚联邦学习里的“客户端”不一定是一台独立机器它可以是一个进程、一个数据分片。单机模拟就是用多个进程或线程扮演多个客户端各自持有不同的数据子集走完整的训练-上传-聚合-下发循环。理解了这一点后面的配置就好懂了。2. TaoToken 统一 Key 的前置准备在动手写配置之前先把模型服务的接入通道准备好。联邦学习实验里每个客户端在本地训练时可能需要调用模型推理或微调接口如果每个客户端都单独配一套 Key管理起来很乱。TaoToken 的思路是提供一个统一的 API 通道你只需要一个 Key就能在多个客户端进程里复用同一套接入配置。先注册并拿到 Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 完成账号注册后进入控制台。控制台地址是 https://taotoken.net/console 在 API Keys 页面可以创建新的 Key。创建时建议给 Key 起一个能区分用途的名字比如fl-sim-local这样后面在多个客户端配置里引用时不容易搞混。拿到 Key 之后API 的基础地址是 https://taotoken.net/api 注意这个地址不带任何查询参数直接作为 base_url 使用。如果你用的是 OpenAI 兼容的 SDK把 base_url 指向它、api_key 填你创建的 Key 就行。这里有个实操细节联邦学习模拟里通常会有 3 到 5 个客户端进程它们可以共用同一个 Key因为 Key 是绑定账号的不是绑定进程的。但要注意并发调用时的速率限制如果你在本地同时起 5 个客户端做推理建议在配置里加上重试和退避逻辑避免某个客户端因为瞬时并发被限流而训练中断。如果你后续要做更长期的编码或 Agent 类实验可以了解下 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。它适合需要持续调用、频繁调试的场景。而如果只是想先验证模型对话是否通可以用模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 快速试一下。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各语言 SDK 的调用示例。API Keys 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 建议收藏后面换 Key 或查用量都用得上。3. 可复制的 settings.json 与 config.toml 骨架现在进入正题把配置骨架搭起来。我设计的结构是一个settings.json管全局参数和 TaoToken 接入信息一个config.toml管联邦学习实验本身的参数客户端数量、训练轮数、聚合策略等。两个文件放在项目根目录客户端进程启动时都读这两个文件。先看settings.json{ taotoken: { base_url: https://taotoken.net/api, api_key: sk-your-key-here, model: gpt-4o-mini, timeout: 30, max_retries: 3, retry_backoff: 1.5 }, federated: { num_clients: 3, local_epochs: 2, batch_size: 32, learning_rate: 0.01, aggregation: fedavg, rounds: 5 }, logging: { level: INFO, log_dir: ./logs, save_model_updates: true } }几个关键字段说明。base_url固定填https://taotoken.net/api不要加斜杠结尾。api_key换成你在控制台创建的那串。model按你实际要调用的模型名填。max_retries和retry_backoff是给并发场景兜底的客户端多的时候很有用。federated段里num_clients设成 3表示单机模拟 3 个客户端。local_epochs是每个客户端本地训练几轮设小一点方便快速看到聚合效果。aggregation填fedavg这是最常用的联邦平均算法。rounds是全局聚合轮数5 轮足够你观察 loss 变化趋势。再看config.toml[experiment] name fl-local-sim seed 42 device cpu [data] dataset synthetic num_samples_per_client 500 num_features 20 num_classes 2 non_iid true dirichlet_alpha 0.5 [client] local_train_script client_train.py upload_format json compress_updates false [server] aggregate_script server_aggregate.py min_clients_per_round 2 dropout_tolerance 0.1 [checkpoint] save_every_round true checkpoint_dir ./checkpointsnon_iid true和dirichlet_alpha 0.5是为了模拟非独立同分布数据这是联邦学习里很真实的场景——每个客户端的数据分布不一样。如果你想让实验更简单可以先设non_iid false跑通流程再改成 true 观察聚合难度。min_clients_per_round 2表示至少要有 2 个客户端上传了更新才触发聚合这是防止某个客户端掉线导致聚合失败的保护机制。两个文件建好后目录结构大概是这样fl-local-sim/ ├── settings.json ├── config.toml ├── client_train.py ├── server_aggregate.py ├── data/ └── logs/4. 客户端训练与聚合的接入写法配置有了接下来写客户端训练脚本。核心逻辑是每个客户端读取自己的数据分片本地训练若干轮然后把模型更新序列化成 JSON 上传到聚合器。这里用 TaoToken 的 API 通道来做模型推理部分的调用。先写一个公共的 TaoToken 客户端封装import json import time import requests class TaoTokenClient: def __init__(self, settings_pathsettings.json): with open(settings_path, r) as f: cfg json.load(f) self.base_url cfg[taotoken][base_url] self.api_key cfg[taotoken][api_key] self.model cfg[taotoken][model] self.timeout cfg[taotoken][timeout] self.max_retries cfg[taotoken][max_retries] self.backoff cfg[taotoken][retry_backoff] def chat(self, messages): url f{self.base_url}/v1/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } payload { model: self.model, messages: messages, temperature: 0.2 } for attempt in range(self.max_retries): try: resp requests.post(url, headersheaders, jsonpayload, timeoutself.timeout) resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: if attempt self.max_retries - 1: raise wait self.backoff ** attempt time.sleep(wait)这个封装做了三件事从settings.json读配置、拼装请求、带退避的重试。注意 URL 是base_url加上/v1/chat/completions这是 OpenAI 兼容格式的路径。然后是客户端训练脚本client_train.pyimport json import numpy as np from taotoken_client import TaoTokenClient def load_client_data(client_id, config): np.random.seed(config[experiment][seed] client_id) n config[data][num_samples_per_client] d config[data][num_features] X np.random.randn(n, d) y (X[:, 0] X[:, 1] 0).astype(int) return X, y def local_train(X, y, global_weights, lr, epochs): w np.array(global_weights) for _ in range(epochs): preds X w grad X.T (preds - y) / len(y) w - lr * grad return w.tolist() def main(client_id): import toml with open(config.toml, r) as f: config toml.load(f) with open(settings.json, r) as f: settings json.load(f) X, y load_client_data(client_id, config) global_weights [0.0] * config[data][num_features] updated local_train( X, y, global_weights, settings[federated][learning_rate], settings[federated][local_epochs] ) update_payload { client_id: client_id, weights: updated, num_samples: len(y) } with open(flogs/client_{client_id}_update.json, w) as f: json.dump(update_payload, f) print(fclient {client_id} update saved, norm{np.linalg.norm(updated):.4f}) if __name__ __main__: import sys main(int(sys.argv[1]))这个脚本做了本地训练并保存更新。num_samples字段很重要FedAvg 聚合时要按样本数加权平均。聚合脚本server_aggregate.pyimport json import glob import numpy as np def fedavg(update_files): all_weights [] all_counts [] for fp in update_files: with open(fp, r) as f: data json.load(f) all_weights.append(np.array(data[weights])) all_counts.append(data[num_samples]) counts np.array(all_counts, dtypefloat) weights np.array(all_weights) weighted np.average(weights, axis0, weightscounts) return weighted.tolist() def main(): files sorted(glob.glob(logs/client_*_update.json)) if len(files) 2: print(not enough client updates, skip aggregation) return global_weights fedavg(files) with open(logs/global_model.json, w) as f: json.dump({weights: global_weights, round: 1}, f) print(faggregated {len(files)} clients, global norm{np.linalg.norm(global_weights):.4f}) if __name__ __main__: main()跑一轮的流程是先启动 3 个客户端进程各自生成更新文件然后运行聚合脚本读取所有更新文件做加权平均输出全局模型。5. 验证请求与聚合结果检查配置和脚本都就位后跑一次完整流程验证。先起客户端python client_train.py 0 python client_train.py 1 python client_train.py 2每个客户端会输出类似client 0 update saved, norm2.3456的日志。三个都跑完后检查logs/目录下是否生成了三个更新文件ls logs/client_*_update.json应该看到client_0_update.json、client_1_update.json、client_2_update.json。然后跑聚合python server_aggregate.py输出应该是aggregated 3 clients, global norm...。这时候打开logs/global_model.json能看到聚合后的权重向量。怎么确认聚合是正确的做两个检查动作。第一手动算一下加权平均和脚本输出对比import json import numpy as np files [logs/client_0_update.json, logs/client_1_update.json, logs/client_2_update.json] ws, cs [], [] for fp in files: d json.load(open(fp)) ws.append(d[weights]) cs.append(d[num_samples]) manual np.average(np.array(ws), axis0, weightsnp.array(cs, dtypefloat)) auto json.load(open(logs/global_model.json))[weights] print(max diff:, np.max(np.abs(manual - np.array(auto))))如果max diff接近 0比如小于 1e-10说明聚合逻辑没问题。第二检查每个客户端的更新是否真的不同。如果三个客户端的权重完全一样那说明数据分片没生效或者随机种子冲突了。可以打印每个客户端更新向量的前几个值对比python -c import json for i in range(3): d json.load(open(flogs/client_{i}_update.json)) print(fclient {i}:, [round(x, 4) for x in d[weights][:5]]) 正常情况下三个客户端的输出应该有明显差异因为它们的本地数据不同。如果你想验证 TaoToken 通道本身是否通可以单独跑一个最小请求from taotoken_client import TaoTokenClient client TaoTokenClient() resp client.chat([{role: user, content: reply with ok}]) print(resp[choices][0][message][content])返回ok就说明 Key 和通道都正常。这一步建议在跑联邦实验之前先做避免训练脚本报错时分不清是通道问题还是逻辑问题。6. 本篇常见错误排查第一个高频错误是401 Unauthorized。原因通常是settings.json里的api_key没换成你自己的或者复制时带了多余空格。检查方法是把 Key 打印出来看首尾有没有空白字符。另外确认base_url是https://taotoken.net/api不要写成带/v1的地址路径拼接逻辑已经在代码里处理了。第二个是ConnectionError或超时。如果你在客户端并发跑的时候遇到先降低并发数把num_clients从 3 改成 2 试试。同时确认max_retries和retry_backoff配置生效了。如果公司网络有出口限制确认taotoken.net可以正常访问。第三个是聚合时报not enough client updates。这说明logs/目录下的更新文件少于 2 个。检查每个客户端进程是否都正常退出有没有报错但被忽略。可以单独跑一个客户端看它是否生成了文件。另外注意文件路径脚本里用的是相对路径logs/如果你在别的目录执行脚本路径就对不上了。第四个是ModuleNotFoundError: No module named toml。这是缺依赖装一下就行pip install toml requests numpy第五个是聚合结果和手动计算对不上。最常见的原因是浮点精度问题如果max diff在 1e-8 量级可以忽略。如果差得很多检查num_samples字段是否被正确读取以及np.average的weights参数是否传对了。第六个是客户端更新完全一样。检查load_client_data里的随机种子seed client_id确保每个客户端种子不同。如果用的是真实数据集检查数据分片逻辑是不是把同一份数据发给了所有客户端。排障时如果怀疑是 Key 或通道问题直接去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 确认 Key 状态或者翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 对照请求格式。模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 可以快速验证模型是否可用。7. 把实验扩展到更真实的场景单机模拟跑通之后你可以做几件事让它更接近真实联邦学习。第一把non_iid打开观察 FedAvg 在数据分布不均衡时的收敛速度变化。第二增加rounds看全局模型的 loss 是否持续下降。第三把客户端更新加上简单的差分隐私噪声体验一下隐私保护对模型精度的影响。如果你要长期跑这类实验或者把联邦学习逻辑接入到更大的 Agent 工作流里Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 会比按次调用更省心。它的定位就是给需要持续编码和调试的场景用的。最后提醒一个实操细节每次改完settings.json或config.toml记得清空logs/目录再重跑否则旧的更新文件会混进聚合结果里导致你看到的数据对不上。这个坑我踩过排查了半天才发现是残留文件的问题。
