ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

大模型学习·第42天:LangChain进阶——自定义函数入链与对话记忆

大模型学习·第42天:LangChain进阶——自定义函数入链与对话记忆

一、RunnableLambda:自定义函数加入Chain

普通函数不能直接用|加入链,需要用RunnableLambda包装(也可直接传入lambda,会自动转换)。

fromlangchain_core.runnablesimportRunnableLambda# 函数接收AIMessage,返回dictmy_func=RunnableLambda(lambdaai_msg:{"name":ai_msg.content})# 直接传lambda也可以,自动转换chain=first_prompt|model|(lambdaai_msg:{"name":ai_msg.content})|second_prompt|model|str_parser

两种入链方式:封装为RunnableLambda对象再入链,或直接将函数入链自动转换。函数在链中接收上一个组件的输出,返回值传给下一个组件。可以在链中用自定义函数做数据格式转换、打印调试信息等。

二、Memory临时会话记忆

RunnableWithMessageHistory在原链基础上创建带历史记录的新链,InMemoryChatMessageHistory在内存中存储历史。

fromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistory store={}defget_history(session_id):ifsession_idnotinstore:store[session_id]=InMemoryChatMessageHistory()returnstore[session_id]conversation_chain=RunnableWithMessageHistory(base_chain,# 被增强的原链get_history,# 通过session_id获取历史记录对象input_messages_key="input",# 用户输入在模板中的占位符名history_messages_key="chat_history"# 历史记录在模板中的占位符名)# 调用时需传入session_idsession_config={"configurable":{"session_id":"user_001"}}res=conversation_chain.invoke({"input":"小明有2个猫"},session_config)

工作流程:每次调用时根据session_id获取该用户的历史记录,把历史填入模板的chat_history位置,模型结合历史回答问题,新的问答自动追加到历史中。不同session_id的历史互不影响。InMemoryChatMessageHistory存在内存中,重启程序后消失,适合临时测试。

三、Memory长期会话记忆:FileChatMessageHistory

继承BaseChatMessageHistory,重写三个方法,以session_id为文件名存到本地。

fromlangchain_core.chat_historyimportBaseChatMessageHistoryclassFileChatMessageHistory(BaseChatMessageHistory):def__init__(self,session_id,storage_path):self.session_id=session_id self.file_path=os.path.join(storage_path,session_id)os.makedirs(os.path.dirname(self.file_path),exist_ok=True)defadd_messages(self,messages):all_messages=list(self.messages)all_messages.extend(messages)new_messages=[message_to_dict(m)forminall_messages]withopen(self.file_path,"w",encoding="utf-8")asf:json.dump(new_messages,f)@propertydefmessages(self):try:withopen(self.file_path,encoding="utf-8")asf:returnmessages_from_dict(json.load(f))exceptFileNotFoundError:return[]defclear(self):withopen(self.file_path,"w",encoding="utf-8")asf:json.dump([],f)

只需把get_history函数换成return FileChatMessageHistory(session_id, "./chat_history"),历史记录就能持久化到磁盘文件。@property装饰器让messages方法像属性一样访问。

四、两种记忆方案对比

方案存储位置重启后适用场景
InMemoryChatMessageHistory内存丢失临时测试
FileChatMessageHistory(自实现)本地文件保留持久化存储

核心流程base_chain(原有链)→RunnableWithMessageHistory(增强为带记忆的链)→ 调用时传session_config指定会话ID。

今日核心总结

  1. RunnableLambda:将自定义函数转为Runnable实例入链,lambda可直接传入
  2. 临时记忆RunnableWithMessageHistory+InMemoryChatMessageHistory,按session_id隔离
  3. 长期记忆:继承BaseChatMessageHistory,实现文件读写,持久化保存
  4. 调用格式:invoke时传{"configurable": {"session_id": "xxx"}},自动关联历史
  5. 调试技巧:链中加入自定义函数print信息,原样返回不破坏业务
返回列表