点击开始动手实验


背景痛点:当对话越长,AI 越“健忘”

在生产级对话系统里,上下文管理是绕不过去的“硬骨头”。ChatGPT 看似能聊很久,实则受限于 4K/8K/32K 的 token 天花板;一旦超限,早期信息被无情截断,用户却以为它“记得”。更糟的是:

  • 记忆碎片化:多轮会话被拆成独立请求,服务端默认无状态,导致“昨天订的机票”今天再也找不到。
  • 长期记忆检索效率:把几十轮历史全塞进 prompt 做向量相似度,延迟高、精度低,还浪费钱。
  • 数据一致性:同一用户在不同设备提问,返回的答案前后矛盾,体验瞬间“出戏”。

这些痛点倒逼我们在系统层做“外挂记忆”,而不是单纯依赖模型窗口。

技术对比:三种存储方案谁更适合对话场景

方案 优势 劣势 适用场景
向量数据库(Pinecone、Weaviate) 语义检索精准、可水平扩展 写入后需索引刷新,延迟 100ms+ 百万级记忆、语义模糊查询
KV 存储(Redis、DynamoDB) 毫秒级点查、成本低 仅支持精确 key,无语义能力 用户画像、短期缓存
关系型数据库(PostgreSQL) 事务一致、复杂过滤 全文/向量检索需插件,扩容重 强一致性账单、订单类记忆

结论:向量库存“说了什么”,KV 存“是谁说的”,关系型存“事实源”。三者互补,而非互斥。

核心实现:注意力 + 外部存储的双层记忆

ChatGPT 官方 Memory 机制并未开源,但从 API 行为可反推出两条主线:

  1. 短期上下文:仍走 Transformer 自注意力,窗口内 token 全部加载,保证细节不丢失。

  2. 长期记忆池:服务端维护一张用户级 Memory Table,结构类似 (user_id, memory_id, text_emb, summary, timestamp, ttl)
    每次请求前,系统用当前 query 向量做 ANN 搜索,取 Top-K 相关记忆,经重排序后拼成“系统提示”注入对话,模型据此生成带记忆的回复。
    关键算法流程:

    query_vec = embed(query)
    cand = vector_search(query_vec, top_k=20)
    cand = rerank_by_time_and_score(cand)  # 时间衰减 + 分数
    memory_prompt = format_to_prompt(cand[:5])
    final_prompt = system_prompt + memory_prompt + chat_history
    

    通过“向量召回 + 精排 + 动态注入”,既突破窗口限制,又避免一次性灌入过多噪声。

代码示例:用 FAISS 做轻量级记忆检索

下面给出最小可运行片段,演示如何把用户历史对话编码、入库、再召回。依赖:faiss-cpusentence-transformersnumpy

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer
from datetime import datetime
import json, os

EMB_DIM = 768
INDEX_FILE = "memory.index"
META_FILE = "memory_meta.json"

class MemoryStore:
    def __init__(self):
        self.encoder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        if os.path.exists(INDEX_FILE):
            self.index = faiss.read_index(INDEX_FILE)
            with open(META_FILE) as f:
                self.meta = json.load(f)  # list of dict
        else:
            self.index = faiss.IndexFlatIP(EMB_DIM)  # 内积相似度
            self.meta = []

    def add(self, text: str, user_id: str, ttl_days: int = 30):
        """新增一条记忆"""
        vec = self.encoder.encode([text], normalize_embeddings=True).astype("float32")
        self.index.add(vec)
        self.meta.append({
            "text": text,
            "user_id": user_id,
            "ts": datetime.utcnow().isoformat(),
            "ttl_days": ttl_days
        })
        self._save()

    def search(self, query: str, user_id: str, top_k: int = 5):
        """仅返回该用户的记忆"""
        if self.index.ntotal == 0:
            return []
        qvec = self.encoder.encode([query], normalize_embeddings=True).astype("float32")
        scores, idxs = self.index.search(qvec, top_k * 3)  # 先放大窗口
        results = []
        for score, idx in zip(scores[0], idxs[0]):
            if idx == -1:
                continue
            item = self.meta[idx]
            if item["user_id"] != user_id:
                continue
            results.append((item["text"], float(score)))
            if len(results) >= top_k:
                break
        return results

    def _save(self):
        faiss.write_index(self.index, INDEX_FILE)
        with open(META_FILE, "w") as f:
            json.dump(self.meta, f, ensure_ascii=False, indent=2)

# 使用示例
if __name__ == "__main__":
    store = MemoryStore()
    store.add("我喜欢坐靠窗的位置", user_id="alice")
    store.add("我对花生过敏", user_id="alice")
    hits = store.search("我不能吃什么?", user_id="alice")
    for text, score in hits:
        print(f"{score:.2f} {text}")

要点解读:

  • 采用内积索引 + 归一化向量,等价于余弦相似度,速度快。
  • 先放大 top_k*3 再按用户过滤,避免向量库返回结果不足。
  • 落盘双文件:.index 存向量,.json 存元数据,方便 TTL 与后期清理。

性能考量:生产环境的三座大山

  1. 内存占用IndexFlatIP 全内存,768 维 × 4 字节 × 1 千万条 ≈ 28 GB,破亿级请换 IndexIVFPQHNSW
  2. 检索延迟IndexFlat 在 100 万条内 < 5 ms;超过后考虑分片或把热数据放 Redis,向量库仅存冷数据。
  3. 数据一致性:写操作先落日志(Kafka/Pulsar),后台异步建索引,读链路用 Redis 缓存最新 1 分钟记录,做到“最终一致”。

避坑指南:踩出来的 5 条血泪经验

  1. 记忆也要“减肥”:长对话先让 LLM 总结成 50 字摘要再入库,向量维度不变,存储减半,精度损失 < 2%。
  2. 时间衰减权重:召回后按 score * exp(-Δt/τ) 重新排序,防止 3 个月前的“我喜欢可乐”覆盖刚说的“我在戒糖”。
  3. 冷启动优化:新用户无记忆时,用“同类用户”聚类模板兜底,避免空记忆导致答非所问。
  4. 敏感字段加密:记忆落盘前对手机、地址做 NER 脱敏,再 AES 加密,合规审计少踩坑。
  5. 灰度删除:TTL 到期不立即物理删除,先打“已失效”标,7 天内可快速回滚,防止用户投诉“记忆消失术”。

开放问题:你的 NLP 项目需要哪种记忆?

  • 如果做智能客服,你会把订单状态放关系型,把 FAQ 放向量库,还是全量丢给 LLM 做 Few-shot?
  • 多租户场景下,如何既保证用户隔离,又利用“群体知识”提升整体回答质量?
  • 记忆压缩与隐私合规永远矛盾,你会优先缩减存储成本,还是加密粒度更细?

期待在评论区看到你的实战思路,一起把“会遗忘的 AI”变成“有记忆的朋友”。


我按上面的思路撸了一遍后,发现从零手搓虽然能跑,但真要把“听、想、说”串成低延迟的实时通话,还要兼顾回声消除、语音活动检测、流式 ASR/TTS,坑比想象的多。后来直接上手了从0打造个人豆包实时通话AI动手实验,官方把火山引擎的豆包语音全家桶都包好了:流式语音识别、大模型对话、音色克隆、WebRTC 一键跑通,半小时就能在浏览器里跟 AI 语音唠嗑。小白也能顺利体验,建议先跑通 baseline,再把自己优化的 Memory 模块插进去,效率翻倍。

点击开始动手实验


Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