企业官网建设流程全解析

1. 为什么去中心化交易需要 AI Agent 介入

去中心化交易平台从最早的订单簿模式一路演化到自动做市商模式，解决了撮合效率问题，但新的瓶颈很快暴露出来。流动性碎片化、无常损失、滑点控制、MEV 提取，这些问题本质上都是信息处理与决策速度的问题。人类交易者面对链上每秒数十笔的状态变化，反应速度远远不够。AI Agent 的出现，恰好补上了这块短板。

所谓 AI Agent，不是简单的脚本机器人，而是一个具备感知、推理、决策、执行闭环的智能体。它能够读取链上数据、调用大语言模型进行语义理解、根据预设策略或学习到的经验做出判断，然后自动发起交易。和传统的交易机器人相比，AI Agent 的核心差异在于自主性和适应性——它不需要人为设定每一条 if-else 规则，而是能在动态环境中调整自己的行为。

这篇文章适合三类人看：一是正在做 DEX 产品、想引入智能化能力的开发者；二是对 AI Agent 开发感兴趣、想找一个真实落地场景练手的技术人；三是想理解这个赛道到底在发生什么、值不值得投入时间的从业者。我会从架构设计讲到代码实操，从策略逻辑讲到踩坑经验，尽量把每个环节的“为什么”说清楚。

2. AI Agent 与 DEX 结合的核心架构拆解

2.1 Agent 的四个核心模块

一个能在 DEX 场景下跑起来的 AI Agent，拆开来看至少包含四个模块：

• 感知层：负责采集链上数据，包括池子状态、价格变动、Gas 费、待处理交易等。数据源可以是节点 RPC、链上索引服务或者预言机。
• 推理层：这是 AI Agent 区别于普通机器人的关键。它通常由大语言模型驱动，负责理解当前市场状态、评估风险、生成决策建议。比如“当前 ETH/USDC 池子的流动性正在快速撤出，是否应该调整做市区间”。
• 执行层：把推理结果转化为具体的链上操作，包括构造交易、估算 Gas、签名发送、监控回执。
• 记忆层：存储历史决策和结果，供后续推理参考。这一层决定了 Agent 能否从经验中学习。

注意：很多团队一上来就把推理层做得极其复杂，却忽略了感知层的数据质量。实际上，垃圾数据喂给再强的模型也出不来好决策。我建议先把数据采集做扎实，再考虑模型的事。

2.2 为什么选择链下推理 + 链上执行

目前主流方案是链下跑 Agent 逻辑，链上只做最终执行。原因很直接：大语言模型的推理成本高、延迟大，不可能放在链上跑。链上只负责验证和执行，保证最终结果的可信度。

这种架构下，Agent 的决策过程是链下完成的，但执行结果是链上可验证的。对于需要更高可信度的场景，可以把决策摘要和关键参数上链存证，方便事后审计。

2.3 Hybrid DEX 模式下的 Agent 定位

Hybrid DEX 结合了订单簿和 AMM 的特点，既有链下撮合的高效，又有链上结算的透明。在这种模式下，AI Agent 可以扮演几个角色：

每个角色的 Agent 可以独立运行，也可以通过多智能体框架协同。比如路由 Agent 发现一笔大额交易，可以通知做市 Agent 调整报价，同时让风控 Agent 评估敞口变化。

3. 从零搭建一个 DEX 交易 Agent 的实操过程

3.1 环境准备与依赖安装

先明确技术栈。我用的方案是 Python 作为主语言，web3.py 做链交互，LangChain 做 Agent 编排，OpenAI 兼容接口做推理。这套组合的好处是生态成熟、文档多、上手快。

pip install web3 langchain langchain-openai python-dotenv

环境变量文件里放好 RPC 地址和模型 API Key：

# .env RPC_URL=https://your-rpc-endpoint PRIVATE_KEY=your-private-key OPENAI_API_KEY=your-api-key

提示：私钥千万不要硬编码在代码里，也不要用主钱包私钥。建议专门生成一个交易钱包，只放必要的资金。

3.2 感知层：链上数据采集

感知层的核心是稳定、低延迟地拿到链上状态。我封装了一个简单的数据采集类：

from web3 import Web3 import json class ChainSensor: def __init__(self, rpc_url, pool_address, abi_path): self.w3 = Web3(Web3.HTTPProvider(rpc_url)) with open(abi_path) as f: abi = json.load(f) self.pool = self.w3.eth.contract( address=Web3.to_checksum_address(pool_address), abi=abi ) def get_pool_state(self): slot0 = self.pool.functions.slot0().call() liquidity = self.pool.functions.liquidity().call() return { "sqrtPriceX96": slot0[0], "tick": slot0[1], "liquidity": liquidity } def get_gas_price(self): return self.w3.eth.gas_price

这里拿的是集中流动性池子的核心状态。sqrtPriceX96是价格的平方根编码，tick是价格区间的离散化表示。这两个值决定了当前价格和流动性分布。

