LangChain ChatMessageHistory 实践：聊天记录、短期记忆与生产持久化

聊天机器人要“记住上下文”，最朴素的做法就是保存消息历史。

用户说一句，系统保存一条 HumanMessage；模型回答一句，系统保存一条 AIMessage。下一轮请求时，把这些历史消息重新放回模型上下文，模型就能知道前面聊过什么。

这就是 ChatMessageHistory 这类组件的基本价值：它不是长期记忆系统，也不是向量检索，更不是 Agent 状态机。它负责管理一段会话里的消息列表，让对话链路能围绕标准的 Message 对象读写历史。

不过在当前 LangChain 里，需要先讲清楚边界：

• 如果你在用 create_agent 构建 Agent，短期记忆更推荐使用 checkpointer 和 thread_id，让 LangGraph state 持久化消息、工具结果和中间状态。
• 如果你在写 LCEL chain、普通聊天接口、轻量问答服务，ChatMessageHistory 和 RunnableWithMessageHistory 仍然很适合。
• 如果你要跨会话保存用户偏好、项目事实、长期画像，那已经不是 ChatMessageHistory 的职责，而是长期记忆模块的职责。

一句话：ChatMessageHistory 适合管理“当前会话的消息历史”，不适合承载完整记忆系统。

从类关系上看，BaseChatMessageHistory 定义了消息历史的基本协议；InMemoryChatMessageHistory 是 LangChain Core 中的内存实现；Redis、Postgres、File 等实现通常属于外部集成或社区扩展。无论底层存储是什么，核心能力都围绕读取消息列表、追加用户消息、追加 AI 消息、批量追加消息、清空历史和对应异步方法展开。

Message 是聊天记忆的最小单位

LangChain 的聊天上下文不是普通字符串，而是一组 message。

常见 message 类型包括：

• SystemMessage：系统规则和角色设定。
• HumanMessage：用户输入。
• AIMessage：模型回复。
• ToolMessage：工具调用结果。

示例：

from langchain.messages import AIMessage, HumanMessage, SystemMessage


messages = [
    SystemMessage("你是一个严谨的 Python 后端助手。"),
    HumanMessage("Flask 项目里为什么需要数据库迁移？"),
    AIMessage("因为模型变更和真实数据库 schema 之间需要可追踪的演进过程。"),
    HumanMessage("那上线前要检查什么？"),
]

模型看到的不是“拼接后的一大段文本”，而是带角色的消息序列。角色非常重要：系统规则、用户输入、模型历史回复、工具结果在模型侧的权重和语义都不同。

所以生产里不要把历史对话简单拼成：

history_text = "\n".join(history)

更好的方式是保留 message 结构，让模型明确知道每句话是谁说的。

InMemoryChatMessageHistory：最小可运行示例

InMemoryChatMessageHistory 是最简单的消息历史实现，适合本地开发和单进程测试。

from langchain_core.chat_history import InMemoryChatMessageHistory


history = InMemoryChatMessageHistory()

history.add_user_message("你好，我叫 Alice。")
history.add_ai_message("你好 Alice，很高兴认识你。")
history.add_user_message("我叫什么？")

for message in history.messages:
    print(type(message).__name__, message.content)

这段代码做了三件事：

• 把用户消息保存成 HumanMessage。
• 把模型消息保存成 AIMessage。
• 通过 history.messages 读取完整消息列表。

它非常适合理解组件接口，但不适合生产持久化。进程重启后，内存里的历史就没了；多实例部署时，不同实例之间也无法共享历史。

BaseChatMessageHistory 的核心协议

各种消息历史实现背后，都应该遵守同一套基本协议。

核心能力包括：

• 读取消息列表。
• 追加一条或多条消息。
• 清空当前会话历史。

可以把它理解成一个面向聊天消息的 repository。

from langchain_core.chat_history import BaseChatMessageHistory
from langchain.messages import BaseMessage


class CustomChatMessageHistory(BaseChatMessageHistory):
    @property
    def messages(self) -> list[BaseMessage]:
        ...

    def add_messages(self, messages: list[BaseMessage]) -> None:
        ...

    def clear(self) -> None:
        ...

