AI大模型企业应用实战(17)-利用memory为LLM解决长短时记忆问题

JavaEdge

修改于 2026-05-08 07:41:39

4190

文章被收录于专栏：AIGC大模型应用AIGC大模型应用

本文已收录在Github，关注我，紧跟本系列专栏文章，咱们下篇再续！

🚀 魔都架构师 | 全网30W技术追随者
🔧 大厂分布式系统/数据中台实战专家
🏆 主导交易系统百万级流量调优 & 车联网平台架构
🧠 AIGC应用开发先行者 | 区块链落地实践者
🌍 以技术驱动创新，我们的征途是改变世界！
👉 实战干货：编程严选网

1 内存实现短时记忆

将完整对话历史（HumanMessage/AIMessage）缓存于内存，不持久化。适合短对话，但长对话可能导致上下文过长。

实现

# 创建简单的缓冲区记忆，存储所有对话历史（用户和AI消息）
memory = ConversationBufferMemory()
memory.chat_memory.add_user_message("你好，我是人类！")
memory.chat_memory.add_ai_message("你好，我是AI，有什么可以帮助你的吗？")

# 返回包含所有历史对话的变量，用于后续LLM调用
memory.load_memory_variables({})

output：
{'history': 'Human: 你好，我是人类！\nAI: 你好，我是AI，有什么可以帮助你的吗？'}

内部结构：依赖 ChatMessageHistory 管理消息列表。

适用场景

短对话演示场景，无需复杂管理。

缺点

无长度控制，对话越长 prompt 越臃肿

2 对话窗口记忆

实现一个最近的对话窗口，超过窗口条数的对话将被删除。

滑动窗口机制：控制记忆大小，避免内存溢出。save_context 添加新上下文。

实现

from langchain_classic.memory import  ConversationBufferWindowMemory

# 核心参数：`k=N` 只保留最近 N 轮对话
# 丢弃策略：超出窗口的旧消息被彻底丢弃
memory = ConversationBufferWindowMemory(k=1)

memory.save_context({"input":"你好，我是人类!"},{"output":"你好，我是AI，有什么可以帮助你的吗？"})
memory.save_context({"input":"我想吃鸡肉"},{"output":"好的，我帮你找找鸡肉的做法"})

memory.load_memory_variables({})

output：
{'history': 'Human: 我想吃鸡肉\nAI: 好的，我帮你找找鸡肉的做法'}

适用

简单多轮问答，token 成本敏感场景

优点

高效，防止长对话累积。

3 利用Entity memory构建实体记忆

利用 LLM 自动提取对话中的命名实体，维护实体→描述的映射表。

工作流程

实体提取 → 增量更新存储 → 新问题匹配已有实体并注入 prompt

示例：从"胡八一和王胖子雪莉杨经常在一起冒险"提取实体"胡八一""王胖子""雪莉杨"

实现

memory = ConversationEntityMemory(llm=llm)
_input = {
    "input":"胡八一和王胖子雪莉杨经常在一起冒险，合称盗墓铁三角."
}
memory.load_memory_variables(_input)
memory.save_context(
    _input,
    {
        "output":"听起来很刺激，我也想加入他们！"
    }
)

memory.load_memory_variables({"input":"铁三角是谁?"})

4 利用知识图谱来构建记忆

将对话信息转化为知识三元组（Subject → Predicate → Object）。

v.s EntityMemory

EntityMemory 是键值对，KGMemory 是图结构，支持关系链推理

实现

memory = ConversationKGMemory(llm=llm,return_messages=True)

memory.save_context(
    {"input":"帮我找一下JavaEdge"},
    {"output":"对不起请问什么是JavaEdge？"}
)

memory.save_context(
    {"input":"JavaEdge 是一个技术大佬"},
    {"output":"好的，我知道了。"}
)

