feat(memory): P1 长期记忆升级 —— 异步攒批 Consolidate + 软删 + importance/last_seen

把"逐轮盲写抽取"升级为 Mem0 式对账（方案见 memory_industry_analysis.md 落地节）： mcp-go： - Profile 加 Importance(1~10, poignancy) + LastSeenAt（为 Generative Agents 读路径 Score=w1·Relevance+w2·Recency+w3·Importance 铺路）。 - Upsert 收 importance + 每次置 last_seen（印证）；新增 Delete（软删，BaseModel.DeletedAt 已具备，失效不物删可审计）+ Touch；memory_upsert 透传 importance、新增 memory_delete 工具。 dispatcher： - extractMemory → consolidateMemory：一次 LLM 调用同时做抽取+对账，输出 [{op:ADD|UPDATE|DELETE|NOOP,key,value,importance}]；ADD/UPDATE→upsert、DELETE→软删； sanitizeOps 防幻删(DELETE 须命中已有)/夹 importance[1,10]/同key保末个/丢 NOOP。 - 攒批：每 3 轮（per-session 计数）才 consolidate 一次，省成本，对齐 ChatGPT 周期整理。从根上解决 exact-key 盲写的记忆腐烂。验证：parseOps/sanitizeOps/parseProfile 纯逻辑单测；store 集成测试(真 PG)覆盖 importance/last_seen 写入 + 软删(live 0 / 物理 1)；dispatcher -race 全过。（注：完整多轮 LLM consolidate 未做实跑，属构造性验证 + 沿用已证 pool.Chat 模式。） P2 待做：读路径按 Score(Recency+Importance) 排序/衰减/截断 + 桌面端记忆面板。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-22 14:27:16 +08:00
parent b06c768f11
commit 1674252d81
8 changed files with 352 additions and 98 deletions
@@ -7,74 +7,157 @@ import (
 	"log"
 	"strings"

+	"github.com/cloudwego/eino/schema"
+
 	"github.com/sundynix/sundynix-dispatcher/internal/llm"
 	"github.com/sundynix/sundynix-shared/contract"
 )

