Files
Cyrene/backend/memory-service/internal/service/memory_service.go
T
AskaEth 91c9ee4b2d fix: 修复 AI 回复无法送达发送者 + 重复消息 + action角色泄露 + OS环境支持
广播逻辑重构:
- AI 回复 (stream_start/response/stream_segments/multi_message/stream_end) 改用 broadcastToUser 发送给所有客户端
- 用户消息回显保持 broadcastToUserExcept 排除发送者

消息去重与角色修复:
- CacheMessage(user) 移至回复生成后,避免本轮 LLM 调用出现重复用户消息
- action 角色消息在 DB 存储时映射为 assistant,DeepSeek 等模型不支持自定义角色
- stream_end defer 机制确保错误路径也会终止客户端思考指示器

OS 完整环境支持:
- host 包重构为 HostBackend 接口 + Direct/WSL/Docker 三种后端
- 新增 os_exec/os_file/os_system 工具供 AI 在完整 Linux 环境中自由操作

其他:
- 视觉模型注入 + 图片预处理后清空 Images 避免传给 Chat 模型
- 图片 URL 相对路径→绝对 URL 转换
- DevTools 链路追踪页面 + 重启修复
- 记忆搜索模糊匹配增强
- 后台思考定时调度支持
- 管理后台页面 (模型配置/用户管理等)
- docs/api 更新广播机制说明

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-29 12:46:17 +08:00

323 lines
8.5 KiB
Go

