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

文章目录

• 一、消息内存缓存
  • 核心概念
  • 关键组件
  • 代码流程
  • 运行效果
• 二、消息过滤
  • 核心概念
  • 关键函数
  • 过滤参数
  • 代码示例
  • 过滤逻辑
• 三、消息合并
  • 核心概念
  • 关键函数
  • 代码示例
  • 合并效果
  • 两种使用方式
• 四、流式输出
  • 什么是流式输出
  • 为什么需要？
  • 典型应用
• 五、同步 vs 异步流式
  • 核心区别
  • 工作原理
  • 何时使用异步？
• 六、流式输出基础用法
  • 同步流式
  • 异步流式
• 七、输出解析器
• 八、流式输出实际应用
  • 1. 聊天机器人
  • 2. 多用户并发
  • 3. FastAPI 集成
• 九、常见问题
  • 1. 没有流式效果？
  • 2. async for 报错？
  • 3. 性能对比
• 十、总结对比
  • 典型应用场景
  • 流式输出要点

一、消息内存缓存

核心概念

通过 InMemoryChatMessageHistory 将对话历史存储在内存中，使模型能"记住"之前的对话内容。

关键组件

代码流程

# 1. 创建内存存储字典 memory_store ={}# 2. 定义获取会话历史的函数（按 session_id 区分会话）defget_session_history(session_id:str):if session_id notin memory_store: memory_store[session_id]= InMemoryChatMessageHistory()return memory_store[session_id]# 3. 用 RunnableWithMessageHistory 包装模型 message_model = RunnableWithMessageHistory(model, get_session_history)# 4. 通过 config 指定会话 ID config ={"configurable":{"session_id":"123"}}# 5. 多轮对话，模型自动记住上下文 response1 = message_model.invoke({"input":"你好，我是小明"}, config=config) response2 = message_model.invoke({"input":"我叫什么名字？"}, config=config)# → 模型能回答出"小明"，因为历史被缓存了

运行效果

• 第一轮：用户说"我是小明"，AI 正常打招呼
• 第二轮：用户问"我叫什么名字"，AI 能从历史中回忆出"小明"

从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 消息，且排除 的消息 filtered_messages = filter_messages( messages, include_types=["ai"], exclude_ids=["4"],)# → 结果只剩 的 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 asyncdefmain(): chain = model | parser asyncfor chunk in chain.astream("写一个关于程序员的笑话"):print(chunk, end="|", flush=True)if __name__ =="__main__": asyncio.run(main())

关键点：

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

七、输出解析器

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

作用：

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

自定义解析器：

defcustom_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")asyncdefchat_stream(question:str):asyncdefgenerate():asyncfor 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