【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 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

本站文章一览:

相关实践学习
AnalyticDB PostgreSQL 企业智能数据中台:一站式管理数据服务资产
企业在数据仓库之上可构建丰富的数据服务用以支持数据应用及业务场景;ADB PG推出全新企业智能数据平台,用以帮助用户一站式的管理企业数据服务资产,包括创建, 管理,探索, 监控等; 助力企业在现有平台之上快速构建起数据服务资产体系
相关文章
|
9天前
|
人工智能 缓存 监控
使用LangChain4j构建Java AI智能体:让大模型学会使用工具
AI智能体是大模型技术的重要演进方向,它使模型能够主动使用工具、与环境交互,以完成复杂任务。本文详细介绍如何在Java应用中,借助LangChain4j框架构建一个具备工具使用能力的AI智能体。我们将创建一个能够进行数学计算和实时信息查询的智能体,涵盖工具定义、智能体组装、记忆管理以及Spring Boot集成等关键步骤,并展示如何通过简单的对话界面与智能体交互。
194 1
|
9天前
|
人工智能 Java API
AI 超级智能体全栈项目阶段一:AI大模型概述、选型、项目初始化以及基于阿里云灵积模型 Qwen-Plus实现模型接入四种方式(SDK/HTTP/SpringAI/langchain4j)
本文介绍AI大模型的核心概念、分类及开发者学习路径,重点讲解如何选择与接入大模型。项目基于Spring Boot,使用阿里云灵积模型(Qwen-Plus),对比SDK、HTTP、Spring AI和LangChain4j四种接入方式,助力开发者高效构建AI应用。
376 122
AI 超级智能体全栈项目阶段一:AI大模型概述、选型、项目初始化以及基于阿里云灵积模型 Qwen-Plus实现模型接入四种方式(SDK/HTTP/SpringAI/langchain4j)
|
1月前
|
SQL 人工智能 自然语言处理
阿里云 CIO 蒋林泉:AI 大模型时代,我们如何用 RIDE 实现 RaaS 的首次落地?
本文整理自阿里云智能集团 CIO 蒋林泉在 AICon 2025 深圳的演讲,分享了阿里云在大模型应用落地中的实践经验。通过多个数字人项目案例,探讨了企业在 AI 应用中的组织转型、业务识别、产品定义与工程落地等关键环节,并提出了 RIDE 方法论(重组、识别、定义、执行),助力企业实现 AI 有效落地。
|
3月前
|
存储 运维 数据挖掘
革新智能驾驶数据挖掘检索效率!某国内新能源汽车未来出行领导者选择阿里云Milvus构建多模态检索引擎
在智能驾驶技术快速发展中,数据成为驱动算法进步的核心。某新能源汽车领军企业基于阿里云Milvus向量数据库构建智能驾驶数据挖掘平台,利用其高性能、可扩展的相似性检索服务,解决了大规模向量数据检索瓶颈问题,显著降低20%以上成本,缩短模型迭代周期,实现从数据采集到场景挖掘的智能化闭环,加速智能驾驶落地应用。
革新智能驾驶数据挖掘检索效率!某国内新能源汽车未来出行领导者选择阿里云Milvus构建多模态检索引擎
|
3月前
|
存储 机器学习/深度学习 缓存
阿里云AirCache技术实现多模态大模型高效推理加速,入选国际顶会ICCV2025
阿里云研发的AirCache技术被计算机视觉顶会ICCV2025收录,该技术通过激活跨模态关联、优化KV缓存压缩策略,显著提升视觉语言模型(VLMs)的推理效率与存储性能。实验表明,在保留仅10%视觉缓存的情况下,模型性能下降小于1%,解码延迟最高降低66%,吞吐量提升达192%。AirCache无需修改模型结构,兼容主流VLMs,已在教育、医疗、政务等多个行业落地应用,助力多模态大模型高效赋能产业智能化升级。
291 1
|
5月前
|
程序员 定位技术 开发者
试了试阿里云的通义灵码 2.5 版
通义灵码 2.5 版是个特别实用的工具,无论是个人开发者还是企业团队,都能从中受益。如果你也在找能提升开发效率的工具,通义灵码绝对值得一试!
196 33
试了试阿里云的通义灵码 2.5 版
|
4月前
|
存储 人工智能 云计算
挑战杯专属支持资源|阿里云-AI大模型算力及实验资源丨云工开物
阿里云发起的“云工开物”高校支持计划,助力AI时代人才培养与科研创新。为“挑战杯”参赛选手提供专属算力资源、AI模型平台及学习训练资源,包括300元免费算力券、百炼大模型服务、PAI-ArtLab设计平台等,帮助学生快速掌握AI技能并构建优秀作品,推动产学研融合发展。访问链接领取资源:https://university.aliyun.com/action/tiaozhanbei。
|
13天前
|
存储 机器学习/深度学习 人工智能
云栖 2025|阿里云 Qwen3 系列领衔:AI 模型全栈突破与开发者落地指南
阿里云发布Qwen3全栈AI体系,七大模型升级、性能全球领先,开源生态稳居第一。从底层基建到开发工具链全面优化,助力企业高效落地AI应用,共建超级AI云生态。
449 11
|
11天前
|
机器学习/深度学习 人工智能 数据安全/隐私保护
阿里云 Qwen3 全栈 AI 模型:技术解析、开发者实操指南与 100 万企业落地案例
阿里云发布Qwen3全栈AI体系,推出Qwen3-Max、Qwen3-Next等七大模型,性能全球领先,开源生态超6亿次下载。支持百万级上下文、多模态理解,训练成本降90%,助力企业高效落地AI。覆盖制造、金融、创作等场景,提供无代码与代码级开发工具,共建超级AI云生态。
195 6

热门文章

最新文章