LitServeで推論APIを構築｜初心者でも使えるAIツール完全ガイド
あなたは、完璧に訓練したAIモデルを本番環境にデプロイする際、思わぬ壁にぶつかった経験はありませんか？
モデルの精度は申し分ない。テストも完璧に通過した。でも、いざ本番環境で動かそうとすると、FastAPIの設定に悩まされ、複雑なMLOpsパイプラインと格闘し、YAMLファイルの山に埋もれてしまう……。
そんな課題を一気に解決してくれるのが、LitServeです。
LitServeは、PyTorch Lightningの開発元であるLightning AIが提供する、柔軟で超高速なAIモデルサーバーエンジンです。FastAPIと比較して最低でも2倍以上の速度を実現しながら、複雑な設定は一切不要。LLM、画像認識、音声処理、さらにはマルチモーダルシステムまで、あらゆるAIモデルをシンプルに、そして高速にデプロイできます。
✅ こんな方におすすめ！
- AIモデルを初めて本番環境にデプロイする開発者
- FastAPIの複雑な設定にうんざりしているエンジニア
- GPU環境で最大限のパフォーマンスを引き出したい方
- チャットボットやリアルタイムAIアプリを構築したい方
この記事では、LitServeの基本から実践的な使い方まで、コード例とともに詳しく解説していきます！
目次
- 1 1. はじめに：AIモデルデプロイの課題
- 2 2. LitServeとは？
- 3 3. LitServeを選ぶべき理由
- 4 4. 基本コンセプトの理解
- 5 5. インストール方法
- 6 6. 機能①：基本的なサーバーセットアップ
- 7 7. 機能②：複数モデルを使った推論エンジン
- 8 8. 機能③：動的バッチ処理で処理速度を劇的に向上
- 9 9. 機能④：リアルタイムストリーミング機能
- 10 10. 機能⑤：GPU自動スケーリング
- 11 11. 機能⑥：AIエージェントの構築
- 12 12. 実践例：Hugging Faceモデルのデプロイ
- 13 13. 応用機能の紹介
- 14 14. パフォーマンスベンチマーク
- 15 15. デプロイ方法
- 16 16. ベストプラクティス
- 17 17. 活用シーン
- 18 18. まとめ
1. はじめに：AIモデルデプロイの課題
AIモデルの訓練は、実は本番デプロイに比べれば簡単な作業かもしれません。
何週間もかけてモデルを調整し、テストデータで素晴らしい精度を達成した後、いよいよ本番環境へ……そこで直面するのが、こんな課題です：
- FastAPIの膨大なボイラープレートコード
- 複雑なMLOpsパイプラインの設定
- 終わりのないYAMLファイルとの戦い
- スケーリングに関する頭痛のタネ
でも、もっと簡単な方法があるとしたら？
LitServeは、まさにそんな「もっと簡単な方法」を実現するツールです。PyTorch Lightningの開発チームであるLightning AIが作った柔軟で超高速なサーバーエンジンで、モデルサービングの複雑さを排除しながら、FastAPIより最低でも2倍以上高速な処理を実現します。しかも、AI処理に特化した最適化が施されています。
LLM、コンピュータビジョンモデル、音声処理パイプライン、複雑なマルチモーダルシステムなど、どんなモデルをデプロイする場合でも、LitServeなら低レベルな細かい制御と、高レベルなシンプルさの両方を手に入れることができます。
2. LitServeとは？
LitServeは、FastAPIをベースに構築された、AIモデル専用の柔軟なサーバーエンジンです。vLLMのような特定のモデルタイプにしか対応していない硬直的なツールとは異なり、LitServeは推論処理の方法を完全にコントロールできます。
FastAPIとの主な違い
- AI処理に最適化されたパフォーマンス：
AIワークロードに特化したマルチワーカー処理により、FastAPIより高速 - 組み込みバッチ処理：
複雑な設定なしで、動的なリクエストバッチ処理が可能 - GPU自動スケーリング：
複数のGPUに自動的にスケールアウト - ストリーミング対応：
リアルタイムアプリケーション向けのネイティブストリーミング機能 - 設定ファイル不要：
YAMLファイルや複雑なMLOpsセットアップは一切不要
設計思想：完全な制御 vs 硬直的な抽象化
vLLMのようなサーバーエンジンが特定のモデルタイプに硬直的な抽象化を提供するのに対し、LitServeはあらゆるモデルタイプ（画像、音声、テキスト、マルチモーダル）に対応し、推論の動作を正確に定義できます。バッチ処理、キャッシング、ストリーミング、ルーティング、複数モデルのオーケストレーションまで、すべてをコントロールできるのです。
FastAPIでもAIモデルをデプロイできるのに、わざわざLitServeを使う必要があるんですか？
FastAPIでもデプロイは可能ですが、LitServeはAI処理に特化した最適化が施されているため、FastAPIより最低でも2倍以上高速です。さらに、バッチ処理やGPU自動スケーリングといったAI特有の機能が標準で組み込まれているため、複雑な設定なしで高性能なAPIを構築できます。YAMLファイルやMLOpsの複雑なセットアップも不要なので、開発に集中できます。
3. LitServeを選ぶべき理由
🚀 圧倒的なパフォーマンス
AI処理に最適化されたマルチワーカー処理により、FastAPIと比較して最低でも2倍の高速化を実現。バッチ処理などの機能を活用すれば、さらに2倍を大きく超えるパフォーマンス向上も可能です。
🎯 驚くほどシンプル
MLOpsの複雑な接続処理や、YAMLファイルの設定は不要。マルチモデルシステム、AIエージェント、RAG（Retrieval-Augmented Generation）など、完全なAIシステムをシンプルなPythonコードだけで構築できます。
🔧 圧倒的な柔軟性
あらゆるモデルタイプに対応：
- LLM（大規模言語モデル）
- 画像認識モデル
- 音声処理
- マルチモーダルシステム
- 従来の機械学習モデル
- カスタムパイプライン
⚡ 標準搭載の強力機能
- 動的バッチ処理：設定可能なタイムアウト付き
- レスポンスストリーミング：リアルタイムアプリケーション対応
- GPU自動スケーリング：複数デバイスに自動分散
- OpenAPI準拠：最初から対応済み
- OpenAI API互換性：既存ツールとの統合が簡単
🌐 柔軟なデプロイオプション
どこにでもセルフホスティングできるほか、Lightning AIのクラウドサービス（Lightning Cloud）にワンクリックでデプロイすれば、自動スケーリング、モニタリング、99.995%の稼働率を実現できます。
4. 基本コンセプトの理解
LitServeは、2つの主要なコンセプトで構成されています。
LitAPI｜サービングロジックを定義
LitAPIは、モデルがリクエストをどのように処理するかを定義する抽象基底クラスです。主なメソッドは以下の通り：
- setup() – 初回起動時の一度だけ実行される処理（モデルの読み込み、データベース接続など）
- decode_request() – HTTPリクエストをモデルの入力形式に変換
- predict() – モデルを使った推論処理を実行
- encode_response() – モデルの出力をHTTPレスポンスに変換
LitServer｜インフラを管理
LitServerは、すべてのインフラ関連の処理を担当します：
- 複数ワーカーへのリクエスト分散
- 入力の動的バッチ処理
- レスポンスのストリーミング
- GPUリソース管理
- ロードバランシングと自動スケーリング
5. インストール方法
LitServeの導入は驚くほど簡単です！
# LitServeのインストール
pip install litserve
# 追加機能が必要な場合
pip install litserve[extra]たったこれだけ！複雑なセットアップも、設定ファイルも、最初にDockerイメージをビルドする必要もありません。
6. 機能①：基本的なサーバーセットアップ
まずは、シンプルな例でコアコンセプトを理解しましょう。
import litserve as ls
class SimpleLitAPI(ls.LitAPI):
def setup(self, device):
# サーバー起動時に一度だけ呼ばれる
# モデルの読み込み、データベース接続などを行う
self.model1 = lambda x: x**2 # シンプルな数学モデル
self.model2 = lambda x: x**3 # 別の数学モデル
print(f"モデルが以下のデバイスに読み込まれました: {device}")
def decode_request(self, request):
# HTTPリクエストをモデル入力に変換
# リクエストは自動的にJSONとして解析される
return request["input"]
def predict(self, x):
# 推論を実行 - 複数モデルを使用可能
squared = self.model1(x)
cubed = self.model2(x)
output = squared + cubed
return {"output": output}
def encode_response(self, output):
# モデル出力をHTTPレスポンスに変換
return {"result": output["output"]}
if __name__ == "__main__":
# APIインスタンスを作成
api = SimpleLitAPI()
# デバイスを自動検出してサーバーを作成
server = ls.LitServer(api, accelerator="auto", port=8000)
# サーバーを起動
server.run()
実行とテスト方法
上記のコードをserver.pyとして保存し、実行します：
python server.py
curlコマンドでテストしてみましょう！
curl -X POST http://127.0.0.1:8000/predict \
-H "Content-Type: application/json" \
-d '{"input": 4.0}'
# レスポンス: {"result": 80.0} # 4^2 + 4^3 = 16 + 64 = 80
✅ たったこれだけで動作します！
7. 機能②：複数モデルを使った推論エンジン
LitServeの強力な機能の一つは、複数のモデルを組み合わせて使う複合AIシステムを簡単に構築できることです。ここでは、テキスト処理と感情分析を組み合わせた実用的な例を紹介します。
import litserve as ls
import torch
from transformers import pipeline
class MultiModelAPI(ls.LitAPI):
def setup(self, device):
# 異なるタスク用の複数モデルを読み込む
self.text_classifier = pipeline(
"sentiment-analysis",
model="distilbert-base-uncased-finetuned-sst-2-english",
device=0 if device == "cuda" else -1
)
self.summarizer = pipeline(
"summarization",
model="facebook/bart-large-cnn",
device=0 if device == "cuda" else -1
)
print("マルチモデル推論エンジンの準備が完了しました！")
def decode_request(self, request):
return {
"text": request["text"],
"task": request.get("task", "both") # "sentiment", "summary", "both"のいずれか
}
def predict(self, inputs):
text = inputs["text"]
task = inputs["task"]
results = {}
# リクエストに基づいて条件付きでモデルを実行
if task in ["sentiment", "both"]:
sentiment = self.text_classifier(text)[0]
results["sentiment"] = {
"label": sentiment["label"],
"confidence": round(sentiment["score"], 4)
}
if task in ["summary", "both"]:
# テキストが十分に長い場合のみ要約を実行
if len(text.split()) > 50:
summary = self.summarizer(text, max_length=130, min_length=30)[0]
results["summary"] = summary["summary_text"]
else:
results["summary"] = "テキストが短すぎて要約できません"
return results
def encode_response(self, output):
return {
"status": "成功",
"results": output,
"model_info": "DistilBERT + BART"
}
if __name__ == "__main__":
api = MultiModelAPI()
server = ls.LitServer(api, accelerator="auto")
server.run(port=8000)
マルチモデルAPIのテスト
# 感情分析のみをテスト
curl -X POST http://127.0.0.1:8000/predict \
-H "Content-Type: application/json" \
-d '{
"text": "LitServeを使ったAIプロジェクトが大好きです！",
"task": "sentiment"
}'
# 感情分析と要約の両方をテスト
curl -X POST http://127.0.0.1:8000/predict \
-H "Content-Type: application/json" \
-d '{
"text": "長い記事のテキストをここに...",
"task": "both"
}'
💡 ポイント：リクエストに応じて異なるモデルを使い分けることで、柔軟で効率的なAPIを構築できます。
8. 機能③：動的バッチ処理で処理速度を劇的に向上
バッチ処理は、特にGPU推論において、スループット最適化の鍵となります。LitServeは、複数のリクエストを自動的にグループ化して効率的に処理する動的バッチ処理を提供します。
import litserve as ls
import torch
import numpy as np
from typing import List
class BatchedImageClassifier(ls.LitAPI):
def setup(self, device):
# シンプルな画像分類モデルを読み込む
# 実際にはResNet、EfficientNetなどを使用します
self.device = device
self.model = torch.nn.Sequential(
torch.nn.Linear(224*224*3, 512),
torch.nn.ReLU(),
torch.nn.Linear(512, 10) # 10クラス分類
)
if device == "cuda":
self.model = self.model.cuda()
print(f"画像分類器が{device}に読み込まれました")
def decode_request(self, request):
# base64画像または画像配列をテンソルに変換
image_array = np.array(request["image"]) # 前処理済みと仮定
return torch.tensor(image_array, dtype=torch.float32)
def batch(self, inputs: List[torch.Tensor]) -> torch.Tensor:
# 個別の入力をバッチに結合
# inputsは複数のリクエストからのテンソルのリスト
return torch.stack(inputs)
def predict(self, batch: torch.Tensor) -> torch.Tensor:
# バッチ全体を一度に処理 - ここで効率化が実現される
if self.device == "cuda":
batch = batch.cuda()
with torch.no_grad():
# シンプルなモデル用に画像を平坦化
batch_flattened = batch.view(batch.size(0), -1)
logits = self.model(batch_flattened)
probabilities = torch.softmax(logits, dim=1)
return probabilities.cpu()
def unbatch(self, batch_output: torch.Tensor) -> List[torch.Tensor]:
# バッチ出力を個別のレスポンスに分割
return [batch_output[i] for i in range(batch_output.size(0))]
def encode_response(self, output: torch.Tensor):
# テンソルをJSON化可能な形式に変換
probs = output.numpy()
predicted_class = int(np.argmax(probs))
confidence = float(np.max(probs))
return {
"predicted_class": predicted_class,
"confidence": confidence,
"all_probabilities": probs.tolist()
}
if __name__ == "__main__":
api = BatchedImageClassifier()
# バッチ処理のパラメータを設定
server = ls.LitServer(
api,
accelerator="auto",
max_batch_size=8, # バッチあたりの最大リクエスト数
batch_timeout=0.1 # 最大待機時間（100ms）
)
server.run(port=8000)
バッチ処理のメリット
適切なバッチ処理を行えば、GPU推論で5〜10倍のスループット向上を実現できます。GPUが一度に複数のリクエストを処理するため、1つずつ処理するよりも大幅に効率的になるのです。
バッチ処理って、そもそもどういう仕組みで速くなるんですか？難しそうで自分で実装できる気がしないんですが…
バッチ処理は、複数のリクエストをまとめてGPUに送ることで効率化する仕組みです。GPUは並列処理が得意なので、1つずつ処理するより、まとめて処理する方が圧倒的に高速になります。LitServeではbatch()とunbatch()メソッドを定義するだけで自動的にバッチ処理してくれるので、複雑なロジックを自分で書く必要はありません。
9. 機能④：リアルタイムストリーミング機能
チャットボットやライブ推論のようなリアルタイムアプリケーションでは、ストリーミングレスポンスがより良いユーザー体験を提供します。
import litserve as ls
import time
import json
class StreamingChatAPI(ls.LitAPI):
def setup(self, device):
# 実際のシナリオでは、ここでLLMを読み込みます
self.device = device
print("ストリーミングチャットAPIの準備が完了しました！")
def decode_request(self, request):
return {
"message": request["message"],
"stream": request.get("stream", True)
}
def predict(self, inputs):
message = inputs["message"]
stream = inputs["stream"]
# ストリーミングレスポンスをシミュレート（実際にはLLM生成になります）
response_text = f"あなたのメッセージ「{message}」を受け取りました。このトピックについて詳しく説明します..."
if stream:
# ストリーミング用のジェネレータを返す
return self._generate_streaming_response(response_text)
else:
# 完全なレスポンスを返す
return {"response": response_text}
def _generate_streaming_response(self, text):
"""レスポンスチャンクを生成するジェネレータ"""
words = text.split()
for i, word in enumerate(words):
chunk = {
"chunk": word + " ",
"index": i,
"finished": i == len(words) - 1
}
yield f"data: {json.dumps(chunk)}\n\n"
time.sleep(0.1) # 処理遅延をシミュレート
def encode_response(self, output):
# ストリーミングの場合、LitServeが自動的にジェネレータを処理
if hasattr(output, '__iter__') and not isinstance(output, (str, dict)):
return output # ストリーミング用にジェネレータをそのまま返す
else:
return output # 非ストリーミング用に辞書を返す
if __name__ == "__main__":
api = StreamingChatAPI()
server = ls.LitServer(api, stream=True) # ストリーミングサポートを有効化
server.run(port=8000)
ストリーミングのテスト
# ストリーミングレスポンスをテスト
curl -X POST http://127.0.0.1:8000/predict \
-H "Content-Type: application/json" \
-d '{"message": "こんにちは！", "stream": true}' \
--no-buffer
# 非ストリーミングレスポンスをテスト
curl -X POST http://127.0.0.1:8000/predict \
-H "Content-Type: application/json" \
-d '{"message": "こんにちは！", "stream": false}'
💬 ストリーミングが役立つシーン：チャットボット、リアルタイム翻訳、ライブ文字起こしなど
10. 機能⑤：GPU自動スケーリング
LitServeは、最大スループットを実現するために、複数のGPUとワーカーに自動的にスケールできます。
import litserve as ls
import torch
class ScalableModelAPI(ls.LitAPI):
def setup(self, device):
# デバイスはLitServeによって自動的に割り当てられる（cuda:0、cuda:1など）
self.device = device
# 割り当てられたデバイスにモデルを読み込む
self.model = torch.nn.Linear(100, 10)
if device != "cpu":
self.model = self.model.to(device)
print(f"モデルがデバイス{device}に読み込まれました")
def decode_request(self, request):
return torch.tensor(request["input"], dtype=torch.float32)
def predict(self, x):
if self.device != "cpu":
x = x.to(self.device)
with torch.no_grad():
output = self.model(x)
return output.cpu().numpy().tolist()
def encode_response(self, output):
return {"prediction": output}
if __name__ == "__main__":
api = ScalableModelAPI()
# 複数のGPUと複数のワーカーでスケール
server = ls.LitServer(
api,
accelerator="auto",
devices="auto", # 利用可能なすべてのGPUを使用
workers_per_device=2, # GPU1台あたり2ワーカー
max_batch_size=16,
batch_timeout=0.05
)
# 別の方法：正確なデバイスを指定
# server = ls.LitServer(
# api,
# devices=[0, 1, 2, 3], # GPU 0、1、2、3を使用
# workers_per_device=2 # 合計：8ワーカー
# )
server.run(port=8000)
スケーリング設定のポイント
4つのGPUとGPUあたり2ワーカーの設定で、合計8つの並列ワーカーが自動的にロードバランシングとリクエスト分散を処理します。
🚀 メリット：複数のGPUを活用することで、処理能力を大幅に向上させることができます。
11. 機能⑥：AIエージェントの構築
LitServeは、複数のサービスやAPIを統合したAIエージェントの構築にも優れています。
import litserve as ls
import openai
import requests
import re
from datetime import datetime
class NewsAnalysisAgent(ls.LitAPI):
def setup(self, device):
# OpenAIクライアントを初期化（本番環境ではAPIキーを環境変数から取得）
self.openai_client = openai.OpenAI(
api_key="your-openai-api-key"
)
print("ニュース分析エージェントの準備が完了しました！")
def decode_request(self, request):
return {
"website_url": request.get("website_url", "https://text.npr.org/"),
"analysis_type": request.get("analysis_type", "summary"),
"max_articles": request.get("max_articles", 3)
}
def predict(self, inputs):
website_url = inputs["website_url"]
analysis_type = inputs["analysis_type"]
try:
# ステップ1：Webサイトのコンテンツを取得してクリーニング
response = requests.get(website_url, timeout=10)
response.raise_for_status()
# HTMLタグを削除してテキストを抽出
clean_text = re.sub(r'<[^>]+>', ' ', response.text)
clean_text = re.sub(r'\s+', ' ', clean_text).strip()
# API効率化のためテキスト長を制限
if len(clean_text) > 8000:
clean_text = clean_text[:8000] + "..."
# ステップ2：リクエストタイプに基づいてOpenAIで分析
if analysis_type == "summary":
prompt = f"このコンテンツから主要なニュースストーリーを要約してください：{clean_text}"
elif analysis_type == "sentiment":
prompt = f"これらのニュースストーリーの全体的な感情を分析してください：{clean_text}"
elif analysis_type == "topics":
prompt = f"これらのニュースストーリーから主要なトピックとテーマを抽出してください：{clean_text}"
else:
prompt = f"このニュースコンテンツの包括的な分析を提供してください：{clean_text}"
# ステップ3：LLMレスポンスを取得
# ※注：gpt-3.5-turboは現在も利用可能ですが、より新しいモデル（gpt-4o-miniなど）の使用も検討できます
llm_response = self.openai_client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "あなたは有能なニュースアナリストです。"},
{"role": "user", "content": prompt}
],
max_tokens=500,
temperature=0.7
)
analysis = llm_response.choices[0].message.content.strip()
return {
"success": True,
"analysis": analysis,
"analysis_type": analysis_type,
"source_url": website_url,
"timestamp": datetime.now().isoformat()
}
except requests.RequestException as e:
return {
"success": False,
"error": f"Webサイトの取得に失敗しました：{str(e)}",
"analysis_type": analysis_type
}
except Exception as e:
return {
"success": False,
"error": f"分析に失敗しました：{str(e)}",
"analysis_type": analysis_type
}
def encode_response(self, output):
return {
"agent": "NewsAnalysisAgent",
"result": output,
"version": "1.0"
}
if __name__ == "__main__":
api = NewsAnalysisAgent()
server = ls.LitServer(api, accelerator="cpu") # このエージェントにはCPUで十分
server.run(port=8000)
エージェントのテスト
# ニュース要約を取得
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{
"website_url": "https://text.npr.org/",
"analysis_type": "summary"
}'
# 感情分析
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{
"website_url": "https://text.npr.org/",
"analysis_type": "sentiment"
}'
🤖 活用シーン：ニュース分析、市場調査、競合分析、トレンド把握など
12. 実践例：Hugging Faceモデルのデプロイ
ここでは、BERTモデルをテキスト分類用にデプロイする完全な例を紹介します。
import litserve as ls
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from typing import List
class BERTClassificationAPI(ls.LitAPI):
def setup(self, device):
self.device = device
# 事前訓練されたBERT感情分析モデルを読み込む
model_name = "nlptown/bert-base-multilingual-uncased-sentiment"
print(f"BERTモデルを{device}に読み込んでいます...")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
if device != "cpu":
self.model = self.model.to(device)
# モデルを評価モードに設定
self.model.eval()
# クラスラベルを定義（このモデルは1〜5つ星評価を出力）
self.labels = ["1つ星", "2つ星", "3つ星", "4つ星", "5つ星"]
print("BERTモデルが正常に読み込まれました！")
def decode_request(self, request):
# 単一テキストとテキストのバッチの両方に対応
if isinstance(request["text"], str):
return [request["text"]]
return request["text"]
def batch(self, inputs: List[List[str]]) -> List[str]:
# リストのリストを単一のリストに平坦化
flattened = []
for item in inputs:
flattened.extend(item)
return flattened
def predict(self, texts: List[str]) -> List[dict]:
# バッチ内のすべてのテキストをトークン化
inputs = self.tokenizer(
texts,
padding=True,
truncation=True,
max_length=512,
return_tensors="pt"
)
if self.device != "cpu":
inputs = {k: v.to(self.device) for k, v in inputs.items()}
# 推論を実行
with torch.no_grad():
outputs = self.model(**inputs)
predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
# CPUに変換して結果を処理
predictions = predictions.cpu().numpy()
results = []
for i, prediction in enumerate(predictions):
predicted_class = prediction.argmax()
confidence = float(prediction[predicted_class])
results.append({
"text": texts[i],
"predicted_rating": self.labels[predicted_class],
"confidence": round(confidence, 4),
"all_scores": {
label: round(float(score), 4)
for label, score in zip(self.labels, prediction)
}
})
return results
def unbatch(self, batch_output: List[dict]) -> List[List[dict]]:
# 個別の結果を返す（各リクエストはその結果を含むリストを取得）
return [[result] for result in batch_output]
def encode_response(self, output: List[dict]):
# 単一アイテムの場合はそのまま返し、複数の場合はリストを返す
if len(output) == 1:
return {
"status": "成功",
"prediction": output[0],
"model": "BERT多言語感情分析"
}
else:
return {
"status": "成功",
"predictions": output,
"count": len(output),
"model": "BERT多言語感情分析"
}
if __name__ == "__main__":
api = BERTClassificationAPI()
server = ls.LitServer(
api,
accelerator="auto", # GPU利用可能なら使用
max_batch_size=8, # 一度に最大8テキストを処理
batch_timeout=0.1, # バッチ形成の最大待機時間100ms
workers_per_device=1 # メモリ効率のためデバイスあたり1ワーカー
)
server.run(port=8000)
BERT APIのテスト
# 単一テキスト分類
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{"text": "この製品が大好きです！品質が素晴らしく、配送も迅速でした。"}'
# バッチテキスト分類
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{
"text": [
"素晴らしいサービスで、強くお勧めします！",
"ひどい経験で、二度と買いません。",
"普通の製品で、特別なことは何もありません。"
]
}'
最適化のヒント
✅ モデル量子化：torch.quantizationを使用してメモリ使用量を削減
✅ ONNX変換：モデルをONNXに変換してより高速な推論を実現
✅ バッチサイズ調整：GPUメモリに基づいてmax_batch_sizeを実験
✅ キャッシング：繰り返しリクエストに対してレスポンスキャッシングを実装
13. 応用機能の紹介
非同期処理
import litserve as ls
import asyncio
import aiohttp
class AsyncAPI(ls.LitAPI):
async def predict(self, inputs):
# I/Oバウンドタスクには非同期操作を使用
async with aiohttp.ClientSession() as session:
async with session.get(inputs["url"]) as response:
data = await response.json()
return data
OpenAI API互換性
# LitServeはOpenAI互換エンドポイントを提供できる
server = ls.LitServer(
api,
spec=ls.OpenAISpec(), # OpenAI互換性を有効化
port=8000
)
認証機能
# 認証ミドルウェアを追加
server = ls.LitServer(
api,
middlewares=[ls.TokenAuthMiddleware(token="your-secret-token")]
)
カスタムエンドポイント
class MultiEndpointAPI(ls.LitAPI):
@ls.endpoint("/health")
def health_check(self):
return {"status": "正常"}
@ls.endpoint("/metrics")
def get_metrics(self):
return {"requests": self.request_count}
14. パフォーマンスベンチマーク
LitServeは、標準的なサービングソリューションと比較して大幅なパフォーマンス向上を実現します。
🚀 速度向上の実績
- 最低でもFastAPIより2倍高速
- バッチ処理を有効にすると5〜10倍高速
- ほとんどのAIワークロードでTorchServeを上回る
- 追加GPUでリニアにスケール
バッチ処理が2倍以上の改善をもたらすケース
✅ GPUバウンドモデル：画像分類、物体検出
✅ 大規模モデル：計算要求の高いTransformerモデル
✅ 高スループットシナリオ：複数の同時リクエスト
✅ 標準化された入力：類似の入力サイズと処理要件
ベンチマーク結果
画像分類とテキスト処理タスクの標準化されたテストにおいて：
| 構成 | リクエスト/秒 |
|---|---|
| FastAPIベースライン | 100 |
| LitServe（バッチ処理なし） | 200+ |
| LitServe（バッチ処理あり） | 800+ |
| LitServe（マルチGPU） | 2000+ |
15. デプロイ方法
Dockerでのセルフホスティング
# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY server.py . EXPOSE 8000 CMD ["python", "server.py"] # ビルドと実行 docker build -t my-litserve-app . docker run -p 8000:8000 my-litserve-app
Lightning Cloudへのワンクリックデプロイ
# 自動スケーリング付きでLightning Cloudにデプロイ lightning deploy server.py --cloud # カスタム設定でデプロイ lightning deploy server.py --cloud --instance-type gpu_1x_a100
Lightning Cloudの利点
- 自動スケーリング
- 99.995%の稼働率
- 組み込みモニタリング
- 使った分だけ支払う料金体系
- エンタープライズセキュリティ機能
本番デプロイチェックリスト
✅ 認証とレート制限を有効化
✅ 適切なログ記録とモニタリングを設定
✅ ヘルスチェックとメトリクスエンドポイントをセットアップ
✅ 適切なエラーハンドリングを実装
✅ WebアプリケーションにCORSを設定
✅ シークレットに環境変数を使用
✅ 自動テストとCI/CDをセットアップ
16. ベストプラクティス
バッチ処理を使うべきとき
- GPU集約型モデル：コンピュータビジョン、大規模言語モデル
- 類似のリクエストパターン：一貫した入力サイズと処理時間
- 高スループット要件：多数の同時リクエストが予想される
⚠️ 避けるべき場合：即座のレスポンスが必要なリアルタイムアプリケーション
GPUメモリ管理
def setup(self, device):
# モデルを読み込む前にGPUキャッシュをクリア
if device != "cpu":
torch.cuda.empty_cache()
# メモリ効率の良い設定でモデルを読み込む
self.model = AutoModel.from_pretrained(
model_name,
torch_dtype=torch.float16, # 半精度を使用
device_map="auto" # 自動デバイスマッピング
)
エラーハンドリング
def predict(self, inputs):
try:
result = self.model(inputs)
return {"status": "成功", "result": result}
except torch.cuda.OutOfMemoryError:
torch.cuda.empty_cache()
return {"status": "エラー", "message": "GPUメモリ超過"}
except Exception as e:
logger.error(f"予測に失敗しました：{str(e)}")
return {"status": "エラー", "message": "予測に失敗しました"}
モニタリングとログ記録
import logging
from time import time
class MonitoredAPI(ls.LitAPI):
def __init__(self):
self.request_count = 0
self.logger = logging.getLogger(__name__)
def predict(self, inputs):
start_time = time()
self.request_count += 1
try:
result = self._run_inference(inputs)
processing_time = time() - start_time
self.logger.info(
f"リクエスト{self.request_count}が{processing_time:.3f}秒で完了"
)
return result
except Exception as e:
self.logger.error(f"リクエスト失敗：{str(e)}")
raise
17. 活用シーン
LLMとRAGシステム
- チャットボット：カスタマーサービス、パーソナルアシスタント
- ドキュメントQ&A：企業ナレッジベース
- コード生成：プログラミングアシスタント
- コンテンツ作成：ライティング支援、マーケティングコピー
コンピュータビジョンAPI
- 画像分類：医療画像、品質管理
- 物体検出：セキュリティシステム、在庫管理
- 画像生成：クリエイティブツール、パーソナライゼーション
- OCRとドキュメント処理：フォーム処理、デジタル化
音声処理
- 音声認識：文字起こしサービス、音声コマンド
- 音声合成：アクセシビリティ、コンテンツ作成
- 音声分類：音楽推薦、音声検出
- 音声クローニング：パーソナライズされた音声アシスタント
マルチモーダルアプリケーション
- 視覚的質問応答：教育ツール、アクセシビリティ
- 画像キャプション：コンテンツモデレーション、SEO
- ビデオ分析：コンテンツ理解、監視
- ドキュメントAI：マルチフォーマットドキュメント処理
従来の機械学習モデル
- 推薦システム：Eコマース、コンテンツプラットフォーム
- 不正検出：金融サービス、セキュリティ
- 予測分析：ビジネスインテリジェンス、予測
- A/Bテスト：実験プラットフォーム
18. まとめ
LitServeは、AIモデルサービングをシンプルかつ高性能にするために設計されたサービングエンジンです。FastAPIの柔軟性とAI特化の強力な最適化を組み合わせ、AIモデルのデプロイに最適な選択肢となっています。
主要なメリットのおさらい
- パフォーマンス：FastAPIより最低2倍高速、バッチ処理で10倍以上も可能
- シンプルさ：複雑なMLOpsセットアップ、YAML設定、ボイラープレートコード不要
- 柔軟性：あらゆるモデルタイプとカスタム推論ロジックに対応
- 組み込み機能：バッチ処理、ストリーミング、GPU自動スケーリングなどが標準装備
- デプロイオプション：どこにでもセルフホスト可能、Lightning Cloudへワンクリックデプロイ
LitServeを選ぶべきとき
✅ こんな場合にLitServeを選択：
柔軟性、パフォーマンス、推論パイプラインの完全な制御が必要な場合
✅ 最適な用途：
カスタムモデル、マルチモデルシステム、複雑な前処理、エージェント、RAGシステム
⚠️ 代替案を検討すべき場合：
カスタマイズゼロでそのまま使えるLLMサーバーが必要な場合（vLLM/Ollamaを直接使用）
次のステップ
さあ、実際にLitServeを使ってみましょう！以下のリソースを活用して、今日からAI推論APIの構築を始められます。
📚 参考リソース
この記事で紹介した基本をマスターしたら、ぜひ自分のプロジェクトに応用してみてください。プロンプトを工夫したり、さまざまなパラメータを調整したりしながら試行錯誤を重ねることで、理想のAI推論システムを構築できるようになります。
💡 最後に：LitServeは、AIモデルのデプロイをもっとシンプルに、もっと高速にするために開発されました。この記事が、あなたのAIプロジェクトを次のレベルに引き上げる助けになれば幸いです。
質問や困ったことがあれば、ぜひLitServeのGitHubコミュニティに参加してみてください。世界中の開発者が、あなたの挑戦をサポートしてくれるはずです！
この記事の著者
Md Amanatullah
生成AI、LLM、NLPを専門とするAI/ML開発者兼MLOpsエンジニア。
生成AI、大規模言語モデル（LLM）、自然言語処理（NLP）を専門とするAI/ML開発者兼MLOpsエンジニアで、5年以上の実務経験を有している。クラウドプラットフォームや最先端のフレームワークを活用し、実運用レベルのAIアプリケーションの構築とデプロイに精通している。AI技術の発展と、グローバルなコミュニティへの知識共有に情熱を注いでいる。
この記事は著者の許可を得て公開しています。
元記事：LitServe: Build Lightning-Fast AI Inference APIs with Ease — A Complete Guide
この記事の監修・コメント
池田朋弘（監修）
Workstyle Evolution代表。18万人超YouTuber＆著書『ChatGPT最強の仕事術』は4万部突破。
株式会社Workstyle Evolution代表取締役。YouTubeチャンネル「いけともch（チャンネル）」では、 AIエージェント時代の必須ノウハウ・スキルや、最新AIツールの活用法を独自のビジネス視点から解説し、 チャンネル登録数は18万人超（2025年7月時点）。
■合わせて読みたい
■関連記事
公式LINEで最新ニュースをゲット
最新のAIニュースを
毎週お届け
生成AIの業務別の
ビジネス活用シーン
がわかるAIチャット
過去のAIニュースから
事実を確認できる
何でもAI相談チャット
ニュース動画の
アーカイブ
