从入门到实战：理解short-term memory与long-term memory在AI应用中的协同机制

1次阅读

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

在构建智能对话系统时，记忆管理是一个核心挑战。作为刚接触 AI 开发的新手，理解 short-term memory（短期记忆）和 long-term memory（长期记忆）的差异及协同方式尤为重要。本文将带你从基础概念到代码实践，逐步掌握这两种记忆机制的应用方法。

Short-term memory（短期记忆）通常指系统在单次会话中临时保存的上下文信息，比如最近几次对话的内容。它有以下特点：

存储在内存中，访问速度快
容量有限，通常只保留最近的几条记录
会话结束后自动清除

Long-term memory（长期记忆）则是系统长期积累的知识库，比如存储在向量数据库中的文档或问答对。它的特点是：

持久化存储，数据量大
检索速度相对较慢
需要定期更新维护

特性	Short-term memory	Long-term memory
存储介质	内存	数据库 / 向量库
访问延迟	纳秒级	毫秒级
典型容量	KB~MB 级	GB~TB 级
数据生命周期	会话级	永久
典型实现	LRU 缓存	FAISS/Pinecone

from functools import lru_cache
from typing import Dict, Any

class ConversationContext:
    """
    使用 LRU 缓存实现的短期记忆管理器
    :param max_size: 最大缓存对话轮数
    """
    def __init__(self, max_size: int = 5):
        self._cache = {}
        self.max_size = max_size

    def add_message(self, session_id: str, role: str, content: str) -> None:
        """添加对话消息到上下文"""
        if session_id not in self._cache:
            self._cache[session_id] = []

        # 维护缓存大小
        if len(self._cache[session_id]) >= self.max_size:
            self._cache[session_id].pop(0)

        self._cache[session_id].append({'role': role, 'content': content})

    def get_context(self, session_id: str) -> list[Dict[str, Any]]:
        """获取当前会话的完整上下文"""
        return self._cache.get(session_id, [])

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class KnowledgeBase:
    """
    基于 FAISS 构建的长期记忆系统
    :param model_name: 使用的嵌入模型名称
    :param dim: 向量维度
    """def __init__(self, model_name: str ='all-MiniLM-L6-v2', dim: int = 384):
        self.encoder = SentenceTransformer(model_name)
        self.index = faiss.IndexFlatL2(dim)
        self.documents = []

    def add_document(self, text: str) -> None:
        """添加文档到知识库"""
        embedding = self.encoder.encode(text)
        self.index.add(np.array([embedding]))
        self.documents.append(text)

    def search(self, query: str, k: int = 3) -> list[str]:
        """检索最相关的 k 个文档"""
        query_embed = self.encoder.encode(query)
        _, indices = self.index.search(np.array([query_embed]), k)
        return [self.documents[i] for i in indices[0] if i < len(self.documents)]

class HybridMemorySystem:
    """结合短期和长期记忆的混合系统"""
    def __init__(self):
        self.short_memory = ConversationContext()
        self.long_memory = KnowledgeBase()

    def respond(self, session_id: str, user_input: str) -> str:
        """
        生成响应：1. 将用户输入加入短期记忆
        2. 从长期记忆检索相关信息
        3. 结合上下文生成回复
        """
        # 更新短期记忆
        self.short_memory.add_message(session_id, "user", user_input)
        context = self.short_memory.get_context(session_id)

        # 从长期记忆检索
        relevant_info = self.long_memory.search(user_input)

        # 这里应该是实际的 LLM 生成逻辑
        return f"基于上下文和知识库生成的回复。相关参考：{relevant_info[:1]}"