こんにちは！ (株)Qualiteg プロダクト開発部 です！

本稿では、 ChatStream を FastAPI の Web API として実装する方法についてご説明いたします！

エンドポイントの実装

/chat_stream という URL パスに、ストリーミングチャット用のWebエンドポイントをつくるには
以下のように handle_chat_stream_request を呼び出します。

これだけで、ユーザーからのリクエストは　文章生成の同時実行数を制御したストリーミングチャットの実装は完了です

@app.post("/chat_stream")
async def stream_api(request: Request):
    # handling FastAPI/Starlette's Request
    response = await chat_stream.handle_chat_stream_request(request)
    return response

メッセージインターセプト

FastAPI/Starlette を利用している場合、エンドポイントで await request.body() や await request.json() を実行すると、
リクエストストリームを消費(consume)してしまうため、 ChatStream にリクエストを委譲する前にリクエストをインターセプトをする場合は以下のように実装します

import json
from fastapi import FastAPI, Request

@app.post("/chat_stream")
async def stream_api(request: Request):

    # Request を インターセプトする場合
    request_body = await request.body()
    data = json.loads(request_body)
    
    user_input = data["user_input"]
    regenerate = data["regenerate"]

    print(f"user_input:{user_input} regenerate:{regenerate}")
    
    # インターセプトした場合は `request_body` を指定する
    response = await chat_stream.handle_chat_stream_request(request, request_body)

    return response

チャットストリームの送出完了のコールバックを受け取る

ChatStream では、ストリーミングレスポンスを行うため、エンドポイントで return reponse を行ったタイミングが文章生成処理の終了ではありません。

そこで、文章生成の完了のタイミングをキャッチしたい場合、
エンドポイントの実装で、 handle_chat_stream_request の引数 callback にコールバック関数を指定します。

文章生成が完了すると、指定したコールバック関数が呼び出されます

@app.post("/chat_stream")
async def stream_api(request: Request):

    def callback_func(request, message):
        # 文章生成が終了したとき
        
        # ここでは、セッションに格納されている ChatPrompt を取得して、これまでの会話履歴をもとにプロンプトを生成する例
        session_mgr = getattr(request.state, "session", None)
        session = session_mgr.get_session()
        chat_prompt = session.get("chat_prompt")
        print(chat_prompt.create_prompt())

    pass

    response = await chat_stream.handle_chat_stream_request(request, callback=callback_func)

    return response

文章生成終了時のコールバック関数のパラメータ message の取り得る値と意味