package service
import (
"context"
"fmt"
"github.com/yourname/cyrene-ai/pkg/logger"
"strings"
"github.com/yourname/cyrene-ai/memory-service/internal/model"
"github.com/yourname/cyrene-ai/memory-service/internal/store"
)
// deDupThreshold 去重相似度阈值
const deDupThreshold = 0.75
// MemoryService 记忆业务逻辑
type MemoryService struct {
store *store.Store
}
// NewMemoryService 创建记忆服务
func NewMemoryService(s *store.Store) *MemoryService {
return &MemoryService{store: s}
}
// CreateMemory 创建/保存记忆
func (svc *MemoryService) CreateMemory(ctx context.Context, entry *model.MemoryEntry) error {
if entry.UserID == "" {
return fmt.Errorf("user_id 不能为空")
}
if entry.Content == "" {
return fmt.Errorf("content 不能为空")
}
if entry.Category == "" {
entry.Category = model.CategoryKnowledge
}
if entry.Importance < 1 {
entry.Importance = 5
}
// Priority 范围检查:扩展为 [0, 10] 以支持用户自定义高优先级记忆
// 低于 0 的视为无效,重置为 normal;高于 10 的截断到 10
if entry.Priority < 0 {
entry.Priority = model.MemoryNormal
}
if entry.Priority > 10 {
entry.Priority = 10
}
if entry.Source == "" {
entry.Source = "manual"
}
// 去重检查
similar, err := svc.findSimilar(ctx, entry.UserID, entry)
if err == nil && similar != nil {
// 合并到已有记忆
return svc.mergeMemory(ctx, similar, entry)
}
return svc.store.Save(ctx, entry)
}
// GetMemory 获取单个记忆
func (svc *MemoryService) GetMemory(ctx context.Context, id string) (*model.MemoryEntry, error) {
return svc.store.GetByID(ctx, id)
}
// ListMemories 列出用户所有记忆
func (svc *MemoryService) ListMemories(ctx context.Context, userID string, category string, minImportance int, limit int, offset int) ([]model.MemoryEntry, error) {
if limit <= 0 {
limit = 50
}
q := model.MemoryQuery{
UserID: userID,
MinImportance: minImportance,
Limit: limit,
Offset: offset,
}
if category != "" {
q.Category = model.MemoryCategory(category)
}
return svc.store.Query(ctx, q)
}
// UpdateMemory 更新记忆
func (svc *MemoryService) UpdateMemory(ctx context.Context, entry *model.MemoryEntry) error {
if entry.ID == "" {
return fmt.Errorf("id 不能为空")
}
return svc.store.Update(ctx, entry)
}
// DeleteMemory 删除记忆
func (svc *MemoryService) DeleteMemory(ctx context.Context, id string) error {
return svc.store.Delete(ctx, id)
}
// QueryMemories 语义查询 + 关键词匹配
func (svc *MemoryService) QueryMemories(ctx context.Context, userID string, queryText string, category string, minImportance int, limit int) ([]model.MemoryEntry, error) {
if limit <= 0 {
limit = 10
}
var allEntries []model.MemoryEntry
seen := make(map[string]bool)
// 1. 关键词匹配检索
keywordEntries, err := svc.store.SearchByKeyword(ctx, userID, queryText, limit*2)
if err == nil {
for _, e := range keywordEntries {
if !seen[e.ID] {
seen[e.ID] = true
allEntries = append(allEntries, e)
}
}
}
// 2. 补充按分类/重要性查询
q := model.MemoryQuery{
UserID: userID,
MinImportance: minImportance,
Limit: limit,
}
if category != "" {
q.Category = model.MemoryCategory(category)
}
categoryEntries, err := svc.store.Query(ctx, q)
if err == nil {
for _, e := range categoryEntries {
if !seen[e.ID] {
seen[e.ID] = true
allEntries = append(allEntries, e)
}
}
}
// 3. 在应用层做内容匹配过滤
queryLower := strings.ToLower(queryText)
var matched []model.MemoryEntry
for _, entry := range allEntries {
contentLower := strings.ToLower(entry.Content)
summaryLower := strings.ToLower(entry.Summary)
if strings.Contains(contentLower, queryLower) || strings.Contains(summaryLower, queryLower) {
matched = append(matched, entry)
continue
}
for _, kw := range entry.Keywords {
if strings.Contains(queryLower, strings.ToLower(kw)) ||
strings.Contains(strings.ToLower(kw), queryLower) {
matched = append(matched, entry)
break
}
}
}
if len(matched) == 0 {
matched = allEntries
}
// 4. 去重合并
matched = svc.deduplicate(matched)
// 5. 按重要性降序
sortByImportance(matched)
if len(matched) > limit {
matched = matched[:limit]
}
return matched, nil
}
// ConsolidateMemories 合并相似记忆
func (svc *MemoryService) ConsolidateMemories(ctx context.Context, userID string) (int, error) {
return svc.store.ConsolidateMemories(ctx, userID)
}
// DecayMemories 衰减旧记忆
func (svc *MemoryService) DecayMemories(ctx context.Context, userID string) (int, int, error) {
return svc.store.DecayMemories(ctx, userID)
}
// GetCategories 获取用户分类统计
func (svc *MemoryService) GetCategories(ctx context.Context, userID string) (map[string]int, error) {
return svc.store.GetCategories(ctx, userID)
}
// findSimilar 查找相似记忆
func (svc *MemoryService) findSimilar(ctx context.Context, userID string, newMem *model.MemoryEntry) (*model.MemoryEntry, error) {
existing, err := svc.store.Query(ctx, model.MemoryQuery{
UserID: userID,
Limit: 100,
})
if err != nil {
return nil, err
}
for i := range existing {
score := existing[i].SimilarityScore(newMem)
if score >= deDupThreshold {
return &existing[i], nil
}
}
return nil, nil
}
// mergeMemory 合并新记忆到已有记忆
func (svc *MemoryService) mergeMemory(ctx context.Context, existing *model.MemoryEntry, newMem *model.MemoryEntry) error {
// 更新内容(如果新内容更有价值)
if newMem.Importance > existing.Importance || len(newMem.Content) > len(existing.Content) {
existing.Content = newMem.Content
existing.Summary = newMem.Summary
}
// 合并关键词
keywordSet := make(map[string]bool)
for _, k := range existing.Keywords {
keywordSet[k] = true
}
for _, k := range newMem.Keywords {
keywordSet[k] = true
}
mergedKeywords := make([]string, 0, len(keywordSet))
for k := range keywordSet {
mergedKeywords = append(mergedKeywords, k)
}
existing.Keywords = mergedKeywords
// 取最高重要性
if newMem.Importance > existing.Importance {
existing.Importance = newMem.Importance
}
// 取最高优先级
if newMem.Priority > existing.Priority {
existing.Priority = newMem.Priority
}
existing.AccessCount++
existing.Source = "merged"
logger.Printf("[memory-service] 合并记忆 [%s|%d★]: %s (去重)", existing.Category, existing.Importance, existing.Summary)
return svc.store.Update(ctx, existing)
}
// deduplicate 去重合并
func (svc *MemoryService) deduplicate(entries []model.MemoryEntry) []model.MemoryEntry {
if len(entries) < 2 {
return entries
}
result := make([]model.MemoryEntry, 0, len(entries))
discarded := make(map[int]bool)
for i := 0; i < len(entries); i++ {
if discarded[i] {
continue
}
for j := i + 1; j < len(entries); j++ {
if discarded[j] {
continue
}
score := entries[i].SimilarityScore(&entries[j])
if score >= deDupThreshold {
if entries[j].Importance > entries[i].Importance ||
(entries[j].Importance == entries[i].Importance && entries[j].Priority > entries[i].Priority) {
discarded[i] = true
break
} else {
discarded[j] = true
}
}
}
if !discarded[i] {
result = append(result, entries[i])
}
}
return result
}
// sortByImportance 按 Importance 降序, Priority 降序排列
func sortByImportance(entries []model.MemoryEntry) {
for i := 0; i < len(entries); i++ {
for j := i + 1; j < len(entries); j++ {
if entries[j].Importance > entries[i].Importance ||
(entries[j].Importance == entries[i].Importance && entries[j].Priority > entries[i].Priority) {
entries[i], entries[j] = entries[j], entries[i]
}
}
}
}
// SaveThinkingLog 保存自主思考日志
func (svc *MemoryService) SaveThinkingLog(ctx context.Context, tl *model.ThinkingLog) error {
if tl.Content == "" {
return fmt.Errorf("content 不能为空")
}
if tl.UserID == "" {
tl.UserID = "admin"
}
return svc.store.SaveThinkingLog(ctx, tl)
}
// QueryThinkingLogs 分页查询思考日志
func (svc *MemoryService) QueryThinkingLogs(ctx context.Context, q model.ThinkingQuery) ([]model.ThinkingLog, error) {
return svc.store.QueryThinkingLogs(ctx, q)
}
// GetThinkingLogByID 获取单条思考日志
func (svc *MemoryService) GetThinkingLogByID(ctx context.Context, id string) (*model.ThinkingLog, error) {
return svc.store.GetThinkingLogByID(ctx, id)
}
// GetThinkingStats 获取思考日志统计
func (svc *MemoryService) GetThinkingStats(ctx context.Context, userID string) (*model.ThinkingStats, error) {
return svc.store.GetThinkingStats(ctx, userID)
}