从零搭建AI算力共享平台：设备接入、任务调度与奖励结算

最近在逛 Hacker News 时，看到了一个很有意思的项目 Leiolai。它的核心思路用一句话概括：用户把电脑、手机、平板等设备的空闲算力贡献给 AI 平台，平台根据实际贡献给用户支付奖励。这个方向其实并不算新，但 Leiolai 把“设备算力 + AI 任务 + 用户激励”三者串在了一条链路上，在个人闲置设备与 AI 计算需求之间搭了一座桥。本文不打算只停留在项目介绍层面，而是会结合这类分布式算力平台的通用设计，带大家从零搭建一个简化版的 Leiolai 节点原型，包含设备注册、能力上报、任务分发、结果提交、奖励入账这几个核心环节。读完这篇文章，你会理解这类“AI 算力共享 + 激励”平台的基本架构，也能用代码跑通一个最小系统。

需要提前说明的是，Leiolai 目前更像是一个早期项目，具体接口和奖励规则可能随时调整。所以本文不会凭空去“复刻”它的官方实现，而是从工程角度拆解它背后的通用技术模型。即使你以后接入了别的同类平台，这套思路也可以复用。

1. Leiolai 是什么：让闲置设备参与 AI 计算

1.1 从算力闲置到算力共享

先看一个很常见的现象：很多开发者电脑性能并不差，尤其是配备了独立显卡的机器，平时写代码、浏览网页、看视频时，CPU 和 GPU 利用率可能连 20% 都不到。与此同时，AI 模型的推理和训练需要大量算力，个人或小团队如果要部署一个稍大一点的模型，往往需要购买云服务器或 GPU 实例，成本并不低。

Leiolai 想做的事情，就是把这两端连起来。用户设备安装一个客户端，通过网络接入平台，平台把 AI 计算任务拆分成小片，分发给不同设备执行。设备完成任务后，平台根据任务难度、计算量、完成质量等维度，给用户发放积分或现金奖励。这本质上是一种“算力共享经济”。

从技术角度看，这种模式并不神秘。类似的思想在很多分布式计算项目中出现过：早期有 SETI@home 利用闲置 CPU 分析射电信号，后来有 BOINC 平台聚合全球志愿者的算力，近几年也有不少项目尝试用区块链或积分体系激励用户共享 GPU 算力。Leiolai 的特点在于它把重心放在了 AI 任务上，并且强调“用户能拿到实际回报”。

1.2 Leiolai 的价值链路

我们可以把 Leiolai 这类平台的价值链路拆成四段：

第一段是设备接入。用户设备需要安装客户端，客户端负责采集设备信息、建立网络连接、接收任务、执行计算、返回结果。这一段的难点在于设备类型差异大，可能是 x86 的 Windows 电脑，也可能是 ARM 架构的 Linux 服务器，还可能是手机。客户端要有较好的兼容性。

第二段是任务分解。AI 任务不一定都能整包扔给一台设备。比如一个文本生成任务，可以按照请求拆分成多个推理请求；一个模型训练任务，可以按照数据分片拆分给多个设备做分布式训练。任务怎么拆，直接决定了计算效率。

第三段是调度分发。平台需要知道当前哪些设备在线、各自有多少算力、网络延迟如何。然后结合任务优先级和奖励预算，决定把任务派给哪台设备。好的调度器要尽量做到“让合适的设备做合适的事”。

第四段是验证结算。设备返回计算结果后，平台需要验证结果是否正确、是否被篡改。验证通过后，再把奖励计入用户账户。这一段最容易出问题：如果验证太松，恶意设备会随便返回垃圾数据骗取奖励；如果验证太严，又会增加平台成本。

1.3 适用场景与用户画像

谁适合关注 Leiolai 这类项目？大致可以分为三类。第一类是普通用户，手头有闲置电脑或游戏显卡，想利用空闲时间获得一点额外回报。第二类是开发者，想研究分布式计算、任务调度、激励系统，甚至想在自己项目里实现类似机制。第三类是有 AI 计算需求但又不想直接买昂贵 GPU 的小团队，希望以更低成本获取算力。

当然，这种模式也有明显的边界。个人设备的算力稳定性、网络带宽、在线时长都无法和专业数据中心相比，所以它更适合对实时性要求不高的 AI 任务，比如批量推理、数据增强、模型微调、联邦学习等。对于需要低延迟响应的在线业务，个人设备并不适合。

