CoeFontの機械学習推論を支える技術

翻訳全文

同期処理エンドポイントの例は下記になります。この例では、リクエストがあるたびに hello 関数が呼び出され、その結果がレスポンスとして順次返されます。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")

def hello():

return {"Hello": "World"}

下記の図は、FastAPI サーバーが複数の HTTP リクエストを受け取った場合の、ワーカースレッドによる同期処理の流れを示しています。この図解では

リクエスト 1 をワーカースレッド 1 で行い、レスポンスが返されます。

リクエスト 2 をワーカースレッド 2 で行い、レスポンスが返されます。

同期処理の場合、リクエストを受けたスレッドは処理が完了するまでブロックされます。たとえば、外部サービスとの通信やデータベースクエリといった I/O バウンド（I/O bound）の処理が発生すると、そのスレッドは応答が返るまで待機状態になり、他の処理は行われません。

FastAPI は、Gunicorn などの Web サーバーと組み合わせて利用することが推奨されており、公式ドキュメント参照ではワーカー数を「CPU コア数 × 2 + 1」に設定することが推奨されています

これにより、複数の FastAPI が立ちあげられ、並列でのリクエスト処理が可能になります。また、各ワーカーで設定されるスレッド数（Python3.12 だと min(32, (os.process_cpu_count() or 1) + 4)。参照）がパフォーマンスに影響を与えます。

しかし、プロセスやスレッド間のコンテキストスイッチには高いオーバーヘッドが伴うため、数を増やすことがパフォーマンスが向上するとは限らず、大量のリクエストを処理する際には限界が生じる可能性があります。

そこで使うのが、非同期処理です。同期処理エンドポイントの例は下記になります。この例では、hello 関数は非同期関数として定義され、await asyncio.sleep(1) により 1 秒間待機しています。

from fastapi import FastAPI

import asyncio

app = FastAPI()

@app.get("/async")

async def hello():

await asyncio.sleep(1)

return {"message": "This is an async response"}

FastAPI 上では、イベントループが各非同期タスクの進捗を管理します。例えば、I/O 待ち中に他のタスクを実行することで、全体のパフォーマンスが向上する可能性があります。また、タスク間の切り替えはコンテキストスイッチよりもはるかに低負荷です。

非同期処理エンドポイントの注意点

しかし、注意する点もあります。パフォーマンスの大幅な悪化を防ぐために、非同期エンドポイントの中で、同期処理を行ってはいけません。

下記の図は、FastAPI が複数の HTTP リクエスト（短い I/O と長い I/O）を受け、同期処理がイベントループ上でブロックするため、タスクの開始が遅れている状態を示しています。

非同期に対応していないライブラリ（例：boto3）などは、別のライブラリ（例：aiboto3）に切り替えたり、無理やりイベントループから外部スレッドプールのスレッドに処理を退避させて実行するなど、徹底的に非同期化する必要があります。外部スレッドプールのスレッドに処理を退避させる例は下記です。

import asyncio

import concurrent.futures

import time

def sync_inference(input_data):

# ここでは 2 秒間のスリープで重い処理をシミュレーション

time.sleep(2)

return f"同期推論結果：{input_data}"

async def run_sync_inference_in_threadpool(input_data):

loop = asyncio.get_running_loop()

# ThreadPoolExecutor を利用して、同期処理を非同期的に実行

with concurrent.futures.ThreadPoolExecutor() as executor:

result = await loop.run_in_executor(executor, sync_inference, input_data)

return result

async def main_sync():

result = await run_sync_inference_in_threadpool("サンプルデータ")

print(result)

非同期処理は、実装の複雑さやレビュー負荷の増大という側面や、実装ミスによるパフォーマンス悪化という側面もあるため、性能向上のメリットとこれらのコストとのバランスを考慮して、適切な手法を選択することが重要です。

FastAPI における同期処理と非同期処理の特徴やパフォーマンスの違いをより深く理解するために、以下の資料がおすすめです。

Pycon 2024 FastAPI での async def と def の使い分け