生产里通常不会直接使用内存实现，而是根据 session_id 或 thread_id 从 Redis、Postgres、MongoDB 等存储中读写消息。

接口简单，但边界要想清楚：

• 一段历史属于哪个用户。
• 一段历史属于哪个租户。
• 一段历史属于哪个会话。
• 消息是否需要软删除。
• 消息是否需要脱敏。
• 消息是否有保留周期。
• 消息是否可以被用户导出或删除。

消息历史是用户数据，不只是技术缓存。

手动管理历史

最直接的方式，是手动读取历史、调用模型、再写回新消息。

from langchain.messages import HumanMessage
from langchain_openai import ChatOpenAI


model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
history = InMemoryChatMessageHistory()


def chat(user_input: str) -> str:
    messages = history.messages + [HumanMessage(user_input)]

    response = model.invoke(messages)

    history.add_user_message(user_input)
    history.add_ai_message(response.content)

    return response.content

这种方式最容易理解，但生产里会很快遇到问题：

• 每个接口都要手写读写历史。
• 容易漏写 AI 回复或用户输入。
• 失败时不知道该不该写入历史。
• 多个请求并发时可能顺序错乱。
• 很难统一接 trace、裁剪、摘要和持久化。

所以手动管理适合理解原理，不适合作为复杂项目的最终形态。

RunnableWithMessageHistory：给链路包一层历史

LCEL 链路里，更推荐使用 RunnableWithMessageHistory。

它做的事很直接：在调用链路前读取历史消息，把当前输入和历史一起送进 chain；调用结束后，把本轮用户输入和模型输出写回历史。

先定义 Prompt：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder


prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业、克制、准确的技术助手。"),
    MessagesPlaceholder("chat_history"),
    ("human", "{question}"),
])

再定义 chain：

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI


model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | model | StrOutputParser()

准备一个按 session_id 获取历史的函数：

from langchain_core.chat_history import InMemoryChatMessageHistory


store: dict[str, InMemoryChatMessageHistory] = {}


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

包装成带历史的链路：

from langchain_core.runnables.history import RunnableWithMessageHistory


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)

调用时传入 session_id：

answer = chain_with_history.invoke(
    {"question": "我叫 Alice。"},
    config={"configurable": {"session_id": "session_001"}},
)

answer = chain_with_history.invoke(
    {"question": "我叫什么？"},
    config={"configurable": {"session_id": "session_001"}},
)

这里最关键的是 configurable.session_id。它不是给模型看的业务输入，而是给历史管理层用来定位会话的运行配置。

数据流到底怎么走

RunnableWithMessageHistory 可以这样理解：

调用输入 + session_id
  -> 根据 session_id 读取 ChatMessageHistory
  -> 把 history.messages 放入 chat_history
  -> 调用原始 chain
  -> 把本轮 HumanMessage 和 AIMessage 追加回 history
  -> 返回 chain 输出

它并没有让模型产生真正长期记忆，只是自动维护当前会话的消息列表。

如果你发现模型能回答“我叫什么”，原因不是模型记住了用户，而是应用把上一轮消息重新传给了模型。

这点非常重要。很多记忆系统的设计错误，都来自把“上下文回放”误认为“模型记忆”。

Agent 短期记忆的新边界

如果你使用的是当前 LangChain 的 create_agent，短期记忆一般不再从 ChatMessageHistory 开始，而是用 checkpointer。

示例：

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver


agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "thread_001"}}

agent.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫 Alice。"}]},
    config,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "我叫什么？"}]},
    config,
)

thread_id 会把多轮交互归到同一个线程，checkpointer 负责保存 Agent state。这个 state 不只包含普通消息，还可能包含工具结果、中间步骤、自定义字段和检查点。

所以选型可以这样看：

场景	推荐
普通 LCEL 对话链	RunnableWithMessageHistory
本地理解消息历史	InMemoryChatMessageHistory
Agent 短期记忆	create_agent + checkpointer
跨会话偏好和事实	长期记忆 store
复杂状态恢复和分支	LangGraph checkpoint

不要把 ChatMessageHistory 当成所有记忆问题的答案。它只是会话消息历史这一层。

生产持久化怎么做

内存历史不适合生产。一个更接近生产的实现，至少要存这些字段：

