当前位置：首页 > news >正文

decentralized_agent.py

news 2026/7/31 20:45:35

一、去中心化AI推理：从叙事到工程化的拐点

2026年7月，去中心化AI推理领域经历了从"为什么需要去中心化推理"到"如何实现可验证的去中心化推理"的话语转折。这个转折的标志性事件包括：Ritual的Infernet主网启动并处理了超过50万次链上推理请求、Bittensor的子网生态从年初的32个增长到78个、Gensyn完成了针对大模型分布式训练的测试网基准测试。

去中心化AI的核心命题可以拆解为三个子问题：推理的正确性验证（如何确信矿工/节点真的运行了指定的模型）、推理的可用性保证（如何在节点下线或作恶时仍能获得推理结果）、以及模型的分发与版本管理（如何确保全网节点运行的是同一个模型文件/同一个版本的权重）。

本文将7月中在AI+Web3交叉领域中最值得关注的技术进展整合为一套选型框架，覆盖推理网络、Agent框架和模型托管三个维度。

二、三大去中心化推理方案的架构对比

Bittensor的核心理念是"用经济激励替代信任"——任何运行模型并提供API的节点都可以注册为矿工，验证者通过发送测试查询并评估响应质量来评判矿工。如果矿工返回错误结果或使用劣质模型，其权益（staked TAO）会被削减。这种模式的优势在于开放性和无需许可（任何人可以成为矿工），但验证机制的可靠性是一个持续被讨论的问题——如果验证者本身不具备评估优质响应的能力（例如对专业领域的文本生成），激励系统可能奖励"听起来合理"而非"事实上正确"的回答。

Ritual走的是与Bittensor互补的路线——它不依赖纯粹的经济激励，而是通过TEE（可信执行环境）和链上智能合约来实现推理的可验证性。Ritual的Infernet SDK将AI推理包装为一个"链上可验证的预言机"：用户在链上发起推理请求，链下节点在TEE中执行推理（TEE签名证明节点确实运行了指定的模型和输入），结果通过Guardian验证者上链。这种模式在"可验证性"维度上远强于纯激励模式，但TEE带来了硬件依赖（需要支持Intel SGX或AMD SEV的CPU）。

Gensyn聚焦于另一个维度——去中心化的分布式训练。Bittensor和Ritual解决了"推理"的去中心化，Gensyn解决的是"训练"的去中心化。其核心技术挑战是：如何在不信任的分布式节点上完成大模型的训练任务，并验证每个节点确实执行了分配的计算（而非伪造梯度更新）。Gensyn的spot-check协议通过随机抽查节点的中间计算结果来检测作弊，这比完整验证所有节点的计算更经济。

三、Agent框架在去中心化AI中的编排实现

以下是一个基于LangChain + Ritual Infernet SDK的Agent编排示例，展示如何在去中心化推理网络上构建AI Agent：