memory.load_memory_variables({"input":"JavaEdge 是谁?"})
# get_current_entities(query)提取实体
memory.get_current_entities("JavaEdge 最喜欢做什么事?")
# get_knowledge_triplets(query)提取关系三元组
memory.get_knowledge_triplets("JavaEdge 最喜欢打游戏")

5 利用对话摘要来兼容内存中的长对话

5.1 SummaryMemory

当对话持续进行且对话内容很多时，可用ConversationSummaryBufferMemory来存储对话摘要。

memory = ConversationSummaryMemory(llm=llm)
memory.save_context(
    {"input":"帮我找一下 JavaEdge"},
    {"output":"对不起请问什么是 JavaEdge？"}
)
memory.save_context(
    {"input":"JavaEdge 是一个技术大佬"},
    {"output":"好的，我知道了。"}
)

memory.load_memory_variables({})

output：
{'history': '\n\n人类请求AI查找“JavaEdge”，AI起初表示不清楚并询问具体指什么，随后人类解释该词指的是一位技术大佬，AI表示知晓。'}

messages = memory.chat_memory.messages
# 摘要生成的核心方法
memory.predict_new_summary(messages,"")

output：
'\n\n人类请求AI查找JavaEdge，AI询问其具体含义，人类解释JavaEdge是一位技术大佬，AI随后表示已了解。'

ChatMessageHistory

快速获得对话摘要：

hisiory = ChatMessageHistory()
hisiory.add_user_message("你好，我是人类！")
hisiory.add_ai_message("你好，我是AI小丸子，有什么可以帮助你的吗？")

memory = ConversationSummaryMemory.from_messages(
    llm,
   chat_memory=hisiory,
   return_messages=True
)

memory = ConversationSummaryMemory(
    llm,
    return_messages=True,
    buffer="\nThe AI introduces itself as AI Little Maruko and asks if there is anything it can help the human with.",
    chat_memory=hisiory
)

# memory.buffer
memory.load_memory_variables({})

output：
{'history': [SystemMessage(content='\nThe AI introduces itself as AI Little Maruko and asks if there is anything it can help the human with.', additional_kwargs={}, response_metadata={})]}

5.2 SummaryBufferMemory

当对话持续进行且对话内容很多时，可用ConversationSummaryBufferMemory来存储对话摘要。这是一种非常有用方式，它按token数来自动判断是否需要进行摘要。当token数量超过阈值，自动进行摘要。

在缓冲区中，会保留最近的k条对话。比较久的对话会被删除，在删除前会进行摘要。

# 为不原生支持 get_num_tokens_from_messages 的模型注入 token 计数器
# 继承 ChatOpenAI 以保证 Pydantic 类型校验通过
import tiktoken
from langchain_openai import ChatOpenAI
enc = tiktoken.get_encoding("cl100k_base")
def token_counter(messages):
    text = "".join(m.content or "" for m in messages)
    return len(enc.encode(text))

class _TokenCountingLLM(ChatOpenAI):
    def get_num_tokens_from_messages(self, messages):
        return token_counter(messages)

# 重建 llm，使其同时具备 ChatOpenAI 类型和 token 计数能力
llm = _TokenCountingLLM(
    model=llm.model_name,
    openai_api_base=llm.openai_api_base,
    openai_api_key=llm.openai_api_key,
    streaming=llm.streaming,
    temperature=llm.temperature,
)

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=10, # 控制何时触发摘要/丢弃
    return_messages=True
)
memory.save_context(
    {"input":"帮我找一下JavaEdge"},
    {"output":"对不起请问什么是JavaEdge？"}
)
memory.save_context(
    {"input":"JavaEdge是一个培训讲师"},
    {"output":"好的，我知道了。"}
)
memory.save_context(
    {"input":"今天他要讲一门关于RAG的课程"},
    {"output":"好的，我知道了。需要RAG的资料吗？"}
)

memory.load_memory_variables({})

5.3 使用token来刷新内存缓冲区

Conversation Token Buffer使用token长度来决定什么时候刷新内存

def token_counter(messages):
    text = "".join(m.content or "" for m in messages)
    return len(enc.encode(text))

