AI Agent开发学习系列 - langchain之LCEL(4)：Memory

alex100

731人浏览 · 2025-07-26 13:27:30

alex100 · 2025-07-26 13:27:30 发布

Memory的添加方式

from operator import itemgetter

from 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_dotenv

load_dotenv()

model = ChatOpenAI(
    api_key=SecretStr(os.environ.get("HUNYUAN_API_KEY", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="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 Optional

from 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_dotenv

load_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", "")),  # 混元 APIKey
    base_url="https://api.hunyuan.cloud.tencent.com/v1",  # 混元 endpoint
    model="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，提供了高性能、可扩展的对话历史管理功能。