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

AI Agent 与链上自动化协作：从意图到交易的自驱引擎

news 2026/6/22 4:32:21

AI Agent 与链上自动化协作：从意图到交易的自驱引擎

一、链上交互的"最后一公里"：为什么需要 AI Agent

DeFi 用户的日常操作流程堪称折磨：连接钱包 → 切换网络 → 授权代币 → 确认滑点 → 签名交易 → 等待确认。一个简单的跨链 swap 可能需要 6 次以上交互。更别提套利、再质押、流动性管理这些复杂策略，手动操作根本跟不上市场节奏。

AI Agent 的价值在于：用户只需表达意图，Agent 负责拆解任务、规划路径、执行交易。从"告诉计算机怎么做"到"告诉计算机我想要什么"，这是交互范式的根本转变。

但链上环境比传统自动化复杂得多。Gas 价格波动、MEV 攻击、交易回滚——Agent 必须在这些不确定性中做出实时决策。这不是简单的 API 调用链，而是一个需要感知、推理、行动的自主系统。

二、AI Agent 链上协作的技术架构

2.1 意图驱动的 Agent 架构

链上 AI Agent 不是单一模型，而是一个多模块协作系统。核心设计原则是意图（Intent）驱动：用户声明目标，Agent 自主规划执行路径。

graph LR A[用户意图输入] --> B[意图解析器] B --> C[任务规划器] C --> D[链上状态感知] D --> E[执行路径生成] E --> F{模拟执行} F -->|成功| G[交易构建与签名] F -->|失败| C G --> H[MEV 保护提交] H --> I[执行结果反馈] I --> D

2.2 核心模块拆解

意图解析器：将自然语言意图转为结构化的操作描述。例如"帮我把 ETH 换成 USDC，滑点不超过 0.5%"→{action: swap, from: ETH, to: USDC, slippage: 0.5%}。

任务规划器：基于当前链上状态，生成最优执行路径。可能涉及多跳路由、跨链桥接、Gas 优化等。

链上状态感知：实时监控 Gas 价格、流动性深度、交易池状态。这是 Agent 做出正确决策的基础。

MEV 保护提交：通过 Flashbots 或私有内存池提交交易，避免被抢跑。

2.3 工具调用与链上交互

AI Agent 通过 Function Calling 机制与链上交互。每个链上操作（swap、质押、授权）被封装为一个工具函数，Agent 根据推理结果选择调用。

三、生产级 Agent 实现与最佳实践

3.1 意图解析与任务规划

