LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@KENTAROSZK

PydanticのBaseSettingsで機械学習の実験設定をスッキリ管理する

Posted at

概要

  • pydanticで環境変数を自動でパースして、読み込むことができることを知ったので、試してみた。

機械学習の実験で、検証用の少量データと本番用の全量データで処理を分けたいとき、どうしていますか？

def preprocess(is_validation, **kwargs):
    if is_validation:
        # 検証用の設定
    else:
        # 本番用の設定

のような分岐がコードのあちこちに散らばって、管理が煩雑になっていませんか？

この記事では、pydantic-settings を使って、そうした設定値を外部ファイル（.env）に分離し、型安全かつスマートに管理する方法を紹介する。この方法を導入することで、設定の切り替え忘れを防ぎ、コードの可読性をに向上させることができる（はず）。

簡単な使い方

まずは、pydanticのBaseSettingsの使い方や動きを確認する。
（機械学習の実験設定での使い方をさくっと確認したい人はこの章をスキップしてOKです。）

▼ notebookで環境変数をテキトーに作成する

%%bash
cat << 'EOF' > .env
DATABASE_URL=postgresql://localhost:5432/mydb
SECRET_KEY=your-secret-key
DEBUG=true
MAX_CONNECTIONS=100
EOF

▼ 作成結果を確認する

!cat .env

# ----- 出力 -----
# DATABASE_URL=postgresql://localhost:5432/mydb
# SECRET_KEY=your-secret-key
# DEBUG=true
# MAX_CONNECTIONS=100

▼ 環境変数をpydanticで読み込む

# pydanticを使って環境変数を自動で取得する
from pydantic_settings import BaseSettings
from dotenv import load_dotenv
import os

# .envを環境変数化する
load_dotenv(".env")

# フィールド名を大文字に変換したものを環境変数名としてマッピングされる
### DATABASE_URL(環境変数) <--> database_url(pydanticのフィールド名)
class Settings(BaseSettings):
    database_url: str
    secret_key: str
    debug: bool

settings = Settings()

print(settings.database_url)
print(settings.secret_key)
print(settings.debug)

# ----- 出力 -----
# postgresql://localhost:5432/mydb
# your-secret-key
# True

注目ポイント

  • DEBUG=true という文字列が、debug: bool という型定義によって自動的に True というブール値に変換されている！
  • 同様に、数値も自動で int や float にパースしてくれる。

機械学習での利用の仕方

機械学習の場合、モデリングやデータの前処理で時間がかかることが多い。そのため、デバッグや検証でコードを実行させるときは、一部のデータだけを使ってコーディングを進めることがある。

機械学習の実験で、検証用の少量データと本番用の全量データで処理を分けたいとき、どうしているかといえば、

def preprocess(is_validation, **kwargs):
    if is_validation:
        # 検証用の設定
    else:
        # 本番用の設定

のような分岐がコードをあちこちに作成しまうことがある。

前処理の関数に渡す引数としてis_validation定義して、検証時と本番時とでデータ量を変更しているが、これらを前処理、モデル学習、モデル検証、モデル指標の可視化、etc…など機械学習モデリングに関する処理全てに作るのは面倒臭いし、カッコ悪い。

また、notebookなどを複数作成して、いろいろなパターンで実験している時などは、この実験条件の変数を全てのnotebookで修正する必要があるなど、実験条件を変更するときの負荷が高い。

そこで、検証条件と本番条件とのそれぞれの条件を環境変数として作成し、その環境変数をプログラムの中から読み出す形式にすることで、この面倒臭さから解放されようと考えた。

具体実装

まず、検証と本番用とでそれぞれの環境変数を作成する

# .env.validation (検証用)
%%bash
cat << 'EOF' > .env.validation
RUN_MODE=validation
MAX_RECORDS=1000
SAMPLE_RATIO=0.01
DATA_PATH=/data/sample
OUTPUT_PATH=/output/validation
BATCH_SIZE=16
EPOCHS=2
NUM_WORKERS=2
USE_GPU=false
EOF

# .env.production (本番用)
%%bash
cat << 'EOF' > .env.production
RUN_MODE=production
# MAX_RECORDS=  # 指定しないので全量
# SAMPLE_RATIO=  # 指定しないので全量
DATA_PATH=/data/full
OUTPUT_PATH=/output/production
BATCH_SIZE=128
EPOCHS=100
NUM_WORKERS=8
USE_GPU=true
EOF

# 各環境変数をちゃんと作成できているか確認する
!cat .env.validation
!echo "-----"
!cat .env.production

