news 2026/7/31 20:45:35

decentralized_agent.py

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
decentralized_agent.py

一、去中心化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需要增加"证明验证"的工程步骤。

五、总结

去中心化AI推理在7月不再是概念展示,而是进入了"有生产级案例可参考"的阶段。Ritual的50万次链上推理请求证明了TEE方案的工程可行性,Bittensor的78个子网展现了激励驱动网络的扩展性,Gensyn的分布式训练测试网为"去中心化训练大模型"这个更困难的问题迈出了第一步。

当前阶段的技术选型应该从"信仰驱动"转向"场景驱动":你的应用是否需要可验证的推理确定性(选Ritual)?你的应用是内容生成为主且对错误容忍度较高(选Bittensor)?还是你需要在大模型训练成本上获得突破(关注Gensyn进展)?这三个问题比"哪个项目更去中心化"更有工程指导意义。

8月值得关注的事件:Ritual计划公布的TEE多厂商支持(AMD SEV + Intel TDX)、Bittensor的Dynamic TAO升级(改进通胀分配机制)、以及OpenAI发布的新模型在去中心化推理网络上的部署可行性评估。

资料说明

本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论,不应视为行业事实。可参考 0731 资料来源索引,并在发布前将具体来源贴到对应断言之后。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/7/31 20:45:33

Skywork-Reward-V2-Qwen3-8B分布式部署教程:SGLang实现高吞吐量推理

Skywork-Reward-V2-Qwen3-8B分布式部署教程&#xff1a;SGLang实现高吞吐量推理 【免费下载链接】Skywork-Reward-V2-Qwen3-8B 项目地址: https://ai.gitcode.com/hf_mirrors/Skywork/Skywork-Reward-V2-Qwen3-8B Skywork-Reward-V2-Qwen3-8B是一款基于Qwen3-8B架构的奖…

作者头像 李华
网站建设 2026/7/31 20:41:22

OpenControl源码探秘:核心组件设计与AI工具调用实现原理

OpenControl源码探秘&#xff1a;核心组件设计与AI工具调用实现原理 【免费下载链接】opencontrol ⏣ Control your infrastructure with AI. 项目地址: https://gitcode.com/gh_mirrors/op/opencontrol OpenControl是一个创新的开源项目&#xff0c;旨在通过AI技术实现…

作者头像 李华
网站建设 2026/7/31 20:40:33

IRust与Jupyter集成:打造强大的Rust数据分析工作流

IRust与Jupyter集成&#xff1a;打造强大的Rust数据分析工作流 【免费下载链接】IRust Cross Platform Rust Repl 项目地址: https://gitcode.com/gh_mirrors/ir/IRust IRust作为一款跨平台的Rust交互式解释器&#xff08;REPL&#xff09;&#xff0c;与Jupyter的集成开…

作者头像 李华
网站建设 2026/7/31 20:40:05

如何用metrics-spring监控Spring应用?5分钟快速上手教程

如何用metrics-spring监控Spring应用&#xff1f;5分钟快速上手教程 【免费下载链接】metrics-spring Spring integration for Metrics 项目地址: https://gitcode.com/gh_mirrors/me/metrics-spring metrics-spring是一款专为Spring应用打造的监控集成工具&#xff0c;…

作者头像 李华