我们在使用大模型网站时，有时会看见有些模型的输出是一个字一个字蹦出来的，而不是我们之前API调用那样一下子返回完整结果。实际上我们可以通过设置流式输出来达到这样的效果。

stream同步流式传输

我们可以使用模型的.stream()方法传消息列表并获取迭代器，使用for遍历迭代器，进行流式输出

1
2
3
4
5
6
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

for chunk in model.stream("讲一个100字的冷笑话"):
    print(chunk.content,end="|")
1
2
3
4
5
D:\program_software\AnaConda\envs\langChainP13\python.exe D:\codes\code_pycharm\langChainTool\stream.py 
|为什么|书|总|是|很|冷|？

|因为|它|们|有|很多|“|章节|”|！||||
Process finished with exit code 0

astream

langchain也支持了异步流式调用，为编程提供了更大的灵活性

1
2
3
4
5
6
7
8
9
10
11
12
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")


async def async_output():
    print("开始异步调用")
    async for chunk in model.astream("讲一个100个汉字字的冷笑话"):
        print(chunk.content, end="|")

import asyncio
asyncio.run(async_output())
1
2
3
4
5
6
7
8
9
10
D:\program_software\AnaConda\envs\langChainP13\python.exe D:\codes\code_pycharm\langChainTool\stream.py 
开始异步调用
|有|一天|，一|只|鸭|子|走|进|药|店|，|问|店|员|：“|你|们|这里|有|卖|葡|萄|吗|？”|店|员|愣|了一|下|，|回答|：“|我们|这里|不|卖|葡|萄|。”|鸭|子|点|点|头|，|转|身|离|开|。

|第二|天|，|鸭|子|又|来了|，|问|：“|你|们|这里|有|卖|葡|萄|吗|？”|店|员|无|奈|地|说|：“|我|不是|告诉|过|你|吗|？|我们|不|卖|葡|萄|！”

|第三|天|，|鸭|子|再|来|，|问|：“|你|们|这里|有|卖|葡|萄|吗|？”|店|员|生|气|地|说|：“|如果|你|再|来|问|，我|就|用|钉|子|把|你|钉|住|！”

|第四|天|，|鸭|子|又|来了|，|问|：“|你|们|这里|有|卖|钉|子|吗|？”|店|员|回答|：“|没有|。”|鸭|子|笑|着|说|：“|那|你|们|有|卖|葡|萄|吗|？”||||
Process finished with exit code 0

使用其它Runnable实例实现流式输出

我们再来看一下Runnable接口的功能描述

Runnable定义了⼀个标准接⼝，允许Runnable组件：

  • Invoked（调⽤）:单个输⼊转换为输出。
  • Batched（批处理）:多个输⼊被有效地转换为输出。
  • Streamed（流式传输）:输出在⽣成时进⾏流式传输。
  • Inspected（检查）:可以访问有关Runnable的输⼊、输出和配置的原理图信息。
  • Composed（组合）:可以组合多个Runnable，以使⽤LCEL协同⼯作以创建复杂的管道。
  • …..

可以看到只要是Runnable接口都支持流式输出，因此我们也可以在原本的模型调用的基础上，给链增加一个输出解析器StrOutParser并流式输出

1
2
3
4
5
6
7
8
9
10
11
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")

parser = StrOutputParser()

chain = model | parser

for chunk in chain.stream("讲一个50汉字的冷笑话"):
    print(chunk,end='|')

输出如下

1
2
3
D:\program_software\AnaConda\envs\langChainP13\python.exe D:\codes\code_pycharm\langChainTool\parser.py 
|有|一天|，一|只|蚂|蚁|和|大|象|一起|散|步|，|蚂|蚁|突然|摔|了一|跤|。|大|象|关|心|地|问|：“|你|没|事|吧|？”|蚂|蚁|摇|摇|头|说|：“|没|事|，只|是|摔|了一|跤|，|没|想到|一|跤|摔|出了|别|的|‘|高度|’|！”||||
Process finished with exit code 0

自定义流式输出解析器

既然我们可以使用输出解析器来修饰输出，那么我们就可以自定义一个流式解析器，让它保留流式输出的功能，但是是一句话一句话地输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing import Iterator,List

model = ChatOpenAI(model="gpt-4o-mini")

parser = StrOutputParser()

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 [buffer[:stop_index].strip()]
            buffer = buffer[stop_index + 1 :]

    yield [buffer.strip()]

chain = model | parser | split_into_list

for chunk in chain.stream("讲一个100汉字的冷笑话，每句话用中文句号分割"):
    print(chunk,end='|')

输出如下

1
2
3
4
D:\program_software\AnaConda\envs\langChainP13\python.exe D:\codes\code_pycharm\langChainTool\parser.py 
['有一天，香蕉和苹果一起去旅行']|['它们在路上遇到了一只大象']|['香蕉问：“你怎么不带鸟儿一起去？”苹果回应：“因为它们太烦了，整天喳喳叫']|['”大象听了，低头说：“其实，我更喜欢喝果汁']|['”香蕉和苹果面面相觑，心想：“它竟然会喝果汁！”这时，香蕉突然掉了下来，大象笑着说：“看来你也喝‘汁’了！”']|
Process finished with exit code 0