abtop扩展实战:5步实现AgentCollector trait,为你的AI编码Agent新增监控支持
【免费下载链接】abtopLike htop, but for AI coding agents. Monitor Claude Code & Codex CLI sessions, tokens, context window, rate limits, and ports in real-time.项目地址: https://gitcode.com/gh_mirrors/ab/abtop
abtop 是一款类 htop 的终端 AI 编码 Agent 监控工具,能实时查看 Claude Code、Codex CLI、OpenCode 会话的 Token 用量、上下文窗口、速率限制与端口占用。如果你日常使用其他 AI 编码 Agent,想让它也出现在同一块仪表板上,本文就用5 步带你实现核心扩展点AgentCollectortrait,为 abtop 新增监控支持。
为什么 abtop 适合二次扩展
abtop 的设计是"一个 Agent 一个收集器":主循环每 2 秒执行一次 tick,把所有已注册的收集器统一调度,汇总后渲染到终端面板。这意味着新增一个 Agent 监控,不需要改动任何 UI 代码,只需提供一个新的收集器实现并注册进去。
整体协作流程如下:
MultiCollector ──每tick──> 各 AgentCollector.collect(shared) │ │ └── SharedProcessData(进程/子进程/端口,全局只采集一次)- 统一调度入口:MultiCollector
- 共享进程数据(避免重复执行
ps/lsof):SharedProcessData - 三个官方参考实现:claude.rs、codex.rs、opencode.rs
先获取源码仓库开始动手:
git clone https://gitcode.com/gh_mirrors/ab/abtop第1步:读懂数据模型 AgentSession
collect()的返回值是Vec<AgentSession>,它就是 abtop 渲染所有面板的"通用货币"。读懂它,就完成了一半。
AgentSession定义在 src/model/session.rs,核心字段一览:
| 字段 | 含义 | 必填程度 |
|---|---|---|
agent_cli | Agent 标识,如"claude"、"codex",静态字符串 | 必填 |
pid | Agent 主进程 PID(0 表示进程不可用) | 必填 |
session_id/cwd/project_name | 会话 ID、工作目录、项目名 | 必填 |
status | 状态:Thinking/Executing/Waiting/Done等 | 必填 |
context_percent/context_window | 上下文窗口占用百分比与窗口大小 | 建议填 |
total_input_tokens/total_output_tokens | Token 累计值 | 建议填 |
current_tasks/mem_mb/children | 当前任务、内存、子进程(含端口) | 可选 |
💡 小技巧:字段很多,但参考最精简的 OpenCodeCollector 就能看到一个"合格答案"长什么样——填不准的字段用
0、空字符串或空Vec占位即可。
第2步:定位进程,发现你的 Agent 会话
监控的前提是"找到正在运行的 Agent 进程"。好消息是:ps结果已经通过SharedProcessData共享给你了,不要自己去再跑一遍ps。
典型做法(参考 OpenCodeCollector::find_opencode_pids):
let my_agent_pids = shared.process_info .iter() .filter(|(_, info)| process::cmd_has_binary(&info.command, "my-agent")) .map(|(pid, _)| *pid) .collect::<Vec<_>>();然后从 Agent 自己的本地状态(配置文件、日志、SQLite 数据库等)中读取会话详情,用 PID 与工作目录做匹配,把"活的进程"和"会话记录"对应起来。abtop 对数据源的原则是:只读本地文件 + 进程信息,不发任何网络请求。
第3步:填充 AgentSession 关键字段
拿到会话数据后,按 AgentSession 的字段逐一对应填入。几个高频要点:
- 状态判断:可参考 SessionStatus 的注释——进程存活且最近有活动 →
Thinking/Executing;长时间无活动 →Waiting;进程已退出 →Done(Done的会话会被自动过滤)。 - 上下文窗口:数据源里通常没有现成数值,可用现成的 context_window_for_model 按模型名推算窗口大小(200K/1M),再计算百分比。
- 安全清洗:显示到 TUI 的文本建议过一遍 sanitize_terminal_text 与 redact_secrets,避免控制字符和密钥泄漏进界面。
第4步:实现 AgentCollector trait 并注册到 MultiCollector
扩展点只有 15 行,一个必选方法 + 两个可选方法:
| 方法 | 必需 | 作用 |
|---|---|---|
collect(&mut self, shared: &SharedProcessData) -> Vec<AgentSession> | ✅ | 每个 tick 返回该 Agent 的存活会话列表 |
live_rate_limit(&self) -> Option<RateLimitInfo> | ❌ | 提供账户级速率限制(配额面板展示) |
discovered_config_dirs(&self) -> Vec<PathBuf> | ❌ | 上报发现的配置目录,供跨目录查询限额 |
最小骨架(可对照 OpenCodeCollector 的 impl):
pub struct MyAgentCollector { /* 缓存字段放这里 */ } impl AgentCollector for MyAgentCollector { fn collect(&mut self, shared: &SharedProcessData) -> Vec<AgentSession> { // 第2~3步的成果在这里返回 } // 不需要配额面板?两个可选方法留默认实现即可 }最后一步是注册:在 with_hidden_and_claude_config_dirs 的collectors列表中加入Box::new(MyAgentCollector::new())。注意agent_cli标识(如"my-agent")会参与hidden_agents配置的大小写不敏感匹配——用户可在~/.config/abtop/config.toml中通过hidden_agents = ["my-agent"]隐藏你的收集器,请保持一致。
⚡ 性能约定:昂贵的发现操作(扫磁盘、读
/proc)放到shared.slow_tick == true时执行(约每 10 秒一次),快速 tick 复用缓存。参考 OpenCodeCollector 的缓存策略。
第5步:构建、运行与测试验证
cargo build # 编译通过 cargo test # 建议为你的收集器补上单元测试(可参考各 collector 文件底部 #[cfg(test)]) cargo run -- --once # 单帧快照,快速验证会话是否出现 cargo run # 启动 TUI,观察你的 Agent 是否进入 sessions 面板验证清单:
- ✅
--once输出中出现你的 Agent 会话行 - ✅ sessions 面板中状态(Thinking/Waiting)随实际活动切换
- ✅ 子进程与端口正确归属到会话,端口面板无重复
- ✅ 杀掉 Agent 进程后会话消失(
Done自动过滤) - ✅ 10 秒级慢 tick 下不卡顿(昂贵 I/O 已缓存在 fast tick)
参考资料
- 扩展入口 trait:AgentCollector
- 会话数据模型:src/model/session.rs
- 最简实现范本(SQLite 数据源 + 缓存):src/collector/opencode.rs
- 复杂实现范本(大文件增量解析):src/collector/claude.rs
- 速率限制读取:src/collector/rate_limit.rs
- 项目架构与数据源说明:AGENTS.md
按照以上 5 步,你就完成了 abtop 的 Agent 监控扩展。从"读懂数据模型"到"注册收集器",核心代码量不过百行——现在,让你的 AI 编码 Agent 也登上这块实时监控仪表板吧 🚀
【免费下载链接】abtopLike htop, but for AI coding agents. Monitor Claude Code & Codex CLI sessions, tokens, context window, rate limits, and ports in real-time.项目地址: https://gitcode.com/gh_mirrors/ab/abtop
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考