AI驱动的去中心化推理服务弹性调度：链上负载感知的自动扩缩容

AI驱动的去中心化推理服务弹性调度：链上负载感知的自动扩缩容

一、搜索困境：为什么 DApp 不能只有 SQL LIKE？

AI 驱动架构的核心优势在于将不确定性推理转化为确定性调度。在去中心化推理服务场景中，节点的 GPU 算力、网络带宽、内存容量等资源是异构且动态变化的——某个推理节点可能在处理批量请求时突发负载，而另一个节点则处于空闲状态。传统的手动扩缩容或静态负载均衡策略无法感知链上推理任务的实时分布，更无法根据区块高度、Gas 价格、链上活动量等 Web3 原生指标做出智能决策。这正是 AI 驱动的弹性调度要解决的问题：通过 Embedding 将节点负载状态、任务队列深度、历史推理延时等多维特征向量化，利用语义检索在调度空间中匹配最优节点，再由启发式算法动态调整节点池的规模——既避免资源浪费，又防止单点过载导致的推理超时。

更深层的问题在于链上推理任务本身缺乏语义标注。一个用户的推理请求（"请分析这笔交易的 MEV 风险"）在调度层看来只是一个 JSON-RPC 调用，不携带其对计算资源的需求信息。调度器需要根据请求的语义特征（涉及多少合约调用、是否需要历史区块回溯、预期输出复杂度）推断出资源开销——这正是 AI 检索增强生成（RAG, Retrieval-Augmented Generation）架构与弹性调度交汇的地方。将历史推理请求的特征向量化存入向量数据库，新请求到来时先通过语义检索找到最相似的历史任务，利用其实际执行数据（GPU 占用、推理延时、内存峰值）作为负载预估的依据，再结合当前节点池的健康状态，做出"是否扩缩容"的实时决策。

RAG 架构提供了把这种调度逻辑工程化的路径：把推理请求的特征、节点负载快照、历史扩缩容决策三者预先向量化存入向量数据库，调度器在每次决策时先通过语义检索找到最相关的历史案例和当前最佳节点，再交由启发式引擎生成具体的扩缩容指令。

flowchart TD A[链上数据源] --> B[数据提取层] B --> C1[事件日志] B --> C2[合约 ABI] B --> C3[协议文档] C1 --> D[分块器 Chunker] C2 --> D C3 --> D D --> E[Embedding 模型] E --> F[(Milvus 向量库)] G[用户自然语言查询] --> H[Query Embedding] H --> I[向量相似度检索] F --> I I --> J[Top-K 候选片段] J --> K[LLM 重排序 + 生成] K --> L[结构化答案 + 链上验证链接] subgraph 索引链路 A --> B --> D --> E --> F end subgraph 检索链路 G --> H --> I --> J --> K --> L end

二、RAG 在链上数据场景的特殊挑战

通用 RAG 架构通常处理的是文本文档 —— PDF、网页、API 文档等。但链上数据有四个特殊性质需要针对性处理：

时序性。链上数据是严格按区块高度排序的时间序列。一次"Swap"操作的语义取决于它前后的相关交易。如果 Embedding 时只考虑单条事件的文本描述，会丢失交易序列的上下文。解决方案是在分块（Chunking）时把相关事件的因果链信息注入到 chunk 的 metadata 中。

数值敏感性。用户搜索"APY 大于 20% 的池子"时，向量检索本身不擅长数值范围过滤。需要采用混合检索：向量检索负责语义匹配，结构化过滤负责数值条件。Milvus 的标量过滤 + 向量检索的混合模式恰好匹配这个需求。

地址作为一等实体。0x 地址在 DApp 搜索中是最常见的查询目标，但 Embedding 模型对十六进制字符串的语义理解几乎为零。需要为地址建立知识图谱：将地址解析为协议标签、交互历史和资产画像，然后将这些语义丰富的描述文本送入 Embedding。

可验证性要求。DApp 场景下，用户需要对检索结果做链上验证。RAG 的每一步检索结果都应附带链上证据链接（Tx Hash 或区块高度），让用户可以直接在区块浏览器上溯源。

