企业官网建设流程全解析

1. 联邦学习入门：本地训练 + 只传更新的最小闭环

联邦学习（Federated Learning, FL）说白了就是一句话：数据不动，模型动。多个参与方各自在本地用自己的数据训练模型，只把梯度或权重更新上传到中心服务器，服务器用 FedAvg 之类的算法聚合出一个全局模型，再发回去继续下一轮。整个过程原始数据始终留在本地，适合医疗、金融、IoT 这类隐私敏感场景。

我第一次接触 FL 的时候，最困惑的不是 FedAvg 的公式，而是"怎么证明它真的没上传原始数据"。后来发现，光看论文没用，得自己跑一轮，把上传的 payload 打印出来看看到底传了什么。这篇就按这个思路来：先搭一个本地模拟的多参与方环境，跑通一轮 FedAvg 聚合，检查上传的梯度里到底有没有原始数据，最后把模型调用统一走 TaoToken 的 API 通道，方便后续接入真实的大模型做联邦微调实验。

适合谁看：刚入门联邦学习、想动手跑通一轮聚合、又不想在环境配置上卡半天的同学。你不需要 GPU，一台普通笔记本就能跑完本文所有代码。核心检索词就是"联邦学习 FedAvg 本地模拟"，跟着做大概 20 分钟能出结果。

先说清楚本文的边界：我们做的是本地模拟，不是真实跨机构部署。参与方是同一台机器上的多个进程，数据是人工切分的。这样做的目的是把 FL 的通信协议和聚合逻辑看清楚，等这套跑通了，再换成真实网络通信只是替换传输层的事。

环境准备很简单，Python 3.9+，装三个包：

pip install torch numpy requests

PyTorch 用来做本地训练，numpy 做聚合计算，requests 用来调 TaoToken 的 API。如果你已经有 conda 环境，直接在里面装就行，不用新建。

这里有个容易踩的坑：很多人一上来就想搞"真实分布式"，开好几个 Docker 容器互相通信，结果卡在网络配置上三天没跑通一轮。我的建议是先单机模拟，把 FedAvg 的加权平均逻辑、梯度上传格式、聚合后的模型分发这三件事搞明白，再考虑分布式。顺序反了会浪费大量时间。

下面我会分几步走：先讲清楚 TaoToken 统一 Key 在这里扮演什么角色，再给可复制的配置和代码，然后跑一轮验证，最后把常见报错列出来。每一步都有完整命令和预期输出，你可以直接复制粘贴。

2. TaoToken 统一 Key 与 API 通道前置准备

在联邦学习里，中心服务器除了做 FedAvg 聚合，往往还需要调用大模型来做一些辅助任务，比如对聚合后的模型做效果评估、生成训练报告、或者用 LLM 做参与方的异常检测。这时候如果每个参与方各自去申请模型 API Key，管理起来会很乱。TaoToken 的统一 Key 就是解决这个问题的：一个 Key 走统一通道，所有参与方的模型调用都从这里过，计费和权限也好统一管理。

TaoToken 是什么？它是一个统一的模型 API 接入通道，你拿一个 Key 就能调用多种模型，不用为每个模型单独配一套鉴权和地址。对联邦学习场景来说，它的价值在于：中心服务器可以用同一个 Key 去调模型做评估，参与方如果需要本地调用模型做预处理，也能复用这套通道，省去重复配置。

适合谁用：做 FL 实验需要频繁调模型做评估、又不想在 Key 管理上花时间的同学。你只需要在中心服务器配一次，参与方通过环境变量读取即可。

前置准备分三步。第一步，拿到统一 Key。访问 https://taotoken.net/api-keys 创建你的 API Key，注意这个 Key 只在创建时显示一次，复制下来存好。第二步，确认 API 基础地址是 https://taotoken.net/api，所有请求都走这个地址。第三步，把 Key 写进环境变量，不要硬编码在代码里：

export TAOTOKEN_API_KEY="你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

Windows 用户用set或者直接在系统环境变量里配。配完之后验证一下：

echo $TAOTOKEN_API_KEY

