搭建Leiolai式算力共享系统：从设备注册到任务调度实战

最近在调研分布式 AI 算力调度时，注意到一个很有意思的新兴项目模式：Leiolai。它把传统“中心化 GPU 集群”的逻辑反过来，让普通用户的 PC、笔记本甚至移动设备，在闲置时贡献出自己的计算资源参与 AI 任务，并因此获得收益。这个模式听起来像“赛博出租”，但落地到工程上，需要考虑设备注册、任务分发、算力证明、结果回传等一系列架构问题。

本文不吹概念，直接以 Leiolai 的思路为蓝本，带大家从零搭建一套“设备算力供给 AI 任务”的完整闭环模拟系统。内容包含核心概念拆解、环境准备、Python 完整实战代码、常见问题排查以及工程化最佳实践。无论你是对 Web3 与 AI 结合感兴趣的开发者，还是想了解分布式任务调度的后端工程师，都能在这篇文章里找到可直接复用的方案。

1. 项目背景与核心概念

1.1 什么是 Leiolai：AI 算力共享新模式

Leiolai 本质上是一个算力交易平台。传统意义上，训练和运行大模型，需要购买昂贵的 GPU 服务器，或者租用云厂商的算力。这导致两个问题：一是成本高，中小企业或独立开发者难以承受；二是资源浪费，个人电脑里的 GPU 在绝大多数时间都处于闲置状态，利用率可能不到 20%。

Leiolai 想要解决的就是这个错配问题。它建立了一个网络，将设备所有者和 AI 任务需求方连接起来：

• 设备所有者（Provider）：安装对应的客户端，贡献自己的 CPU、GPU 和内存资源。
• AI 任务需求方（Requester）：通过平台提交推理任务（比如批量生成图片、处理自然语言数据），支付平台积分或代币。
• 平台（Coordinator）：充当撮合与验证的角色，负责健康检查、任务分发、结果校验和收益结算。

一句话概括：设备出力换收益，AI 任务低成本跑起来。它属于分布式计算，与传统的 BOINC（伯克利开放式网络计算平台）相比，Leiolai 更聚焦 AI 深度学习相关的张量计算任务，且对任务贡献的验证模型更严格，因为神经网络模型具有很强的参数依赖特征，传统随机数切块方式并不完全适用。

1.2 设备算力变现的底层逻辑与价值

为什么这种模式能成立？我们算一笔简单的账。

假设你有一台消费级显卡 RTX 4060，功耗约 115W，每天开机 8 小时，其中可能有一半时间是空闲的。如果这 4 小时的空闲被利用起来，计算模型推理任务，平台的成本远低于专门租用云 GPU 实例。Leiolai 的核心技术价值不在于“能跑”，而在于“怎么管”。

管理体现在三个维度：

1. 动态调度：设备网速、显卡显存大小各不相同，平台需要根据任务复杂度进行动态切分，把大任务化成小任务分发。
2. 状态感知：个人设备随时可能关机、断网、被用户抢走玩游戏，节点随时会掉线，平台必须支持断点重传和状态冗余。
3. 信任证明：如何确定设备真的跑了 10 万亿次浮点运算，而不是用 CPU 写了个死循环骗算力？Leiolai 引入了基于结果哈希校验和抽检机制，甚至可以采用零知识证明验证计算正确性。

掌握 Leiolai 这一类系统架构，可以帮助你深入理解现代分布式计算、边缘推理和算力经济模型。在很多边缘 AI 场景中，直接把模型扔到本地设备执行，反而比上传云端更快更安全，Leiolai 相当于把这种边缘能力打包成了可交易的资产。

1.3 别误解：去中心化算力不等于绝对安全可靠

在动手写代码之前，需要给大家泼一盆冷水。任何“共享计算资源”的公共网络都面临两个硬伤：

1. 异构环境带来的不确定性：个人 PC 的软件环境千奇百怪，驱动版本不一致、缺少 CUDA 库、杀毒软件拦截脚本等等，都会导致任务执行失败。平台必须假设所有设备都是“恶意且脆弱的”。
2. 通信延迟与带宽瓶颈：模型参数和训练数据动辄几百 MB，通过家用宽带上行传输是致命的。所以 Leiolai 更偏向推理任务（模型小、参数固定）而不是训练任务。

了解清楚这些限制，我们再进入实际操作环节，就可以少走很多弯路。我们下面的实战项目，重点会放在任务的注册、分配、执行和结果上传上，这正是整个系统的骨架。

