feat: 方案 D - 流式消费 hermes /v1/runs SSE，每个 tool 边界 → 一条独立微信消息

旧路径：/v1/chat/completions 同步阻塞拿完整 reply → 一条微信消息。长 agent loop 用户体验差（"稍等" 之后再无下文）。新路径： 1) POST /v1/runs { input, conversation_history, session_id, instructions } 拿 run_id 立即返回 2) GET /v1/runs/{run_id}/events SSE 长连接，消费事件： - message.delta { delta } → 累积到 textBuf - tool.started → flush(false): textBuf → onUtterance - run.completed { output } → flush(true): 最后一段 - run.failed/cancelled → reject 3) onUtterance 在串行 Promise 队列（sendChain）里跑： - idx===0 的 utterance 带 quoteText + atSender（群里 @ 原消息） - 后续 idx>0 仅文本，不重复 quote/@ - 发完后 await sleep(14000 或 9000) 等 A11y 跑完整一轮 chat-list-click → quote → @ → type → send，避免下一个 broadcast 打断当前 FSM 状态 handleMessage： - 移除旧的 this.replyGenerator + sendTextMessageBroadcast 同步路径 - 改成 callHermesRunStream(messages, sessionId, onUtterance) - state.markReplied 在 run 整体完成后才标记（避免中间步骤已发但失败时重发漏掉）诊断：增加 [STREAM EV] 日志统计事件类型 + textBuf 长度，方便诊断"agent loop 跑很久但没 utterance"等情况（实测发现深度探索时 hermes 几乎只调 tool 不发 text，多步分割效果依赖 LLM 行为）。 keepalive.sh: HERMES_TIMEOUT 180s → 600s，长 agentic run 经常 3-5min。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

