大模型学习·第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方法像属性一样访问。

四、两种记忆方案对比

核心流程：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信息，原样返回不破坏业务