如何管理 AI Agent 的记忆之一：既记得牢、想得起，又不烧钱

flowchart TD
    A[用户请求] --> B[工作记忆<br/>当前上下文]
    B -->|检索相关的长期记忆| C[长期记忆]
    subgraph C [长期记忆]
        direction LR
        C1[情景记忆<br/>发生过的事]
        C2[语义记忆<br/>学到的事实]
        C3[程序记忆<br/>怎么做]
    end
    C -->|回答之后再更新记忆| D[记忆巩固<br/>合并 / 压缩]
    D -.回流.-> C

目标:
    做一个 REST API

当前步骤:
    先写 OpenAPI

已知约束:
    Java 21
    Spring Boot

待定问题:
    要不要上 OAuth?

昨天

用户要:
    - OAuth
    - Spring Security
    - JWT

结果:
    API 顺利完成。

{
    "task": "...",
    "timestamp": "...",
    "outcome": "success",
    "summary": "...",
    "embeddings": [ ... ]
}

用户偏好 Go。
用户在 X 公司。
项目跑在 Kubernetes 上。
数据库是 MySQL。

主语      谓语        宾语
用户      偏好        Go
项目      部署在      Kubernetes

(用户)-[:偏好]->(Go)
(项目)-[:部署于]->(Kubernetes)

部署服务

1. 构建 Docker 镜像
2. 推送镜像
3. 更新 Helm
4. 等待 rollout 完成

对话
  ↓
候选记忆（先当"待定"，不落库）
  ↓
重要性评估（值不值得永久记？）
  ↓
去重（是不是已经有了？）
  ↓
摘要（压成一句能复用的话）
  ↓
存储

hello
谢谢
好的

给下面这段对话打个 1-10 分。
这段内容值得被永久记住吗？

对话:
...

昨天：问了 Kubernetes。
今天：问了 K8S。
明天：问了 Pod。

用户正在搞 Kubernetes 基础设施。

score = 0.45 * 向量相似度
      + 0.25 * 新近度
      + 0.20 * 重要性
      + 0.10 * 图关系距离

flowchart TD
    Agent["AI Agent (LangGraph)"] --> API["记忆 API / MCP Server"]
    API --> Redis["工作缓存<br/>Redis"]
    API --> PG["语义存储<br/>PostgreSQL + pgvector"]
    API --> Qdrant["情景存储<br/>Qdrant"]
    Redis --> KG["知识图谱<br/>Neo4j / Graphiti"]
    PG --> KG
    Qdrant --> KG
    API -.触发.-> Consolidate["后台记忆巩固<br/>摘要 / 去重 / 衰减"]
    Consolidate -.整理后写回.-> PG

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE memories (
    id          BIGSERIAL PRIMARY KEY,
    user_id     TEXT NOT NULL,          -- 用户隔离：命名空间
    kind        TEXT NOT NULL,          -- semantic / episodic
    content     TEXT NOT NULL,          -- 提炼后的记忆，不是原始对话
    importance  REAL NOT NULL,          -- 1-10 重要性分
    source      TEXT,                   -- provenance：这条哪来的
    created_at  TIMESTAMPTZ DEFAULT now(),
    embedding   vector(1536)
);

-- 向量索引 + 用户维度过滤，检索又快又不串味
CREATE INDEX ON memories USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON memories (user_id, kind);

import numpy as np, psycopg2, json
from openai import OpenAI

client = OpenAI()  # 或任意 OpenAI 兼容网关
db = psycopg2.connect("dbname=postgres user=postgres password=pass host=localhost")

def embed(text: str) -> list[float]:
    return client.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding

def score_importance(text: str) -> float:
    """让 LLM 给这条内容打 1-10 分，只有值得长期记的才存"""
    prompt = (
        "给下面这段内容打 1-10 分，评估它是否值得被 AI 助手永久记住。"
        "用户的长期偏好、项目关键事实给高分；寒暄、临时性内容给低分。"
        f"只回一个数字。\n\n内容：{text}"
    )
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    try:
        return float(resp.choices[0].message.content.strip())
    except ValueError:
        return 5.0

def remember(user_id: str, text: str, kind="semantic",
             source="chat", threshold=7.0, dedup_sim=0.92):
    # 1) 重要性闸门：不够重要直接丢
    imp = score_importance(text)
    if imp < threshold:
        print(f"[skip] importance={imp:.1f} < {threshold}: {text[:30]}")
        return

    vec = embed(text)

    # 2) 去重：和已有记忆太像就不新增（真正生产里应改成"合并/更新"）
    with db.cursor() as cur:
        cur.execute(
            """SELECT id, content, 1 - (embedding <=> %s::vector) AS sim
               FROM memories WHERE user_id=%s
               ORDER BY embedding <=> %s::vector LIMIT 1""",
            (vec, user_id, vec),
        )
        row = cur.fetchone()
        if row and row[2] >= dedup_sim:
            print(f"[dedup] 命中已有记忆(sim={row[2]:.2f}): {row[1][:30]}")
            return

        # 3) 落库
        cur.execute(
            """INSERT INTO memories(user_id, kind, content, importance, source, embedding)
               VALUES (%s,%s,%s,%s,%s,%s)""",
            (user_id, kind, text, imp, source, vec),
        )
        db.commit()
        print(f"[store] importance={imp:.1f}: {text[:30]}")

