上一章的 Agent 已经能调用工具、做多步规划和错误恢复,但它还有一个明显短板:每次对话都是“新来的”。用户上一句说“我喜欢喝咖啡”,下一句问“我刚才说了什么”,它大概率答不上来。
这一章我们要解决的就是这个问题:让 Agent 拥有短期记忆和长期记忆,能够记住对话历史、用户偏好,并在程序重启后仍然保留关键信息。
一、为什么 Agent 需要记忆?
大语言模型本身是无状态的。它不会自动记住你之前说过什么。每次调用模型时,模型看到的只是当前请求里的文本。
所以,要让 Agent “记住”内容,必须把历史对话、用户偏好、工具结果等信息保存下来,并在下一次调用时重新送入上下文。
可以把记忆理解成三类:
| 记忆类型 | 作用 | 典型实现 |
|---|---|---|
| 短期记忆 | 记住当前对话上下文 | 消息列表、滑动窗口 |
| 长期记忆 | 跨会话记住用户偏好 | 文件、SQLite、数据库 |
| 摘要记忆 | 压缩早期对话,节省 Token | LLM 生成摘要 |
简单理解:
- 短期记忆:解决“刚才说了什么”。
- 长期记忆:解决“这个用户平时喜欢什么”。
- 摘要记忆:解决“历史太长,上下文塞不下”。
二、记忆系统的工作流程
一个带记忆的 Agent,通常遵循四步流程:
读取历史
根据session_id或thread_id找到当前会话的历史消息。拼入提示词
把历史消息放入MessagesPlaceholder占位符,让模型看到完整上下文。调用模型
模型结合历史、当前输入和可用工具生成响应。写回记忆
把本轮用户输入和模型输出保存起来,供下一轮使用。
如果历史无限增长,会带来两个问题:
- Token 消耗越来越高;
- 超过模型上下文窗口后会报错。
因此,实际项目中通常会使用滑动窗口或摘要压缩来控制记忆长度。
三、本章项目结构
本章在第三章基础上增加记忆能力,文件结构如下:
1project/ 2├── .env 3├── agent_with_memory.py 4├── verify_memory.py 5└── agent_memory/ 6 ├── chat_history.json 7 └── user_profile.json其中:
agent_with_memory.py:带记忆和上下文管理的 Agent;verify_memory.py:验证 Agent 是否能记住历史、偏好和跨会话信息;agent_memory/:持久化存储目录。
四、完整可运行代码:agent_with_memory.py
1""" 2第四章示例:记忆与上下文管理 3 4运行方式: 51. 在项目根目录创建 .env 文件,写入 OPENAI_API_KEY 62. 安装依赖: 7 pip install langchain langchain-openai python-dotenv 83. 运行:python agent_with_memory.py 9""" 10 11import os 12import json 13from datetime import datetime 14from dotenv import load_dotenv 15from langchain_openai import ChatOpenAI 16from langchain_core.tools import tool 17from langchain.agents import create_tool_calling_agent, AgentExecutor 18from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder 19from langchain_core.messages import HumanMessage, AIMessage 20 21# ---------------------------- 22# 1. 加载环境变量 23# ---------------------------- 24load_dotenv() 25 26OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") 27MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o") 28MODEL_BASE_URL = os.getenv("MODEL_BASE_URL", "").strip() 29 30if not OPENAI_API_KEY: 31 raise RuntimeError( 32 "未找到 OPENAI_API_KEY,请在项目根目录创建 .env 文件并写入 API Key。" 33 ) 34 35# ---------------------------- 36# 2. 初始化模型 37# ---------------------------- 38llm_kwargs = { 39 "model": MODEL_NAME, 40 "temperature": 0.2, 41 "api_key": OPENAI_API_KEY, 42} 43 44if MODEL_BASE_URL: 45 llm_kwargs["base_url"] = MODEL_BASE_URL 46 47llm = ChatOpenAI(**llm_kwargs) 48 49# ---------------------------- 50# 3. 定义工具 51# ---------------------------- 52 53@tool 54def get_weather(city: str) -> str: 55 """查询指定城市的天气信息。 56 57 当用户询问某个城市的天气时调用此工具。 58 59 参数: 60 - city: 城市名称,例如“郑州”“北京”“上海” 61 """ 62 63 if not city or not city.strip(): 64 return "缺少城市名称,无法查询天气。" 65 66 city = city.strip() 67 68 weather_data = { 69 "郑州": "晴,22°C,微风", 70 "北京": "多云,18°C,北风2级", 71 "上海": "小雨,20°C,东南风3级", 72 "广州": "阴,26°C,湿度较高", 73 "深圳": "晴,28°C,适合出行", 74 } 75 76 if city in weather_data: 77 return f"{city}天气:{weather_data[city]}" 78 79 return f"未找到“{city}”的天气数据,请确认城市名称。" 80 81 82@tool 83def get_current_time(dummy: str = "") -> str: 84 """获取当前日期和时间。 85 86 当用户询问时间、日期、现在几点时调用此工具。 87 88 参数: 89 - dummy: 可选占位参数,不需要传入实际内容 90 """ 91 92 now = datetime.now() 93 return now.strftime("%Y年%m月%d日 %H:%M:%S") 94 95 96@tool 97def convert_currency(amount: float, from_currency: str, to_currency: str) -> str: 98 """进行简单的货币换算。 99 100 当用户询问汇率或进行金额换算时调用此工具。 101 102 参数: 103 - amount: 需要换算的金额 104 - from_currency: 源货币,例如 USD、CNY 105 - to_currency: 目标货币,例如 CNY、USD 106 """ 107 108 if amount <= 0: 109 return "换算金额必须大于 0。" 110 111 from_currency = from_currency.upper() 112 to_currency = to_currency.upper() 113 114 rates = { 115 "USD_CNY": 7.2, 116 "CNY_USD": 1 / 7.2, 117 "EUR_CNY": 7.8, 118 "CNY_EUR": 1 / 7.8, 119 } 120 121 key = f"{from_currency}_{to_currency}" 122 123 if key not in rates: 124 return ( 125 f"暂不支持 {from_currency} 到 {to_currency} 的换算," 126 "请尝试 USD/CNY 或 EUR/CNY。" 127 ) 128 129 result = amount * rates[key] 130 result = round(result, 2) 131 132 return f"{amount} {from_currency} ≈ {result} {to_currency}" 133 134 135tools = [get_weather, get_current_time, convert_currency] 136 137# ---------------------------- 138# 4. 记忆管理器 139# ---------------------------- 140 141MEMORY_DIR = "agent_memory" 142CHAT_HISTORY_FILE = os.path.join(MEMORY_DIR, "chat_history.json") 143USER_PROFILE_FILE = os.path.join(MEMORY_DIR, "user_profile.json") 144 145 146def ensure_memory_dir(): 147 """确保记忆存储目录存在。""" 148 os.makedirs(MEMORY_DIR, exist_ok=True) 149 150 151def load_chat_history(session_id: str) -> list: 152 """从文件加载指定会话的聊天历史。 153 154 返回格式: 155 [ 156 {"role": "human", "content": "..."}, 157 {"role": "ai", "content": "..."}, 158 ] 159 """ 160 161 ensure_memory_dir() 162 163 if not os.path.exists(CHAT_HISTORY_FILE): 164 return [] 165 166 try: 167 with open(CHAT_HISTORY_FILE, "r", encoding="utf-8") as f: 168 data = json.load(f) 169 except Exception: 170 return [] 171 172 # data 结构示例: 173 # { 174 # "session_001": [ 175 # {"role": "human", "content": "..."}, 176 # {"role": "ai", "content": "..."} 177 # ] 178 # } 179 if not isinstance(data, dict): 180 return [] 181 182 return data.get(session_id, []) 183 184 185def save_chat_history(session_id: str, history: list): 186 """将指定会话的聊天历史写入文件。""" 187 188 ensure_memory_dir() 189 190 # 先读取已有数据,避免覆盖其他会话 191 all_histories = {} 192 if os.path.exists(CHAT_HISTORY_FILE): 193 try: 194 with open(CHAT_HISTORY_FILE, "r", encoding="utf-8") as f: 195 all_histories = json.load(f) 196 if not isinstance(all_histories, dict): 197 all_histories = {} 198 except Exception: 199 all_histories = {} 200 201 all_histories[session_id] = history 202 203 with open(CHAT_HISTORY_FILE, "w", encoding="utf-8") as f: 204 json.dump(all_histories, f, ensure_ascii=False, indent=2) 205 206 207def load_user_profile() -> dict: 208 """加载用户长期偏好。 209 210 示例结构: 211 { 212 "name": "小明", 213 "city": "郑州", 214 "preferences": { 215 "weather_reminder": True 216 } 217 } 218 """ 219 220 ensure_memory_dir() 221 222 if not os.path.exists(USER_PROFILE_FILE): 223 return {} 224 225 try: 226 with open(USER_PROFILE_FILE, "r", encoding="utf-8") as f: 227 profile = json.load(f) 228 return profile if isinstance(profile, dict) else {} 229 except Exception: 230 return {} 231 232 233def save_user_profile(profile: dict): 234 """保存用户长期偏好。""" 235 236 ensure_memory_dir() 237 238 with open(USER_PROFILE_FILE, "w", encoding="utf-8") as f: 239 json.dump(profile, f, ensure_ascii=False, indent=2) 240 241 242def update_user_profile_from_conversation(profile: dict, last_human: str, last_ai: str) -> dict: 243 """从最近一轮对话中抽取用户偏好,更新长期记忆。 244 245 这里使用规则抽取作为示例,后续可以换成 LLM 抽取。 246 """ 247 248 profile = profile.copy() 249 250 # 简单抽取城市偏好 251 city_keywords = ["郑州", "北京", "上海", "广州", "深圳", "杭州", "成都"] 252 for city in city_keywords: 253 if city in last_human and "city" not in profile: 254 profile["city"] = city 255 256 # 简单抽取用户姓名 257 name_prefixes = ["我叫", "名字是", "叫我"] 258 for prefix in name_prefixes: 259 if prefix in last_human: 260 idx = last_human.find(prefix) + len(prefix) 261 name = last_human[idx:].strip().split(",")[0].split(",")[0].strip() 262 if name and "name" not in profile: 263 profile["name"] = name 264 265 return profile 266 267 268def trim_history(messages: list, max_rounds: int = 10) -> list: 269 """对聊天历史做滑动窗口裁剪。 270 271 每轮对话包含一条 HumanMessage 和一条 AIMessage。 272 这里保留最近 max_rounds 轮。 273 """ 274 275 if len(messages) <= max_rounds * 2: 276 return messages 277 278 # 只保留最后 max_rounds * 2 条消息 279 return messages[-max_rounds * 2:] 280 281 282def build_history_messages(history: list) -> list: 283 """将 JSON 格式的历史转成 LangChain 消息对象。""" 284 285 messages = [] 286 for item in history: 287 role = item.get("role") 288 content = item.get("content", "") 289 290 if role == "human": 291 messages.append(HumanMessage(content=content)) 292 elif role == "ai": 293 messages.append(AIMessage(content=content)) 294 295 return messages 296 297 298def serialize_history(messages: list) -> list: 299 """将 LangChain 消息对象转成可保存的 JSON 列表。""" 300 301 result = [] 302 for msg in messages: 303 if isinstance(msg, HumanMessage): 304 result.append({"role": "human", "content": msg.content}) 305 elif isinstance(msg, AIMessage): 306 result.append({"role": "ai", "content": msg.content}) 307 return result 308 309 310# ---------------------------- 311# 5. 创建带记忆的 Agent 312# ---------------------------- 313 314prompt = ChatPromptTemplate.from_messages( 315 [ 316 ( 317 "system", 318 "你是一个智能助手。你可以调用工具查询天气、时间和汇率。" 319 "你会记住用户的基本信息和偏好。" 320 "如果工具返回不确定或错误信息,不要编造答案,应如实提示用户。" 321 "如果用户之前提到过自己的城市、姓名或偏好,请在回答时自然引用。" 322 ), 323 # 短期记忆占位符:历史对话会插入到这里 324 MessagesPlaceholder(variable_name="chat_history"), 325 ("human", "{input}"), 326 ("placeholder", "{agent_scratchpad}"), 327 ] 328) 329 330agent = create_tool_calling_agent(llm, tools, prompt) 331 332# 注意:这里不直接使用 AgentExecutor.memory, 333# 而是手动管理 chat_history,方便演示短期记忆裁剪和长期偏好存储。 334agent_executor = AgentExecutor( 335 agent=agent, 336 tools=tools, 337 verbose=True, 338 max_iterations=8, 339 handle_parsing_errors=True, 340) 341 342 343# ---------------------------- 344# 6. 带记忆的调用入口 345# ---------------------------- 346 347def ask_agent(session_id: str, user_input: str) -> str: 348 """执行一次带记忆的 Agent 调用。 349 350 流程: 351 1. 读取历史对话; 352 2. 加载用户长期偏好; 353 3. 裁剪历史,避免上下文过长; 354 4. 调用 Agent; 355 5. 保存本轮对话; 356 6. 更新用户长期偏好。 357 """ 358 359 # 1. 读取短期记忆 360 raw_history = load_chat_history(session_id) 361 history_messages = build_history_messages(raw_history) 362 363 # 2. 滑动窗口裁剪,防止历史过长 364 history_messages = trim_history(history_messages, max_rounds=8) 365 366 # 3. 读取长期记忆 367 profile = load_user_profile() 368 369 # 4. 把用户偏好注入到输入中,让模型更容易利用长期记忆 370 enriched_input = user_input 371 if profile: 372 profile_hint = "用户已知信息:" 373 if profile.get("name"): 374 profile_hint += f" 姓名={profile['name']};" 375 if profile.get("city"): 376 profile_hint += f" 所在城市={profile['city']};" 377 if profile.get("preferences"): 378 profile_hint += f" 偏好={profile['preferences']};" 379 enriched_input = profile_hint + "\n用户当前问题:" + user_input 380 381 # 5. 调用 Agent 382 result = agent_executor.invoke( 383 { 384 "input": enriched_input, 385 "chat_history": history_messages, 386 } 387 ) 388 389 output = result.get("output", "") 390 391 # 6. 更新短期记忆 392 raw_history.append({"role": "human", "content": user_input}) 393 raw_history.append({"role": "ai", "content": output}) 394 395 # 再次裁剪后保存,避免文件无限增长 396 trimmed = trim_history(build_history_messages(raw_history), max_rounds=8) 397 raw_history = serialize_history(trimmed) 398 399 save_chat_history(session_id, raw_history) 400 401 # 7. 更新长期记忆 402 profile = update_user_profile_from_conversation(profile, user_input, output) 403 save_user_profile(profile) 404 405 return output 406 407 408# ---------------------------- 409# 7. 运行示例 410# ---------------------------- 411 412def run_demo(): 413 """运行多轮对话示例,展示短期记忆、长期记忆和跨会话隔离。""" 414 415 session_id = "session_demo_001" 416 417 questions = [ 418 "你好,我叫小明,我住在郑州。", 419 "郑州今天天气怎么样?", 420 "我刚才说我住在哪个城市?", 421 "我叫什么名字?", 422 "帮我查一下北京天气,然后和郑州比较一下。", 423 "100美元能换多少人民币?", 424 "火星今天天气如何?", 425 ] 426 427 print(f"会话ID:{session_id}\n") 428 429 for question in questions: 430 print("=" * 60) 431 print(f"用户:{question}") 432 print("-" * 60) 433 434 answer = ask_agent(session_id, question) 435 436 print(f"AI:{answer}") 437 print() 438 439 # 演示跨会话隔离 440 print("\n" + "=" * 60) 441 print("切换到新会话:session_demo_002") 442 print("=" * 60) 443 444 new_session_id = "session_demo_002" 445 answer = ask_agent(new_session_id, "我叫什么名字?") 446 print(f"用户:我叫什么名字?") 447 print(f"AI:{answer}") 448 print() 449 print("说明:新会话没有之前对话历史,所以无法知道姓名。") 450 451 452if __name__ == "__main__": 453 run_demo()五、关键代码解释
1. 为什么用MessagesPlaceholder?
1MessagesPlaceholder(variable_name="chat_history")这个占位符告诉 Agent:在执行时,把chat_history里的消息插入到提示词中。
如果没有它,模型就看不到历史对话,也就无法回答“我刚才说了什么”这类问题。
2. 短期记忆为什么要裁剪?
1history_messages = trim_history(history_messages, max_rounds=8)如果一直保存全部历史,会出现三个问题:
- 每次调用都要把全部历史发给模型;
- Token 费用会越来越高;
- 超过模型上下文窗口后会直接报错。
这里采用滑动窗口策略,只保留最近 8 轮对话。更复杂的项目也可以结合摘要记忆,把早期对话压缩成一段总结。
3. 长期记忆如何保存?
1save_user_profile(profile)这里把用户姓名、城市等偏好保存到user_profile.json。这样即使程序重启,Agent 仍然能恢复这些信息。
真实项目中,长期记忆可以存到 SQLite、PostgreSQL、Redis 或向量数据库中。
4. 会话隔离为什么重要?
1ask_agent(session_id, question)每个会话都有独立session_id。这样不同用户、不同对话不会混在一起。
如果不做会话隔离,所有用户的消息都会进入同一份历史,既不安全,也不符合真实业务需求。
六、完整验证脚本:verify_memory.py
1""" 2记忆能力验证脚本 3 4运行方式: 51. 先运行:python agent_with_memory.py 62. 再运行:python verify_memory.py 7""" 8 9import os 10import json 11from agent_with_memory import ( 12 ask_agent, 13 load_chat_history, 14 load_user_profile, 15 save_chat_history, 16 save_user_profile, 17) 18 19# ---------------------------- 20# 1. 测试用例 21# ---------------------------- 22 23MEMORY_TEST_CASES = [ 24 { 25 "name": "短期记忆:记住上一轮城市", 26 "session_id": "verify_session_001", 27 "steps": [ 28 {"input": "我住在郑州。", "expect_keywords": ["郑州"]}, 29 {"input": "我住在哪个城市?", "expect_keywords": ["郑州"]}, 30 ], 31 }, 32 { 33 "name": "短期记忆:记住姓名", 34 "session_id": "verify_session_002", 35 "steps": [ 36 {"input": "我叫小华。", "expect_keywords": ["小华"]}, 37 {"input": "我叫什么名字?", "expect_keywords": ["小华"]}, 38 ], 39 }, 40 { 41 "name": "长期记忆:程序重启后仍能读取偏好", 42 "session_id": "verify_session_003", 43 "steps": [ 44 {"input": "我叫阿明,我喜欢北京。", "expect_keywords": []}, 45 {"input": "我叫什么名字?", "expect_keywords": ["阿明"]}, 46 {"input": "我喜欢哪个城市?", "expect_keywords": ["北京"]}, 47 ], 48 }, 49 { 50 "name": "跨会话隔离:新会话不应继承旧偏好", 51 "session_id": "verify_session_004_new", 52 "steps": [ 53 {"input": "我叫什么名字?", "expect_not_keywords": ["阿明"]}, 54 ], 55 }, 56] 57 58# ---------------------------- 59# 2. 验证逻辑 60# ---------------------------- 61 62def check_keywords(text: str, keywords: list) -> bool: 63 """检查文本中是否包含任意一个关键词。""" 64 return any(k in text for k in keywords) 65 66 67def run_memory_verification(): 68 """逐用例执行记忆验证。""" 69 70 report = [] 71 passed = 0 72 failed = 0 73 74 for case in MEMORY_TEST_CASES: 75 session_id = case["session_id"] 76 77 # 清理该会话的历史,避免受之前运行影响 78 save_chat_history(session_id, []) 79 80 print(f"\n{'='*60}") 81 print(f"测试:{case['name']}") 82 print(f"会话ID:{session_id}") 83 84 step_results = [] 85 86 for step in case["steps"]: 87 user_input = step["input"] 88 expect_keywords = step.get("expect_keywords", []) 89 expect_not_keywords = step.get("expect_not_keywords", []) 90 91 print(f"\n 用户:{user_input}") 92 93 answer = ask_agent(session_id, user_input) 94 95 print(f" AI:{answer}") 96 97 # 检查必须出现的关键词 98 keyword_ok = True 99 for kw in expect_keywords: 100 if kw not in answer: 101 keyword_ok = False 102 print(f" 期望回答中包含“{kw}”,但未找到。") 103 104 # 检查不应出现的关键词 105 not_keyword_ok = True 106 for kw in expect_not_keywords: 107 if kw in answer: 108 not_keyword_ok = False 109 print(f" 期望回答中不包含“{kw}”,但找到了。") 110 111 step_passed = keyword_ok and not_keyword_ok 112 status = "PASS" if step_passed else "FAIL" 113 114 step_results.append({ 115 "input": user_input, 116 "output": answer, 117 "status": status, 118 "expect_keywords": expect_keywords, 119 "expect_not_keywords": expect_not_keywords, 120 }) 121 122 if step_passed: 123 passed += 1 124 else: 125 failed += 1 126 127 report.append({ 128 "name": case["name"], 129 "session_id": session_id, 130 "steps": step_results, 131 }) 132 133 # 打印汇总 134 print(f"\n{'='*60}") 135 print(f"记忆验证汇总:共 {passed + failed} 个步骤,{passed} 通过,{failed} 失败") 136 print(f"{'='*60}") 137 138 # 保存报告 139 output_dir = "agent_traces" 140 os.makedirs(output_dir, exist_ok=True) 141 142 report_file = os.path.join(output_dir, "memory_verify_report.json") 143 144 with open(report_file, "w", encoding="utf-8") as f: 145 json.dump(report, f, ensure_ascii=False, indent=2) 146 147 print(f"\n记忆验证报告已保存到:{report_file}") 148 149 150if __name__ == "__main__": 151 run_memory_verification()七、运行验证脚本
先运行主程序:
1python agent_with_memory.py再运行验证脚本:
1python verify_memory.py验证脚本会依次检查:
- 短期记忆是否能记住上一轮城市;
- 短期记忆是否能记住姓名;
- 长期记忆在程序重启后是否仍可读取;
- 新会话是否不会继承旧会话的偏好。
运行结束后,会在agent_traces/目录下生成memory_verify_report.json,方便逐条查看每一步的输入、输出和验证结果。
八、进阶:摘要记忆
前面使用的是滑动窗口,也就是只保留最近若干轮对话。它简单可靠,但有一个问题:早期对话会被直接丢弃。
如果想保留更多信息,可以采用摘要记忆。思路是:
- 当历史消息超过一定长度时,调用一次模型;
- 让模型把早期对话压缩成一段摘要;
- 用摘要替换早期对话;
- 新对话继续追加在摘要后面。
示例实现如下:
1from langchain_core.messages import SystemMessage 2 3def summarize_early_history(messages: list, keep_last: int = 4) -> list: 4 """把早期对话压缩成摘要,保留最近 keep_last 条消息。 5 6 注意:这里只是示例,真实项目中可以加入缓存、失败降级等逻辑。 7 """ 8 9 if len(messages) <= keep_last + 2: 10 return messages 11 12 # 早期消息用于摘要 13 early_messages = messages[:-keep_last] 14 recent_messages = messages[-keep_last:] 15 16 # 把早期消息转成文本 17 early_text = "\n".join( 18 f"{msg.type}: {msg.content}" for msg in early_messages 19 ) 20 21 summary_llm = ChatOpenAI( 22 model=MODEL_NAME, 23 temperature=0, 24 api_key=OPENAI_API_KEY, 25 **( 26 {"base_url": MODEL_BASE_URL} if MODEL_BASE_URL else {} 27 ), 28 ) 29 30 summary_prompt = ( 31 "以下是用户和助手之前的对话。请用 2-3 句话总结用户的关键偏好、" 32 "事实和当前任务进展。不要编造信息。" 33 "\n\n对话内容:\n" + early_text 34 ) 35 36 try: 37 summary_response = summary_llm.invoke([HumanMessage(content=summary_prompt)]) 38 summary_text = summary_response.content 39 except Exception: 40 # 如果摘要生成失败,就保留原始消息,避免丢失上下文 41 return messages 42 43 # 用系统消息保存摘要 44 summary_message = SystemMessage( 45 content=f"【对话摘要】{summary_text}" 46 ) 47 48 return [summary_message] + recent_messages使用时,把原来的trim_history替换成summarize_early_history即可:
1history_messages = summarize_early_history(history_messages, keep_last=6)摘要记忆适合长对话场景,但它会额外消耗一次模型调用,所以通常只在历史较长时触发。
九、生产环境建议
本章示例为了便于理解,使用了本地 JSON 文件保存记忆。真实项目里可以这样演进:
| 能力 | 示例方案 | 适用场景 |
|---|---|---|
| 短期记忆 | Redis / 内存缓存 | 高并发会话 |
| 长期记忆 | SQLite / PostgreSQL | 用户画像、偏好 |
| 向量记忆 | FAISS / Chroma / Milvus | 文档检索、知识库 |
| 会话隔离 | thread_id/user_id | 多用户系统 |
| 记忆清理 | 定期归档、用户删除接口 | 合规与隐私保护 |
另外,长期记忆涉及用户隐私,建议注意三点:
- 不要默认保存敏感信息,例如身份证号、银行卡号、密码;
- 提供用户查看和删除记忆的入口;
- 对记忆做最小化存储,只保留真正有助于后续对话的信息。
十、本章小结
| 概念 | 作用 | 本章实现方式 |
|---|---|---|
| 短期记忆 | 记住当前对话上下文 | chat_history.json+ 滑动窗口 |
| 长期记忆 | 跨会话记住用户偏好 | user_profile.json |
| 会话隔离 | 不同用户互不干扰 | session_id |
| 摘要记忆 | 压缩早期对话 | LLM 生成历史摘要 |
| 记忆抽取 | 从对话中提取偏好 | 规则抽取 + 可替换为 LLM 抽取 |
到这里,Agent 已经具备:
- 工具调用能力;
- 多步规划能力;
- 错误恢复能力;
- 短期记忆与长期记忆能力。