链上 AI 预言机设计：可信数据喂入 + 推理结果上链的工程方案
一、智能合约不能直接调用 ChatGPT
智能合约运行在区块链的确定性环境中，无法发起 HTTP 请求，也无法调用 OpenAI 的 API。这就产生了一个矛盾：DeFi 协议需要 AI 来做风险评估，NFT 市场需要 AI 来验证内容原创性，但智能合约本身没有 AI 能力。
预言机（Oracle）就是桥梁——将链外数据传送给链上合约。而 AI 预言机更进一步，不仅传递数据，还传递模型的推理结果。
但这里有一个信任问题：你怎么证明链外的 AI 推理没有被篡改？
二、AI 预言机架构：数据喂入 + 推理验证
flowchart TB
subgraph OffChain["链下（Off-Chain）"]
Data["外部数据源\\n（行情、新闻、链上数据）"] –> Preprocess["数据预处理\\n清洗 + 标准化"]
Preprocess –> Model["AI 模型推理\\n（风险评分、内容审核等）"]
Model –> Proof["生成推理证明\\nZK-SNARK / TEE 签名"]
end
subgraph Oracle["预言机网络"]
Proof –> Node1["预言机节点 1\\nAI 推理"]
Proof –> Node2["预言机节点 2\\nAI 推理"]
Proof –> Node3["预言机节点 3\\nAI 推理"]
Node1 –> Agg["结果聚合\\n多数一致 + 中位数"]
Node2 –> Agg
Node3 –> Agg
Agg –> Sign["多方签名\\n门限签名（TSS）"]
end
subgraph OnChain["链上（On-Chain）"]
Sign –> Contract["智能合约"]
Contract –> Verify["验证签名 + 检查结果一致性"]
Verify –> Execute["执行业务逻辑\\n（清算、定价、版税）"]
end
subgraph Trust["信任机制"]
Trust1["经济激励: 节点质押 + 罚没"]
Trust2["TEE: 可信执行环境（Intel SGX）"]
Trust3["ZK: 零知识证明（推理过程可验证）"]
end
三、Python 实现 AI 预言机节点
预言机节点核心逻辑
import hashlib
import json
import time
from typing import List, Optional, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
from web3 import Web3
from eth_account import Account
from eth_account.messages import encode_defunct
class OracleNodeStatus(Enum):
ACTIVE = "active"
SLASHED = "slashed" # 被罚没
INACTIVE = "inactive"
@dataclass
class AIInferenceResult:
"""AI 推理结果——需要上链的结构"""
request_id: str # 请求唯一 ID
model_name: str # 模型名称（如 "risk_scorer_v2"）
model_version: str # 模型版本哈希
input_hash: str # 输入数据的哈希（用于验证输入一致性）
result: Dict[str, Any] # 推理结果
confidence: float # 模型置信度 [0, 1]
timestamp: int # Unix 时间戳
node_address: str # 预言机节点地址
tss_signature: Optional[str] = None # 门限签名（聚合后添加）
@dataclass
class OracleNode:
"""AI 预言机节点"""
private_key: str
contract_address: str
contract_abi: list
web3_provider: str
stake_amount: int # 质押金额（Wei）
status: OracleNodeStatus = OracleNodeStatus.INACTIVE
_web3: Optional[Web3] = field(default=None, repr=False)
def __post_init__(self):
self._web3 = Web3(Web3.HTTPProvider(self.web3_provider))
self.account = Account.from_key(self.private_key)
def stake(self) -> str:
"""质押——成为活跃的预言机节点"""
try:
contract = self._web3.eth.contract(
address=Web3.to_checksum_address(self.contract_address),
abi=self.contract_abi,
)
tx = contract.functions.stake().build_transaction({
"from": self.account.address,
"value": self.stake_amount,
"nonce": self._web3.eth.get_transaction_count(self.account.address),
"gas": 300000,
"gasPrice": self._web3.eth.gas_price,
})
signed = self.account.sign_transaction(tx)
tx_hash = self._web3.eth.send_raw_transaction(signed.raw_transaction)
receipt = self._web3.eth.wait_for_transaction_receipt(tx_hash)
if receipt.status == 1:
self.status = OracleNodeStatus.ACTIVE
return tx_hash.hex()
raise Exception("质押交易失败")
except Exception as e:
print(f"质押失败: {e}")
raise
def process_inference_request(
self,
request_id: str,
input_data: bytes,
model_func, # AI 推理函数
) -> AIInferenceResult:
"""
处理推理请求
1. 计算输入哈希
2. 执行 AI 推理
3. 生成签名
"""
# 计算输入哈希（确保所有节点使用相同输入）
input_hash = hashlib.sha3_256(input_data).hexdigest()
# 执行 AI 推理
try:
result, confidence = model_func(input_data)
except Exception as e:
print(f"推理失败: {e}")
return AIInferenceResult(
request_id=request_id,
model_name="error",
model_version="",
input_hash=input_hash,
result={"error": str(e)},
confidence=0.0,
timestamp=int(time.time()),
node_address=self.account.address,
)
# 构造推理结果
inference = AIInferenceResult(
request_id=request_id,
model_name=getattr(model_func, "__name__", "unknown"),
model_version=hashlib.md5(
str(model_func.__code__).encode()
).hexdigest()[:8],
input_hash=input_hash,
result=result,
confidence=confidence,
timestamp=int(time.time()),
node_address=self.account.address,
)
return inference
def sign_result(self, inference: AIInferenceResult) -> str:
"""
对推理结果签名
——用于链上验证该结果来自可信节点
"""
# 构造签名消息
message = json.dumps({
"request_id": inference.request_id,
"input_hash": inference.input_hash,
"result_hash": hashlib.sha3_256(
json.dumps(inference.result, sort_keys=True).encode()
).hexdigest(),
"confidence": inference.confidence,
"timestamp": inference.timestamp,
}, sort_keys=True)
# 以太坊格式签名
encoded = encode_defunct(text=message)
signed = self.account.sign_message(encoded)
return signed.signature.hex()
聚合器：多节点结果共识
from collections import Counter
from statistics import median
class OracleAggregator:
"""预言机聚合器——从多个节点收集结果并达成共识"""
def __init__(self, min_confirmations: int = 3, tolerance: float = 0.1):
"""
min_confirmations: 至少需要多少个节点返回结果
tolerance: 结果之间的允许偏差（用于判断一致）
"""
self.min_confirmations = min_confirmations
self.tolerance = tolerance
# 待处理的推理请求
self._pending: Dict[str, List[AIInferenceResult]] = {}
def collect_result(self, inference: AIInferenceResult) -> Optional[AIInferenceResult]:
"""
收集单个节点的推理结果
如果收集到足够多的结果，返回聚合后的最终结果
"""
rid = inference.request_id
if rid not in self._pending:
self._pending[rid] = []
self._pending[rid].append(inference)
# 检查是否收集满
if len(self._pending[rid]) >= self.min_confirmations:
return self._aggregate(self._pending.pop(rid))
return None
def _aggregate(self, results: List[AIInferenceResult]) -> AIInferenceResult:
"""聚合多个节点的推理结果"""
# 策略一：多数一致（适合分类任务）
# 策略二：取中位数（适合回归任务）
# 策略三：加权平均（按质押量加权）
# 检查输入哈希是否一致
input_hashes = Counter(r.input_hash for r in results)
if len(input_hashes) > 1:
print(f"[WARN] 输入数据不一致! 哈希分布: {input_hashes}")
# 取多数哈希
majority_hash = input_hashes.most_common(1)[0][0]
results = [r for r in results if r.input_hash == majority_hash]
# 取置信度中位数（稳健统计）
confidences = [r.confidence for r in results if r.confidence > 0]
median_confidence = median(confidences) if confidences else 0.0
# 聚合结果：取出现最多的结果项
result_items = [
json.dumps(r.result, sort_keys=True) for r in results
]
majority_result_str = Counter(result_items).most_common(1)[0][0]
aggregated_result = json.loads(majority_result_str)
return AIInferenceResult(
request_id=results[0].request_id,
model_name=results[0].model_name,
model_version=results[0].model_version,
input_hash=results[0].input_hash,
result=aggregated_result,
confidence=median_confidence,
timestamp=int(time.time()),
node_address="aggregated",
)
def submit_to_chain(
self,
web3: Web3,
contract_address: str,
abi: list,
inference: AIInferenceResult,
private_key: str,
) -> Optional[str]:
"""将聚合后的结果提交到链上合约"""
try:
account = Account.from_key(private_key)
contract = web3.eth.contract(
address=Web3.to_checksum_address(contract_address),
abi=abi,
)
# 调用合约方法提交结果
tx = contract.functions.submitInferenceResult(
inference.request_id,
json.dumps(inference.result),
int(inference.confidence * 10000), # 存为整数
inference.timestamp,
).build_transaction({
"from": account.address,
"nonce": web3.eth.get_transaction_count(account.address),
"gas": 500000,
"gasPrice": web3.eth.gas_price,
})
signed = account.sign_transaction(tx)
tx_hash = web3.eth.send_raw_transaction(signed.raw_transaction)
receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
if receipt.status == 1:
return tx_hash.hex()
return None
except Exception as e:
print(f"上链提交失败: {e}")
return None
四、边界分析与 Trade-offs
去中心化程度 vs 延迟：
- 更多预言机节点 → 更高的去中心化程度 → 更长的聚合等待时间
- 3-5 个节点是性价比拐点
推理可验证性：
- 链上无法重放 AI 推理，因此无法通过计算验证结果正确性
- 替代方案：TEE（Intel SGX）证明推理运行在可信环境，或 ZK-ML 证明推理过程的正确性
- 当前实践中，经济激励（质押 + 罚没）比技术证明更实用
数据新鲜度：
- 链上需要的是某个时间点的推理结果，而不是实时流
- 预言机节点应附带数据的时间戳，链上合约检查新鲜度窗口
Gas 成本控制：
- 每条推理结果上链需要消耗 Gas
- 建议：仅上链哈希 + 关键字段，完整结果存在 IPFS 上
- 合约仅验证签名和哈希，不存储大量数据
五、总结
AI 预言机设计的核心是解决"链下 AI 推理结果如何在链上被信任"：
记住：不要试图在链上做 AI 推理，也不要在链下放弃信任机制。预言机的价值在于两者之间的平衡设计。
