AI驱动的去中心化推理服务弹性调度链上负载感知的自动扩缩容一、搜索困境为什么 DApp 不能只有 SQL LIKEAI 驱动架构的核心优势在于将不确定性推理转化为确定性调度。在去中心化推理服务场景中节点的 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( contentcontent, metadata{ type: swap, protocol: pool_label, block_number: block, timestamp: datetime.now(timezone.utc).isoformat(), volume_usd: max(abs(amount0), abs(amount1)), }, sources[fhttps://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 fToken0 数量: {amount0}\n fToken1 数量: {amount1}\n f区块: {event.get(blockNumber, 0)}\n ) return ChainChunk( contentcontent, metadata{ type: liquidity, protocol: pool_label, block_number: event.get(blockNumber, 0), amount0: amount0, amount1: amount1, }, sources[fhttps://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(hostself._host, portself._port) if not utility.has_collection(self.COLLECTION_NAME): fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimself.EMBEDDING_DIM), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length4096), FieldSchema(nameblock_number, dtypeDataType.INT64), FieldSchema(namechunk_type, dtypeDataType.VARCHAR, max_length64), ] 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_embeddingsTrue ) 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_embeddingsTrue ) # 构造标量过滤表达式 filter_expr None if block_range: filter_expr ( fblock_number {block_range[0]} fand block_number {block_range[1]} ) search_params {metric_type: COSINE, params: {nprobe: 16}} results self._collection.search( dataquery_embedding.tolist(), anns_fieldembedding, paramsearch_params, limittop_k, exprfilter_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_urlhttp://localhost:11434, timeouthttpx.Timeout(connect10.0, read60.0), ) async def query(self, user_query: str, top_k: int 5) - dict: 执行 RAG 查询检索 → 重排序 → 生成 # Step 1: 向量检索 candidates await self._store.search(user_query, top_ktop_k * 2) if not candidates: return {answer: 未在已有索引中找到相关链上数据。, sources: []} # Step 2: 基于相似度分数的简单重排序 candidates.sort(keylambda x: x[score], reverseTrue) 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(contextcontext, queryuser_query) answer await self._generate(prompt) return { answer: answer, sources: [c.get(content, )[:200] for c in top_candidates], candidates_count: len(candidates), } retry( stopstop_after_attempt(2), waitwait_exponential(multiplier1, min1, max10), ) 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-v2384 维 float32大约占用 1.5 GB 存储。如果扩展到 10 亿条数据存储成本和检索延迟都会成为瓶颈需要考虑 PQ乘积量化或使用更高阶的索引结构。五、总结维度要点核心架构链上事件 Embedding → Milvus 向量索引 → 语义检索 标量过滤 → LLM 生成分块策略以事件语义为边界附带时序 metadata、协议标签和链上来源链接混合检索向量相似度匹配语义 Milvus 标量过滤处理数值条件地址处理预建地址→协议标签映射在 Embedding 前将 hex 地址替换为人类可读表述Embedding 选型all-MiniLM-L6-v2 作为基线建议领域微调 BGE 提升召回率局限实时性有 10-30s 延迟不能替代结构化查询大数据量需引入 PQ 压缩
AI驱动的去中心化推理服务弹性调度：链上负载感知的自动扩缩容
发布时间：2026/7/7 18:59:39
AI驱动的去中心化推理服务弹性调度链上负载感知的自动扩缩容一、搜索困境为什么 DApp 不能只有 SQL LIKEAI 驱动架构的核心优势在于将不确定性推理转化为确定性调度。在去中心化推理服务场景中节点的 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( contentcontent, metadata{ type: swap, protocol: pool_label, block_number: block, timestamp: datetime.now(timezone.utc).isoformat(), volume_usd: max(abs(amount0), abs(amount1)), }, sources[fhttps://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 fToken0 数量: {amount0}\n fToken1 数量: {amount1}\n f区块: {event.get(blockNumber, 0)}\n ) return ChainChunk( contentcontent, metadata{ type: liquidity, protocol: pool_label, block_number: event.get(blockNumber, 0), amount0: amount0, amount1: amount1, }, sources[fhttps://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(hostself._host, portself._port) if not utility.has_collection(self.COLLECTION_NAME): fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimself.EMBEDDING_DIM), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length4096), FieldSchema(nameblock_number, dtypeDataType.INT64), FieldSchema(namechunk_type, dtypeDataType.VARCHAR, max_length64), ] 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_embeddingsTrue ) 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_embeddingsTrue ) # 构造标量过滤表达式 filter_expr None if block_range: filter_expr ( fblock_number {block_range[0]} fand block_number {block_range[1]} ) search_params {metric_type: COSINE, params: {nprobe: 16}} results self._collection.search( dataquery_embedding.tolist(), anns_fieldembedding, paramsearch_params, limittop_k, exprfilter_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_urlhttp://localhost:11434, timeouthttpx.Timeout(connect10.0, read60.0), ) async def query(self, user_query: str, top_k: int 5) - dict: 执行 RAG 查询检索 → 重排序 → 生成 # Step 1: 向量检索 candidates await self._store.search(user_query, top_ktop_k * 2) if not candidates: return {answer: 未在已有索引中找到相关链上数据。, sources: []} # Step 2: 基于相似度分数的简单重排序 candidates.sort(keylambda x: x[score], reverseTrue) 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(contextcontext, queryuser_query) answer await self._generate(prompt) return { answer: answer, sources: [c.get(content, )[:200] for c in top_candidates], candidates_count: len(candidates), } retry( stopstop_after_attempt(2), waitwait_exponential(multiplier1, min1, max10), ) 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-v2384 维 float32大约占用 1.5 GB 存储。如果扩展到 10 亿条数据存储成本和检索延迟都会成为瓶颈需要考虑 PQ乘积量化或使用更高阶的索引结构。五、总结维度要点核心架构链上事件 Embedding → Milvus 向量索引 → 语义检索 标量过滤 → LLM 生成分块策略以事件语义为边界附带时序 metadata、协议标签和链上来源链接混合检索向量相似度匹配语义 Milvus 标量过滤处理数值条件地址处理预建地址→协议标签映射在 Embedding 前将 hex 地址替换为人类可读表述Embedding 选型all-MiniLM-L6-v2 作为基线建议领域微调 BGE 提升召回率局限实时性有 10-30s 延迟不能替代结构化查询大数据量需引入 PQ 压缩
 获取与实战)