2. 环境准备与版本说明

2.1 硬件与操作系统要求

由于我们要真正模拟出“设备提供算力”的效果，不仅需要有服务端（调度方），还需要有 Worker 节点（计算方）。开发调试阶段，你只需要一台电脑，通过localhost模拟；如果条件允许，建议准备两台电脑（一台 Linux 一台 Windows），或者用两台虚拟机，效果更逼真。

硬件要求：

• 最低配置：2 核 CPU、4GB 内存、20GB 磁盘空间。
• 推荐配置：具备 NVIDIA 显卡（可选，本文代码会兼容无 GPU 环境的 CPU 推理），4 核 CPU、8GB 内存。
• 操作系统：Ubuntu 20.04 / Windows 10 均可，本文示例以 Windows 和 Linux 双环境兼容为默认目标。

2.2 Python 环境与依赖库安装

我建议使用 Python 3.9 或 3.10 版本，这两个版本对于深度学习库的支持目前最稳定。我们需要安装以下核心依赖：

pip install requests flask psutil numpy

如果设备上有 NVIDIA 显卡，可选择性安装 PyTorch CPU 或 GPU 版本来承担真实推理任务：

# CPU 版 pip install torch torchvision --index-url https://download.pytorch.org/whl/cpu # GPU 版（CUDA 11.8 为例，请根据自己驱动调整） pip install torch torchvision --index-url https://download.pytorch.org/whl/cu118

版本提醒：截止我目前写这篇文章的时间，PyTorch 2.x 已经成为主流。但如果你用的是老旧显卡或旧驱动，安装 1.13 版本会更稳妥。请根据你的实际环境调整，本文重点演示调度系统逻辑，不是深度学习炼丹。

2.3 测试环境说明

我们将搭建一个包含 3 个角色的模拟系统：

• server.py：负责接收 Worker 心跳、下发任务、校验结果。
• worker.py：负责注册设备、拉取任务、执行计算（真正调用 Python 函数计算）。
• task_gen.py：负责生成模拟的 AI 任务存入服务端队列。

本节我们先创建基本的项目目录结构，代码在下一节逐行拆解。

mkdir leiolai_project cd leiolai_project mkdir logs tasks

3. 核心原理拆解：计算、验证与奖励机制

3.1 计算委托进程

Leiolai 模式中，客户端（Worker）启动后，先向服务器发送注册请求，上报自己的硬件信息。

这个过程在工程上叫“握手”（Handshake）。服务器收到硬件配置后，会给 Worker 一个唯一的 Worker ID。后续的任务请求都会附带这个 ID，方便统计算力贡献。

那么，任务如何委托呢？并不是所有 AI 任务都能上传到普通 PC 跑，Leiolai 通常采用的是“分片推理”策略。举例来说，一个批量图片分类任务，需要处理 1000 张图片，服务器不会一次性发给一个 Worker，而是切成 50 个任务包，每个包 20 张图，分发给 50 个设备并行处理。每个任务包就是一个可独立执行的最小单元。

在代码实现时，这个最小单元可以用一个 JSON 字符串表示，包含任务 ID、任务类型、数据 Base64 编码、期望返回结果格式等。

3.2 贡献证明与奖励发放逻辑

在模拟系统中，我们用“积分”代替真实的代币机制。关键在于如何证明“你已经做了工作”。

最朴素的做法是使用 Base64 编码的样本数据，让 Worker 计算后返回一个哈希值。服务器端持有同样的数据和代码，随机抽取 20% 的任务进行结果比对，如果比对一致，则给 Worker 发放积分；如果不一致，则扣除信誉分并标记该 Worker 存在潜在作弊行为。

除了哈希校验，生产环境中还会加上时间戳。例如，一个任务预计在 1 分钟内完成，如果某个 Worker 在 3 秒内就返回了结果，而且计算结果哈希一致，那大概率是拿旧结果伪造的，必须判定为异常节点。

在下面的实战里，我们将简化这一过程，用计算结果的 MD5 值和耗时参数共同作为工作量证明依据。

3.3 通信协议设计（任务拆包与结果回传）

为了保证通信稳定，我们使用 RESTful API 风格，数据格式采用 JSON。

关键接口如下表所示：

这里必须注意：很多初学者会把任务拉取做成 GET 请求。但 Worker 每次拉取任务时都需要携带加密签名或密钥，为避免签名出现在 URL 中（网关日志容易泄露），强烈建议统一使用 POST 请求传递拉取参数。