「FastAPI での async/def と def の違い：非同期を使うならいつ使わないのか」では、同期処理と非同期処理の基本概念やそれぞれの実装上のメリット・デメリットが丁寧に解説されています。

FastAPI で async def と def をちゃんと使い分ける

こちらの記事では、非同期処理内での同期処理がブロッキングを引き起こすのか、また同期処理をスレッドプールにオフロードすることでパフォーマンスが向上するかどうかの検証結果が示されています。実際の検証結果に基づいた実践的な知見が得られるため、パフォーマンス改善のヒントになります。

これらの資料を参照することで、FastAPI の同期処理と非同期処理の適材適所な使い分けや、パフォーマンス向上のための具体的な手法について、より深い理解を得ることができるでしょう。

Server Sent Event で通信する

Server Sent Event とは

従来の HTTP 通信は、リクエストごとに一度だけレスポンスを返す仕組みです。すべての機械学習による推論処理が完了してから、応答が返されます。

一方、Server Sent Event（以下 SSE）はサーバーからクライアントへの一方向のストリーミングに特化しており、実装がシンプルでリソース効率が高く、モデル推論結果などの逐次的なデータ送信に最適です。例えば ChatGPT などで使われています。下記は、SSE を利用して、サーバーから文字列がリアルタイムに送信される様子を示しています

モデルの推論結果をストリーミングで返す場合、SSE を活用すると、部分的な結果を早期に表示でき、ユーザーエクスペリエンスが向上します。以下は、サーバーからクライアントへ文字列データをリアルタイムに送信する例です。

詳細な仕様については、MDN の公式ドキュメントなども参考にされると良いでしょう.

Server Sent Event のエラーハンドリング

SSE では、接続が正常に確立されるとサーバーは HTTP ステータスコード 200（OK）を返し、クライアントへ継続的なイベント送信が可能な状態となります。

一方、データの送信中にエラーが発生した場合は、HTTP ステータスコード以外の方法でエラー情報を伝える必要があります。弊社では、例えば以下のような形式でエラー情報を送信し、クライアント側で適切にハンドリングできるようにしています。

{"event": "error", "data": "invalid-request-error"}

全体のコード例は次のようになります。

翻訳全文

from fastapi import FastAPI

from sse_starlette.sse import EventSourceResponse

import asyncio

app = FastAPI()

def fakeLLM_generator():

messages = [

"これは Server Sent Event のテスト文章です。",

"この文章は、LLM の出力を模倣するために作られた長文であり、",

"様々な情報が盛り込まれています。",

"ランダムなタイミングで切り取られることにより、",

"部分的な情報の伝達が行われます。",

"まるで実際の会話のように、途中で途切れることもありますが、",

"全体としては一貫した内容を持っています。",

"これにより、ユーザーは逐次的なデータの受信を体験することができます。",

"さらに、このテストはシステムのストリーミング機能を確認するためのものです。",

"出力が途中で切れても、後続のメッセージで続きが送られます。"

]

for msg in messages:

yield msg

async def event_generator():

try:

llm_iterator = fakeLLM_generator()

yield {"event": "start", "data": "開始します"}

for message in llm_iterator:

yield {"data": message}

await asyncio.sleep(1)

yield {"event": "complete", "data": "送信完了"}

# 何かしらのクライアントエラー

except CustomClientError as e:

yield {"event": "error", "data": "invalid-request-error"}

# それ以外のエラー

except Exception as e:

yield {"event": "error", "data": "unknown-error"}

@app.get("/sse")

async def sse_endpoint():

return EventSourceResponse(event_generator())

中間層なしでのリクエストの処理

システムアーキテクチャとして、よく見られる形態は「クライアント ⇔ バックエンド ⇔ 推論エンドポイント」という構成です。これは、従来のバックエンド（例：登録・認証・決済）と複雑な推論システムとの役割を明確に分離することで、システム全体の管理性や拡張性を向上させるアプローチとなっています。

