LangChain 基于 Runnable 封装记忆链：从手写自动管理到新版生产实践

记忆链的本质并不复杂：模型调用前，把历史消息加载进 prompt；模型调用后，把本轮用户输入和模型输出保存回记忆系统。

如果把它拆成执行流程，就是两步：

• 调用前：load_memory_variables(input)，得到 history。
• 调用后：save_context(inputs, outputs)，保存本轮对话。

Runnable 的组合能力足够灵活，所以可以自己把这两步封装进 LCEL 链里。常见做法是：用 RunnablePassthrough.assign() 在链前面注入历史，用 with_listeners(on_end=...) 在链结束后保存上下文。

这篇文章就从这个手写封装讲起，但重点不是鼓励生产里重复造轮子，而是看清它背后的机制，以及为什么现在更推荐 RunnableWithMessageHistory 或 LangGraph checkpointer。

记忆链的运行流程

一个自动管理记忆的 Runnable 链，大概长这样：

可以把它理解成一条带前置读取和后置写入的链：

1. 用户传入当前问题。
2. 链从运行配置里找到对应的 memory。
3. 调用 memory 加载历史消息。
4. 把历史消息注入 prompt。
5. 调用模型生成回答。
6. 解析输出。
7. 在生命周期结束时保存本轮输入和输出。

用 LCEL 写出来，大概是：

from operator import itemgetter
from typing import Any

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnablePassthrough
from langchain_core.tracers.schemas import Run


def load_history(input: dict[str, Any], config: RunnableConfig) -> list:
    memory = config.get("configurable", {}).get("memory")
    if memory is None:
        return []
    variables = memory.load_memory_variables(input)
    return variables.get("history", [])


def save_history(run: Run, config: RunnableConfig) -> None:
    memory = config.get("configurable", {}).get("memory")
    if memory is None:
        return
    memory.save_context(run.inputs, run.outputs)


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个可靠的业务助手。"),
        ("placeholder", "{history}"),
        ("human", "{query}"),
    ]
)

chain = (
    RunnablePassthrough.assign(
        history=RunnableLambda(load_history)
    )
    | prompt
    | llm
    | StrOutputParser()
).with_listeners(on_end=save_history)

answer = chain.invoke(
    {"query": "我刚才说我的项目叫什么？"},
    config={"configurable": {"memory": memory}},
)

这个写法能工作，也很适合理解 Runnable 的机制：config["configurable"] 可以传运行时对象，RunnableLambda 可以读配置，with_listeners() 可以在结束时拿到 run 输入输出。

但它也有不少生产问题。

这个手写方案的核心机制

手写记忆链靠三个 Runnable 能力拼出来。

第一是 RunnableConfig.configurable。它把本次运行需要的记忆对象传进链：

config={
    "configurable": {
        "memory": memory,
    }
}

第二是 RunnablePassthrough.assign()。它保留原始输入，同时追加一个 history 字段：

RunnablePassthrough.assign(
    history=RunnableLambda(load_history)
)

输入：

{"query": "继续解释"}

会变成：

{
    "query": "继续解释",
    "history": [...],
}

第三是 with_listeners(on_end=...)。模型和 parser 执行完成后，监听器拿到 Run 对象，再把输入输出保存到 memory。

这就是“记忆自动管理”的来源：业务层只调用 chain.invoke()，读取和保存都包在链里。

为什么这不是当前首选

这个方案的问题不在于不能用，而在于边界太容易失控。

第一，configurable 里直接塞 memory 实例，会让配置变成对象传递通道。生产系统里，config 更适合传 session_id、thread_id、tenant_id 这类稳定标识，由服务端工厂函数查出对应的历史对象。

第二，on_end 依赖最终输出格式。如果链输出字符串、字典、AIMessage、结构化对象，save_context() 的处理逻辑都不一样。稍微换一个 parser，保存逻辑就可能坏。

第三，错误场景不好处理。模型成功但保存失败怎么办？保存成功但响应返回失败怎么办？重试后会不会重复保存？流式输出中途失败时要不要保存半截结果？

第四，并发和幂等要自己处理。同一个 session 的多次请求同时进来，历史顺序可能错乱；请求重试后可能重复写入。

第五，旧式 BaseMemory 抽象已经不是新项目的首选。现在更推荐基于消息历史、checkpointer 和 store 来管理状态。

所以，这个方案适合作为理解 Runnable 的练习，也适合非常小的原型。生产项目应该优先使用更明确的记忆封装。

当前推荐：RunnableWithMessageHistory

对于普通 LCEL 聊天链，当前更推荐 RunnableWithMessageHistory。它就是专门为“包装一个 Runnable，并自动读取和更新聊天历史”设计的。

典型写法：

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from pydantic import BaseModel, Field


class InMemoryHistory(BaseChatMessageHistory, BaseModel):
    messages: list[BaseMessage] = Field(default_factory=list)

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

    def clear(self) -> None:
        self.messages = []


store: dict[str, InMemoryHistory] = {}


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


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个可靠的业务助手。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ]
)

