LangChain 消息处理：缓存、过滤、合并与流式输出实战

从 LangChain 的 v0.3 版本开始，官方建议 LangChain 用户不要使用 RunnableWithMessageHistory，而是利用 LangGraph 持久性来完成。

二、消息过滤

核心概念

使用 filter_messages 函数对消息列表进行筛选，按类型或ID过滤消息。

关键函数

from langchain_core.messages import filter_messages 

过滤参数

代码示例

messages = [
    HumanMessage(content="你好，我是小明", id="1"),
    AIMessage(content="你好，小明！很高兴认识你！", id="2"),
    HumanMessage(content="我想知道我之前的名字", id="3"),
    AIMessage(content="你之前的名字是小绿！", id="4"),
]
# 过滤：只保留 AI 消息，且排除 id=4 的消息
filtered_messages = filter_messages(
    messages,
    include_types=["ai"],
    exclude_ids=["4"],
)
# → 结果只剩 id=2 的 AIMessage: "你好，小明！很高兴认识你！"

过滤逻辑

原始消息 → include_types=["ai"] 筛掉 Human 消息 → exclude_ids=["4"] 再排除 id=4 → 最终结果

原始：[Human#1, AI#2, Human#3, AI#4]
↓ include_types=["ai"]
中间：[AI#2, AI#4]
↓ exclude_ids=["4"]
结果：[AI#2]

三、消息合并

核心概念

使用 merge_message_runs 将连续的同类型消息合并为一条，避免多条连续 Human 或 AI 消息导致模型报错或行为异常。

关键函数

from langchain_core.messages import merge_message_runs 

代码示例

messages = [
    HumanMessage(content="你好", id="1"),
    HumanMessage(content="我是小明", id="2"), # 连续两条 Human
    AIMessage(content="你好，小明！", id="3"),
    AIMessage(content="很高兴认识你！", id="4"), # 连续两条 AI
]
merged_messages = merge_message_runs(messages)

合并效果

合并前（4 条）: human: 你好
human: 我是小明
ai: 你好，小明！
ai: 很高兴认识你！
合并后（2 条）: human: 你好\n我是小明
ai: 你好，小明！\n很高兴认识你！

两种使用方式

# 方式一：直接调用函数合并后传给模型
merged_messages = merge_message_runs(messages)
model.invoke(merged_messages)

# 方式二：通过管道（pipe）操作，合并与模型调用串联
chain = merge_message_runs | model
response = chain.invoke(messages)

管道方式更简洁，适合在 LangChain 链式调用中使用。

四、流式输出

什么是流式输出

流式输出（Streaming） 是指 AI 模型逐字返回内容，而不是等待全部生成完毕后一次性返回。就像 ChatGPT 那样，文字一个个"打"出来，而不是突然全部出现。

为什么需要？

AI 生成长文本可能需要几秒甚至更长时间。传统方式用户需要等待整个响应完成才能看到内容，体验很差。流式输出实时展示生成过程，让用户感觉响应更快，交互更自然。

典型应用

1. 聊天机器人：像 ChatGPT 一样逐字显示
2. 文章生成：实时展示生成过程
3. 代码生成：逐行显示代码
4. 实时翻译：边翻译边显示

五、同步 vs 异步流式

LangChain 提供两种流式方式：同步（stream）和异步（astream）。

核心区别

工作原理

同步流式： 阻塞当前线程，处理一个请求时无法处理其他请求。就像排队买咖啡，必须等前一个人买完。

异步流式： 使用协程机制，等待 AI 响应时可以切换到其他任务。就像服务员可以同时为多桌客人点单。

何时使用异步？

推荐：

• 多用户 Web 应用
• 高并发聊天机器人
• 与其他异步操作结合

不需要：

• 简单的单次调用
• 学习测试阶段

六、流式输出基础用法

同步流式

from langchain_deepseek import ChatDeepSeek
from langchain_core.output_parsers import StrOutputParser

model = ChatDeepSeek(model="deepseek-chat", streaming=True)
parser = StrOutputParser()
chain = model | parser

for chunk in chain.stream("写一个关于程序员的笑话"):
    print(chunk, end="|", flush=True)

关键点：

• streaming=True：必须设置
• flush=True：立即刷新输出

异步流式

import asyncio

async def main():
    chain = model | parser
    async for chunk in chain.astream("写一个关于程序员的笑话"):
        print(chunk, end="|", flush=True)

if __name__ == "__main__":
    asyncio.run(main())

关键点：

• async def：定义异步函数
• async for：异步迭代
• asyncio.run()：运行入口

七、输出解析器

StrOutputParser 是最常用的解析器，将模型输出转换为纯文本。

作用：

• 提取文本内容
• 去除多余格式
• 统一输出格式

自定义解析器：

def custom_parser(output: str) -> str:
    return output.strip().replace("。", "！")

chain = model | parser | custom_parser 

应用场景：

• 格式转换（Markdown → HTML）
• 内容过滤审核
• 特殊字符处理

八、流式输出实际应用

1. 聊天机器人

用户发送消息后，AI 回复逐字显示，像真人打字。使用异步流式提高响应速度。

2. 多用户并发

Web 应用中多个用户同时请求，异步流式可以并发处理。

性能对比：

• 同步：3 个请求需要 15 秒（串行）
• 异步：3 个请求只需 5 秒（并发）

3. FastAPI 集成

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.get("/chat")
async def chat_stream(question: str):
    async def generate():
        async for chunk in chain.astream(question):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

九、常见问题

1. 没有流式效果？

原因： 忘记 streaming=True 或 flush=True

2. async for 报错？

原因： 使用了 ainvoke() 而不是 astream()

ainvoke() 返回完整结果，astream() 返回流式迭代器。

3. 性能对比

• 单个请求：同步和异步相近
• 多个并发：异步快 3 倍

十、总结对比

典型应用场景

• 内存缓存：多轮对话场景，用户问"我之前说了什么"时模型能回答
• 消息过滤：只提取 AI 回复做摘要、排除某些敏感消息
• 消息合并：用户连续发了多条消息时，合并后再发给模型，避免格式错误
• 流式输出：聊天机器人逐字显示、长文本生成、FastAPI 接口集成

流式输出要点

1. 流式输出 = 实时返回，提升体验
2. 同步 = 简单，适合学习
3. 异步 = 高性能，适合生产
4. 必须设置 streaming=True 和 flush=True