弊社では、クライアントが直接 Resource サーバー（推論サーバー）からレスポンスを受け取る方式へ変更している箇所があります。初回の認証は Authorization サーバー（通常のバックエンド）で実施され、その後の各リクエストでは、Resource サーバーが JWS 署名の検証を行い、リソースを返却します。この仕組みにより、不要な仲介層を排除し、通信経路を短縮することで、応答速度および全体パフォーマンスが大幅に向上します。

なお、ユーザー認証には JWS 認証を採用し、高速な検証を実現しています。以下のシーケンス図は、新しいフローを示しています。

このアーキテクチャでは、各リクエストがデータベースにアクセスする頻度が減少するため、システム全体の応答速度が向上します。

推論の信頼性向上を実現する技術

システムの信頼性を向上させるためのモニタリング、エラーハンドリング、インフラ管理などの取り組みについて説明します。たとえ高速な推論結果を提供できるシステムであっても、信頼性が担保されなければユーザーの信頼を得ることは難しいため、信頼性向上は高速応答施策と連動して、全体のサービス品質を支える重要な要素となります。

各種モニタリングツールを活用する

アプリケーションモニタリング・インフラ監視・外形監視

Sentry はアプリケーションモニタリング（APM）ツールとして、発生したエラーをリアルタイムにキャッチし、詳細なコンテキスト（スタックトレース、ユーザー情報、環境情報など）を提供します。料金もかなり抑えめに利用できます。

エラー発生時のスタックトレースの表示

エラー時のリクエストペイロードの表示

一定の閾値を超えるエラーに対する通知

特定エンドポイントのエラー率・パフォーマンスの表示

リクエストのプロファイリングの表示

音声合成 AI の場合、本番環境でのユーザーからのさまざまな文字入力がなされ、予期せぬエラーを吐いているケースがあります。ペイロードを詳細に確認できる機能は、問題発生から発見・解決までの時間短縮に大いに役立っています。

また、弊社では Datadog も用いています。料金は相対的に高いですが、インフラ、アプリケーション、ログ、メトリクス、トレースなどを一元管理でき、さまざまな機能を提供しています。

監視という観点だと、Datadog Synthetic Monitoring を用いた AWS ネットワーク外からの監視により、システム全体が正常に動作しているかを継続的にチェックしています。

SLI SLI は、サービスのパフォーマンスや品質を数値で表現する指標です。具体的には、システムの稼働率、レスポンスタイム、エラーレートなどが該当します。

SLO SLO は、サービス提供側がユーザーに対して約束する具体的な性能目標です。たとえば、「99.9% の稼働率」や「平均レスポンス時間 500 ミリ秒以内」など、明確な数値目標が設定されます。

サービスレベルの目標（SLO）を策定し、その達成状況を示す指標（SLI）を継続的にモニタリングしています。GPU を使った計算コストが高い処理や動作が不安定な外部 API（LLM）の影響により、予期せぬ変動が生じる可能性があるため、常に数値を確認しています。

チームで、外形監視による稼働率やレイテンシの SLO と、実ユーザー向けの稼働率・レイテンシの SLO を設定し、毎週確認を行っています。エラーバジェットを考慮しながら、各タスクとして改善に取り組んだ結果、チーム発足以降、SLI は大幅に改善しました。

また、PO/PdM との共通指標としてシステムの状態を数値化することで、トラブル発生時の対応や改善策の議論がスムーズに進むようになっています。

また、機械学習システムでは、単にシステム自体の稼働状況をモニタリングするだけでなく、出力されるコンテンツの品質チェックも不可欠です。 たとえば、

音声生成の推論結果のロギング・モニタリング

推論の信頼性向上を実現する技術

システムの信頼性を向上させるためのモニタリング、エラーハンドリング、インフラ管理などの取り組みについて説明します。たとえ高速な推論結果を提供できるシステムであっても、信頼性が担保されなければユーザーの信頼を得ることは難しいため、信頼性向上は高速応答施策と連動して、全体のサービス品質を支える重要な要素となります。

