LangChain学习之旅(三)：用Memory赋予模型记忆

一、 痛点重现：大模型为什么会“失忆”？

1. 破除迷思：API 是无状态的

很多初学者误以为大模型像人一样有“脑子”，能记住刚才说过的话。但残酷的事实是：大模型的 API 调用本质上是无状态的 (stateless)。每一次invoke()对你来说是对话的延续，但对模型来说，它看到的只是一个孤立的请求——仿佛宇宙在这一刻刚刚诞生。

类比：你把大模型想象成一个极其聪明但患有严重失忆症的音频处理专家。每次你走进他的办公室，他都不记得你是谁，也不记得你五分钟前问过他什么。你必须在每次提问时，把之前所有的对话内容重新复述一遍，他才能“想起来”。

2. 反面实战：裸调 API 的翻车现场

为了直观展示这个痛点，我们先写一段没有记忆的循环对话代码（对应项目中的03_no_memory.py）：

# 模拟多轮对话（无记忆）questions=["请用 librosa 写一段 Python 代码：读取音频文件并提取它的梅尔频谱。注意：请把变量名命名为 `my_super_mel_data`。只要代码，不要解释。","在刚才那段代码的基础上，帮我加一行代码：把提取出来的频谱画成图并保存为 mel.png。只要代码，不要解释。",]fori,qinenumerate(questions,1):print(f"\n👤 第{i}轮用户:{q}")response=llm.invoke(q)print(f"🤖 模型回答:{response.content}")

运行结果是这样的：

👤 第1轮用户: 请用 librosa 写一段 Python 代码：读取音频文件并提取它的梅尔频谱。注意：请把变量名命名为 `my_super_mel_data`。只要代码，不要解释。 🤖 模型回答: import librosa y, sr = librosa.load('audio_file.wav') my_super_mel_data = librosa.feature.melspectrogram(y=y, sr=sr) 👤 第2轮用户: 在刚才那段代码的基础上，帮我加一行代码：把提取出来的频谱画成图并保存为 mel.png。只要代码，不要解释。 🤖 模型回答: import matplotlib.pyplot as plt plt.figure(figsize=(10,4)) plt.imshow(mel_spectrogram, aspect='auto', origin='lower', cmap='viridis') plt.colorbar(format='%+2.0f dB') plt.tight_layout() plt.savefig('mel.png', dpi=300) plt.close()

彻底翻车——模型完全不记得我们在第一轮专门命名的my_super_mel_data变量，而在第二轮里自顾自地使用了它“脑补”出来的变量名mel_spectrogram。如果你直接把这两段代码复制粘贴到一起运行，当场就会报错NameError: name 'mel_spectrogram' is not defined。这就是生产环境中绝对不能接受的“失忆症”。

3. 核心原理：Memory 的本质是什么？

要让模型“记住”，其实很简单：每次提问时，我们把之前的聊天记录偷偷塞进 Prompt 里，假装模型本来就知道。

这是 Memory 组件的核心思想：历史对话 -> 注入 Prompt -> 模型感知上下文。

注意，这里不仅塞入用户的提问，还会完整地塞入模型之前的回答，以维持对话语义的连贯性。

[用户第1轮输入]:"请用 librosa 写一段 Python 代码：读取音频文件并提取它的梅尔频谱。注意：请把变量名命名为 `my_super_mel_data`。只要代码，不要解释。"[模型第1轮输出]:"import librosa\ny, sr = librosa.load('audio_file.wav')\nmy_super_mel_data = librosa.feature.melspectrogram(y=y, sr=sr)"当用户发起第2轮提问时，LangChain 在底层实际发给大模型的完整 Prompt 结构如下：SystemMessage:"你是一个音频算法工程师。请严格按照用户的要求输出代码..."History (也就是之前发生过的对话):HumanMessage:"请用 librosa 写一段 Python 代码：读取音频文件并提取它的梅尔频谱。注意：请把变量名命名为 `my_super_mel_data`。只要代码，不要解释。"AIMessage:"import librosa\ny, sr = librosa.load('audio_file.wav')\nmy_super_mel_data = librosa.feature.melspectrogram(y=y, sr=sr)"HumanMessage (当前用户最新的输入):"在刚才那段代码的基础上，帮我加一行代码：把提取出来的频谱画成图并保存为 mel.png。只要代码，不要解释。"

