UnityでChatGPT APIのServer-Sent Eventsに対応してデータを順次受信する

ChatGPT APIにて、streamオプションを有効にしServer-Sent Eventsという技術を利用することで、徐々に文章を受信し表示する場合のUnity実装コードについて紹介します。

以下のような動作になります。ChatGPTのWebサイトのような動作ですね。

これにより、データを一括で送受信（もしくは推論）するよりも、ユーザに対してなるべく早く応答を開始できるので体感を向上されられるかと思います。

なお、紹介するコードはAsyncEnumerableを利用しているため、C#8/Unity 2020.2以降が必要です。

＊2023/3/17: OpenAIChatCompletionAPI.csの実装をよりシンプルかつ安全にできたのでコードを修正しました。

サンプルコード・プロジェクト

こちらのリポジトリのstreamブランチに実装を上げておきました。

通信部分のコードと解説

以下がAPI通信部分の該当部分を抜き出したコードです。その他の実装も含む全コードはリポジトリ内のこちらを参照ください。また、サンプルのためオプションの渡し方やインターフェースの行儀が多少悪いですがご容赦を。以降に簡単な解説を続けます。

OpenAIChatCompletionAPI.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using System.Net;
using System.Net.Http;
using System.Runtime.CompilerServices;
using Newtonsoft.Json;
using System.IO;
using System.Text;

// https://platform.openai.com/docs/api-reference/chat/create
public class OpenAIChatCompletionAPI
{
    const string API_URL = "https://api.openai.com/v1/chat/completions";
    string apiKey;
    JsonSerializerSettings settings = new JsonSerializerSettings();
    HttpClient httpClient;

    public OpenAIChatCompletionAPI(string apiKey)
    {
        this.httpClient = new HttpClient();
        this.apiKey = apiKey;
        settings.NullValueHandling = NullValueHandling.Ignore;
    }

    public async IAsyncEnumerable<ResponseChunkData> CreateCompletionRequestAsStream(RequestData requestData, [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        if (!requestData.stream.HasValue || !requestData.stream.Value)
        {
            throw new AggregateException("stream must be true.");
        }

        var json = JsonConvert.SerializeObject(requestData, settings);

        using var request = new HttpRequestMessage(HttpMethod.Post, API_URL);
        request.Content = new StringContent(json, System.Text.Encoding.UTF8, "application/json");
        request.Headers.Add("Authorization", $"Bearer {apiKey}");

        using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken);

        if (response.IsSuccessStatusCode)
        {
            using var stream = await response.Content.ReadAsStreamAsync();
            cancellationToken.ThrowIfCancellationRequested();

            using var reader = new StreamReader(stream, Encoding.UTF8);

            string line = null;
            while ((line = await reader.ReadLineAsync()) != null)
            {
                cancellationToken.ThrowIfCancellationRequested();
                if (string.IsNullOrEmpty(line)) continue;
                var chunkString = line.Substring(6); // "data: "
                if (chunkString == "[DONE]") break;
                yield return JsonConvert.DeserializeObject<ResponseChunkData>(chunkString);
            }
        }
        else
        {
            var message = await response.Content.ReadAsStringAsync();
            cancellationToken.ThrowIfCancellationRequested();
            throw new WebException($"request failed. {(int)response.StatusCode} {response.StatusCode}\n{message}");
        }
    }

    [System.Serializable]
    public class RequestData
    {
        public string model = "gpt-3.5-turbo";
        public List<Message> messages;
        public float? temperature = null; // [0.0 - 2.0]
        public float? top_p = null;
        public int? n = null;
        public bool? stream = null;
        public List<string> stop = null;
        public int? max_tokens = null;
        public float? presence_penalty = null;
        public float? frequency_penalty = null;
        public Dictionary<int, int> logit_bias = null;
        public string user = null;
    }

    [System.Serializable]
    public class Message
    {
        public string role;
        public string content;
    }

    [System.Serializable]
    public class ChunkChoice
    {
        public Message delta;
        public int index;
        public object finish_reason;
    }

    [System.Serializable]
    public class ResponseChunkData
    {
        public string id;
        public string @object;
        public int created;
        public string model;
        public List<ChunkChoice> choices;
    }
}

上記コードでは、System.Net.HttpのHttpClientを通信に利用しています。ここでHttpClientのSendAsyncメソッドにて、HttpCompletionOption.ResponseHeadersReadを指定することがポイントです。これにより、ResponseHeaderを受け取った時点で、処理が開始できるようになります。

