【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推出全新企业智能数据平台,用以帮助用户一站式的管理企业数据服务资产,包括创建, 管理,探索, 监控等; 助力企业在现有平台之上快速构建起数据服务资产体系
目录
打赏
0
0
0
0
17
分享
相关文章
阿里云通义千问发布多款AI大模型 多模态、长文本能力全面升级!
阿里云通义千问发布多款AI大模型 多模态、长文本能力全面升级!
《2025年度企业用云十大趋势》发布,AI云全面支撑大模型落地
《2025年度企业用云十大趋势》发布,AI云全面支撑大模型落地
与上亿玩家一同闯荡江湖!逆水寒联手通义共创AI大模型竞技场
与上亿玩家一同闯荡江湖!逆水寒联手通义共创AI大模型竞技场
【活动报名】​AI应用启航workshop:瓴羊+通义助力企业迈入AI驱动的数智营销时代
【活动报名】​AI应用启航workshop:瓴羊+通义助力企业迈入AI驱动的数智营销时代
从零开始即刻拥有 DeepSeek-R1 满血版并使用 Dify 部署 AI 应用
本文介绍了如何使用阿里云提供的DeepSeek-R1大模型解决方案,通过Chatbox和Dify平台调用百炼API,实现稳定且高效的模型应用。首先,文章详细描述了如何通过Chatbox配置API并开始对话,适合普通用户快速上手。接着,深入探讨了使用Dify部署AI应用的过程,包括选购云服务器、安装Dify、配置对接DeepSeek-R1模型及创建工作流,展示了更复杂场景下的应用潜力。最后,对比了Chatbox与Dify的输出效果,证明Dify能提供更详尽、精准的回复。总结指出,阿里云的解决方案不仅操作简便,还为专业用户提供了强大的功能支持,极大提升了用户体验和应用效率。
252 18
从零开始即刻拥有 DeepSeek-R1 满血版并使用 Dify 部署 AI 应用
容器化浪潮下的AI赋能:智能化运维与创新应用
近年来,容器技术以其轻量、高效、可移植的特性成为云原生时代的基石,推动应用开发和部署方式革新。随着容器化应用规模扩大,传统运维手段逐渐力不从心。AI技术的引入为容器化生态带来新活力,实现智能监控、自动化故障诊断与修复及智能资源调度,提升运维效率和可靠性。同时,AI驱动容器化创新应用,如模型训练、边缘计算和Serverless AI服务,带来更多可能性。未来,AI与容器技术的融合将更加紧密,推动更智能、高效的运维平台和丰富的创新应用场景,助力数字化转型。
Spring AI与DeepSeek实战一:快速打造智能对话应用
在 AI 技术蓬勃发展的今天,国产大模型DeepSeek凭借其低成本高性能的特点,成为企业智能化转型的热门选择。而Spring AI作为 Java 生态的 AI 集成框架,通过统一API、简化配置等特性,让开发者无需深入底层即可快速调用各类 AI 服务。本文将手把手教你通过spring-ai集成DeepSeek接口实现普通对话与流式对话功能,助力你的Java应用轻松接入 AI 能力!虽然通过Spring AI能够快速完成DeepSeek大模型与。
82 10
Serverless + AI 让应用开发更简单,加速应用智能化
Serverless + AI 让应用开发更简单,加速应用智能化
Java 也能快速搭建 AI 应用?一文带你玩转 Spring AI 可观测性
Java 也能快速搭建 AI 应用?一文带你玩转 Spring AI 可观测性

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等