各種モニタリングツールを活用する

アプリケーションモニタリング・インフラ監視・外形監視

Sentry はアプリケーションモニタリング（APM）ツールとして、発生したエラーをリアルタイムにキャッチし、詳細なコンテキスト（スタックトレース、ユーザー情報、環境情報など）を提供します。料金もかなり抑えめに利用できます。

エラー発生時のスタックトレースの表示

エラー時のリクエストペイロードの表示

一定の閾値を超えるエラーに対する通知

特定エンドポイントのエラー率・パフォーマンスの表示

リクエストのプロファイリングの表示

音声合成 AI の場合、本番環境でのユーザーからのさまざまな文字入力がなされ、予期せぬエラーを吐いているケースがあります。ペイロードを詳細に確認できる機能は、問題発生から発見・解決までの時間短縮に大いに役立っています。

また、弊社では Datadog も用いています。料金は相対的に高いですが、インフラ、アプリケーション、ログ、メトリクス、トレースなどを一元管理でき、さまざまな機能を提供しています。

監視という観点だと、Datadog Synthetic Monitoring を用いた AWS ネットワーク外からの監視により、システム全体が正常に動作しているかを継続的にチェックしています。

SLI SLI は、サービスのパフォーマンスや品質を数値で表現する指標です。具体的には、システムの稼働率、レスポンスタイム、エラーレートなどが該当します。

SLO SLO は、サービス提供側がユーザーに対して約束する具体的な性能目標です。たとえば、「99.9% の稼働率」や「平均レスポンス時間 500 ミリ秒以内」など、明確な数値目標が設定されます。

サービスレベルの目標（SLO）を策定し、その達成状況を示す指標（SLI）を継続的にモニタリングしています。GPU を使った計算コストが高い処理や動作が不安定な外部 API（LLM）の影響により、予期せぬ変動が生じる可能性があるため、常に数値を確認しています。

チームで、外形監視による稼働率やレイテンシの SLO と、実ユーザー向けの稼働率・レイテンシの SLO を設定し、毎週確認を行っています。エラーバジェットを考慮しながら、各タスクとして改善に取り組んだ結果、チーム発足以降、SLI は大幅に改善しました。

また、PO/PdM との共通指標としてシステムの状態を数値化することで、トラブル発生時の対応や改善策の議論がスムーズに進むようになっています。

また、機械学習システムでは、単にシステム自体の稼働状況をモニタリングするだけでなく、出力されるコンテンツの品質チェックも不可欠です。 たとえば、

音声生成の推論結果のロギング・モニタリング

LLM の監視（例：langfuse の利用）

などが求められます。弊社でも絶賛取組中の領域になっています。

現行の Text2Speech の仕様では、リクエスト数だけでなく、リクエストごとの文字数もサーバー負荷にほぼ線形で影響を与えます。同じ RPM でも、文字数の違いにより実際の負荷が 100 倍に達する可能性があるため、単純なリクエスト数のみでの負荷状況は把握できません。

そこで、各リクエストの文字数情報も Cloud Watch Metrics に流し込み、迅速かつ柔軟なスケーリングを行い、システム全体のパフォーマンスを改善しようとしています。

クライアントが直接リクエストを送信する設計は、不正アクセスのリスクを増加させる可能性があります。

そのため、AWS WAF を活用した DDoS 攻撃や脆弱性を狙った攻撃への防御、ライブラリの脆弱性管理、そして IDS（侵入検知システム）など、包括的なセキュリティ対策が不可欠となります。

本記事でご紹介した各技術と取り組みは、我々のサービスが迅速かつ安定して提供されるための基盤となっています。MLOps チームは、非同期処理の最適化やリアルタイム通信、そして継続的なモニタリングを通じて、システム全体のパフォーマンスと信頼性の向上に努めています。これからも最新技術の積極的な採用と運用改善により、さらなる品質向上を追求していきます。

CoeFontPublicationAI 音声プラットフォーム「CoeFont（コエフォント）」の公式テックブログです。