这也解释了为什么“上下文窗口 (Context Window)”如此珍贵——因为每次对话都会占用 token 配额，窗口越大，能记住的历史越多，但 API 调用的成本也就越高。

二、 LangChain 的基础解法：RunnableWithMessageHistory

在 LangChain 1.0+ 的 LCEL 架构中，官方推荐使用RunnableWithMessageHistory来为链添加记忆功能。这不仅是一层简单的包装，而是它优雅地分离了核心处理逻辑 (Chain)与会话状态管理 (Session History)。

1. 拆解RunnableWithMessageHistory的四大核心要素

当你使用RunnableWithMessageHistory包装一个链时，你需要告诉它四个关键信息：

1. runnable: 你要包装的那个没有记忆的原始链（比如prompt | llm）。
2. get_session_history: 一个获取历史记录的回调函数。大模型每次被调用时，都会传入当前用户的session_id。这个函数需要去数据库（或本地文件/Redis）里查出这个 ID 对应的聊天记录，返回一个BaseChatMessageHistory对象。
3. input_messages_key: 告诉包装器，用户当前说的话（比如“帮我降噪”），在你的 Prompt 模板里对应的变量名叫什么（通常是"input"）。
4. history_messages_key: 告诉包装器，从数据库里查出来的历史聊天记录，应该塞到 Prompt 模板里的哪个占位符（通常是"history"）。

搞懂了这四个要素，我们就能基于第二篇的音频调度器，给它装上具备真实本地文件持久化能力的记忆（对应项目中的03_buffer_memory.py）。

💡 行业迷思纠正：聊天记录存在哪？
很多开发者以为调用 OpenAI 或火山引擎的接口，模型就会在服务器上帮你存下对话历史。这是完全错误的！
大模型的 API 是纯无状态的（除非你使用了 Assistant API 这种有状态的高级封装）。在标准的 LangChain LCEL 架构中，历史记录必须由开发者自己存储在本地（或开发者自己的数据库如 Redis 中）。每次调用模型时，LangChain 只是把本地取出的历史数据打包，和当前问题一起作为一长串文本发送给大模型服务器。

💣 隐蔽的工程深坑：Memory 与 Structured Output 的冲突
如果你直接把with_structured_output绑定在带有记忆的链上，你会发现历史文件永远是空的。原因是：Memory 组件期望保存的是标准对话消息（AIMessage），而结构化输出强行将其转成了 Pydantic 对象，导致保存机制崩溃。
正规军解法：让 Memory 组件包裹基础模型（保持原生对话），拿到AIMessage后，再在最外层使用结构化模型对其内容进行提取。

importosfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistoryfromlangchain_core.messagesimportmessages_to_dictfrompydanticimportBaseModel,Field load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__),'..','.env'))# ---------- 1. 定义结构化输出模型 ----------classAudioMetadata(BaseModel):task_type:str=Field(description="音频处理任务类型，如：降噪(OfflineNS)、语音识别(ASR)、文本转语音(TTS)")target_sample_rate:int=Field(description="目标采样率（Hz），如果未提及默认返回 16000")language:str=Field(description="处理语言，如：zh, en。如果未提及默认返回 zh")priority:str=Field(description="任务优先级(high/low)。如果用户提到'紧急'、'立刻'则为 high，否则默认为 low")is_batch:bool=Field(description="是否为批量处理。如果用户提到'这批'、'所有'等复数词汇则为 True，否则为 False")defmain():# ---------- 2. 初始化模型和链 ----------frompydanticimportSecretStr llm=ChatOpenAI(api_key=SecretStr(os.getenv("ARK_API_KEY")oros.getenv("OPENAI_API_KEY")or""),base_url=os.getenv("OPENAI_API_BASE"),model=os.getenv("LLM_MODEL_NAME")or"doubao-seed-2-0-mini-260428",temperature=0.1,max_completion_tokens=2000)structured_llm=llm.with_structured_output(AudioMetadata)prompt=ChatPromptTemplate.from_messages([("system","你是一个资深的音频处理专家。请根据用户的自然语言描述，提取出音频处理任务的核心元数据。"),MessagesPlaceholder(variable_name="history"),("human","{input}")])# ---------- 3. 配置记忆 (使用核心包的内存记忆) ----------# ⚠️ 架构笔记：LangChain 1.0+ 推荐使用 langchain-core 的 InMemoryChatMessageHistory 用于开发。# 生产环境中推荐迁移至 langchain-redis 或 langchain-postgres 等专用持久化包。# （旧版中的 FileChatMessageHistory 来自 langchain_community，已进入维护模式）store={}defget_session_history(session_id:str):ifsession_idnotinstore:store[session_id]=InMemoryChatMessageHistory()returnstore[session_id]# 注意：为了让 Memory 组件能正确保存 AI 的原始文本回复，# 我们不能直接把与 with_structured_output() 绑定的链放进去。# 正确做法：组装基础带记忆的链（它吐出的是 AIMessage）base_chain_with_memory=RunnableWithMessageHistory(prompt|llm,get_session_history,input_messages_key="input",# 对应 Prompt 中的 {input}history_messages_key="history"# 对应 Prompt 中的 MessagesPlaceholder(variable_name="history"))# ---------- 4. 模拟多轮对话 ----------session_id="user_audio_002"# 每次运行使用唯一会话 ID# 运行前清理掉历史遗留文件，保证每次演示结果一致history_file=os.path.join(os.path.dirname(__file__),"chat_histories",f"{session_id}.json")ifos.path.exists(history_file):os.remove(history_file)questions=["我有一段英语的播客音频，需要做降噪处理，采样率统一重采样到 44100Hz。","等等，刚才说错了，是中文的访谈，不是英语。","为了节省存储空间，采样率还是降到 16000Hz 吧。","处理完之后，顺便把这段音频转写成文字记录。","最后，用这段 16000Hz 的中文音频作为参考音色，合成一段新的语音（TTS）。"]fori,qinenumerate(questions,1):print(f"\n👤 第{i}轮用户:{q}")# 触发带有记忆的基础流水线ai_msg=base_chain_with_memory.with_retry(stop_after_attempt=3).invoke({"input":q},config={"configurable":{"session_id":session_id}})# 在拿到 AI 的原生消息后，我们在外部用 structured_llm 进行一轮“提取翻译”# 这样既保证了记忆里存的是正常对话，又拿到了结构化参数result=structured_llm.invoke(ai_msg.content)print(f"🤖 模型提取参数:{result}")# ---------- 5. 模拟持久化到本地 JSON ----------# 为了在博客中直观展示底层到底存了什么数据，我们把 InMemoryChatMessageHistory 导出来存为 JSON。importjson history_file=os.path.join(os.path.dirname(__file__),"chat_histories",f"{session_id}.json")os.makedirs(os.path.dirname(history_file),exist_ok=True)withopen(history_file,"w",encoding="utf-8")asf:# messages_to_dict 会把 AIMessage/HumanMessage 序列化成标准化字典json.dump(messages_to_dict(store[session_id].messages),f,ensure_ascii=False,indent=2)if__name__=="__main__":main()

我们在终端中真实运行这段代码，结果如下：