# decentralized_agent.py # 去中心化AI推理Agent —— 使用Ritual Infernet实现可验证的推理调用 # # 设计决策： # 1. 在 Agent 层构建"推理来源抽象" —— 通过 InferenceProvider 接口分离 # 具体推理来源（Bittensor / Ritual / 本地模型），Agent核心逻辑不关心 # 推理请求的物理执行位置 # 2. 对链上推理请求附加"验证要求" —— 不同的推理任务有不同的验证严格度: # - 内容生成(category=creative): 接受概率性验证(经济激励) # - 代码生成(category=code): 要求TEE证明(确定性验证) # - 金融推理(category=finance): 要求TEE + 共识多重确认 # 3. 推理结果带过期时间缓存 —— 对于相同输入的不变性查询(如"模型元数据"), # 在TTL内使用缓存结果,减少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 import requests class VerificationLevel(Enum): """推理验证级别 —— 从宽松到严格""" BEST_EFFORT = "best_effort" # 经济激励保证，不验证计算正确性 TEE_ATTESTED = "tee_attested" # TEE硬件签名，证明特定模型运行 MULTI_SIG = "multi_sig" # 多节点共识，3/5+ 节点结果一致 class ReasoningCategory(Enum): """推理任务类别 → 映射到验证级别""" CREATIVE = VerificationLevel.BEST_EFFORT # 文本/图像生成 CODE = VerificationLevel.TEE_ATTESTED # 代码生成 FINANCE = VerificationLevel.MULTI_SIG # 市场分析/交易决策 @dataclass class InferenceRequest: """推理请求数据结构""" model_id: str # 模型注册表ID,如 "llama-3-8b-instruct" prompt: str max_tokens: int = 1024 temperature: float = 0.7 category: ReasoningCategory = ReasoningCategory.CREATIVE def request_hash(self) -> str: """计算请求的内容哈希 —— 用于缓存去重""" payload = json.dumps({ "model": self.model_id, "prompt": self.prompt, "max_tokens": self.max_tokens, "temperature": self.temperature, }, sort_keys=True) return hashlib.sha256(payload.encode()).hexdigest() @dataclass class InferenceResult: """推理结果""" output: str model_id: str verification_level: VerificationLevel attestation_proof: Optional[bytes] = None # TEE签名或ZK proof latency_ms: int = 0 cost_wei: int = 0 class InferenceProvider(ABC): """推理来源抽象接口""" @abstractmethod def infer(self, request: InferenceRequest) -> InferenceResult: """执行推理""" pass @abstractmethod def verify(self, request: InferenceRequest, result: InferenceResult) -> bool: """验证推理结果 —— 不同提供者的验证逻辑不同""" pass class RitualInferenceProvider(InferenceProvider): """Ritual Infernet 推理提供者 通过TEE验证确保推理结果的正确性。 适用场景: 对推理结果有可验证要求的DeFi/DAO应用 """ INFERNET_API = "https://api.ritual.network/v1" MODEL_REGISTRY = "0x..." # Ritual模型注册表合约地址 def __init__(self, api_key: str, min_attestation_level: str = "SGX"): self.api_key = api_key self.min_attestation_level = min_attestation_level # 支持模型列表缓存 —— TTL=3600s，减少链上查询 self._supported_models: dict[str, float] = {} def infer(self, request: InferenceRequest) -> InferenceResult: """通过Ritual Infernet执行推理 流程: 链上创建推理任务 → 链下TEE节点执行 → Guardian验证 → 结果返回 """ start = time.time() # Step 1: 验证模型是否在Ritual注册表中 if request.model_id not in self._get_supported_models(): raise ValueError(f"Model {request.model_id} not in Ritual registry") # Step 2: 提交推理请求并指定验证要求 verification = request.category.value # finance类别需要多节点确认 if request.category == ReasoningCategory.FINANCE: verification = "tee_attested_multi" response = requests.post( f"{self.INFERNET_API}/infer", headers={"Authorization": f"Bearer {self.api_key}"}, json={ "model_id": request.model_id, "prompt": request.prompt, "max_tokens": request.max_tokens, "temperature": request.temperature, "verification_level": verification, "min_attestation": self.min_attestation_level, }, timeout=120, # 链上确认可能需要较长时间 ) data = response.json() latency = int((time.time() - start) * 1000) return InferenceResult( output=data["output"], model_id=request.model_id, verification_level=VerificationLevel.TEE_ATTESTED, attestation_proof=bytes.fromhex(data.get("attestation", "")), latency_ms=latency, cost_wei=int(data.get("gas_used", 0)), ) def verify(self, request: InferenceRequest, result: InferenceResult) -> bool: """验证TEE证明的有效性""" if not result.attestation_proof: return False # 链上验证TEE签名 response = requests.post( f"{self.INFERNET_API}/verify", json={ "request_hash": request.request_hash(), "attestation": result.attestation_proof.hex(), "expected_model": request.model_id, }, ) return response.json().get("valid", False) def _get_supported_models(self) -> dict: """获取Ritual支持的模型列表(带缓存)""" now = time.time() if self._supported_models and next(iter(self._supported_models.values())) > now: return self._supported_models response = requests.get( f"{self.INFERNET_API}/models", headers={"Authorization": f"Bearer {self.api_key}"}, ) models = response.json()["models"] # 缓存1小时 self._supported_models = {m["id"]: now + 3600 for m in models} return self._supported_models class HybridAgent: """混合AI Agent —— 根据任务类别自动选择推理提供者""" def __init__(self): self.providers = { VerificationLevel.TEE_ATTESTED: RitualInferenceProvider( api_key="ritual_api_key" ), # 可扩展其他提供者: Bittensor、本地Ollama等 } # 推理结果LRU缓存 —— 避免重复请求 self._cache: dict[str, tuple[float, InferenceResult]] = {} self._cache_ttl = 300 # 5分钟 def think(self, request: InferenceRequest) -> InferenceResult: """Agent思考入口 —— 自动路由到合适的推理提供者""" req_hash = request.request_hash() # 检查缓存 cached = self._cache.get(req_hash) if cached and time.time() < cached[0]: return cached[1] # 根据验证级别选择提供者 provider = self.providers.get(request.category.value) if not provider: raise ValueError(f"No provider for {request.category}") result = provider.infer(request) # 写入缓存 self._cache[req_hash] = (time.time() + self._cache_ttl, result) return result def execute_chain_action(self, reasoning: str, action: str, params: dict): """基于推理结果执行链上操作 这是一个概念性的占位——实际的链上操作需要连接 wagmi/viem或ethers.js来发送交易 """ # 验证推理的可验证性 # 仅在推理结果附带有效证明时才执行链上操作 pass

四、三个方案的适用边界与切换成本

Bittensor的适用边界：最适合"质量容错"的推理场景——文本摘要、情感分析、创意写作。这些场景下，结果的质量差异不会造成经济损失，因此纯粹的经济激励就足够。不适合的场景是财务计算、合约审计、代码生成——这些场景需要确定性正确性保证（错误的推理可能造成链上资金损失），Bittensor的验证机制只保证"质量相对高低"而非"结果绝对正确"。

Ritual的适用边界：最适合需要"链上可验证性"的DeFi和DAO场景——价格预言、风险分析、治理提案生成。TEE证明提供了"这个推理确实由指定模型在特定硬件上执行"的加密学保证。但局限性在于TEE本身的安全性争议——SGX在过去几年多次被发现侧信道漏洞（如SGAxe、Platypus等），使得一些安全团队对TEE的绝对安全性持保留态度。

切换成本：从Bittensor切换到Ritual，API层的修改量不大（都是HTTP POST + 模型ID + prompt），但验证逻辑完全不同。Bittensor的验证在链下通过子网验证者完成（对调用方透明），Ritual的验证需要调用方主动验证attestation proof。这意味着从Bittensor迁移到Ritual需要增加"证明验证"的工程步骤。