尧图网页设计 
最近在逛 Hacker News 时看到了一个很有意思的项目 Leiolai。它的核心思路用一句话概括用户把电脑、手机、平板等设备的空闲算力贡献给 AI 平台平台根据实际贡献给用户支付奖励。这个方向其实并不算新但 Leiolai 把“设备算力 AI 任务 用户激励”三者串在了一条链路上在个人闲置设备与 AI 计算需求之间搭了一座桥。本文不打算只停留在项目介绍层面而是会结合这类分布式算力平台的通用设计带大家从零搭建一个简化版的 Leiolai 节点原型包含设备注册、能力上报、任务分发、结果提交、奖励入账这几个核心环节。读完这篇文章你会理解这类“AI 算力共享 激励”平台的基本架构也能用代码跑通一个最小系统。需要提前说明的是Leiolai 目前更像是一个早期项目具体接口和奖励规则可能随时调整。所以本文不会凭空去“复刻”它的官方实现而是从工程角度拆解它背后的通用技术模型。即使你以后接入了别的同类平台这套思路也可以复用。1. Leiolai 是什么让闲置设备参与 AI 计算1.1 从算力闲置到算力共享先看一个很常见的现象很多开发者电脑性能并不差尤其是配备了独立显卡的机器平时写代码、浏览网页、看视频时CPU 和 GPU 利用率可能连 20% 都不到。与此同时AI 模型的推理和训练需要大量算力个人或小团队如果要部署一个稍大一点的模型往往需要购买云服务器或 GPU 实例成本并不低。Leiolai 想做的事情就是把这两端连起来。用户设备安装一个客户端通过网络接入平台平台把 AI 计算任务拆分成小片分发给不同设备执行。设备完成任务后平台根据任务难度、计算量、完成质量等维度给用户发放积分或现金奖励。这本质上是一种“算力共享经济”。从技术角度看这种模式并不神秘。类似的思想在很多分布式计算项目中出现过早期有 SETIhome 利用闲置 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.mdserver/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(titleLeiolai Demo Server) # 数据模型 class DeviceRegisterRequest(BaseModel): 设备注册请求 device_name: str cpu_cores: int Field(..., descriptionCPU 核心数) 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_factorytime.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_code404, detail设备不存在) return devices[device_id]4.2 服务端实现设备注册与心跳注册接口负责创建设备 ID并把设备信息保存下来。设备 ID 使用 UUID 生成这样不容易冲突。# 文件路径server/main.py续 app.post(/api/register, response_modelDeviceInfo) def register_device(req: DeviceRegisterRequest): device_id uuid.uuid4().hex device DeviceInfo( device_iddevice_id, device_namereq.device_name, cpu_coresreq.cpu_cores, memory_gbreq.memory_gb, gpu_namereq.gpu_name, gpu_vram_gbreq.gpu_vram_gb, statusonline, last_heartbeattime.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_iduuid.uuid4().hex, task_typetask_type, payloadpayload, rewardreward, ) 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_code404, detail任务不存在) task tasks[req.task_id] if task.assigned_device ! device.device_id: raise HTTPException(status_code400, detail任务未分配给该设备) if not req.result: raise HTTPException(status_code400, 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的角色是一台接入平台的设备。它要做的事情包括启动时向服务端注册拿到设备 ID周期性发送心跳并在心跳响应中检查是否有任务如果有任务模拟执行计算把计算结果提交给服务端。为了模拟真实设备客户端会通过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(logicalTrue) 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, jsonspec) 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(interval1) 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, jsonpayload) 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, jsonpayload) 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_typeinferencepayloadhelloreward10 curl -X POST http://localhost:8000/api/tasks?task_typeinferencepayloadworldreward10 curl -X POST http://localhost:8000/api/tasks?task_typetrainingpayloadfederated-round-1reward50如果客户端正在运行它会自动领取任务并提交结果。接着查询设备余额可以看到奖励逐步累计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 奖励计算争议奖励是用户最关心的部分。如果用户发现贡献了很多算力但奖励很少需要平台给出可解释的积分明细。真实系统里应该记录每笔奖励的来源任务、执行时长、算力贡献等原始信息并提供查询接口。下面是一个常见的排查表问题现象常见原因解决思路设备显示在线但领不到任务任务类型与设备能力不匹配检查任务类型和设备 GPU 信息任务一直 running设备离线或执行超时增加任务超时重置机制提交结果失败任务已被重置或设备 ID 不正确核对设备 ID 和任务状态奖励未到账结果验证未通过检查结果提交时的校验逻辑服务端重启后数据丢失使用了内存存储引入 MySQL/PostgreSQL 或 Redis 持久化6. 生产环境落地的最佳实践原型能跑通只是第一步。如果你想基于 Leiolai 的思路做一个真实可用的平台还需要在多个维度做加固。6.1 安全与信任边界最容易被忽略的是安全问题。平台接收来自个人设备的计算结果这些设备完全不可控。恶意设备可能提交伪造结果、抓包重放请求、甚至尝试通过接口注入恶意数据。因此在生产环境中需要注意几点。第一所有 API 请求必须使用 HTTPS避免传输过程被劫持。第二设备身份不能只靠设备 ID需要引入 API Token 或签名机制。设备注册时可以发放一个 secret后续请求携带签名服务端验签。第三对任务结果要设计验证逻辑同一任务可以交给多个设备计算对比结果一致后再发放奖励。6.2 算力评估与任务分片设备上报的信息不能直接作为调度依据。更好的方式是平台主动下发基准测试任务测量设备的 CPU 浮点运算能力、内存带宽、GPU 算力等指标。然后把这些指标作为设备评分保存在数据库中。任务分片也要考虑设备差异。有些任务适合用 GPU 批量计算有些任务 CPU 就能胜任。在调度系统里可以给每类任务打上“资源需求标签”例如gpu_required4GB、cpu_cores_min8。调度器根据标签和设备评分做匹配避免 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 这类项目重点关注的应该是平台对设备隐私的保护、奖励规则的透明度以及任务是否会被用于恶意用途。对于开发者我建议先把本文的原型跑起来再逐步添加持久化、验证、监控等模块。亲手实现一遍你对分布式算力系统的理解会比只看文档深刻得多。如果本文对你有帮助可以收藏备用。后续如果有新的项目动态或工程实践我会继续整理成同样风格的文章分享出来。 从零搭建AI算力共享平台：设备接入、任务调度与奖励结算
最近在逛 Hacker News 时看到了一个很有意思的项目 Leiolai。它的核心思路用一句话概括用户把电脑、手机、平板等设备的空闲算力贡献给 AI 平台平台根据实际贡献给用户支付奖励。这个方向其实并不算新但 Leiolai 把“设备算力 AI 任务 用户激励”三者串在了一条链路上在个人闲置设备与 AI 计算需求之间搭了一座桥。本文不打算只停留在项目介绍层面而是会结合这类分布式算力平台的通用设计带大家从零搭建一个简化版的 Leiolai 节点原型包含设备注册、能力上报、任务分发、结果提交、奖励入账这几个核心环节。读完这篇文章你会理解这类“AI 算力共享 激励”平台的基本架构也能用代码跑通一个最小系统。需要提前说明的是Leiolai 目前更像是一个早期项目具体接口和奖励规则可能随时调整。所以本文不会凭空去“复刻”它的官方实现而是从工程角度拆解它背后的通用技术模型。即使你以后接入了别的同类平台这套思路也可以复用。1. Leiolai 是什么让闲置设备参与 AI 计算1.1 从算力闲置到算力共享先看一个很常见的现象很多开发者电脑性能并不差尤其是配备了独立显卡的机器平时写代码、浏览网页、看视频时CPU 和 GPU 利用率可能连 20% 都不到。与此同时AI 模型的推理和训练需要大量算力个人或小团队如果要部署一个稍大一点的模型往往需要购买云服务器或 GPU 实例成本并不低。Leiolai 想做的事情就是把这两端连起来。用户设备安装一个客户端通过网络接入平台平台把 AI 计算任务拆分成小片分发给不同设备执行。设备完成任务后平台根据任务难度、计算量、完成质量等维度给用户发放积分或现金奖励。这本质上是一种“算力共享经济”。从技术角度看这种模式并不神秘。类似的思想在很多分布式计算项目中出现过早期有 SETIhome 利用闲置 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.mdserver/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(titleLeiolai Demo Server) # 数据模型 class DeviceRegisterRequest(BaseModel): 设备注册请求 device_name: str cpu_cores: int Field(..., descriptionCPU 核心数) 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_factorytime.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_code404, detail设备不存在) return devices[device_id]4.2 服务端实现设备注册与心跳注册接口负责创建设备 ID并把设备信息保存下来。设备 ID 使用 UUID 生成这样不容易冲突。# 文件路径server/main.py续 app.post(/api/register, response_modelDeviceInfo) def register_device(req: DeviceRegisterRequest): device_id uuid.uuid4().hex device DeviceInfo( device_iddevice_id, device_namereq.device_name, cpu_coresreq.cpu_cores, memory_gbreq.memory_gb, gpu_namereq.gpu_name, gpu_vram_gbreq.gpu_vram_gb, statusonline, last_heartbeattime.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_iduuid.uuid4().hex, task_typetask_type, payloadpayload, rewardreward, ) 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_code404, detail任务不存在) task tasks[req.task_id] if task.assigned_device ! device.device_id: raise HTTPException(status_code400, detail任务未分配给该设备) if not req.result: raise HTTPException(status_code400, 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的角色是一台接入平台的设备。它要做的事情包括启动时向服务端注册拿到设备 ID周期性发送心跳并在心跳响应中检查是否有任务如果有任务模拟执行计算把计算结果提交给服务端。为了模拟真实设备客户端会通过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(logicalTrue) 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, jsonspec) 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(interval1) 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, jsonpayload) 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, jsonpayload) 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_typeinferencepayloadhelloreward10 curl -X POST http://localhost:8000/api/tasks?task_typeinferencepayloadworldreward10 curl -X POST http://localhost:8000/api/tasks?task_typetrainingpayloadfederated-round-1reward50如果客户端正在运行它会自动领取任务并提交结果。接着查询设备余额可以看到奖励逐步累计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 奖励计算争议奖励是用户最关心的部分。如果用户发现贡献了很多算力但奖励很少需要平台给出可解释的积分明细。真实系统里应该记录每笔奖励的来源任务、执行时长、算力贡献等原始信息并提供查询接口。下面是一个常见的排查表问题现象常见原因解决思路设备显示在线但领不到任务任务类型与设备能力不匹配检查任务类型和设备 GPU 信息任务一直 running设备离线或执行超时增加任务超时重置机制提交结果失败任务已被重置或设备 ID 不正确核对设备 ID 和任务状态奖励未到账结果验证未通过检查结果提交时的校验逻辑服务端重启后数据丢失使用了内存存储引入 MySQL/PostgreSQL 或 Redis 持久化6. 生产环境落地的最佳实践原型能跑通只是第一步。如果你想基于 Leiolai 的思路做一个真实可用的平台还需要在多个维度做加固。6.1 安全与信任边界最容易被忽略的是安全问题。平台接收来自个人设备的计算结果这些设备完全不可控。恶意设备可能提交伪造结果、抓包重放请求、甚至尝试通过接口注入恶意数据。因此在生产环境中需要注意几点。第一所有 API 请求必须使用 HTTPS避免传输过程被劫持。第二设备身份不能只靠设备 ID需要引入 API Token 或签名机制。设备注册时可以发放一个 secret后续请求携带签名服务端验签。第三对任务结果要设计验证逻辑同一任务可以交给多个设备计算对比结果一致后再发放奖励。6.2 算力评估与任务分片设备上报的信息不能直接作为调度依据。更好的方式是平台主动下发基准测试任务测量设备的 CPU 浮点运算能力、内存带宽、GPU 算力等指标。然后把这些指标作为设备评分保存在数据库中。任务分片也要考虑设备差异。有些任务适合用 GPU 批量计算有些任务 CPU 就能胜任。在调度系统里可以给每类任务打上“资源需求标签”例如gpu_required4GB、cpu_cores_min8。调度器根据标签和设备评分做匹配避免 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 这类项目重点关注的应该是平台对设备隐私的保护、奖励规则的透明度以及任务是否会被用于恶意用途。对于开发者我建议先把本文的原型跑起来再逐步添加持久化、验证、监控等模块。亲手实现一遍你对分布式算力系统的理解会比只看文档深刻得多。如果本文对你有帮助可以收藏备用。后续如果有新的项目动态或工程实践我会继续整理成同样风格的文章分享出来。