From 41ded07b2ff5c2509cfe0c772e46d1c1c06bb2b1 Mon Sep 17 00:00:00 2001 From: yibo Date: Sun, 24 May 2026 22:59:11 +0800 Subject: [PATCH] feat: 方案 D - 流式消费 hermes /v1/runs SSE，每个 tool 边界 → 一条独立微信消息 --- server/services/wechat/auto-reply.js | 236 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------- wechat_plain/scripts/local-testkit/keepalive.sh | 4 ++-- 2 files changed, 194 insertions(+), 46 deletions(-) diff --git a/server/services/wechat/auto-reply.js b/server/services/wechat/auto-reply.js index 96d5744..adba284 100644 --- a/server/services/wechat/auto-reply.js +++ b/server/services/wechat/auto-reply.js @@ -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 + * 串行调用；返回的 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); } } diff --git a/wechat_plain/scripts/local-testkit/keepalive.sh b/wechat_plain/scripts/local-testkit/keepalive.sh index b27d4ad..2a96589 100755 --- a/wechat_plain/scripts/local-testkit/keepalive.sh +++ b/wechat_plain/scripts/local-testkit/keepalive.sh @@ -27,7 +27,7 @@ CHECK_INTERVAL_SEC="${CHECK_INTERVAL_SEC:-15}" # 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 \ -- libgit2 0.21.4