4. 完整实战案例：构建 Leiolai 式 Worker 节点

从这一节开始，我们会写真正能运行的代码。先把路径标明清楚，再逐文件解释。

4.1 创建项目核心辅助函数（utils.py）

在项目根目录创建utils.py，负责处理设备信息采集、结果指纹计算和 Base64 编解码。

# 文件路径：leiolai_project/utils.py import platform import socket import hashlib import json import time import psutil import base64 def get_device_info(): """采集设备硬件信息，用于 Worker 注册""" info = { "hostname": socket.gethostname(), "platform": platform.system(), "platform_release": platform.release(), "cpu_cores": psutil.cpu_count(logical=True), "cpu_freq": psutil.cpu_freq().max if psutil.cpu_freq() else 0, "memory_total_gb": round(psutil.virtual_memory().total / (1024 ** 3), 2), "ip_address": socket.gethostbyname(socket.gethostname()) } # 尝试获取 GPU 信息（简单读取 nvidia-smi 的输出版本号） try: import subprocess result = subprocess.run(["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], capture_output=True, text=True, timeout=5) if result.returncode == 0: info["gpu_name"] = result.stdout.strip().split("\n")[0] else: info["gpu_name"] = "CPU-Only" except Exception: info["gpu_name"] = "CPU-Only" return info def compute_task_signature(task_data: str, exec_start_time: float, exec_end_time: float) -> str: """根据任务数据、执行开始和结束时间生成工作量证明指纹""" payload = json.dumps({ "task_data": task_data, "start": exec_start_time, "end": exec_end_time }, sort_keys=True).encode("utf-8") return hashlib.sha256(payload).hexdigest() def encode_data(data: dict) -> str: """将字典数据编码为 Base64 字符串，便于在 JSON 中传输""" return base64.b64encode(json.dumps(data).encode("utf-8")).decode("utf-8") def decode_data(data_str: str) -> dict: """将 Base64 字符串还原为字典""" return json.loads(base64.b64decode(data_str.encode("utf-8"))) if __name__ == "__main__": # 本地测试设备信息采集 print(json.dumps(get_device_info(), indent=2, ensure_ascii=False))

这段代码的核心作用有两个：get_device_info用于注册时向服务器报告资源，compute_task_signature用于生成一个和任务内容、执行时间强相关的结果指纹。因为我们是模拟算力贡献，不能只返回计算结果，还必须包含时间参数，服务器才能判断你是否真的在本地执行了计算。

4.2 编写服务器端调度程序（server.py)

服务器端使用 Flask 实现轻量 API。它维护三个全局字典：workers（设备清单）、task_queue（任务队列）、task_results（结果存储）。

为了演示效果，我们会在服务器启动时，通过一个内置接口预置若干“AI 图片分类”任务。每个任务里包含一个input_data字段，模拟需要处理的图像字节码。

