概要

生成AIを組み込んだ業務アプリケーションでは、応答性の高いUIと、AI／MLライブラリを扱いやすいバックエンドの両立が求められます。DigitalBaseでは、この要件を満たす構成として フロントエンドにNext.js、バックエンドにFastAPI を採用するケースが多くあります。

この構成を前提に、業務AIアプリケーションの本番運用に関わるデータアクセスの集約、DB接続管理、APIのアクセス制御、認証、パフォーマンス最適化の設計指針を整理します。

システムアーキテクチャ

基本的な構成は以下のとおりです。

[ユーザー] <--> [Next.js Frontend]
                    |
                    | REST API / WebSocket
                    |
                    v
              [FastAPI Backend]
                    |
                    +-- [SQLAlchemy ORM]
                    |        |
                    |        v
                    |   [PostgreSQL]
                    |
                    +-- [AI Services]
                           |
                           +-- LLM API（Claude / OpenAI など）
                           +-- LangChain
                           +-- ベクトルDB（pgvector / Pinecone / Weaviate）

フロントエンドはAPI経由でのみデータにアクセスし、DBへの直接アクセスはバックエンドに集約する構成を基本とします。

データベース設計の指針

Prisma と SQLAlchemy の使い分け

DigitalBaseで推奨する基本構成は次のとおりです。

  • データアクセスはバックエンド（SQLAlchemy）に集約：Next.jsから直接DBへアクセスしない
  • スキーマ・マイグレーション管理：Alembic（SQLAlchemy用）で一元管理

そのうえで、以下のようなケースではPrismaの併用も選択肢になります。

Prismaを併用するメリット

  • TypeScriptファーストで型安全性が高い
  • Prisma StudioでデータをGUIから確認・管理できる
  • Server ActionsやRoute Handlerで軽量なDBアクセスを行う用途に適している

SQLAlchemyを主軸にするメリット

  • PythonのAI／MLライブラリとの連携が容易
  • 複雑なクエリやトランザクション制御に強い
  • FastAPIとの統合が成熟している

データアクセス層を二重化するとスキーマの整合性管理が煩雑になるため、原則は SQLAlchemyに一本化 し、Prismaは限定的な用途に留めることを推奨します。

データベース接続管理

非同期エンジンを用いる場合の接続管理の例です。

# backend/database.py

import os
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker,
)
from sqlalchemy.orm import declarative_base

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_async_engine(
    DATABASE_URL,
    echo=False,        # 開発環境ではTrue、本番ではFalse
    pool_size=20,
    max_overflow=0,
    pool_pre_ping=True,  # 接続の有効性を確認
)

AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

Base = declarative_base()

# 依存性注入用
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

SQLAlchemy 2.0系では、非同期セッションのファクトリに async_sessionmaker の使用が推奨されます。async with を抜ける際にセッションは自動的にクローズされます。

FastAPIでのAPI設計

Pydanticによるデータ検証と、ユーザーIDによるアクセス制御を組み合わせます。

Pydantic v2では、ORMオブジェクトからの変換に model_config = ConfigDict(from_attributes=True) を用います。

ユーザーIDによる絞り込みを各クエリに含めることで、他ユーザーのリソースへのアクセスを防ぎます。

AI機能の統合

LLMの呼び出しはバックエンドに集約します。

機密データを社外のAPIへ送信できない環境では、ローカルLLM（Ollama / vLLMなど）をOpenAI互換エンドポイントとして構築します。プロバイダを抽象化した設計により、後からの切り替えが容易になります。

認証・認可の実装

JWTを用いたトークン認証では、署名鍵とトークンの有効期限を管理します。

SECRET_KEY は十分なエントロピーを持つ値を環境変数で管理し、コードにハードコードしないことが前提です。アクセストークンの有効期限は短めに設定し、安全性の観点から、長期セッションにはリフレッシュトークンの使用を推奨します。

パフォーマンス最適化

1. データベース最適化

  • インデックスの作成：頻繁に検索されるカラムにインデックスを付与する
  • コネクションプール：pool_size をワークロードに合わせて設定する
  • N+1問題の解消：joinedload() や selectinload() でEager Loadingを行う

2. キャッシュ戦略

# Redisキャッシュの利用例
import json
from redis import asyncio as aioredis

redis = aioredis.from_url("redis://localhost")

async def get_cached_conversation(conversation_id: str):
    cached = await redis.get(f"conversation:{conversation_id}")
    if cached:
        return json.loads(cached)
    return None

async def set_cached_conversation(conversation_id: str, data: dict):
    await redis.setex(
        f"conversation:{conversation_id}",
        300,  # 5分間キャッシュ
        json.dumps(data),
    )

3. 非同期処理の活用

  • 入出力待ちが発生する処理は async def で実装する
  • 長時間かかる処理はBackgroundTasksやCelery、あるいはメッセージキューで非同期に実行する

要点

Next.js + FastAPI でAIアプリケーションを構築する際の設計指針は、次の6点です。

  1. データアクセスはバックエンド（SQLAlchemy）に集約し、フロントはAPI経由でアクセスする
  2. Pydantic（v2）でデータ検証と型安全性を確保する
  3. 入出力待ちが発生する処理に非同期処理を活用する
  4. JWTによる認証と、ユーザーIDによるリソースへのアクセス制御を実装する
  5. インデックス・コネクションプール・Redisキャッシュでパフォーマンスを最適化する
  6. LLMプロバイダを抽象化し、外部APIとローカルLLMを切り替えられる設計にする

自社の要件やデータの機密性に合わせて、これらの設計を調整します。

関連記事