chain = prompt | llm

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

response = chain_with_history.invoke(
    {"question": "我叫小林"},
    config={"configurable": {"session_id": "thread_001"}},
)

第二次调用同一个 session_id：

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

它会自动把历史消息注入 history，并在调用后写回消息历史。

这比手写 assign + with_listeners 更清晰，因为它把输入消息字段、历史字段、输出消息字段都显式建模了。

多租户生产写法

生产里一般不会只靠一个 session_id。更常见的是用租户、账号、会话三类信息定位历史：

from langchain_core.runnables import ConfigurableFieldSpec


def get_session_history(
    *,
    tenant_id: str,
    actor_hash: str,
    thread_id: str,
) -> BaseChatMessageHistory:
    return history_repository.get_or_create(
        tenant_id=tenant_id,
        actor_hash=actor_hash,
        thread_id=thread_id,
    )


chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=get_session_history,
    input_messages_key="question",
    history_messages_key="history",
    history_factory_config=[
        ConfigurableFieldSpec(
            id="tenant_id",
            annotation=str,
            name="租户 ID",
            description="用于隔离不同租户的聊天历史",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="actor_hash",
            annotation=str,
            name="用户哈希",
            description="脱敏后的用户标识",
            default="",
            is_shared=True,
        ),
        ConfigurableFieldSpec(
            id="thread_id",
            annotation=str,
            name="会话 ID",
            description="一次连续对话的线程标识",
            default="",
            is_shared=True,
        ),
    ],
)

调用时：

response = chain_with_history.invoke(
    {"question": "继续刚才的方案"},
    config={
        "configurable": {
            "tenant_id": "tenant_a",
            "actor_hash": "usr_8f4a",
            "thread_id": "thread_001",
        }
    },
)

这比传 memory 对象更适合生产：

• 配置里是稳定标识，不是可变对象。
• 历史仓储由服务端统一管理。
• 可以做租户隔离。
• 可以做权限校验。
• 可以接 Redis、Postgres、MongoDB 等持久化。
• 可以统一加锁、去重、审计和过期策略。

持久化不要用内存

示例里的 InMemoryHistory 只适合开发和测试。生产环境至少要解决这些问题：

• 服务重启后历史不能丢。
• 多实例部署时历史要共享。
• 同一个 thread 的并发写入要有顺序。
• 历史要支持 TTL 和清理。
• 敏感内容要加密或脱敏。
• 删除历史要符合用户隐私要求。
• 历史写入失败要能重试或补偿。

可以把 BaseChatMessageHistory 接到自己的仓储层：

class PostgresChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, repository, tenant_id: str, thread_id: str):
        self.repository = repository
        self.tenant_id = tenant_id
        self.thread_id = thread_id

    @property
    def messages(self) -> list[BaseMessage]:
        return self.repository.list_messages(
            tenant_id=self.tenant_id,
            thread_id=self.thread_id,
        )

    def add_messages(self, messages: list[BaseMessage]) -> None:
        self.repository.append_messages(
            tenant_id=self.tenant_id,
            thread_id=self.thread_id,
            messages=messages,
        )

    def clear(self) -> None:
        self.repository.clear_messages(
            tenant_id=self.tenant_id,
            thread_id=self.thread_id,
        )

注意优先实现 add_messages()，而不是每条消息单独 add_message()。这样可以减少数据库往返，也更容易做事务。

Agent 场景：优先 checkpointer

如果是 Agent，不建议手写 RunnableWithMessageHistory 去包每一步。当前 LangChain Agent 的短期记忆更推荐交给 LangGraph checkpointer。

官方思路是：Agent 的短期记忆属于图状态，状态通过 checkpointer 按 thread 持久化。每次调用时传入 thread_id：

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


agent = create_agent(
    model="openai:gpt-5.5",
    tools=[get_user_info],
    checkpointer=InMemorySaver(),
)

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

agent.invoke(
    {"messages": [{"role": "user", "content": "我叫小林"}]},
    config,
)

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

这种方式更适合 Agent，因为 Agent 的状态不只有聊天历史，还可能包含：

• 工具调用过程。
• 中间推理步骤。
• 文件状态。
• 用户偏好。
• 子任务状态。
• 可恢复的执行快照。

手写 memory 链很难覆盖这些状态管理需求。

流式输出怎么处理

手写 on_end 保存上下文有一个隐藏问题：流式输出。

如果使用 stream() 或 astream()，模型输出可能是一段一段产生的。你要决定：

• 是等完整输出结束后保存。
• 还是边生成边缓存。
• 如果中途异常，是否保存部分输出。
• 用户取消请求时，是否保存。
• 输出被内容安全拦截时，是否保存。

RunnableWithMessageHistory 已经处理了常见消息输入输出场景，但复杂流式产品仍然建议在服务层维护一个响应缓冲区。只有当输出完整、通过安全检查、成功返回给用户后，再写入历史。

生产里不要把“模型生成了”直接等同于“用户看到了”。记忆应该记录用户实际经历过的对话。

