feat: 消息并行处理 + QQ卡片完整解析 + 视觉OCR融合格式修复
- platform-bridge: 8-worker per-session 并行分发，同会话保序跨会话并行 - platform-bridge: 静默消息 fire-and-forget，不阻塞同用户后续消息 - QQ卡片: html.UnescapeString 解码 NapCat HTML实体，正确解析卡片JSON - QQ卡片: 输出含应用名/简介/来源/封面URL，封面注入图片管线走视觉 - ai-core: 视觉+OCR结果融合为单句，单图不编号，避免LLM误解为多张图 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -5,10 +5,12 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
@@ -171,16 +173,38 @@ func main() {
|
||||
}
|
||||
groupSessionID := fmt.Sprintf("platform_%s_%s", msg.Platform, msg.ChannelID)
|
||||
|
||||
// Helper: fire-and-forget a silent observation. Silent paths don't
|
||||
// produce visible responses, so we can background them to avoid
|
||||
// blocking the worker for the next message from the same user.
|
||||
fireSilent := func(namespace string, imgs, vids, voices []string) {
|
||||
go func() {
|
||||
_, err := forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, imgs, vids, voices, isAdmin)
|
||||
if err != nil {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
Success: false,
|
||||
Error: err.Error(),
|
||||
})
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
switch {
|
||||
case isMessageHistorical(msg, router):
|
||||
msg.RouteType = "silent"
|
||||
namespace := buildMemoryNamespace(msg.Platform, msg.ChannelType, msg.ChannelID)
|
||||
response, routeErr = forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, nil, videoURLs, voiceURLs, isAdmin)
|
||||
fireSilent(namespace, nil, videoURLs, voiceURLs)
|
||||
response = &bridge.UnifiedResponse{Messages: []bridge.ResponseMessage{{DisplayType: "silent"}}, Platform: msg.Platform}
|
||||
|
||||
case isAdmin && !isBotMentioned && shouldAdminBeSilent(msg, router):
|
||||
msg.RouteType = "silent"
|
||||
namespace := buildMemoryNamespace(msg.Platform, msg.ChannelType, msg.ChannelID)
|
||||
response, routeErr = forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, imageURLs, videoURLs, voiceURLs, isAdmin)
|
||||
fireSilent(namespace, imageURLs, videoURLs, voiceURLs)
|
||||
response = &bridge.UnifiedResponse{Messages: []bridge.ResponseMessage{{DisplayType: "silent"}}, Platform: msg.Platform}
|
||||
|
||||
case isAdmin:
|
||||
msg.RouteType = "normal"
|
||||
@@ -191,30 +215,16 @@ func main() {
|
||||
response, routeErr = forwardToAICore(cfg, msg, "text", chatUserID, groupSessionID, imageURLs, videoURLs, voiceURLs, isAdmin)
|
||||
|
||||
case isMentioned:
|
||||
// Non-admin user mentioned an admin. Don't respond in channel —
|
||||
// the admin already gets QQ's native @notification. Observe silently.
|
||||
msg.RouteType = "silent"
|
||||
namespace := buildMemoryNamespace(msg.Platform, msg.ChannelType, msg.ChannelID)
|
||||
response, routeErr = forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, imageURLs, videoURLs, voiceURLs, isAdmin)
|
||||
fireSilent(namespace, imageURLs, videoURLs, voiceURLs)
|
||||
response = &bridge.UnifiedResponse{Messages: []bridge.ResponseMessage{{DisplayType: "silent"}}, Platform: msg.Platform}
|
||||
|
||||
case isSilent:
|
||||
msg.RouteType = "silent"
|
||||
namespace := buildMemoryNamespace(msg.Platform, msg.ChannelType, msg.ChannelID)
|
||||
silentResponse, silentErr := forwardToAICore(cfg, msg, "platform_silent", namespace, namespace, imageURLs, videoURLs, voiceURLs, isAdmin)
|
||||
if silentErr != nil {
|
||||
msgLogger.Log(logging.LogEntry{
|
||||
Timestamp: time.Now(),
|
||||
Direction: "outgoing",
|
||||
Platform: msg.Platform,
|
||||
ChannelID: msg.ChannelID,
|
||||
SenderID: msg.OriginalSenderUID,
|
||||
Success: false,
|
||||
Error: silentErr.Error(),
|
||||
})
|
||||
return nil, silentErr
|
||||
}
|
||||
response = silentResponse
|
||||
routeErr = nil
|
||||
fireSilent(namespace, imageURLs, videoURLs, voiceURLs)
|
||||
response = &bridge.UnifiedResponse{Messages: []bridge.ResponseMessage{{DisplayType: "silent"}}, Platform: msg.Platform}
|
||||
|
||||
default:
|
||||
msg.RouteType = "normal"
|
||||
@@ -367,62 +377,86 @@ func startQQReaders(router *bridge.PlatformRouter) {
|
||||
qqReaderCancelsMu.Unlock()
|
||||
|
||||
adapter := qqAdapter // capture for goroutine
|
||||
qqMsgCh := make(chan *qqadapter.OBv11Message, 100)
|
||||
go adapter.ReadMessages(ctx, qqMsgCh)
|
||||
const numWorkers = 8
|
||||
|
||||
// Single reader pumps raw messages from the adapter.
|
||||
rawCh := make(chan *qqadapter.OBv11Message, 100)
|
||||
go adapter.ReadMessages(ctx, rawCh)
|
||||
|
||||
// Per-worker channels.
|
||||
var workerChs [numWorkers]chan *qqadapter.OBv11Message
|
||||
for i := 0; i < numWorkers; i++ {
|
||||
workerChs[i] = make(chan *qqadapter.OBv11Message, 50)
|
||||
}
|
||||
|
||||
// Dispatcher: route to worker by session hash so same conversation stays ordered.
|
||||
go func() {
|
||||
for msg := range qqMsgCh {
|
||||
response, err := router.RouteMessage(adapterKey, msg)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq:%s] route error: %v\n", adapterKey, err)
|
||||
continue
|
||||
}
|
||||
if response != nil && len(response.Messages) > 0 && !hasOnlySilentMessages(response.Messages) {
|
||||
messageType := msg.MessageType
|
||||
userID := msg.UserID
|
||||
groupID := msg.GroupID
|
||||
// Filter non-empty messages.
|
||||
var toSend []bridge.ResponseMessage
|
||||
for _, rm := range response.Messages {
|
||||
if rm.Content != "" {
|
||||
rm.Content = qqadapter.ConvertMarkdownToQQ(rm.Content)
|
||||
toSend = append(toSend, rm)
|
||||
}
|
||||
}
|
||||
interval := time.Duration(adapter.SendIntervalMs()) * time.Millisecond
|
||||
if interval <= 0 {
|
||||
interval = 2 * time.Second
|
||||
}
|
||||
for i, rm := range toSend {
|
||||
if i > 0 && interval > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
// Re-acquire current adapter for hot-reload safety.
|
||||
cur, err := router.GetAdapter(adapterKey)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
curQQ, ok := cur.(*qqadapter.Adapter)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var sendErr error
|
||||
switch messageType {
|
||||
case "private":
|
||||
sendErr = curQQ.SendMessage("private", userID, 0, rm.Content)
|
||||
case "group":
|
||||
sendErr = curQQ.SendMessage("group", 0, groupID, rm.Content)
|
||||
}
|
||||
if sendErr != nil {
|
||||
fmt.Printf("[qq:%s] send msg error: %v\n", adapterKey, sendErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
for msg := range rawCh {
|
||||
idx := hashQQSession(msg) % numWorkers
|
||||
workerChs[idx] <- msg
|
||||
}
|
||||
for i := 0; i < numWorkers; i++ {
|
||||
close(workerChs[i])
|
||||
}
|
||||
}()
|
||||
|
||||
// Worker pool: each worker processes messages sequentially; workers run in parallel.
|
||||
for _, wch := range workerChs {
|
||||
wch := wch // capture
|
||||
go func() {
|
||||
for msg := range wch {
|
||||
response, err := router.RouteMessage(adapterKey, msg)
|
||||
if err != nil {
|
||||
fmt.Printf("[qq:%s] route error: %v\n", adapterKey, err)
|
||||
continue
|
||||
}
|
||||
if response != nil && len(response.Messages) > 0 && !hasOnlySilentMessages(response.Messages) {
|
||||
messageType := msg.MessageType
|
||||
userID := msg.UserID
|
||||
groupID := msg.GroupID
|
||||
// Filter non-empty messages.
|
||||
var toSend []bridge.ResponseMessage
|
||||
for _, rm := range response.Messages {
|
||||
if rm.Content != "" {
|
||||
rm.Content = qqadapter.ConvertMarkdownToQQ(rm.Content)
|
||||
toSend = append(toSend, rm)
|
||||
}
|
||||
}
|
||||
interval := time.Duration(adapter.SendIntervalMs()) * time.Millisecond
|
||||
if interval <= 0 {
|
||||
interval = 2 * time.Second
|
||||
}
|
||||
for i, rm := range toSend {
|
||||
if i > 0 && interval > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
cur, err := router.GetAdapter(adapterKey)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
curQQ, ok := cur.(*qqadapter.Adapter)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var sendErr error
|
||||
switch messageType {
|
||||
case "private":
|
||||
sendErr = curQQ.SendMessage("private", userID, 0, rm.Content)
|
||||
case "group":
|
||||
sendErr = curQQ.SendMessage("group", 0, groupID, rm.Content)
|
||||
}
|
||||
if sendErr != nil {
|
||||
fmt.Printf("[qq:%s] send msg error: %v\n", adapterKey, sendErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -892,6 +926,25 @@ func hasOnlySilentMessages(messages []bridge.ResponseMessage) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
// hashQQSession returns a hash for dispatching a QQ message to a worker.
|
||||
// Messages from the same conversation (private or group) get the same hash,
|
||||
// preserving ordering within a session while allowing cross-session parallelism.
|
||||
func hashQQSession(msg *qqadapter.OBv11Message) uint32 {
|
||||
h := fnv.New32a()
|
||||
h.Write([]byte(msg.MessageType))
|
||||
h.Write([]byte(":"))
|
||||
if msg.MessageType == "group" {
|
||||
// Hash by group+user: same user's messages in a group stay ordered,
|
||||
// but different users in the same group can be processed in parallel.
|
||||
h.Write([]byte(strconv.FormatInt(msg.GroupID, 10)))
|
||||
h.Write([]byte(":"))
|
||||
h.Write([]byte(strconv.FormatInt(msg.UserID, 10)))
|
||||
} else {
|
||||
h.Write([]byte(strconv.FormatInt(msg.UserID, 10)))
|
||||
}
|
||||
return h.Sum32()
|
||||
}
|
||||
|
||||
func parseIntOr(s string, defaultVal int) int {
|
||||
n := 0
|
||||
for _, c := range s {
|
||||
|
||||
Reference in New Issue
Block a user