package background import ( "context" "encoding/json" "fmt" "log" "os" "strconv" "strings" "sync" "time" ctxbuild "github.com/yourname/cyrene-ai/ai-core/internal/context" "github.com/yourname/cyrene-ai/ai-core/internal/llm" "github.com/yourname/cyrene-ai/ai-core/internal/memory" "github.com/yourname/cyrene-ai/ai-core/internal/model" "github.com/yourname/cyrene-ai/ai-core/internal/persona" "github.com/yourname/cyrene-ai/ai-core/internal/tools" ) // PendingThought 待推送的后台思考 type PendingThought struct { Content string `json:"content"` CreatedAt time.Time `json:"created_at"` Consumed bool `json:"consumed"` } // Thinker 后台思考器(增强版:支持工具调用、记忆管理、5分钟定时循环) type Thinker struct { mu sync.Mutex enabled bool personaLoader *persona.Loader memRetriever *memory.Retriever llmAdapter *llm.Adapter iotClient *tools.IoTClient idleTimeout time.Duration // 闲置超时 thinkInterval time.Duration // 两次思考最小间隔 iotQueryInterval time.Duration // IoT查询最小间隔 // 新增字段:记忆管理 memoryStore *memory.Store // 直接操作记忆(衰减、合并) memoryExtractor *memory.Extractor // 从思考结果中提取记忆 // 新增字段:工具调用 toolRegistry *tools.Registry // 工具注册中心 // 新增字段:会话上下文 convStore *ctxbuild.ConversationStore // 管理员对话历史 adminUserID string // 管理员用户 ID adminSessionID string // 管理员主对话 session ID // 记忆服务 HTTP 客户端(用于持久化思考日志) memClient *memory.Client pendingThoughts []*PendingThought lastUserMessage time.Time lastThinkTime time.Time lastIoTQuery time.Time stopCh chan struct{} wg sync.WaitGroup } // ThinkerConfig 后台思考配置 type ThinkerConfig struct { Enabled bool IdleTimeout time.Duration ThinkInterval time.Duration IoTQueryInterval time.Duration } // DefaultThinkerConfig 默认配置 func DefaultThinkerConfig() ThinkerConfig { return ThinkerConfig{ Enabled: getEnvBool("ENABLE_BACKGROUND_THINKING", true), IdleTimeout: getEnvDuration("THINK_IDLE_TIMEOUT_SEC", 120), ThinkInterval: getEnvDuration("THINK_INTERVAL_SEC", 300), IoTQueryInterval: getEnvDuration("IOT_QUERY_INTERVAL_SEC", 600), } } // NewThinker 创建增强版后台思考器 func NewThinker( cfg ThinkerConfig, personaLoader *persona.Loader, memRetriever *memory.Retriever, llmAdapter *llm.Adapter, iotClient *tools.IoTClient, memoryStore *memory.Store, memoryExtractor *memory.Extractor, toolRegistry *tools.Registry, convStore *ctxbuild.ConversationStore, adminUserID string, adminSessionID string, memClient *memory.Client, ) *Thinker { return &Thinker{ enabled: cfg.Enabled, personaLoader: personaLoader, memRetriever: memRetriever, llmAdapter: llmAdapter, iotClient: iotClient, idleTimeout: cfg.IdleTimeout, thinkInterval: cfg.ThinkInterval, iotQueryInterval: cfg.IoTQueryInterval, memoryStore: memoryStore, memoryExtractor: memoryExtractor, toolRegistry: toolRegistry, convStore: convStore, adminUserID: adminUserID, adminSessionID: adminSessionID, memClient: memClient, pendingThoughts: make([]*PendingThought, 0), lastUserMessage: time.Now(), stopCh: make(chan struct{}), } } // Start 启动后台思考循环(5分钟定时器) func (t *Thinker) Start() { if !t.enabled { log.Println("[后台思考] 已禁用 (ENABLE_BACKGROUND_THINKING=false)") return } t.wg.Add(1) go t.loop() log.Printf("[后台思考] 已启动 (思考间隔=%v, IoT查询间隔=%v, 管理员=%s)", t.thinkInterval, t.iotQueryInterval, t.adminUserID) } // Stop 停止后台思考 func (t *Thinker) Stop() { close(t.stopCh) t.wg.Wait() log.Println("[后台思考] 已停止") } // RecordUserMessage 记录用户活动时间(管理员对话时调用) func (t *Thinker) RecordUserMessage() { t.mu.Lock() t.lastUserMessage = time.Now() t.mu.Unlock() } // GetPendingThoughts 获取并消费所有待处理的后台思考 func (t *Thinker) GetPendingThoughts() []*PendingThought { t.mu.Lock() defer t.mu.Unlock() if len(t.pendingThoughts) == 0 { return nil } result := t.pendingThoughts t.pendingThoughts = make([]*PendingThought, 0) // 标记已消费 for _, pt := range result { pt.Consumed = true } return result } // HasPendingThoughts 检查是否有待处理的思考 func (t *Thinker) HasPendingThoughts() bool { t.mu.Lock() defer t.mu.Unlock() return len(t.pendingThoughts) > 0 } // loop 后台主循环(5分钟定时器) func (t *Thinker) loop() { defer t.wg.Done() // 启动后等待 10 秒再执行首次思考(让服务完全就绪) initialDelay := time.NewTimer(10 * time.Second) ticker := time.NewTicker(t.thinkInterval) defer ticker.Stop() // 思考计数器(用于周期性记忆维护) thinkCount := 0 for { select { case <-t.stopCh: initialDelay.Stop() return case <-initialDelay.C: t.performThink() thinkCount++ t.maybeMaintainMemories(thinkCount) case <-ticker.C: t.performThink() thinkCount++ t.maybeMaintainMemories(thinkCount) } } } // maybeMaintainMemories 周期性执行记忆维护(每6次思考约30分钟) func (t *Thinker) maybeMaintainMemories(thinkCount int) { if thinkCount%6 != 0 { return } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if t.memoryStore != nil && t.memoryStore.IsReady() { // 衰减旧记忆 if err := t.memoryStore.DecayMemories(ctx, t.adminUserID); err != nil { log.Printf("[后台思考] 记忆衰减失败: %v", err) } // 合并相似记忆 if err := t.memoryStore.ConsolidateMemories(ctx, t.adminUserID); err != nil { log.Printf("[后台思考] 记忆合并失败: %v", err) } } } // performThink 执行一次增强版后台思考(支持工具调用和记忆管理) func (t *Thinker) performThink() { ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) defer cancel() log.Println("[后台思考] 开始执行思考周期...") // 1. 加载人格配置 personaConfig, err := t.personaLoader.Get("cyrene") if err != nil { log.Printf("[后台思考] 加载人格失败: %v", err) return } // 2. 检索相关记忆 var memories []memory.MemoryEntry if t.memRetriever != nil { memories, err = t.memRetriever.Retrieve(ctx, t.adminUserID, "最近发生了什么 重要的事情 用户偏好 个人信息") if err != nil { log.Printf("[后台思考] 记忆检索失败: %v", err) } } // 3. 获取管理员对话历史 var convHistory []model.LLMMessage if t.convStore != nil && t.adminSessionID != "" { convHistory = t.convStore.GetHistory(t.adminSessionID, 30) if len(convHistory) > 0 { log.Printf("[后台思考] 加载管理员对话历史 %d 条", len(convHistory)) } } // 4. 查询 IoT 设备状态(节制) var deviceSummary string if t.iotClient != nil { t.mu.Lock() canQuery := time.Since(t.lastIoTQuery) >= t.iotQueryInterval t.mu.Unlock() if canQuery { devices := t.iotClient.GetDevicesForContext() if len(devices) > 0 { deviceSummary = formatDeviceContext(devices) } t.mu.Lock() t.lastIoTQuery = time.Now() t.mu.Unlock() } } // 5. 构建思考提示词 systemPrompt := t.buildThinkingSystemPrompt(personaConfig) userPrompt := t.buildThinkingUserPrompt(memories, convHistory, deviceSummary) messages := []model.LLMMessage{ {Role: model.RoleSystem, Content: systemPrompt}, {Role: model.RoleUser, Content: userPrompt}, } // 6. 准备工具定义 openAITools := t.buildOpenAITools() // 7. 调用 LLM(支持工具调用,最多 3 轮) maxToolRounds := 3 var finalContent string var totalToolCalls int var toolCallRecords []map[string]interface{} for round := 0; round <= maxToolRounds; round++ { resp, err := t.llmAdapter.ChatWithTools(ctx, messages, openAITools) if err != nil { log.Printf("[后台思考] LLM调用失败 (round=%d): %v", round, err) return } // 如果 LLM 没有请求工具调用,这就是最终回复 if len(resp.ToolCalls) == 0 { finalContent = resp.Content break } log.Printf("[后台思考] LLM 请求 %d 个工具调用 (round=%d)", len(resp.ToolCalls), round) // 将助手消息(含工具调用)加入上下文 assistantMsg := model.LLMMessage{ Role: model.RoleAssistant, Content: resp.Content, ToolCalls: resp.ToolCalls, ReasoningContent: resp.ReasoningContent, } messages = append(messages, assistantMsg) // 执行每个工具调用 for _, tc := range resp.ToolCalls { var args map[string]interface{} if err := json.Unmarshal([]byte(tc.Arguments), &args); err != nil { log.Printf("[后台思考] 工具 %s 参数解析失败: %v", tc.Name, err) args = make(map[string]interface{}) } result, execErr := t.toolRegistry.Execute(ctx, tc.Name, args) if execErr != nil { log.Printf("[后台思考] 工具 %s 执行失败: %v", tc.Name, execErr) } resultJSON, _ := json.Marshal(result) messages = append(messages, model.LLMMessage{ Role: model.RoleTool, Content: string(resultJSON), ToolCallID: tc.ID, }) totalToolCalls++ toolCallRecords = append(toolCallRecords, map[string]interface{}{ "name": tc.Name, "args": args, }) } // 最后一轮:即使有 tool_calls 也强制停止 if round == maxToolRounds { // 再做一次不带工具的调用获取最终总结 finalResp, finalErr := t.llmAdapter.Chat(ctx, messages) if finalErr != nil { log.Printf("[后台思考] 最终总结调用失败: %v", finalErr) finalContent = resp.Content } else { finalContent = finalResp.Content } break } } if finalContent == "" { log.Println("[后台思考] 未获得有效思考内容,跳过") return } // 序列化工具调用记录 toolCallsJSON := "[]" if len(toolCallRecords) > 0 { if data, err := json.Marshal(toolCallRecords); err == nil { toolCallsJSON = string(data) } } // 8. 存储思考结果(内存队列 + 持久化到 memory-service) t.storeThought(finalContent, toolCallsJSON, totalToolCalls) log.Printf("[后台思考] 完成 (内容长度=%d, 工具调用=%d次)", len(finalContent), totalToolCalls) // 9. 从思考结果中提取记忆(异步) if t.memoryExtractor != nil { go t.extractMemoriesFromThinking(finalContent) } } // buildThinkingSystemPrompt 构建思考用的系统提示词 func (t *Thinker) buildThinkingSystemPrompt(personaConfig *persona.PersonaConfig) string { basePrompt := personaConfig.BuildSystemPrompt("开拓者", 1) thinkingInstructions := ` 【后台思考模式】 你正在进行定期的自我反思。开拓者暂时没有说话,这是你主动思考和整理信息的时间。 请完成以下内省任务: 1. **回顾最近的对话**:总结开拓者表达了什么、情绪如何、有什么新信息值得关注 2. **评估现有记忆**:基于记忆检索结果,判断哪些记忆仍然相关、哪些可能需要更新 3. **识别重要信息**:是否有值得保存的用户偏好、个人信息、计划任务、关系信息等 4. **使用工具获取信息**:如果需要了解当前时间、搜索实时信息等,请使用可用工具 5. **记忆操作建议**:判断是否需要创建新记忆、更新旧记忆或合并重复记忆 完成反思后,请输出结构化的思考总结,包含: - **关键洞察**:从最近对话中提炼的核心发现 - **记忆更新建议**:需要创建/更新/合并的记忆条目 - **下次关注事项**:下次思考时需要跟进的话题或任务 注意: - 这是内部思考,不是直接与开拓者对话 - 请以自省和观察的方式思考,不要用"你"来称呼开拓者 - 有机会就使用工具获取实时信息(如当前时间) - 思考要简洁有深度,不需要太长` return basePrompt + thinkingInstructions } // buildThinkingUserPrompt 构建思考用的用户提示词 func (t *Thinker) buildThinkingUserPrompt( memories []memory.MemoryEntry, convHistory []model.LLMMessage, deviceSummary string, ) string { var sb strings.Builder sb.WriteString("现在是你的后台思考时间。请基于以下信息进行深度反思。\n") // 对话历史 if len(convHistory) > 0 { sb.WriteString("\n【最近的对话历史】\n") msgCount := 0 for _, msg := range convHistory { if msg.Role == model.RoleUser || msg.Role == model.RoleAssistant { roleLabel := "开拓者" if msg.Role == model.RoleAssistant { roleLabel = "昔涟" } content := msg.Content runes := []rune(content) if len(runes) > 200 { content = string(runes[:200]) + "…" } sb.WriteString(fmt.Sprintf("[%s]: %s\n", roleLabel, content)) msgCount++ } } if msgCount == 0 { sb.WriteString("(暂无对话历史)\n") } } else { sb.WriteString("\n【最近的对话历史】\n(暂无对话历史,这是首次思考或对话历史为空)\n") } // 现有记忆 if len(memories) > 0 { sb.WriteString("\n【现有相关记忆】\n") for i, m := range memories { if i >= 15 { sb.WriteString(fmt.Sprintf("... 还有 %d 条记忆未列出\n", len(memories)-15)) break } sb.WriteString(fmt.Sprintf("- [%s|重要度%d] %s\n", m.Category.DisplayName(), m.Importance, m.Content)) } } else { sb.WriteString("\n【现有相关记忆】\n(暂无相关记忆)\n") } // IoT 设备状态 if deviceSummary != "" { sb.WriteString("\n" + deviceSummary) } sb.WriteString("\n请开始你的后台思考。如果需要获取当前时间或搜索信息,请使用可用工具。") return sb.String() } // buildOpenAITools 将工具注册中心的定义转换为 LLM 层的 OpenAITool 格式 func (t *Thinker) buildOpenAITools() []llm.OpenAITool { if t.toolRegistry == nil || !t.toolRegistry.IsEnabled() { return nil } defs := t.toolRegistry.GetDefinitions() if len(defs) == 0 { return nil } result := make([]llm.OpenAITool, 0, len(defs)) for _, d := range defs { result = append(result, llm.OpenAITool{ Type: "function", Function: llm.OpenAIToolFunc{ Name: d.Name, Description: d.Description, Parameters: d.Parameters, }, }) } return result } // storeThought 存储思考结果到待推送队列,并异步持久化到 memory-service func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCount int) { t.mu.Lock() t.pendingThoughts = append(t.pendingThoughts, &PendingThought{ Content: content, CreatedAt: time.Now(), Consumed: false, }) // 只保留最近 10 条 if len(t.pendingThoughts) > 10 { t.pendingThoughts = t.pendingThoughts[len(t.pendingThoughts)-10:] } t.mu.Unlock() log.Printf("[后台思考] 思考已存储 (当前累积 %d 条待推送思考)", len(t.pendingThoughts)) // 异步持久化到 memory-service (不阻塞思考循环) if t.memClient != nil { go func() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := t.memClient.SaveThinkingLog(ctx, t.adminUserID, content, toolCallsJSON, toolCallCount, len(content)); err != nil { log.Printf("[后台思考] 持久化思考日志失败: %v", err) } else { log.Printf("[后台思考] 思考日志已持久化 (长度=%d, 工具调用=%d)", len(content), toolCallCount) } }() } } // extractMemoriesFromThinking 从思考结果中提取记忆(异步执行) func (t *Thinker) extractMemoriesFromThinking(thinkingContent string) { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() log.Println("[后台思考] 开始从思考结果中提取记忆...") // 使用 memoryExtractor.ExtractAndStore 提取记忆 // 将思考内容作为"昔涟的自省"传递给提取器 t.memoryExtractor.ExtractAndStore( ctx, t.adminUserID, t.adminSessionID, "【系统触发】后台思考时间 — 昔涟进行了自我反思,以下是她的思考内容", thinkingContent, ) } // formatDeviceContext 格式化设备状态为文本 func formatDeviceContext(devices []tools.IoTDevice) string { if len(devices) == 0 { return "" } summary := "[当前IoT设备状态]\n" for _, d := range devices { switch d.Type { case "light": if d.Status == "on" { summary += fmt.Sprintf("- %s: 开启 (亮度%d%%, %s)\n", d.Name, d.Brightness, d.Color) } else { summary += fmt.Sprintf("- %s: 关闭\n", d.Name) } case "ac": if d.Status == "on" { summary += fmt.Sprintf("- %s: 运行中 (%s%.0f°C)\n", d.Name, modeLabel(d.Mode), d.Temperature) } else { summary += fmt.Sprintf("- %s: 关闭\n", d.Name) } case "curtain": statusLabel := "已关闭" if d.Status == "open" { statusLabel = "已打开" } summary += fmt.Sprintf("- %s: %s\n", d.Name, statusLabel) case "sensor": summary += fmt.Sprintf("- %s: %.1f%s\n", d.Name, d.Value, d.Unit) case "lock": statusLabel := "已锁定" if d.Status == "unlocked" { statusLabel = "已解锁" } summary += fmt.Sprintf("- %s: %s (电量%d%%)\n", d.Name, statusLabel, d.Battery) } } return summary } func modeLabel(mode string) string { switch mode { case "cool": return "制冷" case "heat": return "制热" case "auto": return "自动" default: return mode } } func getEnvBool(key string, fallback bool) bool { v := os.Getenv(key) if v == "" { return fallback } b, err := strconv.ParseBool(v) if err != nil { return fallback } return b } func getEnvDuration(key string, fallbackSec int) time.Duration { v := os.Getenv(key) if v == "" { return time.Duration(fallbackSec) * time.Second } sec, err := strconv.Atoi(v) if err != nil { return time.Duration(fallbackSec) * time.Second } return time.Duration(sec) * time.Second }