上下文窗口管理

记忆不是把所有历史无限塞回 prompt。上下文窗口、成本和噪声都会限制历史长度。

常见策略包括：

• 只保留最近 N 轮。
• 按 token 数裁剪。
• 摘要旧消息。
• 删除无效消息。
• 对工具消息做压缩。
• 对长文档只存引用，不存全文。

Agent 场景可以使用 LangChain 的 summarization middleware，在接近 token 阈值时自动总结旧消息。普通 LCEL 链也可以在 get_session_history() 或 prompt 之前增加裁剪逻辑。

真正的生产记忆系统，重点不是“能保存”，而是“保存什么、召回什么、何时压缩、何时删除”。

什么时候还会手写封装

虽然新项目不建议把手写封装作为默认方案，但它仍然有价值。

适合手写的场景：

• 掌握 Runnable 的配置传递和生命周期。
• 做一个非常小的内部工具。
• 需要兼容已有 BaseMemory 抽象。
• 需要在某个非聊天 Runnable 前后读写业务状态。
• 需要临时验证某种记忆策略。

不适合手写的场景：

• 多租户聊天产品。
• 多实例部署。
• Agent 复杂状态。
• 需要流式输出可靠保存。
• 需要严格审计和隐私删除。
• 需要长期维护的生产系统。

手写封装可以帮助理解，但不要让它变成生产架构的地基。

问题

手写 Runnable 记忆链最主要的问题，是把太多职责塞进链里。

读取历史、格式化 prompt、调用模型、解析输出、保存上下文、处理错误、处理并发、处理流式、处理持久化，这些职责如果都散在 LCEL 表达式和监听器里，很快会变得难以测试和维护。

另外，with_listeners(on_end=...) 的保存时机也容易误导。on_end 只能说明 Runnable 正常结束，不一定说明响应已经安全地交付给用户。生产里还要考虑网关超时、用户取消、内容安全、保存失败和重复请求。

拓展

这个思路可以拓展到更通用的“状态链”：

• 调用前读取用户画像。
• 调用前读取权限上下文。
• 调用前读取实验配置。
• 调用后记录审计日志。
• 调用后更新任务状态。
• 调用后保存模型输出摘要。

但只要状态开始变复杂，就应该抽出状态管理层，而不是一直往 Runnable 监听器里塞逻辑。

更好的方向是：

• 简单聊天链：RunnableWithMessageHistory。
• Agent 短期记忆：LangGraph checkpointer。
• 长期记忆：LangGraph Store 或业务知识库。
• 观测和审计：callbacks / LangSmith / 日志平台。
• 自定义状态：明确的 repository 和 service 层。

实际生产是否使用

生产里不太建议直接使用“把 memory 实例放进 configurable，再用 on_end 保存”的手写方案。

生产里更常见的是：

• 用 RunnableWithMessageHistory 管普通 LCEL 聊天历史。
• 用 Redis/Postgres/MongoDB 等持久化 BaseChatMessageHistory。
• 用 LangGraph checkpointer 管 Agent thread 状态。
• 用 Store 或向量库管理跨会话长期记忆。
• 在服务层做会话权限、历史裁剪、摘要和审计。

手写方案可以保留在掌握、原型、兼容旧代码和少数特殊 Runnable 状态管理场景里。

现在是否抛弃

要分开看。

Runnable 组合、RunnablePassthrough.assign()、RunnableLambda、with_listeners() 都没有被抛弃，它们仍然是当前 Runnable 体系里的能力。

但是，用旧式 BaseMemory 加手写监听器来封装聊天记忆，已经不是新项目首选。现在更推荐 RunnableWithMessageHistory，而 Agent 场景更推荐 checkpointer。旧方案可以理解机制，也可以维护存量项目，但新项目不应该优先照搬。

最新生产如何实现

最新生产实现建议按场景选型。

普通 LCEL 聊天链：

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

response = chain_with_history.invoke(
    {"question": question},
    config={
        "configurable": {
            "tenant_id": tenant_id,
            "actor_hash": actor_hash,
            "thread_id": thread_id,
        }
    },
)

Agent 短期记忆：

agent = create_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer,
)

agent.invoke(
    {"messages": [{"role": "user", "content": question}]},
    {"configurable": {"thread_id": thread_id}},
)

长期记忆：

memory = store.search(
    namespace=("tenant", tenant_id, "profile"),
    query=question,
)

生产治理：

• 历史按租户和 thread 隔离。
• 写入使用幂等键。
• 持久化层支持事务或乐观锁。
• 输入输出默认脱敏。
• 历史有 TTL、删除和导出能力。
• 长对话有裁剪和摘要策略。
• trace 记录 thread_id，但不记录完整敏感内容。

最终原则是：Runnable 负责组合，MessageHistory 负责聊天历史，checkpointer 负责 Agent 状态，Store 负责长期记忆，业务服务层负责权限、持久化和治理。这样记忆自动管理才不会从一个方便的小封装，长成后期难以维护的隐性状态系统。