GPTのストリーム機能を使いたい方へ（その2）

Outline

• 注意事項
• 前書き
• 本編
  • 先に結論を
  • GPTのストリーミング仕様
  • Pythonからストリームをコールするには
  • なぜFetch Event Source推しなのか
  • Fetch Event Sourceの使い方
• Reference

注意事項

本記事は「富士通株式会社 デジタルシステムプラットフォーム本部 Advent Calendar 2023」の25日目の記事です。記事の掲載内容は私自身の見解であり、所属する組織を代表するものではありません（お約束）。

前書き

本記事は「GPTのストリーム機能を使いたい方へ（その1）」の続編です。

以下のようなニーズに応えたいと思っています。

• OpenAIのストリーム機能について
  • ストリームによるメリットを知りたい
  • Node.jsからじゃなく、Python（openai）からストリームを実現したい
• そもそもストリームって何？
• FetchとEventSourceの違いを知りたい

本編

• GPTのストリーム機能を使いたい方へ（その1）
  ストリーミングの動作イメージや、メリットについて解説していきます。
  
  

• GPTのストリーム機能を使いたい方へ（その2）
  以下について解説していきます。（いまココ！）

  • GPTのストリーミングをPythonから呼び出す方法
  • Fetch / EventSource / FetchEventSourceの違い

先に結論を

本記事ではGPTのストリーミングを利用する際のインタフェースとして、JSONがPOST出来ること、SSEイベントを扱えることから、Fetch Event Sourceを推しています。

以下に機能の比較を記載します。

インタフェース	メリット	デメリット
Streams API	・追加パッケージなしに利用できる ・JSONをPOST出来る	・SSEイベントを扱えない
EventSource	・追加パッケージなしに利用できる ・SSEイベントを扱える	・JSONをPOST出来ない
Fetch-Event-Source	・JSONをPOST出来る ・SSEイベントを扱える	・インストールが必要

GPTのストリーミング仕様

OpenAIのAPI Referenceでは以下のようにServer-sent events（以降、SSEと記載）に従うとあります。

Streaming

The OpenAI API provides the ability to stream responses back to a client in order to allow partial results for certain requests. To achieve this, we follow the Server-sent events standard.

SSEの仕様を例を使って押さえていきましょう。
SSEはいくつかのフィールドからMessageが定義されますが、特に大切なのはeventとdataの2つのフィールドです。

こちらで押さえておきたいのは
eventフィールドで、ストリーミングに名前を付けることが出来ること。および
dataフィールドが細切れにされたデータ(Chunk)を送信する機能を持つということです。

さらっと書いてますが、以下の仕様に注意してください。
　・eventフィールドは、dataよりも前で定義する必要があります
　・各Messageは\n\nで区切る必要があります

「前回の投稿」で記載した「ストリームが有効の場合」を見ると、上述したdataフィールドが返ってきていることが分かりますね。

Pythonからストリームをコールするには

「前回の投稿」ではPythonからストリームでレスポンスを取得するにはstream:Trueとしました。

単純にレスポンスを得るためであれば、上記だけで問題ありませんが、アプリケーションを意識すると、さらに工夫が必要で、ジェネレーター関数としてChunkを返してあげてください。

以下はエラーハンドリングやKey管理、PEP8などをサボっているVerです。

• 実行サンプルコード

  ## Import OpenAI Library
  import openai
  ## Import Flask Library
  from flask import Flask, Response, stream_with_context
  ## Parameter Setting
  ## 注意
  ## - きちんとしたアプリケーションとして構成するときはハードコーディングせずKey管理サービス(Ex: Azure Key Vault等)を利用しましょう！
  openai.api_type    = "azure"
  openai.api_key     = "<YOUR_API_KEY>"
  openai.api_base    = "<YOUR_ENDPOINT>"
  openai.api_version = "<YOUR_API_VERSION>"
  
  stream   = True
  messages = [
      {
          "role":     "user",
          "content":  "以下の指示に従って下さい。 \n\n" \
                      "「こんにちは、人間。」\n" \
                      "「」の中の文章を必ず＜主張＞を記載した後に、そのまま返信してください。"
      }
  ]
  app = Flask(__name__)
  
  @app.route('/')
  def start():
      ## ストリーム形のChunkを返すジェネレータ関数
      def gpt_stream():
          generator = openai.ChatCompletion.create(
                  deployment_id="<YOUR_MODEL_NAME>",
                  messages=messages,
                  # ストリームを有効化する場合は`True`
                  stream=stream,
                  max_tokens=500
              )
          for chunk in generator:
              if chunk['choices'][0]['finish_reason'] is None:
                  stream_token = chunk['choices'][0]['delta'].get('content', '')
                  time.sleep(2)
                  yield "event: <YOUR_EVENT>\ndata: {}\n\n".format(stream_token)
  
      response = Response(stream_with_context(gpt_stream()), mimetype='text/event-stream')
      return response
  
  if __name__ == "__main__":
      app.run()
  