# 文件路径：leiolai_project/server.py import json import uuid import time import threading from flask import Flask, request, jsonify app = Flask(__name__) # 全局任务队列 task_queue = [] # 存储 Worker 注册信息，key 为 worker_id workers = {} # 存储任务执行结果，key 为 task_id task_results = {} # 模拟任务数量 TASK_NUM = 20 def generate_mock_tasks(): """生成模拟 AI 任务，每个任务包含一个随机的输入特征向量""" for i in range(TASK_NUM): task = { "task_id": str(uuid.uuid4()), "task_type": "image_classify", "status": "pending", # pending, running, completed, failed "assigned_worker": None, "create_time": time.time(), "input_data": f"fake_image_data_{i}", # 模拟待推理的数据 "timeout": 30 # 任务超时时间（秒） } task_queue.append(task) # 在应用启动时生成任务 generate_mock_tasks() @app.route("/api/worker/register", methods=["POST"]) def register_worker(): """设备注册接口：接收硬件信息，分配 worker_id""" data = request.get_json() if not data or "device_info" not in data: return jsonify({"code": 400, "msg": "缺少设备信息"}), 400 worker_id = "wk_" + uuid.uuid4().hex[:10] workers[worker_id] = { "device_info": data["device_info"], "status": "online", "score": 0, "total_tasks_done": 0, "register_time": time.time() } print(f"[REGISTER] 新设备注册: {worker_id}, 硬件: {data['device_info']['gpu_name']}") return jsonify({"code": 0, "worker_id": worker_id, "msg": "注册成功"}), 200 @app.route("/api/worker/pull_task", methods=["POST"]) def pull_task(): """任务拉取接口：从队列中找一个未分配且未超时的任务""" data = request.get_json() worker_id = data.get("worker_id") if worker_id not in workers: return jsonify({"code": 401, "msg": "Worker 未注册或不存在"}), 401 # 查找一个 pending 状态的任务，并将其标记为 running，分配给当前 Worker for task in task_queue: if task["status"] == "pending": task["status"] = "running" task["assigned_worker"] = worker_id task["start_time"] = time.time() print(f"[TASK_ASSIGN] 分配任务 {task['task_id']} 给 {worker_id}") return jsonify({"code": 0, "task": task}), 200 # 没有任务可分配 return jsonify({"code": 20010, "msg": "当前没有待处理任务"}), 200 @app.route("/api/worker/report_result", methods=["POST"]) def report_result(): """结果上报接口：接收计算结果，校验指纹，发放积分""" data = request.get_json() worker_id = data.get("worker_id") task_id = data.get("task_id") result_data = data.get("result_data") signature = data.get("signature") cost_time = data.get("cost_time") if worker_id not in workers: return jsonify({"code": 401, "msg": "Worker 未注册"}), 401 # 找到对应任务 task = next((t for t in task_queue if t["task_id"] == task_id), None) if not task: return jsonify({"code": 404, "msg": "任务不存在"}), 404 # 简单校验：任务必须由该 Worker 领取 if task["assigned_worker"] != worker_id: return jsonify({"code": 403, "msg": "无权上报该任务"}), 403 # 计算这单任务的积分（模拟：用时越短，算力贡献越多，积分越高） base_score = 10 if cost_time and cost_time < 5: score = base_score * 2 else: score = base_score # 记录结果 task_results[task_id] = { "worker_id": worker_id, "result": result_data, "signature": signature, "cost_time": cost_time, "score": score } # 更新 Worker 积分 workers[worker_id]["score"] += score workers[worker_id]["total_tasks_done"] += 1 # 将任务标记为完成 task["status"] = "completed" task["end_time"] = time.time() print(f"[RESULT_OK] 任务 {task_id} 由 {worker_id} 完成，耗时 {cost_time}s，积分 +{score}") return jsonify({"code": 0, "msg": "任务接收成功，积分已发放", "current_score": workers[worker_id]["score"]}), 200 @app.route("/api/worker/balance", methods=["GET"]) def get_balance(): """查询当前积分""" worker_id = request.args.get("worker_id") if worker_id not in workers: return jsonify({"code": 401, "msg": "Worker 未注册"}), 401 return jsonify({"code": 0, "worker_id": worker_id, "balance": workers[worker_id]["score"]}), 200 @app.route("/health", methods=["GET"]) def health(): """服务健康检查""" return jsonify({"code": 0, "status": "leiolai server is running"}), 200 if __name__ == "__main__": print("Leiolai 模拟调度服务已启动，监听端口 8000") print(f"共生成 {TASK_NUM} 个待处理 AI 任务") app.run(host="0.0.0.0", port=8000, debug=False)

这里我特意把积分规则写得简单粗暴：在 5 秒内完成的任务，奖励翻倍。这模拟了算力越强、任务处理越快、激励机制越明确的思路。在实际系统里，还要加入贡献值衰减、抽检惩罚等逻辑。

4.3 编写 Worker 端程序（worker.py）

Worker 的作用是尝到甜头的一方。它需要完成注册、定时拉取任务、执行计算、上报结果四个循环。

为了模拟真实的 AI 计算，我这里写了一个execute_task函数，它会对任务里的input_data字符串做一次 SHA256 哈希循环计算，并模拟矩阵乘法。这样即便在无 GPU 的普通电脑上也能跑出明显的耗时差异。