# ---- 実行結果 -----
# RUN_MODE=validation
# MAX_RECORDS=1000
# SAMPLE_RATIO=0.01
# DATA_PATH=/data/sample
# OUTPUT_PATH=/output/validation
# BATCH_SIZE=16
# EPOCHS=2
# NUM_WORKERS=2
# USE_GPU=false
-----
# RUN_MODE=production
# # MAX_RECORDS=  # 指定しないので全量
# # SAMPLE_RATIO=  # 指定しないので全量
# DATA_PATH=/data/full
# OUTPUT_PATH=/output/production
# BATCH_SIZE=128
# EPOCHS=100
# NUM_WORKERS=8
# USE_GPU=true

ちゃんと設定されていることを確認できた。


▼ pydanticを使って環境変数を読み込むための処理を作る

from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # 実行モード
    run_mode: str
    
    # データセット制限
    max_records: Optional[int] = None  # Noneなら全量
    sample_ratio: Optional[float] = None  # 0.1なら10%サンプリング
    
    # データソース設定
    data_path: str
    output_path: str
    
    # モデル設定
    batch_size: int = 32
    epochs: int = 10
    learning_rate: float = 0.001
    
    # パフォーマンス設定
    num_workers: int = 4
    use_gpu: bool = True
    
    class Config:
        env_file = ".env" # これを設定することで、自動で`.env`を読み込みしてくれる!!

▼ 検証用の条件で実行させる（検証用の環境変数を読み込む） 
# 検証用を実行する
!cp .env.validation .env

# 環境変数から設定を読み込み
settings = Settings()
print(f"{settings.run_mode=}")
print(f"{settings.max_records=}")
print(f"{settings.sample_ratio=}")
print(f"{settings.data_path=}")
print(f"{settings.output_path=}")

# ----- 実行結果 -----
# settings.run_mode='validation'
# settings.max_records=1000
# settings.sample_ratio=0.01
# settings.data_path='/data/sample'
# settings.output_path='/output/validation'

▼ 本番用の条件で実行させる（本番用の環境変数を読み込む） 
# 本番用を実行する
!cp .env.production .env

# 環境変数から設定を読み込み
settings = Settings()
print(f"{settings.run_mode=}")
print(f"{settings.max_records=}")
print(f"{settings.sample_ratio=}")
print(f"{settings.data_path=}")
print(f"{settings.output_path=}")

# ----- 実行結果 -----
# settings.run_mode='production'
# settings.max_records=None
# settings.sample_ratio=None
# settings.data_path='/data/full'
# settings.output_path='/output/production'

検証用で実行したときは、検証用の変数が読み込まれ、本番用の場合は、本番用の変数が読み込まれていることを確認できた。

注目ポイント

本番用の .env.production では MAX_RECORDS をコメントアウトした。このようにキーが存在しない、または値が空の場合、max_records: Optional[int] = None のように定義しておくと、自動的に None が代入される。これにより、設定の有無を柔軟に扱うことができる。

あとは、機械学習の処理を以下のように作成すれば、読み込む環境変数を変えるだけで、検証用と本番用とですぐに条件を切り替えることができる。

import pandas as pd

def load_data(settings: Settings) -> pd.DataFrame:
    df = pd.read_csv(settings.data_path)
    
    # データ制限の適用
    if settings.run_mode == 'validation':
        if settings.max_records:
            df = df.head(settings.max_records)
            print(f"Limited to {settings.max_records} records")
        
        if settings.sample_ratio:
            df = df.sample(frac=settings.sample_ratio)
            print(f"Sampled {settings.sample_ratio*100}% of data")
    
    print(f"Final dataset size: {len(df)}")
    return df

def preprocess_data(df: pd.DataFrame, settings: Settings) -> pd.DataFrame:
    if settings.run_mode == 'validation':
        print("Running lightweight preprocessing for validation")
        return df.dropna()
    else:
        print("Running full preprocessing for production")
        # 重い前処理をここに
        return df

def train_model(df: pd.DataFrame, settings: Settings):
    print(f"Training with batch_size={settings.batch_size}, epochs={settings.epochs}")
    
    # 実際の学習処理
    # model = create_model()
    # model.fit(df, batch_size=settings.batch_size, epochs=settings.epochs)
    
    # 結果保存
    print(f"Saving results to: {settings.output_path}")

df = load_data(settings) # データ読み込み
df_processed = preprocess_data(df, settings) # 前処理
train_model(df_processed, settings) # モデル学習

まとめ

  • pydanticを使って環境変数を読み込む方法を調べて実験した。
  • それを応用して機械学習の実験管理を便利にする方法を（擬似）コードで作成した。
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?