onchain_ai_service.py-平芜编程栈

一、链上AI服务的工程化：从概念验证到可维护系统

将AI模型的推理结果写入区块链，在技术层面上已经不是新鲜事——Ritual Infernet、ORA的opML、Bittensor的链上推理都可以实现"AI推理→链上存储/验证"的基本流程。但7月的工程实践让我们看到了一个更深层的问题：链上AI服务的建设成本不主要在"实现推理"上，而在"保证推理的可验证性"和"维护服务的持续可用性"上。

构建一条端到端的链上AI服务管线，涉及的工程层次包括：模型推理部署（计算资源选型、推理API封装）、验证机制选择（TEE/ZK/OpML/经济激励）、链上合约设计（请求-响应模式、费用模型、治理机制）、以及运维保障（节点健康监控、模型版本升级、灾难恢复）。这条管的每个环节都有工程权衡，本文整合7月的实践经验为一份全链路方案提炼。

二、链上AI服务的四层工程架构

每一层都有独立的技术选型和容错策略，但层与层之间的"接口协议"决定了整个系统的可靠性和可演进性。7月总结的最重要经验是：在L2（验证机制层）和L3（链上合约层）之间定义清晰的"证明格式协议"——无论底层使用TEE还是ZK，链上合约只需要验证一个标准化的证明结构（包含模型ID哈希、输入哈希、输出哈希和签名），这使得验证机制可以在不修改合约的前提下升级。

三、核心链路实现：从推理到链上存储

推理部署与验证封装

# onchain_ai_service.py # 链上AI服务的核心引擎 —— 模型推理 + 证明生成 + 链上提交 # # 设计决策： # 1. 模型引擎抽象为 InferenceEngine 接口 —— # 支持多种后端: vLLM(TGI兼容)、Ollama(本地)、OpenAI API(云端) # 切换引擎不影响上层逻辑(请求提交、证明生成) # 2. 证明策略工厂模式 —— ProveStrategy 根据模型类型和成本预算 # 自动选择最优证明方案: # - 7B以下模型 → ZK证明(EZKL) # - 7B-70B模型 → TEE证明(SGX/TDX) # - 70B以上模型 → OpML(乐观验证 + 欺诈证明期) # 阈值可以通过治理合约动态调整 # 3. 链上提交采用"乐观提交 + 可挑战"模式 —— # 推理结果先上链(用户立即可用),在挑战期内(通常1-7天) # 任何验证者可以提交欺诈证明推翻结果 # 这种模式兼顾了用户体验(低延迟)和安全性(事后验证) # 4. 费用估算在推理前进行 —— # 同时估算推理计算成本(GPU费用)和链上提交成本(Gas) # 用户预先支付,多退少补 from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import Optional import hashlib import time import json from web3 import Web3 class ProveStrategy(Enum): ZK = "zk" # 零知识证明: 小模型(≤7B) TEE = "tee" # 可信执行环境: 中等模型(7B-70B) OPTIMISTIC = "op" # 乐观验证: 大模型(>70B) class ModelSize(Enum): SMALL = "small" # ≤7B 参数 MEDIUM = "medium" # 7B-70B LARGE = "large" # >70B @dataclass class InferenceTask: """推理任务生命周期对象""" task_id: str model_id: str model_size: ModelSize input_text: str output_text: str = "" prove_strategy: Optional[ProveStrategy] = None proof: bytes = b"" ipfs_cid: str = "" tx_hash: str = "" status: str = "pending" # pending → inferring → proving → submitting → complete created_at: float = field(default_factory=time.time) class InferenceEngine(ABC): """模型推理引擎抽象""" @abstractmethod def infer(self, model_id: str, prompt: str, max_tokens: int = 1024) -> str: pass @abstractmethod def get_model_size(self, model_id: str) -> ModelSize: pass class VLLMEngine(InferenceEngine): """vLLM推理引擎 —— 高性能批处理推理""" def __init__(self, api_url: str, api_key: str = ""): self.api_url = api_url self.api_key = api_key # 模型大小注册表 —— 用于自动选择证明策略 self._model_sizes = { "llama-3-8b": ModelSize.SMALL, "mistral-7b": ModelSize.SMALL, "llama-3-70b": ModelSize.MEDIUM, "mixtral-8x7b": ModelSize.MEDIUM, "llama-3-405b": ModelSize.LARGE, } def infer(self, model_id: str, prompt: str, max_tokens: int = 1024) -> str: response = requests.post( f"{self.api_url}/v1/completions", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model": model_id, "prompt": prompt, "max_tokens": max_tokens, "temperature": 0.0, # 链上推理建议 temperature=0 确保确定性 }, timeout=300, ) return response.json()["choices"][0]["text"] def get_model_size(self, model_id: str) -> ModelSize: return self._model_sizes.get(model_id, ModelSize.MEDIUM) class ProveStrategySelector: """证明策略自动选择器""" # 策略选择矩阵: (模型大小, 延迟要求) → 证明策略 @staticmethod def select(model_size: ModelSize, max_latency_ms: int = 5000) -> ProveStrategy: if model_size == ModelSize.SMALL: # 小模型: ZK证明延迟通常在2-5秒,可接受 return ProveStrategy.ZK elif model_size == ModelSize.MEDIUM: # 中等模型: ZK证明生成耗时10-60秒 if max_latency_ms < 10000: return ProveStrategy.TEE # TEE延迟: 1-3秒增量 return ProveStrategy.ZK else: # 大模型: ZK/TEE都不现实,使用乐观验证 return ProveStrategy.OPTIMISTIC @staticmethod def generate_proof( strategy: ProveStrategy, model_id: str, input_text: str, output_text: str ) -> bytes: """生成证明""" if strategy == ProveStrategy.TEE: # TEE: 在安全飞地中执行推理并签名结果 # 实际实现需要 Intel SGX SDK 或类似工具 # 此处为简化示例: 对(模型ID + 输入哈希 + 输出哈希)做HMAC签名 content = f"{model_id}:{hashlib.sha256(input_text.encode()).hexdigest()}:{hashlib.sha256(output_text.encode()).hexdigest()}" return hashlib.sha256(content.encode()).digest() elif strategy == ProveStrategy.ZK: # ZK: 使用ezkl或其他ZkML框架生成证明 # ezkl需要将模型转换为ONNX格式 + 生成电路 # 此处为占位 return hashlib.sha256(f"zk_proof:{input_text[:50]}".encode()).digest() else: # OpML: 乐观验证不需要预先生成证明 # 返回空证明, 欺诈证明在挑战期生成 return b"" class OnchainAIService: """链上AI服务主类""" def __init__(self, engine: InferenceEngine, web3_provider: str, contract_address: str): self.engine = engine self.w3 = Web3(Web3.HTTPProvider(web3_provider)) self.contract_address = contract_address self.tasks: dict[str, InferenceTask] = {} def submit_request(self, model_id: str, input_text: str) -> InferenceTask: """提交推理请求 —— 用户入口""" task_id = hashlib.sha256( f"{model_id}:{input_text}:{time.time()}".encode() ).hexdigest()[:16] task = InferenceTask( task_id=task_id, model_id=model_id, model_size=self.engine.get_model_size(model_id), input_text=input_text, prove_strategy=ProveStrategySelector.select( self.engine.get_model_size(model_id) ), ) self.tasks[task_id] = task return task def execute(self, task: InferenceTask) -> InferenceTask: """执行全链路: 推理 → 证明 → 链上提交""" # Step 1: 模型推理 task.status = "inferring" task.output_text = self.engine.infer(task.model_id, task.input_text) # Step 2: 生成证明 task.status = "proving" task.proof = ProveStrategySelector.generate_proof( task.prove_strategy, task.model_id, task.input_text, task.output_text, ) # Step 3: 上传结果到IPFS task.ipfs_cid = self._upload_to_ipfs(task) # Step 4: 提交到链上 task.status = "submitting" task.tx_hash = self._submit_onchain(task) task.status = "complete" return task def _upload_to_ipfs(self, task: InferenceTask) -> str: """上传推理结果到IPFS —— 生成标准化的元数据""" metadata = { "model_id": task.model_id, "input_hash": hashlib.sha256(task.input_text.encode()).hexdigest(), "output": task.output_text, "prove_strategy": task.prove_strategy.value, "proof_hex": task.proof.hex() if task.proof else "", "timestamp": int(task.created_at), "version": "1.0", } # 实际上传使用 Pinata/Web3.Storage API # 此处简化为返回内容哈希 return hashlib.sha256(json.dumps(metadata).encode()).hexdigest() def _submit_onchain(self, task: InferenceTask) -> str: """提交推理结果到链上合约""" # 调用链上合约的 submitInferenceResult 方法 # 包含: task_id, ipfs_cid, proof # 实际实现需要 wagmi/viem 或 ethers.js return f"0x{task.task_id}"

