【AI大模型应用开发】【LangChain系列】9. 实用技巧:大模型的流式输出在 OpenAI 和 LangChain 中的使用

本文涉及的产品
阿里云百炼推荐规格 ADB PostgreSQL,4核16GB 100GB 1个月
简介: 【AI大模型应用开发】【LangChain系列】9. 实用技巧:大模型的流式输出在 OpenAI 和 LangChain 中的使用
  • 大家好,我是同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • +v: jasper_8017 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:


当大模型的返回文字非常多时,返回完整的结果会耗费比较长的时间。如果等待大模型形成完整的答案再展示给用户,明显会给用户不好的体验。所以,现在市面上大多数的AI应用,在给用户结果时,都是以流式输出的方式展示给用户的。所谓的流式输出,就是类似打字机式的方式,一个字或一个词的输出,给用户一种答案逐渐出现的动画效果。

今天我们来学习下如何流式输出大模型的返回结果。本文将涵盖 LangChain 的流式输出方式和 OpenAI 原生的流式输出方式。

0. LangChain的流式输出 Streaming

0.1 实现流式输出

我们在 【AI大模型应用开发】【LangChain系列】实战案例4:再战RAG问答,提取在线网页数据,并返回生成答案的来源 代码的基础上,增加流式输出。

原代码:

import bs4
from langchain import hub
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
# Load, chunk and index the contents of the blog.
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
# Retrieve and generate using the relevant snippets of the blog.
retriever = vectorstore.as_retriever()
prompt = hub.pull("rlm/rag-prompt")
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
from langchain_core.runnables import RunnableParallel
rag_chain_from_docs = (
    RunnablePassthrough.assign(context=(lambda x: format_docs(x["context"])))
    | prompt
    | llm
    | StrOutputParser()
)
rag_chain_with_source = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
).assign(answer=rag_chain_from_docs)
result = rag_chain_with_source.invoke("What is Task Decomposition")
print(result)

修改为流式输出:

# result = rag_chain_with_source.invoke("What is Task Decomposition")
# print(result)
for chunk in rag_chain_with_source.stream("What is Task Decomposition"):
    print(chunk)

修改方式很简单,LangChain的Chain中已经帮我们封装好了 stream 接口,调用该接口获取的结果即为流式输出的结果。其输出的结果如下(每次输出一个词,词前面加一个Key,用来标识这是答案的哪一部分):

我们可以利用Key来组装答案:

output = {}
curr_key = None
for chunk in rag_chain_with_source.stream("What is Task Decomposition"):
    for key in chunk:
        if key not in output:
            output[key] = chunk[key]
        else:
            output[key] += chunk[key]
        if key != curr_key:
            print(f"\n\n{key}: {chunk[key]}", end="", flush=True)
        else:
            print(chunk[key], end="", flush=True)
        curr_key = key

这样我们看到的答案的打印过程就是一个词一个词的出现了。最后展示完跟非流式输出一样。

1. OpenAI 原生的流式输出

1.1 启动 OpenAI 的流式输出

只需要在OpenAI接口调用时,将stream参数置为True,就启用了流式输出。

response = client.chat.completions.create(
    model = model,
    messages = messages,
    temperature = temperature,
    stream=True,    # 启动流式输出
)

1.2 流式输出结果组装

结果的组装过程如下,流式输出的结果在 msg.choices[0].delta 中存着:

text = ""
print("====Streaming====")
# 需要把 stream 里的 token 拼起来,才能得到完整的 call
for msg in response:
    delta = msg.choices[0].delta
    if delta.content:
        text_delta = delta.content
        print(text_delta)
        text = text + text_delta
print("====done!====")
if text:
    print(text)

1.3 完整的流式输出测试程序

from openai import OpenAI
# 加载 .env 到环境变量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())
client = OpenAI()
###### 这里封装成函数 #######
def get_openai_chat_completion(messages, temperature, model = "gpt-3.5-turbo-1106"):
    response = client.chat.completions.create(
        model = model,
        messages = messages,
        temperature = temperature,
        stream=True,    # 启动流式输出
    )
    return response
SYSTEM_PROMPT = """
你是一名资深教师,你叫“同学小张”,用户会给你一个提示,你根据用户给的提示,来为用户设计关于此课程的学习大纲。
你必须遵循以下原则:
1. 你有足够的时间思考,确保在得出答案之前,你已经足够理解用户需求中的所有关键概念,并给出关键概念的解释。
2. 输出格式请使用Markdown格式, 并保证输出内容清晰易懂。
3. 至少输出10章的内容, 每章至少有5个小节
不要回答任何与课程内容无关的问题。
"""
if __name__ == "__main__":
    user_input = "大模型应用开发"
    
    messages = [
        {
            "role": "system",
            "content": SYSTEM_PROMPT,
        },
        {
            "role": "user",
            "content": user_input,
        }   
    ]
    response = get_openai_chat_completion(messages, 0.5)
    
    text = ""
    print("====Streaming====")
    # 需要把 stream 里的 token 拼起来,才能得到完整的 call
    for msg in response:
        delta = msg.choices[0].delta
        if delta.content:
            text_delta = delta.content
            print(text_delta)
            text = text + text_delta
    print("====done!====")
    if text:
        print(text)

流式输出过程如下:

组装后结果如下:

如果觉得本文对你有帮助,麻烦点个赞和关注呗 ~~~


  • 大家好,我是 同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习持续干货输出
  • +v: jasper_8017 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:

