企业官网建设流程全解析

前言

本期我们将要介绍的聊天模型中的流式传输。什么是流式传输？简单来说流式传输就是其通过逐步显示输出，甚至在完整的响应准备就绪之前显示部分输出。相较于在这之前所学习的.invoke()方法的非流式输出，其输出内容是全量返回的，需要所有内容准备就绪后才一并进行输出。因此导致了用户等待输出时间长的问题。而流式输出的能力将有效改善用户的体验。

一、流式传输方式

（一）stream() 同步传输

在 LangChain 聊天模型中，可以使用其 .stream() 方法，来同步地生成流式响应的效果。而 .stream() 方法返回的是一个迭代器，该迭代器在生成输出时同步产生输出消息块。

import os import asyncio from langchain_openai import ChatOpenAI model = ChatOpenAI( model = "deepseek-flash", api_key = os.environ.get("DEEPSEEK_API_KEY"), base_url = "https://api.deepseek.com", ) # 同步流式输出 # chunk: AIMessageChunk # 返回一个迭代器，产生的消息块 chunks = [] for chunk in model.stream("写一篇关于薛定谔的猫的短篇故事，字数在100字以内"): chunks.append(chunk) print(chunk.content, end=" ", flush=True) tmp_chunks = chunks[0] + chunks[1] + chunks[2] +chunks[3] print(tmp_chunks)

通过调试，可看到chunk其实是一个叫做 AIMessageChunk 的东西，它代表 AIMessage 的一部分，也就是消息块。这些消息块还可以直接相加。

（二）astream() 异步传输

1、异步概念

关于异步方式asyncio之前，我们要介绍协程和事件循环另外两个概念。

关于协程我们要先知道：

• 多进程通常利用的是多核 CPU 的优势，同时执行多个计算任务。每个进程有自己独立的内存管理，所以不同进程之间要进行数据通信会比较麻烦。
• 多线程是在一个 cpu 上创建多个子任务，当某一个子任务休息的时候其他任务接着执行。多线程的控制是由 python 自己控制的。而线程存在数据同步问题，所以会有锁机制。
• 协程的实现是在一个线程内实现的，相当于流水线作业。由于线程切换的消耗比较大，所以对于并发编程，可以优先使用协程。

所以协程作为⼀种轻量级的并发编程模型，可以被视为用户态的“轻量级线程”。其与传统线程相比，协程的核心优势在于其调度完全由用户空间掌控，避免了操作系统内核的频繁介入，从而显著降低了上下文切换的开销。
协程的切换则由程序员和编程语言控制，程序员决定在何时暂停或恢复协程。协程是一个特殊的函数，它可以在执行过程中暂停，并在稍后恢复执行。其用 async def 定义，并在需要暂停的地方使用 await。

事件循环则是 asyncio（Python 标准库中的模块，用于编写异步 I/O 操作的代码）的核心，你可以把它想象成一个总调度员或一个高效的待办事项 (To-Do List) 的管理员。

事件循环的工作流程大致如下：

1. 维护着一个任务列表。
2. 不断地循环检查每个任务。如果任务处于 “等待I/O” 状态，就暂停它，立即去执行下⼀个已经 “就绪” 的任务；如果任务的等待时间到了或者 I/O 操作完成了，事件循环就恢复执行这个任务。

在这里我们举出烧水与发消息两个事件一起分别用同步、异步的方式来执行的例子来进一步理解。

# 同步IO import time def boil_water(): print("开始烧水...") time.sleep(5) # 模拟烧水5s, CPU 完全空闲 print("烧水结束...") def send_message(): print("开始发消息...") time.sleep(2) print("发消息完成...") def main(): boil_water() send_message() # 共耗时7s main() # 异步IO import asyncio # 协程 async def boil_water_async(): print("开始烧水...") await asyncio.sleep(5) print("烧水结束...") async def send_message_async(): print("开始发消息...") await asyncio.sleep(2) print("发消息完成...") async def main_async(): task1 = asyncio.create_task(boil_water_async()) task2 = asyncio.create_task(send_message_async()) await task1 await task2 # 5s asyncio.run(main_async())

2、asyncio 具体使用

我们可以通过 .astream() 方法，来异步生成流式响应的效果。我们也可以在
其他异步代码中使用它来实现相同的实时流式处理行为。

model = ChatOpenAI( model = "deepseek-flash", api_key = os.environ.get("DEEPSEEK_API_KEY"), base_url = "https://api.deepseek.com", ) async def async_stream(): print("===异步调用===") async for chunk in model.astream("写一篇关于薛定谔的猫的短篇故事，字数在100字以内"): print(chunk.content,end=" ",flush=True) asyncio.run(async_stream())

二、自定义流式输出解析器

在介绍完如何让聊天模型进行流式输出后，我们发现上面的结果都是一个字或两个字的间歇性的输出。如果此时我们想要将输出改为一句话一句话的输出，同时保留流式处理功能，此时便需要在链中使用生成器函数，即可完成自定义流式输出的能力。

所以为实现一句话一句话的输出，同时保留流式处理功能的效果，我们要使用生成器函数来生成自定义的流式输出解析器来对聊天模型的 .stream() 方法的返回内容进行解析。
我们知道 .stream() 的是一个迭代器，该迭代器在生成输出时同步产生输出 消息块。此时我们用句号分隔消息块列表来实现解析内容。

from langchain_core.output_parsers import StrOutputParser from langchain_deepseek import ChatDeepSeek from typing import Iterator, List # 自定义输出格式 # 组件1: 聊天模型 model = ChatDeepSeek(model="deepseek-v4-flash") # 组件2: 输出解析器 parser = StrOutputParser() # 组件3: 自定义生成器 def split_into_list(input: Iterator[str])->Iterator[List[str]]: buffer = "" for chunk in input: buffer += chunk # 设为遇到。就需要刷新 while "。" in buffer: # 找到句号的位置 stop_index = buffer.index("。") # yield 用于创造生成器 yield [buffer[:stop_index + 1].strip()] buffer = buffer[stop_index + 1:] # 处理buffer最后几个字 yield [buffer.strip()] # 定义链 chain = model | parser | split_into_list # 返回一个迭代器，产生的消息块 for chunk in chain.stream("写一段关于爱情的歌词，共5句话，每句用中文句号“。”隔开"): print(chunk,end=" ",flush=True)

这样便实现了我们所想要的效果。

结语

以上便是Langchain中聊天模型流式传输的内容了。如果喜欢我的内容，请点赞、收藏加关注，多多支持我，以及欢迎各位在评论区讨论交流，并指出我的不足，谢谢大家。