From a01fe21ab15bce7a0bcbf3227406b4e8798132d5 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 14 Sep 2026 16:09:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(qq):=20=E5=90=8C=E4=B8=80=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E8=BF=9E=E7=BB=AD=E6=B6=88=E6=81=AF=E5=90=88=E5=B9=B6?= =?UTF-8?q?=E4=B8=BA=E4=B8=80=E6=AC=A1=E4=B8=AD=E6=96=AD=20+=20=E7=A4=BA?= =?UTF-8?q?=E4=BE=8B=20SDK=20=E6=8C=87=E5=9B=9E=E4=BB=93=E5=BA=93=E6=BA=90?= =?UTF-8?q?=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 需求（jianf）：同一个人连发的数条消息应打包成一次中断，别逐条唤醒 Agent。 - debounce 合并：同一会话 + 同一发送者（群聊按 群号+QQ、私聊按 QQ）在 batch_window_ms（默认 1500）内的连续消息合成一批，每来一条重置计时； 整批不超过 batch_max_ms（默认 30000），避免对方持续刷屏时一直不投。 - n>1 时中断说明「短时间连续发来 N 条」并列出 message_id，建议一次 get_history 拿全上下文；n==1 沿用原文，行为与合并前逐字一致。 - 可配置 batch_window_ms / batch_max_ms，0 = 关闭合并（逐条投递）。 - Stop 时 flush 未到点批次，别把对方消息吞掉。 - 新增 4 条测试（同发送者合并 / 不同发送者不合并 / 窗口 0 逐条 / 单条沿用原文）。 顺带：deepsearch / vikunja 的 go.mod 与 plg.json 此前指向本地安装的 SDK 1.2.0， 导致无法用当前 SDK 重编（缺 InjectOptions.Priority）。改回 ../../ 仓库源码， 与其余示例一致。 --- example/deepsearch/go.mod | 4 +- example/deepsearch/plg.json | 11 +- example/qq/plugin.go | 193 ++++++++++++++++++++++++++++++++++-- example/qq/plugin_test.go | 77 ++++++++++++++ example/vikunja/go.mod | 4 +- example/vikunja/plg.json | 11 +- 6 files changed, 281 insertions(+), 19 deletions(-) diff --git a/example/deepsearch/go.mod b/example/deepsearch/go.mod index 8797bad..0d2a34d 100644 --- a/example/deepsearch/go.mod +++ b/example/deepsearch/go.mod @@ -2,7 +2,7 @@ module deepsearch-plugin go 1.25.0 -require gitcode.com/JianFeeeee/homeagent-sdk v1.2.0 +require gitcode.com/JianFeeeee/homeagent-sdk v0.0.0 @@ -18,4 +18,4 @@ require gitcode.com/JianFeeeee/homeagent-sdk v1.2.0 -replace gitcode.com/JianFeeeee/homeagent-sdk => /root/.homeagent/hmapdev/sdk/v1.2.0 +replace gitcode.com/JianFeeeee/homeagent-sdk => ../../ diff --git a/example/deepsearch/plg.json b/example/deepsearch/plg.json index 554f50c..ce6957f 100644 --- a/example/deepsearch/plg.json +++ b/example/deepsearch/plg.json @@ -6,7 +6,12 @@ "description": "为 agent 提供真正的联网信息检索：本地 SearXNG 聚合多引擎（返回标题/URL/摘要/时间），支持新闻、时间范围、指定引擎；并提供网页正文抽取与「搜索+读前K篇」的深检索", "author": "HomeAgent", "entry": "plugin.bin", - "sdk": "1.2.0", - "tags": ["search", "web", "searxng", "retrieval", "news"], + "tags": [ + "search", + "web", + "searxng", + "retrieval", + "news" + ], "targets": "linux/amd64" -} +} \ No newline at end of file diff --git a/example/qq/plugin.go b/example/qq/plugin.go index 7e53be8..5168d76 100644 --- a/example/qq/plugin.go +++ b/example/qq/plugin.go @@ -155,6 +155,41 @@ type Plugin struct { msgMu sync.Mutex msgMap map[int64]msgRef // message_id → {peer, time} chats map[int64]*chatMeta // peerID → 会话状态（群号或 QQ 号） + + // 消息合并（debounce）：同一会话、同一发送者在 batchWindow 内连续到达的消息 + // 合并成一次中断。同一个人连发「在吗」「帮我看看」「报错是这个」三条， + // 逐条注入会把 Agent 唤醒三次，且前两次拿到的信息都不完整。 + batchMu sync.Mutex + batches map[string]*pendingBatch + batchWindow time.Duration // 最后一条到达后再等多久（<=0 = 关闭合并，逐条投递） + batchMax time.Duration // 一批最长等多久（防持续刷屏时永远不投） + + // injectHook 仅供测试：非 nil 时 injectInterrupt 走它而不是真实 SDK。 + injectHook func(text string) +} + +// pendingBatch 是一批待投递的消息（同一会话、同一发送者、短时间内的连续消息）。 +type pendingBatch struct { + key string + isGroup bool + userID int64 + groupID int64 + nickname string + msgIDs []int64 + single string // 单条时沿用的原文（含所有者/高危前缀），保证 n==1 行为不变 + owner bool + highRisk bool + first time.Time + timer *time.Timer +} + +// qqBatchKey 同一会话 + 同一发送者 = 一组。私聊按 QQ 号；群聊按 (群号, QQ 号)—— +// 群里不同人各发各的，不该并成一条。 +func qqBatchKey(msgType string, groupID, userID int64) string { + if msgType == "group" { + return fmt.Sprintf("g:%d:%d", groupID, userID) + } + return fmt.Sprintf("p:%d", userID) } type typingState struct { @@ -316,6 +351,8 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { s.Settings().RegisterDef(sdk.ConfigDef{Key: "remote_dir", Default: "/home/program/qq-workspace/remote", Type: "string", DisplayName: "NapCat容器共享目录", Description: "与NapCat容器共享的文件目录，主机路径。发文件时文件会复制到此目录，NapCat内部映射为/app/files/", Category: "qq"}) s.Settings().RegisterDef(sdk.ConfigDef{Key: "webhook_token", Default: "", Type: "string", DisplayName: "Webhook 令牌", Description: "NapCat 上报请求头 X-Webhook-Token 校验值，留空则不校验", Category: "qq"}) s.Settings().RegisterDef(sdk.ConfigDef{Key: "agentfs_dir", Default: "/home/newqqagent/agentfs/merged", Type: "string", DisplayName: "AgentFS目录", Description: "文件读写的工作目录，read_document/video_download 等工具的默认工作目录", Category: "qq"}) + s.Settings().RegisterDef(sdk.ConfigDef{Key: "batch_window_ms", Default: "1500", Type: "int", DisplayName: "消息合并窗口(毫秒)", Description: "同一会话同一发送者的连续消息在该窗口内合并成一次中断并告知共几条；0=关闭合并（逐条投递）", Category: "qq"}) + s.Settings().RegisterDef(sdk.ConfigDef{Key: "batch_max_ms", Default: "30000", Type: "int", DisplayName: "消息合并上限(毫秒)", Description: "一批消息最长等这么久就投递，避免对方持续刷屏时一直不唤醒 Agent", Category: "qq"}) settings := s.Settings() @@ -339,12 +376,18 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { p.filesDir = strings.TrimRight(getSetting[string](settings, "files_dir", "/home/newqqagent/agentfs/merged/qq_files"), "/") p.agentfsDir = strings.TrimRight(getSetting[string](settings, "agentfs_dir", "/home/newqqagent/agentfs/merged"), "/") p.remoteDir = strings.TrimRight(getSetting[string](settings, "remote_dir", "/home/program/qq-workspace/remote"), "/") + p.batchWindow = time.Duration(getSetting[int64](settings, "batch_window_ms", 1500)) * time.Millisecond + p.batchMax = time.Duration(getSetting[int64](settings, "batch_max_ms", 30000)) * time.Millisecond + if p.batchWindow < 0 { + p.batchWindow = 0 + } os.MkdirAll(p.remoteDir, 0755) p.httpClient = &http.Client{Timeout: 30 * time.Second} // msg_id → peer 映射 + 会话状态（不缓存正文） p.msgMap = make(map[int64]msgRef) + p.batches = make(map[string]*pendingBatch) p.chats = make(map[int64]*chatMeta) // 从 NapCat 获取 Bot 身份（阻塞等待，最多 5s） @@ -697,6 +740,8 @@ func (p *Plugin) Stop() error { } } p.typingMu.Unlock() + // 停机前把未到点的合并批次立刻投出去，别把对方的消息吞掉。 + p.flushAllBatches() if p.srv != nil { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -761,6 +806,10 @@ func getSetting[T string | int64 | float64](s sdk.SettingsAPI, key string, fallb } case int64: switch val := v.(type) { + case int: + return any(int64(val)).(T) + case int64: + return any(val).(T) case float64: return any(int64(val)).(T) case string: @@ -1299,6 +1348,136 @@ func requiresConfirmFriendCommand(cmd string) bool { } } +// enqueueInterrupt 把一条已通过策略/@ 检查的消息并入待投批次，并重置 debounce 计时。 +// +// batchWindow<=0 时退回逐条投递（合并前行为）。 +func (p *Plugin) enqueueInterrupt(msgType string, userID, groupID, messageID int64, nickname, single string, owner, highRisk bool) { + if p.sdk == nil && p.injectHook == nil { + return + } + if p.batchWindow <= 0 { + p.injectInterrupt(single) + return + } + key := qqBatchKey(msgType, groupID, userID) + p.batchMu.Lock() + if p.batches == nil { + p.batches = make(map[string]*pendingBatch) + } + b := p.batches[key] + if b == nil { + b = &pendingBatch{key: key, first: time.Now()} + p.batches[key] = b + } + b.isGroup = msgType == "group" + b.userID, b.groupID, b.nickname = userID, groupID, nickname + b.msgIDs = append(b.msgIDs, messageID) + b.single = single + b.owner = b.owner || owner + b.highRisk = b.highRisk || highRisk + // debounce：每来一条就推迟；但整体不超过 batchMax（否则持续刷屏会一直不投）。 + delay := p.batchWindow + if p.batchMax > 0 { + if remain := p.batchMax - time.Since(b.first); remain < delay { + delay = remain + } + } + if delay < 0 { + delay = 0 + } + if b.timer != nil { + b.timer.Stop() + } + b.timer = time.AfterFunc(delay, func() { p.flushBatch(key) }) + p.batchMu.Unlock() +} + +// injectInterrupt 投递一条中断提示（NoMemory：HTTP 侧来的不是对话内容； +// Priority L1：QQ 消息是低级别中断，既非实时工作也非紧急工作，完全可等）。 +func (p *Plugin) injectInterrupt(text string) { + if text == "" { + return + } + if p.injectHook != nil { + p.injectHook(text) + return + } + if p.sdk == nil { + return + } + p.sdk.InjectInterruptTextOpts(p.name, p.name, text, sdk.InjectOptions{ + NoMemory: true, + Priority: sdk.PriorityL1, + }) +} + +// flushBatch 投递一批：n==1 沿用单条原文；n>1 生成「共几条」的合并中断。 +func (p *Plugin) flushBatch(key string) { + p.batchMu.Lock() + b := p.batches[key] + delete(p.batches, key) + p.batchMu.Unlock() + if b == nil { + return + } + text := b.single + if len(b.msgIDs) > 1 { + text = p.buildBatchInterrupt(b) + } + p.injectInterrupt(text) +} + +// flushAllBatches 停机前把未到点的批次立刻投出去（best effort）。 +func (p *Plugin) flushAllBatches() { + p.batchMu.Lock() + keys := make([]string, 0, len(p.batches)) + for k := range p.batches { + keys = append(keys, k) + } + p.batchMu.Unlock() + for _, k := range keys { + p.flushBatch(k) + } +} + +// buildBatchInterrupt 生成合并中断：说清「一共几条」「分别是哪些 message_id」， +// 并给出一次拿全上下文的建议（get_history），避免模型逐条 get_message。 +func (p *Plugin) buildBatchInterrupt(b *pendingBatch) string { + tp := p.name + "_" + outputTool := "output_send__" + p.name + n := len(b.msgIDs) + ids := formatMsgIDs(b.msgIDs) + var s string + if b.isGroup { + s = fmt.Sprintf("来自「%s」在群里短时间内连续发来 %d 条消息(message_id=%s)。建议先用%sget_history(group_id=%d, count=%d)一次拉取这几条上下文再统一回复；也可用%sget_message 取单条。用%s回复群聊", + b.nickname, n, ids, tp, b.groupID, n+5, tp, outputTool) + } else { + s = fmt.Sprintf("来自「%s」的私聊短时间内连续发来 %d 条消息(message_id=%s, user_id=%d)。建议先用%sget_history(user_id=%d, count=%d)一次拉取这几条上下文再统一回复；也可用%sget_message 取单条。用%s回复对方", + b.nickname, n, ids, b.userID, tp, b.userID, n+5, tp, outputTool) + } + if b.highRisk { + s = "【⚠️ 高危信息，谨慎处理】" + s + } + if b.owner { + s = "【重要！Bot 所有者消息】" + s + } + return s +} + +// formatMsgIDs 把 message_id 列表压成一行；过多时截断，避免中断文字过长。 +func formatMsgIDs(ids []int64) string { + const capN = 12 + parts := make([]string, 0, len(ids)+1) + for i, id := range ids { + if i >= capN { + parts = append(parts, "…") + break + } + parts = append(parts, strconv.FormatInt(id, 10)) + } + return strings.Join(parts, ",") +} + func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { if r.Method != "POST" { http.Error(w, "", http.StatusMethodNotAllowed) @@ -1417,7 +1596,9 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) return } + highRisk := false if highRiskRe.MatchString(text) { + highRisk = true interrupt = "【⚠️ 高危信息，谨慎处理】" + interrupt } @@ -1446,15 +1627,9 @@ func (p *Plugin) handleWebhook(w http.ResponseWriter, r *http.Request) { p.startTyping(evt.UserID) } - if p.sdk != nil { - // NoMemory：HTTP 侧来的中断提示，不是对话内容。 - // Priority：QQ 消息是**低级别中断**——既不是时钟那样的实时工作， - // 也不是紧急工作，所以声明 L1（完全可等）。 - p.sdk.InjectInterruptTextOpts(p.name, p.name, interrupt, sdk.InjectOptions{ - NoMemory: true, - Priority: sdk.PriorityL1, - }) - } + // 合并投递：同一会话同一发送者在 batchWindow 内的连续消息并成一次中断。 + p.enqueueInterrupt(evt.MessageType, evt.UserID, evt.GroupID, evt.MessageID, nickname, interrupt, p.isOwner(evt.UserID), highRisk) + w.WriteHeader(http.StatusOK) } diff --git a/example/qq/plugin_test.go b/example/qq/plugin_test.go index 76e8cde..04828b1 100644 --- a/example/qq/plugin_test.go +++ b/example/qq/plugin_test.go @@ -7,6 +7,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" "gitcode.com/JianFeeeee/homeagent-sdk/sdk" ) @@ -235,3 +236,79 @@ func TestDowngradedAuthStillAllowsQQOutput(t *testing.T) { t.Fatalf("读取类工具在降权时应被当前会话限制挡住: %#v", ctx2.Response) } } + +// ---- 消息合并（debounce）---- + +// collectInterrupts 用注入钩子收集中断文本（避免测试依赖真实 SDK）。 +func collectInterrupts(p *Plugin) *[]string { + got := []string{} + p.injectHook = func(s string) { got = append(got, s) } + return &got +} + +func TestConsecutiveMessagesFromSameSenderAreBatched(t *testing.T) { + p := newPermissionTestPlugin(t) + got := collectInterrupts(p) + p.batchWindow = 20 * time.Millisecond + p.batchMax = time.Second + + for i := 0; i < 3; i++ { + p.enqueueInterrupt("private", 10001, 0, int64(100+i), "小明", "单条", false, false) + } + time.Sleep(120 * time.Millisecond) + + if len(*got) != 1 { + t.Fatalf("同一发送者连发 3 条应合并成 1 次中断，实际 %d 次: %#v", len(*got), *got) + } + if !strings.Contains((*got)[0], "3 条消息") { + t.Fatalf("合并中断应说明一共几条，实际: %s", (*got)[0]) + } + // 三个 message_id 都要带上，模型才能取全 + for _, id := range []string{"100", "101", "102"} { + if !strings.Contains((*got)[0], id) { + t.Fatalf("合并中断漏了 message_id=%s: %s", id, (*got)[0]) + } + } +} + +func TestDifferentSendersAreNotBatchedTogether(t *testing.T) { + p := newPermissionTestPlugin(t) + got := collectInterrupts(p) + p.batchWindow = 20 * time.Millisecond + p.batchMax = time.Second + + p.enqueueInterrupt("private", 10001, 0, 1, "小明", "a", false, false) + p.enqueueInterrupt("private", 10002, 0, 2, "小红", "b", false, false) + time.Sleep(120 * time.Millisecond) + + if len(*got) != 2 { + t.Fatalf("不同发送者不该合并，应有 2 次中断，实际 %d: %#v", len(*got), *got) + } +} + +func TestBatchWindowZeroFallsBackToPerMessage(t *testing.T) { + p := newPermissionTestPlugin(t) + got := collectInterrupts(p) + p.batchWindow = 0 + + for i := 0; i < 3; i++ { + p.enqueueInterrupt("private", 10001, 0, int64(i), "小明", "原文", false, false) + } + if len(*got) != 3 { + t.Fatalf("关闭合并时应逐条投递（3 次），实际 %d: %#v", len(*got), *got) + } +} + +func TestSingleMessageKeepsOriginalText(t *testing.T) { + p := newPermissionTestPlugin(t) + got := collectInterrupts(p) + p.batchWindow = 20 * time.Millisecond + p.batchMax = time.Second + + p.enqueueInterrupt("group", 10001, 20002, 7, "小明", "单条原文", true, false) + time.Sleep(120 * time.Millisecond) + + if len(*got) != 1 || (*got)[0] != "单条原文" { + t.Fatalf("单条消息应沿用原文（含所有者前缀），实际 %#v", *got) + } +} diff --git a/example/vikunja/go.mod b/example/vikunja/go.mod index 2ba8ae8..ed39de3 100644 --- a/example/vikunja/go.mod +++ b/example/vikunja/go.mod @@ -2,7 +2,7 @@ module vikunja-plugin go 1.25.0 -require gitcode.com/JianFeeeee/homeagent-sdk v1.2.0 +require gitcode.com/JianFeeeee/homeagent-sdk v0.0.0 // 与同目录其它示例一致：SDK 指向仓库内的 vendored 副本 @@ -12,4 +12,4 @@ require gitcode.com/JianFeeeee/homeagent-sdk v1.2.0 -replace gitcode.com/JianFeeeee/homeagent-sdk => /root/.homeagent/hmapdev/sdk/v1.2.0 +replace gitcode.com/JianFeeeee/homeagent-sdk => ../../ diff --git a/example/vikunja/plg.json b/example/vikunja/plg.json index c3ac6dc..5c32f62 100644 --- a/example/vikunja/plg.json +++ b/example/vikunja/plg.json @@ -6,7 +6,12 @@ "description": "Vikunja 待办/任务管理：任务增删改查、项目与看板桶、标签、指派、评论、关联、附件、保存筛选器、团队与分享、通知、订阅、Webhook、时间跟踪、数据导入、实例管理；并附通用 API 直通工具兜底", "author": "HomeAgent", "entry": "plugin.bin", - "sdk": "1.2.0", - "tags": ["vikunja", "todo", "task", "gtd", "productivity"], + "tags": [ + "vikunja", + "todo", + "task", + "gtd", + "productivity" + ], "targets": "linux/amd64" -} +} \ No newline at end of file