From 13a2c69d5eb30228b770dd8db222a2eb64adb989 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Fri, 3 Jul 2026 12:25:05 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=E5=8A=9B=E5=B7=AE=E5=92=8C=E8=B7=A8=E7=BE=A4=E8=81=8A=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E6=B3=84=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 记忆嵌入修复: - 新增 memory.APIEmbedder 使用 text-embedding-3-small 替代 SimpleEmbedder - Extractor 保存记忆时自动生成向量嵌入 - Embedder 接口增加 IsAvailable() 方法 跨群聊上下文隔离: - Thinker 新增 thinkSessionID 字段，performThink 启动时绑定会话 - storeThought 优先使用绑定的 session 推送思考结果 - 防止思考过程中其他群消息改变 activeSessionID 导致串台 Co-Authored-By: Claude --- backend/ai-core/cmd/main.go | 11 +- .../ai-core/internal/background/thinker.go | 19 +++- backend/ai-core/internal/memory/embedder.go | 104 ++++++++++++++++++ backend/ai-core/internal/memory/extractor.go | 26 ++++- backend/ai-core/internal/memory/retriever.go | 4 + 5 files changed, 155 insertions(+), 9 deletions(-) create mode 100644 backend/ai-core/internal/memory/embedder.go diff --git a/backend/ai-core/cmd/main.go b/backend/ai-core/cmd/main.go index 876c878..2e22022 100644 --- a/backend/ai-core/cmd/main.go +++ b/backend/ai-core/cmd/main.go @@ -119,12 +119,21 @@ func main() { memStore = memory.NewStore(cfg.DatabaseURL) defer memStore.Close() - memRetriever = memory.NewRetriever(memStore, nil) + // 使用真实嵌入模型 (text-embedding-3-small)，通过 OpenAI 兼容 API + memEmbedder := memory.NewAPIEmbedder(cfg.LLMBaseURL, cfg.LLMAPIKey, "text-embedding-3-small") + if memEmbedder.IsAvailable() { + log.Println("记忆嵌入服务已就绪 (text-embedding-3-small)") + } else { + log.Println("⚠ 记忆嵌入服务未配置 API Key，降级为字符频率嵌入（检索质量较差）") + } + + memRetriever = memory.NewRetriever(memStore, memEmbedder) // 记忆提取器使用 memory purpose 适配器 memExtractor = memory.NewExtractor(memStore, func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) { return memoryAdapter.Chat(ctx, messages) }) + memExtractor.SetEmbedder(memEmbedder) log.Println("记忆提取器已就绪") } diff --git a/backend/ai-core/internal/background/thinker.go b/backend/ai-core/internal/background/thinker.go index b9c4764..712cc4e 100644 --- a/backend/ai-core/internal/background/thinker.go +++ b/backend/ai-core/internal/background/thinker.go @@ -229,6 +229,7 @@ type Thinker struct { lastProactiveNs atomic.Int64 // UnixNano (replaces lastProactiveMsgTime) lastOnlineChange time.Time userSessionID string // 当前活跃的 session ID (用于重连) + thinkSessionID string // 当前思考周期绑定的 session（防止跨群串台） // 时区设置 (默认 Asia/Shanghai，可通过 TZ 环境变量覆盖) timeLocation *time.Location @@ -1114,6 +1115,11 @@ func (t *Thinker) performThink(triggerReason string) { log.Printf("[后台思考] 开始思考周期 (触发原因=%s, 计数=%d)...", triggerReason, currentCount) + // 捕获当前活跃 session，防止思考过程中其他群的消息改变 activeSessionID 导致串台 + t.muLock() + t.thinkSessionID = t.activeSessionID + t.muUnlock() + // 0. 让步于前台——如果用户最近有活动（非post_chat），跳过本次思考。 if triggerReason != "post_chat" { t.muLock() @@ -1133,15 +1139,13 @@ func (t *Thinker) performThink(triggerReason string) { return } - // 2. 获取当前活跃会话的对话历史（优先活跃会话，回退到管理员主会话） + // 2. 获取思考周期绑定会话的对话历史，回退到管理员主会话 var convHistory []model.LLMMessage if t.convStore != nil { - t.muLock() - sessionID := t.activeSessionID + sessionID := t.thinkSessionID if sessionID == "" { sessionID = t.adminSessionID } - t.muUnlock() if sessionID != "" { convHistory = t.convStore.GetHistory(sessionID, 30) @@ -1922,8 +1926,11 @@ func (t *Thinker) storeThought(content string, toolCallsJSON string, toolCallCou // Extract proactive message and optional platform target. proactiveMsg, proactiveTarget := t.extractProactiveMessage(content) - // Prefer active session, fall back to admin main session. - pushSessionID := t.activeSessionID + // 优先使用思考周期绑定的 session（防止跨群串台），其次活跃 session，最后管理员主会话 + pushSessionID := t.thinkSessionID + if pushSessionID == "" { + pushSessionID = t.activeSessionID + } if pushSessionID == "" { pushSessionID = t.adminSessionID } diff --git a/backend/ai-core/internal/memory/embedder.go b/backend/ai-core/internal/memory/embedder.go new file mode 100644 index 0000000..875ce9e --- /dev/null +++ b/backend/ai-core/internal/memory/embedder.go @@ -0,0 +1,104 @@ +package memory + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// APIEmbedder generates text embeddings via OpenAI-compatible API. +type APIEmbedder struct { + baseURL string + apiKey string + model string + httpClient *http.Client +} + +// NewAPIEmbedder creates a new embedding service. +func NewAPIEmbedder(baseURL, apiKey, model string) *APIEmbedder { + return &APIEmbedder{ + baseURL: baseURL, + apiKey: apiKey, + model: model, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +type embRequest struct { + Input []string `json:"input"` + Model string `json:"model"` +} + +type embResponse struct { + Data []embData `json:"data"` + Error *embError `json:"error,omitempty"` +} + +type embData struct { + Embedding []float64 `json:"embedding"` +} + +type embError struct { + Message string `json:"message"` +} + +// Embed generates an embedding vector for the given text. +func (e *APIEmbedder) Embed(ctx context.Context, text string) ([]float64, error) { + if !e.IsAvailable() { + return nil, fmt.Errorf("embedding service not available") + } + + reqBody := embRequest{ + Input: []string{text}, + Model: e.model, + } + + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("marshal embedding request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", e.baseURL+"/embeddings", bytes.NewReader(jsonBody)) + if err != nil { + return nil, fmt.Errorf("create embedding request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+e.apiKey) + + resp, err := e.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("embedding request failed: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read embedding response: %w", err) + } + + var embResp embResponse + if err := json.Unmarshal(body, &embResp); err != nil { + return nil, fmt.Errorf("parse embedding response: %w", err) + } + + if embResp.Error != nil { + return nil, fmt.Errorf("embedding API error: %s", embResp.Error.Message) + } + + if len(embResp.Data) == 0 { + return nil, fmt.Errorf("no embedding returned") + } + + return embResp.Data[0].Embedding, nil +} + +// IsAvailable checks if the embedding service is configured. +func (e *APIEmbedder) IsAvailable() bool { + return e.apiKey != "" && e.baseURL != "" +} diff --git a/backend/ai-core/internal/memory/extractor.go b/backend/ai-core/internal/memory/extractor.go index 8047137..3046c31 100644 --- a/backend/ai-core/internal/memory/extractor.go +++ b/backend/ai-core/internal/memory/extractor.go @@ -6,14 +6,16 @@ import ( "fmt" "git.yeij.top/AskaEth/Cyrene/pkg/logger" "strings" + "time" "git.yeij.top/AskaEth/Cyrene/ai-core/internal/model" ) // Extractor 记忆提取器 —— 从对话中提取结构化记忆 type Extractor struct { - store *Store - llmChat func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) + store *Store + llmChat func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) + embedder Embedder // 可选：为保存的记忆生成向量嵌入 } // NewExtractor 创建记忆提取器 @@ -26,6 +28,11 @@ func NewExtractor(store *Store, llmChat func(ctx context.Context, messages []mod } } +// SetEmbedder sets the embedder for generating vector embeddings on saved memories. +func (e *Extractor) SetEmbedder(embedder Embedder) { + e.embedder = embedder +} + // ExtractAndStore 从一轮对话中提取记忆并存储 // 异步执行，不阻塞主流程 func (e *Extractor) ExtractAndStore(ctx context.Context, userID, sessionID, userMessage, assistantResponse string) { @@ -54,6 +61,21 @@ func (e *Extractor) storeMemories(ctx context.Context, userID, sessionID string, mem.SessionID = sessionID mem.Source = "conversation" + // 生成向量嵌入（异步，不阻塞主流程） + if e.embedder != nil && e.embedder.IsAvailable() { + embedCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + embedding, embErr := e.embedder.Embed(embedCtx, mem.Content) + cancel() + if embErr != nil { + logger.Printf("[memory] 嵌入生成失败: %v，将保存无嵌入的记忆", embErr) + } else { + mem.Embedding = make([]float32, len(embedding)) + for i, v := range embedding { + mem.Embedding[i] = float32(v) + } + } + } + existing, err := e.findSimilar(ctx, userID, &mem) if err == nil && existing != nil { e.mergeMemory(ctx, existing, &mem) diff --git a/backend/ai-core/internal/memory/retriever.go b/backend/ai-core/internal/memory/retriever.go index c7bfec3..4424dea 100644 --- a/backend/ai-core/internal/memory/retriever.go +++ b/backend/ai-core/internal/memory/retriever.go @@ -20,6 +20,7 @@ type Retriever struct { // Embedder 文本嵌入接口 type Embedder interface { Embed(ctx context.Context, text string) ([]float64, error) + IsAvailable() bool } // SimpleEmbedder 基于关键词的简单嵌入（MVP阶段可用，无需外部API） @@ -43,6 +44,9 @@ func (e *SimpleEmbedder) Embed(ctx context.Context, text string) ([]float64, err return vec, nil } +// IsAvailable returns true (SimpleEmbedder is always available as fallback). +func (e *SimpleEmbedder) IsAvailable() bool { return true } + // NewRetriever 创建记忆检索器 func NewRetriever(store *Store, embedder Embedder) *Retriever { if embedder == nil {