当前位置: 首页 > news >正文

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,提供了高性能、可扩展的对话历史管理功能。

http://www.lryc.cn/news/600695.html

相关文章:

  • x86汇编语言入门基础(三)汇编指令篇5 串操作
  • 【架构】Docker简单认知构建
  • JAVA学习-练习试用Java实现“深度优先搜索(DFS):实现八数码问题的解法(最短路径搜索)”
  • LangChain4j低阶+高阶Api+日志配置+监听器+重试机制+超时机制
  • 【LeetCode 热题 100】131. 分割回文串——回溯
  • 算法竞赛阶段二-数据结构(35)数据结构单链表模拟实现
  • Android-广播详解
  • golang实现一个定时引擎,功能包括按照corntab的时间任务实时增加、修改、删除定时任务
  • 常见sql深入优化( 二)
  • 一文学会c++list
  • 激光雷达-相机标定工具:支持普通相机和鱼眼相机的交互式标定
  • 二叉搜索树(Binary Search Tree)详解与java实现
  • Linux 如何统计系统上各个用户登录(或者登出)记录出现的次数?
  • Android-三种持久化方式详解
  • 摘录-打造第二大脑
  • J2EE模式---表现层集成模式
  • C++ TAP(基于任务的异步编程模式)
  • Web后端进阶:springboot原理(面试多问)
  • React入门学习——指北指南(第五节)
  • JavaScript手录06-函数
  • 【RK3568 PWM 子系统(SG90)驱动开发详解】
  • 数据赋能(336)——技术平台——智能化运营
  • Java动态调试技术原理
  • 【RocketMQ】一分钟了解RocketMQ
  • 告别复杂配置!Spring Boot优雅集成百度OCR的终极方案
  • Windows 平台源码部署 Dify教程(不依赖 Docker)
  • 《C++ list 完全指南:从基础到高效使用》
  • Linux——线程同步
  • InvokeRepeating避免嵌套调用
  • C++编程学习(第16天)