1️⃣ 分清 AI 两大记忆：短记忆、长记忆

做 AI 对话开发，绝大多数人都会踩一个坑：把「持久化」等同于「长期记忆」。

其实记忆按业务只分两类，而且短记忆、长记忆都可以做持久化，区别根本不在 “存不存库”，而在存储内容和是否跨会话。

1. 持久化短记忆（会话级记忆，当新建对话后忘记之前的记录）

代表：LangChain 原生整套能力

• 存储内容：当前聊天窗口的完整对话流水
• 存储介质：Redis / MySQL 做持久化
• 配套能力：用 trim_messages 做上下文裁剪
• 核心特征：只保本次会话，不跨会话
• 作用：保证多轮对话连贯、能正常追问承接

2. 持久化长记忆（用户级记忆）

代表：Mem0 + 向量数据库

• 存储内容：从对话提炼用户人设、偏好、习惯、固定信息
• 存储介质：Milvus / Qdrant 等向量库持久化
• 核心特征：跨会话、跨设备、长期有效
• 作用：AI 记住你这个人，实现千人千面个性化回复

LangChain 原生只能做：会话级持久化短记忆
本身没有记忆抽取、向量存储、跨会话记忆能力，做不了真正的长记忆。

---

企业级标准做法：
LangChain 负责本次聊天上下文（短记忆） + Mem0 负责用户长期人设（长记忆）

2️⃣ 短记忆（持久化短记忆）

一、LangChain 短记忆管理简述

LangChain 记忆由两部分组成：

1. 裁剪规则
负责管理历史：保留最近 N 轮对话、总结、截断等。

• 裁剪器 - trim_messages # 按token数或消息条数进行裁剪（LangChain0.3 推荐）
• 滑动窗口 - ConversationBufferWindowMemory(k=2) # 保留N轮对话 （已淘汰）
• 压缩总结 - ConversationSummaryMemory（已淘汰）

2. 持久化存储（持久化短期记忆）
通过 Redis / MySQL 等实现会话聊天记录落地持久化。

• RedisChatMessageHistory 【Redis 存储】
• SQLChatMessageHistory 【Mysql 存储】
• FileChatMessageHistory 【本地文件存储】

二、内存级上下级记忆（测试场景使用）

## ConversationBufferWindowMemory 类
ConversationBufferWindowMemory 继承自 ConversationMemory，属于窗口型缓存记忆类。
该类仅保存最近“指定轮次”的对话交互记录，超出窗口范围的历史信息会自动丢弃，能够有效限制上下文内存占用大小，适配各类对话场景的上下文长度约束需求。

例如：

