Next.js + FastAPIによるAIアプリ基盤｜SQLAlchemyを軸にしたDB・API設計の指針

はじめに

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

本稿では、この構成を前提に、データアクセス層（Prisma / SQLAlchemy の使い分け）、データベース設計、API設計、認証・認可、パフォーマンス最適化について、実装の指針とコード例を整理します。

---

なぜNext.js + FastAPIなのか

Next.jsの強み

1. Reactベースのモダンなフレームワーク：コンポーネント指向で保守性が高い
2. SSR / SSG対応：サーバーサイドレンダリングと静的生成によりSEOとパフォーマンスを両立
3. App Router：ネストレイアウトやローディング状態の管理を宣言的に記述できる
4. TypeScript完全サポート：型安全な開発が可能

FastAPIの強み

1. 高速なAPI開発：Pydanticによるデータ検証とOpenAPIドキュメントの自動生成
2. 非同期処理：async / await による高スループットなAPI
3. Pythonエコシステム：LangChain、Transformers、各種LLM SDKとの連携が容易
4. 自動ドキュメント：Swagger UIとReDocが標準で生成される

AI機能はPythonライブラリへの依存が大きく、バックエンドをPythonに寄せることで連携コストを下げられる点が、この構成を選ぶ主な理由です。

---

システムアーキテクチャ

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

[ユーザー] <--> [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は限定的な用途に留めることを推奨します。

SQLAlchemyでのモデル設計例

# backend/models/user.py

from sqlalchemy import Column, String, DateTime, Boolean
from sqlalchemy.dialects.postgresql import UUID
from datetime import datetime
import uuid

from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String, unique=True, nullable=False, index=True)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # リレーション設定例
    # conversations = relationship("Conversation", back_populates="user")

# backend/models/conversation.py

from sqlalchemy import Column, String, DateTime, ForeignKey, Text
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid

from database import Base

class Conversation(Base):
    __tablename__ = "conversations"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # AI用のメタデータをJSONBで保存
    meta = Column(JSONB, default={})

    # リレーション
    user = relationship("User", back_populates="conversations")
    messages = relationship(
        "Message", back_populates="conversation", cascade="all, delete-orphan"
    )

class Message(Base):
    __tablename__ = "messages"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    conversation_id = Column(
        UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False
    )
    role = Column(String, nullable=False)  # "user" or "assistant"
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    # AI生成情報をJSONBで保存（token数、モデル名など）
    ai_metadata = Column(JSONB, default={})

    conversation = relationship("Conversation", back_populates="messages")

補足：metadata はSQLAlchemyのDeclarative APIで予約されている属性名のため、カラム名には meta など別名を用いる点に注意してください。

データベース接続管理

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

# 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設計

アプリケーションの初期化

# backend/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI(
    title="AI Application API",
    description="Next.js + FastAPI AI Application",
    version="1.0.0",
)

# CORS設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Next.jsのデフォルトポート
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ルーターの登録
from routers import auth, conversations, ai

app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(conversations.router, prefix="/api/conversations", tags=["conversations"])
app.include_router(ai.router, prefix="/api/ai", tags=["ai"])

@app.get("/")
def read_root():
    return {"message": "AI Application API"}

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Pydanticスキーマの定義

# backend/schemas/conversation.py

from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from typing import List
from uuid import UUID

class MessageBase(BaseModel):
    role: str = Field(..., description="user or assistant")
    content: str

class MessageCreate(MessageBase):
    pass

class MessageResponse(MessageBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    conversation_id: UUID
    created_at: datetime

class ConversationBase(BaseModel):
    title: str

class ConversationCreate(ConversationBase):
    pass

class ConversationResponse(ConversationBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    user_id: UUID
    created_at: datetime
    updated_at: datetime
    messages: List[MessageResponse] = []

Pydantic v2では、ORMオブジェクトからの変換に model_config = ConfigDict(from_attributes=True) を用います（v1の class Config: orm_mode = True の後継）。

ルーターの実装

# backend/routers/conversations.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List
from uuid import UUID

from database import get_db
from models.conversation import Conversation
from schemas.conversation import ConversationCreate, ConversationResponse
from auth.dependencies import get_current_user

router = APIRouter()

@router.post("/", response_model=ConversationResponse)
async def create_conversation(
    conversation: ConversationCreate,
    db: AsyncSession = Depends(get_db),
    current_user=Depends(get_current_user),
):
    db_conversation = Conversation(
        title=conversation.title,
        user_id=current_user.id,
    )
    db.add(db_conversation)
    await db.commit()
    await db.refresh(db_conversation)
    return db_conversation

@router.get("/", response_model=List[ConversationResponse])
async def list_conversations(
    db: AsyncSession = Depends(get_db),
    current_user=Depends(get_current_user),
):
    result = await db.execute(
        select(Conversation)
        .where(Conversation.user_id == current_user.id)
        .order_by(Conversation.updated_at.desc())
    )
    return result.scalars().all()

@router.get("/{conversation_id}", response_model=ConversationResponse)
async def get_conversation(
    conversation_id: UUID,
    db: AsyncSession = Depends(get_db),
    current_user=Depends(get_current_user),
):
    result = await db.execute(
        select(Conversation)
        .where(Conversation.id == conversation_id)
        .where(Conversation.user_id == current_user.id)
    )
    conversation = result.scalar_one_or_none()

    if not conversation:
        raise HTTPException(status_code=404, detail="Conversation not found")

    return conversation

ユーザーIDによる絞り込みを各クエリに含めることで、他ユーザーのリソースへアクセスできない設計にしている点が重要です。

---

AI機能の統合

LLMの呼び出しはバックエンドに集約します。ここでは一例としてOpenAIのPython SDK（v1系）を用いた実装を示しますが、Claude（Anthropic SDK）など他のプロバイダでも同様の構成で置き換え可能です。

# backend/routers/ai.py

import os
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncGenerator
from openai import AsyncOpenAI

from database import get_db
from models.conversation import Message
from schemas.ai import ChatRequest, ChatResponse
from auth.dependencies import get_current_user

router = APIRouter()
client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
MODEL = "gpt-4o"

@router.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    db: AsyncSession = Depends(get_db),
    current_user=Depends(get_current_user),
):
    # ユーザーメッセージを保存
    db.add(Message(
        conversation_id=request.conversation_id,
        role="user",
        content=request.message,
    ))
    await db.commit()

    try:
        response = await client.chat.completions.create(
            model=MODEL,
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": request.message},
            ],
            temperature=0.7,
        )

        assistant_content = response.choices[0].message.content

        # AIメッセージを保存
        db.add(Message(
            conversation_id=request.conversation_id,
            role="assistant",
            content=assistant_content,
            ai_metadata={
                "model": MODEL,
                "tokens": response.usage.total_tokens,
            },
        ))
        await db.commit()

        return ChatResponse(message=assistant_content)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.post("/chat/stream")
