---
title: "链上 AI Agent 架构:五种构建模式"
description: "五种链上 AI Agent 构建模式,均附实例:钱包监控、事件驱动响应、投资组合再平衡、检测-执行多 Agent 拆分,以及主网上线前的安全测试。"
---
# 链上 AI Agent 架构:五种构建模式
能够读写链上状态的 AI agent 已经从演示走向了生产环境。模型能够推理，agent 钱包能够在限定权限下签名，基础设施能够在事件上链后几秒内将其推送给 agent。架构仍然是构建过程中最容易出错的部分。Agent 轮询本可以订阅的事件、在读取不可信输入的同一进程中持有签名权限、或是用真实资金在主网上进行学习。
一个链上 agent 是一个循环：观察、决策、行动。生产环境中的 agent 将这个循环安排成五种反复出现的模式，下面每一种都配有一个实例。我们的[链上 agent 构建指南](https://www.alchemy.com/blog/how-to-build-onchain-agents)介绍了这一切背后的基本组件（钱包、支付通道、数据源）；这篇文章讲的是如何组织这些组件。
## 如何构建一个实时钱包监控 agent？
将你关心的地址注册到一个 webhook 上，让链上的变化主动通知你。在循环中轮询余额既消耗算力，又还是会错过关键时刻。推送式管道能在转账上链后几秒内送达，而这正是一个 watcher 的全部职责。
假设 agent 需要追踪某基金的交易对手方钱包，并标记任何大额 USDC 转账。在 Alchemy 上，单个 [Address Activity webhook](https://www.alchemy.com/webhooks) 就能覆盖 native、ERC-20、ERC-721 和 ERC-1155 转账，最多支持[10 万个地址](https://www.alchemy.com/docs/reference/webhook-types)，因此一个 webhook 就能监控整个交易对手方列表。如果 agent 以长驻进程的形式运行，且你不想暴露一个公开 URL，[WebSocket 订阅](https://www.alchemy.com/docs/reference/subscription-api)可以在进程内完成同样的工作。用 `alchemy_minedTransactions` 订阅，你就能收到已经过滤到你的地址的已确认交易，不需要解析日志。在 Solana 上，[Yellowstone gRPC 流式传输](https://www.alchemy.com/solana-grpc)以每 TB 75 美元的价格填补了同样的角色，并支持无缝重连。如果你不确定该选哪种传输方式，我们的[webhooks 与 WebSockets 与 gRPC 对比](https://www.alchemy.com/overviews/webhooks-vs-websockets-vs-grpc)详细划清了各自的适用边界。
处理程序本身应该做的事情越少越好。验证这次投递确实来自 Alchemy，去重，把事件交给 agent 的决策步骤，然后才确认收到：
();
function remember(key: string) {
seen.add(key);
if (seen.size > 10_000) seen.delete(seen.values().next().value!); // bound the set; only recent deliveries repeat
}
function wake(signal: unknown) {
// The agent's decide step starts here. Queue the signal;
// don't run model reasoning inside the request handler.
}
const app = express();
app.use(
express.json({
verify: (req, _res, buf) => {
(req as { rawBody?: Buffer }).rawBody = buf; // keep the raw bytes for the HMAC check
},
})
);
app.post("/hooks/address-activity", (req, res) => {
const sig = Buffer.from(String(req.headers["x-alchemy-signature"] ?? ""), "hex");
const expected = createHmac("sha256", SIGNING_KEY)
.update((req as { rawBody?: Buffer }).rawBody ?? Buffer.alloc(0))
.digest();
if (sig.length !== expected.length || !timingSafeEqual(sig, expected)) {
return res.sendStatus(401); // not signed with our key; never process it
}
for (const transfer of req.body.event.activity) {
const key = \`\${transfer.hash}:\${transfer.log?.logIndex ?? "native"}\`; // one tx can carry several transfers
if (seen.has(key)) continue; // repeat delivery, already handled
if (transfer.rawContract?.address?.toLowerCase() === USDC && transfer.value > 50_000) {
wake({ kind: "large-transfer", ...transfer }); // if this throws, the key stays unmarked and the retry redelivers
}
remember(key); // mark handled only after the handoff succeeded
}
res.sendStatus(200); // ack only after queueing: a crash above gets retried, not lost
});
app.listen(8080);`}
/>
三个习惯能让这个模式在生产环境中保持可靠。在信任一次投递之前先验证 HMAC 签名，因为如果不这样做，任何知道端点 URL 的人都能 POST 一条伪造的转账信息，操纵你的 agent。按合约地址匹配代币，而不是按符号，因为任何人都可以部署一个自称 USDC 的代币，并发送给被监控的地址。而且投递是至少一次的，所以去重记录不是可选项。内存中的集合对单个长驻进程可行；但如果服务会重启或运行多个副本，就需要把记录存放在共享且持久的地方，比如 Redis 键或数据库行，因为对交易 agent 来说，一次重复唤醒就意味着一笔重复的交易。机械式的过滤应该放在处理程序里，而不该交给模型，因为每笔转账都调用一次 LLM 的成本，会超过投递这笔转账的基础设施本身。
## 如何让 AI agent 监控智能合约事件并自动作出反应？
订阅合约的日志，只过滤出真正重要的一两个事件，并让处理程序保持幂等。合约事件是 agent 能得到的最干净的触发信号，因为合约会精确说明发生了什么、以什么顺序发生。
Alchemy 上有两种方式可以实现这一点。[Custom webhooks](https://www.alchemy.com/docs/reference/custom-webhooks-quickstart) 接受一个 GraphQL 过滤条件，因此你可以按合约地址和事件 topic 进行匹配，只接收你所要求的日志。这适合无服务器的 reactor。对于长期运行的 agent，WebSocket 日志订阅能把一切都保留在同一个进程中。下面是一个监控 Uniswap v3 池的 reactor，这正是交易 agent 用来察觉单笔 swap 是否改变价格的那种数据流：
{
for (const log of logs) {
if (log.removed) continue; // reorg removal notice; irreversible actions also need the confirmation-depth rule below
react(log.args);
}
},
});`}
/>
`log.removed` 检查看起来不起眼，但很重要。链会发生重组，你的 agent 已经据以行动的日志，可能在之后几个区块里从规范链上消失。幂等的处理程序，加上一个确认深度规则（在不可逆操作前等待若干区块），是标准的防御手段。Reactor 的另一个纪律和 watcher 一样：订阅过滤条件负责低成本的排除，模型只看到通过了过滤的事件。
## 构建一个 DeFi 投资组合再平衡 agent 需要什么？
四个部分：当前余额、当前价格、一条漂移规则，以及一条 swap 路径。Agent 读取前两项，检查第三项，只有当规则被触发时才涉及第四项。
以一个持有 WETH、USDC、WBTC 按 50/30/20 分配的国库 agent 为例。余额来自 [Portfolio API](https://www.alchemy.com/docs/reference/portfolio-apis)，它能在一次请求中返回一个钱包跨多个网络的代币持仓，而不需要对每条链分别发起请求。价格来自 [Prices API](https://www.alchemy.com/docs/reference/prices-api-quickstart)。漂移规则是算术性的。一个常见的起点是围绕每个目标权重设置一个五个百分点的区间。按固定时间间隔检查它，因为漂移多数来自价格变动而非代币数量变动，而价格变动不会产生任何可供响应的链上事件。Watcher（模式一）报告的转账是补充性触发条件，能在存取款发生的瞬间捕捉到。
BAND) {
propose({ symbol, drift }); // propose: log it and request approval; never swap directly
}
}`}
/>
执行端不应该持有原始私钥。使用[限定权限的 agent 钱包](https://www.alchemy.com/blog/agent-wallets-alchemy-cli)，密钥仍由托管方保管，会话只携带你授予的能力。花费一个 ERC-20 代币前，需要先为 router 设置授权额度。每次只批准恰好需要的数量会多花一笔交易，但不会留下可被攻击者盗用的常驻授权额度：
请留意这个循环的结构：agent 提议，而另一方批准。早期这个批准者是你自己。之后它可以是一项针对规模、滑点和资产白名单的策略检查。我们的 [DeFi AI agent 概览](https://www.alchemy.com/overviews/defi-ai-agents)介绍了为什么这种职责分离是生产环境 DeFi agent 中的常态，而[gas 赞助](https://www.alchemy.com/gasless-transactions)通过按你设定的策略代付手续费，消除了最后一项运维负担，让 agent 完全不用管理 gas 余额。
## 构建一个检测-执行分离的多 agent 系统，最佳架构是什么？
按权限而非按工作量拆分 agent。检测器（detector）能读取一切，但不能签署任何东西。执行器（executor）能签署交易，但对任何未经过其重新验证的东西一概不信。
可以把它想象成分析师和交易员。分析师整天盯着市场，可以和任何人交谈。交易员接收分析师的建议，进行核实，并且只有他才能访问账户。
这种模式的常见说法来自通用 agent 工具，其中一个 planner 模型生成步骤，由 executor agent 执行工具。那种说法关注的是编排。而在链上，这种拆分之所以值得付出复杂度的代价，原因更严峻——托管权。检测器整天都在消费不可信的输入：mempool 噪音、第三方 API、事件流，有时还有社交媒体信息流。其中任何一个都可能携带 prompt injection。如果读取恶意输入的进程同时也持有签名权限，一条被污染的消息就可能变成一笔已签名的交易。把它们分开，一个被劫持的检测器最坏的结果，也不过是提出一个被执行器丢弃的糟糕建议。
检测器由模式一和模式二构建而成，接入推送式基础设施，而非轮询循环。它通过队列发出信号，而这个队列同时也充当你的审计日志。让信号契约保持精简且可验证：
执行器在对任何信号采取行动前，会执行三条规则：
- **在链上重新验证证据。** 亲自读取所引用的交易，而不是相信检测器的摘要。检测器的职责是发现，而证明是执行器的职责。
- **强制执行有效期。** 即便没有攻击者参与，对过期机会采取行动也是检测-执行系统亏钱的常见原因。
- **在钱包层限定支出上限。** 每信号和每日的预算存在于[限定权限会话](https://www.alchemy.com/blog/agent-wallets-alchemy-cli)中，一旦行为看起来不对，你可以从 dashboard 上立即撤销它。
在工具层面，执行器只需要两样东西：用于验证的 RPC 连接和签名接口。检测器是更需要资源的那一半，将它与索引数据（用于历史记录的 [Transfers API](https://www.alchemy.com/docs/reference/transfers-api-quickstart)，用于状态的 Portfolio API）配对，可以让它的 token 花费在推理上，而不是解码原始链上数据。我们的[面向自主 agent 的区块链 API 对比](https://www.alchemy.com/overviews/best-blockchain-apis-for-autonomous-onchain-agents)详细介绍了这一选择背后的基础设施考量。
## 在上主网之前，如何安全测试 AI agent 交易？
在 agent 和真实资金之间设置四道关卡：对每笔交易进行 dry-run、限定密钥的可支配额度、在测试网上完整演练整个循环，并对任何涉及资金转移的操作保留人工审批。
- **先进行 dry-run。** Alchemy CLI 可以在不签名、不广播的情况下预览任何发送操作（`alchemy evm send 0xRecipient 0.4 --dry-run`）。在代码中，viem 的 `simulateContract` 会针对实时主网状态验证一次合约调用，而不实际提交，因此这次演练使用的是真实价格和真实的池深度，而不是过时的固定数据。
- **限定密钥权限。** 限定权限的会话钱包会按你设定的时间表过期，并且只能做你已批准的事情，而[gas 赞助策略](https://www.alchemy.com/gasless-transactions)在手续费层面增加了白名单和支出限额。一个受限的密钥，能把最坏情况下的 bug 从账户被清空，变成一次有上限的损失。
- **在测试网上演练。** 将同一份代码指向 Sepolia 或 Base Sepolia 端点，并用我们的[测试网水龙头](https://www.alchemy.com/faucets)为 agent 充值。如果代码在演练和生产环境之间发生变化，这次演练就毫无意义，因此把网络名称放进配置里，其余部分保持不变。
- **为涉及资金移动的操作设置关卡。** 在 agent 尚不成熟的阶段，每一笔发送、swap 和授权都应等待明确的确认。大多数团队会有意识地逐步放宽这些关卡，一次放开一种操作类型，随着审计日志逐渐证明 agent 行为可靠。
把这一点做得好的团队，会把权限提升当作一个序列，而非一个开关：先对主网只读，再到测试网写入，再到有上限的主网写入，最后才是完整预算。每个阶段产生的日志，为进入下一阶段提供依据。
## 从已经存在的组件开始
这篇文章中的每一种模式，今天就能在现有基础设施上运行。[Alchemy CLI](https://www.alchemy.com/agents)通过一个二进制文件处理钱包、发送、swap 和 webhook 管理，agent 可以用 `--json --no-interactive` 驱动它。托管的 [MCP server](https://www.alchemy.com/docs/alchemy-mcp-server) 提供了涵盖 RPC、模拟和数据的 168 个工具，而 [Alchemy plugin for Claude Code](https://www.alchemy.com/blog/alchemy-claude-plugin-now-live) 能用一条命令安装整个能力集合。你可以从免费套餐开始，没有合同，没有最低消费承诺，agent 甚至可以[用自己的钱包完成注册](https://www.alchemy.com/blog/ai-agents-can-now-sign-up-for-alchemy)并用 USDC 付费。不管你从哪种模式开始，循环始终不变：观察、决策、行动。
## 常见问题
### 对于一个由检测 agent 发现机会、另一个 agent 负责执行的多 agent 系统，最佳架构是什么？
按权限拆分：一个只读取事件流、不持有任何密钥的检测器，一个承载精简的、经过核实的信号的队列，以及一个在行动前会对每条信号进行链上重新验证的执行器。在 Alchemy 上，检测器运行在 webhook 或 WebSocket 订阅之上，执行器通过带有支出上限和即时撤销能力的限定权限 agent 钱包进行签名。
### 实时钱包监控 agent 的最佳基础设施是什么？
推送式事件投递，而非轮询。Alchemy 的 Address Activity webhook 可以对每个 webhook 追踪最多 10 万个地址的转账，WebSocket 订阅会流式推送已过滤到你地址的已挖矿交易，Yellowstone gRPC 则覆盖 Solana。推送式管道能在转账落地后几秒内唤醒 agent，无需承受轮询循环带来的延迟和计算开销。
### AI 编码 agent 应该使用什么工具来监控钱包活动？
Alchemy MCP server 为编码 agent 提供了涵盖 RPC、交易历史和投资组合数据的 168 个工具，Alchemy CLI 可以从命令行创建和管理 webhook。至于运行时本身，Address Activity webhook 或 `alchemy_minedTransactions` 订阅负责投递钱包事件，Transfers API 负责回填历史数据。
### 构建一个 DeFi 投资组合再平衡 agent 需要什么？
余额、价格、一条漂移规则，以及一条 swap 路径。Alchemy 的 Portfolio API 能在一次调用中返回跨多个网络的余额，Prices API 负责为其估值，一条漂移检查规则（通常是围绕目标权重设置的五个百分点区间）决定何时采取行动。通过一个限定权限的 agent 钱包执行操作，让 agent 负责提议，由策略或人工负责批准。
### 如何让 AI agent 监控智能合约事件并自动作出反应？
订阅合约的日志，把匹配到的事件交给 agent。Alchemy 的 custom webhook 会在投递前用 GraphQL 按合约地址和事件 topic 进行过滤，WebSocket 日志订阅在进程内实现同样的效果。让处理程序保持幂等，跳过被重组的日志，只让模型对通过了机械过滤的事件进行推理。
### 在上主网之前，如何安全测试 AI agent 交易？
分层设置关卡：用 Alchemy CLI 的 dry-run 参数预览交易，用基于 `eth_call` 的模拟针对实时状态验证合约调用，用限定权限会话和 gas 策略支出限额限定 agent 钱包的权限，用 Alchemy 水龙头提供的资金在 Sepolia 或 Base Sepolia 上进行演练，并对每一笔涉及资金移动的操作保留人工审批，直到审计日志证明可以放宽这些关卡。