2. 核心概念与系统拆解

要理解 Leiolai 的工程实现，需要先认识几个核心概念。

2.1 设备节点

设备节点是参与计算的最小单元。可以是一台电脑、一块开发板、一部手机，也可以是一台虚拟机。每个节点在平台上有一个唯一标识，通常是一串设备 ID。节点在首次接入时需要注册，之后通过心跳机制保持在线状态。

节点需要上报自己的硬件信息，包括 CPU 型号和核心数、内存大小、GPU 型号和显存、磁盘空间、网络上行带宽等。这些信息会用于任务匹配。比如某个 AI 推理任务需要 4GB 显存，调度器就不会把它分发给没有 GPU 的设备。

这里要注意，设备上报的信息并不一定可信。出于安全考虑，平台不能完全相信客户端上报的“我有 8 核 CPU”，最好通过基准测试（benchmark）或短时任务来评估设备的真实算力。

2.2 AI 任务

在 Leiolai 场景下，AI 任务可以分成两大类。

一类是推理任务。例如给出一段文本让模型生成回答，或者给出一张图片让模型识别物体。推理任务通常单个请求计算量不大，但并发量可能很高，非常适合分发给大量终端设备。

另一类是训练任务。例如让设备用本地数据做模型微调，然后把更新后的模型参数上传到平台，再通过联邦聚合生成全局模型。这类任务对设备算力和网络稳定性要求更高。

任务在平台内部有明确的状态流转：待调度、已分发、执行中、已提交、验证中、已完成、失败。理解这个状态机，对后面写服务端代码很有帮助。

2.3 奖励机制

奖励是 Leiolai 这种平台最敏感的部分。如果奖励规则不透明，用户很快会流失；如果奖励发放有漏洞，平台又会被刷单攻击。

常见的奖励维度包括：设备实际计算时长、CPU/GPU 占用率、完成任务的数量、任务难度系数、结果质量。更精细的平台还会考虑设备的在线时长和稳定贡献率。Leiolai 的奖励机制具体细节只有官方能确认，但从工程角度，我们至少需要设计一个“任务积分”模型，后续可以对接支付或链上结算。

2.4 调度与验证

调度器和验证器是平台的“大脑”和“裁判”。调度器负责把任务分配给合适的设备，验证器负责确认设备提交的结果有效。验证方式有很多种：同一任务分配给多台设备做交叉验证；平台内置抽样复核；对于确定性任务，平台可以快速重算；对于非确定性任务，则需要设计更复杂的验收规则。

在原型项目中，为了降低复杂度，我通常会先做最简验证：记录任务下发时的计算参数，设备提交结果后，服务端通过简单的哈希校验对比结果是否一致。

3. 环境准备与架构选型

3.1 本地开发环境

在开始写代码之前，建议先准备好环境。本文示例代码以 Python 为主，因为 Python 在 AI 生态中非常成熟，写原型也比较快。操作系统方面，Windows、macOS、Linux 都可以。需要安装的基础工具包括：

• Python 3.10 或更高版本；
• pip 包管理工具；
• Docker（如果你希望通过容器方式跑起服务端）；
• 一个趁手的 API 测试工具，例如 curl 或 Postman。

本文代码不依赖特殊硬件，普通 CPU 电脑就能运行。如果你有 NVIDIA GPU，也可以把示例中的“模拟计算”替换成真实的 AI 推理任务，只需要在设备端装好 PyTorch 或 ONNX Runtime 即可。

3.2 技术栈建议

服务端我建议使用 FastAPI，理由有几点：性能不错、自动生成 API 文档、代码量少、适合快速搭建原型。客户端则使用 requests 库做 HTTP 通信，用 psutil 库采集设备 CPU、内存信息。为方便本地演示，我们暂时不引入数据库，所有数据保存在内存中。真实项目中，设备表、任务表、奖励流水表至少需要一套持久化存储，可以选择 PostgreSQL 或 MySQL。

版本不需要完全固定，以下是一个可用的依赖清单：

fastapi uvicorn pydantic requests psutil

如果你希望服务端跨域，可以额外安装fastapi-cors，但原型阶段不强制。