相关实践学习
阿里云百炼xAnalyticDB PostgreSQL构建AIGC应用
通过该实验体验在阿里云百炼中构建企业专属知识库构建及应用全流程。同时体验使用ADB-PG向量检索引擎提供专属安全存储,保障企业数据隐私安全。
AnalyticDB PostgreSQL 企业智能数据中台:一站式管理数据服务资产
企业在数据仓库之上可构建丰富的数据服务用以支持数据应用及业务场景;ADB PG推出全新企业智能数据平台,用以帮助用户一站式的管理企业数据服务资产,包括创建, 管理,探索, 监控等; 助力企业在现有平台之上快速构建起数据服务资产体系
相关文章
|
29天前
|
机器学习/深度学习 人工智能 算法
高考理科题AI秒解!昆仑万维开源多模态推理模型 Skywork-R1V 2.0
Skywork-R1V 2.0是昆仑万维最新开源的多模态推理模型,通过混合强化学习和多模态奖励模型实现复杂推理任务,在理科题目解答和科研分析中展现出色性能。
172 11
高考理科题AI秒解!昆仑万维开源多模态推理模型 Skywork-R1V 2.0
|
29天前
|
机器学习/深度学习 人工智能 编解码
OpenAI 最新多模态图像生成模型 GPT-image-1:一句话生成商业图+智能修图
GPT-image-1是OpenAI推出的新一代多模态图像生成模型,通过API提供专业级图像生成与编辑能力,支持多种分辨率、格式和艺术风格定制。
173 10
OpenAI 最新多模态图像生成模型 GPT-image-1:一句话生成商业图+智能修图
|
1月前
|
人工智能 搜索推荐
「社会实验室」成真!SocioVerse:复旦联合小红书开源社会模拟世界模型,用AI预演群体行为
SocioVerse是由复旦大学联合小红书等机构开源的社会模拟框架,基于大语言模型和千万级真实用户数据构建,能精准模拟群体行为并预测社会事件演化趋势。
122 2
「社会实验室」成真!SocioVerse:复旦联合小红书开源社会模拟世界模型,用AI预演群体行为
|
1月前
|
人工智能 编解码 算法
AI生成视频告别剪辑拼接!MAGI-1:开源自回归视频生成模型,支持一镜到底的长视频生成
MAGI-1是Sand AI开源的全球首个自回归视频生成大模型,采用创新架构实现高分辨率流畅视频生成,支持无限扩展和精细控制,在物理行为预测方面表现突出。
256 1
AI生成视频告别剪辑拼接!MAGI-1:开源自回归视频生成模型,支持一镜到底的长视频生成
|
1月前
|
数据采集 人工智能 自动驾驶
从虚拟到现实!Aether:上海AI Lab开源的生成式世界模型,4D动态重建+视觉规划全搞定
Aether是上海AI Lab开源的生成式世界模型,通过三维时空建模与生成式建模的深度融合,实现了4D动态重建、动作条件视频预测和目标导向视觉规划三大核心能力。
98 1
从虚拟到现实!Aether:上海AI Lab开源的生成式世界模型,4D动态重建+视觉规划全搞定
|
24天前
|
人工智能 自然语言处理 搜索推荐
AI 搜索开放平台重磅发布:Qwen3 模型上线啦
阿里云AI搜索开放平台重磅发布最新Qwen3模型,为企业和开发者提供全栈智能搜索解决方案。Qwen3作为最新一代大模型,在推理、多语言支持和Agent能力上表现卓越。用户可通过三步快速体验Qwen3服务,助力业务在AI时代抢占先机。
195 12
|
2天前
|
人工智能 负载均衡 API
长连接网关技术专题(十二):大模型时代多模型AI网关的架构设计与实现
随着 AI 技术快速发展,业务对 AI 能力的渴求日益增长。当 AI 服务面对处理大规模请求和高并发流量时,AI 网关从中扮演着至关重要的角色。AI 服务通常涉及大量的计算任务和设备资源占用,此时需要一个 AI 网关负责协调这些请求来确保系统的稳定性与高效性。因此,与传统微服务架构类似,我们将相关 API 管理的功能(如流量控制、用户鉴权、配额计费、负载均衡、API 路由等)集中放置在 AI 网关层,可以降低系统整体复杂度并提升可维护性。 本文要分享的是B站在大模型时代基于多模型AI的网关架构设计和实践总结,希望能带给你启发。
28 4
|
4天前
|
SQL 人工智能 自然语言处理
阿里云 AI 搜索开放平台新功能发布:新增GTE自部署模型
阿里云 AI搜索开放平台正式推出 GTE 多语言通用文本向量模型(iic/gte_sentence-embedding_multilingual-base)
|
2天前
|
机器学习/深度学习 人工智能 算法
Python+YOLO v8 实战:手把手教你打造专属 AI 视觉目标检测模型
本文介绍了如何使用 Python 和 YOLO v8 开发专属的 AI 视觉目标检测模型。首先讲解了 YOLO 的基本概念及其高效精准的特点,接着详细说明了环境搭建步骤,包括安装 Python、PyCharm 和 Ultralytics 库。随后引导读者加载预训练模型进行图片验证,并准备数据集以训练自定义模型。最后,展示了如何验证训练好的模型并提供示例代码。通过本文,你将学会从零开始打造自己的目标检测系统,满足实际场景需求。
20 0
Python+YOLO v8 实战:手把手教你打造专属 AI 视觉目标检测模型
|
1月前
|
机器学习/深度学习 人工智能 编解码
这个AI能拍无限长电影!SkyReels-V2:昆仑万维开源无限时长电影生成模型!
SkyReels-V2是昆仑万维推出的突破性视频生成模型,基于扩散强迫框架和多模态大语言模型技术,支持生成理论上无限时长的连贯视频内容,在影视制作、广告创意等领域展现强大潜力。
323 7
这个AI能拍无限长电影!SkyReels-V2:昆仑万维开源无限时长电影生成模型!

热门文章

最新文章