0. 代碼演示

from langchain_community.chat_message_histories import SQLChatMessageHistorydef get_session_history(session_id):# 通過 session_id 區分對話歷史，并存儲在 sqlite 數據庫中return SQLChatMessageHistory(session_id, "sqlite:///memory.db")

from langchain_core.messages import HumanMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParsermodel = ChatOpenAI(model="gpt-4o-mini", temperature=0)runnable = model | StrOutputParser()runnable_with_history = RunnableWithMessageHistory(runnable, # 指定 runnableget_session_history, # 指定自定義的歷史管理方法
)runnable_with_history.invoke([HumanMessage(content="你好，我叫麥酷")],config={"configurable": {"session_id": "wzr"}},
)'你好，麥酷！很高興再次見到你。有什么想聊的或者需要幫助的呢？'runnable_with_history.invoke([HumanMessage(content="你知道我叫什么名字")],config={"configurable": {"session_id": "wzr"}},
)'是的，你叫麥酷。有什么我可以幫助你的嗎？'runnable_with_history.invoke([HumanMessage(content="你知道我叫什么名字")],config={"configurable": {"session_id": "test"}},
)'抱歉，我無法知道你的名字。你可以告訴我你的名字，或者如果你有其他問題，我也很樂意幫助你！'

---

代碼功能解析

這段代碼實現了一個帶持久化歷史記憶的對話系統，通過 session_id 區分不同用戶的對話歷史，并存儲到 SQLite 數據庫中。以下是核心模塊的解析：

---

1. 對話歷史管理模塊

from langchain_community.chat_message_histories import SQLChatMessageHistorydef get_session_history(session_id):# 每個 session_id 對應獨立的數據庫記錄return SQLChatMessageHistory(session_id=session_id, connection_string="sqlite:///memory.db" # SQLite 數據庫路徑)

• 核心作用: 為每個用戶/會話創建獨立的歷史存儲
• 技術細節:
  • 使用 SQLite 數據庫存儲對話記錄（文件名為 memory.db）
  • session_id 作為主鍵區分不同對話（如用戶ID、設備ID等）
  • 實際表結構包含 id, session_id, message, timestamp 等字段
• 擴展性：可替換為其他存儲后端（如PostgreSQL、Redis）

---

2. 對話鏈構建模塊

from langchain_core.runnables.history import RunnableWithMessageHistorymodel = ChatOpenAI(model="gpt-4o-mini", temperature=0)
runnable = model | StrOutputParser()  # 基礎問答鏈runnable_with_history = RunnableWithMessageHistory(runnable=runnable,                # 原始鏈get_session_history=get_session_history, # 歷史管理方法input_messages_key="input",       # 輸入消息字段名（默認）history_messages_key="history"    # 歷史消息字段名（默認）
)

• 組件連接:
• 關鍵參數:
  • input_messages_key: 輸入消息在上下文中的鍵名
  • history_messages_key: 歷史消息的鍵名（模型需支持上下文窗口）

---

3. 對話調用示例

response = runnable_with_history.invoke([HumanMessage(content="你好，我叫麥酷")], # 當前消息config={"configurable": {"session_id": "wzr"}} # 指定會話
)

• 執行流程:
  1. 根據 session_id="wzr" 從數據庫加載歷史消息
  2. 將當前消息 "你好，我叫麥酷" 添加到歷史記錄
  3. 組合歷史消息 + 當前輸入 → 發送給 GPT-4
  4. 解析模型輸出 → 返回最終響應
  5. 將新消息對（用戶輸入 + 模型回復）保存到數據庫

---

4. 數據庫操作示例

假設進行三次連續對話：

調用順序	用戶輸入	數據庫存儲內容（session_id=“wzr”）
第一次	“你好，我叫麥酷”	[Human: 你好，我叫麥酷, AI: 回復1]
第二次	“記住我的名字了嗎？”	添加 [Human: 記住我的名字了嗎？, AI: 回復2]
第三次	“我是誰？”	添加 [Human: 我是誰？, AI: 回復3]

模型在第三次調用時，實際接收的上下文包含前兩次對話歷史，因此能正確回答姓名。

---

5. 關鍵技術點

5.1 歷史注入機制

# 偽代碼展示實際發送給模型的內容
full_context = [{"role": "user", "content": "你好，我叫麥酷"},{"role": "assistant", "content": "回復1"},{"role": "user", "content": "記住我的名字了嗎？"},{"role": "assistant", "content": "回復2"},{"role": "user", "content": "我是誰？"}
]
response = model.generate(full_context)

5.2 自動歷史管理

• 自動追加：每次調用自動添加新消息到歷史
• 上下文截斷：當歷史超過模型窗口時需處理（此示例未展示）

---

6. 優化建議

6.1 歷史長度控制

from langchain.memory import ConversationBufferWindowMemory# 僅保留最近3輪對話
memory = ConversationBufferWindowMemory(k=3)
runnable_with_history.memory = memory

6.2 自定義歷史格式

def custom_history_formatter(history):return "\n".join([f"{msg.type}: {msg.content}" for msg in history])chain = runnable_with_history.configure(history_formatter=custom_history_formatter
)

6.3 多模態歷史支持

from langchain_core.messages import ImageMessage# 支持圖片消息存儲
history.add_message(ImageMessage(content="path/to/image.png"))

---

7. 常見問題排查

現象	可能原因	解決方案
數據庫無寫入	文件權限問題	檢查 memory.db 可寫權限
歷史消息未生效	session_id 不一致	確認每次調用使用相同 session_id
響應時間越來越長	歷史消息過多未截斷	添加窗口記憶或摘要記憶
中文內容存儲亂碼	數據庫編碼問題	使用 sqlite:///memory.db?charset=utf8

---

8. 典型應用場景

1. 客服系統

   # 根據用戶手機號保持對話連續性
   session_id = user.phone_number
   

2. 教育機器人

   # 為每個學生保存學習進度
   session_id = f"{student_id}-{course_id}"
   

3. 多設備同步

   # 通過用戶賬戶實現跨設備同步
   session_id = user.account_id
   

---

該代碼展示了如何快速構建具備長期記憶能力的對話系統，通過簡潔的接口實現復雜的狀態管理，是開發智能對話應用的基石。

通過 LCEL，還可以實現

1. 配置運行時變量：https://python.langchain.com/v0.2/docs/how_to/configure/
2. 故障回退：https://python.langchain.com/v0.2/docs/how_to/fallbacks
3. 并行調用：https://python.langchain.com/v0.2/docs/how_to/parallel/
4. 邏輯分支：https://python.langchain.com/v0.2/docs/how_to/routing/
5. 動態創建 Chain: https://python.langchain.com/v0.2/docs/how_to/dynamic_chain/

更多例子：https://python.langchain.com/v0.2/docs/how_to/lcel_cheatsheet/