CREATE TABLE chat_messages (
    id BIGSERIAL PRIMARY KEY,
    tenant_id VARCHAR(64) NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    session_id VARCHAR(128) NOT NULL,
    role VARCHAR(32) NOT NULL,
    content TEXT NOT NULL,
    metadata JSONB NOT NULL DEFAULT '{}',
    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
    deleted_at TIMESTAMP NULL
);

CREATE INDEX idx_chat_messages_session
ON chat_messages (tenant_id, session_id, created_at);

关键点：

• tenant_id 用来做租户隔离。
• user_id 用来做权限校验和查询。
• session_id 用来定位会话历史。
• role 对应 human、ai、system、tool 等消息类型。
• metadata 保存 message id、token、tool call、模型名等附加信息。
• deleted_at 支持软删除。

真实系统里，还要考虑加密、归档、保留周期和用户删除权。

一个 PostgresChatMessageHistory 的骨架

可以把持久化历史封装成一个类，让业务链路只依赖 BaseChatMessageHistory 协议。

from langchain_core.chat_history import BaseChatMessageHistory
from langchain.messages import AIMessage, BaseMessage, HumanMessage


class PostgresChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, *, pool, tenant_id: str, session_id: str):
        self.pool = pool
        self.tenant_id = tenant_id
        self.session_id = session_id

    @property
    def messages(self) -> list[BaseMessage]:
        rows = self._load_rows()

        result: list[BaseMessage] = []
        for row in rows:
            if row["role"] == "human":
                result.append(HumanMessage(row["content"]))
            elif row["role"] == "ai":
                result.append(AIMessage(row["content"]))

        return result

    def add_messages(self, messages: list[BaseMessage]) -> None:
        rows = []

        for message in messages:
            if isinstance(message, HumanMessage):
                role = "human"
            elif isinstance(message, AIMessage):
                role = "ai"
            else:
                role = message.type

            rows.append({
                "tenant_id": self.tenant_id,
                "session_id": self.session_id,
                "role": role,
                "content": message.content,
            })

        self._insert_rows(rows)

    def clear(self) -> None:
        self._soft_delete_session()

    def _load_rows(self):
        ...

    def _insert_rows(self, rows: list[dict]) -> None:
        ...

    def _soft_delete_session(self) -> None:
        ...

这里省略了数据库细节，重点是边界：chain 不关心你用什么数据库，chain 只关心能不能读写消息历史。

生产实现还要补：

• 批量写入。
• 事务。
• 连接池。
• 重试。
• 消息排序。
• 幂等 message id。
• 内容脱敏。
• 查询权限。

会话 ID 不能随便传

session_id 是消息历史隔离的核心。

如果用户可以随便指定 session_id，就可能读到别人的历史。生产里不要直接信任前端传来的会话 ID。

更安全的做法：

• session 属于当前登录用户。
• 查询历史时同时校验 tenant_id 和 user_id。
• session id 使用不可猜测 ID。
• 服务端根据认证信息绑定会话。
• 对跨租户访问做硬隔离。

示例：

def get_session_history(session_id: str) -> PostgresChatMessageHistory:
    session = session_repo.get(
        tenant_id=current_tenant_id(),
        user_id=current_user_id(),
        session_id=session_id,
    )

    if session is None:
        raise PermissionError("session not found")

    return PostgresChatMessageHistory(
        pool=db_pool,
        tenant_id=session.tenant_id,
        session_id=session.session_id,
    )

消息历史看起来只是聊天记录，本质上是用户数据。权限边界不能省。

控制历史长度

历史消息会越积越多。即使数据库能存，模型上下文也不能无限接收。

常见策略：

• 只取最近 N 条。
• 按 token 预算裁剪。
• 删除低价值消息。
• 把较早历史压缩成摘要。
• 工具结果只保留摘要。
• 大文件和图片只保留引用。

简单裁剪：

def trim_recent_messages(messages, max_messages: int = 20):
    return messages[-max_messages:]

更好的方式是按 token 控制，而不是按条数。因为一条工具结果可能比十轮普通聊天还长。

在 Agent 场景里，可以使用 summarization middleware 或 LangGraph state 里的裁剪策略；在普通 LCEL 链路里，可以在 get_session_history 或 Prompt 前加一层裁剪。