实际跑的时候，我建议加一个轮询间隔控制，不要每个区块都查，根据策略频率来定。做市策略一般 3-5 秒查一次就够了，套利策略可能需要更快。

3.3 推理层：让 Agent 学会“思考”

推理层是整个 Agent 的大脑。我的做法是把链上状态转成自然语言描述，喂给大语言模型，让它输出结构化的决策。

from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate class TradingBrain: def __init__(self, api_key): self.llm = ChatOpenAI( model="gpt-4o-mini", api_key=api_key, temperature=0.1 ) self.prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个去中心化交易策略助手。 根据当前池子状态，输出是否调整仓位的建议。 输出格式必须是 JSON： {{"action": "hold|adjust|exit", "reason": "...", "confidence": 0-1}}"""), ("human", """当前状态： 价格 tick: {tick} 流动性: {liquidity} Gas 价格: {gas_price} 最近 10 笔交易方向: {recent_trades}""") ]) def decide(self, state): chain = self.prompt | self.llm result = chain.invoke(state) return result.content

这里有几个关键设计点。第一，temperature设成 0.1，因为交易决策需要稳定性，不能太随机。第二，要求输出 JSON 格式，方便后续程序解析。第三，把最近交易方向也作为输入，让模型能感知市场情绪。

实操心得：一开始我用的是纯规则引擎，后来换成 LLM 驱动，发现最大的提升不是准确率，而是处理边界情况的能力。规则引擎遇到没预设过的情况就卡住了，LLM 能给出一个合理的兜底决策。

3.4 执行层：把决策变成链上交易

执行层要处理的事情比想象中多：构造 calldata、估算 Gas、设置滑点保护、发送交易、监控回执、处理失败重试。

class TradeExecutor: def __init__(self, w3, private_key, router_address, router_abi): self.w3 = w3 self.account = w3.eth.account.from_key(private_key) self.router = w3.eth.contract( address=Web3.to_checksum_address(router_address), abi=router_abi ) def swap(self, token_in, token_out, amount_in, min_amount_out): deadline = self.w3.eth.get_block('latest')['timestamp'] + 300 tx = self.router.functions.swapExactTokensForTokens( amount_in, min_amount_out, [token_in, token_out], self.account.address, deadline ).build_transaction({ 'from': self.account.address, 'nonce': self.w3.eth.get_transaction_count(self.account.address), 'gas': 300000, 'gasPrice': self.w3.eth.gas_price }) signed = self.w3.eth.account.sign_transaction(tx, self.account.key) tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction) return tx_hash.hex()

min_amount_out是滑点保护的关键参数。设太松容易被夹，设太紧交易容易失败。我的经验是稳定币对设 0.1%，波动性大的对设 0.5%-1%，具体看池子深度。

3.5 记忆层：让 Agent 从历史中学习

记忆层不需要多复杂，一个本地 SQLite 就够了。记录每次决策的输入状态、输出动作、执行结果。

import sqlite3 class AgentMemory: def __init__(self, db_path="agent_memory.db"): self.conn = sqlite3.connect(db_path) self.conn.execute(""" CREATE TABLE IF NOT EXISTS decisions ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp INTEGER, state TEXT, action TEXT, result TEXT, pnl REAL ) """) def record(self, state, action, result, pnl): self.conn.execute( "INSERT INTO decisions (timestamp, state, action, result, pnl) VALUES (?, ?, ?, ?, ?)", (int(time.time()), json.dumps(state), action, result, pnl) ) self.conn.commit() def recent(self, n=10): cursor = self.conn.execute( "SELECT state, action, pnl FROM decisions ORDER BY id DESC LIMIT ?", (n,) ) return cursor.fetchall()

这些历史记录可以在下一轮推理时作为上下文喂给模型，让它知道自己之前的决策效果如何。这就是所谓的“经验注入”。

4. 策略设计与参数调优的实战细节

4.1 做市策略的区间设定逻辑

集中流动性做市的核心是选择价格区间。区间太窄，资金效率高但容易出区间；区间太宽，资金效率低但省心。我的做法是让 Agent 根据历史波动率动态计算。

具体来说，取过去 24 小时的价格数据，计算标准差，然后以当前价格为中心，设置 ±1.5 倍标准差的区间。这个系数是可以调的，波动大的时候放宽，波动小的时候收窄。

import numpy as np def calc_range(prices, current_price, k=1.5): std = np.std(prices) lower = current_price * (1 - k * std / current_price) upper = current_price * (1 + k * std / current_price) return lower, upper

注意：这个公式是简化版，实际用的时候要考虑 tick 间距对齐。不同费率档位的池子 tick 间距不一样，0.05% 费率是 10，0.3% 费率是 60。区间边界必须对齐到 tick 间距上，否则交易会失败。

4.2 Gas 成本与收益的平衡

Agent 每次调整仓位都要花 Gas。如果调整带来的预期收益小于 Gas 成本，这笔操作就是亏的。所以推理层必须把 Gas 成本纳入决策。

我的做法是在 prompt 里明确告诉模型当前 Gas 价格，并要求它在 confidence 低于某个阈值时输出 hold。同时代码层面加一道硬性检查：

def should_execute(expected_gain, gas_cost, threshold=1.5): return expected_gain > gas_cost * threshold

threshold设 1.5 的意思是，预期收益至少要覆盖 1.5 倍的 Gas 成本才值得动手。这个系数根据策略频率调整，高频策略可以设低一点，低频策略设高一点。

4.3 多 Agent 协同的编排方式

当你有多个 Agent 同时运行时，需要一个协调机制。我用的是简单的消息队列 + 优先级规则。

风控 Agent 的决策可以覆盖其他 Agent 的动作。比如做市 Agent 想加仓，但风控 Agent 判断当前敞口过大，就会否决这笔操作。这种优先级机制能有效防止 Agent 在极端行情下做出危险决策。

5. 常见问题与排查技巧实录

5.1 交易失败的五种典型原因

在实际运行中，交易失败是家常便饭。我整理了最常见的五种情况：

实操心得：我一开始没做失败重试，结果一笔交易失败后 Agent 就卡住了。后来加了指数退避重试机制，失败后等 2 秒、4 秒、8 秒再试，最多重试 3 次。这个改动让 Agent 的稳定性提升了一个档次。

5.2 模型输出不稳定的处理

LLM 的输出有时候会不符合预期格式，比如该输出 JSON 的时候输出了一段解释文字。我的处理方式是加一层解析容错：

import re def parse_decision(text): try: return json.loads(text) except json.JSONDecodeError: match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group()) except: pass return {"action": "hold", "reason": "parse_failed", "confidence": 0}

解析失败时默认返回 hold，这是最安全的兜底动作。宁可不动，也不要乱动。

5.3 防止 Agent 陷入死循环

有一种情况很隐蔽：Agent 发现价格偏离区间，决定调整；调整后价格又偏离，又调整；来回几次 Gas 费吃掉大量利润。这就是死循环。

我的解决方案是加一个冷却时间。每次调整仓位后，至少等 5 分钟才能再次调整同一池子。同时记录最近 1 小时的调整次数，超过 10 次就强制暂停，等待人工检查。

class CooldownManager: def __init__(self, cooldown_seconds=300, max_actions_per_hour=10): self.cooldown = cooldown_seconds self.max_actions = max_actions_per_hour self.last_action = {} self.action_count = {} def can_act(self, pool_id): now = time.time() if pool_id in self.last_action: if now - self.last_action[pool_id] < self.cooldown: return False hour_key = int(now // 3600) count = self.action_count.get((pool_id, hour_key), 0) if count >= self.max_actions: return False return True

5.4 数据延迟导致的决策偏差

链上数据从产生到被 Agent 读取，中间有延迟。如果 Agent 基于过期数据做决策，结果往往不理想。我踩过的一个坑是：Agent 看到池子价格是 2000，实际已经变成 2010，结果按 2000 的价格挂单，一挂进去就被套利。

解决办法有两个：一是尽量用低延迟的数据源，比如直接连节点而不是走第三方 API；二是在决策时加入价格变化率的判断，如果最近几秒价格波动剧烈，就降低 confidence 或者直接 hold。

6. 这套方案后续还能怎么扩展

跑通基础版本之后，我陆续加了一些扩展能力，效果不错，分享几个方向。

第一个是多链部署。把感知层和执行层做成可配置的，通过配置文件切换不同的链和池子。推理层和记忆层可以复用。这样一套 Agent 逻辑能同时服务多条链。

第二个是回测框架。在正式跑之前，用历史数据回测策略效果。我搭了一个简单的回测器，把历史价格和流动性数据喂给推理层，记录每次决策的模拟收益。这个框架帮我省了不少真金白银的学费。

第三个是多模态输入。除了链上数据，还可以把社交媒体情绪、新闻事件作为输入。比如某个代币突然上了热门讨论，Agent 可以提前感知到波动加剧，主动收窄做市区间。

第四个是Agent 之间的协作协议。当多个 Agent 分属不同团队时，需要一个标准化的通信协议来交换信息和协调动作。这块目前还在早期，但我觉得是接下来一两年会快速发展的方向。

最后分享一个我在调试过程中总结的小技巧：把 Agent 的每次决策都打上详细日志，包括输入状态、模型原始输出、解析后的动作、执行结果。出问题的时候，翻日志比猜原因快得多。我现在的日志格式是这样的：

logger.info(f"[DECISION] pool={pool_id} tick={tick} action={action} " f"confidence={conf} gas={gas} result={result} pnl={pnl}")

一行日志包含所有关键信息，grep 一下就能定位问题。这个习惯让我排查效率至少提升了一倍。