3.3 项目目录结构

为了让代码清晰，建议按下面结构组织项目：

leiolai-demo/ ├── server/ │ ├── main.py │ └── requirements.txt ├── client/ │ ├── agent.py │ └── requirements-client.txt ├── docker-compose.yml └── README.md

server/main.py是平台服务端，提供设备注册、心跳上报、任务分发、结果提交、余额查询等 API。client/agent.py是设备客户端，模拟节点端行为。docker-compose.yml用于一键启动服务端和客户端容器。接下来，我们逐步实现这些文件。

4. 从零实现一个 Leiolai 节点原型

这一节我们实现一个简化版的算力共享平台。代码不追求生产级健壮性，而是为了把核心流程跑通，让读者理解每个环节之间的关系。

4.1 定义设备节点与任务模型

先编写服务端server/main.py。我们需要定义几个 Pydantic 模型，用于 API 请求和响应。

# 文件路径：server/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from typing import Optional, Dict, List import uuid import time import hashlib import random app = FastAPI(title="Leiolai Demo Server") # ========== 数据模型 ========== class DeviceRegisterRequest(BaseModel): """设备注册请求""" device_name: str cpu_cores: int = Field(..., description="CPU 核心数") memory_gb: float = Field(..., description="内存大小，单位 GB") gpu_name: Optional[str] = None gpu_vram_gb: Optional[float] = None class DeviceInfo(BaseModel): """设备信息""" device_id: str device_name: str cpu_cores: int memory_gb: float gpu_name: Optional[str] = None gpu_vram_gb: Optional[float] = None status: str = "offline" last_heartbeat: float = 0.0 balance: int = 0 class HeartbeatRequest(BaseModel): """心跳上报请求""" device_id: str cpu_load: float = Field(..., description="当前 CPU 使用率 0-100") memory_used_gb: float = Field(..., description="已使用内存 GB") gpu_load: Optional[float] = None class AI_Task(BaseModel): """AI 任务""" task_id: str task_type: str # inference / training payload: str status: str = "pending" assigned_device: Optional[str] = None result: Optional[str] = None reward: int = 10 created_at: float = Field(default_factory=time.time) class SubmitResultRequest(BaseModel): """提交任务结果请求""" device_id: str task_id: str result: str

这里每个模型都承担一个职责。设备注册请求里，我们把 CPU、内存、GPU 信息作为设备能力的“声明”；心跳请求则携带更实时的负载数据；任务模型包含奖励字段，方便后续结算。

在内存中保存设备和任务：

# 文件路径：server/main.py（续） devices: Dict[str, DeviceInfo] = {} tasks: Dict[str, AI_Task] = {} def get_device_or_404(device_id: str) -> DeviceInfo: if device_id not in devices: raise HTTPException(status_code=404, detail="设备不存在") return devices[device_id]

4.2 服务端实现设备注册与心跳

注册接口负责创建设备 ID，并把设备信息保存下来。设备 ID 使用 UUID 生成，这样不容易冲突。

# 文件路径：server/main.py（续） @app.post("/api/register", response_model=DeviceInfo) def register_device(req: DeviceRegisterRequest): device_id = uuid.uuid4().hex device = DeviceInfo( device_id=device_id, device_name=req.device_name, cpu_cores=req.cpu_cores, memory_gb=req.memory_gb, gpu_name=req.gpu_name, gpu_vram_gb=req.gpu_vram_gb, status="online", last_heartbeat=time.time(), ) devices[device_id] = device return device

心跳接口做的事情有两件：一是更新设备在线状态和最新负载，二是看看有没有合适的任务可以派发。为了让逻辑直观，心跳响应里除了返回设备最新信息，还会返回一个assigned_task字段。如果暂时没有任务，该字段为None。

# 文件路径：server/main.py（续） @app.post("/api/heartbeat") def heartbeat(req: HeartbeatRequest): device = get_device_or_404(req.device_id) device.status = "online" device.last_heartbeat = time.time() assigned_task = None for task in tasks.values(): if task.status == "pending" and task.assigned_device is None: # 简单匹配：训练任务尽量分给有 GPU 的设备 if task.task_type == "training" and not device.gpu_name: continue task.status = "running" task.assigned_device = device.device_id assigned_task = task break return { "device_id": device.device_id, "status": device.status, "assigned_task": assigned_task }

