Outline
注意事項
本記事は「富士通株式会社 デジタルシステムプラットフォーム本部 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>
毎にブラウザ側でも処理させる仕組みのフレームができました。
不明点や○○の解説記事が欲しい!などありましたらコメントいただけると幸いです。