FastAPIとSQLAlchemyを使ってRAGのバックエンドを開発した話

1. はじめに

今回はFastAPIとSQLAlchemyを使ってRAG(Retrieval-Augmented Generation)のバックエンドを開発した経験についてお話しします。

FastAPIはAPIを構築するための比較的新しいWebフレームワークで、DjangoやFlaskと同程度の人気があります。
また、SQLAlchemyはSQLを操作するためのライブラリで、デファクトスタンダードなORM(Object Relational Mapping)ツールとして広く使用されています。
FastAPIとSQLAlchemyは、どちらもPythonで記述できるのでPythonに慣れている場合は簡単に習得することができます。

2. RAGとは

RAGとは、LLM(大規模言語モデル)と大量の社内ドキュメントなどの外部情報を組み合わせた技術のことで、検索拡張生成や取得拡張生成と訳されます。
ユーザーからの質問に対して大量にある社内ドキュメントなどから回答に必要な情報を検索して取得し、その結果をベースにLLMに回答を生成させます。

3. FastAPIの特徴

FastAPIの特徴を3つに絞って紹介したいと思います。

型ヒントで実装できる

FastAPIでは、Pythonの型ヒントを使用しクエリパラメータやリクエストボディの構造を明示的に定義することで、データのバリデーションを非常に簡単に実現できます。
例えば、リクエストボディを型ヒントで実装した場合のソースコードが以下になります。

from typing import Union

from fastapi import FastAPI

from pydantic import BaseModel

# リクエストボディの構造を定義するためにPydanticのBaseModelを継承したItemクラスを定義

class Item(BaseModel):

name: str

description: Union[str, None] = None

price: float

tax: Union[float, None] = None

app = FastAPI()

# @app.get("/items/")でHTTP GETメソッドで/items/にアクセスされたときのエンドポイントを定義

# create_item(item: Item)はリクエストボディのデータをItemクラスのインスタンスとして受け取り、そのインスタンスを返却

@app.post("/items/")

async def create_item(item: Item):

return item

FastAPIの公式ドキュメントより引用

簡単に非同期処理を実装できる

FastAPIでは、予約語async/awaitを使用し簡単に非同期処理を実装することができます。

@app.get('/')

async def read_results():

results = await some_library()

return results

FastAPIの公式ドキュメントより引用

自動生成されるAPIドキュメントがある

FastAPIはデフォルトで以下のようなAPIドキュメントが生成されるため開発を効率的に行うことができます。
サーバを起動し http://localhost:8000/docs にアクセスするとAPIドキュメントを確認することができます。

FastAPIの公式ドキュメントより引用

4. SQLAlchemyの特徴

SQLAlchemyの特徴を3つに絞って紹介したいと思います。

ORM(Object Relational Mapping)

SQLAlchemyの特徴の一つにORMがあります。
ORMとは、データベースのテーブルをPythonのクラスにマッピングし、データベース操作をオブジェクト指向的に行えるようにした技術です。
これにより、SQL文を書くことなくデータの取得や保存が簡単に行えます。

特定のDB製品に依存しない

SQLAlchemyは、PostgreSQL、MySQL、SQLite、Oracleなど、さまざまなデータベースと互換性があります。
これにより、特定のデータベースに依存せずにアプリケーションを開発できます。

Alembicによるマイグレーション管理が便利

Alembicは、SQLAlchemyと併用されるデータベースマイグレーションツールです。
Alembicを使用すると、データベースのスキーマ変更(テーブルの追加、カラムの変更、インデックスの追加など)を簡単に管理できます。
また、データベースのバージョンを管理し、特定のバージョンへのアップグレードやダウングレードも簡単に行えます。

5. ドキュメント周りの処理を少し見てみる

ここからはFastAPIとSQLAlchemyに焦点を当てて、実際に開発で使用したソースコードを一部抜粋し改変したものを見ていきたいと思います。
ソースコードの内容はドキュメントをアップロードする処理となります。
FastAPIとSQLAlchemyの使い方の雰囲気を掴めると思います。

まず、ドキュメントの情報が登録されるdocumentsテーブルを定義します。

from sqlalchemy import Boolean, Column, DateTime, String

from sqlalchemy.dialects.postgresql import UUID

from sqlalchemy.sql import text

class Document(Base):

# テーブル名を定義

__tablename__ = "documents"

# テーブルのカラムを定義

file_id = Column(

UUID(as_uuid=True),

primary_key=True,

nullable=False,

index=False,

server_default=text("gen_random_uuid()"),

)

document_name = Column(String, nullable=False, index=True)

status = Column(String, nullable=False, index=False)

is_active = Column(Boolean, nullable=False, index=False)

registered_at = Column(DateTime, nullable=False, index=False)

permalink = Column(String, nullable=False, index=False)

次に先ほど定義したdocumentsテーブルに情報を登録するための関数を実装します。

from datetime import datetime

from sqlalchemy.ext.asyncio import AsyncSession

async def register_document(

db: AsyncSession,

document_name: str,

download_url: str,

) -> Document:

document = Document(

document_name=document_name,

status="registered",

is_active=True,

registered_at=datetime.now(),

permalink=download_url,

)

db.add(document)

await db.commit()

await db.refresh(document)

return document

最後にAPIエンドポイントを実装します。

from pydantic import BaseModel

from typing import Annotated

from fastapi import Depends, Form, UploadFile, status

from sqlalchemy.ext.asyncio import AsyncSession

class DocumentRegisterResponse(BaseModel):

detail: str

@router.post(

"/register", response_model=DocumentRegisterResponse, status_code=status.HTTP_200_OK

)

async def register_documents(

db: Annotated[AsyncSession, Depends(get_db)],

documents: list[UploadFile],

tags: Annotated[list[str], Form()] = [],

):

# ... (省略) ... #

db_file = await register_document(

db, document_name, download_url

)

# ... (省略) ... #

return {"detail": "ドキュメントの登録完了"}

6. FastAPIとSQLAlchemyを使った所感

今回の開発でFastAPIとSQLAlchemyを使用し印象的だったのは、習得するのが非常に簡単だったことです。
FastAPIはソースコードが非常にシンプルで実装しやすく、一方でSQLAlchemyもテーブルの定義やCRUD操作の実装が直感的で理解しやすかったです。
また、両方ともPythonで実装できた点も学習コストを抑えられた要因だった思います。