👤 第1轮用户: 我有一段英语的播客音频，需要做降噪处理，采样率统一重采样到 44100Hz。 🤖 模型提取参数: task_type='降噪(OfflineNS)' target_sample_rate=44100 language='en' priority='low' is_batch=False 👤 第2轮用户: 等等，刚才说错了，是中文的访谈，不是英语。 🤖 模型提取参数: task_type='降噪(OfflineNS)' target_sample_rate=44100 language='zh' priority='low' is_batch=False 👤 第3轮用户: 为了节省存储空间，采样率还是降到 16000Hz 吧。 🤖 模型提取参数: task_type='降噪(OfflineNS)' target_sample_rate=16000 language='zh' priority='low' is_batch=False 👤 第4轮用户: 处理完之后，顺便把这段音频转写成文字记录。 🤖 模型提取参数: task_type='降噪(OfflineNS),语音识别(ASR)' target_sample_rate=16000 language='zh' priority='low' is_batch=False 👤 第5轮用户: 最后，用这段 16000Hz 的中文音频作为参考音色，合成一段新的语音（TTS）。 🤖 模型提取参数: task_type='降噪(OfflineNS),语音识别(ASR),文本转语音(TTS)' target_sample_rate=16000 language='zh' priority='low' is_batch=False

注意看这连续 5 轮对话的绝妙表现：模型不仅像人类一样记住了前面说的所有需求，还能根据你“朝令夕改”的指令，动态修正并叠加参数。从第一轮的单纯降噪，一路演变成了最后包含降噪、ASR 转写、TTS 合成的复合音频处理调度流。

这就是大模型作为“大脑”的威力：记忆生效了，且下游的音频处理管道成功接管了执行！

💡 深度解密：本地到底存了什么？

我们在上面提到了，不能直接把with_structured_output塞进记忆里。为了让你知其然更知其所以然，我们直接打开刚才生成的本地文件/chat_histories/user_audio_002.json，看看它底层的数据结构（截取其中一段）：

[{"type":"human","data":{"content":"我有一段英语的播客音频，需要做降噪处理，采样率统一重采样到 44100Hz。"}},{"type":"ai","data":{"content":"核心元数据：\n1. 音频内容类型：英语播客\n2. 处理任务：降噪处理、采样率重采样\n3. 重采样目标参数：44100Hz"}},{"type":"human","data":{"content":"等等，刚才说错了，是中文的访谈，不是英语。"}}]

看到了吗？JSON 文件里保存的是大模型“最原始、最自然”的思考过程，而不是冷冰冰的 Pydantic 参数对象（比如task_type='降噪(OfflineNS)'）。

正是因为有了这份包含完整自然语言推理细节的历史记录，在后续提问时，模型才能准确推断出前因后果。这也是我们在架构上必须使用base_chain_with_memory获取AIMessage，再在最外层用structured_llm提取 JSON 的根本原因。

三、 架构进阶：Token 爆炸与工业级记忆方案

1. 新的痛点：全量记忆的代价

像我们刚才演示的那种全量记录历史消息的方式，虽然简单有效，但在真实的工业场景中存在一个致命缺陷：随着对话轮次增加，历史记录会无限膨胀。

很多初学者以为历史记录膨胀仅仅是“更费钱”（API 调用成本线性上升），但实际上它会带来更致命的工程灾难。历史记录不仅会占用输入 Token，更是直接受限于大模型的“上下文窗口（Context Window）”物理上限（如 32K、128K）。

在一个完整的对话请求中：
上下文窗口上限 = System Prompt + 历史对话 (History) + 当前用户输入 + 模型输出内容 (含 CoT 推理消耗)

一旦历史记录的无序膨胀挤占了过多空间，就会引发以下惨案：

1. API 级熔断（硬超载）：总 Token 超出绝对物理上限，API 直接抛出context_length_exceeded错误，导致下游业务宕机。
2. 推理截断与哑火（软超载）：这是当前带有深度思考（CoT）能力的大模型最容易踩的坑。假设模型窗口为 128K，历史对话吃掉了 127.5K，留给模型生成的空间只剩区区 500 Token。大模型在启动<think>推理过程时，这 500 Token 瞬间被耗尽，最终抛出LengthFinishReasonError（达到长度限制），表现为模型思考了很久，最后却返回了一段空白。

因此，对历史记忆进行物理或语义上的“垃圾回收”，是任何工业级 AI 应用的必修课。LangChain 为此提供了两种核心的内存管理方案：