sequenceDiagram participant User as 用户 participant API as 搜索 API participant ES as Milvus 向量库 participant Graph as 地址知识图谱 participant RPC as 链上 RPC participant LLM as LLM 推理 User->>API: "最近24小时 TVL 变化最大的借贷协议" API->>API: Query Embedding API->>ES: 向量检索 + 标量过滤(time > now-24h) ES-->>API: Top-20 候选 alt 候选涉及合约地址 API->>Graph: 解析地址标签 Graph-->>API: 协议名称 + 分类 end API->>LLM: Prompt + Top-5 候选片段 LLM-->>API: 生成结构化答案 API->>RPC: 实时校验最新 TVL RPC-->>API: 当前链上数据 API-->>User: 答案 + 数据来源链接

三、工程实现：从索引到检索的完整链路

3.1 数据提取与分块策略

""" chain_indexer.py — 链上数据抽取与分块器 设计考量： - 分块大小不是固定的 token 数：一条完整的 Swap 日志块包含 TxHash + 相关事件 + 上下文 - 使用 metadata 字段携带时序信息（block_number, tx_index），让 LLM 了解数据的时间顺序 - 地址自动解析：合约地址通过标签库转换为协议名，提升 Embedding 语义质量 """ import asyncio from dataclasses import dataclass, field from datetime import datetime, timezone from typing import AsyncIterator import httpx @dataclass class ChainChunk: """链上数据块""" content: str # Embedding 的输入文本 metadata: dict # 结构化元数据（用于混合检索的标量过滤） sources: list[str] # 链上证据链接 @dataclass class BlockRange: start: int end: int # 地址到协议标签的映射表，每次启动时从配置文件加载 # 为什么不用链上查询：ENS 反向解析速度慢且标签不完整 ADDRESS_LABELS: dict[str, str] = {} def load_labels(path: str = "config/address_labels.json"): """加载地址标签映射表""" import json with open(path) as f: ADDRESS_LABELS.update(json.load(f)) def resolve_address(addr: str) -> str: """将合约地址解析为人类可读的协议名称""" label = ADDRESS_LABELS.get(addr.lower(), "") if label: return f"{label}({addr[:10]}...)" return addr class EventChunker: """ 事件分块器 一条 Swap 事件块的结构： - 区块高度 / 时间戳 - 协议名称 - 操作类型 - Token 信息（输入/输出/数量） - 关联事件（如 Sync 事件中的池子储备变化） """ def chunk_swap_event(self, event: dict) -> ChainChunk: pool_label = resolve_address(event.get("address", "")) token0 = event.get("args", {}).get("token0", "未知") token1 = event.get("args", {}).get("token1", "未知") amount0 = event.get("args", {}).get("amount0", 0) amount1 = event.get("args", {}).get("amount1", 0) block = event.get("blockNumber", 0) content = ( f"协议: {pool_label}\n" f"操作: Swap\n" f"输入: {amount0} {token0}\n" f"输出: {amount1} {token1}\n" f"区块: {block}\n" f"交易哈希: {event.get('transactionHash', '')}\n" ) return ChainChunk( content=content, metadata={ "type": "swap", "protocol": pool_label, "block_number": block, "timestamp": datetime.now(timezone.utc).isoformat(), "volume_usd": max(abs(amount0), abs(amount1)), }, sources=[f"https://etherscan.io/tx/{event.get('transactionHash', '')}"], ) def chunk_liquidity_event(self, event: dict) -> ChainChunk: """流动性添加/移除事件的分块处理""" pool_label = resolve_address(event.get("address", "")) event_type = "AddLiquidity" if event.get("event") == "Mint" else "RemoveLiquidity" amount0 = event.get("args", {}).get("amount0", 0) amount1 = event.get("args", {}).get("amount1", 0) content = ( f"协议: {pool_label}\n" f"操作: {event_type}\n" f"Token0 数量: {amount0}\n" f"Token1 数量: {amount1}\n" f"区块: {event.get('blockNumber', 0)}\n" ) return ChainChunk( content=content, metadata={ "type": "liquidity", "protocol": pool_label, "block_number": event.get("blockNumber", 0), "amount0": amount0, "amount1": amount1, }, sources=[f"https://etherscan.io/tx/{event.get('transactionHash', '')}"], )

