Rust重写Nostr中继buzz：内嵌AI Agent的架构设计与实践

1. 从 buzz 这个名字说起一个用 Rust 重写的 Nostr 中继到底想解决什么第一次看到 buzz 这个项目名我脑子里蹦出来的其实是消息队列里那个 buzz或者是某个即时通讯的提示音。但把关键词 Nostr、relay、Rust、AI agents 摆在一起之后方向就清楚了这是一个用 Rust 实现的 Nostr 中继relay服务而且它把 AI agents 作为一等公民来对待。换句话说它不是又一个能跑就行的 relay 玩具而是冲着让 AI 代理在去中心化社交协议上真正干活这个目标去的。先把背景铺一下方便不熟悉的朋友跟上。Nostr 是一套极简的去中心化社交协议核心就两样东西客户端用密钥对签名事件event中继负责接收、存储、转发这些事件。协议本身不规定中继怎么实现、怎么存、怎么收费、怎么过滤这就留出了巨大的发挥空间。市面上大多数 relay 是 Node.js 或 Go 写的功能上够用但一旦你要在上面挂 AI agent——比如让一个 agent 监听特定关键词、自动回复、做内容审核、做语义检索——就会遇到几个硬骨头并发模型不够细、类型系统兜不住复杂的事件流、扩展点不清晰、性能在事件量上来之后掉得厉害。buzz 选择 Rust本质上是在赌三件事。第一Rust 的 ownership 和 async 生态tokio axum 这一套能让 relay 在高并发连接下保持低延迟和内存稳定这对一个要长期挂着 AI agent 的服务来说是刚需。第二强类型系统能把 Nostr 事件的各种 kind、tag、签名校验逻辑在编译期就约束住减少运行时的惊喜。第三Rust 的 FFI 和进程内嵌能力让 AI agent 可以直接跑在 relay 进程里而不是靠外部 webhook 来回折腾省掉一层网络开销和状态同步的麻烦。这篇文章我会按设计思路 → 核心细节 → 实操落地 → 问题排查这条线走把 buzz 这类 Rust Nostr relay 的里里外外拆开讲。适合两类人看一是想自己搭一个 relay 并且打算在上面跑点自动化逻辑的开发者二是正在学 Rust、想找一个真实项目练手尤其是 axum tokio 这套组合的人。我会尽量把为什么这么设计讲透而不是只丢一段配置让你抄。2. 整体架构与设计取舍为什么是 Rust axum 内嵌 Agent2.1 为什么不用现成的 Node/Go relay 改一改我最早也是拿现成的 relay 改的Node.js 那套上手快一天就能跑起来。但改到第三周就发现不对劲我要给 relay 加一个当收到包含特定 tag 的事件时触发一个本地模型做摘要并回帖的逻辑。在 Node 里这件事要么写成中间件塞进请求链路要么起一个独立进程订阅 relay 自己的 WebSocket。前者会把 AI 推理的耗时直接压到客户端请求的响应时间上后者则要处理两套状态、两套重连、两套鉴权。Go 的情况好一些goroutine 处理并发很舒服但 Go 的类型系统对 Nostr 这种事件结构固定、tag 语义千变万化的场景表达力有限你很容易写出一堆interface{}和类型断言。而且 Go 的 AI 生态相对薄很多推理库要么走 cgo要么走 HTTP 调外部服务绕不开。Rust 的吸引力在于tokio 的 task 模型天然适合每个连接一个 task、每个 agent 一个 task的结构axum 把 HTTP 和 WebSocket 统一在一个 tower 中间件栈里类型安全贯穿始终。更重要的是Rust 里可以直接把推理逻辑不管是调本地模型还是走 HTTP封装成一个实现了特定 trait 的组件编译进同一个二进制。部署的时候就是一个文件没有 node_modules没有运行时依赖。2.2 核心模块划分一个像 buzz 这样的 relay我习惯把它拆成五块每块职责单一方便替换和测试模块职责关键技术选型替换成本接入层处理 WebSocket/HTTP 连接、协议握手axum tokio-tungstenite低接口稳定事件校验验签、检查 kind/tag 合法性secp256k1 serde中涉及密码学存储层事件持久化、索引、查询SQLite/Postgres sqlx中看查询复杂度订阅分发维护订阅、匹配事件、推送内存索引 broadcast channel高性能敏感Agent 运行时加载、调度、隔离 AI agenttokio task trait object高设计空间大这个划分不是拍脑袋来的。接入层和校验层放一起会导致连接处理和密码学运算耦合压测时不好定位瓶颈存储和分发放一起会让写多读少和读多写少两种负载互相干扰。分开之后每一层都可以单独压测、单独替换。比如你一开始用 SQLite事件量上到百万级想换 Postgres只要存储层的 trait 定义清楚换实现不影响其他模块。2.3 Agent 作为一等公民的设计含义AI agents 是一等公民这句话落到代码上意味着 agent 不是外挂而是和订阅存储平级的内部概念。具体表现有三点第一agent 有自己的生命周期管理。relay 启动时读取配置实例化 agent给每个 agent 分配独立的 tokio task 和资源配额比如最多占用多少内存、每秒最多触发多少次推理。agent 崩溃不会拖垮 relay 主进程主进程会记录日志并按策略重启。第二agent 通过内部事件总线接收事件而不是走网络。relay 收到一个事件、校验通过、写入存储之后会往一个 broadcast channel 发一份所有订阅了这个 channel 的 agent 都能拿到。这比让 agent 自己开 WebSocket 连回 relay 要省太多——没有序列化开销没有网络往返没有鉴权问题。第三agent 的输出走统一的事件提交接口。agent 想发帖、想回复、想打标签都调用同一个内部函数这个函数会做和外部客户端一样的校验签名、限流、权限保证 agent 不能绕过规则乱来。这一点很重要否则 agent 就成了后门。提示把 agent 设计成内部客户端而不是特权模块是保证系统可审计的关键。任何 agent 的动作都应该在日志里留下和外部用户一样的痕迹。3. 核心细节拆解事件流、并发模型与 Agent 调度3.1 Nostr 事件在 Rust 里怎么表达Nostr 事件的结构其实很规整id、pubkey、created_at、kind、tags、content、sig。用 serde 定义一个结构体字段类型明确反序列化的时候就能挡掉一批脏数据。但真正的坑在 tags——它是一个二维字符串数组语义完全靠约定。比如[e, event_id]表示引用事件[p, pubkey]表示提及某人[t, topic]表示话题标签。我的做法是底层用VecVecString原样存保证不丢信息上层再包一层解析函数把常用的 tag 提取成强类型。这样既不会因为协议扩展导致解析失败又能在业务逻辑里享受类型安全。举个例子#[derive(Debug, Clone, Serialize, Deserialize)] pub struct Event { pub id: String, pub pubkey: String, pub created_at: u64, pub kind: u64, pub tags: VecVecString, pub content: String, pub sig: String, } impl Event { pub fn tag_values(self, name: str) - Vecstr { self.tags .iter() .filter(|t| t.first().map(|s| s.as_str()) Some(name)) .filter_map(|t| t.get(1).map(|s| s.as_str())) .collect() } }这段代码看着简单但tag_values这个辅助函数在整个项目里会被调用几百次。把它写好后面所有 agent 逻辑都轻松。注意filter_map那里用了get(1)因为有些 tag 可能只有名字没有值直接索引会 panic。这种细节在 Rust 里编译器不会帮你挡得自己养成习惯。3.2 并发模型连接、订阅、Agent 三类 tasktokio 的 task 是轻量的但也不是免费的。我的经验是分三类 task各自有明确的职责边界连接 task每个 WebSocket 连接一个负责读写帧、心跳、断线清理。这类 task 数量最多但逻辑最简单尽量不做重活。订阅 task每个活跃订阅一个负责从 broadcast channel 收事件、按 filter 匹配、推给对应连接。订阅数量通常远小于连接数。Agent task每个 agent 一个负责监听感兴趣的事件、执行逻辑、提交输出。数量最少但单个 task 可能很重比如跑推理。关键点是连接 task 和订阅 task 之间通过 channel 通信而不是共享锁。我见过有人用ArcMutexHashMap...管理订阅结果高并发下锁竞争把 CPU 吃满。改成每个连接持有自己的mpsc::Sender订阅 task 往 sender 里发连接 task 从 receiver 收锁就没了。Agent task 稍微特殊因为它可能要跑几秒甚至几十秒推理耗时。这类 task 绝对不能阻塞事件分发的主循环。我的做法是给 agent 配一个有界队列事件先入队agent 慢慢消费。队列满了就丢弃最旧的事件或者按优先级丢弃具体策略看业务。这样即使某个 agent 卡住也不会影响 relay 接收新事件。3.3 Agent 调度与资源隔离Agent 调度这块我踩过最大的坑是一个 agent 拖垮全局。早期版本里 agent 直接在事件分发的 task 里同步执行结果一个调外部 API 超时的 agent 把整个分发链路堵死了。后来改成完全异步 超时控制每个 agent 的每次执行都有 deadline超时就取消。资源隔离我用了两个手段。一是 tokio 的Semaphore限制单个 agent 的并发执行数比如最多同时处理 4 个事件超出的排队。二是给 agent 的推理调用设内存上限如果跑本地模型或者 QPS 上限如果调外部服务。这些配额写在配置文件里热加载不用重启。pub struct AgentRuntime { name: String, semaphore: ArcSemaphore, timeout: Duration, handler: Boxdyn AgentHandler Send Sync, } impl AgentRuntime { pub async fn dispatch(self, event: Event) { let permit match self.semaphore.clone().try_acquire_owned() { Ok(p) p, Err(_) { tracing::warn!(agent %self.name, agent busy, dropping event); return; } }; let handler self.handler.clone_box(); let timeout self.timeout; tokio::spawn(async move { let _permit permit; match tokio::time::timeout(timeout, handler.handle(event)).await { Ok(Ok(_)) {} Ok(Err(e)) tracing::error!(error %e, agent error), Err(_) tracing::warn!(agent timeout), } }); } }这段代码里try_acquire_owned而不是acquire是故意的——agent 忙的时候我宁愿丢事件也不愿排队堆积因为对社交场景来说过时的事件处理了也没意义。这个取舍要看具体业务如果是金融类事件就不能丢得改成阻塞等待加持久化队列。注意Boxdyn AgentHandler要能 clone得手动实现clone_box这是 Rust trait object 的经典痛点。别想着用#[derive(Clone)]trait object 不支持。4. 实操落地从零搭一个带 Agent 的 Rust Relay4.1 环境准备与依赖选型先把 Rust 环境弄好。我推荐用 rustup 装版本跟到 stable 就行别追 nightly除非你要用某些实验特性。装完之后cargo --version确认一下。国内网络环境下配置镜像源能省不少时间在~/.cargo/config.toml里加[source.crates-io] replace-with ustc [source.ustc] registry sparsehttps://mirrors.ustc.edu.cn/crates.io-index/依赖这块核心就几个axum做 HTTP/WStokio做运行时tokio-tungstenite处理 WebSocket 帧secp256k1验签sqlx连数据库serde/serde_json处理序列化tracing做日志。别一上来就堆一堆库先把最小可运行版本跑通再加。[dependencies] axum { version 0.7, features [ws] } tokio { version 1, features [full] } tokio-tungstenite 0.21 secp256k1 { version 0.29, features [rand, serde] } sqlx { version 0.7, features [runtime-tokio, sqlite] } serde { version 1, features [derive] } serde_json 1 tracing 0.1 tracing-subscriber 0.3选 axum 0.7 是因为它的 WebSocket 支持和 tower 中间件整合得比较顺而且社区活跃遇到问题好搜。secp256k1 用官方那个 crate别用纯 Rust 实现性能差一截。sqlx 选 SQLite 起步单文件、零运维事件量到几十万再考虑换。4.2 最小可运行 Relay 的骨架先搭一个能接收事件、验签、存库、转发的骨架。入口大概长这样#[tokio::main] async fn main() - anyhow::Result() { tracing_subscriber::fmt().with_env_filter(info).init(); let pool sqlx::SqlitePool::connect(sqlite:buzz.db?moderwc).await?; sqlx::query( CREATE TABLE IF NOT EXISTS events ( id TEXT PRIMARY KEY, pubkey TEXT NOT NULL, created_at INTEGER NOT NULL, kind INTEGER NOT NULL, tags TEXT NOT NULL, content TEXT NOT NULL, sig TEXT NOT NULL ) ).execute(pool).await?; let (tx, _) tokio::sync::broadcast::channel::Event(1024); let state AppState { pool, bus: tx }; let app axum::Router::new() .route(/, axum::routing::get(ws_handler)) .with_state(state); let listener tokio::net::TcpListener::bind(0.0.0.0:8080).await?; axum::serve(listener, app).await?; Ok(()) }broadcast::channel(1024)这个容量要斟酌。太小会导致慢消费者丢事件太大吃内存。1024 是我在中等负载下的经验值事件峰值高的话可以调到 4096但别超过 8192否则内存增长会很明显。WebSocket handler 里做三件事解析客户端发来的消息通常是[EVENT, {...}]或[REQ, ...]、校验、分发。校验包括验签和检查 id 是否匹配内容哈希。验签用 secp256k1 的 schnorr 签名验证这块代码不长但容易写错建议直接抄官方示例再改。4.3 接入第一个 Agent关键词监听与自动回复骨架跑通之后加一个最简单的 agent监听包含特定关键词的事件自动回复一条。这个 agent 能验证整条链路是否通畅。struct KeywordAgent { keyword: String, reply_template: String, signer: Signer, } #[async_trait::async_trait] impl AgentHandler for KeywordAgent { async fn handle(self, event: Event) - anyhow::Result() { if !event.content.contains(self.keyword) { return Ok(()); } if event.pubkey self.signer.pubkey() { return Ok(()); // 别回复自己 } let reply EventBuilder::new(1, self.reply_template) .tag(vec![e, event.id]) .tag(vec![p, event.pubkey]) .sign(self.signer)?; self.submit(reply).await } }这里有几个细节值得说。第一event.pubkey self.signer.pubkey()这个自检必须有否则 agent 回复自己的回复无限循环。第二回复事件要带etag 引用原事件、ptag 提及原作者这是 Nostr 的社交惯例不带的话客户端不会把它显示成回复。第三submit走的是内部提交接口会重新校验和限流agent 不能绕过。启动 agent 的时候把它注册到事件总线上let agent KeywordAgent { /* ... */ }; let runtime AgentRuntime::new(keyword-bot, agent, Duration::from_secs(10), 4); let mut rx state.bus.subscribe(); tokio::spawn(async move { while let Ok(event) rx.recv().await { runtime.dispatch(event).await; } });注意state.bus.subscribe()要在 spawn 之前调用否则可能漏掉早期事件。这个顺序问题我调试了半小时才发现broadcast channel 的订阅是从订阅那一刻开始收不是从头收。4.4 存储与查询优化事件存进 SQLite 之后查询性能取决于索引。至少要给pubkey、kind、created_at建索引因为客户端订阅的 filter 基本都围绕这几个字段。复合索引(kind, created_at)对某类事件按时间倒序这种查询帮助很大。CREATE INDEX IF NOT EXISTS idx_events_pubkey ON events(pubkey); CREATE INDEX IF NOT EXISTS idx_events_kind_created ON events(kind, created_at DESC); CREATE INDEX IF NOT EXISTS idx_events_created ON events(created_at DESC);tags 的查询是个难点因为它是 JSON 数组。SQLite 有 JSON1 扩展可以用json_each展开查询但性能一般。我的做法是对高频查询的 tag比如e、p、t单独建一张关联表写入事件的时候同步插入。这样查询走普通索引快很多。代价是写入多一次但 relay 通常是读多写少划算。CREATE TABLE event_tags ( event_id TEXT NOT NULL, tag_name TEXT NOT NULL, tag_value TEXT NOT NULL ); CREATE INDEX idx_event_tags_lookup ON event_tags(tag_name, tag_value);提示关联表方案在事件量超过百万后优势明显但要注意事务一致性。事件和 tag 的写入必须在同一个事务里否则会出现事件存在但 tag 查不到的诡异情况。5. 常见问题与排查技巧实录5.1 连接频繁断开怎么查WebSocket 断连是最常见的问题原因五花八门。我的排查顺序是先看 relay 日志有没有 panic再看是不是心跳没配最后看是不是客户端主动断的。心跳这块axum 的 WebSocket 默认不发 ping长时间空闲的连接会被中间设备掐掉。解决办法是起一个定时 task每隔 30 秒给所有连接发 ping收不到 pong 就清理。这个逻辑要写在连接 task 里别用全局定时器遍历所有连接那样锁竞争严重。let mut interval tokio::time::interval(Duration::from_secs(30)); loop { tokio::select! { _ interval.tick() { if socket.send(Message::Ping(vec![])).await.is_err() { break; } } msg socket.recv() { match msg { Some(Ok(m)) handle_message(m).await?, _ break, } } } }tokio::select!这个模式在处理既要定时又要收消息的场景里非常好用比开两个 task 加 channel 简洁。5.2 Agent 不触发或重复触发Agent 不触发八成是事件总线的订阅时机问题或者 filter 写错了。先确认 agent 的 task 真的起来了打日志再确认事件真的进了总线在分发处打日志。如果都正常那就是 filter 逻辑的问题把事件和 filter 都打印出来对比。重复触发通常是两个原因一是 agent 被注册了两次配置重复加载二是事件被重复分发比如存储和转发各发了一次总线。前者检查配置加载逻辑后者检查分发点是否唯一。我建议在分发处加一个事件 id 的去重集合用lrucrate 做个有界缓存能挡掉大部分重复。5.3 内存持续增长Rust 服务内存涨通常不是泄漏而是缓存或 channel 堆积。先看 broadcast channel 的容量和消费者速度慢消费者会导致消息在 channel 里堆积。再看有没有无界 channelunbounded_channel这个能不用就不用一旦生产快于消费内存就起飞。还有一个隐蔽的点Arc循环引用。如果 agent 持有 relay 的Arcrelay 又持有 agent 的Arc就永远不会释放。用Weak打破循环或者干脆别让 agent 持有 relay 引用改成通过 channel 通信。现象可能原因排查手段解决方向连接频繁断心跳缺失/中间设备超时抓包看 ping/pong加定时 pingAgent 不触发订阅时机/filter 错误分发处打日志调整订阅顺序Agent 重复触发重复注册/重复分发检查配置和分发点加去重缓存内存持续涨channel 堆积/循环引用看 channel 长度改有界/用 Weak验签失败率高编码问题/时钟偏移对比 id 和内容哈希统一 UTF-8/校时5.4 验签失败的几个坑Nostr 事件的 id 是内容序列化后的 SHA256序列化格式必须和协议规定的一模一样[0, pubkey, created_at, kind, tags, content]字段顺序、空格、转义都不能差。我见过有人用 serde_json 默认序列化结果多了空格id 全对不上。正确做法是手动拼这个数组或者用专门的序列化函数。签名验证用 schnorr公钥是 x-only 的 32 字节。从 hex 解码的时候注意别搞成 compressed 格式那是 33 字节。这个错误很隐蔽因为解码不报错只是验签一直失败。6. 性能调优与扩展方向6.1 压测与瓶颈定位压测我一般用websocat或者自己写个小脚本模拟 N 个连接、每秒发 M 个事件。观察三个指标CPU、内存、事件处理延迟。CPU 打满通常是验签或序列化内存涨是 channel 或缓存延迟高是锁竞争或 agent 阻塞。验签是 CPU 大户secp256k1 的 schnorr 验证单次大概几十微秒每秒几千事件就能吃满一个核。优化方向是批量验签或者用多核并行tokio 的spawn_blocking可以把验签丢到阻塞线程池避免占用 async 运行时。6.2 水平扩展的边界单个 relay 实例的能力有上限事件量再大就得考虑分片或集群。Nostr 协议本身不要求 relay 之间同步所以最简单的扩展是按 pubkey 分片——不同用户连不同 relay。但这样会牺牲全局可见性用户看不到其他分片的内容。另一种是 relay 之间互相转发但这会带来环路和重复问题实现复杂。我的建议是单实例先优化到极限几万并发连接、每秒几千事件真到瓶颈了再考虑分片别过早设计分布式。6.3 Agent 能力的扩展Agent 这块的想象空间很大。除了关键词回复还可以做语义检索把事件 embedding 存起来支持相似度查询、内容审核本地模型判断是否违规、自动摘要把长贴浓缩成短贴、跨语言翻译。每个能力都可以封装成一个独立的AgentHandler按需加载。关键是保持接口统一。所有 agent 都实现同一个 trait输入是事件输出是零个或多个待提交事件。这样新增 agent 不用改核心代码配置里加一行就行。这种插件化设计在 Rust 里用 trait object 实现很自然比动态语言还清爽因为编译器帮你检查了所有实现。我在实际项目里最深的体会是Rust 写 relay 前期慢后期快。类型系统和所有权检查会在编译期逼你把并发和生命周期想清楚一旦编译通过运行时的问题少一大半。buzz 这类项目真正的价值不在于用 Rust 重写而在于它把 AI agent 当成系统的一部分来设计而不是事后贴上去的补丁。如果你也在做类似的东西建议先把事件总线和 agent 生命周期这两块设计好剩下的都是水到渠成。