用LangChain LCEL 重构「记忆链」：四行代码搞定可扩展对话系统

版本：Python 3.9+、LangChain ≥ 0.1
目标：抛弃 LLMChain，全程 Runnable；同步 / 异步 / 流式一键切换

---

0. 环境一键复制

pip install -U langchain langchain-openai python-dotenv

.env 同目录

OPENAI_API_KEY=sk-xxx

# 0_env.py
import os, dotenv
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

---

1. 核心思路：记忆也是 Runnable

LCEL 规定 一切皆 Runnable → 把「历史记录」当成一个可插拔的组件，与 prompt | llm | parser 同级拼接即可。

---

2. 四行骨架：可运行最小记忆链

文件：2_mem_lcel_minimal.py

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain.memory import ChatMessageHistory
from 0_env import llm

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是助手，尽量简洁"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{question}"),
])

chain = prompt | llm  # ① 先拼普通链

# ② 把 chain 包一层“记忆”外壳
mem_chain = RunnableWithMessageHistory(
    chain,
    lambda session_id: ChatMessageHistory(),  # 工厂：每个会话一个历史实例
    input_messages_key="question",            # 用户新消息字段
    history_messages_key="history",           # 历史消息字段
)

# ③ 调用（需带 session_id 用于区分用户）
print(mem_chain.invoke(
    {"question": "我叫啥？"},
    config={"configurable": {"session_id": "user_123"}}
))
# ④ 继续对话
print(mem_chain.invoke(
    {"question": "刚才我自我介绍了吗？"},
    config={"configurable": {"session_id": "user_123"}}
))

终端打印：

您还没有告诉我您的名字。
刚才您没有进行自我介绍，只询问了我是否知道您的名字。

---

3. 把记忆换成「滑动窗口」：仅需改工厂函数

文件：3_window_lcel.py

from langchain.memory import ConversationBufferWindowMemory

def get_window_history(session_id: str):
    # 返回 ConversationBufferWindowMemory 的底层 ChatMessageHistory
    mem = ConversationBufferWindowMemory(k=3, return_messages=True)
    return mem.chat_memory

mem_chain = RunnableWithMessageHistory(
    prompt | llm,
    get_window_history,
    input_messages_key="question",
    history_messages_key="history",
)

# 同一会话连续调用 5 轮，可见最早两轮被“遗忘”
for i in range(5):
    ans = mem_chain.invoke(
        {"question": f"第{i+1}轮"},
        config={"configurable": {"session_id": "wnd"}}
    )
    print(f"轮{i+1}", ans)

---

4. 流式 + 记忆：逐字返回，内存零等待

文件：4_stream_mem.py

for chunk in mem_chain.stream(
    {"question": "用 20 字介绍 LCEL"},
    config={"configurable": {"session_id": "stream_user"}}
):
    print(chunk, end="", flush=True)

效果：文字逐蹦，历史仍自动写进内存。

---

5. 异步 + 记忆：asyncio 同样一行

文件：5_async_mem.py

import asyncio

async def main():
    ans = await mem_chain.ainvoke(
        {"question": "异步有什么好处？"},
        config={"configurable": {"session_id": "async_user"}}
    )
    print(ans)

asyncio.run(main())

---

6. 多会话管理：用字典缓存历史实例

文件：6_multi_user.py

def get_history(session_id: str):
    if session_id not in history_pool:
        history_pool[session_id] = ChatMessageHistory()
    return history_pool[session_id]

mem_chain = RunnableWithMessageHistory(
    prompt | llm,
    get_history,
    input_messages_key="question",
    history_messages_key="history",
)

# 模拟两个用户
print(mem_chain.invoke({"question": "我叫 Alice"}, config={"configurable": {"session_id": "alice"}}))
print(mem_chain.invoke({"question": "我叫 Bob"},   config={"configurable": {"session_id": "bob"}}))
print(mem_chain.invoke({"question": "我叫啥？"},   config={"configurable": {"session_id": "alice"}}))

---

7. 与 Retriever 组合：RAG + 记忆同样管道化

from langchain_core.runnables import RunnablePassthrough

retriever = vectordb.as_retriever()
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt_with_context
    | llm
)

# 再给 RAG 包一层记忆
mem_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_history,
    input_messages_key="question",
    history_messages_key="history",
)

---

8. 迁移对比：老写法 vs LCEL

老命令式	LCEL 声明式
LLMChain(prompt=p, llm=llm, memory=mem)	RunnableWithMessageHistory(prompt | llm, factory, ...)
流式需额外封装	原生 .stream() / .astream()
多会话手动维护	工厂函数自动管理

---

9. 小结：一张图带走

用户输入 ─► RunnableWithMessageHistory ─► 历史记录 ─► prompt ─► llm ─► 输出
                              ▲
                     工厂函数自动提供 ChatMessageHistory

• 全程无继承、无冗长 API
• 同一条链支持 sync / async / stream / batch
• 记忆、检索、工具任意拼接，真正的「可运行图」

---

复制以上文件即可 python xxx.py 直接运行，把你的旧链用 LCEL 重构，享受「一次编写，多路调用」的丝滑体验！
点赞 + 收藏，下篇见！