上面这段代码体现了最简单的任务匹配策略：遍历待分配任务，找到第一个可以执行的任务。在真实系统中，这一步会换成复杂的调度算法，比如基于设备得分、网络延迟、任务截止时间来排序。

4.3 任务注入与结果提交

平台需要有一个“注入任务”的接口，方便模拟平台侧产生 AI 任务。正常运行时，这个接口应该由任务生产者调用，而不是直接暴露给普通用户。

# 文件路径：server/main.py（续） @app.post("/api/tasks") def create_task(task_type: str, payload: str, reward: int = 10): task = AI_Task( task_id=uuid.uuid4().hex, task_type=task_type, payload=payload, reward=reward, ) tasks[task.task_id] = task return task

客户端执行完计算后，会调用结果提交接口。服务端在这里做两件事：把任务状态改为“已完成”，给设备增加奖励。为了让原型更接近真实，我们还简单校验了一下结果是否非空，以及设备是否真的领取过这个任务。

# 文件路径：server/main.py（续） @app.post("/api/submit") def submit_result(req: SubmitResultRequest): device = get_device_or_404(req.device_id) if req.task_id not in tasks: raise HTTPException(status_code=404, detail="任务不存在") task = tasks[req.task_id] if task.assigned_device != device.device_id: raise HTTPException(status_code=400, detail="任务未分配给该设备") if not req.result: raise HTTPException(status_code=400, detail="结果不能为空") task.status = "completed" task.result = req.result device.balance += task.reward return { "task_id": task.task_id, "status": task.status, "reward": task.reward, "balance": device.balance }

最后是查询余额接口：

# 文件路径：server/main.py（续） @app.get("/api/balance/{device_id}") def get_balance(device_id: str): device = get_device_or_404(device_id) return { "device_id": device.device_id, "balance": device.balance }

到这一步，服务端的最小闭环已经完成：注册 -> 心跳领取任务 -> 执行计算 -> 提交结果 -> 获得奖励。

4.4 客户端实现：模拟设备节点

客户端client/agent.py的角色是一台接入平台的设备。它要做的事情包括：

1. 启动时向服务端注册，拿到设备 ID；
2. 周期性发送心跳，并在心跳响应中检查是否有任务；
3. 如果有任务，模拟执行计算；
4. 把计算结果提交给服务端。

为了模拟真实设备，客户端会通过psutil采集 CPU 使用率、内存占用等信息。如果psutil未安装，代码中会给出提示。

# 文件路径：client/agent.py import time import random import requests import platform import hashlib try: import psutil HAS_PSUTIL = True except ImportError: HAS_PSUTIL = False SERVER_URL = "http://localhost:8000" def collect_device_spec(): """收集设备硬件信息""" if HAS_PSUTIL: cpu_cores = psutil.cpu_count(logical=True) or 2 memory_gb = round(psutil.virtual_memory().total / (1024 ** 3), 2) gpu_name = None gpu_vram_gb = None else: cpu_cores = 4 memory_gb = 8.0 gpu_name = None gpu_vram_gb = None return { "device_name": platform.node() or "unknown-device", "cpu_cores": cpu_cores, "memory_gb": memory_gb, "gpu_name": gpu_name, "gpu_vram_gb": gpu_vram_gb, } def register_device(): """注册设备""" spec = collect_device_spec() resp = requests.post(f"{SERVER_URL}/api/register", json=spec) resp.raise_for_status() data = resp.json() print(f"[注册成功] device_id={data['device_id']}") return data["device_id"] def send_heartbeat(device_id): """发送心跳并接收任务""" if HAS_PSUTIL: cpu_load = psutil.cpu_percent(interval=1) memory_used_gb = round(psutil.virtual_memory().used / (1024 ** 3), 2) else: cpu_load = random.uniform(10, 60) memory_used_gb = round(random.uniform(2, 6), 2) payload = { "device_id": device_id, "cpu_load": cpu_load, "memory_used_gb": memory_used_gb, "gpu_load": None } resp = requests.post(f"{SERVER_URL}/api/heartbeat", json=payload) resp.raise_for_status() return resp.json() def execute_task(task): """模拟执行 AI 计算任务""" print(f"[执行任务] task_id={task['task_id']} type={task['task_type']} payload={task['payload']}") # 模拟 CPU 密集计算 time.sleep(2) fake_result = hashlib.sha256(task["payload"].encode()).hexdigest() return fake_result def submit_result(device_id, task_id, result): """提交任务结果""" payload = { "device_id": device_id, "task_id": task_id, "result": result } resp = requests.post(f"{SERVER_URL}/api/submit", json=payload) resp.raise_for_status() return resp.json() def main(): device_id = register_device() while True: try: data = send_heartbeat(device_id) task = data.get("assigned_task") if task: result = execute_task(task) submit_resp = submit_result(device_id, task["task_id"], result) print(f"[提交成功] balance={submit_resp['balance']}") else: print("[等待任务] 当前无任务，休眠 5 秒") time.sleep(5) except requests.RequestException as exc: print(f"[网络异常] {exc}") time.sleep(10) if __name__ == "__main__": main()