class _TokenCountingLLM(ChatOpenAI):
    def get_num_tokens_from_messages(self, messages):
        return token_counter(messages)

# 重建 llm，使其同时具备 ChatOpenAI 类型和 token 计数能力
llm = _TokenCountingLLM(
    model=llm.model,
    openai_api_base=llm.openai_api_base,
    openai_api_key=llm.openai_api_key,
    streaming=llm.streaming,
    temperature=llm.temperature,
)

memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=150
)
memory.save_context(
    {"input":"帮我找一下JavaEdge"},
    {"output":"对不起请问什么是JavaEdge？"}
)
memory.save_context(
    {"input":"JavaEdge是一个技术大佬"},
    {"output":"好的，我知道了。"}
)
memory.save_context(
    {"input":"今天他要写一门关于RAG的专栏"},
    {"output":"好的，我知道了。需要RAG的资料吗？"}
)
memory.save_context(
    {"input":"不需要资料了，谢谢"},
    {"output":"好的，那我就不打扰你了。"}
)

memory.load_memory_variables({})

5.4 小结

类型	存储策略	Token 效率
SummaryMemory	全文摘要压缩	最高
SummaryBufferMemory	原始消息+摘要混合，token 阈值触发压缩	高
TokenBufferMemory	保留原始消息，按 token 数丢弃最早消息	中

6 长时记忆的实现

通过向量数据库存储之前的对话内容，有的向量数据库服务提供自动摘要等，每次对话时，都会从向量数据库里查询最相关的文档或历史对话。

6.1 架构设计

flowchart LR
    A[对话文本] --> B[Embedding 向量化]
    B --> C[FAISS 存储]
    D[用户查询] --> E[查询向量化]
    E --> F[FAISS 语义检索]
    C --> F
    F --> G[返回相似对话片段]

    classDef input fill:#E3F2FD,stroke:#1E88E5,stroke-width:2px,color:#0D47A1;
    classDef process fill:#E8F5E9,stroke:#43A047,stroke-width:2px,color:#1B5E20;
    classDef storage fill:#FFF3E0,stroke:#FB8C00,stroke-width:2px,color:#E65100;
    classDef output fill:#FCE4EC,stroke:#D81B60,stroke-width:2px,color:#880E4F;

    class A,D input;
    class B,E,F process;
    class C storage;
    class G output;

关键类：VectorStoreRetrieverMemory 封装检索器
优势：跨 Session 持久化、语义检索、可扩展（支持 Pinecone/Chroma 等）

6.2 实现

memory = ConversationBufferMemory()
memory.save_context(
    {"input":"帮我找一下JavaEdge"},
    {"output":"对不起请问什么是JavaEdge？"}
)
memory.save_context(
    {"input":"JavaEdge是一个培训讲师"},
    {"output":"好的，我知道了。"}
)
memory.save_context(
    {"input":"今天他要讲一门关于RAG的课程"},
    {"output":"好的，我知道了。需要RAG的资料吗？"}
)
memory.save_context(
    {"input":"不需要资料了，谢谢"},
    {"output":"好的，那我就不打扰你了。"}
)

vectorstore = FAISS.from_texts(
    memory.buffer.split("\n"),
    create_qwen_embeddings()
)
FAISS.save_local(vectorstore,"test_faiss")

FAISS.load_local("test_faiss",create_qwen_embeddings(model="text-embedding-3-small"),allow_dangerous_deserialization=True).similarity_search("JavaEdge是什么职业?")

r1 = FAISS.load_local("test_faiss",create_qwen_embeddings(),allow_dangerous_deserialization=True)
r2 = r1.as_retriever(
    search_kwargs={"k":1}
)
memory2 = VectorStoreRetrieverMemory(
    retriever=r2
)
memory2.load_memory_variables({"prompt":"JavaEdge是什么职业?"})

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2024-06-23，如有侵权请联系 cloudcommunity@tencent.com 删除

memory