AskaEth/Cyrene
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: Phase 1+2 架构进化 — 连续思考链/主动消息决策/情感状态机/离线自主思考 (86文件)

Browse Source 
Phase 1 (基础设施):
- ThinkChain 思考链连续性 + 差异化思考提示词 (persistent)
- AutonomousToolPolicy 工具安全策略 (safe/unsafe/conditional)
- MessageScheduler 自适应消息节奏 (Idle/Available/Busy)
- SessionEnrichmentStore 渐进式上下文丰富 (5层)
- ConversationBus 事件总线 + ResponseCache (dedup)
- pkg/logger 统一日志 + 所有 handler 替换 fmt.Printf
- NPE 守卫/链路优化/数据库表修复/Go workspace

Phase 2 (人格交互):
- EmotionState/EmotionTracker 情感状态机 (5种心情, 情绪衰减)
- ProactiveGuard 主动消息多维决策 (静默时段/紧急度/频率/校验)
- Gateway↔ai-core 在线状态感知链路 (presence notification)
- 离线思考频率控制 + 重连问候 + 离线消息排队

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
AskaEth
2026-05-23 15:25:12 +08:00
parent b123a36aae
commit 87214b9441
86 changed files with 3085 additions and 582 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/gateway/internal/ws/hub.go
+93 -15
View File
@@ -3,9 +3,10 @@ package ws
import (
"encoding/json"
"fmt"
"log"
"github.com/yourname/cyrene-ai/pkg/logger"
"net/http"
"os"
"strings"
"sync"
"time"
@@ -70,6 +71,11 @@ type Hub struct {
// 闲置超时时间
idleTimeout time.Duration
// Phase 2: 离线主动消息队列 + 在线状态通知
pendingProactive map[string][]json.RawMessage // userID -> queued messages
aiCoreURL string
internalToken string
}
// SetStore 设置持久化存储 (可选)
@@ -93,6 +99,7 @@ func NewHub() *Hub {
sessions: make(map[string]*SessionState),
iotStopCh: make(chan struct{}),
idleTimeout: 30 * time.Minute, // 默认30分钟
pendingProactive: make(map[string][]json.RawMessage),
}
}
@@ -102,7 +109,7 @@ func (h *Hub) StartIdleCleanup() {
go func() {
defer func() {
if r := recover(); r != nil {
log.Printf("[WS] 闲置会话清理 panic 恢复: %v", r)
logger.Printf("[WS] 闲置会话清理 panic 恢复: %v", r)
}
}()
ticker := time.NewTicker(5 * time.Minute)
@@ -112,7 +119,7 @@ func (h *Hub) StartIdleCleanup() {
h.cleanupIdleSessions()
}
}()
log.Printf("[WS] 闲置会话清理已启动 (超时: %v)", h.idleTimeout)
logger.Printf("[WS] 闲置会话清理已启动 (超时: %v)", h.idleTimeout)
}
// cleanupIdleSessions 标记超时会话为 idle（不删除状态）
@@ -148,7 +155,7 @@ func (h *Hub) cleanupIdleSessions() {
}
if idleCount > 0 {
log.Printf("[WS] 闲置清理: %d 个会话标记为 idle", idleCount)
logger.Printf("[WS] 闲置清理: %d 个会话标记为 idle", idleCount)
}
}
@@ -170,6 +177,58 @@ func (h *Hub) GetAllActiveSessions() []*SessionState {
return result
}
// SetAICoreConfig sets the ai-core URL and internal token for presence notifications.
func (h *Hub) SetAICoreConfig(url, token string) {
h.aiCoreURL = url
h.internalToken = token
}
// QueueProactiveMessage queues a proactive message for offline delivery.
func (h *Hub) QueueProactiveMessage(userID string, msg json.RawMessage) {
h.mu.Lock()
defer h.mu.Unlock()
h.pendingProactive[userID] = append(h.pendingProactive[userID], msg)
// Keep only the most recent 3 messages
if len(h.pendingProactive[userID]) > 3 {
h.pendingProactive[userID] = h.pendingProactive[userID][1:]
}
}
// FlushPendingProactive returns and clears queued proactive messages for a user.
func (h *Hub) FlushPendingProactive(userID string) []json.RawMessage {
h.mu.Lock()
defer h.mu.Unlock()
msgs := h.pendingProactive[userID]
delete(h.pendingProactive, userID)
return msgs
}
// notifyAICorePresence sends a presence update to ai-core.
func (h *Hub) notifyAICorePresence(userID, status, sessionID string) {
if h.aiCoreURL == "" || h.internalToken == "" {
return
}
body, _ := json.Marshal(map[string]string{
"user_id": userID,
"status": status,
"session_id": sessionID,
"timestamp": fmt.Sprintf("%d", time.Now().Unix()),
})
go func() {
req, _ := http.NewRequest("POST", h.aiCoreURL+"/api/v1/internal/presence", strings.NewReader(string(body)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Internal-Token", h.internalToken)
client := &http.Client{Timeout: 5 * time.Second}
resp, err := client.Do(req)
if err != nil {
logger.Printf("[presence] 通知 ai-core 失败: %v", err)
return
}
resp.Body.Close()
logger.Printf("[presence] 通知 ai-core: user=%s status=%s", userID, status)
}()
}
// Run 启动Hub主循环
func (h *Hub) Run() {
for {
@@ -195,10 +254,29 @@ func (h *Hub) Run() {
MessageCount: 0,
}
}
h.mu.Unlock()
// Phase 2: 检测是否为重连 (之前处于离线状态)
wasOffline := len(h.userClients[client.UserID]) == 1 // 刚加入，之前为0
h.mu.Unlock()
log.Printf("[WS] 客户端连接: user=%s session=%s (当前连接数: %d)",
client.UserID, client.SessionID, len(h.clients))
// 重连后推送积压的主动消息
if wasOffline {
pending := h.FlushPendingProactive(client.UserID)
if len(pending) > 0 {
logger.Printf("[proactive] 推送 %d 条积压消息给重连用户 %s", len(pending), client.UserID)
// 只推送最新的一条
go func() {
// small delay for WS connection to stabilize
time.Sleep(500 * time.Millisecond)
h.SendToUser(client.UserID, pending[len(pending)-1])
}()
}
}
// 通知 ai-core 用户上线
h.notifyAICorePresence(client.UserID, "online", client.SessionID)
logger.Printf("[WS] 客户端连接: user=%s session=%s (当前连接数: %d)",
client.UserID, client.SessionID, len(h.clients))
case client := <-h.unregister:
h.mu.Lock()
@@ -233,7 +311,7 @@ func (h *Hub) Run() {
}
h.mu.Unlock()
log.Printf("[WS] 客户端断开: user=%s session=%s (当前连接数: %d)",
logger.Printf("[WS] 客户端断开: user=%s session=%s (当前连接数: %d)",
client.UserID, client.SessionID, len(h.clients))
case message := <-h.broadcast:
@@ -287,7 +365,7 @@ func (h *Hub) Run() {
}
h.mu.Unlock()
log.Printf("[WS] 广播清理 %d 个失效客户端 (当前连接数: %d)",
logger.Printf("[WS] 广播清理 %d 个失效客户端 (当前连接数: %d)",
len(staleClients), len(h.clients))
}
}
@@ -504,12 +582,12 @@ func (h *Hub) StartIoTBroadcast(iotServiceURL string) {
go func() {
defer func() {
if r := recover(); r != nil {
log.Printf("[IoT广播] 轮询循环 panic 恢复: %v", r)
logger.Printf("[IoT广播] 轮询循环 panic 恢复: %v", r)
}
}()
h.iotPollLoop()
}()
log.Printf("[IoT广播] 已启动 (IoT服务地址: %s)", iotServiceURL)
logger.Printf("[IoT广播] 已启动 (IoT服务地址: %s)", iotServiceURL)
}
// StopIoTBroadcast 停止 IoT 设备广播
@@ -522,7 +600,7 @@ func (h *Hub) StopIoTBroadcast() {
}
close(h.iotStopCh)
h.iotPollRunning = false
log.Println("[IoT广播] 已停止")
logger.Println("[IoT广播] 已停止")
}
// iotPollLoop IoT 设备轮询循环
@@ -563,7 +641,7 @@ func (h *Hub) pollAndBroadcastIoT() {
devices, err := fetchIoTDevices(url)
if err != nil {
log.Printf("[IoT广播] 获取设备失败: %v", err)
logger.Printf("[IoT广播] 获取设备失败: %v", err)
// 即使失败也发送空列表，让前端知道 IoT 服务状态
devices = []IotDeviceInfo{}
}
@@ -576,7 +654,7 @@ func (h *Hub) pollAndBroadcastIoT() {
data, err := json.Marshal(msg)
if err != nil {
log.Printf("[IoT广播] 消息序列化失败: %v", err)
logger.Printf("[IoT广播] 消息序列化失败: %v", err)
return
}
@@ -586,7 +664,7 @@ func (h *Hub) pollAndBroadcastIoT() {
for _, d := range devices {
deviceNames = append(deviceNames, d.Name)
}
log.Printf("[IoT广播] 已推送 %d 个设备状态到 %d 个客户端: %v", len(devices), h.ClientCount(), deviceNames)
logger.Printf("[IoT广播] 已推送 %d 个设备状态到 %d 个客户端: %v", len(devices), h.ClientCount(), deviceNames)
}
// fetchIoTDevices 从 IoT 调试服务获取设备列表