这段客户端代码是一个典型的“心跳轮询模型”。真实平台为了降低网络开销，可能会改用 WebSocket 或 gRPC 长连接，但轮询模型最容易理解和调试。

4.5 运行与验证

先启动服务端。在server目录下安装依赖并运行：

cd server pip install -r requirements.txt uvicorn main:app --host 0.0.0.0 --port 8000

看到Uvicorn running on http://0.0.0.0:8000后，再启动客户端：

cd client pip install -r requirements-client.txt python agent.py

为了验证完整流程，我们需要先创建几个任务。打开另一个终端，用 curl 创建两个推理任务和一个训练任务：

curl -X POST "http://localhost:8000/api/tasks?task_type=inference&payload=hello&reward=10" curl -X POST "http://localhost:8000/api/tasks?task_type=inference&payload=world&reward=10" curl -X POST "http://localhost:8000/api/tasks?task_type=training&payload=federated-round-1&reward=50"

如果客户端正在运行，它会自动领取任务并提交结果。接着查询设备余额，可以看到奖励逐步累计：

curl http://localhost:8000/api/balance/<device_id>

浏览器访问http://localhost:8000/docs，还能看到 FastAPI 自动生成的 Swagger 文档，方便测试所有接口。

上面的代码虽然简单，但已经覆盖了 Leiolai 类平台最核心的流程。稍加扩展，就可以把它改造成一个真实可用的算力共享平台。

5. 常见问题与排查思路

在开发和运行这类系统时，有几个典型问题需要特别注意。

5.1 设备一直领不到任务

如果客户端日志总是显示“当前无任务”，常见原因有三个。一是任务还没创建，需要在平台侧注入任务。二是已有任务都被分配给了其他设备，导致看不到 pending 任务。三是任务类型与设备能力不匹配，比如训练任务只分发给有 GPU 的设备，而你的客户端没有 GPU，就会一直拿不到训练任务。

遇到这类情况，可以先检查服务端内存里的任务状态。也可以给/api/tasks接口加一个状态过滤参数，或者在响应中返回所有任务列表，方便定位。

5.2 设备离线后任务超时

原型代码里没有处理任务超时。真实情况下，设备可能领到任务后立刻断网，或者执行过程中崩溃。如果平台不处理超时，任务会一直卡在running状态，永远无法被重新分配。

解决思路是给任务增加一个assigned_at时间戳，服务端启动一个定时巡检任务，如果发现某个任务长时间处于running状态，就把它重置为pending，并清空assigned_device。这样其他在线设备就可以继续领取。

5.3 客户端提交结果被拒绝

结果提交失败最常见的报错是“任务未分配给该设备”。这是因为任务状态已经被服务端重置，或者设备 ID 不匹配。排查时先确认客户端使用的设备 ID 是否一致，再确认任务是否已经被其他设备完成。另一种情况是结果为空，客户端在计算失败时提交了空字符串，服务端会返回 400。

5.4 奖励计算争议

奖励是用户最关心的部分。如果用户发现贡献了很多算力但奖励很少，需要平台给出可解释的积分明细。真实系统里应该记录每笔奖励的来源任务、执行时长、算力贡献等原始信息，并提供查询接口。下面是一个常见的排查表：

6. 生产环境落地的最佳实践

