Memory V1：让 AI 记住你的关键信息

目标：

• 每次用户发言后，后端抽取“用户事实”
• 按 session_id 保存记忆
• 聊天时自动把记忆拼到 system prompt
• 前端右侧展示“当前会话记忆”

项目就从“多会话聊天”升级成了“有记忆的 AI”。

1）新增文件

server/memory_store.py

import json
import os
from typing import Dict, List

MEMORY_FILE = "memory_store.json"


def _load() -> Dict[str, List[str]]:
    if not os.path.exists(MEMORY_FILE):
        return {}
    try:
        with open(MEMORY_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def _save(data: Dict[str, List[str]]):
    with open(MEMORY_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def get_session_memories(session_id: str) -> List[str]:
    data = _load()
    return data.get(session_id, [])


def add_session_memories(session_id: str, new_memories: List[str]):
    if not session_id or not new_memories:
        return

    data = _load()
    old_memories = data.get(session_id, [])

    merged = old_memories[:]
    for item in new_memories:
        item = item.strip()
        if item and item not in merged:
            merged.append(item)

    data[session_id] = merged[:20]
    _save(data)

2） server/app.py

新增 import

from typing import List, Literal, Optional
from fastapi import FastAPI, HTTPException
from memory_store import get_session_memories, add_session_memories

新增一个响应模型

class MemoryResponse(BaseModel):
    memories: List[str]

新增两个函数

def build_memory_prompt(memories: List[str]) -> str:
    if not memories:
        return ""

    lines = "n".join([f"- {item}" for item in memories])
    return f"""以下是你已经记住的用户信息，请在回复时自然参考，但不要生硬复述：
{lines}
"""


def extract_user_memories(user_text: str) -> List[str]:
    text = (user_text or "").strip()
    if not text:
        return []

    prompt = f"""
你是一个信息抽取助手。
请从下面这段用户发言中，提取“适合长期记忆的用户事实”。

要求：
1. 只提取长期有价值的信息，比如：职业、兴趣、目标、偏好、正在做的项目
2. 不要提取一次性寒暄
3. 每条一句话，简洁
4. 最多返回 5 条
5. 只返回 JSON 数组，不要输出其他内容

用户发言：
{text}
"""

    try:
        completion = client.ch@t.completions.create(
            model=MODEL_NAME,
            messages=[
                {"role": "system", "content": "你是一个严谨的用户事实提取助手。"},
                {"role": "user", "content": prompt},
            ],
            temperature=0.2,
        )

        content = completion.choices[0].message.content or "[]"
        import json
        result = json.loads(content)

        if isinstance(result, list):
            return [str(item).strip() for item in result if str(item).strip()]
        return []
    except Exception:
        return []

更新 /api/ch@t

@app.post("/api/ch@t", response_model=ChatResponse)
def ch@t(req: ChatRequest):
    messages = [m.model_dump() for m in req.messages]

    session_memories = get_session_memories(req.session_id or "")
    memory_prompt = build_memory_prompt(session_memories)

    final_messages = []
    for item in messages:
        if item["role"] == "system":
            final_messages.append({
                "role": "system",
                "content": f'{item["content"]}nn{memory_prompt}'.strip()
            })
        else:
            final_messages.append(item)

    try:
        completion = client.ch@t.completions.create(
            model=MODEL_NAME,
            messages=final_messages,
            temperature=0.7,
        )
        reply = completion.choices[0].message.content or ""
    except Exception as e:
        print("ch@t error:", e)
        reply = "AI服务异常，请稍后再试"

    if req.session_id:
        latest_user_text = ""
        for item in reversed(messages):
            if item["role"] == "user":
                latest_user_text = item["content"]
                break

        if latest_user_text:
            new_memories = extract_user_memories(latest_user_text)
            add_session_memories(req.session_id, new_memories)

    return ChatResponse(reply=reply)

新增一个查看记忆接口

@app.get("/api/memory/{session_id}", response_model=MemoryResponse)
def get_memory(session_id: str):
    if not session_id:
        raise HTTPException(status_code=400, detail="session_id不能为空")
    return MemoryResponse(memories=get_session_memories(session_id))

3） web/src/App.vue

先改 import

import { computed, ref, watch, onMounted } from 'vue'

新增状态

放在 loading 下面：

const memoryList = ref([])

新增获取记忆方法

放在 formatTime 下面：

const fetchMemories = async () => {
  if (!currentSessionId.value) return
  try {
    const res = await axios.get(`${currentSessionId.value}`)
    memoryList.value = res.data.memories || []
  } catch (error) {
    memoryList.value = []
    console.error(error)
  }
}

新增生命周期和

放在 watch(sessions...) 后面：

watch(currentSessionId, () => {
  fetchMemories()
})

onMounted(() => {
  fetchMemories()
})

在 sendMessage 成功后，追加一行

找到这里：

updateCurrentSession(session => ({
  ...session,
  updatedAt: Date.now(),
  messages: [
    ...session.messages,
    {
      role: 'assistant',
      content: res.data.reply,
    },
  ],
}))

紧接着加：

await fetchMemories()

模板里给 .container 增加右侧记忆面板

<aside class="memory-panel">
  <div class="memory-header">会话记忆</div>
  <div v-if="memoryList.length" class="memory-list">
    <div v-for="(item, index) in memoryList" :key="index" class="memory-item">
      {{ item }}
    </div>
  </div>
  <div v-else class="memory-empty">暂时还没有提取到长期记忆</div>
</aside>

样式新增

.memory-panel {
  width: 280px;
  background: #fff;
  border-radius: 16px;
  padding: 16px;
  box-sizing: border-box;
  height: calc(100vh - 48px);
  overflow-y: auto;
}
.memory-header {
  font-size: 18px;
  font-weight: 600;
  margin-bottom: 12px;
}
.memory-list {
  display: flex;
  flex-direction: column;
  gap: 10px;
}
.memory-item {
  padding: 12px;
  border-radius: 12px;
  background: #f9fafb;
  line-height: 1.6;
  color: #111827;
  border: 1px solid #e5e7eb;
}
.memory-empty {
  color: #6b7280;
  font-size: 14px;
}

4）做完后效果

现在会变成这样：

• 你说：“我是前端工程师，最近在做 AI Agent 项目”

• 后端会抽取出类似记忆：

  • 我是前端工程师
  • 最近在做 AI Agent 项目

• 下次继续聊时，AI 会自动参考这些记忆

• 右侧面板会把这些记忆展示出来

5）这版的价值

这一版虽然还不是向量记忆，但已经具备 3 个非常重要的点：

• 记忆抽取
• 记忆持久化
• 记忆注入上下文

可以讲：

再问第二次时

会话记忆存在