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

# Outline
- [注意事項](#注意事項)
- [前書き](#前書き)
- [本編](#本編)
  - [先に結論を](#先に結論を) 
  - [GPTのストリーミング仕様](#gptのストリーミング仕様)
  - [Pythonからストリームをコールするには](#pythonからストリームをコールするには)
  - [なぜFetch Event Source推しなのか](#なぜfetch-event-source推しなのか)
  - [Fetch Event Sourceの使い方](#fetch-event-sourceの使い方)
- [Reference](#reference)

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

# 前書き
本記事は「[GPTのストリーム機能を使いたい方へ（その1）](https://qiita.com/eto_ryota/items/fa5f41b9c7a7e5bce1eb)」の続編です。

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

- OpenAIのストリーム機能について
  - ストリームによるメリットを知りたい 
  - Node.jsからじゃなく、Python（[openai](https://pypi.org/project/openai/)）からストリームを実現したい
- そもそもストリームって何？
- FetchとEventSourceの違いを知りたい

# 本編
- **[GPTのストリーム機能を使いたい方へ（その1）](https://qiita.com/eto_ryota/items/fa5f41b9c7a7e5bce1eb)**
   ストリーミングの動作イメージや、メリットについて解説していきます。
   <br>
 
- **GPTのストリーム機能を使いたい方へ（その2）**
   以下について解説していきます。（**いまココ！**） 
   - GPTのストリーミングをPythonから呼び出す方法
   - Fetch / EventSource / FetchEventSourceの違い

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

以下に機能の比較を記載します。
| インタフェース | メリット | デメリット | 
| :----: | :----: | :----: | 
| Streams API | ・追加パッケージなしに利用できる <br> ・JSONをPOST出来る |・SSEイベントを扱えない
| EventSource |・追加パッケージなしに利用できる <br> ・SSEイベントを扱える |・JSONをPOST出来ない | 
| Fetch-Event-Source |・JSONをPOST出来る <br> ・SSEイベントを扱える | ・インストールが必要

## GPTのストリーミング仕様
[OpenAIのAPI Reference](https://platform.openai.com/docs/api-reference/streaming)では以下のように`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](https://html.spec.whatwg.org/multipage/server-sent-events.html#server-sent-events) standard.

SSEの仕様を例を使って押さえていきましょう。
SSEはいくつかのフィールドから`Message`が定義されますが、特に大切なのは`event`と`data`の2つのフィールドです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3658085/cb2cd58b-5926-1488-d49b-d11a904af803.png)

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

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

[「前回の投稿」](https://qiita.com/eto_ryota/items/fa5f41b9c7a7e5bce1eb#%E3%81%BE%E3%81%9A%E5%8B%95%E4%BD%9C%E3%82%A4%E3%83%A1%E3%83%BC%E3%82%B8%E3%81%8B%E3%82%89)で記載した「ストリームが有効の場合」を見ると、上述した`data`フィールドが返ってきていることが分かりますね。

## Pythonからストリームをコールするには
[「前回の投稿」](https://qiita.com/eto_ryota/items/fa5f41b9c7a7e5bce1eb#%E3%83%A1%E3%83%AA%E3%83%83%E3%83%88%E3%81%9D%E3%81%AE%E6%99%82%E7%82%B9%E3%81%BE%E3%81%A7%E3%81%AE%E7%94%9F%E6%88%90%E7%B5%90%E6%9E%9C%E3%82%92%E5%BE%97%E3%82%89%E3%82%8C%E3%82%8B)ではPythonからストリームでレスポンスを取得するには`stream:True`としました。

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

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


- **実行サンプルコード**
    ```py
    ## 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()
    ```
<br>

- **応答結果**
![streaming_yield.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3658085/de417f9e-76da-a159-c064-e9dd03ebf956.gif)

<br>

- **ジェネレータ関数の動き**
  ジェネレータ関数と聞いてピンとこない方も多いのではないかと思いますが、**Pythonでは`yield`を含む関数がジェネレータ関数となります。**
  この動作は少し特殊で、アクセスした際に**③をスキップし**、②まで処理が行われた後、③が生成されたChunkの数だけ繰り返されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3658085/60a39bed-7df8-ad9b-191d-037bef8044c7.png)


<br>

- **応答結果から分かること**
  `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.)が発生した場合

<br>

- **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](https://github.com/Azure/fetch-event-source)のインストール&使い方について見ていきましょう。

まずはインストールからですね、npmから導入可能です。
- `Fetch Event Source`のインストール
  ```sh
   npm install @microsoft/fetch-event-source
  ```

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

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

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

  ```vue
  <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](https://ja.javascript.info/server-sent-events)
- [Github：Fetch Event Srouce](https://github.com/Azure/fetch-event-source)