AI Agent開發學習系列 - langchain之LCEL(4)：Memory

Memory的添加方式

from operator import itemgetterfrom langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI
from pydantic import SecretStr
import os
from dotenv import load_dotenvload_dotenv()model = ChatOpenAI(api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKeybase_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpointmodel="hunyuan-lite",  # 模型名稱temperature=0
)
prompt = ChatPromptTemplate.from_messages([("system", "你是一個可以幫助人類的機器人"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),]
)

MessagesPlaceholder 是 LangChain 中用于在聊天提示模板中插入消息列表的占位符。它主要用于處理對話歷史和動態消息內容。

variable_name: 指定占位符的變量名，用于在運行時匹配對應的消息列表
這個變量名需要與 Memory 組件的 memory_key 或 history_messages_key 保持一致

memory = ConversationBufferMemory(return_messages=True)

ConversationBufferMemory 是 LangChain 中最基礎、最常用的記憶組件，用于存儲對話歷史。它將所有的對話消息按順序保存在內存中，形成一個完整的對話緩沖區。
return_message參數：

True：返回消息對象列表（HumanMessage, AIMessage）
False：返回格式化的字符串

memory.load_memory_variables({})

load_memory_variables 是 LangChain 中所有 Memory 組件的核心方法，用于從記憶組件中加載記憶變量。它返回一個字典，包含當前存儲的記憶內容。
示例：基于當前輸入查詢相關記憶
variables = memory.load_memory_variables({"input": "Where is Lisa?"})

# 增加一條鏈
chain = (RunnablePassthrough.assign(history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"))| prompt| model
)

RunnablePassthrough 是一個特殊的可運行組件，用于在鏈中傳遞數據。它可以：

直接傳遞輸入數據
使用assign方法添加新的鍵值對到輸入字典中
作為數據流的"橋梁"

RunnableLambda 是 LangChain LCEL 中用于將任意 Python 函數包裝成可運行組件的工具。它允許您在鏈中插入自定義函數邏輯。
itemgetter 是 Python 標準庫 operator 模塊中的一個函數，用于從字典、列表等可索引對象中提取特定位置的元素。在 LangChain LCEL 中，它常用于數據流處理中提取特定的鍵值。

這里 itemgetter(“history”) 從 memory.load_memory_variables() 返回的字典中提取 “history” 鍵對應的值。

這段代碼構建了一個帶記憶功能的對話鏈。RunnablePassthrough.assign() 在保持原始輸入的同時，通過 RunnableLambda(memory.load_memory_variables) 從記憶組件中加載歷史對話，然后用 itemgetter(“history”) 提取出對話歷史列表，將其作為 history 字段添加到輸入數據中。這樣，當數據流經 prompt 模板時，歷史對話會被注入到提示詞中，最終讓 model 能夠基于完整的對話上下文生成回復，實現連續對話的記憶功能。

inputs = {"input": "你好，我是小明，很高興認識你。"}
response = chain.invoke(inputs)
response

輸出：

AIMessage(content='你好，小明！我也很高興認識你，希望我們能成為好朋友😄 無論是聊天、玩耍還是其他方面，都可以隨時找我哦。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 34, 'prompt_tokens': 18, 'total_tokens': 52, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': '202f461322c14a2b37232dcb3e72cd56', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--e16a133a-6c92-46d0-a6a0-32deff6d2d7a-0', usage_metadata={'input_tokens': 18, 'output_tokens': 34, 'total_tokens': 52, 'input_token_details': {}, 'output_token_details': {}})

#保存記憶
memory.save_context(inputs, {"output": str(response.content)})
memory.load_memory_variables({})

輸出：

{'history': [HumanMessage(content='你好，我是小明，很高興認識你。', additional_kwargs={}, response_metadata={}),AIMessage(content='你好，小明！我也很高興認識你，希望我們能成為好朋友😄 無論是聊天、玩耍還是其他方面，都可以隨時找我哦。', additional_kwargs={}, response_metadata={})]}

inputs = {"input": "我是誰？"}
response = chain.invoke(inputs)
response

AIMessage(content='你是小明呀，我記得你。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 9, 'prompt_tokens': 57, 'total_tokens': 66, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': '021589cde3e5617bedeacaeff5363d11', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--20559b8d-df04-447a-ad7f-a66524eabb0d-0', usage_metadata={'input_tokens': 57, 'output_tokens': 9, 'total_tokens': 66, 'input_token_details': {}, 'output_token_details': {}})

這個 AIMessage 顯示模型回答"你是小明呀，我記得你"，說明記憶功能正常工作。記憶的實現原理是：

歷史注入：通過 RunnablePassthrough.assign() 將 memory.load_memory_variables() 返回的歷史對話作為 history 字段添加到輸入中
提示詞構建：prompt 模板中的 MessagesPlaceholder(variable_name=“history”) 會將歷史對話插入到提示詞中，形成完整的上下文
上下文感知：模型接收到包含歷史對話的完整提示詞，能夠理解之前的對話內容，因此能夠記住用戶之前介紹過自己叫"小明"

使用Redis來實現長時記憶

先確保本地安裝了redis sever并啟動：

brew install redis
brew services start redis

pip install redis

from typing import Optionalfrom langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from pydantic import SecretStr
import os
from dotenv import load_dotenvload_dotenv()prompt = ChatPromptTemplate.from_messages([("system", "你是一個擅長{ability}的助手"),MessagesPlaceholder(variable_name="history"),("human", "{question}"),]
)chain = prompt | ChatOpenAI(api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKeybase_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpointmodel="hunyuan-lite",  # 模型名稱temperature=0
)chain_with_history = RunnableWithMessageHistory(chain,# 使用redis存儲聊天記錄lambda session_id: RedisChatMessageHistory(session_id, url="redis://localhost:6379/0"),input_messages_key="question",history_messages_key="history",
)# 每次調用都會保存聊天記錄，需要有對應的session_id
chain_with_history.invoke({"ability": "歷史", "question": "中國建都時間最長的城市是哪個？"},config={"configurable": {"session_id": "alex"}},
)

結果：

AIMessage(content='中國建都時間最長的城市是北京。北京作為中國的首都，有著悠久的歷史和豐富的文化底蘊。早在西周時期，北京就是諸侯國的都城之一，稱為“燕京”。在漫長的歷史進程中，北京曾多次成為中國的都城，包括金朝的燕京、元朝的大都（今北京）、明朝的首都以及清朝的京師。經過這些朝代的統治，北京成為了中國政治、經濟、文化的中心，至今已有超過800年的建都歷史。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 105, 'prompt_tokens': 18, 'total_tokens': 123, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'hunyuan-lite', 'system_fingerprint': '', 'id': '51335c0cd7911a579bb1db283adb907b', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--64edee75-ae46-49de-9802-9f6a9ed21f53-0', usage_metadata={'input_tokens': 18, 'output_tokens': 105, 'total_tokens': 123, 'input_token_details': {}, 'output_token_details': {}})

RunnableWithMessageHistory 是 LangChain 中用于為任何可運行組件（Runnable）自動添加消息歷史管理功能的包裝器。它簡化了記憶功能的實現，特別適合需要持久化對話歷史的場景。
RedisChatMessageHistory 是 LangChain 中基于 Redis 數據庫實現的消息歷史存儲類，用于持久化存儲聊天對話歷史。它繼承自 BaseChatMessageHistory，提供了高性能、可擴展的對話歷史管理功能。