fix: 修复记忆力差和跨群聊上下文泄漏
记忆嵌入修复: - 新增 memory.APIEmbedder 使用 text-embedding-3-small 替代 SimpleEmbedder - Extractor 保存记忆时自动生成向量嵌入 - Embedder 接口增加 IsAvailable() 方法 跨群聊上下文隔离: - Thinker 新增 thinkSessionID 字段,performThink 启动时绑定会话 - storeThought 优先使用绑定的 session 推送思考结果 - 防止思考过程中其他群消息改变 activeSessionID 导致串台 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -119,12 +119,21 @@ func main() {
|
|||||||
memStore = memory.NewStore(cfg.DatabaseURL)
|
memStore = memory.NewStore(cfg.DatabaseURL)
|
||||||
defer memStore.Close()
|
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 适配器
|
// 记忆提取器使用 memory purpose 适配器
|
||||||
memExtractor = memory.NewExtractor(memStore, func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
|
memExtractor = memory.NewExtractor(memStore, func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error) {
|
||||||
return memoryAdapter.Chat(ctx, messages)
|
return memoryAdapter.Chat(ctx, messages)
|
||||||
})
|
})
|
||||||
|
memExtractor.SetEmbedder(memEmbedder)
|
||||||
log.Println("记忆提取器已就绪")
|
log.Println("记忆提取器已就绪")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -229,6 +229,7 @@ type Thinker struct {
|
|||||||
lastProactiveNs atomic.Int64 // UnixNano (replaces lastProactiveMsgTime)
|
lastProactiveNs atomic.Int64 // UnixNano (replaces lastProactiveMsgTime)
|
||||||
lastOnlineChange time.Time
|
lastOnlineChange time.Time
|
||||||
userSessionID string // 当前活跃的 session ID (用于重连)
|
userSessionID string // 当前活跃的 session ID (用于重连)
|
||||||
|
thinkSessionID string // 当前思考周期绑定的 session(防止跨群串台)
|
||||||
|
|
||||||
// 时区设置 (默认 Asia/Shanghai,可通过 TZ 环境变量覆盖)
|
// 时区设置 (默认 Asia/Shanghai,可通过 TZ 环境变量覆盖)
|
||||||
timeLocation *time.Location
|
timeLocation *time.Location
|
||||||
@@ -1114,6 +1115,11 @@ func (t *Thinker) performThink(triggerReason string) {
|
|||||||
|
|
||||||
log.Printf("[后台思考] 开始思考周期 (触发原因=%s, 计数=%d)...", triggerReason, currentCount)
|
log.Printf("[后台思考] 开始思考周期 (触发原因=%s, 计数=%d)...", triggerReason, currentCount)
|
||||||
|
|
||||||
|
// 捕获当前活跃 session,防止思考过程中其他群的消息改变 activeSessionID 导致串台
|
||||||
|
t.muLock()
|
||||||
|
t.thinkSessionID = t.activeSessionID
|
||||||
|
t.muUnlock()
|
||||||
|
|
||||||
// 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。
|
// 0. 让步于前台——如果用户最近有活动(非post_chat),跳过本次思考。
|
||||||
if triggerReason != "post_chat" {
|
if triggerReason != "post_chat" {
|
||||||
t.muLock()
|
t.muLock()
|
||||||
@@ -1133,15 +1139,13 @@ func (t *Thinker) performThink(triggerReason string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// 2. 获取当前活跃会话的对话历史(优先活跃会话,回退到管理员主会话)
|
// 2. 获取思考周期绑定会话的对话历史,回退到管理员主会话
|
||||||
var convHistory []model.LLMMessage
|
var convHistory []model.LLMMessage
|
||||||
if t.convStore != nil {
|
if t.convStore != nil {
|
||||||
t.muLock()
|
sessionID := t.thinkSessionID
|
||||||
sessionID := t.activeSessionID
|
|
||||||
if sessionID == "" {
|
if sessionID == "" {
|
||||||
sessionID = t.adminSessionID
|
sessionID = t.adminSessionID
|
||||||
}
|
}
|
||||||
t.muUnlock()
|
|
||||||
|
|
||||||
if sessionID != "" {
|
if sessionID != "" {
|
||||||
convHistory = t.convStore.GetHistory(sessionID, 30)
|
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.
|
// Extract proactive message and optional platform target.
|
||||||
proactiveMsg, proactiveTarget := t.extractProactiveMessage(content)
|
proactiveMsg, proactiveTarget := t.extractProactiveMessage(content)
|
||||||
// Prefer active session, fall back to admin main session.
|
// 优先使用思考周期绑定的 session(防止跨群串台),其次活跃 session,最后管理员主会话
|
||||||
pushSessionID := t.activeSessionID
|
pushSessionID := t.thinkSessionID
|
||||||
|
if pushSessionID == "" {
|
||||||
|
pushSessionID = t.activeSessionID
|
||||||
|
}
|
||||||
if pushSessionID == "" {
|
if pushSessionID == "" {
|
||||||
pushSessionID = t.adminSessionID
|
pushSessionID = t.adminSessionID
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 != ""
|
||||||
|
}
|
||||||
@@ -6,14 +6,16 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"git.yeij.top/AskaEth/Cyrene/pkg/logger"
|
"git.yeij.top/AskaEth/Cyrene/pkg/logger"
|
||||||
"strings"
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.yeij.top/AskaEth/Cyrene/ai-core/internal/model"
|
"git.yeij.top/AskaEth/Cyrene/ai-core/internal/model"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Extractor 记忆提取器 —— 从对话中提取结构化记忆
|
// Extractor 记忆提取器 —— 从对话中提取结构化记忆
|
||||||
type Extractor struct {
|
type Extractor struct {
|
||||||
store *Store
|
store *Store
|
||||||
llmChat func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error)
|
llmChat func(ctx context.Context, messages []model.LLMMessage) (*model.LLMResponse, error)
|
||||||
|
embedder Embedder // 可选:为保存的记忆生成向量嵌入
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewExtractor 创建记忆提取器
|
// 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 从一轮对话中提取记忆并存储
|
// ExtractAndStore 从一轮对话中提取记忆并存储
|
||||||
// 异步执行,不阻塞主流程
|
// 异步执行,不阻塞主流程
|
||||||
func (e *Extractor) ExtractAndStore(ctx context.Context, userID, sessionID, userMessage, assistantResponse string) {
|
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.SessionID = sessionID
|
||||||
mem.Source = "conversation"
|
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)
|
existing, err := e.findSimilar(ctx, userID, &mem)
|
||||||
if err == nil && existing != nil {
|
if err == nil && existing != nil {
|
||||||
e.mergeMemory(ctx, existing, &mem)
|
e.mergeMemory(ctx, existing, &mem)
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ type Retriever struct {
|
|||||||
// Embedder 文本嵌入接口
|
// Embedder 文本嵌入接口
|
||||||
type Embedder interface {
|
type Embedder interface {
|
||||||
Embed(ctx context.Context, text string) ([]float64, error)
|
Embed(ctx context.Context, text string) ([]float64, error)
|
||||||
|
IsAvailable() bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// SimpleEmbedder 基于关键词的简单嵌入(MVP阶段可用,无需外部API)
|
// SimpleEmbedder 基于关键词的简单嵌入(MVP阶段可用,无需外部API)
|
||||||
@@ -43,6 +44,9 @@ func (e *SimpleEmbedder) Embed(ctx context.Context, text string) ([]float64, err
|
|||||||
return vec, nil
|
return vec, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// IsAvailable returns true (SimpleEmbedder is always available as fallback).
|
||||||
|
func (e *SimpleEmbedder) IsAvailable() bool { return true }
|
||||||
|
|
||||||
// NewRetriever 创建记忆检索器
|
// NewRetriever 创建记忆检索器
|
||||||
func NewRetriever(store *Store, embedder Embedder) *Retriever {
|
func NewRetriever(store *Store, embedder Embedder) *Retriever {
|
||||||
if embedder == nil {
|
if embedder == nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user