```python
import os
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain_openai import OpenAI

# 初始化LLM - 使用通义千问
llm = OpenAI(
temperature=0.3,
openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
model_name="qwen-turbo"
)

# 创建窗口记忆 - 只保留最近2轮对话
memory = ConversationBufferWindowMemory(k=2)

# 创建对话链
conversation_with_memory = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)

# 第1轮：客户自我介绍
print("第1轮对话：")
response1 = conversation_with_memory.predict(
input="你好，我是张先生，我经常需要寄送电子产品到全国各地"
)
print(f"客服回复: {response1}\n")

# 第2轮：说明业务需求
print("第2轮对话：")
response2 = conversation_with_memory.predict(
input="我主要做电商生意，每天大概有50-100个包裹需要发货，主要是手机配件和数码产品"
)
print(f"客服回复: {response2}\n")

# 第3轮：询问物流方案
print("第3轮对话：")
response3 = conversation_with_memory.predict(
input="你能为我推荐一个适合的物流方案吗？我希望价格合理，时效稳定"
)
print(f"客服回复: {response3}\n")

# 第4轮：询问更多选项
print("第4轮对话：")
response4 = conversation_with_memory.predict(
input="除了刚才的方案，你还能给出更多选项吗？我想对比一下"
)
print(f"客服回复: {response4}\n")

# 第5轮：测试记忆窗口（应该忘记第1轮的内容）
print("第5轮对话（测试记忆窗口）：")
response5 = conversation_with_memory.predict(
input="你还记得我的名字吗？我是做什么生意的？"
)
print(f"客服回复: {response5}\n")

# 查看当前记忆内容
print("=== 当前记忆内容 ===")
print(memory.buffer)

三、LangChain - trim_messages 消息裁剪器 + 持久化（推荐）

1. Redis 持久化（RedisChatMessageHistory）

1.1. 安装依赖

pip install langchain langchain-core langchain-openai langchain-redis redis

1.2. 基础用法：RedisChatMessageHistory 直接操作消息历史

案例：将对话消息存储到 Redis，并查询。

from langchain_core.messages import HumanMessage, AIMessage
from langchain_redis import RedisChatMessageHistory  # 需要导入

# 创建 Redis 消息历史实例
history = RedisChatMessageHistory(
    session_id="user_1001",  # 对话标识/用户ID/窗口标识
    url="redis://localhost:6379",  
    key_prefix="chat:history:",   # 自定义 key 前缀
    ttl=3600  # 1小时过期
)

# 添加消息
history.add_message(HumanMessage(content="你好！我叫小明"))
history.add_message(AIMessage(content="你好小明！很高兴认识你，有什么我可以帮你的吗？"))

# 获取所有消息
print(history.messages)
# 输出: [HumanMessage(content='你好！我叫小明'), AIMessage(content='你好小明！很高兴认识你...')]

# 查看消息数量
print(f"共有 {len(history.messages)} 条消息")

1.3. 案例1、LangChain 对话 + Redis 持久化

这是最常用的生产级模式，使用 RunnableWithMessageHistory 自动管理历史记录的读写。

"""
生产级 LangChain 0.3 + Redis 对话记忆 - 精简版
支持：多用户隔离 | 会话自动过期 | 自动持久化 | 开箱即用
"""
import os
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI
from langchain_redis import RedisChatMessageHistory

# ========== 核心配置 ==========
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")

# ========== 获取Redis会话历史 ==========
def get_session_history(session_id: str):
    """
    根据用户ID+会话ID获取历史
    自动读写Redis，支持多用户隔离
    注意：生产场景设置全局单例 Redis 客户端，避免重复创建连接
    
    参数:
        session_id: 会话窗口标识（如：user123:sess456）
    返回:
        RedisChatMessageHistory: 支持自动持久化的历史记录对象
    """
    return RedisChatMessageHistory(
        session_id=session_id, # 定义存储key
        redis_url=REDIS_URL,
        key_prefix="chat:history:",  # Redis key前缀，最终格式：chat:history:user123:sess456
        ttl=604800  # 过期时间7天
    )

# ========== 创建带记忆的聊天机器人 ==========
def create_chat_bot():
    # 提示词模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是AI助手，记住上下文，连贯回答"),
        MessagesPlaceholder(variable_name="history"),  # 历史消息占位符
        ("human", "{input}"),  # 用户当前输入
    ])
    
    # 大模型
    llm = ChatOpenAI(
        model="gpt-3.5-turbo", 
        api_key=OPENAI_API_KEY, 
        temperature=0.7  # 控制回复的创造性（0-2，越高越随机）
    )

    # 定义修剪器，避免上下文超过模型限制
    trimmer = trim_messages(
        max_tokens=3000,  # 裁剪后历史消息的最大Token数（GPT-3.5上下文16K，建议预留空间）
        strategy="last",  # 保留最近的对话，丢弃最旧的
        token_counter=llm,  # 使用LLM精确计数（注意：会产生API调用费用），或使用len按字符数估算（免费，够用）
        # token_counter="approximate",  # 替代方案：快速估算，不产生费用
        include_system=True,  # 始终保留系统提示词
        start_on="human",  # 确保裁剪后的第一条消息是用户消息（符合对话格式）
        allow_partial=False,  # 不允许在消息中间截断（会完整保留或丢弃整条消息）
    )

    # 构建处理链：修剪历史 -> 格式化提示词 -> 调用模型
    chain = (
        # 从输入字典中提取history字段，应用修剪器后重新赋值
        RunnablePassthrough.assign(history=lambda x: trimmer(x["history"]))
        | prompt  # 将修剪后的history和其他变量传给prompt模板
        | llm  # 调用大模型生成回复
    )
    
    # 绑定记忆（核心）
    return RunnableWithMessageHistory(
        runnable=chain,  # 处理链
        get_session_history=get_session_history,  # 历史记录获取函数
        input_messages_key="input",  # 指定输入字典中用户消息的key
        history_messages_key="history",  # 指定输入字典中历史消息的key
    )

# ========== 使用示例 ==========
if __name__ == "__main__":
    bot = create_chat_bot()
    # config中的session_id会被传递给get_session_history函数
    config = {"configurable": {"session_id": "user123:sess456"}}

    # 对话测试（自动记忆上下文）
    print(bot.invoke({"input": "我叫张三，是软件工程师"}, config).content)
    print(bot.invoke({"input": "我的职业？"}, config).content)

    # ========== 常用操作 ==========
    # 1. 查看历史
    history = get_session_history("user123:sess456")
    print(f"\n历史消息数量：{len(history.messages)}")
    
    # 2. 清空历史（异步方法）
    # import asyncio
    # asyncio.run(history.aclear())
    
    # 3. 查看所有历史消息内容
    # for msg in history.messages:
    #     print(f"{msg.type}: {msg.content}")

1.4. 案例2、Agent + Redis 记忆集成

# ================= Agent + Redis记忆 + 修剪器（标准方式）==================
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import trim_messages
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_redis import RedisChatMessageHistory
from datetime import datetime
import os
from dotenv import load_dotenv

load_dotenv()

# ========== Redis配置 ==========
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
SESSION_TTL = 3600  # 会话过期时间：1小时（3600秒）

def get_session_history(session_id: str):
    """
    获取Redis会话历史（工厂函数）
    
    作用：告诉 RunnableWithMessageHistory 如何为每个用户/会话创建历史记录管理器
    
    工作原理：
    1. 每次对话时，RunnableWithMessageHistory 会调用此函数获取历史对象
    2. 然后自动调用 history.messages 读取历史（获取上下文）
    3. 对话结束后自动调用 history.add_message() 保存新消息（新增上下文）
    
    参数[只支持单参数]：
        session_id: 会话标识（如：user123:sess456）
    
    返回：
        RedisChatMessageHistory: 支持自动持久化的历史记录对象
    """
    return RedisChatMessageHistory(
        session_id=session_id,  # Redis key：user123:sess456
        redis_url=REDIS_URL,  # Redis 连接地址
        key_prefix="agent:chat:",  # Redis key前缀，最终格式：agent:chat:user123:sess456
        ttl=3600  # 过期时间1小时
    )

# ========== 定义工具 ==========
@tool
def calculator(expression: str) -> str:
    """数学计算器"""
    try:
        return f"结果：{expression} = {eval(expression)}"
    except:
        return "表达式格式错误"

@tool
def get_current_time() -> str:
    """获取当前日期时间"""
    return f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

tools = [calculator, get_current_time]

# ========== 创建带记忆和修剪器的Agent ==========
def create_agent_with_memory():
    # ===== 1. 提示词模板 =====
    # 注意：Agent 的提示词必须包含三个关键部分
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是智能助手，严格按照以下规则执行：
			1. 常识类问题直接回答，无需调用工具
			2. 涉及数学计算 → 调用 calculator 工具
			3. 询问当前时间 → 调用 get_current_time 工具
			4. 记住对话历史，能回答上下文相关问题
        """),
        # 历史消息占位符：这里会被 RunnableWithMessageHistory 自动填充历史对话
        MessagesPlaceholder(variable_name="chat_history"),
        # 用户当前输入占位符
        ("user", "{input}"),
        # Agent思考空间占位符（必需！存储工具调用和推理过程）
        # 注意：这个变量名是固定的，不能改，create_tool_calling_agent 要求必须有它
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
    
    # ===== 2. 初始化大模型 =====
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        api_key=os.getenv("OPENAI_API_KEY"),
        temperature=0.1  # 低温度 = 更确定性的回答
    )
    
    # ===== 3. 创建Agent和执行器 =====
    # 3.1 创建工具调用Agent（LangChain 0.3 标准方式）
    agent = create_tool_calling_agent(llm, tools, prompt)
    
    # 3.2 创建Agent执行器（负责运行Agent、调用工具、处理错误）
    agent_executor = AgentExecutor(
        agent=agent,  # Agent实例
        tools=tools,  # 工具列表
        verbose=True,  # 开启调试日志，查看思考&调用过程（生产环境建议False）
        max_iterations=5,  # 限制最大工具调用次数，防止死循环
        handle_parsing_errors=True  # 自动处理解析异常
    )

    # ===== 4. 定义消息修剪器 =====
    # 作用：防止历史消息超过模型上下文限制（GPT-3.5是16K，这里预留到3K）
    trimmer = trim_messages(
        max_tokens=3000,  # 裁剪后历史消息的最大Token数
        strategy="last",  # 保留最近的对话，丢弃最旧的
        token_counter=len,  # 按字符数估算，虽然不精确但足够用于修剪（如果使用llm会产生费用）
        include_system=False,  # 不包含系统提示词（因为system已在prompt中单独定义）
        start_on="human",  # 确保裁剪后的第一条消息是用户消息（符合对话格式）
        allow_partial=False,  # 不允许在消息中间截断（会完整保留或丢弃整条消息）
    )
    
    # ===== 5. 构建处理链 =====
    # 数据流向：输入 → 修剪历史 → Agent执行 → 输出
    chain = (
        # 5.1 修剪历史：从输入中提取 chat_history，应用修剪器后重新赋值
        RunnablePassthrough.assign(
            chat_history=lambda x: trimmer(x.get("chat_history", []))  # 如果没有历史则传空列表
        )
        # 5.2 执行Agent：将修剪后的历史+用户输入传给Agent
        # Agent会自动处理：prompt格式化 → 模型调用 → 工具调用（如果需要）→ 返回最终答案
        | agent_executor
    )
    
    # ===== 6. 绑定记忆（核心！） =====
    # RunnableWithMessageHistory 的作用：
    # - 自动从Redis读取历史消息（通过 get_session_history）
    # - 将历史消息注入到 chain 的输入中（字段名为 chat_history）
    # - 执行 chain
    # - 自动将本轮对话保存到Redis（用户输入 + AI输出）
    return RunnableWithMessageHistory(
        runnable=chain,  # 处理链（已包含修剪+Agent）
        get_session_history=get_session_history,  # 历史记录获取函数（连接Redis）
        input_messages_key="input",  # 指定输入字典中用户消息的key
        history_messages_key="chat_history",  # 指定历史消息在输入字典中的key（必须与prompt中的占位符一致）
    )

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 创建带记忆的Agent
    agent_with_memory = create_agent_with_memory()
    
    # 配置用户和会话（这些参数会传递给 get_session_history）
    config = {"configurable": {"session_id": "user123:sess456"}}
    
    print("=== 测试带记忆和修剪器的Agent ===\n")
    
    # 第一轮：介绍自己（会保存到Redis）
    res1 = agent_with_memory.invoke(
        {"input": "我叫张三，是一名软件工程师，我喜欢Python"}, 
        config
    )
    print(f"助手: {res1['output']}\n")
    
    # 第二轮：测试记忆（会从Redis读取历史，应该记得我的职业）
    res2 = agent_with_memory.invoke(
        {"input": "我的职业是什么？"}, 
        config
    )
    print(f"助手: {res2['output']}\n")
    
    # 第三轮：使用工具 + 记忆（结合历史信息和工具调用）
    res3 = agent_with_memory.invoke(
        {"input": "我刚才说的编程语言是什么？帮我算一下这个语言的字母数"}, 
        config
    )
    print(f"助手: {res3['output']}\n")
    
    # ===== 可选：手动查看Redis中的历史 =====
    # 直接调用 get_session_history 可以查看原始存储的历史（未修剪的完整历史）
    history = get_session_history("user123:sess456")
    print(f"\n===== Redis中存储的原始历史 =====")
    print(f"历史消息总数: {len(history.messages)}")
    for i, msg in enumerate(history.messages):
        print(f"{i+1}. {msg.type}: {str(msg.content)[:80]}...")

2. Mysql 持久化（SQLChatMessageHistory）

2.1. 安装依赖

pip install langchain langchain-core langchain-openai langchain-community
pip install sqlalchemy pymysql  # SQLAlchemy 是核心，pymysql 是 MySQL 驱动

2.2. 基础用法：SQLChatMessageHistory 直接操作（通用所有 SQL 数据库）

from langchain_core.messages import HumanMessage, AIMessage
from langchain_community.chat_message_histories import SQLChatMessageHistory
from sqlalchemy import create_engine

# ========== 1. 配置数据库连接（MySQL 为例） ==========
# 连接格式：mysql+pymysql://用户名:密码@主机:端口/数据库名
MYSQL_URL = "mysql+pymysql://root:你的数据库密码@localhost:3306/langchain_chat"
engine = create_engine(MYSQL_URL, echo=False)  # echo=True 可打印 SQL 日志

# ========== 2. 创建 SQL 消息历史实例 ==========
history = SQLChatMessageHistory(
    session_id="user_1001", # 标识
    connection=engine,  # 传入 SQLAlchemy 引擎
    table_name="chat_message_history"  # 自动创建表，无需手动创建
)

# ========== 3. 手动操作消息 ==========
# 添加对话
history.add_message(HumanMessage(content="你好！我叫张三"))
history.add_message(AIMessage(content="你好张三！有什么可以帮你？"))

# 获取全部历史
print(history.messages)
print(f"当前会话消息总数：{len(history.messages)}")

# 清空当前会话历史
# history.clear()

案例1、LangChain 对话 + MySQL 持久化（生产级，通用 SQL）

"""
LangChain 0.3+ 通用 SQL 持久化会话短记忆（MySQL/PostgreSQL/SQLite 通用）
特性：多用户会话隔离、历史自动落库、搭配 trim_messages 裁剪上下文
"""
import os
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.runnables import RunnablePassthrough
from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import SQLChatMessageHistory
from sqlalchemy import create_engine

# ========== 数据库 & 模型配置 ==========
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
# MySQL 连接格式：mysql+pymysql://用户名:密码@主机:端口/数据库名
MYSQL_URL = "mysql+pymysql://root:你的数据库密码@localhost:3306/langchain_chat"
engine = create_engine(MYSQL_URL, echo=False)

# ========== 获取 SQL 会话历史 ==========
"""
当单用户单会话的聊天记录非常多（例如超过 50～100 条）时，LangChain 官方默认的 SQLChatMessageHistory 会一次性读取该 session_id 下的所有历史消息。
随着会话增长，会出现数据读取冗余、MySQL 查询效率变低、内存占用增加等问题。

生产环境推荐优化方案：
通过自定义类继承 SQLChatMessageHistory，重写消息读取逻辑，自定义查询SQL加上 LIMIT，只查询最近 N 条消息，避免全量加载历史数据。

搭配 trim_messages 上下文裁剪器，形成 「数据库层 LIMIT + 模型层 Token 裁剪」 双重防护
"""
def get_sql_session_history(session_id: str):
    """
    根据会话ID获取 SQL 存储的聊天历史
    支持 MySQL/PostgreSQL/SQLite，自动建表、自动持久化、会话隔离
    """
    return SQLChatMessageHistory(
        session_id=session_id, # 用它来存/取对应记忆
        connection=engine,
        table_name="chat_message_history"
    )

# ========== 创建带记忆的聊天机器人 ==========
def create_sql_chat_bot():
    # 提示词模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是AI助手，记住上下文，连贯回答"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ])

    # 大模型
    llm = ChatOpenAI(
        model="gpt-3.5-turbo",
        api_key=OPENAI_API_KEY,
        temperature=0.7
    )

    # 消息修剪器
    trimmer = trim_messages(
        max_tokens=3000,  # 每次请求最多带多少 token 的历史消息给 LLM
        strategy="last",  # 保留最近的对话，丢弃最旧的
        token_counter=len,  # 按字符数计算，虽然不精确但足够用于修剪（如果使用llm会产生费用）【推荐：使用 Qwen 官方 tokenizer 库精确计数，本地计算零费用】
        include_system=True,  # 在裁剪计算时，是否把 system 消息的 token 也算进 max_tokens 的限制里。
        start_on="human",  # 确保裁剪后的第一条消息是用户消息（符合对话格式）
        allow_partial=False,  # 不允许在消息中间截断（会完整保留或丢弃整条消息）
    )

    # 构造链路：裁剪历史 -> 提示词 -> 大模型
    chain = (
        RunnablePassthrough.assign(history=lambda x: trimmer(x["history"]))
        | prompt
        | llm
    )

    # 绑定 SQL 记忆（LangChain 0.3+ 官方推荐的记忆绑定工具）
    return RunnableWithMessageHistory(
    	 # ========== 核心参数 1：业务处理链 ==========
    	# 传入我们之前构造好的完整链路：裁剪历史 -> 提示词 -> 大模型
        runnable=chain,
     
     	# ========== 核心参数 2：记忆获取函数 ==========
        # 传入定义的 get_sql_session_history 函数
    	# LangChain 会自动调用它，根据 session_id 获取/存储对应会话的历史记录
        get_session_history=get_sql_session_history,
     
     	# ========== 核心参数 3：用户输入的 key ==========
        # 指定我们传给 invoke() 的输入字典中，哪一个 key 是用户的当前输入
	    # 对应：bot.invoke({"input": "你好"}, config) 里的 "input"
        input_messages_key="input",
     
     	# ========== 核心参数 4：历史消息的 key ==========
        # 指定我们传给 prompt 的输入字典中，哪一个 key 是历史消息列表
	    # 对应：prompt 里的 MessagesPlaceholder(variable_name="history")
        history_messages_key="history",
    )

# ========== 使用测试 ==========
if __name__ == "__main__":
    bot = create_sql_chat_bot()
    
    # 定义session_id，通常 session_id 包含用户 ID + 会话 ID
    # 这样既能实现多用户隔离，又能实现同一用户不同会话隔离。
    config = {"configurable": {"session_id": "user123:sess789"}}

    # 多轮对话自动记忆落库
    print(bot.invoke({"input": "我是一名Python开发工程师"}, config).content)
    print(bot.invoke({"input": "我的职业是什么？"}, config).content)

    # 查看 SQL 中存储的历史
    history = get_sql_session_history("user123:sess789")
    print(f"\nSQL 存储历史消息数：{len(history.messages)}")

3️⃣ 长记忆（持久化长记忆）

额外记录：基于 LangChain + LangGraph + FAISS 框架，自定义了一套长期记忆的业务编排工具：
D:\dev\python-projects\ai-engineer-training-main\ai-engineer-training-main\week07\p09-faissMEM.py

一、Mem0 长记忆框架介绍

Mem0是什么？Mem0 是专门给大模型看的「结构化长期记忆」，不是给人看的原始聊天流水记录

---

Mem0 和 LangChain 的边界

• LangChain 原生记忆：只管单会话、短期上下文（trim_messages 裁剪 + Redis/MySQL 持久化）
• Mem0：只管跨会话、长期用户记忆（向量库存储 + 语义检索）
  关系：协作，非从属——LangChain 管 “当前聊天”，Mem0 管 “记住你这个人”。

---

Mem0 AI 记忆框架：独立开源的长期记忆（跨会话）管理框架，不属于 LangChain 原生体系；负责自动抽取、向量化、存储、语义检索用户长期记忆，与 LangChain 的 ** 短期会话记忆（trim_messages + Redis/MySQL）** 互补。

Mem0 它让AI能够"记住"之前的对话和用户偏好，实现真正的个性化交互，可跨客户端【底层通过问题/回答向量化 >> 从向量数据库查询用户历史记录】。
—— 解决“模型记性差，跨会话不认人”的问题。
—— 突破了传统LLM的上下文窗口限制，能够跨会话保持用户信息和偏好支持长期记忆积累和演化。

---

Mem0的工作方式
Mem0 是一个需要手动维护的长期记忆库：必须手动查询记忆、手动新增交互记录，它不会自动读取或保存对话。

用户提问
↓ 你手动调用 search() 检索记忆
↓ 你手动把记忆拼进 Prompt 给大模型
↓ 大模型生成回答
↓ 你手动调用 add() 保存本次对话

二、LangChain + Mem0 案例（长记忆）

Mem0 的核心价值在于语义检索，依赖向量数据库。
mem = Memory() 在首次运行时，会自动配置并使用本地文件模式的 Qdrant向量数据库 和 SQLite（可以切换其他向量数据库）

"""
教学案例：LangChain + Mem0 智能问答助手
核心重点：**Mem0 全部常用API详解教学**
仅用LangChain完成基础大模型问答，所有篇幅重点讲解Mem0记忆操作
"""
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from mem0 import Memory

# 加载环境变量
load_dotenv()

# ====================== 1. 初始化大模型（LangChain 基础部分） ======================
# 使用阿里云通义千问，兼容OpenAI调用格式
llm = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-turbo",
    temperature=0.3
)

# 提示词模板：注入用户历史记忆 + 用户当前问题
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能问答助手，请结合用户的历史记忆信息回答用户问题。"),
    ("system", "用户历史记忆信息：{memory_context}"),
    ("user", "用户当前问题：{user_query}")
])

# 构建LangChain基础调用链
chain = prompt | llm | StrOutputParser()

# ====================== 2. 初始化 Mem0 长期记忆模块 ======================
# 默认配置：本地SQLite持久化存储 mem0.db，程序重启记忆不丢失
mem = Memory()

# 定义唯一用户ID，用于多用户记忆隔离（所有Mem0方法都需要user_id）
USER_ID = "test_user_001"


# ====================== 3. 定义问答对话主函数 ======================
def chat_answer(user_query: str):
    """
    完整问答流程：
    1. 手动检索用户相关历史记忆
    2. 记忆信息注入提示词，大模型生成回答
    3. 对话结束后，手动保存本次交互到Mem0长期记忆
    """
    
    # ============= 【Mem0 核心方法1：get_all() 获取user_id所有记录，可做多窗口隔离】 ===========
    # 作用：拉取指定用户**所有**历史记忆（全量返回）
    all_memory = mem.get_all(user_id=USER_ID)
    print("\n1. mem.get_all(user_id) 【获取用户全部记忆】")
    print(all_memory)

    # 解析检索结果，拼接成文本注入提示词
    memory_context = "\n".join([item.memory for item in all_memory]) if all_memory else "暂无用户历史记忆"
    print(f"\n【Mem0 检索到的相关记忆】\n{memory_context}")

    # ====================== LangChain 生成回答（基础问答逻辑） ======================
    answer = chain.invoke({
        "memory_context": memory_context,
        "user_query": user_query
    })
    print(f"\n【AI 回答】\n{answer}")

    # ====================== 【Mem0 核心方法2：add() 新增/更新长期记忆】 ======================
    # 作用：手动提交本轮对话内容，存入用户长期记忆
    # Mem0 内部**自动处理**：自动抽取关键信息、自动去重、自动更新冲突记忆、自动持久化存储
    # 参数1：messages 对话消息列表，格式固定 role + content
    # 参数2：user_id 用户ID，实现多用户记忆隔离
    mem.add(
        messages=[
            {"role": "user", "content": user_query}, # 问题
            {"role": "assistant", "content": answer} # 回答
        ],
        user_id=USER_ID
    )
    print("\n>>> 本次对话内容已自动抽取并保存至Mem0长期记忆库")

    return answer


# ====================== 4. 演示 Mem0 其余全部常用方法 ======================
def show_all_mem0_api(user_query: str):   # 修复：增加参数 user_query，用于 search 演示
    """单独演示Mem0所有剩余常用API，用于记忆查看、管理、删除"""
    print("【Mem0 全部剩余常用方法演示】")

    # ====================== 【Mem0 核心方法3：search() 语义检索记忆】 ======================
    # 底层使用向量化 + 向量数据库实现【不建议用！！！】
    # 作用：根据用户当前提问，**语义相似度检索**该用户相关的历史记忆，步骤：取出全部记忆 → 丢给大模型 → 让大模型筛选
    # 参数：query=检索问题、user_id=用户唯一标识
    memory_result = mem.search(query=user_query, user_id=USER_ID)

    # 提取全部记忆文本展示（注意：search 返回的也是对象列表，用 .memory 访问）
    if memory_result:
        all_text = "\n".join([m.memory for m in memory_result])
        print("语义检索到的记忆内容：\n", all_text)
    else:
        print("未检索到相关记忆")

    # ====================== 【Mem0 方法4：get() 根据ID精准获取单条记忆】 ======================
    # 作用：通过单条记忆的唯一id，精准查询某一条记忆详情
    # 修复：先获取所有记忆，再取第一条的 id；方法名改为 get()，且不需要 user_id
    all_memory = mem.get_all(user_id=USER_ID)   # 获取全部记忆用于演示
    if all_memory:
        first_memory_id = all_memory[0].id      # 修复：用 .id 而不是 ["id"]
        single_mem = mem.get(memory_id=first_memory_id)   # 修复：方法名 get，去掉 user_id
        print("\n2. mem.get(memory_id) 【根据ID精准查询单条记忆】")
        print(single_mem)

    # ====================== 【Mem0 方法5：delete() 删除单条记忆】 ======================
    # 作用：根据memory_id，删除指定的某一条记忆
    # 谨慎使用，删除后不可恢复
    # if all_memory:
    #     mem.delete(memory_id=first_memory_id)
    #     print("\n3. mem.delete(memory_id) 【单条记忆已删除】")

    # ====================== 【Mem0 方法6：delete_all() 清空用户全部记忆】 ======================
    # 作用：一键清空当前用户所有的全部记忆
    # 全部清空，不可恢复，一般用于重置用户记忆
    # 修复：方法名 delete_all，而不是 clear
    # mem.delete_all(user_id=USER_ID)
    # print("\n4. mem.delete_all(user_id) 【该用户所有记忆已全部清空】")


# ====================== 5. 程序运行测试 ======================
if __name__ == "__main__":
    print("========== LangChain + Mem0 问答教学案例 ==========")
    print("重点：Mem0 全API用法详解、手动调用流程、内部自动记忆处理\n")

    # 第一轮对话：新增用户信息记忆
    chat_answer("我平时喜欢网购，经常寄快递，寄件偏好使用顺丰")

    # 第二轮对话：语义检索验证记忆
    chat_answer("我寄件一般选择哪家快递？")

    # 第三轮对话：用户修改信息，验证Mem0自动更新记忆
    chat_answer("我现在更改寄件偏好，以后优先使用中通快递")

    # 第四轮对话：再次检索，验证记忆已更新
    chat_answer("我现在的寄件偏好是什么？")

    # 演示Mem0 全部管理API：查看、获取、删除相关方法
    # 修复：传入一个查询字符串，用于 search 演示
    show_all_mem0_api("寄件偏好")

4️⃣ 补充其他记忆（摘要记忆、向量记忆、知识图谱记忆）

1. SummarizationNode 摘要记忆（LangGraph 的新方案）

摘要记忆： 一种通过 LLM 将超长历史消息压缩成精炼摘要的长期记忆策略。

核心思路是：当对话历史超过设定阈值时，调用 LLM 将早期消息压缩成一段精炼摘要（例如不超过 300 token），然后摘要 + 最近若干条完整消息共同进入上下文，既保留关键语义信息，又控制 token 成本。

**典型使用场景 ：**客服、面试、辅导等长对话场景，以及任何需要长期追踪用户偏好、需求或关键决策，但又不能无限制消耗上下文窗口的对话系统。

D:\dev\python-projects\ai-engineer-training-main\ai-engineer-training-main\week07\p06-summaryMEM.py

2. 手写向量记忆的原理（类比Mem0框架）

内存级：D:\dev\python-projects\ai-engineer-training-main\ai-engineer-training-main\week07\p08-vectorMEM.py

持久化：D:\dev\python-projects\ai-engineer-training-main\ai-engineer-training-main\week07\p09-faissMEM.py

3. 知识图谱记忆

D:\dev\python-projects\ai-engineer-training-main\ai-engineer-training-main\week07\p10-KnowledgeTripleMEM.py