LangChain记忆与对话历史

大语言模型本身是无状态的，每次请求都是独立的，它不会自动记住上一轮说了什么。要实现像 ChatGPT 那样的多轮对话，需要我们手动管理对话历史。

1. 手动管理历史（最基础）

最直接的方式：自己维护一个消息列表，每轮对话后追加消息：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")

history = [
    SystemMessage(content="你是一位友善的助手，记住用户告诉你的信息。"),
]

def chat(user_input: str) -> str:
    history.append(HumanMessage(content=user_input))
    response = llm.invoke(history)
    history.append(AIMessage(content=response.content))
    return response.content

print(chat("我叫小明，今年 22 岁。"))
print(chat("我喜欢 Python 编程。"))
print(chat("你知道我叫什么吗？我喜欢什么？"))

这是最透明的实现方式，便于理解原理。缺点是历史消息会越积越多，最终超出模型的上下文限制。

2. 使用 ChatMessageHistory

LangChain 提供了 ChatMessageHistory 类来封装历史管理：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")
history = InMemoryChatMessageHistory()

system_message = SystemMessage(content="你是一位友善的助手。")

def chat(user_input: str) -> str:
    history.add_user_message(user_input)
    messages = [system_message] + history.messages
    response = llm.invoke(messages)
    history.add_ai_message(response.content)
    return response.content

print(chat("我叫小红。"))
print(chat("我的爱好是画画。"))
print(chat("能介绍一下我自己吗？"))

3. RunnableWithMessageHistory（推荐用法）

RunnableWithMessageHistory 是 LangChain 推荐的方式，它自动将历史管理集成到链中，并支持多用户/多会话：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位友善的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

chain = prompt | llm | parser

# 使用字典存储多个会话的历史
sessions = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in sessions:
        sessions[session_id] = InMemoryChatMessageHistory()
    return sessions[session_id]

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 用 session_id 区分不同会话
config_alice = {"configurable": {"session_id": "alice"}}
config_bob = {"configurable": {"session_id": "bob"}}

# Alice 的对话
print(chain_with_history.invoke({"input": "我叫 Alice"}, config=config_alice))
print(chain_with_history.invoke({"input": "你记得我叫什么？"}, config=config_alice))

# Bob 的独立对话（不会看到 Alice 的历史）
print(chain_with_history.invoke({"input": "我叫 Bob"}, config=config_bob))
print(chain_with_history.invoke({"input": "你记得我叫什么？"}, config=config_bob))

每个 session_id 对应独立的历史记录，非常适合多用户场景。

4. 限制历史长度

随着对话增多，历史消息会越来越长，消耗大量 token。可以只保留最近几轮：

from langchain_core.chat_history import InMemoryChatMessageHistory

class TrimmedChatHistory(InMemoryChatMessageHistory):
    """只保留最近 N 轮（2N 条消息）的历史"""

    def __init__(self, max_rounds: int = 5):
        super().__init__()
        self.max_rounds = max_rounds

    def add_user_message(self, message: str):
        super().add_user_message(message)
        self._trim()

    def _trim(self):
        # 每轮 2 条消息（human + ai），保留最近 max_rounds 轮
        max_messages = self.max_rounds * 2
        if len(self.messages) > max_messages:
            self.messages = self.messages[-max_messages:]

或者使用 LangChain 内置的 trim_messages 工具：

from langchain_core.messages import trim_messages

trimmed = trim_messages(
    history.messages,
    max_tokens=1000,                # 最多保留 1000 token
    strategy="last",                # 保留最新的消息
    token_counter=ChatOpenAI(model="gpt-4o-mini"),
    include_system=True,            # 始终保留 system 消息
)

5. 持久化存储历史

InMemoryChatMessageHistory 保存在内存里，程序重启就消失了。如果需要持久化，可以使用数据库后端：

pip install langchain-community

使用 SQLite 存储：

from langchain_community.chat_message_histories import SQLChatMessageHistory

def get_session_history(session_id: str):
    return SQLChatMessageHistory(
        session_id=session_id,
        connection="sqlite:///chat_history.db",
    )

使用 Redis 存储：

pip install redis

from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_session_history(session_id: str):
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379",
    )

只需替换 get_session_history 函数，其他代码不需要改变。

总结

模型本身无状态，需要手动将历史消息传入每次请求
手动管理历史最简单直观，但需要自己追加消息
RunnableWithMessageHistory 是推荐方式，通过 session_id 支持多会话
历史消息越积越多会消耗 token，需要用截断策略控制长度
替换 get_session_history 函数可以切换 SQLite、Redis 等持久化存储