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

慈云数据 7个月前 (04-05) 技术支持 87 0
  • 大家好,我是同学小张,日常分享AI知识和实战案例
  • 欢迎 点赞 + 关注 👏,持续学习,持续干货输出。
  • +v: jasper_8017 一起交流💬,一起进步💪。
  • 微信公众号也可搜【同学小张】 🙏

    本站文章一览:

    在这里插入图片描述


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

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

    文章目录

    • 0. LangChain的流式输出 Streaming
      • 0.1 实现流式输出
      • 1. OpenAI 原生的流式输出
        • 1.1 启动 OpenAI 的流式输出
        • 1.2 流式输出结果组装
        • 1.3 完整的流式输出测试程序

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

            本站文章一览:

            在这里插入图片描述

微信扫一扫加客服

微信扫一扫加客服

点击启动AI问答
Draggable Icon