一、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。
今日核心总结
- RunnableLambda:将自定义函数转为Runnable实例入链,lambda可直接传入
- 临时记忆:
RunnableWithMessageHistory+InMemoryChatMessageHistory,按session_id隔离 - 长期记忆:继承
BaseChatMessageHistory,实现文件读写,持久化保存 - 调用格式:invoke时传
{"configurable": {"session_id": "xxx"}},自动关联历史 - 调试技巧:链中加入自定义函数print信息,原样返回不破坏业务