-// Pref 是从对话抽取出的一条长期偏好（key/value）。
-type Pref struct {
-	Key   string `json:"key"`
-	Value string `json:"value"`
+// memOp 是 consolidate 的一条记忆操作（Mem0 式对账结果）。
+type memOp struct {
+	Op         string  `json:"op"` // ADD / UPDATE / DELETE / NOOP
+	Key        string  `json:"key"`
+	Value      string  `json:"value"`
+	Importance float64 `json:"importance"` // 1~10，poignancy（读路径权重）
+	Reason     string  `json:"reason"`
 }

-// extractMemory 写回阶段（异步、离开热路径）：从本轮对话用 LLM 抽取用户长期偏好，
-// 与已有画像去重后经 memory_upsert 登记。模型/工具不可用或输入过短则跳过。
-func (o *Orchestrator) extractMemory(ctx context.Context, uid, input, answer string) {
+// consolidateMemory 写回阶段（异步、攒批）：把近期对话与已有画像交给 LLM 一次对账，
+// 输出 ADD/UPDATE/DELETE/NOOP 操作并执行。ADD/UPDATE→upsert(带 importance)、DELETE→软删。
+// 一次 LLM 调用同时做"抽取+对账"，从根上解决 exact-key 盲写导致的记忆腐烂。
+func (o *Orchestrator) consolidateMemory(ctx context.Context, uid string, hist []*schema.Message) {
 	if uid == "" || o.tools == nil || o.pool == nil || !o.pool.Ready() {
 		return
 	}
-	if len([]rune(strings.TrimSpace(input))) < 2 || len([]rune(strings.TrimSpace(answer))) < 20 {
-		return // 太短，不值得抽取
+	dialogue := renderDialogue(hist)
+	if strings.TrimSpace(dialogue) == "" {
+		return
 	}
 	existing := parseProfile(o.fetchMemory(ctx, uid, ""))

 	cctx, cancel := llmCtx(ctx)
 	defer cancel()
-	sys := "你从对话中提取【用户的长期稳定偏好或事实】（如称呼、语言、职业、专业领域、口味、常用工具、固定要求等），" +
-		"忽略一次性的临时信息与你自己的话。"
-	user := fmt.Sprintf("用户输入：%s\n助手回答：%s\n请抽取。只输出 JSON 数组 [{\"key\":\"偏好维度\",\"value\":\"值\"}]，"+
-		"没有可抽取的就输出 []，不要任何多余文字。", truncate(input, 800), truncate(answer, 1200))
-	txt, err := o.pool.Chat(cctx, []llm.ChatMessage{{Role: "system", Content: sys}, {Role: "user", Content: user}})
+	sys := "你是用户长期记忆管理器。对照【已有记忆】与【近期对话】，只针对用户长期稳定的偏好/事实做对账，" +
+		"忽略一次性临时信息。"
+	txt, err := o.pool.Chat(cctx, []llm.ChatMessage{
+		{Role: "system", Content: sys},
+		{Role: "user", Content: consolidatePrompt(dialogue, existing)},
+	})
 	if err != nil {
-		log.Printf("[eino] (writeback) 偏好抽取失败 user=%s: %v", uid, err)
+		log.Printf("[eino] (writeback) 记忆对账失败 user=%s: %v", uid, err)
 		return
 	}
-	fresh := filterNewPrefs(parsePrefs(txt), existing)
-	for _, p := range fresh {
-		o.upsertMemory(ctx, uid, p.Key, p.Value)
+	ops := sanitizeOps(parseOps(txt), existing)
+	var add, del int
+	for _, op := range ops {
+		switch op.Op {
+		case "DELETE":
+			o.deleteMemory(ctx, uid, op.Key)
+			del++
+		default: // ADD / UPDATE
+			o.upsertMemory(ctx, uid, op.Key, op.Value, op.Importance)
+			add++
+		}
 	}
-	if len(fresh) > 0 {
-		log.Printf("[eino] (writeback) 已登记 %d 条新偏好 user=%s", len(fresh), uid)
+	if add+del > 0 {
+		log.Printf("[eino] (writeback) 记忆对账 user=%s：写入/更新 %d，删除 %d", uid, add, del)
 	}
 }

-// upsertMemory 经 mcp-go memory_upsert 工具登记一条偏好。
-func (o *Orchestrator) upsertMemory(ctx context.Context, uid, key, value string) {
-	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
-	defer cancel()
-	if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("memory_upsert"),
-		&contract.ToolCall{Tool: "memory_upsert", Args: map[string]any{"user_id": uid, "key": key, "value": value}}); err != nil {
-		log.Printf("[eino] memory_upsert 失败 %s=%s: %v", key, value, err)
+// consolidatePrompt 拼对账提示词：给出已有记忆 + 近期对话，要求输出操作列表。
+func consolidatePrompt(dialogue string, existing map[string]string) string {
+	var ex strings.Builder
+	if len(existing) == 0 {
+		ex.WriteString("（暂无）")
+	} else {
+		for k, v := range existing {
+			fmt.Fprintf(&ex, "- %s：%s\n", k, v)
+		}
 	}
+	return fmt.Sprintf(`【已有记忆】
+%s
+【近期对话】
+%s
+
+请对账并只输出 JSON 数组，每项 {"op":"ADD|UPDATE|DELETE|NOOP","key":"维度","value":"值","importance":1到10的整数,"reason":"简述"}：
+- ADD：对话中出现、已有记忆里没有的新长期偏好；
+- UPDATE：与已有某条同维度但值变了（**key 必须复用已有的那个**）；
+- DELETE：已有记忆被对话明确推翻/过时（key 用已有的）；
+- NOOP：无需变更的不要输出。
+importance 表示该偏好的重要程度（随口一提=低，强烈/反复强调=高）。只输出 JSON，无多余文字。`, strings.TrimRight(ex.String(), "\n"), dialogue)
 }

-// parsePrefs 解析 LLM 抽取结果（容忍 ```json 围栏）为 []Pref，过滤空项。
-func parsePrefs(txt string) []Pref {
-	var ps []Pref
-	if json.Unmarshal([]byte(stripFence(txt)), &ps) != nil {
+// renderDialogue 把近期消息渲染为"用户/助手"对话文本（取末尾若干条，限长）。
+func renderDialogue(hist []*schema.Message) string {
+	const maxMsgs = 8
+	start := 0
+	if len(hist) > maxMsgs {
+		start = len(hist) - maxMsgs
+	}
+	var b strings.Builder
+	for _, m := range hist[start:] {
+		role := "用户"
+		if m.Role == schema.Assistant {
+			role = "助手"
+		} else if m.Role == schema.System {
+			continue
+		}
+		fmt.Fprintf(&b, "%s：%s\n", role, truncate(m.Content, 500))
+	}
+	return strings.TrimRight(b.String(), "\n")
+}
+
+// parseOps 解析 LLM 对账结果（容忍 ```json 围栏）为 []memOp。
+func parseOps(txt string) []memOp {
+	var ops []memOp
+	if json.Unmarshal([]byte(stripFence(txt)), &ops) != nil {
 		return nil
 	}
-	out := make([]Pref, 0, len(ps))
-	for _, p := range ps {
-		p.Key, p.Value = strings.TrimSpace(p.Key), strings.TrimSpace(p.Value)
-		if p.Key != "" && p.Value != "" {
-			out = append(out, p)
+	return ops
+}
+
+// sanitizeOps 校验/规整操作：规范 op 名；ADD/UPDATE 需 key+value；DELETE 需 key 且必须命中已有；
+// importance 夹到 [1,10]；同 key 去重（保留末个）；过滤 NOOP。
+func sanitizeOps(ops []memOp, existing map[string]string) []memOp {
+	byKey := map[string]memOp{}
+	order := []string{}
+	for _, op := range ops {
+		op.Op = strings.ToUpper(strings.TrimSpace(op.Op))
+		op.Key = strings.TrimSpace(op.Key)
+		op.Value = strings.TrimSpace(op.Value)
+		if op.Key == "" {
+			continue
 		}
+		switch op.Op {
+		case "ADD", "UPDATE":
+			if op.Value == "" {
+				continue
+			}
+			if op.Importance < 1 {
+				op.Importance = 1
+			} else if op.Importance > 10 {
+				op.Importance = 10
+			}
+		case "DELETE":
+			if _, ok := existing[op.Key]; !ok { // 只删确实存在的，防误删/幻删
+				continue
+			}
+		default: // NOOP 或未知 → 丢弃
+			continue
+		}
+		if _, seen := byKey[op.Key]; !seen {
+			order = append(order, op.Key)
+		}
+		byKey[op.Key] = op
+	}
+	out := make([]memOp, 0, len(order))
+	for _, k := range order {
+		out = append(out, byKey[k])
 	}
 	return out
 }

-// parseProfile 把 memory_get 渲染的画像（"- 维度：值" 多行）解析回 map，供去重。
+// parseProfile 把 memory_get 渲染的画像（"- 维度：值" 多行）解析回 map，供对账。
 func parseProfile(s string) map[string]string {
 	m := map[string]string{}
 	for _, line := range strings.Split(s, "\n") {
@@ -95,19 +178,22 @@ func parseProfile(s string) map[string]string {
 	return m
 }

-// filterNewPrefs 保留新增或值有变化的偏好（同批同 key 去重；已有且相同则跳过）。
-func filterNewPrefs(extracted []Pref, existing map[string]string) []Pref {
-	out := make([]Pref, 0, len(extracted))
-	seen := map[string]bool{}
-	for _, p := range extracted {
-		if seen[p.Key] {
-			continue
-		}
-		seen[p.Key] = true
-		if cur, ok := existing[p.Key]; ok && cur == p.Value {
-			continue
-		}
-		out = append(out, p)
+// upsertMemory 经 mcp-go memory_upsert 登记/更新一条偏好（带 importance）。
+func (o *Orchestrator) upsertMemory(ctx context.Context, uid, key, value string, importance float64) {
+	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
+	defer cancel()
+	if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("memory_upsert"),
+		&contract.ToolCall{Tool: "memory_upsert", Args: map[string]any{"user_id": uid, "key": key, "value": value, "importance": importance}}); err != nil {
+		log.Printf("[eino] memory_upsert 失败 %s=%s: %v", key, value, err)
+	}
+}
+
+// deleteMemory 经 mcp-go memory_delete 软删一条偏好。
+func (o *Orchestrator) deleteMemory(ctx context.Context, uid, key string) {
+	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
+	defer cancel()
+	if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("memory_delete"),
+		&contract.ToolCall{Tool: "memory_delete", Args: map[string]any{"user_id": uid, "key": key}}); err != nil {
+		log.Printf("[eino] memory_delete 失败 %s: %v", key, err)
 	}
-	return out
 }
@@ -2,19 +2,50 @@ package eino

 import "testing"

-func TestParsePrefs(t *testing.T) {
-	got := parsePrefs("```json\n[{\"key\":\"称呼\",\"value\":\"Dexter\"},{\"key\":\"语言\",\"value\":\"中文\"},{\"key\":\"\",\"value\":\"空\"}]\n```")
-	if len(got) != 2 {
-		t.Fatalf("应解析出 2 条（过滤空 key），got %d: %v", len(got), got)
+func TestParseOps(t *testing.T) {
+	got := parseOps("```json\n[{\"op\":\"ADD\",\"key\":\"称呼\",\"value\":\"Dexter\",\"importance\":7},{\"op\":\"DELETE\",\"key\":\"旧公司\"}]\n```")
+	if len(got) != 2 || got[0].Op != "ADD" || got[0].Key != "称呼" || got[0].Importance != 7 || got[1].Op != "DELETE" {
+		t.Fatalf("parseOps 解析错: %+v", got)
 	}
-	if got[0].Key != "称呼" || got[0].Value != "Dexter" {
-		t.Errorf("解析错: %v", got[0])
-	}
-	if parsePrefs("不是 JSON") != nil {
+	if parseOps("不是 JSON") != nil {
 		t.Error("非 JSON 应返回 nil")
 	}
 }

+func TestSanitizeOps(t *testing.T) {
+	existing := map[string]string{"语言": "中文", "旧公司": "A厂"}
+	in := []memOp{
+		{Op: "add", Key: "重要事", Value: "X", Importance: 99}, // 大小写规范 + importance 夹到 10
+		{Op: "UPDATE", Key: "语言", Value: "英文", Importance: 0}, // importance 提到 1
+		{Op: "ADD", Key: "空值", Value: ""},                     // 无 value → 丢
+		{Op: "DELETE", Key: "旧公司"},                            // 命中已有 → 留
+		{Op: "DELETE", Key: "不存在"},                            // 未命中 → 丢（防幻删）
+		{Op: "NOOP", Key: "语言"},                               // NOOP → 丢（不覆盖已收的 UPDATE）
+		{Op: "ADD", Key: "称呼", Value: "Dexter", Importance: 7},
+		{Op: "ADD", Key: "称呼", Value: "Dex", Importance: 3}, // 同 key 重复 → 保留末个（值+importance）
+	}
+	out := sanitizeOps(in, existing)
+	if len(out) != 4 {
+		t.Fatalf("应剩 4 条（重要事/语言/删旧公司/称呼），得 %d: %+v", len(out), out)
+	}
+	byKey := map[string]memOp{}
+	for _, o := range out {
+		byKey[o.Key] = o
+	}
+	if byKey["重要事"].Importance != 10 {
+		t.Errorf("重要事 importance 应夹到 10: %+v", byKey["重要事"])
+	}
+	if byKey["语言"].Op != "UPDATE" || byKey["语言"].Importance != 1 {
+		t.Errorf("语言 应为 UPDATE 且 importance 提到 1: %+v", byKey["语言"])
+	}
+	if byKey["旧公司"].Op != "DELETE" {
+		t.Errorf("旧公司 应保留 DELETE: %+v", byKey["旧公司"])
+	}
+	if byKey["称呼"].Value != "Dex" || byKey["称呼"].Importance != 3 {
+		t.Errorf("称呼 应取末个 op（Dex/3）: %+v", byKey["称呼"])
+	}
+}
+
 func TestParseProfile(t *testing.T) {
 	m := parseProfile("- 称呼：Dexter\n- 语言: 中文\n\n- 职业：律师")
 	if m["称呼"] != "Dexter" || m["语言"] != "中文" || m["职业"] != "律师" {
@@ -24,20 +55,3 @@ func TestParseProfile(t *testing.T) {
 		t.Error("空画像应得空 map")
 	}
 }
-
-func TestFilterNewPrefs(t *testing.T) {
-	existing := map[string]string{"称呼": "Dexter", "语言": "中文"}
-	in := []Pref{
-		{"称呼", "Dexter"}, // 已有且相同 → 跳
-		{"语言", "英文"},     // 已有但变了 → 留
-		{"职业", "律师"},     // 新 → 留
-		{"职业", "工程师"},    // 同批重复 key → 跳（保留首个）
-	}
-	got := filterNewPrefs(in, existing)
-	if len(got) != 2 {
-		t.Fatalf("应剩 2 条（语言变更 + 新职业），got %d: %v", len(got), got)
-	}
-	if got[0].Key != "语言" || got[0].Value != "英文" || got[1].Key != "职业" || got[1].Value != "律师" {
-		t.Errorf("过滤结果不符: %v", got)
-	}
-}
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"sync"
 	"time"

 	"github.com/cloudwego/eino/schema"
@@ -46,6 +47,9 @@ type Orchestrator struct {
 	sink    TokenSink
 	tools   ToolCaller
 	exec    ExecSink
+
+	turnMu sync.Mutex     // 保护 turns（攒批计数，多任务 goroutine 共享）
+	turns  map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate
 }

 // NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。
@@ -169,8 +173,10 @@ func (o *Orchestrator) fetchHistory(ctx context.Context, sessionID string) []*sc
 	return msgs
 }

-// memorize 写回阶段：把本轮对话落进短期历史，并（TODO）抽取长期偏好记忆。
-// 异步执行，离开热路径。
+// consolidateEveryTurns 控制记忆对账的攒批节奏：每 N 轮 consolidate 一次（不逐轮，省成本）。
+const consolidateEveryTurns = 3
+
+// memorize 写回阶段（异步、离热路径）：落短期历史；每 N 轮做一次记忆对账（consolidate）。
 func (o *Orchestrator) memorize(t *contract.Task, answer string) {
 	uid, _ := t.Meta[contract.MetaUserID].(string)
 	sid, _ := t.Meta[contract.MetaSessionID].(string)
@@ -179,9 +185,20 @@ func (o *Orchestrator) memorize(t *contract.Task, answer string) {
 		o.appendHistory(sid, "assistant", answer)
 		log.Printf("[eino] (writeback) task %s 已落会话历史 session=%s", t.ID, sid)
 	}
-	if uid != "" {
-		// 从本轮对话抽取长期偏好 → 去重 → memory_upsert（离开热路径，已在 goroutine 内）。
-		o.extractMemory(context.Background(), uid, dsl.Compile(t.Graph).Query, answer)
+	if uid == "" || sid == "" || o.tools == nil {
+		return
+	}
+	// 攒批：累计轮次，每 N 轮才把近期对话与已有画像交给 LLM 对账一次。
+	o.turnMu.Lock()
+	if o.turns == nil {
+		o.turns = map[string]int{}
+	}
+	o.turns[sid]++
+	n := o.turns[sid]
+	o.turnMu.Unlock()
+	if n%consolidateEveryTurns == 0 {
+		ctx := context.Background()
+		o.consolidateMemory(ctx, uid, o.fetchHistory(ctx, sid))
 	}
 }