その後、HttpResponseMessageオブジェクトからStreamを取得し、Stream.ReadAsyncメソッドを利用することでStreamReaderを噛ませReadLineAsyncメソッドを利用することで、送信データの内容を順次（フレームを跨いで）処理できるようになります。

また、ChatGPTのサーバーはstreamオプションを有効にすると以下のようなBodyを順次送ってきます。

data: {"id":"chatcmpl-6uFHcfHo1s0FY4NimSuomBDqy4m85","object":"chat.completion.chunk","created":1678863136,"model":"gpt-3.5-turbo-0301","choices":[{"delta":{"role":"assistant"},"index":0,"finish_reason":null}]}

data: {"id":"chatcmpl-6uFHcfHo1s0FY4NimSuomBDqy4m85","object":"chat.completion.chunk","created":1678863136,"model":"gpt-3.5-turbo-0301","choices":[{"delta":{"content":"は"},"index":0,"finish_reason":null}]}

(...省略...)

data: {"id":"chatcmpl-6uFHcfHo1s0FY4NimSuomBDqy4m85","object":"chat.completion.chunk","created":1678863136,"model":"gpt-3.5-turbo-0301","choices":[{"delta":{"content":"。"},"index":0,"finish_reason":null}]}

data: {"id":"chatcmpl-6uFHcfHo1s0FY4NimSuomBDqy4m85","object":"chat.completion.chunk","created":1678863136,"model":"gpt-3.5-turbo-0301","choices":[{"delta":{},"index":0,"finish_reason":"stop"}]}

data: [DONE]

こちらを、「data: 」の単位でjsonパースすればOKです。データとしては、choices[0].delta.contentに差分の文字列が送られてきます。最終的にdata: [DONE]を受け取ったら全ての受信処理が完了です。なお、roleは最初のチャンクのみ送られてくるみたいです。

さらに、APIクラスで非同期に受け取った差分データ（Chunk）を呼び出し側に送信するするのは、今回はAsyncEnumerableを利用しています。これにより、ChunkごとのJsonを処理した単位でyield returnするため、その単位で呼び出し側では差分文字列を非同期に受け取れるようにしています。

利用側のコード

こちらも、全文のコードはこちらを参照してください。該当部分のみを抜き出すと以下になります。

Chat.cs

...
        ChatMessageView view = null;
        var message = new OpenAIChatCompletionAPI.Message();

        await foreach (var chunk in chatCompletionAPI.CreateCompletionRequestAsStream(requestData, cancellationToken))
        {
            if (view == null) view = CreatedMessage();

            string role = chunk.choices[0].delta.role;

            if (role != null)
            {
                message.role = role;
            }
            message.content += chunk.choices[0].delta.content;

            view.Role = message.role;
            view.Content = message.content;
            scrollRect.verticalNormalizedPosition = 0;
        }

        context.Add(message);
...

ここで、

await foreach (var chunk in chatCompletionAPI.CreateCompletionRequestAsStream())

とすることで、APIクラス側で非同期に順次処理されるChunkデータをforeach文で扱えます。今までの構文を少し拡張するだけで、直感的にイベントストリームのような処理を扱えるAsyncEnumerableすばらしいです。

ここの実装はActionによるコールバックでも、Observableなパターンでもやりやすい方法でよいかもしれません。

応答時間の計測

一括でデータを受信する場合と、今回のstreamオプションを利用する場合の応答時間を計測してみました。以下の文章をuser roleで送信し、初めてユーザーが何かしらの文字を目にするまでの時間を測定しています。

クジラの生態についての記事の構成を考えてください

なお、計測時刻はJST 14:30です。また、これにあたり一応temperatureオプションに0を指定したのですが、微妙に出力される文章に差はあるようでした

stream オプションoff

• Trial1: 00:00:15.6562905sec
• Trial2: 00:00:13.8869909sec
• Trial3: 00:00:16.6130076sec

stream オプションon

• Trial1: 00:00:01.3007185sec
• Trial2: 00:00:01.2786250sec
• Trial3: 00:00:01.2480494sec

結果として、streamオプションを有効にした場合だと、14〜15秒ほど文字列表示開始までの時間が早くなりました。完全な文章でなくても、なるべく早くユーザに結果を提示し始めたい場合はstreamオプションの利用は有効そうです。もちろん全ての文字列を受信しきって表示するまでには同様に時間がかかります。