# 文件路径：leiolai_project/worker.py import time import json import requests import hashlib import numpy as np from utils import get_device_info, compute_task_signature, encode_data # 服务器地址（如果跨机器部署，请改成服务器 IP） SERVER_URL = "http://127.0.0.1:8000" class LeiolaiWorker: def __init__(self): self.worker_id = None self.session = requests.Session() def register(self): """向服务器注册设备，获取 Worker ID""" device_info = get_device_info() resp = self.session.post(f"{SERVER_URL}/api/worker/register", json={"device_info": device_info}) data = resp.json() if data.get("code") == 0: self.worker_id = data["worker_id"] print(f"[WORKER] 注册成功，Worker ID: {self.worker_id}") else: print(f"[WORKER] 注册失败: {data.get('msg')}") def execute_task(self, task): """真正执行计算任务, 返回结果字符串及耗时""" input_str = task["input_data"] task_type = task["task_type"] start_time = time.time() # ---------- 模拟 AI 推理计算 ---------- # 1. 先做一次 SHA256 迭代，模拟 IO 密集型预处理 hash_val = hashlib.sha256(input_str.encode("utf-8")).hexdigest() # 2. 构造一个 32x32 的随机矩阵，做 1000 次矩阵乘法，模拟张量运算 matrix_a = np.random.rand(32, 32).astype(np.float32) matrix_result = None for _ in range(1000): matrix_b = np.random.rand(32, 32).astype(np.float32) matrix_result = np.dot(matrix_a, matrix_b) # 3. 将最终结果转为字符串（绝对不允许传回原始输入作为结果） result_str = f"{task_type}:{hash_val}:{matrix_result.var():.6f}" # ------------------------------------- end_time = time.time() cost_time = round(end_time - start_time, 4) return result_str, start_time, end_time, cost_time def run(self): """主循环：注册、拉取任务、执行、上报""" if not self.worker_id: self.register() if not self.worker_id: return while True: try: # 1. 拉取任务 resp = self.session.post(f"{SERVER_URL}/api/worker/pull_task", json={"worker_id": self.worker_id}) data = resp.json() if data.get("code") != 0: # 没有任务或需要等待，休息 2 秒再问 print(f"[WORKER] 暂时没有任务，{data.get('msg','')}，休息 2 秒") time.sleep(2) continue task = data["task"] print(f"[WORKER] 领取任务 {task['task_id']}，类型：{task['task_type']}") result, start_time, end_time, cost_time = self.execute_task(task) signature = compute_task_signature(task["input_data"], start_time, end_time) # 2. 上报结果 report_payload = { "worker_id": self.worker_id, "task_id": task["task_id"], "result_data": encode_data({"result": result}), "signature": signature, "cost_time": cost_time } report_resp = self.session.post(f"{SERVER_URL}/api/worker/report_result", json=report_payload) report_data = report_resp.json() if report_data.get("code") == 0: print(f"[WORKER] 任务 {task['task_id']} 上报成功，获得积分，当前总分：{report_data['current_score']}") else: print(f"[WORKER] 任务上报失败：{report_data.get('msg')}") # 3. 每次任务执行完，稍作休息，避免打满 CPU time.sleep(1) except requests.exceptions.ConnectionError: print("[WORKER] 服务器连接失败，5 秒后重试...") time.sleep(5) except KeyboardInterrupt: print("[WORKER] 收到中断信号，优雅退出") break if __name__ == "__main__": worker = LeiolaiWorker() worker.run()

4.4 运行与验证

首先，在一个终端启动服务器：

cd leiolai_project python server.py

预期输出：

Leiolai 模拟调度服务已启动，监听端口 8000 共生成 20 个待处理 AI 任务 * Running on http://0.0.0.0:8000

紧接着，新开一个终端启动 Worker：

cd leiolai_project python worker.py

预期输出可以看到设备注册信息、任务拉取打印和积分累计打印。你可以同时多开两三个 Worker 终端，模拟多设备并行，观察服务器端任务被分配给不同 Worker 的过程。

4.5 结果说明与验证

跑完 20 个任务后，服务器端终端会显示类似这样的日志：

[TASK_ASSIGN] 分配任务 3f2c... 给 wk_abc1234567 [RESULT_OK] 任务 3f2c... 由 wk_abc1234567 完成，耗时 0.523s，积分 +20

Worker 终端则会显示自己的积分余额。这里包含了几个关键过程：

• 注册成功：意味着服务器端workers字典有了你的设备画像。
• 积分增加：说明服务器校验了任务状态，且认可了你的计算贡献。
• 任务状态流转：从pending到running再到completed，这是分布式任务系统的最小闭环。

你可以用浏览器或 curl 访问查询接口验证余额：

curl "http://127.0.0.1:8000/api/worker/balance?worker_id=wk_abc1234567"

返回 JSON 中balance字段就是当前 Worker 的累计积分。

5. 常见问题与排查思路

在实际接入这类算力共享系统时，你可能会遇到各种各样的问题。下面整理几个高频故障点，并给出排查解决思路。

