From 76e648aac3ac3b8def2223554ddb6152a6ceb423 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 28 Jun 2026 16:58:55 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=9B=9E=E9=80=80thinker.go=E5=88=B0649?= =?UTF-8?q?0c34=E7=A8=B3=E5=AE=9A=E7=89=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ai-core/internal/background/thinker.go | 98 +++++++------------ 1 file changed, 33 insertions(+), 65 deletions(-) diff --git a/backend/ai-core/internal/background/thinker.go b/backend/ai-core/internal/background/thinker.go index d8ba2d6..c8ed50f 100644 --- a/backend/ai-core/internal/background/thinker.go +++ b/backend/ai-core/internal/background/thinker.go @@ -292,31 +292,27 @@ func timePeriod(now time.Time) (string, string) { // SetMessagePusher 设置主动消息推送回调 // SetScheduleLoader sets the dynamic schedule loader for interval calculation. func (t *Thinker) SetScheduleLoader(loader *ScheduleLoader) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.scheduleLoader = loader } func (t *Thinker) SetMessagePusher(pusher func(string, string, string)) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.messagePusher = pusher } // SetPlatformMessagePusher sets the callback for pushing proactive messages to platform adapters (OBv11, etc.). func (t *Thinker) SetPlatformMessagePusher(pusher func(ProactiveTarget, string)) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.platformMessagePusher = pusher } // SetBotUID sets the bot's own platform UID (e.g., OBv11 account). func (t *Thinker) SetBotUID(platform, uid string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() if t.botUIDs == nil { t.botUIDs = make(map[string]string) @@ -328,8 +324,7 @@ func (t *Thinker) SetBotUID(platform, uid string) { // AddOrUpdatePlatformChannel adds or updates a platform channel with resolved display name. func (t *Thinker) AddOrUpdatePlatformChannel(platform, channelType, channelID, channelName, adapterID, adapterName string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() for i, ch := range t.platformChannels { @@ -359,8 +354,7 @@ func (t *Thinker) AddOrUpdatePlatformChannel(platform, channelType, channelID, c // SetEmotionTracker sets the emotion tracker. func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.emotionTracker = et } @@ -368,8 +362,7 @@ func (t *Thinker) SetEmotionTracker(et *persona.EmotionTracker) { // UpdatePresence updates the user online status. // Called by the ai-core presence endpoint when gateway detects connect/disconnect. func (t *Thinker) UpdatePresence(online bool, sessionID string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() wasOffline := !t.isUserOnline() t.setUserOnline(online) @@ -397,8 +390,7 @@ func (t *Thinker) UpdatePresence(online bool, sessionID string) { // TriggerReminderMessage pushes a reminder message generated by LLM to the user. // Called by the internal reminder-trigger endpoint when a reminder fires. func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() pusher := t.messagePusher pushSessionID := sessionID @@ -418,8 +410,7 @@ func (t *Thinker) TriggerReminderMessage(userID, sessionID, message string) { // PushPlatformMessage pushes a message to a platform channel via the platform pusher. func (t *Thinker) PushPlatformMessage(target ProactiveTarget, message string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() pusher := t.platformMessagePusher @@ -433,8 +424,7 @@ func (t *Thinker) PushPlatformMessage(target ProactiveTarget, message string) { // IsUserRecentlyActive returns true if the user has been active within the given duration. func (t *Thinker) IsUserRecentlyActive(d time.Duration) bool { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() return time.Since(t.lastUserTime()) < d } @@ -521,7 +511,7 @@ func NewThinker( lightThinkEnabled: cfg.LightThinkEnabled, lightThinkInterval: cfg.LightThinkInterval, silenceTimeout: cfg.SilenceTimeout, - proactiveMsgMinGap: getEnvDuration("PROACTIVE_MSG_MIN_GAP_SEC", 60), + proactiveMsgMinGap: getEnvDuration("PROACTIVE_MSG_MIN_GAP_SEC", 300), postChatDelay: cfg.PostChatDelay, minThinkGap: cfg.MinThinkGap, offlineThinkGap: cfg.OfflineThinkGap, @@ -567,8 +557,7 @@ func (t *Thinker) restoreContext() { } } if !latest.IsZero() && time.Since(latest) < 24*time.Hour { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.lastUserMessage = latest t.setLastUser(latest) @@ -683,8 +672,7 @@ func (t *Thinker) TriggerPostChatThink() { return } - if !t.muLock() { - return + t.muLock() defer t.muUnlock() canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -754,8 +742,7 @@ func (t *Thinker) resetSilenceTimer() { return case <-t.silenceTimer.C: // 再次检查：用户是否真的沉默了足够久 - if !t.muLock() { - return + t.muLock() defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) canThink := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -873,8 +860,7 @@ func (t *Thinker) performPlatformObservation() { } observationContent := fmt.Sprintf("[平台观察 %s]\n%s", time.Now().In(t.timeLocation).Format("15:04"), result.Summary) - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: observationContent, @@ -907,8 +893,7 @@ func (t *Thinker) lightThinkLoop() { log.Println("[后台思考] 轻量思考已停止") return case <-time.After(t.lightThinkInterval): - if !t.muLock() { - return + t.muLock() defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) sinceLastThink := time.Since(t.lastThinkTimeAtomic()) @@ -970,8 +955,7 @@ func (t *Thinker) performLightThink() { // Check if deep think should be woken. if strings.Contains(content, "【需要深思】") { log.Println("[轻量思考] 检测到【需要深思】，唤醒深度思考...") - if !t.muLock() { - return + t.muLock() defer t.muUnlock() canDeep := time.Since(t.lastThinkTimeAtomic()) >= t.minThinkGap @@ -987,8 +971,7 @@ func (t *Thinker) performLightThink() { topic := strings.TrimSpace(content[idx+len("【话题发起】"):]) if topic != "" { log.Printf("[轻量思考] 话题发起: %s", topic) - if !t.muLock() { - return + t.muLock() defer t.muUnlock() pusher := t.messagePusher sessionID := t.activeSessionID @@ -1066,8 +1049,7 @@ func (t *Thinker) periodicThinkLoop() { // GetPendingThoughts 获取并消费所有待处理的后台思考 func (t *Thinker) GetPendingThoughts() []*PendingThought { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() if len(t.pendingThoughts) == 0 { @@ -1085,8 +1067,7 @@ func (t *Thinker) GetPendingThoughts() []*PendingThought { // HasPendingThoughts 检查是否有待处理的思考 func (t *Thinker) HasPendingThoughts() bool { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() return len(t.pendingThoughts) > 0 } @@ -1098,8 +1079,7 @@ func (t *Thinker) HasPendingThoughts() bool { // 防御性速率限制：即使调用方未检查 minThinkGap，performThink 自身也会 // 强制执行最小间隔，防止并发调用或 bug 导致 LLM 配额被快速消耗。 func (t *Thinker) performThink(triggerReason string) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() gapSinceLast := time.Since(t.lastThinkTimeAtomic()) minGap := t.minThinkGap @@ -1120,16 +1100,14 @@ func (t *Thinker) performThink(triggerReason string) { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.thinkCancel = cancel // 0. 让步于前台——如果用户最近有活动（非post_chat），跳过本次思考。 if triggerReason != "post_chat" { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) @@ -1143,8 +1121,7 @@ func (t *Thinker) performThink(triggerReason string) { // 0. 让步于前台——如果用户最近有活动（非post_chat），跳过本次思考。 if triggerReason != "post_chat" { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() sinceLastUser := time.Since(t.lastUserTime()) @@ -1164,8 +1141,7 @@ func (t *Thinker) performThink(triggerReason string) { // 2. 获取当前活跃会话的对话历史（优先活跃会话，回退到管理员主会话） var convHistory []model.LLMMessage if t.convStore != nil { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() sessionID := t.activeSessionID if sessionID == "" { @@ -1283,8 +1259,7 @@ func (t *Thinker) performThink(triggerReason string) { // 4.5 获取最近平台观察（定期触发和对话后触发时注入） var platformObservation string if triggerReason == "periodic" || triggerReason == "post_chat" { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() for i := len(t.pendingThoughts) - 1; i >= 0; i-- { if strings.HasPrefix(t.pendingThoughts[i].Content, "[平台观察") { @@ -1435,8 +1410,7 @@ func (t *Thinker) performThink(triggerReason string) { if len(parts) == 2 { adapterName, groupID := parts[0], parts[1] // Find the channel to get the correct platform type. - if !t.muLock() { - return + t.muLock() defer t.muUnlock() var platform string for _, ch := range t.platformChannels { @@ -1690,8 +1664,7 @@ func (t *Thinker) buildThinkingUserPrompt( case "post_chat": sb.WriteString("刚有人和你聊完天。你想自然地在心里回味一下刚才的对话……\n") case "silence": - if !t.muLock() { - return + t.muLock() defer t.muUnlock() silenceDuration := time.Since(t.lastUserTime()) @@ -1791,8 +1764,7 @@ func (t *Thinker) buildThinkingUserPrompt( } // OBv11 platform identity and available channels for proactive messaging. - if !t.muLock() { - return + t.muLock() defer t.muUnlock() qqChannels := t.platformChannels botUIDs := t.botUIDs @@ -1941,8 +1913,7 @@ func (t *Thinker) buildOpenAITools() []llm.OpenAITool { // storeThought 存储思考结果到待推送队列，并异步持久化到 memory-service func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCount int) { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: content, @@ -2118,8 +2089,7 @@ func (t *Thinker) extractProactiveMessage(content string) (string, *ProactiveTar channelID = "private_" + m[2] } adapterName := platform // fallback to format key - if !t.muLock() { - return + t.muLock() defer t.muUnlock() for _, ch := range t.platformChannels { if ch.Platform == platform && ch.ChannelType == chatType && ch.ChannelID == channelID { @@ -2711,8 +2681,7 @@ func (t *Thinker) setLastProactive(ts time.Time) { t.lastProactiveNs.Store(ts.Un func (t *Thinker) DeadlockDetected(timeout time.Duration) bool { done := make(chan struct{}) go func() { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() close(done) @@ -2738,8 +2707,7 @@ func (t *Thinker) StartDeadlockMonitor() { for range ticker.C { done := make(chan struct{}) go func() { - if !t.muLock() { - return + t.muLock() defer t.muUnlock() close(done)