能打印出你的 Key 就说明环境变量生效了。这一步看着简单，但我见过不少人因为 shell 会话没刷新导致后面请求一直 401，所以务必先确认。

关于模型选择，联邦学习实验里常用的评估模型可以用 claude 系列或者 gpt 系列，具体在 https://taotoken.net/models 看当前可用的模型 ID。你需要在配置里填对 Model ID，这个后面代码里会用到。

这里要强调一点：TaoToken 是统一 API 通道，不是让你把训练数据传上去。联邦学习的核心原则是数据不出本地，TaoToken 在这里只承担模型调用的角色，比如聚合后让模型评估一下全局模型的输出质量。数据流向和模型调用是两回事，别混了。

如果你打算长期做 FL 实验，建议看一下 Coding Plan，它适合需要持续调用模型做实验的场景，比按次调用更划算。入口在 https://taotoken.net/coding-plan。

3. 可复制的本地模拟配置与 FedAvg 代码

这一节是核心，给你一套能直接跑的配置和代码。目录结构建议这样：

fl_demo/ ├── config.json ├── client.py ├── server.py └── data/

先写配置文件config.json，把 TaoToken 的接入信息和 FL 参数都放进去：

{ "taotoken": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "model_id": "claude-3-5-sonnet-20241022" }, "fl": { "num_clients": 3, "rounds": 1, "local_epochs": 2, "lr": 0.01, "batch_size": 16 } }

注意api_key_env写的是环境变量名，不是 Key 本身，这样配置文件可以安全地提交到仓库。model_id你按 https://taotoken.net/models 上实际可用的填。

接下来是参与方（client）的代码client.py。每个 client 在本地用自己的数据训练，然后返回模型更新：

import json import copy import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset class SimpleModel(nn.Module): def __init__(self, in_dim=10, hidden=16, out_dim=2): super().__init__() self.net = nn.Sequential( nn.Linear(in_dim, hidden), nn.ReLU(), nn.Linear(hidden, out_dim) ) def forward(self, x): return self.net(x) def make_local_data(seed, n=200): torch.manual_seed(seed) x = torch.randn(n, 10) y = (x.sum(dim=1) > 0).long() return TensorDataset(x, y) def local_train(model, dataset, epochs=2, lr=0.01, batch_size=16): loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) optimizer = torch.optim.SGD(model.parameters(), lr=lr) criterion = nn.CrossEntropyLoss() model.train() for _ in range(epochs): for xb, yb in loader: optimizer.zero_grad() loss = criterion(model(xb), yb) loss.backward() optimizer.step() return model def get_update(global_state, client_id, cfg): model = SimpleModel() model.load_state_dict(copy.deepcopy(global_state)) dataset = make_local_data(seed=client_id) local_train(model, dataset, epochs=cfg["local_epochs"], lr=cfg["lr"], batch_size=cfg["batch_size"]) update = {k: v.clone() for k, v in model.state_dict().items()} return update, len(dataset)

这段代码的关键点：get_update返回的是模型参数（state_dict），不是数据。你可以打印一下update的内容，里面全是权重张量，没有任何原始样本。这就是"只上传模型更新"的字面含义。

然后是服务器端server.py，负责初始化全局模型、收集更新、做 FedAvg 加权平均：

import json import copy import torch from client import SimpleModel, get_update def fedavg(updates, sizes): total = sum(sizes) avg = {} for k in updates[0].keys(): avg[k] = sum(u[k] * (s / total) for u, s in zip(updates, sizes)) return avg def main(): with open("config.json") as f: cfg = json.load(f)["fl"] global_model = SimpleModel() global_state = global_model.state_dict() for r in range(cfg["rounds"]): updates, sizes = [], [] for cid in range(cfg["num_clients"]): u, n = get_update(global_state, cid, cfg) updates.append(u) sizes.append(n) global_state = fedavg(updates, sizes) print(f"round {r+1} done, aggregated {len(updates)} clients") torch.save(global_state, "global_model.pt") print("saved global_model.pt") if __name__ == "__main__": main()

fedavg函数就是那个公式的代码实现：每个参与方的参数按样本量加权平均。sizes是各参与方的样本数，权重就是n_i / n。

