笔记
Agent的学习记录--6
来源Github的学习指南Agent Learning Hub的学习：
Stage 2——记忆
Agent的记忆主要分为三个方面：
- 短期记忆（Short-term Memory）：即上下文记忆：为当前请求送入 Prompt 的临时消息窗口，受上下文长度与成本制约
- 会话记忆（Session Memory）：记录同一会话 (session_id) 内的多轮对话历史，按会话隔离进行记忆
- 长期记忆（Long-term Memory): 跨会话沉淀的用户偏好、个性化画像与核心事实
import os
import sys
import json
from typing import List, Dict, Any, Optional
from config import get_llm
from langchain_core.messages import (
BaseMessage,
SystemMessage,
HumanMessage,
AIMessage,
trim_messages
)
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
短期记忆管理
对于短期记忆，当消息列表过长时，直接全部送入模型不仅昂贵，还会导致模型注意力分散
因此我们需要动态修剪短期上下文：
- 必须始终保留最开头的 SystemMessage
- 保留最近的 max_messages 条历史对话消息（滑动窗口）
- 修剪丢弃过早的消息
def trim_context_window(messages: List[BaseMessage], max_messages: int = 4) -> List[BaseMessage]:
if len(messages) <= max_messages + 1:
return messages
if messages and isinstance(messages[0], SystemMessage):
system_anchor = messages[0]
last_anchor = messages[1:][-max_messages:]
return [system_anchor] + last_anchor
else:
return messages[-max_messages:]
会话记忆管理
主要原则为负责按 session_id 隔离管理不同会话的多轮对话历史。
class SessionMemoryManager:
def __init__(self):
# 内存存储映射表：session_id -> InMemoryChatMessageHistory
self._session_store: Dict[str, InMemoryChatMessageHistory] = {}
def get_history(self, session_id: str) -> InMemoryChatMessageHistory:
# 如果是新会话，新建一个历史记录本；如果是老会话，取出历史记录本
if not session_id in self._session_store:
self._session_store[session_id] = InMemoryChatMessageHistory()
return self._session_store[session_id]
def add_user_message(self, session_id: str, content: str):
# 追加一条用户发言到该 session 的历史记录本中
history = self.get_history(session_id)
history.add_user_message(content)
def add_ai_message(self, session_id: str, content: str):
# 追加一条 AI 回复到该 session 的历史记录本中
history = self.get_history(session_id)
history.add_ai_message(content)
长期记忆管理
长期记忆，其跨越所有 session，提炼出少量核心标签（如姓名、偏好、习惯）。管理跨越所有会话的永久用户画像与核心事实 (Profile & Facts)。以 JSON 文件形式持久化存储在磁盘上。
LONG_TERM_MEMORY_FILE = os.path.join(os.path.dirname(__file__), "sample_data", "user_long_term_memory.json")
Class LongTermMemoryStore:
def __init__(self, filepath: str = LONG_TERM_MEMORY_FILE):
self.filepath = filepath
self._ensure_file()
def _ensure_file(self):
os.makedirs(os.path.dirname(self.filepath), exist_ok = True)
if not os.path.exist(self.filepath):
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump({"user_profile": {}, "learned_fact": []}, f, ensure_ascii=False, indent=2)
def read_memory(self) -> Dict[str, Any]:
try:
with open(self.filepath, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return {"user_profile": {}, "learned_fact": []}
def update_profile(self, key: str, value: Any):
# 更新用户画像
data = self.read_memory()
data["user_profile"][key] = value
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def add_fact(self, fact:str):
# 更新用户事实
data = self.read_memory()
if fact not in data["learned_fact"]:
data["learn_fact"].append(fact)
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
然后我们通过LLM为我们创建画像：
def extract_and_save_long_term_facts(dialogue_history: List[BaseMessage], memory_store: LongTermMemoryStore):
"""
使用 LLM 分析对话内容，提取出值得永久记忆的用户个性化信息或事实（如研究方向、常用编程语言、喜好）。
"""
llm = get_llm(temperature=0.0)
# 拼接对话历史文本
conversation_text = "\n".join(f"{msg.type}: {msg.content}" for msg in dialogue_history if msg.type in ("human", "ai"))
prompt = (
"你是一个行为学家，擅长从用户对话里提取出关于用户的长期偏好与事实画像。\n"
"请严格以纯 JSON 格式输出,其格式如下：\n"
"{\n"
' "user_profile": {\n'
' "research_interest": "用户研究方向",\n'
' "preferred_language": "偏好使用的框架或语言"\n'
" },\n"
' "learned_facts": ["值得长期记住的事实1", "事实2"]\n'
"}\n\n"
f"【对话记录】:\n{conversation_text}"
)
response = llm.invoke([HumanMessage(content = prompt)])
content = response.content.strip()
# 清洗可能存在的 ```json 代码块包裹
if content.startswith("```"):
lines = content.split("\n")
if lines[0].startswith("```"):
lines = lines[1:]
if lines and lines[-1].strip() == "```":
lines = lines[:-1]
content = "\n".join(lines).strip()
try:
content_json = json.loads(content)
profile = content_json.get("user_profile", {})
fact = content_json.get("learned_facts", [])
for k, v in profile.items():
if k and v:
memory_store.update_profile(k, v)
for f in fact:
if f:
memory_store.add_fact(f)
except Exception as e:
print(f"解析提取结果异常: {e}, 原文: {content}")
这样子我们就梳理完了这三种记忆
现在我们将这三种记忆打包：
综合
综合应用三层记忆架构进行对话：
- 长期记忆：从磁盘读取长期画像，注入 SystemMessage
- 会话记忆：加载该 session_id 下的历史消息
- 短期上下文：使用 trim_context_window 裁剪超长历史，避免超限
def chat_with_full_memory(session_id: str, user_input: str, session_manager: SessionMemoryManager, memory_store: LongTermMemoryStore) -> str:
llm = get_llm(temperature=0.0)
# 取出长期记忆组装SystemMessage
long_term_memory = memory_store.read_memory()
profile = long_term_memory.get("user_profile", {})
facts = long_term_memory.get("learned_facts", [])
system_prompt = (
"你是一名个人专属科研助手。\n"
f"【已知用户长期偏好与画像】: {json.dumps(profile, ensure_ascii=False)}\n"
f"【已知用户长期事实库】: {json.dumps(facts, ensure_ascii=False)}\n"
"请根据用户偏好量身定制回答！"
)
# 获取之前对话记录
history = session_manager.get_history(session_id)
# 组装短期上下文并进行裁切
raw_messages = [SystemMessage(content = system_prompt)] + history.messages + [HumanMessage(content = user_input)]
active_messages = trim_context_window(raw_messages)
response = llm.invoke(active_messages)
history.add_user_message(user_input)
history.add_ai_messaga(response.content)
return response.content