四、全链路的硬约束与运维保障

Gas成本与推理成本的平衡：链上AI服务有两层成本——推理计算成本（GPU时间）和链上提交成本（Gas费）。在Ethereum主网上，单次链上存储（写入storage slot）的成本约为20,000 gas（约$2-5），而一次Llama-3-8B的推理计算成本约为$0.001（A100 GPU上约0.5秒）。也就是说，链上存储成本是推理计算成本的2000-5000倍。这个比例在L2上会改善（Arbitrum上约$0.05-0.1/次存储），但仍然远超计算成本。

因此，链上AI服务的成本优化重点不在推理端，而在存储端。有效策略包括：将推理结果存储在IPFS（去中心化但成本低），仅将IPFS CID写入链上（32字节 vs 完整文本）；使用Event Log而非Storage来存储结果（更低的Gas成本但不可由合约读取）；在L2上部署AI服务合约（降低Gas成本但牺牲一定的去中心化程度）。

模型版本升级的兼容性：当推理模型从Llama-3-8B升级到Llama-3-8B-v2时，同样的输入可能产生不同的输出。这对链上AI服务的"可复现性"是一个挑战——已经上链的推理结果（如一个AI生成的预言机报价）无法在新的模型版本下复现。解决方案：在链上维护模型版本注册表，每个推理结果记录所使用的模型版本号，验证时使用对应版本的模型。

五、总结

链上AI服务的7月工程实践让我们清楚地看到了"推理"只是全链路中的一个环节，而且往往不是最难或最昂贵的环节。可验证性（证明生成和验证）是技术深水区，链上存储是成本瓶颈，运维保障（模型升级、节点容灾、争议处理）是长期可持续性的保障。

三个值得在8月深入的方向：zkML的性能突破（Giza和EZKL的证明生成时间能否降到10秒以内，使ZK验证在大模型上可行）、跨链AI服务的标准化（不同链上的AI推理如何共享结果、如何统一验证格式）、以及链上AI的DAO治理（模型版本升级、费用调整、争议仲裁等决策如何通过去中心化治理完成，而不依赖中心化的"AI服务团队"）。