3.2 Embedding 与向量存储

""" vector_store.py — 向量化与 Milvus 存储 设计考量： - 使用 sentence-transformers 而非 OpenAI API：本地化 Embedding 避免 API 费用和数据泄露 - 连接池使用单例模式：Milvus 连接建立成本高，频繁创建/销毁连接会导致性能回退 - 批量插入 + flush：单条 insert 延迟高，批量操作充分利用 Milvus 的内部缓冲 """ import asyncio from sentence_transformers import SentenceTransformer from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility, ) import numpy as np class VectorStore: # Embedding 维度：sentence-transformers/all-MiniLM-L6-v2 的输出维度 EMBEDDING_DIM = 384 COLLECTION_NAME = "chain_data_chunks" def __init__(self, host: str = "localhost", port: str = "19530"): # 延迟初始化模型：sentence-transformers 模型加载需要 1-3 秒，放在 init 中 # 会让服务启动时阻塞 self._model: SentenceTransformer | None = None self._host = host self._port = port self._collection: Collection | None = None async def _get_model(self) -> SentenceTransformer: if self._model is None: # 在事件循环的线程池中加载模型，避免阻塞异步事件循环 self._model = await asyncio.to_thread( SentenceTransformer, "all-MiniLM-L6-v2" ) return self._model async def connect(self): connections.connect(host=self._host, port=self._port) if not utility.has_collection(self.COLLECTION_NAME): fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.EMBEDDING_DIM), FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=4096), FieldSchema(name="block_number", dtype=DataType.INT64), FieldSchema(name="chunk_type", dtype=DataType.VARCHAR, max_length=64), ] schema = CollectionSchema(fields, "链上数据向量索引") self._collection = Collection(self.COLLECTION_NAME, schema) # 创建 IVF_FLAT 索引：平衡召回率与插入速度 # 对于链上数据场景，每天新增 chunks 在十万级别，IVF_FLAT 比 HNSW 更合适 index_params = { "metric_type": "COSINE", "index_type": "IVF_FLAT", "params": {"nlist": 1024}, } self._collection.create_index("embedding", index_params) else: self._collection = Collection(self.COLLECTION_NAME) self._collection.load() async def insert_chunks(self, chunks: list[ChainChunk]): """批量插入向量化数据块""" if not self._collection: await self.connect() model = await self._get_model() contents = [c.content for c in chunks] # 批量 Embedding：单次推理处理所有文本，避免逐个调用的 GPU 调度开销 embeddings = await asyncio.to_thread( model.encode, contents, normalize_embeddings=True ) insert_data = [ embeddings.tolist(), contents, [c.metadata.get("block_number", 0) for c in chunks], [c.metadata.get("type", "unknown") for c in chunks], ] self._collection.insert(insert_data) self._collection.flush() async def search( self, query: str, top_k: int = 10, block_range: tuple[int, int] | None = None ) -> list[dict]: """ 混合检索：向量相似度 + 标量过滤 :param query: 用户自然语言查询 :param top_k: 返回的最相关结果数 :param block_range: 可选的区块范围过滤（时间窗口） """ if not self._collection: await self.connect() model = await self._get_model() query_embedding = await asyncio.to_thread( model.encode, [query], normalize_embeddings=True ) # 构造标量过滤表达式 filter_expr = None if block_range: filter_expr = ( f"block_number >= {block_range[0]} " f"and block_number <= {block_range[1]}" ) search_params = {"metric_type": "COSINE", "params": {"nprobe": 16}} results = self._collection.search( data=query_embedding.tolist(), anns_field="embedding", param=search_params, limit=top_k, expr=filter_expr, output_fields=["content", "block_number", "chunk_type"], ) return [ {"content": hit.entity.get("content", ""), "score": hit.score, "block_number": hit.entity.get("block_number", 0)} for hit in results[0] ]