import json from typing import Any from openai import AsyncOpenAI from pydantic import BaseModel class UserIntent(BaseModel): """结构化的用户意图描述""" action: str # swap, stake, bridge, claim 等 params: dict[str, Any] constraints: dict[str, Any] # 滑点、Gas 上限、时间约束 class OnChainAgent: """链上自动化 AI Agent""" def __init__(self): self.llm = AsyncOpenAI() # 工具注册表——Agent 可调用的链上操作 self.tools = self._register_tools() def _register_tools(self) -> list[dict]: """注册 Agent 可用的链上工具 为什么用工具模式而非直接调用？ 工具模式让 LLM 自主决定调用顺序和参数， 支持多步推理和条件分支""" return [ { "type": "function", "function": { "name": "get_token_price", "description": "查询代币当前价格", "parameters": { "type": "object", "properties": { "symbol": {"type": "string", "description": "代币符号"} }, "required": ["symbol"], }, }, }, { "type": "function", "function": { "name": "swap_tokens", "description": "在 DEX 上交换代币", "parameters": { "type": "object", "properties": { "from_token": {"type": "string"}, "to_token": {"type": "string"}, "amount": {"type": "string"}, "slippage_bps": {"type": "integer", "description": "滑点，单位基点"}, }, "required": ["from_token", "to_token", "amount"], }, }, }, { "type": "function", "function": { "name": "estimate_gas", "description": "估算交易 Gas 费用", "parameters": { "type": "object", "properties": { "tx_type": {"type": "string", "description": "交易类型"} }, "required": ["tx_type"], }, }, }, ] async def parse_intent(self, user_input: str) -> UserIntent: """将自然语言转为结构化意图""" response = await self.llm.chat.completions.create( model="gpt-4-turbo", messages=[ { "role": "system", "content": ( "你是链上操作意图解析器。" "将用户自然语言输入转为结构化 JSON。" "支持的 action：swap, stake, bridge, claim, approve" ), }, {"role": "user", "content": user_input}, ], response_format={"type": "json_object"}, temperature=0.0, ) data = json.loads(response.choices[0].message.content) return UserIntent(**data) async def execute_intent(self, intent: UserIntent) -> dict: """执行用户意图——多轮工具调用""" messages = [ { "role": "system", "content": ( "你是链上交易执行 Agent。" "根据用户意图，调用合适的工具完成任务。" "执行前必须先查询价格和估算 Gas。" "如果 Gas 过高，建议用户等待。" ), }, { "role": "user", "content": f"执行意图：{intent.model_dump_json()}", }, ] max_rounds = 10 # 防止无限循环 for _ in range(max_rounds): response = await self.llm.chat.completions.create( model="gpt-4-turbo", messages=messages, tools=self.tools, tool_choice="auto", temperature=0.1, ) msg = response.choices[0].message # 如果没有工具调用，说明 Agent 已完成推理 if not msg.tool_calls: return {"result": msg.content} # 执行工具调用 for tool_call in msg.tool_calls: result = await self._execute_tool( tool_call.function.name, json.loads(tool_call.function.arguments), ) # 将工具结果追加到对话历史 messages.append(msg) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result), }) return {"error": "执行轮次超限，任务可能过于复杂"} async def _execute_tool(self, name: str, args: dict) -> dict: """执行具体的链上工具调用""" if name == "get_token_price": # 实际项目中对接 Chainlink 或 DEX API return {"symbol": args["symbol"], "price_usd": "3500.00"} elif name == "swap_tokens": # 对接 1inch/Paraswap 聚合器 API return {"tx_hash": "0x...", "status": "pending"} elif name == "estimate_gas": return {"tx_type": args["tx_type"], "gas_gwei": "25", "cost_usd": "3.50"} return {"error": f"未知工具：{name}"}

3.2 MEV 保护与交易提交

import asyncio from web3 import Web3 class MEVProtectedSubmitter: """MEV 保护的交易提交器 为什么需要 MEV 保护？ 公共内存池中的交易可被 MEV 搜索者监控， 通过 Flashbots 私有提交可避免被抢跑""" def __init__(self, w3: Web3, flashbots_relay: str): self.w3 = w3 self.relay_url = flashbots_relay async def submit_via_flashbots( self, signed_tx: bytes, target_block: int ) -> str: """通过 Flashbots 提交交易""" bundle = [ {"signed_transaction": signed_tx.hex()}, ] # 模拟执行——为什么先模拟？ # 确保交易不会 revert，否则矿工不会打包 sim_result = await self._simulate_bundle(bundle, target_block) if not sim_result["success"]: raise RuntimeError( f"模拟失败：{sim_result['error']}" ) # 提交到 Flashbots relay # 目标区块号需精确，避免交易过期 return await self._send_bundle(bundle, target_block) async def _simulate_bundle(self, bundle: list, block: int) -> dict: """在 Flashbots 模拟环境中执行交易""" # 简化实现，生产环境使用 flashbots-python SDK await asyncio.sleep(0.1) return {"success": True, "gas_used": 150000} async def _send_bundle(self, bundle: list, block: int) -> str: """发送交易包到 Flashbots relay""" await asyncio.sleep(0.1) return "0xbundle_hash..."

3.3 错误处理与重试策略

from tenacity import retry, stop_after_attempt, wait_exponential from web3.exceptions import TransactionNotFound class AgentErrorHandler: """Agent 执行过程中的错误处理与恢复""" @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=15), ) async def wait_for_confirmation( self, w3: Web3, tx_hash: str, timeout: int = 120 ) -> dict: """等待交易确认，带超时和重试 为什么需要重试？RPC 节点偶尔超时， 重试比直接报错更健壮""" receipt = await asyncio.wait_for( self._poll_receipt(w3, tx_hash), timeout=timeout, ) if receipt["status"] == 0: raise RuntimeError(f"交易回滚：{tx_hash}") return receipt async def _poll_receipt(self, w3: Web3, tx_hash: str) -> dict: """轮询交易收据""" while True: receipt = w3.eth.get_transaction_receipt(tx_hash) if receipt is not None: return receipt await asyncio.sleep(2)