不要把所有历史都写入模型上下文

数据库里保存完整历史，和模型调用时使用完整历史，是两回事。

生产里建议分两层：

• 存储层：尽量保留完整、可审计的消息历史。
• 上下文层：只选择当前模型调用需要的历史。

例如：

def select_context_messages(history, max_messages: int = 12):
    messages = history.messages

    return messages[-max_messages:]

如果历史很长，可以用摘要：

context_messages = [
    SystemMessage(f"较早对话摘要：{conversation_summary}"),
    *recent_messages,
]

注意摘要本身也要有版本和来源。摘要错误会影响后续所有回答。

与长期记忆的区别

ChatMessageHistory 保存的是原始对话过程，长期记忆保存的是抽取后的稳定信息。

例如用户说：

以后代码示例默认用 Python。

消息历史会保存这句话本身。长期记忆则可以保存成：

{
  "type": "preference",
  "key": "code_language",
  "value": "python",
  "source": "user_explicit"
}

两者用途不同：

能力	ChatMessageHistory	长期记忆
保存内容	原始消息	抽取后的事实、偏好、事件
作用范围	单会话为主	跨会话
查询方式	按 session 顺序读取	按 key 或语义召回
更新方式	追加消息	upsert、合并、删除
风险	上下文过长	错误记忆长期污染

不要用聊天历史表直接当长期记忆表。聊天历史是原始材料，长期记忆是经过筛选和治理的结果。

与 RAG 的区别

RAG 检索的是外部知识，ChatMessageHistory 管理的是对话历史。

两者经常一起出现：

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是企业知识库助手，只根据上下文回答。"),
    MessagesPlaceholder("chat_history"),
    ("human", "知识库上下文：\n{context}\n\n问题：{question}"),
])

但不要混淆：

• chat_history 解释用户当前问题的上下文。
• context 提供回答问题所需的知识材料。

如果把历史聊天当知识库，模型很容易引用用户随口说过的错误信息；如果把知识库检索结果当聊天历史，模型又会误解角色和时间顺序。

失败和一致性

一个常见问题：模型调用成功了，但写历史失败怎么办？

或者反过来：用户消息写入成功了，模型调用失败了怎么办？

生产里要定义清楚策略。

常见做法：

1. 先写用户消息，状态为 pending。
2. 调用模型。
3. 模型成功后写 AI 消息。
4. 更新用户消息状态为 completed。
5. 模型失败时记录错误消息或失败状态。

这样可以保留用户确实发起过请求的事实，也能区分模型是否成功回复。

简化状态：

ALTER TABLE chat_messages
ADD COLUMN status VARCHAR(32) NOT NULL DEFAULT 'completed';

状态可以包括：

• pending
• completed
• failed
• deleted

不要让失败请求在历史里完全消失。否则用户刷新后会看到“刚才那句话没了”，排障也找不到记录。

隐私和删除

聊天历史通常包含大量敏感内容。

生产里至少要支持：

• 用户删除单条消息。
• 用户清空某个会话。
• 用户删除所有历史。
• 管理端按权限查看。
• 敏感字段脱敏展示。
• 数据保留周期。
• 审计日志。

不要把历史消息永久保存在一个没人管理的表里。记忆能力越强，越需要删除能力。

对于高敏场景，可以考虑：

• 默认不保存原文。
• 只保存脱敏摘要。
• 会话结束后自动过期。
• 高敏租户单独关闭历史。
• 将原文加密存储。

可观测性

消息历史问题很难靠肉眼猜。

建议记录这些指标：

• 每个 session 的消息数。
• 每次注入模型的历史消息数。
• 历史 token 占比。
• 被裁剪消息数量。
• 摘要触发次数。
• 写历史失败率。
• 历史读取延迟。
• 会话并发冲突次数。
• 用户清空历史次数。

同时在 trace metadata 里带上：

• session_id
• thread_id
• history_message_count
• prompt_version
• history_strategy

这样当用户反馈“模型不记得前面说的话”时，你能快速确认是历史没写入、session 错了、裁剪掉了，还是模型没有正确使用历史。

项目目录建议

可以把聊天历史封装到独立模块里：