feat: 方案 D - 流式消费 hermes /v1/runs SSE，每个 tool 边界 → 一条独立微信消息
旧路径：/v1/chat/completions 同步阻塞拿完整 reply → 一条微信消息。长 agent loop 用户体验差（"稍等" 之后再无下文）。新路径： 1) POST /v1/runs { input, conversation_history, session_id, instructions } 拿 run_id 立即返回 2) GET /v1/runs/{run_id}/events SSE 长连接，消费事件： - message.delta { delta } → 累积到 textBuf - tool.started → flush(false): textBuf → onUtterance - run.completed { output } → flush(true): 最后一段 - run.failed/cancelled → reject 3) onUtterance 在串行 Promise 队列（sendChain）里跑： - idx===0 的 utterance 带 quoteText + atSender（群里 @ 原消息） - 后续 idx>0 仅文本，不重复 quote/@ - 发完后 await sleep(14000 或 9000) 等 A11y 跑完整一轮 chat-list-click → quote → @ → type → send，避免下一个 broadcast 打断当前 FSM 状态 handleMessage： - 移除旧的 this.replyGenerator + sendTextMessageBroadcast 同步路径 - 改成 callHermesRunStream(messages, sessionId, onUtterance) - state.markReplied 在 run 整体完成后才标记（避免中间步骤已发但失败时重发漏掉）诊断：增加 [STREAM EV] 日志统计事件类型 + textBuf 长度，方便诊断"agent loop 跑很久但没 utterance"等情况（实测发现深度探索时 hermes 几乎只调 tool 不发 text，多步分割效果依赖 LLM 行为）。 keepalive.sh: HERMES_TIMEOUT 180s → 600s，长 agentic run 经常 3-5min。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
yibo
1 parent 4fec6818
Showing 2 changed files with 194 additions and 46 deletions
server/services/wechat/auto-reply.js
wechat_plain/scripts/local-testkit/keepalive.sh
@@ -136,7 +136,155 @@ function log(...args) {
 // ═══════════════════════════════════════════════════════════════
  
 /**
- * 调用 hermes-agent API (OpenAI-compatible) 生成回复
+ * 调用 hermes-agent 的 /v1/runs 流式 API，按 agent loop 的 tool 边界
+ * 把每一段 assistant 文本作为独立"utterance"通过 onUtterance 回调推出。
+ *
+ * 事件流（SSE on /v1/runs/{id}/events）：
+ *   message.delta { delta: "字符片段" }     → 累积到 textBuf
+ *   tool.started  { tool, preview }          → flush textBuf 作为一条 utterance
+ *   tool.completed                           → 下一个 turn 准备开始
+ *   reasoning.available                       → 不转发（hermes 思考过程）
+ *   run.completed { output, usage }           → 最终 flush
+ *   run.failed / run.cancelled                → reject
+ *
+ * @param {Array} messages       OpenAI 格式（含 system / user / assistant 多轮）
+ * @param {string} sessionId     hermes session（每个 talker 独立）
+ * @param {Function} onUtterance async (text, index, isFinal) => Promise<void>
+ *                               串行调用；返回的 Promise 完成前不会发下一条
+ * @returns {Promise<{output?, usage?}>}
+ */
+async function callHermesRunStream(messages, sessionId, onUtterance) {
+  const userMsg = messages[messages.length - 1]?.content || '';
+  const systemMsg = messages.find(m => m.role === 'system');
+  const history = messages.slice(0, -1)
+    .filter(m => m.role !== 'system')
+    .map(m => ({ role: m.role, content: m.content }));
+
+  // 1) POST /v1/runs 拿 run_id
+  const startBody = JSON.stringify({
+    input: userMsg,
+    conversation_history: history,
+    session_id: sessionId,
+    instructions: systemMsg?.content,
+  });
+
+  const runId = await new Promise((resolve, reject) => {
+    const req = http.request({
+      hostname: CONFIG.hermesHost,
+      port: CONFIG.hermesPort,
+      path: '/v1/runs',
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        'Content-Length': Buffer.byteLength(startBody),
+      },
+      timeout: 15000,
+    }, res => {
+      let buf = '';
+      res.on('data', c => buf += c.toString());
+      res.on('end', () => {
+        try {
+          const j = JSON.parse(buf);
+          if (j.run_id) return resolve(j.run_id);
+          reject(new Error('No run_id in response: ' + buf.substring(0, 200)));
+        } catch (e) { reject(e); }
+      });
+    });
+    req.on('error', reject);
+    req.on('timeout', () => { req.destroy(); reject(new Error('start timeout')); });
+    req.write(startBody);
+    req.end();
+  });
+
+  log('[HERMES] Run started:', runId, 'session:', sessionId);
+
+  // 2) GET /v1/runs/{runId}/events (SSE)
+  return new Promise((resolve, reject) => {
+    const req = http.get({
+      hostname: CONFIG.hermesHost,
+      port: CONFIG.hermesPort,
+      path: `/v1/runs/${runId}/events`,
+      timeout: CONFIG.hermesTimeoutMs,
+    }, res => {
+      let buf = '';
+      let textBuf = '';
+      let utteranceIdx = 0;
+      let sendChain = Promise.resolve(); // 串行队列：每个 flush 在前一个完成后执行
+      let finalResult = {};
+
+      const flush = (isFinal) => {
+        const text = textBuf.trim();
+        textBuf = '';
+        if (!text) return;
+        const idx = utteranceIdx++;
+        // 关键：链式 append，不等待
+        sendChain = sendChain
+          .then(() => onUtterance(text, idx, !!isFinal))
+          .catch(e => log('[onUtterance error]', e.message));
+      };
+
+      // diagnostic counters
+      const evCount = { 'message.delta': 0 };
+      let lastDiagLogAt = Date.now();
+      const diagLog = (evt) => {
+        evCount[evt] = (evCount[evt] || 0) + 1;
+        const now = Date.now();
+        if (evt !== 'message.delta' || now - lastDiagLogAt > 10000) {
+          log('[STREAM EV]', evt, 'counts=', JSON.stringify(evCount), 'textBuf.len=' + textBuf.length);
+          lastDiagLogAt = now;
+        }
+      };
+
+      res.setEncoding('utf-8');
+      res.on('data', chunk => {
+        buf += chunk;
+        let sepIdx;
+        while ((sepIdx = buf.indexOf('\n\n')) >= 0) {
+          const block = buf.substring(0, sepIdx);
+          buf = buf.substring(sepIdx + 2);
+          for (const line of block.split('\n')) {
+            if (!line.startsWith('data: ')) continue;
+            let ev;
+            try { ev = JSON.parse(line.substring(6)); } catch (_) { continue; }
+            diagLog(ev.event);
+            switch (ev.event) {
+              case 'message.delta':
+                textBuf += (ev.delta || '');
+                break;
+              case 'tool.started':
+                // turn 边界：上一段 assistant 文本完成
+                flush(false);
+                break;
+              case 'run.completed':
+                flush(true);
+                finalResult = { output: ev.output, usage: ev.usage };
+                // 等串行队列 drain 完再 resolve
+                sendChain.then(() => resolve(finalResult));
+                break;
+              case 'run.failed':
+                reject(new Error(ev.error || 'run failed'));
+                break;
+              case 'run.cancelled':
+                reject(new Error('run cancelled'));
+                break;
+              // tool.completed / reasoning.available / approval.* 忽略
+            }
+          }
+        }
+      });
+      res.on('end', () => {
+        flush(true);
+        sendChain.then(() => resolve(finalResult));
+      });
+      res.on('error', reject);
+    });
+    req.on('error', reject);
+    req.on('timeout', () => { req.destroy(); reject(new Error('events stream timeout')); });
+  });
+}
+
+/**
+ * 调用 hermes-agent API (OpenAI-compatible) 生成回复 — 老的非流式同步 API
  *
  * @param {Array<{role:string,content:string}>} messages - OpenAI messages 格式
  * @param {string} sessionId - 会话 ID（每个 talker 独立，作为 X-Hermes-Session-Id）
@@ -527,64 +675,64 @@ class AutoReplyBot {
       log('[HISTORY] fetch failed:', e.message);
     }
  
-    let replyText;
-    try {
-      replyText = await this.replyGenerator({
-        type: replyType,
-        talker,
-        displayName,
-        content,
-        senderName,
-        history,
-        raw: msg.raw,
-      });
-    } catch (e) {
-      log('[REPLY GEN ERROR]', e.message);
-      return;
-    }
-
-    if (!replyText) {
-      log('[NO REPLY] Generator returned null');
-      return;
-    }
-
-    if (replyText.length > CONFIG.maxReplyLength) {
-      replyText = replyText.substring(0, CONFIG.maxReplyLength) + '...';
-    }
-
-    // 群里被 @ 时，给 broadcast 带上 quoteText（引用触发消息）+ atSender（回复里 @ 对方）
-    const sendOpts = { toUserDisplayName: displayName };
+    // 流式 hermes 调用直接在 onUtterance 里发送，不再先生成完整文本
+    // 群里被 @ 时，第一条回复带 quote + atSender；多步流式回复后续条仅文本
+    const baseSendOpts = { toUserDisplayName: displayName };
+    let firstSendOpts = { ...baseSendOpts };
     if (replyType === 'group-at' || replyType === 'group-quote') {
-      // 群消息 content 的格式是 "senderWxid:\n实际文本"，提取实际文本作为 quote substring
       let bubbleText = content;
       const colonIdx = bubbleText.indexOf(':\n');
       if (colonIdx > 0) bubbleText = bubbleText.substring(colonIdx + 2);
-      // 取前若干字符作为 substring 匹配；含 XML/特殊字符的部分（如 refermsg）跳过
       const trimmed = bubbleText.replace(/<[^>]+>/g, '').trim();
-      if (trimmed) sendOpts.quoteText = trimmed.substring(0, 30);
-      // senderName 是 wxid（如 'yibooo'），群里 @ 面板显示的是昵称（如 '火星人'），
-      // 必须查 rcontact 拿到昵称，否则 A11y 在选人面板里搜不到
+      if (trimmed) firstSendOpts.quoteText = trimmed.substring(0, 30);
       if (senderName) {
         try {
           const senderNick = await this.tracker.getContactNickname(senderName);
-          sendOpts.atSender = senderNick || senderName;
+          firstSendOpts.atSender = senderNick || senderName;
         } catch (_) {
-          sendOpts.atSender = senderName;
+          firstSendOpts.atSender = senderName;
         }
       }
-      log('[GROUP] quote=' + (sendOpts.quoteText || '-') + ' atSender=' + (sendOpts.atSender || '-'));
+      log('[GROUP] first-reply quote=' + (firstSendOpts.quoteText || '-')
+          + ' atSender=' + (firstSendOpts.atSender || '-'));
     }
  
-    try {
-      const result = await sendTextMessageBroadcast(talker, replyText, sendOpts);
-      if (result.ok) {
-        log('[SENT]', talker, `(as "${displayName}")`, replyText.substring(0, 60));
-        this.state.markReplied(msgId, talker);
-      } else {
-        log('[SEND FAILED]', talker, result.error);
+    // 走流式：每个 hermes turn → 一条独立微信消息
+    // 第一条带 quote + atSender，后续条仅文本（避免重复引用 / @）
+    const onUtterance = async (text, idx, isFinal) => {
+      // 单条限长保护
+      let toSend = text;
+      if (toSend.length > CONFIG.maxReplyLength) {
+        toSend = toSend.substring(0, CONFIG.maxReplyLength) + '...';
+      }
+      const opts = (idx === 0) ? firstSendOpts : baseSendOpts;
+      log('[UTTER]', idx, isFinal ? '(final)' : '', toSend.substring(0, 60));
+      try {
+        const r = await sendTextMessageBroadcast(talker, toSend, opts);
+        if (r.ok) {
+          log('[SENT]', talker, `#${idx}`, toSend.substring(0, 60));
+        } else {
+          log('[SEND FAILED]', talker, r.error);
+        }
+      } catch (e) {
+        log('[SEND ERROR]', e.message);
       }
+      // 等 A11y 跑完 chat-list-click → quote → @ → type → send 完整一轮
+      // 不等的话下一条 broadcast 会打断 A11y 当前状态
+      await sleep(idx === 0 ? 14000 : 9000); // 第一条带 quote+@ 多花时间
+    };
+
+    try {
+      const sessionId = `wx-${talker}`;
+      const messages = buildHermesMessages({
+        type: replyType, talker, displayName, content, senderName, history, raw: msg.raw,
+      });
+      log('[HERMES] Streaming for', talker, 'turns=' + messages.length);
+      const result = await callHermesRunStream(messages, sessionId, onUtterance);
+      log('[HERMES] Run done. usage=' + JSON.stringify(result.usage || {}));
+      this.state.markReplied(msgId, talker);
     } catch (e) {
-      log('[SEND ERROR]', e.message);
+      log('[STREAM ERROR]', e.message);
     }
   }
  
@@ -27,7 +27,7 @@ CHECK_INTERVAL_SEC=&quot;${CHECK_INTERVAL_SEC:-15}&quot;
 # auto-reply 启动用的 env（避开 start-ar.sh 的 shell hook 污染）
 MY_WXID="${MY_WXID:-wxid_51ap3m6p9xs722}"
 MY_NICKNAME="${MY_NICKNAME:-小火子}"
-HERMES_TIMEOUT_MS="${HERMES_TIMEOUT_MS:-180000}"
+HERMES_TIMEOUT_MS="${HERMES_TIMEOUT_MS:-600000}"
 HISTORY_CONTEXT_SIZE="${HISTORY_CONTEXT_SIZE:-8}"
  
 log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; }
@@ -108,7 +108,7 @@ start_auto_reply() {
         MY_NICKNAME=\"${MY_NICKNAME:-小火子}\" \
         HERMES_HOST=127.0.0.1 \
         HERMES_PORT=$HERMES_PORT \
-        HERMES_TIMEOUT=${HERMES_TIMEOUT_MS:-180000} \
+        HERMES_TIMEOUT=${HERMES_TIMEOUT_MS:-600000} \
         NOTIFICATION_ENABLED=true \
         HISTORY_CONTEXT_SIZE=${HISTORY_CONTEXT_SIZE:-8} \
         nohup /data/data/com.termux/files/usr/bin/node auto-reply.js \