Memory 对话历史与流式 / 异步 / 批处理

记忆不是模型的属性，而是 调用前后的一次读写。LLM 本身无状态：每次请求都是独立的。所谓「记住对话」，就是在调用前把历史消息拼进 prompt，调用后把这一轮的新问答追加回某个存储。RunnableWithMessageHistory 做的就是自动化这套 read-then-append 流程，而不是给模型注入魔法。

一、Memory：用 RunnableWithMessageHistory 管理多轮对话

现代 LangChain 的记忆方案由三个部件拼成：① 一个带 MessagesPlaceholder 的 prompt 模板，用来给历史留出注入位置；② 一个 ChatMessageHistory 存储（内存版用 InMemoryChatMessageHistory，生产用 Redis / SQL 等后端）；③ 用 RunnableWithMessageHistory 把链和「按 session_id 取历史」的工厂函数绑在一起。

# 安装核心包 + OpenAI 适配（也可换成 langchain-anthropic 等）
pip install -U langchain langchain-core langchain-openai

# 配置 API key（任选其一）
export OPENAI_API_KEY="sk-..."

import os
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

# 1) prompt 里用 MessagesPlaceholder("history") 给历史留出注入位置
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个简洁友好的助手，用中文回答。"),
    MessagesPlaceholder(variable_name="history"),  # 历史消息会被展开到这里
    ("human", "{input}"),
])

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | model  # 一条普通 LCEL 链，本身无记忆

# 2) 用一个 dict 做最简存储：session_id -> 历史对象
store = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    """工厂函数：按 session_id 返回（或新建）历史存储。
    生产环境把这里换成 Redis/SQL 后端即可，签名不变。"""
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 3) 用 RunnableWithMessageHistory 包裹链
chat = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",    # 链输入里哪个 key 是本轮用户消息
    history_messages_key="history",  # 注入到 prompt 哪个 placeholder
)

# 4) 调用时通过 config 传 session_id（这是记忆隔离的唯一键）
cfg = {"configurable": {"session_id": "user-001"}}
print(chat.invoke({"input": "我叫小明，记住了吗？"}, config=cfg).content)
print(chat.invoke({"input": "我叫什么名字？"}, config=cfg).content)  # 应答出「小明」

# 换一个 session_id，历史互相隔离
cfg2 = {"configurable": {"session_id": "user-002"}}
print(chat.invoke({"input": "我叫什么名字？"}, config=cfg2).content)  # 不知道，因为是新会话

从旧版 ConversationBufferMemory 迁移

LangChain 0.3+ 已将 ConversationBufferMemory、ConversationChain 等有状态记忆类标记为 deprecated。它们的问题在于：记忆状态藏在对象内部，难以序列化、难以按用户隔离、和 LCEL 的无状态组合范式冲突。迁移核心是把「藏在对象里的 buffer」改成「显式按 session_id 存取的 history 工厂」。

二、执行模式：stream / astream / ainvoke / batch

LCEL 的 Runnable 接口天生提供 统一的多模态执行 API：同一条链，不改一行组合逻辑，就能 invoke（同步整段）、stream（同步逐块）、ainvoke（异步整段）、astream（异步逐块）、batch（批量并行）。选哪个只取决于运行环境（同步/异步）和交付形态（要不要边出边显示、要不要批量），与链本身无关。

stream：同步逐 token 流式输出

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("用三句话讲讲 {topic}")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
chain = prompt | model | StrOutputParser()  # StrOutputParser 让流式直接吐字符串

# stream 返回同步生成器，每个 chunk 是新增的一小段文本
for chunk in chain.stream({"topic": "LangChain 的记忆机制"}):
    print(chunk, end="", flush=True)  # 打字机效果，end="" 不换行，flush 立即刷出
print()

ainvoke + astream：异步调用与异步流

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("一句话解释 {topic}")
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

async def main():
    # 1) ainvoke：异步整段调用，await 拿完整结果
    result = await chain.ainvoke({"topic": "协程"})
    print("[ainvoke]", result)

    # 2) astream：异步逐 token，配合 async for 消费
    print("[astream] ", end="")
    async for chunk in chain.astream({"topic": "事件循环"}):
        print(chunk, end="", flush=True)
    print()

# 在 FastAPI 等异步框架里直接 await；脚本里用 asyncio.run 驱动
asyncio.run(main())

batch：批处理与 max_concurrency 并发控制

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("给「{word}」写一个英文同义词，只输出单词")
chain = prompt | ChatOpenAI(model="gpt-4o-mini", temperature=0) | StrOutputParser()

inputs = [{"word": w} for w in ["快乐", "悲伤", "愤怒", "惊讶", "恐惧", "信任"]]

# batch 默认并行执行所有输入（不是 for 循环 invoke），返回顺序与输入一致的结果列表
# config 里的 max_concurrency 限制同时在飞的请求数，防止打爆 API 限速
results = chain.batch(inputs, config={"max_concurrency": 3})
for item, out in zip(inputs, results):
    print(item["word"], "->", out)

# 异步批处理：abatch，签名一致，用在异步环境
# results = await chain.abatch(inputs, config={"max_concurrency": 3})

记忆不是模型记住了什么，而是你在每次调用前后，决定把什么读进 prompt、把什么写回存储。

— LangChain 实战心法

下一章进入 可观测性与上线：用 Callbacks 追踪链内部每一步、定位幻觉与性能瓶颈，盘点高频踩坑，并用 LangServe 把链一键变成带 OpenAPI 文档的生产级 HTTP 服务。