async def chat_stream(
    request: ChatRequest,
    db: AsyncSession = Depends(get_db),
    current_user=Depends(get_current_user),
):
    async def generate() -> AsyncGenerator[str, None]:
        stream = await client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "user", "content": request.message}],
            stream=True,
        )

        full_content = ""
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                full_content += delta
                yield f"data: {delta}\n\n"

        # ストリーミング完了後にDBへ保存
        db.add(Message(
            conversation_id=request.conversation_id,
            role="assistant",
            content=full_content,
        ))
        await db.commit()

    return StreamingResponse(generate(), media_type="text/event-stream")

機密データを社外のAPIへ送信できない環境では、ローカルLLM（Ollama / vLLMなど）をOpenAI互換エンドポイントとして立て、base_url を差し替えるだけで同じコードを流用できます。プロバイダの抽象化を意識した設計にしておくと、後からの切り替えが容易です。

---

Next.jsでのフロントエンド実装

APIクライアントの作成

// frontend/lib/api.ts
import axios from 'axios';

const api = axios.create({
  baseURL: process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000',
  headers: {
    'Content-Type': 'application/json',
  },
});

// リクエストインターセプター：認証トークンを付与
api.interceptors.request.use(
  (config) => {
    const token = localStorage.getItem('token');
    if (token) {
      config.headers.Authorization = `Bearer ${token}`;
    }
    return config;
  },
  (error) => Promise.reject(error)
);

// レスポンスインターセプター：認証エラーのハンドリング
api.interceptors.response.use(
  (response) => response,
  async (error) => {
    if (error.response?.status === 401) {
      // トークン切れの場合はログインページへ
      localStorage.removeItem('token');
      window.location.href = '/login';
    }
    return Promise.reject(error);
  }
);

export default api;

React HooksでのAPI呼び出し

// frontend/hooks/useConversations.ts
import { useState, useEffect } from 'react';
import api from '@/lib/api';

interface Conversation {
  id: string;
  title: string;
  created_at: string;
  updated_at: string;
}

export function useConversations() {
  const [conversations, setConversations] = useState<Conversation[]>([]);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    fetchConversations();
  }, []);

  const fetchConversations = async () => {
    try {
      setLoading(true);
      const response = await api.get('/api/conversations');
      setConversations(response.data);
    } catch (err: any) {
      setError(err.message);
    } finally {
      setLoading(false);
    }
  };

  const createConversation = async (title: string) => {
    try {
      const response = await api.post('/api/conversations', { title });
      setConversations((prev) => [response.data, ...prev]);
      return response.data;
    } catch (err: any) {
      setError(err.message);
      throw err;
    }
  };

  return {
    conversations,
    loading,
    error,
    createConversation,
    refetch: fetchConversations,
  };
}

本番運用では、データ取得・キャッシュ・再検証を一元管理できる TanStack Query（React Query）やSWRの導入も検討に値します。

---

認証・認可の実装

JWTを用いたトークン認証の基本的な実装例です。

# backend/auth/jwt.py

import os
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        return None

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

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アプリケーションを構築する際の設計指針を整理しました。要点は次のとおりです。

1. データアクセスはバックエンド（SQLAlchemy）に集約し、フロントはAPI経由でアクセスする
2. Pydantic（v2）でデータ検証と型安全性を確保する
3. 非同期処理で高スループットを実現する
4. JWTでセキュアな認証・認可を実装する
5. インデックス・コネクションプール・Redisキャッシュでパフォーマンスを最適化する
6. LLMプロバイダを抽象化し、外部APIとローカルLLMを切り替えられる設計にしておく

この構成をベースに、自社の要件やデータの機密性に合わせて調整することで、保守性が高くスケーラブルなAIアプリケーション基盤を構築できます。