跑起来：

cd fl_demo python server.py

预期输出：

round 1 done, aggregated 3 clients saved global_model.pt

到这里一轮 FedAvg 就聚合完了。整个过程数据都在make_local_data里生成，从没离开过 client 的进程。你可以把get_update里的update存下来看一眼，确认里面只有权重。

如果你想把聚合后的模型评估也走 TaoToken，可以在 server 里加一段调用，把全局模型的输出摘要发给模型做质量评估。这部分我放到下一节验证环节一起讲。

4. 验证请求：检查梯度是否含原始数据、确认走统一通道

跑通不等于验证通过。这一节做两件事：确认上传的梯度里没有原始数据，确认模型调用走的是 TaoToken 统一通道。

第一件事，检查梯度内容。在client.py的get_update里加一行调试输出：

def get_update(global_state, client_id, cfg): model = SimpleModel() model.load_state_dict(copy.deepcopy(global_state)) dataset = make_local_data(seed=client_id) local_train(model, dataset, cfg["local_epochs"], cfg["lr"], cfg["batch_size"]) update = {k: v.clone() for k, v in model.state_dict().items()} # 调试：打印上传内容的形状和数值范围 for k, v in update.items(): print(f"upload {k}: shape={tuple(v.shape)}, " f"min={v.min().item():.4f}, max={v.max().item():.4f}") return update, len(dataset)

再跑一次python server.py，你会看到类似：

upload net.0.weight: shape=(16, 10), min=-0.4821, max=0.5103 upload net.0.bias: shape=(16,), min=-0.0912, max=0.0876 ...

这些是权重张量，形状是网络层的维度，数值是训练后的参数。原始数据是 200 条 10 维样本，形状 (200, 10)，跟上传的 (16, 10) 完全对不上。这就从形状上证明了上传的不是原始数据。更严格的话，你可以尝试从梯度反推原始数据，会发现没有额外信息（比如差分隐私噪声）时反推难度很大，这也是 FL 隐私保护的基础。

第二件事，验证 TaoToken 通道。写一个独立的验证脚本verify_taotoken.py：

import os import json import requests with open("config.json") as f: cfg = json.load(f)["taotoken"] api_key = os.environ.get(cfg["api_key_env"]) assert api_key, "TAOTOKEN_API_KEY not set" url = f"{cfg['base_url']}/v1/messages" headers = { "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json" } payload = { "model": cfg["model_id"], "max_tokens": 64, "messages": [ {"role": "user", "content": "用一句话说明联邦学习为什么能保护隐私"} ] } resp = requests.post(url, headers=headers, json=payload, timeout=30) print("status:", resp.status_code) print("body:", resp.text[:300])

运行：

python verify_taotoken.py

预期返回 200，body 里是模型的一句话回答。如果返回 401，说明 Key 没配好或者环境变量没生效；如果返回 404，检查base_url是不是写成了带路径的形式，正确的基础地址是 https://taotoken.net/api，路径/v1/messages由代码拼接。

这里有个细节：不同模型的请求格式可能不一样，claude 系列用/v1/messages，gpt 系列用/v1/chat/completions。你在 https://taotoken.net/models 确认模型 ID 后，按对应格式发请求。文档在 https://taotoken.net/doc 有完整说明。

验证通过后，你就有了一个完整的闭环：本地训练 → 上传梯度 → FedAvg 聚合 → 统一通道调模型评估。数据没出本地，模型调用走统一 Key，两件事都确认了。

5. 本篇常见报错排查

这一节列几个我实际遇到过的报错，对照着排查能省不少时间。

报错一：401 Unauthorized

status: 401 body: {"error": {"type": "authentication_error", "message": "invalid api key"}}

原因通常是 Key 没配或配错。检查三件事：echo $TAOTOKEN_API_KEY能不能打印出 Key；Key 是不是从 https://taotoken.net/api-keys 复制的完整字符串；环境变量是不是在当前 shell 会话里生效的。如果你在 IDE 里跑代码，IDE 可能没继承 shell 的环境变量，需要在 IDE 的运行配置里单独设。

