富士通株式会社デジタルシステムプラットフォーム本部Advent Calendar 2023

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

Last updated at 2023-12-25Posted at 2023-12-25

Outline

注意事項
前書き
本編
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

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up