真实案例：某智能客服机器人使用全量记忆，用户闲聊了 50 轮后，单次请求的 token 消耗从 200 暴涨到 15,000，月成本直接翻了 75 倍！

2. 实用主义方案：滑动窗口记忆 (Window Memory)

在实际的音频任务调度或客服场景中，用户通常只需要模型记住“最近的几句话”即可。因此，工业界最常用的降本手段是引入滑动窗口机制。

它的原理是：只保留最近k轮的对话记录。例如设置k=5，当进行第 6 轮对话时，第 1 轮的记录会被自动丢弃。这保证了 token 消耗永远在一个可控的常数范围内，是性价比最高的生产级选择。

在 LangGraph 或原生实现中，我们通常会通过trim_messages工具在把消息喂给大模型前进行一次裁剪。为了让你直观感受到它的威力，这里提供一个完整可运行的简易对话版本（对应项目中的03_window_memory.py）：

importosfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain_core.runnables.historyimportRunnableWithMessageHistoryfromlangchain_core.chat_historyimportInMemoryChatMessageHistoryfromlangchain_core.messagesimporttrim_messages,messages_to_dict load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__),'..','.env'))defmain():frompydanticimportSecretStr llm=ChatOpenAI(api_key=SecretStr(os.getenv("ARK_API_KEY")oros.getenv("OPENAI_API_KEY")or""),base_url=os.getenv("OPENAI_API_BASE"),model=os.getenv("LLM_MODEL_NAME")or"doubao-seed-2-0-mini-260428",temperature=0.1,max_completion_tokens=2000)prompt=ChatPromptTemplate.from_messages([("system","你是一个音频处理助手，尽量简短回答用户的问题。"),MessagesPlaceholder(variable_name="history"),("human","{input}")])# 🌟 核心改动：设定 max_tokens=3，只保留系统提示词 + 最近的 1 问 1 答# ⚠️ 参数说明：`max_tokens=3` 在此处配合 `token_counter=len` 表示“最多保留 3 条消息”。# 如果按真实 token 数计算，应传入 `token_counter=num_tokens_from_messages` 等函数。trimmer=trim_messages(max_tokens=3,# 按消息“条数”计算token_counter=len,# len 函数统计的是消息条数strategy="last",# 保留最后 3 条allow_partial=False,# 不截断单条消息的内容include_system=True,start_on="human")# ---------- 3. 配置记忆 (使用内存记忆) ----------store={}defget_session_history(session_id:str):ifsession_idnotinstore:store[session_id]=InMemoryChatMessageHistory()returnstore[session_id]# 新的链结构：组装 Prompt -> 裁剪历史 -> 喂给大模型base_chain_with_window=RunnableWithMessageHistory(prompt|trimmer|llm,get_session_history,input_messages_key="input",history_messages_key="history")session_id="user_window_001"# 因为本例最后会将内存中的历史记录保存为 JSON 文件，# 所以在此处先清理掉可能存在的同名遗留文件，保证每次演示结果一致。history_file=os.path.join(os.path.dirname(__file__),"chat_histories",f"{session_id}.json")ifos.path.exists(history_file):os.remove(history_file)os.makedirs(os.path.dirname(history_file),exist_ok=True)questions=["第一句：你好，我叫张三。我有一段会议录音需要做降噪。","第二句：顺便把采样率重采样到 44100Hz。","请问我一开始告诉你我叫什么名字？"]fori,qinenumerate(questions,1):print(f"\n👤 第{i}轮用户:{q}")ai_msg=base_chain_with_window.invoke({"input":q},config={"configurable":{"session_id":session_id}})print(f"🤖 模型回答:{ai_msg.content}")print(f"💰 本轮消耗 Token:{ai_msg.response_metadata['token_usage']['total_tokens']}")importjsonwithopen(history_file,"w",encoding="utf-8")asf:json.dump(messages_to_dict(store[session_id].messages),f,ensure_ascii=False,indent=2)if__name__=="__main__":main()

运行这段代码，你会看到极其直观的 Token 消耗和记忆丢失的对比：

