diff --git a/data/agents/qa-judge/config.json b/data/agents/qa-judge/config.json index 8cb3a53..72e7ca6 100644 --- a/data/agents/qa-judge/config.json +++ b/data/agents/qa-judge/config.json @@ -21,5 +21,9 @@ "maxMemoryMB": 512, "maxDiskMB": 1024, "maxProcesses": 5 + }, + "heartbeat": { + "enabled": true, + "intervalMinutes": 10 } } \ No newline at end of file diff --git a/docs/system-design/23-agent-active-heartbeat.md b/docs/system-design/23-agent-active-heartbeat.md new file mode 100644 index 0000000..27c2575 --- /dev/null +++ b/docs/system-design/23-agent-active-heartbeat.md @@ -0,0 +1,95 @@ +# 23. Agent Active Heartbeat System — 主动心跳与监控架构 + +> **版本**: v1.0 (设计稿) +> **日期**: 2026-03-25 +> **状态**: 规划中 +> **前置**: Phase 3 记忆架构已完成,OKR 目标大盘已上线 + +--- + +## 一、需求背景与痛点 + +当前的 LemonBoss 架构中,所有的 Agent(包含管理者、执行者、QA 仲裁者等)都处于**“被动响应”**状态。 +即:只有当人类或者其他 Agent 发送一条消息(如 `TaskDispatch` 或直接 Chat),它们才会被唤醒并进行一轮推理和动作执行。 + +**痛点:** +随着 OKR 系统(Goal Nodes)复杂度的增加,如果一个任务的负责人(Owner)迟迟不推进进度,或者产出质量不达标。在没有外部输入时,系统的主节点(如 `qa-judge` 质量裁判,或者 `master` 主节点)只能一直假死等待,这严重违背了“**自主智能体(Autonomous Agent)**”能在后台 24/7 监控并主动下发指令的设计初衷。 + +**目标:** +使特定的高级 Agent(如 `qa-judge`)具备内建的相对时间感知能力。通过实现一套主推的“心跳机制(Active Heartbeat)”,使其能每隔 N 分钟自动“苏醒”,根据当前注入体内的 OKR 目标大盘数据,主动进行一次进度巡检、质量总结,并主动向下游 Agent 节点派发反馈任务或催办通知。 + +--- + +## 二、架构设计原理 + +由于 LLM(如 Claude)本身是无状态、无法自主定时的算法黑盒,我们必须通过 Node.js 后端作为起搏器(Pacemaker)来主动触碰 Agent 的执行核心。 + +### 1. 核心流程图 + +```mermaid +sequenceDiagram + participant Cron as Node.js Heartbeat Service + participant Goal as OKR / Goals Data + participant Agent as qa-judge (Agent) + participant Worker as poet (Downstream Agent) + + loop 每 N 分钟 + Cron->>Goal: 获取当前活跃的 OKR 大盘状态 + Cron->>Agent: [SYSTEM HEARTBEAT] 唤醒 (无前端UI展示) + Note over Agent: 挂载 memory、heartbeat.md 及最新 Goal Context + Agent->>Agent: 推理分析 (是否有节点超时/待验收?) + + alt 需要干预 + Agent->>Worker: API调用: TaskDispatch (催促或驳回重做) + Agent->>Agent: "已向负责人发送警告" + else 一切正常 + Agent->>Agent: "保持观察,无需操作" + end + Agent-->>Cron: 结束本次心跳推理 (保存 session 记录入短期记忆) + end +``` + +### 2. 参与模块设计 + +#### (1) 后端起搏器 (`server/services/heartbeat-service.js`) +- 系统级别的 `setInterval` 或使用 `node-cron` 定时任务管理器。 +- 模块初始化时,读取注册了 `cron.json` 或在 `config.json` 中配置了 `active_heartbeat: true` 的全网 Agent。 +- 按设定时间周期(如每 5 分钟、或 15 分钟),发起内部进程级或 HTTP `POST /api/agent/:id/chat` 请求。 +- **消息构造**:必须带有特殊的标记,如 `[SYSTEM HEARTBEAT] 这是系统周期心跳。时间:{当前时间}。如果无需采取行动,请只回复[心跳正常];如果发现 OKR 或任务阻塞,请使用你的 dispatch 工具发送跟进动作。` + +#### (2) Agent 身份与职责声明 (`data/agents/qa-judge/heartbeat.md`) +这部分将作为心跳特供的系统注入层。因为 Agent 被系统唤醒后可能“不知所措”,所以需要在注入时提供明确的**心跳策略**。 +- `qa-judge` 的 `heartbeat.md` 应当规定: + - 检查当前全局进度是低于预期、正常还是卡点。 + - 对于已经标记为 `ready_for_review` 的节点,主动执行抽样检查工具,如果抽样不通过,则主动调度退回给原作者。 + - 规定其只有遇到问题才说话,没事的时候必须节约 Token,直接结案。 + +#### (3) 会话隐藏与静默运行 (隐式聊天机制) +为了不污染用户的主聊天界面: +- 心跳触发的 Conversation 应当拥有独特的 ID(如 `heartbeat-177xxxxx`)。 +- 在前端 UI(AgentChat.vue)中,如果不特别切换到 heartbeat 专区,应该不展示这类后台巡逻日志,以保持人类界面的干净。 +- 但心跳的整个完整推演过程(产生的 session)**必须依然进入现有的 Memory 三层架构**。也就是说,如果 `qa-judge` 在心跳中发出了很多派发任务,这个行为依然会记录进 `short-term/session.md` 里并最终写进日常报表(Daily Report)。 + +--- + +## 三、待实现 TODO 计划 + +- [ ] **1. Heartbeat 后端引擎设计** + - 在 `server/services/heartbeat-service.js` 实现通用的心跳轮询控制器。 + - 能够读取各个 Agent 配置的 `heartbeatInterval`(支持独立配置时间间隔,如 qa-judge 15分钟,master 30分钟)。 +- [ ] **2. 隐式对话 API (`agent-chat.js` 支持)** + - 修改 `agent-chat.js` 中的底层 ACP 唤起逻辑,支持传入 `silent: true` 以及 `isHidden` 参数。 + - 确保这些会话依然会经历 Phase 1 的记忆提取。 +- [ ] **3. `qa-judge` 的专项 prompt 优化** + - 完善 `qa-judge/heartbeat.md` 与 `qa-judge/cron.json`,使其明确自己“裁判+巡检”的作用和判定 OKR 进度的标准。 +- [ ] **4. 拦截发散风暴 (Anti-Loop 制衡机制)** + - 防止 Agent 每次心跳都在给别人发重复的警告。 + - **方案**:因为有 Memory 架构的存在,Agent 在心跳前会读到自己最近的 session(如“我在 15分钟前刚发过催促”),因此可以通过 prompt 指导它:“如果发现最近一次记忆中已经催促过了,请安静等待至少半天再做后续操作”。这完美与最新的 22 号文档中完成的 Memory 系统契合互补。 + +--- + +## 四、技术优势 + +1. **“起搏”+“记忆”打造真实自主引擎**:依赖已实现的海马体与短期记忆,Agent 不再是简单的代码循环,而是拥有自己时间轴与经验累积的鲜活生命体。 +2. **极低侵入性**:依然可以复用现有的 Claude CLI 通道和 `runClaudeCliSession` 函数,仅仅是多了一个虚拟的“系统定时发信人”。 +3. **闭环 OODA loop**:观察(OKR数据注入)- 导向(heartbeat.md规则)- 决策(推理判定)- 行动(ToolDispatch),在此实现完整的智能自主控制闭环。 diff --git a/server/index.js b/server/index.js index e03f00a..8118d26 100644 --- a/server/index.js +++ b/server/index.js @@ -60,7 +60,7 @@ async function start() { ensureDataDirs(); const host = config.host || '0.0.0.0'; - server.listen(config.port, host, () => { + server.listen(Number(config.port), host, () => { console.log(`✅ 服务启动: http://${host}:${config.port}`); }); @@ -70,9 +70,11 @@ async function start() { // 插件系统(异步,不阻塞 HTTP) initPlugins().catch(err => console.error('插件启动失败:', err.message)); - // 恢复中断任务(延迟 5s) + // 恢复中断任务并起搏 Agent (延迟 5s,等 HTTP 服务完全就绪) setTimeout(() => { resumeIncompleteTasks().catch(err => console.error('[Startup] 恢复任务失败:', err.message)); + const heartbeatService = require('./services/heartbeat-service'); + heartbeatService.start(); }, 5000); } catch (error) { console.error('Failed to start server:', error); diff --git a/server/routes/agent-chat.js b/server/routes/agent-chat.js index a8efbba..987eb59 100644 --- a/server/routes/agent-chat.js +++ b/server/routes/agent-chat.js @@ -1151,7 +1151,8 @@ router.post('/:agentId/chat', async (req, res) => { meta = { conversationId: convId, agentId, - title: message.slice(0, 50), + title: req.body.isHeartbeat ? '⏱️ 自主心跳巡检' : message.slice(0, 50), + type: req.body.isHeartbeat ? 'heartbeat' : 'chat', createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), status: 'running', diff --git a/server/services/heartbeat-service.js b/server/services/heartbeat-service.js new file mode 100644 index 0000000..7aae896 --- /dev/null +++ b/server/services/heartbeat-service.js @@ -0,0 +1,125 @@ +const fs = require('fs'); +const path = require('path'); +const http = require('http'); +const config = require('../config'); + +const DATA_DIR = path.resolve(__dirname, '../../data/agents'); + +class HeartbeatService { + constructor() { + this.timers = new Map(); // agentId -> NodeJS.Timeout + this.port = config.port || 3000; + } + + /** + * 初始化并启动全局的心跳监控 + */ + start() { + console.log('[Heartbeat] 💓 开始初始化 Agent 心跳服务...'); + if (!fs.existsSync(DATA_DIR)) return; + + const agents = fs.readdirSync(DATA_DIR).filter(f => { + const p = path.join(DATA_DIR, f); + return fs.lstatSync(p).isDirectory() && fs.existsSync(path.join(p, 'config.json')); + }); + + for (const agentId of agents) { + this._setupAgentHeartbeat(agentId); + } + } + + /** + * 停止所有心跳 + */ + stop() { + for (const [agentId, timer] of this.timers.entries()) { + clearInterval(timer); + } + this.timers.clear(); + console.log('[Heartbeat] 🛑 心跳服务已停止'); + } + + /** + * 为单个 Agent 根据 config 设置心跳 + */ + _setupAgentHeartbeat(agentId) { + try { + const configPath = path.join(DATA_DIR, agentId, 'config.json'); + const cfg = JSON.parse(fs.readFileSync(configPath, 'utf8')); + + if (cfg.heartbeat && cfg.heartbeat.enabled === true) { + const intervalMs = (cfg.heartbeat.intervalMinutes || 15) * 60 * 1000; + + // 确保不会重复启动 + if (this.timers.has(agentId)) { + clearInterval(this.timers.get(agentId)); + } + + const timer = setInterval(() => this._triggerHeartbeat(agentId), intervalMs); + this.timers.set(agentId, timer); + console.log(`[Heartbeat] ✅ 已挂载 ${agentId} 的主动心跳 (间隔: ${cfg.heartbeat.intervalMinutes} 分钟)`); + } + } catch (err) { + console.warn(`[Heartbeat] ⚠️ 读取 ${agentId} 配置失败: ${err.message}`); + } + } + + /** + * 内部方法:向 Agent 的 Chat API 发送一次隐式的系统心跳 + */ + _triggerHeartbeat(agentId) { + const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19); + console.log(`[Heartbeat] 💓 触发 ${agentId} 主动心跳 (${timestamp})...`); + + const convId = `heartbeat-${Date.now().toString(36)}`; + const heartbeatMessage = [ + `[SYSTEM HEARTBEAT]`, + `当前系统时间:${timestamp}`, + `这是一个后台周期的主动心跳触发。`, + `要求:请根据你的身份和 heartbeat.md 规则,扫描当前最新的全局 OKR 目标状态和节点。`, + `1. 如果发现某个负责人超时、卡点或产出需要仲裁,请主动使用 dispatch 工具给他们发送通知或驳回。`, + `2. 如果一切正常,无事发生,请直接回复“[心跳正常]”并且不要使用任何工具。`, + `无需等待人类确认,请完全自主决策。` + ].join('\n'); + + const postData = JSON.stringify({ + message: heartbeatMessage, + conversationId: convId, + isHeartbeat: true // 特殊标记,用于前端或后端识别 + }); + + const options = { + hostname: 'localhost', + port: this.port, + path: `/api/agent/${agentId}/chat`, + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Content-Length': Buffer.byteLength(postData) + } + }; + + const req = http.request(options, (res) => { + let body = ''; + res.on('data', (chunk) => body += chunk); + res.on('end', () => { + if (res.statusCode >= 200 && res.statusCode < 300) { + console.log(`[Heartbeat] 📨 ${agentId} 成功启动心跳推理会话 (${convId})`); + } else { + console.warn(`[Heartbeat] ⚠️ ${agentId} 心跳触发异常: ${res.statusCode} ${body}`); + } + }); + }); + + req.on('error', (err) => { + console.error(`[Heartbeat] ❌ ${agentId} 心跳请求失败: ${err.message}`); + }); + + req.write(postData); + req.end(); + } +} + +// 导出单例 +const heartbeatService = new HeartbeatService(); +module.exports = heartbeatService;