app/
  ai/
    histories/
      base.py
      postgres.py
      redis.py
    chains/
      chat.py
    prompts/
      chat.py
  services/
    chat_service.py

链路构建：

# app/ai/chains/chat.py
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from app.ai.histories.postgres import get_session_history
from app.ai.models import get_default_chat_model


prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业、克制、准确的技术助手。"),
    MessagesPlaceholder("chat_history"),
    ("human", "{question}"),
])


def build_chat_chain():
    chain = prompt | get_default_chat_model() | StrOutputParser()

    return RunnableWithMessageHistory(
        chain,
        get_session_history,
        input_messages_key="question",
        history_messages_key="chat_history",
    )

服务层：

# app/services/chat_service.py
from app.ai.chains.chat import build_chat_chain


def answer(question: str, session_id: str) -> str:
    chain = build_chat_chain()

    return chain.invoke(
        {"question": question},
        config={
            "configurable": {
                "session_id": session_id,
            },
            "metadata": {
                "session_id": session_id,
            },
        },
    )

route 层不要直接操作 message history。它只应该拿到请求，交给 service。

测试建议

先测 history 行为：

def test_in_memory_history_adds_messages():
    history = InMemoryChatMessageHistory()

    history.add_user_message("你好")
    history.add_ai_message("你好，有什么可以帮你？")

    assert len(history.messages) == 2
    assert history.messages[0].content == "你好"

再测 session 隔离：

def test_session_history_is_isolated():
    h1 = get_session_history("s1")
    h2 = get_session_history("s2")

    h1.add_user_message("我是 Alice")
    h2.add_user_message("我是 Bob")

    assert h1.messages[0].content == "我是 Alice"
    assert h2.messages[0].content == "我是 Bob"

最后测带历史的链路：

def test_chain_uses_same_session_history(fake_model):
    chain = build_chat_chain(fake_model=fake_model)

    chain.invoke(
        {"question": "我叫 Alice"},
        config={"configurable": {"session_id": "s1"}},
    )

    chain.invoke(
        {"question": "我叫什么？"},
        config={"configurable": {"session_id": "s1"}},
    )

    history = get_session_history("s1")

    assert len(history.messages) >= 4

测试重点不是让真实模型回答正确，而是验证历史读写、session 隔离、失败处理和裁剪策略。

常见坑

用全局 dict 当生产存储

全局 dict 只适合本地测试。多进程、多实例、重启、扩容都会让历史丢失或不一致。

没有 session 隔离

所有用户共用一个 history，是最危险的错误。轻则上下文串线，重则泄露用户数据。

历史无限增长

上下文越长越好是错觉。长历史会增加成本、延迟和干扰。必须裁剪或摘要。

把历史当长期记忆

聊天历史是原始记录，不是稳定事实。长期偏好要抽取、确认、去重和更新。

忽略工具消息

Agent 或 tool calling 场景里，历史不只有 human / ai。工具调用和工具结果也可能影响下一步模型决策。不要把工具消息简单丢掉。

没有删除能力

用户要求清空历史时，系统必须能删。记忆系统没有删除能力，就不适合生产。

生产 Checklist

上线前至少检查这些点：

• 是否按 tenant、user、session 隔离。
• session id 是否不可猜测。
• 是否校验当前用户有权访问该 session。
• 是否支持持久化而不是只用内存。
• 是否限制注入模型的历史长度。
• 是否支持摘要或裁剪。
• 是否记录写历史失败。
• 是否支持软删除和清空会话。
• 是否有数据保留周期。
• 是否避免把密钥和敏感信息写入历史。
• 是否能在 trace 中看到 session 和历史策略。

这些做好以后，ChatMessageHistory 才不只是 demo 里的消息数组，而是可以支撑真实聊天产品的会话记忆层。

小结

ChatMessageHistory 的核心价值，是把聊天记录管理成标准 message 列表，让普通 LCEL 对话链路具备会话内短期记忆。

它适合保存和回放当前会话的消息历史，配合 RunnableWithMessageHistory 可以减少手写读写历史的重复代码。但在当前 LangChain 体系里，Agent 短期记忆更适合使用 checkpointer 和 thread_id，长期偏好和事实则应该进入独立的长期记忆 store。