👤 第1轮用户: 第一句：你好，我叫张三。我有一段会议录音需要做降噪。 🤖 模型回答: 你好张三，我已了解你需要会议录音降噪的需求，请提供对应的录音文件，我会为你完成降噪处理。 💰 本轮消耗 Token: 259 (第1轮：初始对话) 👤 第2轮用户: 第二句：顺便把采样率重采样到 44100Hz。 🤖 模型回答: 好的，我会将音频重采样至44100Hz。 💰 本轮消耗 Token: 261 (第2轮：历史累积) 👤 第3轮用户: 请问我一开始告诉你我叫什么名字？ 🤖 模型回答: 你并没有告诉我你的名字哦。 💰 本轮消耗 Token: 197 (第3轮：滑动窗口生效，Token 下降！)

看！到了第 3 轮，因为滑动窗口的无情裁剪，模型彻底忘记了“张三”这个名字。但也正因为这种物理截断，你在第 3 轮消耗的总 Tokens 数从 364 不升反降，暴跌到了 174！这就是工业界用来防止 Token 破产的最核心杀手锏。

3. 终极压缩方案：摘要记忆 (Summary Memory)

如果你的业务场景（如心理咨询 AI、长篇剧本创作）确实需要模型记住几百轮之前的核心线索，滑动窗口就不够用了。这时候需要引入摘要压缩的思路。

它的思路非常巧妙：用大模型自己来压缩历史。每次对话后，它会在后台悄悄调用一次大模型，把长篇的历史记录浓缩成一句简短的摘要（例如：“用户正在处理一段英语电话录音，先要求降噪，然后要求转写”），然后只把这段几十个 token 的摘要传给主模型。

它的伪代码逻辑大致如下：

# 当历史对话累积到一定长度时，触发后台的“摘要压缩模型”summary_prompt="请把以下对话记录压缩成 100 字以内的摘要，保留核心的音频处理参数要求：\n{history}"summary_llm=ChatOpenAI(model="廉价模型如 doubao-lite")compressed_history=summary_llm.invoke(summary_prompt)# 将压缩后的摘要覆盖写入数据库save_to_database(session_id,compressed_history)

优点：极大节省上下文 token，理论上支持无限轮对话。
缺点：每次对话后多一次大模型调用，增加了延迟和成本，且摘要过程中可能会丢失细节。

四、 工程陷阱与最佳实践

在将 Memory 组件推向生产环境时，请务必注意以下几点：

1. Session ID 管理：在RunnableWithMessageHistory中，必须为每个用户或每次独立任务分配唯一的session_id。
2. 持久化存储：我们在代码中使用了 Python 的内存字典session_store = {}。在真正的服务器部署中，一旦进程重启记忆就会清空。请务必替换为 Redis、PostgreSQL 或 MongoDB 等外部数据库来持久化历史记录。
3. 成本监控：为对话轮次或单次请求 token 设置熔断机制。遇到超长恶意对话时，必须有策略（如自动清理、强制摘要）防止被薅羊毛。

五、 总结与下期预告

通过本篇的学习，我们掌握了：

• 大模型无状态的本质原因。
• Memory 组件的工作原理：历史对话注入 Prompt。
• 利用RunnableWithMessageHistory优雅地为 LCEL 管道添加记忆。
• 应对 Token 爆炸的进阶选型：滑动窗口与摘要压缩。

然而，Memory 解决的只是短时上下文（多轮对话）的问题。如果用户突然问：“按照我们公司《2026年最新音频质检规范》的第3条，这段降噪后的音频合格吗？”——模型不仅没见过这份私有文档，即使你想把它塞进 Memory，几万字的文档也塞不下！

这就引出了大模型落地的下一个核心痛点：私有知识的注入。

下一篇，我们将迎来 LangChain 最重磅的应用模式——RAG（检索增强生成）：给大模型外挂一个“超级硬盘”，让它能实时检索企业文档、内部规范、知识库，从而精准回答任何私有领域的问题。敬请期待！