2. Agent记忆功能实现
任务说明
为Agent实现短期记忆（当前会话上下文）与长期记忆（跨会话持久化），使Agent能记住用户偏好（如称呼、语言风格、关注话题），并在后续对话中主动引用这些信息。本案例使用Python实现一个最小可运行的记忆增强Agent，不依赖外部大模型API，用规则匹配模拟对话理解，聚焦记忆机制的完整闭环。
核心实现
import json
import os
from datetime import datetime
from typing import Dict, List, Optional
class MemoryStore:
"""长期记忆存储：JSON文件持久化"""
def __init__(self, filepath: str = "agent_memory.json"):
self.filepath = filepath
self.data: Dict = self._load()
def _load(self) -> Dict:
if os.path.exists(self.filepath):
with open(self.filepath, "r", encoding="utf-8") as f:
return json.load(f)
return {"preferences": {}, "history": []}
def save(self):
with open(self.filepath, "w", encoding="utf-8") as f:
json.dump(self.data, f, ensure_ascii=False, indent=2)
def update_preference(self, key: str, value: str):
self.data["preferences"][key] = value
self.save()
def add_history(self, role: str, content: str):
self.data["history"].append({
"role": role,
"content": content,
"time": datetime.now().isoformat()
})
# 只保留最近50条历史
self.data["history"] = self.data["history"][-50:]
self.save()
class Agent:
def __init__(self, memory_file: str = "agent_memory.json"):
self.memory = MemoryStore(memory_file)
self.short_term: List[Dict] = [] # 当前会话短期记忆
def _extract_preference(self, user_input: str) -> Optional[tuple]:
"""从用户输入中提取偏好（规则匹配）"""
rules = [
("称呼", ["叫我", "喊我", "称呼我"]),
("语言风格", ["说中文", "讲英文", "用英文", "用中文"]),
("关注话题", ["关注", "对...感兴趣", "喜欢聊"]),
]
for pref_type, keywords in rules:
for kw in keywords:
if kw in user_input:
value = user_input.split(kw)[-1].strip("，。！？,.!? ")
if value:
return pref_type, value
return None
def _get_preference_context(self) -> str:
"""生成长期记忆上下文提示"""
prefs = self.memory.data["preferences"]
if not prefs:
return ""
parts = [f"{k}:{v}" for k, v in prefs.items()]
return "；".join(parts)
def respond(self, user_input: str) -> str:
# 1. 提取并保存长期偏好
pref = self._extract_preference(user_input)
if pref:
pref_type, value = pref
self.memory.update_preference(pref_type, value)
self.short_term.append({"role": "user", "content": user_input})
return f"好的，我已记住你的{pref_type}：{value}"
# 2. 检查是否引用长期记忆
prefs = self.memory.data["preferences"]
if "称呼" in prefs and ("你好" in user_input or "在吗" in user_input):
name = prefs["称呼"]
self.short_term.append({"role": "user", "content": user_input})
return f"{name}，我在呢！有什么可以帮你？"
# 3. 检查是否引用短期记忆（当前会话前文）
if self.short_term:
last_user = [m for m in self.short_term if m["role"] == "user"]
if last_user and ("刚才" in user_input or "如上" in user_input or "这个" in user_input):
prev = last_user[-1]["content"]
self.short_term.append({"role": "user", "content": user_input})
return f"你刚才说的是“{prev}”，我记着呢。需要我做什么？"
# 4. 默认回复（带长期记忆上下文）
self.short_term.append({"role": "user", "content": user_input})
ctx = self._get_preference_context()
if ctx:
return f"（已记住：{ctx}）收到，请继续说。"
return "收到，请继续说。"
# 最小运行闭环
if __name__ == "__main__":
agent = Agent("demo_memory.json")
# 清理测试文件（保证可重复运行）
if os.path.exists("demo_memory.json"):
os.remove("demo_memory.json")
# 模拟多轮对话
test_inputs = [
"你好，叫我小王",
"我喜欢聊科技",
"你好，在吗",
"刚才说的那个话题，继续聊聊",
]
for text in test_inputs:
print(f"用户: {text}")
print(f"Agent: {agent.respond(text)}")
print("---")
# 验证长期记忆持久化
print("\n[验证] 长期记忆文件内容:")
with open("demo_memory.json", "r", encoding="utf-8") as f:
print(f.read())运行与验证
运行方式：将上述代码保存为 agent_memory.py，在Python 3.8+环境直接执行：
python agent_memory.py验证路径：
- 程序自动执行4轮模拟对话，观察输出：第一轮应识别并保存“称呼=小王”；第二轮保存“关注话题=科技”；第三轮应调用长期记忆中的称呼，回复“小王，我在呢！”；第四轮应引用短期记忆中的前文内容。
- 运行结束后，检查当前目录生成的
demo_memory.json文件，其preferences字段应包含{"称呼": "小王", "关注话题": "科技"}，history字段应包含4条对话记录。 - 再次运行
python agent_memory.py（不删除文件），第三轮输入“你好，在吗”时，Agent仍能回复“小王，我在呢！”，证明长期记忆跨会话持久化生效。
通过条件：上述3个验证点全部符合预期输出，且 demo_memory.json 内容结构完整，即判定案例实现成功。
关键设计说明
短期记忆通过 self.short_term 列表维护当前会话上下文，仅在程序运行期间有效；长期记忆通过 MemoryStore 类以JSON文件持久化，每次更新立即写入磁盘。偏好提取采用关键词规则匹配，覆盖称呼、语言风格、关注话题三类常见用户偏好。实际接入大模型时，只需将 respond() 中的规则回复替换为LLM调用，并将 _get_preference_context() 返回的字符串拼接到系统提示词中即可。