def recall(user_id: str, query: str, k=3,
           w_sim=0.6, w_recency=0.25, w_imp=0.15):
    qvec = embed(query)
    with db.cursor() as cur:
        cur.execute(
            """
            SELECT content, importance,
                   1 - (embedding <=> %s::vector) AS sim,
                   -- 新近度：越新越接近 1，30 天前衰减到 ~0.37
                   exp(-EXTRACT(EPOCH FROM (now()-created_at))/2592000.0) AS recency
            FROM memories WHERE user_id=%s
            ORDER BY embedding <=> %s::vector LIMIT 20
            """,
            (qvec, user_id, qvec),
        )
        rows = cur.fetchall()

    scored = [
        (w_sim*sim + w_recency*rec + w_imp*(imp/10.0), content)
        for content, imp, sim, rec in rows
    ]
    scored.sort(reverse=True)
    return [c for _, c in scored[:k]]   # 只回 top-k，省 Token

remember("u1", "用户偏好 Go，明确不要 Java 方案", source="preference")
remember("u1", "谢谢，辛苦了")          # 会被 importance 闸门拦掉
remember("u1", "用户喜欢用 Go 语言")    # 会被 dedup 拦掉（和第一条太像）

print(recall("u1", "给他推荐个后端技术栈"))
# → ['用户偏好 Go，明确不要 Java 方案']

# memory_mcp_server.py
from mcp.server.fastmcp import FastMCP
from memory_store import remember, recall   # 就是上面实现的那两个函数

mcp = FastMCP("shared-memory")

@mcp.tool()
def save_memory(user_id: str, text: str, kind: str = "semantic") -> str:
    """存一条值得长期记住的信息（自动打分 + 去重，不重要的会被丢弃）。"""
    remember(user_id, text, kind=kind, source="mcp")
    return "ok"

@mcp.tool()
def search_memory(user_id: str, query: str, k: int = 3) -> list[str]:
    """按相关度检索最相关的 top-k 条记忆。"""
    return recall(user_id, query, k=k)

if __name__ == "__main__":
    mcp.run()   # 默认 stdio，三个工具都能连

// Claude Code: 项目根 .mcp.json（或用 `claude mcp add`）
{ "mcpServers": { "shared-memory": {
      "command": "python", "args": ["/path/to/memory_mcp_server.py"] } } }

// OpenCode: opencode.json 的 mcp 段
{ "mcp": { "shared-memory": {
      "type": "local", "command": ["python", "/path/to/memory_mcp_server.py"] } } }

// Codex: 同样支持在其配置里声明 MCP server（配置键以你的版本为准）

---
name: shared-memory
description: 在跨会话记住用户偏好和项目事实时使用。当用户表达长期偏好、
  确认项目关键事实、或说"记住…"时，调用 save_memory；在开始新任务、
  需要用户偏好或项目背景时，先调用 search_memory 回忆。
---

# 记忆使用规矩

## 什么时候“存”（save_memory）
- 用户说出长期偏好：如"我用 Go 不用 Java""提交信息用中文"
- 确认了项目关键事实：技术栈、CI 平台、数据库、目录约定
- 用户明确说"记住这个"
- ⚠️ 别存：寒暄、临时性问题、一次性的调试细节

## 什么时候“取”（search_memory）
- 每个新任务开始前，先按任务关键词检索一次
- 要给建议、选技术栈、写配置前，先看看有没有相关偏好
- 用户说"接着上次的做"时，务必先检索

## 原则
- 存之前先想：三个月后这条还有用吗？没用就别存。
- 取回来的记忆当作"已知事实"，别再反问用户。

flowchart TD
    subgraph Tools["三个编码 Agent"]
        CC[Claude Code]
        CX[Codex]
        OC[OpenCode]
    end
    Skill["AI Skill<br/>什么时候存 / 什么时候取"]
    MCP["AI Tools / MCP<br/>save_memory · search_memory"]
    DB["共享记忆后端<br/>PostgreSQL + pgvector"]

    CC --> Skill
    CX --> Skill
    OC --> Skill
    Skill --> MCP
    MCP --> DB

@startmindmap
<style>
mindmapDiagram {
  node {
    BackgroundColor #F8F9FA
    RoundCorner 10
    Padding 10
    FontSize 13
  }
  :depth(0) {
    BackgroundColor #1E3A5F
    FontColor white
    FontSize 18
    FontStyle bold
  }
  :depth(1) {
    FontSize 15
    FontStyle bold
  }
  :depth(2) {
    FontSize 13
  }
}
</style>

* AI Agent 记忆管理
** 核心问题
*** 该记什么
*** 怎么想起来
*** 什么时候忘
*** 一股脑存向量库是坑
** 分四层（像人脑）
*** 工作记忆（RAM/当前）
*** 情景记忆（发生过的事）
*** 语义记忆（事实/三元组）
*** 程序记忆（怎么做）
** 生命周期
*** 候选→重要性→去重→摘要→存
*** LLM 重要性打分
*** 记忆巩固（合并压缩）
** 检索
*** 别全捞回来
*** 相似度+新近+重要+图距离
** 存储选型
*** 按用途分开存
*** pgvector 常常够用
*** Mem0/Zep/LangGraph
** 参考架构
*** 记忆做成独立服务
*** API/MCP 暴露
*** 多 Agent 共享
** 经验 checklist
*** 存事实不存每句话
*** 狠去重、持续摘要
*** 支持遗忘、分命名空间
@endmindmap