原型能跑通只是第一步。如果你想基于 Leiolai 的思路做一个真实可用的平台，还需要在多个维度做加固。

6.1 安全与信任边界

最容易被忽略的是安全问题。平台接收来自个人设备的计算结果，这些设备完全不可控。恶意设备可能提交伪造结果、抓包重放请求、甚至尝试通过接口注入恶意数据。因此在生产环境中需要注意几点。

第一，所有 API 请求必须使用 HTTPS，避免传输过程被劫持。第二，设备身份不能只靠设备 ID，需要引入 API Token 或签名机制。设备注册时可以发放一个 secret，后续请求携带签名，服务端验签。第三，对任务结果要设计验证逻辑，同一任务可以交给多个设备计算，对比结果一致后再发放奖励。

6.2 算力评估与任务分片

设备上报的信息不能直接作为调度依据。更好的方式是平台主动下发基准测试任务，测量设备的 CPU 浮点运算能力、内存带宽、GPU 算力等指标。然后把这些指标作为设备评分保存在数据库中。

任务分片也要考虑设备差异。有些任务适合用 GPU 批量计算，有些任务 CPU 就能胜任。在调度系统里，可以给每类任务打上“资源需求标签”，例如gpu_required=4GB、cpu_cores_min=8。调度器根据标签和设备评分做匹配，避免 GPU 任务派发给纯 CPU 设备。

6.3 奖励结算的防作弊

奖励结算不能只看“提交了结果”就发钱，因为刷子设备会提交大量假结果来骗取奖励。推荐做法是：

• 给每个任务设置唯一执行 ID，服务端生成随机参数，设备在执行时必须基于该参数计算；
• 对确定性任务做抽查重算，例如按比例 5% 重新执行一遍；
• 设置单设备每日奖励上限，超过后需要人工审核；
• 记录设备行为特征，例如任务完成时间异常短、成功率异常高，都可能是作弊信号。

6.4 可观测性与日志

当平台接入几千台设备后，必然会出现各种异常。没有日志和监控，排查问题会非常痛苦。建议至少做到：

• 每个请求都有唯一 request_id，方便串联设备、任务、结果日志；
• 设备心跳、任务下发、结果提交都记录结构化日志；
• 引入 Prometheus + Grafana 监控设备在线数、任务吞吐量、平均完成时长、奖励发放速率等核心指标；
• 建立任务状态机的审计表，任何状态变更都留下记录。

6.5 持久化与数据一致性

内存字典在原型阶段很方便，但生产环境绝不能这样设计。设备表、任务表、奖励流水表都需要持久化存储。任务状态变更最好通过数据库事务保证一致性，避免出现“任务已完成但奖励未到账”的数据不一致。

另外，设备可能重复提交同一个任务结果。在数据库层，可以对task_id建立唯一约束，或者在服务端加分布式锁，确保同一任务只结算一次奖励。

7. 总结与下一步学习路线

这篇文章从 Leiolai 的项目理念出发，梳理了“AI 算力共享 + 用户激励”平台的通用技术架构，然后用 FastAPI 和 Python 实现了一个最小可运行的节点原型。你现在应该能够理解设备注册、能力上报、心跳轮询、任务分发、结果提交、奖励结算这几个关键环节，也知道了真实生产环境里需要解决的安全、调度、防作弊和可观测性问题。

如果你想继续深入，建议从以下几个方向入手。第一，把内存存储替换成 PostgreSQL，并引入 Redis 做设备在线状态管理。第二，学习任务队列，例如 Celery 或 Arq，让任务分发不再依赖心跳轮询。第三，研究联邦学习框架，了解如何把模型训练任务安全地下发到个人设备，同时保护数据隐私。第四，探索用 WebSocket 或 gRPC 替代 HTTP 轮询，降低通信延迟。

对于普通用户，如果你想参与 Leiolai 这类项目，重点关注的应该是平台对设备隐私的保护、奖励规则的透明度，以及任务是否会被用于恶意用途。对于开发者，我建议先把本文的原型跑起来，再逐步添加持久化、验证、监控等模块。亲手实现一遍，你对分布式算力系统的理解会比只看文档深刻得多。

如果本文对你有帮助，可以收藏备用。后续如果有新的项目动态或工程实践，我会继续整理成同样风格的文章分享出来。