报错二：local proxy failed / connection refused

requests.exceptions.ProxyError: HTTPConnectionPool ... local proxy failed

这个报错说明你的请求被本地代理拦截了。检查环境变量里有没有HTTP_PROXY/HTTPS_PROXY，如果有，临时清掉：

unset HTTP_PROXY HTTPS_PROXY

然后重跑验证脚本。TaoToken 的 API 地址是直连的，不需要额外代理配置。

报错三：reading choices 相关解析错误

KeyError: 'choices'

这个通常发生在你用了 gpt 系列的请求格式去调 claude 模型，或者反过来。claude 的响应体里是content字段，gpt 是choices。检查你的model_id和请求路径是否匹配：claude 用/v1/messages，gpt 用/v1/chat/completions。改对之后重新请求。

报错四：OAuth / token 过期

{"error": {"type": "invalid_request_error", "message": "oauth token expired"}}

如果你用的是 OAuth 方式接入，token 有有效期，过期后需要重新获取。用 API Key 方式接入不会有这个问题。建议在 FL 实验里统一用 API Key，避免训练中途 token 失效导致评估中断。

报错五：FedAvg 聚合后模型不收敛

这个不是请求报错，但很常见。现象是聚合后的全局模型准确率比单个本地模型还低。原因通常是各参与方数据分布差异太大（Non-IID）。解决办法：增加参与方数量、调整本地 epoch 数、或者在聚合时按样本量加权（本文代码已经做了加权）。如果还是不行，可以引入个性化 FL，让每个参与方在全局模型基础上做本地微调。

排查顺序建议：先确认 Key 和地址（401/404），再确认网络（proxy），再确认请求格式（choices/content），最后才是算法层面的问题。大部分卡住的情况都在前三步。

6. 后续接入与统一通道实践

一轮 FedAvg 跑通之后，下一步通常是把模拟环境换成更接近真实的配置。这里给几个方向，都基于本文已经搭好的统一通道。

第一个方向：把参与方拆成独立进程。现在server.py是顺序调用get_update，你可以改成用multiprocessing或者起多个进程，每个进程加载自己的数据分片，通过文件或 socket 交换更新。传输层换了，但fedavg的聚合逻辑不用动。

第二个方向：接入真实模型做联邦微调。如果你想让参与方在本地用大模型做特征提取，可以把SimpleModel换成调用 TaoToken 的 embedding 接口，本地只存特征向量，梯度上传的还是模型参数。这样数据依然不出本地，但用上了大模型的能力。模型 ID 和接口格式在 https://taotoken.net/models 和 https://taotoken.net/doc 查。

第三个方向：加差分隐私。在get_update返回之前给梯度加高斯噪声：

import torch def add_dp_noise(update, sigma=0.01): return {k: v + torch.randn_like(v) * sigma for k, v in update.items()}

这样即使攻击者拿到梯度，也难反推原始数据。噪声大小sigma需要根据隐私预算调，太大影响模型效果，太小保护不够。

第四个方向：把评估流程自动化。每轮聚合后调一次 TaoToken，让模型对全局模型的输出做质量打分，记录到日志里。这样你能看到 FL 训练过程中模型效果的变化趋势。调用代码参考第 4 节的verify_taotoken.py，把 prompt 换成评估任务即可。

如果你打算长期做这类实验，Coding Plan 比按次调用更适合，入口在 https://taotoken.net/coding-plan。需要管理多个 Key 或者看用量，去 https://taotoken.net/console。API Key 创建在 https://taotoken.net/api-keys。

最后说一个我踩过的坑：一开始我把 TaoToken 的 Key 硬编码在config.json里，结果不小心提交到了公开仓库，只能重新生成。后来改成环境变量读取，配置文件里只留变量名，就安全多了。你做实验的时候也注意这点，Key 不要进版本控制。

整套跑下来，你应该已经理解了 FL 的核心：数据在本地，更新走网络，聚合在中心。TaoToken 的统一通道让模型调用这部分不用重复配置，把精力集中在 FL 算法本身。接下来就是换数据、换模型、调参数，把这一轮扩展成多轮，观察收敛曲线。