从本地上下文到向量数据库：构建高效记忆系统的技术选型与实践

1次阅读

共计 1439 个字符，预计需要花费 4 分钟才能阅读完成。

在开发对话系统或推荐引擎时，我们常遇到这样的矛盾：既要快速访问近期交互数据（如最后 5 轮对话），又要从海量历史数据中检索相似场景。前者需要微秒级响应的本地存储，后者要求 GB 级容量的持久化方案。这种短期记忆与长期记忆的协同管理，直接决定了 AI 应用的响应速度和智能水平。

读写延迟 ：0.1~1 微秒（直接内存访问）
容量上限 ：通常限制在进程可用内存（如 4GB）
典型场景 ：存储最近 10 条对话的原始文本和元数据

读写延迟 ：0.1~1 毫秒（网络往返时间占主导）
容量上限 ：取决于机器配置（单实例通常 10~100GB）
优势：支持 TTL 过期和持久化

查询延迟 ：10~100 毫秒（含网络通信和 ANN 搜索）
容量上限 ：支持 PB 级向量存储
核心能力 ：近似最近邻搜索（ANN）和动态索引

flowchart LR
    A[用户输入] --> B{本地 LRU 缓存?}
    B -- 命中 --> C[立即响应]
    B -- 未命中 --> D[查询向量数据库]
    D --> E[更新缓存并返回]

import numpy as np
from lru import LRU
from pinecone import Pinecone

class HybridMemory:
    def __init__(self, cache_size=100, dim=384):
        # 本地缓存使用 LRU 策略
        self.cache = LRU(cache_size)  # O(1) 时间复杂度
        # 连接 Pinecone 长期存储
        self.pc = Pinecone(api_key="YOUR_KEY")
        self.index = self.pc.Index("memory-vectors")

    def query(self, text: str) -> dict:
        """查询记忆系统（含缓存穿透处理）"""
        # 1. 先查本地缓存
        if text in self.cache:
            return {"source": "cache", "data": self.cache[text]}

        # 2. 缓存未命中时查询向量库
        embedding = self._generate_embedding(text)  # 假设已实现
        res = self.index.query(vector=embedding, top_k=3)

        # 3. 更新缓存（异步）self.cache[text] = res.matches
        return {"source": "vector_db", "data": res.matches}