在实际生产项目中，我强烈建议不要把任务状态只存在进程内存中。一旦server.py崩溃，所有任务都会丢失。业界方案是引入 Redis 或数据库持久化状态，任务抢单使用SETNX锁租约，这样才具备真正可运营的基础。

6. 最佳实践与工程化建议

6.1 安全边界：设备黑名单与任务沙箱

能让外部设备执行代码，平台承担着巨大的安全风险。如果设备是恶意的，它完全可以在执行任务的同时植入挖矿病毒或篡改计算结果。

在工程上，推荐采用以下三重防护：

1. 沙箱隔离：强制要求 Worker 在 Docker 容器内执行任务负载，容器只分配 1 个 CPU 核心和 512MB 内存，杀掉容器只影响单个任务，不影响宿主系统。
2. 黑名单机制：对累计报错超过 5 次、结果校验失败超过 3 次的设备，永久拉入黑名单，不再分配高价值任务。
3. 签名通信：我们的示例代码中并没有做身份验证。真实上线时，Worker 请求必须携带由注册时生成的 RSA 私钥签名的随机挑战码，防止中间人攻击和注册接口被刷。

6.2 幂等性设计

分布式系统里，网络抖动是常态。Worker 上报结果后，如果服务器端在返回响应前突然断线，Worker 可能会重试上报。服务器端必须确保同一个task_id被重复上报时，不会给 Worker 重复加积分。

修改report_result接口时需要加入逻辑：

if task_id in task_results: # 如果已经存在，不重复记分，返回当前分数 return jsonify({"code": 0, "msg": "任务已处理过，不重复记分", "current_score": workers[worker_id]["score"]}), 200

6.3 网络与功耗优化

对于个人设备贡献算力，用户最关心的是电费和网络占用。我的建议是：

• 带宽限制：在 Worker 端实现令牌桶限流，每秒最多上传 1MB，确保不影响用户正常浏览网页。
• 空闲触发：客户端检测到用户鼠标键盘 5 分钟无操作，才正式启动任务计算；一旦检测到输入设备活动，立刻暂停当前张量计算，让出 CPU/GPU 资源。
• 温控保护：通过psutil查询 CPU/GPU 温度，当温度超过 85 度时自动进入降频模式，避免烧毁硬件。这不仅是体验问题，更是运营平台的合规底线。

6.4 日志与可观测性

排错依赖日志，分布式场景尤其重要。建议使用loguru库替代 print，按 Worker ID 切割文件，日志级别打到 DEBUG（包含完整 JSON 请求体和响应体）。服务器端建议集成 Prometheus 监控，统计以下核心指标：

• leiolai_task_queue_size：任务队列长度。
• leiolai_task_completed_total：完成任务总数。
• leiolai_worker_online_count：在线设备数。
• leiolai_execution_seconds：任务执行耗时直方图。

有了这些指标，当算力需求突增时，你才能做出精准的扩容和调度策略。

7. 下一步学习路线

到这里，我们已经完整走通了 Leiolai 这类去中心化算力共享模式的三个核心流程：设备注册、任务分发、结果兑现。你不仅学会了 Flask 接口开发，更理解了为什么任务状态需要持久化、为什么设备信任模型比算法本身更难设计。

对于只想跑通 Demo 的朋友，现在就可以把 4.3 节中的execute_task函数替换成你想要的真实 AI 模型，比如接一个transformers库的中文情感分析管线，让设备真正为自然语言处理任务做出算力贡献。

而如果你想把系统真正做到生产可用，我建议下一步集中攻克两个模块：

1. 断点续跑与故障恢复：Worker 执行任务到一半断电，服务器如何通过任务超时机制重新分配？建议研究celery或RocketMQ的延迟队列方案。
2. 基于可信硬件的工作量证明：传统的 CPU 计算结果校验在高强度张量计算中非常耗时，可以查阅 Intel SGX（软件保护扩展）相关的文献，看如何通过可信执行环境在设备本地直接验证计算真实性，而不是依赖服务器重复计算。

本文的所有代码都围绕最小闭环展开，没有引入高深复杂的中间件，就是为了让你能百分之百跑起来。从理解到实践，再到思考优化，这正是进入分布式 AI 基建领域最清晰的路径。如果你在运行这段代码时碰到报错异常，欢迎根据上面的排查表逐一对照，也欢迎多和身边的开发者交流实际踩坑经验。