3.3 RAG 查询引擎

""" rag_engine.py — RAG 查询与生成 设计考量： - LLM 调用包裹 tenacity 重试：Ollama 本地推理在并发下可能偶发超时或 OOM - Prompt 中注入数据来源链接：满足链上数据可验证性要求 - 流式响应用于降低首字延迟：search → generate 两个步骤串行，非流式会增加感知延迟 """ from tenacity import retry, stop_after_attempt, wait_exponential import httpx from vector_store import VectorStore RAG_PROMPT_TEMPLATE = """基于以下链上数据片段回答用户的问题。如果数据不足以回答，请明确说明"当前索引中未找到相关信息"。 ## 检索到的链上数据片段 {context} ## 用户问题 {query} ## 回答要求 - 使用中文回答 - 所有数据引用必需附带 Tx Hash 或区块高度 - 如果数据可能已过期（超过 1 小时），请添加时效性警告""" class RAGEngine: def __init__(self, vector_store: VectorStore, model: str = "qwen2.5:7b"): self._store = vector_store self._model = model self._client = httpx.AsyncClient( base_url="http://localhost:11434", timeout=httpx.Timeout(connect=10.0, read=60.0), ) async def query(self, user_query: str, top_k: int = 5) -> dict: """执行 RAG 查询：检索 → 重排序 → 生成""" # Step 1: 向量检索 candidates = await self._store.search(user_query, top_k=top_k * 2) if not candidates: return {"answer": "未在已有索引中找到相关链上数据。", "sources": []} # Step 2: 基于相似度分数的简单重排序 candidates.sort(key=lambda x: x["score"], reverse=True) top_candidates = candidates[:top_k] # Step 3: 构建上下文 context_parts = [] for i, c in enumerate(top_candidates): context_parts.append(f"[片段{i + 1}，区块 {c['block_number']}]\n{c['content']}") context = "\n\n---\n\n".join(context_parts) # Step 4: LLM 生成 prompt = RAG_PROMPT_TEMPLATE.format(context=context, query=user_query) answer = await self._generate(prompt) return { "answer": answer, "sources": [c.get("content", "")[:200] for c in top_candidates], "candidates_count": len(candidates), } @retry( stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=1, max=10), ) async def _generate(self, prompt: str) -> str: resp = await self._client.post("/api/generate", json={ "model": self._model, "prompt": prompt, "stream": False, "options": {"temperature": 0.3, "num_predict": 1024}, }) resp.raise_for_status() return resp.json().get("response", "") async def close(self): await self._client.aclose()

四、边界分析

Embedding 模型的选择决定检索质量上限。all-MiniLM-L6-v2 是通用文本模型，对链上术语和 DeFi 概念的理解不如领域特化模型。如果有资源，建议使用 BGE 系列模型在 DeFi 语料上做领域微调。模型切换需要注意维度匹配 —— 不同模型的 Embedding 维度不同，切换时需要重建向量索引。

实时性是一个结构性矛盾。链上数据是实时变化的，但向量索引的重建需要时间。从新区块产生到数据被 Embedding 并写入 Milvus，至少存在 10-30 秒的延迟。对于"查询最新区块状态"的场景，RAG 的检索结果需要搭配 RPC 的实时校验。

语义检索不能替代结构化查询。对于"所有 USDC/WETH 池的 TVL 排名"这类查询，直接写 GraphQL 或 Subgraph 查询比 RAG 更高效、更准确。RAG 适合开放域的自然语言查询，不是结构化数据查询的替代品。

成本考量。一个包含 100 万条事件日志的向量索引，使用 all-MiniLM-L6-v2（384 维 float32）大约占用 1.5 GB 存储。如果扩展到 10 亿条数据，存储成本和检索延迟都会成为瓶颈，需要考虑 PQ（乘积量化）或使用更高阶的索引结构。

五、总结