【Python】FastAPIを使って、AI機能を実装してみよう ～応用編～

こんにちは。

FastAPIの紹介も今回で最後になります。

前回API化して自身のプログラム上で結果を得る方法について解説しました。

課題として実用的な使い方をするためには、関数に渡す引数が複数になることが想定されました。

今回はその課題をクリアにするために、機械学習を使った関数呼び出しに挑戦したいと思います。

以下本日の内容です

それではよろしくお願いします。

複数の引数を渡す方法：リクエストボディについて

FastAPIにはリクエストボディというものがあります。

fastapi.tiangolo.com

一言でまとめると、「FastAPIにデータを送信する際の形式」と覚えてください。

なのでデータを渡すためには、このルールに従う必要があります。

pydanticのBaseModelを使って、Python標準型のアノテーションをします。
docs.pydantic.dev

これは何をやっているかというと、送信するデータの型を限定して、それ以外の間違ったものが来たらブロックする仕組み（バリデーション）を敷いています

なんだかややこしいですが、「決まった受信データしか受け付けないようにしますよー」というぐらいの認識で問題ありません。

またpydanticはJSON形式になりますので、送信側のデータはJSON形式に従う必要があります。

これは最後の方でもまた説明します。

コードの全体像

お急ぎの方のためにも、まずはコード全体を先に紹介します。

API側（reg_api.py）は以下のようになります。

from fastapi import FastAPI
from pydantic import BaseModel

import pandas as pd

from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression


class ModelParam(BaseModel):
    random_state: int
    test_size: float
    is_shuffle: bool
    test_number: int


app = FastAPI()


@app.post("/")
def calc_california_price(params: ModelParam):
    california_housing = fetch_california_housing()
    X = pd.DataFrame(california_housing.data, columns=california_housing.feature_names)
    y = pd.DataFrame(california_housing.target)
    # 訓練データとテストデータに分割する
    print(params.random_state, params.test_size, params.is_shuffle, params.test_size)
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, random_state=params.random_state, test_size=params.test_size, shuffle=params.is_shuffle
    )
    # 回帰モデルの作成
    reg = LinearRegression()

    # 学習
    reg.fit(X_train, y_train)

    model_test_pred = reg.predict(X_test)

    pred_num = params.test_number
    # 上限設定
    if pred_num >= len(X_test):
        pred_num = len(X_test) - 1
    return {"predict": model_test_pred[pred_num][0], "answer": y_test.iloc[pred_num].values[0]}

続いてAPI側を呼び出すコード（app.py）は以下になります。

import requests
import json


def main():
    url = "http://127.0.0.1:8000/"
    params = {"random_state": 1, "test_size": 0.5, "is_shuffle": True, "test_number": 1}
    res = requests.post(url, json.dumps(params))
    print(res.json())


if __name__ == "__main__":
    main()

機械学習の実装（reg_api.py）

今回実装する機械学習は、カリフォルニア住宅価格を予想する線形回帰モデル（LinearRegression）を使用します。

コードはこんな感じ。

def calc_california_price(params: ModelParam):
    california_housing = fetch_california_housing()
    X = pd.DataFrame(california_housing.data, columns=california_housing.feature_names)
    y = pd.DataFrame(california_housing.target)
    # 訓練データとテストデータに分割する
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, random_state=params.random_state, test_size=params.test_size, shuffle=params.is_shuffle
    )
    # 回帰モデルの作成
    reg = LinearRegression()

    # 学習
    reg.fit(X_train, y_train)

    model_test_pred = reg.predict(X_test)

    pred_num = params.test_number
    # 上限設定
    if pred_num >= len(X_test):
        pred_num = len(X_test) - 1
    return {"predict": model_test_pred[pred_num][0], "answer": y_test.iloc[pred_num].values[0]}

scikit-learnからカリフォルニア住宅価格のデータセットを呼び出します。

呼び出したデータセットを学習用とテスト用に分割し、学習させます。

後は学習後のモデルから、予想した住宅価格を出力する流れになります。

見慣れないポイントとして、「params: ModelParam」のところでしょうか。

これについて解説します。

リクエストボディの実装（reg_api.py）

リクエストボディは以下のように実装します。

class ModelParam(BaseModel):
    random_state: int
    test_size: float
    is_shuffle: bool
    test_number: int

pydanticのBaseModelを使って、これから受け取るデータの型を宣言しています。

ここの変数はrandom_stateがデータセットをシャッフルするシード値を想定しています。値はint型であればなんでもよいです。

test_sizeがテストデータの分割率です。0.0~1.0の間のfloatを想定しています。

is_shuffleはデータセットを分割する際、シャッフルを行うかどうかをbool値で受け取ることを想定しています。

test_numberは予想結果を返すためのtestデータの位置を指定することを想定しています。

範囲外を指定する可能性もありますので、上限設定しております。

こうすることで、実装したい機械学習関数にデータを渡すことができるようになります。

後は「params: ModelParam」として機械学習関数の引数に渡します。

この「params」の名前にも意味があります。

paramsについて次に説明します。

API呼び出しコードの実装（app.py）

いよいよ呼び出す側のコードです。

ポイントは以下になります。

    params = {"random_state": 1, "test_size": 0.5, "is_shuffle": True, "test_number": 1}
    res = requests.post(url, json.dumps(params))
    print(res.json())

先ほど出てきた「params」はここに対応します。

渡したいデータを辞書型として作成します。

リクエストボディ内の変数名とも一致させる必要があるので注意しましょう。

そして最後にrequests.postメソッドで、APIを立ち上げたURLにparamsをJSON形式で渡せばOKです。

冒頭でお話したJSON形式がここで登場しました。

FastAPIではこのようリクエストボディを使ってJSON形式でデータの送受信をするのです。

後は結果を受け取るときも「res.json()」とJSON形式で機械学習の結果を受け取るという仕組みです。

実行結果

では最後に実行結果を確認しましょう。

まずは「uvicorn reg_api:app --reload」でAPIを立ち上げます。

立ち上げたら、別のプロンプトを立ち上げて「python app.py」を実行します。

すると以下のような結果が得られるかと思います。

{'predict': 0.9512008432757071, 'answer': 0.707}

predictが予想結果で、answerが正解値になります。

渡すデータを変えることで、結果が様々に変わりますので、良かったら試してみてください。

今回のコードも以下のGitで管理してます。

github.com

「requestbody」フォルダ内を参照ください。

もしよかったらスターつけていただけると励みになります！

まとめ

いかがだったでしょうか？

3回に渡ってFastAPIについて紹介してみました。

機械学習で実装した部分を、皆さんの実務にあった機能に置き換えれば、もう明日から使えるのではないでしょうか。

FastAPIを活用して、より効率よく効果的な開発システムが構築できると良いですね。

今回は以上になります。

ここまでご覧いただき、ありがとうございました！