把边界分清楚，系统就会更稳：消息历史负责会话连续性，Agent state 负责执行状态，长期记忆负责跨会话知识。三者各司其职，聊天机器人才能既记得住，又不乱记。

问题

ChatMessageHistory 最大的问题，是它太容易被误解成“完整记忆系统”。

它只保存消息，不负责判断哪些消息应该进入模型上下文，不负责摘要，不负责长期偏好抽取，也不负责 Agent 状态恢复。生产里如果直接把历史消息无限追加，再每轮全部塞进 Prompt，会很快遇到几个问题：

• 历史无限增长，token 成本和延迟持续上升。
• 多用户、多租户、多会话隔离如果没做好，会发生上下文串线。
• 工具消息、失败消息、删除消息如果处理不一致，历史会变得不可复现。
• 用户隐私、密钥、日志、内部链接可能被长期保存。
• 聊天历史被误当成长期事实，导致过期信息持续污染回答。

所以 ChatMessageHistory 的风险不在 API，而在边界。它适合作为消息历史层，不适合作为记忆中枢。

拓展

ChatMessageHistory 可以往三个方向拓展。

第一，接入持久化存储。用 Postgres、Redis、MongoDB 或对象存储保存消息，并按 tenant_id、user_id、session_id 做隔离。

第二，接入上下文治理。在读取消息后增加裁剪、摘要、脱敏、工具消息修复、token 预算计算，只把本轮真正需要的消息交给模型。

第三，接入长期记忆抽取。聊天历史是原始材料，用户偏好、项目事实、稳定画像应该从历史里抽取出来，写入 Store 或业务数据库，而不是继续堆在 message list 里。

一个更完整的架构是：

ChatMessageHistory 保存原始消息
  -> Context selector 选择本轮上下文
  -> Summary / trim 控制 token
  -> Long-term memory extractor 抽取稳定事实
  -> Store 保存跨会话记忆

实际生产是否使用

会使用，但通常不会单独裸用。

在普通 LCEL 聊天链里，ChatMessageHistory + RunnableWithMessageHistory 仍然是很实用的组合，适合 FAQ 助手、普通客服、轻量聊天接口。生产实现会把 InMemoryChatMessageHistory 换成 Redis/Postgres/MongoDB 这类持久化版本，并增加权限、审计、删除和裁剪。

在 Agent 场景里，生产更常用 checkpointer 管理 AgentState["messages"]。这时消息历史仍然存在，但不一定通过 ChatMessageHistory 暴露，而是作为 LangGraph state 的一部分保存。

现在是否抛弃

没有抛弃。

BaseChatMessageHistory、InMemoryChatMessageHistory 和 RunnableWithMessageHistory 仍然是当前 LangChain 里的有效抽象，适合普通 chain 和自定义消息历史管理。

但它的定位变窄了：新 Agent 应用不再把 ChatMessageHistory 当成短期记忆主入口，而是优先使用 create_agent + checkpointer + thread_id。也就是说，ChatMessageHistory 没过时，过时的是把它当成全部记忆系统。

最新生产如何实现

当前生产实现可以按场景选。

普通 LCEL 链：

from langchain_core.runnables.history import RunnableWithMessageHistory


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)

answer = chain_with_history.invoke(
    {"question": "我刚才说我叫什么？"},
    config={"configurable": {"session_id": "tenant-1:user-42:chat-1001"}},
)

生产里的 get_session_history 不应该返回内存对象，而应该返回基于数据库的 BaseChatMessageHistory 实现，并在内部校验当前用户是否有权访问这个 session。

Agent 链路：

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver


with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()

    agent = create_agent(
        model="gpt-5.5",
        tools=[],
        checkpointer=checkpointer,
    )

    result = agent.invoke(
        {"messages": [{"role": "user", "content": "我叫 Alice。"}]},
        {"configurable": {"thread_id": "tenant-1:user-42:chat-1001"}},
    )

这个方案更适合复杂 Agent，因为 checkpoint 保存的不只是消息，还包括工具调用结果、中间状态和自定义 state 字段。

参考：

• LangChain Short-term memory (opens new window)
• LangChain Long-term memory (opens new window)