クリーンな AI コード構築に役立つ Python デコレータ 5 つの強力な例
本記事は、AI・機械学習システムの開発においてコードの可読性と保守性を向上させるために、並行処理制限や構造化ログ出力など、実用的な Python デコレータ 5 つを具体例と共に紹介している。
キーポイント
並行処理の制限（Concurrency Limiter）
LLM の無料枠やレート制限に陥らないよう、セマフォを用いて非同期関数の同時実行数を制御するデコレータの実装例を示している。
構造化機械学習ロギング（Structured Logger）
生産環境でのデバッグを容易にするため、標準的な print 文に代わり、検索可能な JSON 形式で実行情報やエラーを記録するデコレータの手法を提案している。
コードの分離とベストプラクティス
モデルロジックとテスト・検証などのボイラープレート処理を分離し、functools.wraps などを活用してクリーンな AI コードを構築するアプローチを解説している。
推論時のデータ一貫性確保
本デコレータは、モデルをノートブックから生産環境へ移行する際、生データをトレーニング時と同じ変換プロセスで処理し、特徴量生成の整合性を自動保証します。
FastAPI 環境での簡易実装
FastAPI エンドポイントなどの軽量プロダクション環境において、手動でデータ前処理を行う手間を省き、デコレータ内で自動的に 'is_weekend' のような新特徴量を付与できます。
実験とハイパーパラメータ調整の信頼性向上
乱数シードを固定することで、モデル精度の変化が新しい設定によるものか単なる初期化の偶然によるものかを明確に区別し、A/B テストの結果をより信頼性の高いものにします。
外部サービス障害時のフォールバック機能
接続タイムアウトや API 使用制限などのエラーが発生した際、例外をスローするのではなく事前に定義されたモックデータを返すことで、アプリケーションの停止を防ぎます。
影響分析・編集コメントを表示
影響分析
この記事は、AI エンジニアが直面する実務的な課題（レート制限、ログ管理の複雑さ）に対して、標準ライブラリを活用した即座に適用可能な解決策を提供しています。特に大規模な LLM アプリケーション開発において、コードの品質を維持し、運用コストを下げるための具体的なパターンを示しているため、現場の開発生産性向上に直結する価値があります。
編集コメント
一般的な Python のベストプラクティスを紹介する記事ですが、AI/ML システム特有の課題（レート制限や構造化ログ）に焦点を当てている点が実務者にとって有益です。
image**
画像提供：編集者
# イントロダクション
Python デコレータは、AI および機械学習システム開発に関わるプロジェクトにおいて非常に有用です。これらは、モデリングやデータパイプラインといった主要なロジックを、テストや検証、タイミング計測、ログ出力などの他の定型業務から分離する点で優れています。
本記事では、開発者の経験に基づき、AI コードをよりクリーンにするのに効果的であることが実証された、特に有用な Python デコレータ 5 つを紹介します。
以下のコード例は、Python の標準ライブラリやベストプラクティス（例えば functools.wraps など）に基づくシンプルで基礎的なロジックを含んでいます。主な目的は各デコレータの使用方法を示すことにあるため、読者にはデコレータのロジックを自身の AI コーディングプロジェクトに適応させることのみを考慮すれば十分です。
# 1. 並行処理制限器
**
サードパーティ製の大規模言語モデル（LLM）を利用する際によく遭遇する、しばしば厄介な無料枠の制限に対処する際に非常に有用なデコレータです。非同期リクエストを過剰に送信した結果このような制限に達した場合、このパターンはスロットリング機構を導入し、これらの呼び出しをより安全に行います。セマフォを通じて、非同期関数の実行回数を制限することができます：
import asyncio
from functools import wraps
def limit_concurrency(limit=5):
sem = asyncio.Semaphore(limit)
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
async with sem:
return await func(*args, **kwargs)
return wrapper
return decorator
Application
@limit_concurrency(5)
async def fetch_llm_batch(prompt):
return await async_api_client.generate(prompt)
# 2. Structured Machine Learning Logger (構造化機械学習ロガー)
複雑なソフトウェア、特に機械学習システムを制御するものにおいて、標準的な print() 文が埋もれてしまうのは驚くべきことではありません。特に本番環境にデプロイされた後はなおさらです。
以下のロギングデコレータを使用することで、「実行」と「エラー」をキャッチし、迅速なデバッグのために検索しやすい構造化 JSON ログとして整形することが可能になります。以下のコード例は、ニューラルネットワークベースのモデルにおけるトレーニングエポックを定義する関数を装飾するためのテンプレートとして使用できます:
import logging, json, time
from functools import wraps
def json_log(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
try:
res = func(*args, **kwargs)
logging.info(json.dumps({"step": func.__name__, "status": "success", "time": time.time() - start}))
return res
except Exception as e:
logging.error(json.dumps({"step": func.__name__, "error": str(e)}))
raise
return wrapper
アプリケーション
@json_log
def train_epoch(model, training_data):
return model.fit(training_data)
# 3. フィーチャーインジェクター
モデルのデプロイメントおよび推論フェーズで特に有用なデコレーターが登場します！例えば、機械学習モデルをノートブックから軽量な本番環境（例：FastAPI エンドポイント）へ移行する場合を考えてみましょう。エンドユーザーからの生データが、元のトレーニングデータと同じ変換プロセスを経ることを手動で保証するのは、場合によっては少し面倒になることがあります。このフィーチャーインジェクターは、生データから特徴量が生成される方法の一貫性を、すべて内部（アンダーザフード）で保証するのに役立ちます。
これはデプロイメントおよび推論フェーズにおいて非常に有用です。モデルを Jupyter ノートブックから本番環境へ移行する際、大きな頭痛の種となるのが、生データとして流入するユーザーデータをトレーニングデータと同じ変換処理に適用することです。このデコレーターは、データがモデルに到達する前に、特徴量が内部で一貫して生成されることを保証します。
以下の例では、既存のデータフレーム内の日付列に土曜日または日付に関連する日付が含まれているかどうかに基づいて、「is_weekend」という名前のフィーチャーを追加するプロセスを簡略化しています：
from functools import wraps
def add_weekend_feature(func):
@wraps(func)
def wrapper(df, *args, **kwargs):
df = df.copy() # Prevents Pandas mutation warnings
df['is_weekend'] = df['date'].dt.dayofweek.isin([5, 6]).astype(int)
return func(df, *args, **kwargs)
return wrapper
Application
@add_weekend_feature
def process_data(df):
# 'is_weekend' is guaranteed to exist here
return df.dropna()
# 4. Deterministic Seed Setter
**
このデコレータは、AI/機械学習のライフサイクルにおける特定の 2 つの段階、すなわち実験とハイパーパラメータ調整において特に注目すべきものです。これらのプロセスでは通常、モデルの学習率などの重要なハイパーパラメータを調整する際に乱数シードを使用します。例えば、ある値を調整した直後に突然モデルの精度が低下した場合、このパフォーマンス低下の原因が新しいハイパーパラメータ設定によるものなのか、それとも単に重みのランダム初期化が悪かったことによるものなのかを知る必要があるかもしれません。シードを固定することで変数を分離し、A/B テストなどの結果の信頼性を高めることができます。
import random, numpy as np
from functools import wraps
def lock_seed(seed=42):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
random.seed(seed)
np.random.seed(seed)
return func(*args, **kwargs)
return wrapper
return decorator
Application
@lock_seed(42)
def initialize_weights():
return np.random.randn(10, 10)
# 5. Dev-Mode Fallback
ローカル開発環境や CI/CD テストにおいて特に重要な、命を救うデコレータです。例えば、LLM の上にアプリケーション層を構築している場合（例：検索拡張生成 (RAG) システム）を考えてみましょう。デコレータ付きの関数が接続タイムアウトや API 利用制限などの外部要因によって失敗した場合、例外をスローするのではなく、このデコレータがエラーを捕捉し、事前に定義された「モックテストデータ」のセットを返します。
なぜ命を救うのか？それは、このメカニズムにより、外部サービスが一時的に失敗してもアプリケーションが完全に停止しないことを保証できるからです。
from functools import wraps
def fallback_mock(mock_data):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception: # タイムアウトやレート制限を捕捉
return mock_data
return wrapper
return decorator
アプリケーション
@fallback_mock(mock_data=[0.01, -0.05, 0.02])
def get_text_embeddings(text):
return external_api.embed(text)
まとめ
本記事では、構造化された検索可能なログ記録から、データサンプリングやテストなどの側面における制御されたランダムシードに至るまで、さまざまな特定の状況において AI および機械学習のコードをよりクリーンにするのに役立つ 5 つの効果的な Python デコレータを検証しました。
Iván Palomares Carrascosa は、AI、機械学習、深層学習、大規模言語モデル（LLM）におけるリーダー、作家、スピーカー、そしてアドバイザーです。彼は、現実世界で AI を活用する方法を他者に指導・訓練しています。
原文を表示
**
Image by Editor
# Introduction
Python decorators can be incredibly useful in projects involving AI and machine learning system development**. They excel at separating key logic like modeling and data pipelines from other boilerplate tasks, like testing and validation, timing, logging, and so on.
This article outlines five particularly useful Python decorators that, based on developers' experience, have proven themselves effective at making AI code cleaner.
The code examples below include simple, underlying logic based on Python standard libraries and best practices, e.g. functools.wraps. Their primary goal is to illustrate the use of each specific decorator, so that you only need to worry about adapting the decorator's logic to your AI coding project.
# 1. Concurrency Limiter
**
A very useful decorator when dealing with (often annoying) free-tier limits in the use of third-party large language models (LLMs). When hitting such limits as a result of sending too many asynchronous requests, this pattern introduces a throttling mechanism to make these calls safer. Through semaphores, the number of times an asynchronous function executes is limited:
import asyncio
from functools import wraps
def limit_concurrency(limit=5):
sem = asyncio.Semaphore(limit)
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
async with sem:
return await func(*args, **kwargs)
return wrapper
return decorator
# Application
@limit_concurrency(5)
async def fetch_llm_batch(prompt):
return await async_api_client.generate(prompt)# 2. Structured Machine Learning Logger
It is no surprise that in complex software like that governing machine learning systems, standard print() statements get easily lost, especially once deployed in production.
Through the following logging decorator, it is possible to "catch" executions and errors and format them into structured JSON logs that are easily searchable for rapid debugging. The code example below can be used as a template to decorate, for instance, a function that defines a training epoch in a neural network-based model:
import logging, json, time
from functools import wraps
def json_log(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
try:
res = func(*args, **kwargs)
logging.info(json.dumps({"step": func.__name__, "status": "success", "time": time.time() - start}))
return res
except Exception as e:
logging.error(json.dumps({"step": func.__name__, "error": str(e)}))
raise
return wrapper
# Application
@json_log
def train_epoch(model, training_data):
return model.fit(training_data)# 3. Feature Injector
Enter a particularly useful decorator during the model deployment and inference stages! Say you are moving your machine learning model from a notebook into a lightweight production environment, e.g. using a FastAPI** endpoint. Manually ensuring that raw incoming data from end users undergoes the same transformations as the original training data can sometimes become a bit of a pain. The feature injector helps ensure consistency in the way features are generated from raw data, all under the hood.
This is highly useful during the deployment and inference phase. When moving a model from a Jupyter notebook into a production environment, a major headache is ensuring the raw incoming user data gets the same transformations as your training data. This decorator guarantees those features are generated consistently under the hood before the data ever reaches your model.
The example below simplifies the process of adding a feature called 'is_weekend', based on whether a date column in an existing dataframe contains a date associated with a Saturday or Sunday:
from functools import wraps
def add_weekend_feature(func):
@wraps(func)
def wrapper(df, *args, **kwargs):
df = df.copy() # Prevents Pandas mutation warnings
df['is_weekend'] = df['date'].dt.dayofweek.isin([5, 6]).astype(int)
return func(df, *args, **kwargs)
return wrapper
# Application
@add_weekend_feature
def process_data(df):
# 'is_weekend' is guaranteed to exist here
return df.dropna()# 4. Deterministic Seed Setter
**
This one stands out for two specific stages of the AI/machine learning lifecycle: experimentation and hyperparameter tuning. These processes typically entail the use of a random seed as part of adjusting key hyperparameters like a model's learning rate. Say you just adjusted its value, and suddenly, the model accuracy drops. In a situation like this, you may need to know whether the cause behind this performance drop is the new hyperparameter setting or simply a bad random initialization of weights. By locking the seed, we isolate variables, thereby making the results of tests like A/B more reliable.
import random, numpy as np
from functools import wraps
def lock_seed(seed=42):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
random.seed(seed)
np.random.seed(seed)
return func(*args, **kwargs)
return wrapper
return decorator
# Application
@lock_seed(42)
def initialize_weights():
return np.random.randn(10, 10)# 5. Dev-Mode Fallback
A lifesaving decorator, particularly in local development environments and CI/CD testing. Say you are building an application layer on top of an LLM — for instance, a retrieval-augmented generation (RAG) system. If a decorated function fails due to external factors, like connection timeouts or API usage limits, instead of throwing an exception, the error is intercepted by this decorator and a predefined set of "mock test data" is returned.
Why a lifesaver? Because this mechanism can ensure your application does not completely stop if an external service temporarily fails.
from functools import wraps
def fallback_mock(mock_data):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception: # Catches timeouts and rate limits
return mock_data
return wrapper
return decorator
# Application
@fallback_mock(mock_data=[0.01, -0.05, 0.02])
def get_text_embeddings(text):
return external_api.embed(text)# Wrapping Up
This article examined five effective Python decorators that will help make your AI and machine learning code cleaner across a variety of specific situations: from structured, easy-to-search logging to controlled random seeding for aspects like data sampling, testing, and more.
Iván Palomares Carrascosa** is a leader, writer, speaker, and adviser in AI, machine learning, deep learning & LLMs. He trains and guides others in harnessing AI in the real world.
関連記事
今日のまとめ
AI日報で今日の重要ニュースをまとめ読み