• 応答結果
  

• ジェネレータ関数の動き
  ジェネレータ関数と聞いてピンとこない方も多いのではないかと思いますが、Pythonではyieldを含む関数がジェネレータ関数となります。
  この動作は少し特殊で、アクセスした際に③をスキップし、②まで処理が行われた後、③が生成されたChunkの数だけ繰り返されます。
  

• 応答結果から分かること
  yieldで指定した形式でデータがブラウザ側に返却されていることが分かります。
  あとは、ブラウザ側でFetch / EventSource / Fetch Event Source等を用いて、データを上手く扱っていく方法を考えていきましょう。

なぜFetch Event Source推しなのか？

結論で先出しましたが、Fetch Event Sourceは汎用性は限られるものの、両者のいいとこどりをしているインタフェースです。

その有難みを理解するため、FetchとEventSourceでデメリットに記載した内容を深掘りしていきましょう。

• SSEイベントを扱えない
  先述のevent <YOUR EVENT>のように、SSEでは送信するChunkにイベント名を付与することが可能ですがFetch（より正確にはStream API）ではこのイベントを扱うことが出来ません。
  ※ 本記事では便宜上「SSEイベント」と記載していますが正式名称ではありません。

  SSEイベントを扱えないと何が困るのか、以下に列挙しておきます。

  • 生成されたトークンと以下が区別できない
    • token_limitに係る残トークンを計算する場合
    • コンテンツフィルターに検知された場合などのfinish_reasonが異なる場合
    • エラー(openai.error.RateLimitError,etc.)が発生した場合

• JSONをPOSTできない
  こちらはEventSourceの問題ですね。表題の通り、EventSourceではJSONをPOSTすることが出来ません。
  より正確には、https://ドメイン?XXXのようなクエリパラメータ形式でデータをPOSTすることは出来ますが、約2000文字までという制限もあり、GPTの性能を活かすことが難しくなります。

Fetch Event Sourceの使い方

Fetch Event SourceはMircrosoft OSSコミュニティがContributorとして参画している取り組みとして開発されているようです。GitHub：Azure/Fetch Event Sourceのインストール&使い方について見ていきましょう。

まずはインストールからですね、npmから導入可能です。

• Fetch Event Sourceのインストール

   npm install @microsoft/fetch-event-source
  

インストールが完了したら実際のソースコードで利用してみましょう。

Fetch Event SourceはTypeScriptで作られています。
Vue.jsで利用される際にはイベントをアロー関数で記載してください。

• Fetch Event Sourceの利用例（Vue.js）

  <template>
   <!-- Vue.jsでのフロントエンドコード -->
  </template>
  <script>
  // Fetch Event Sourceのインポート
  import { fetchEventSource } from '@microsoft/fetch-event-source';
  methods: {
      // fetchEventSourceを利用してJSONをPOST
      await fetchEventSource('/gpt_stream', {
          method: 'POST',
          headers: {
              'Content-Type': 'application/json'
          },
          body: JSON.stringify(<YOUR_REQUEST_JSON>),
          onopen: async (response) => {
              // レスポンス受信時の処理を記載
          }
          // Python側から送信した<YOUR_EVENT>の処理
          onmessage: (response) => {
              if (response.event === '<YOUR_EVENT>') {
                  // YOUR_ENVENT受信時の処理を記載
              }
          }
          onerror: (error) => {
                // エラー発生時の処理を記載
          }
      }
  }
  

上記のように使ってあげることで、Python側で指定した<YOUR_EVENT>毎にブラウザ側でも処理させる仕組みのフレームができました。

不明点や○○の解説記事が欲しい！などありましたらコメントいただけると幸いです。

Reference

• JAVASCRIPT.INFO：Server Sent Event
• Github：Fetch Event Srouce