概要

https://zenn.dev/keitakn/scraps/49eb2616e82eb9 でLLMとの会話を行う為のAPIを実装しているが、そろそろコードベースが肥大化してきつくなってきた。

そこでオニオンアーキテクチャの導入を試してみる。

以下を参考に進めてみる。

厳密にやると時間がかかるので、ある程度は簡略化する予定です。

keitakn

LLMから応答を取得する部分をリポジトリパターンを使って分割

src/domain/repository/cat_message_repository_interface.py というファイルを用意した。

内容は下記の通り。

from typing import Protocol, List, TypedDict, AsyncGenerator
from domain.message import ChatMessage


class CreateMessageForGuestUserDto(TypedDict):
    user_id: str
    chat_messages: List[ChatMessage]


class CatResponseMessage(TypedDict):
    ai_response_id: str
    message: str


class CatMessageRepositoryInterface(Protocol):
    async def create_message_for_guest_user(
        self, dto: CreateMessageForGuestUserDto
    ) -> AsyncGenerator[CatResponseMessage, None]:
        ...

次に src/infrastructure/repository/cat_message_repository.py を実装。

これは CatMessageRepositoryInterface を満たすような実装とする。

import os
from typing import AsyncGenerator
from openai import ChatCompletion
from domain.repository.cat_message_repository_interface import (
    CreateMessageForGuestUserDto,
    CatResponseMessage,
)


class CatMessageRepository:
    def __init__(self) -> None:
        self.OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

    async def create_message_for_guest_user(
        self, dto: CreateMessageForGuestUserDto
    ) -> AsyncGenerator[CatResponseMessage, None]:
        response = await ChatCompletion.acreate(
            model="gpt-3.5-turbo-0613",
            messages=dto.get("chat_messages"),
            stream=True,
            api_key=self.OPENAI_API_KEY,
            temperature=0.7,
            user=dto.get("user_id"),
        )  # type: ignore

        ai_response_id = ""
        async for chunk in response:
            chunk_message = (
                chunk.get("choices")[0]["delta"].get("content")
                if chunk.get("choices")[0]["delta"].get("content")
                else ""
            )

            if ai_response_id == "":
                ai_response_id = chunk.get("id")

            if chunk_message == "":
                continue

            chunk_body: CatResponseMessage = {
                "ai_response_id": ai_response_id,
                "message": chunk_message,
            }

            yield chunk_body

まだ他の部分のリファクタリングが終わっていないので現時点では CatMessageRepositoryInterface に意味はないが、他の層は基本的に CatMessageRepositoryInterface に依存するようにする事でMockへの差し替えや裏側が他のLLMなどになっても CatMessageRepository の利用側には影響はない。

これがRepositoryパターンの基本だが、難しい場合は無理に抽象化する必要もないと個人的には思う。

例えばRBDなんかも実際に乗り換えが発生するパターンは稀だし、LLMに関しても今はOpenAI前提で作っているので、実際に全てのLLMに対応させるようなインターフェースを切るのは結構難しい。

とは言えMockに置き換え可能になるとテストの書きやすさが段違いに上がるので、一応対応しておく。

ちなみにPythonはインターフェースを構文レベルでサポートしていない。

今回の実装も ABC と Protocol を使うパターンがあるようだが、インターフェースの実装という視点で見ると Protocol を利用するのが適切なように思う。

poetry run python -m mypy --strict src/infrastructure/repository/cat_message_repository.py でちゃんとインターフェースと実装の乖離がないかは確認済み。

以下は対応時のPR。

keitakn

ChatCompletion.acreate の部分は mypy --strict でエラーを回避する事が難しかったので # type: ignore コメントの追加で回避してある。

あまりにもこのコメントが増えてくると mypy の設定自体見直したほうが良いという話になるが、このような外部ライブラリを含む場合はある程度許容して良いと個人的には思う。

keitakn

mypyによる型チェックを導入

コードの整理もかねて mypy で型チェックを行うようにした。

mypy --strict で実行しているので、そこそこ厳しい。

後付だと↓のような小さなコードベースでもそこそこ大変だった。この手の物はプロジェクトの初期に入れておくのが一番良いと思う。

以下は変更時のPR。

keitakn

HTTPのリクエスト処理、レスポンス生成ロジックをプレゼンテーション層に移動

router と controller 2つを作成、ルーティングの処理とHTTPリクエストを受け取ってレスポンスを生成する部分を controller に移動した。

ちなみにFastAPIだとHTTPのリクエストやレスポンスを pydantic を使って表現しているが pydantic や FastAPI等の外部ライブラリに依存するのはこの層にとどめたい。

これも本来のオニオンアーキテクチャであればインフラストラクチャ層に実装を作ってプレゼンテーション層はインターフェースを介して使うのが正式なやり方だが今回は時間が足りない事もあり、プレゼンテーション層から直接FastAPIのHTTP関連のpackageを利用している。

これだけでも main.py のコード量を大幅に削る事が出来た。

また今までインフラストラクチャ層にあったログ出力系の機能を新しく作成した log というpackageに移動。

この log はどの層からも呼ばれる可能性がある。

ドメイン層に定義すればどこからでも使えるがロギングがビジネスロジックかと言われると、そうではないと思うので独自packageを作成する事にした。

以下のように独自packageを作成して良いという意見もあるし、ログ出力は標準のlogger以外には依存関係はないので問題ないと思われる。

以下は対応時のPR（他にも微妙にリファクタリングを進めていたりする）

keitakn

ユースケース層の実装

GenerateCatMessageForGuestUserUseCase を実装。

GenerateCatMessageForGuestUserControllerの処理からHTTPに関わる物以外はこの GenerateCatMessageForGuestUserUseCase に移動した。

ユースケース層はテストの事を考えて、DBライブラリへの依存は避けていのでリポジトリだけでなくトランザクション等を実行する為の DbHandlerInterface を追加して直接DBライブラリへの依存が発生しない形にしている。

これでMockによるテストコードの実装が容易になったので次回はテストコードの実装を行う。

keitakn

全体方針まとめ

各層の大まかな解説

ブートストラップ `src/main.py`

HTTPリクエストを受けて最初に呼ばれるファイルがこれ。

FastAPIの起動とプレゼンテーション層のrouterを読み込んで、後の処理をプレゼンテーション層に任せる。

本当は @app.exception_handler(status.HTTP_401_UNAUTHORIZED) 等のカスタムハンドラも別のファイルにしたかったが今回はここで妥協している。

プレゼンテーション層

ルーティングを担う src/presentation/router/ と Controllerを担う src/presentation/controller/ の2つのサブパッケージを持つ。

ルーティングは文字通り、HTTPのURLのルーティングを行う、ControllerはHTTPリクエストの処理とHTTPレスポンスの処理を行う。

ドメイン層を用いてビジネスロジックを実現する処理はユースケース層に任せる。

プレゼンテーション層にはHTTPレスポンスを生成する処理が集まっている、故にServer Sent Events(SSE)のレスポンスを生成する為の関数郡やBasic認証等のAPIに対する認証もこの層の責務である。

以下は実際の実装例。

なおHTTPを扱う関係上、FastAPI等のフレームワークが提供する機能を利用する機会が多い。（pydanticとか）
本当はこれもインフラストラクチャ層を通じて利用するのが正しいのだが、結構手間がかかるのでここは普通にFastAPIの処理を利用する事にした。

同じくインフラ層のDBコネクションをここで作っていたりもしているが、これも本当はインフラ層に閉じ込めてプレゼンテーション層ではインターフェースに依存するように作るべきだが、ここも省略する事にした。

ユースケース層

src/usecase/ ドメイン層の機能を組み合わせてビジネスロジックを実現する為の層。

この層が各機能の全体の処理の流れを確認出来るので、テストコードはこの層に書くのが最も効率が良い。

実装例は以下の通り。

このアプリケーションではLLMの応答をStreamingで返すという性質上、Streamingの途中でエラーを投げる事が出来ないので辞書で返しているが、通常のAPIの処理の場合は例外を投げても良い。

ただしその場合、汎用Errorクラスを使うのではなく、必ずビジネス上意味のあるErrorを定義してからそれを使うようにする。

インフラストラクチャ層

技術的な問題を解決する為の層。必然的に外部ライブラリに依存する事が多くなる。

例えば以下のようなOpenAI周りの処理等が実装される。

リポジトリに関しては後で記載する。

ドメイン層

ビジネスロジックが実装される外部ライブラリへの依存は原則として行わない。

アーキテクチャに従えば、フレームワークを乗り換える際にもドメイン層（ユースケース層も）だけはそのまま利用する事が出来るハズ。

ちなみに標準ライブラリには依存するように作っている、本当はこれもインターフェースだけ定義してインフラ層で実装したほうが良いのだろうが、外部ライブラリではなく標準パッケージであれば問題ないとしている。

ちなみに時間が取れなかったのでエンティティやバリューオブジェクト等のパターンでは実装していない。

しかしこれでもビジネスロジックをドメイン層に閉じ込めるという最低限の事は出来ている。

その他

リポジトリパターン

以下のデザインパターン。

インターフェースをドメイン層に作り、実装はインフラストラクチャ層に作成する。

以下の2つはインターフェース。

それぞれ会話履歴とねこの人格を設定したLLMとの会話を扱う。

以下はインフラストラクチャ層の実装。

会話履歴は aiomysql というライブラリで実装されているのでファイルパスは src/infrastructure/repository/aiomysql/aiomysql_guest_users_conversation_history_repository.py となっている。

一方でLLMのほうは openai のライブラリを使っているので src/infrastructure/repository/openai/openai_cat_message_repository.py となっている。

ユースケース層はドメイン層のリポジトリのインターフェースにのみ依存しているので、裏側の実装が変わっても（例えばDB操作が SQLAlchemy , LLMの実装が LangChain に変更される等）ユースケース層は変更の必要はなく、リポジトリをMockに置き換える事でテストコードを書く事も容易になる。

と理屈はこうだが、以外と汎用的なインターフェースを作るのが結構難しい。

特にLLMの開発等の比較的新しい物を扱う場合はつい特定のライブラリ（openaiなど）に依存したインターフェースになってしまう事もある。

とは言えテストを書くのが楽になるというメリットはあるので一定の有効性があるデザインパターンではある。

ちなみに変更可能性が薄いのであれば無理にリポジトリパターンを使って抽象化する必要はないと思う。

特にDBに関してはコンテナ等でテスト用のDBを用意にする事が比較的簡単なので、テストの実行が遅くなる等のデメリットを受け入れられればリポジトリパターンを使わないという選択もアリ。

keitakn

テスト戦略

https://zenn.dev/link/comments/cc1190870bcc6d にも書いたが基本的にはユースケース層にテストを書くのが良い。

ここにしっかりテストを書けば、一番重要なドメイン層のコードカバレッジもある程度増えるし、ドメイン層のリファクタリング等を実施した時も機能が壊れていない事が保証しやすい。

以下はユースケースのテストコード例。基本的には、DBやAPI等へのアクセスはRepositoryを利用して行うのでRepositoryのインターフェースを元にMock版のRepositoryを作っている。

ユースケース層のテストが基本ではあるが、場合によってはリポジトリやドメイン層の複雑なロジックのテストも必要に応じて書いていく。

リポジトリはデータベース等の操作だけでなく、データベースのデータを元にドメイン層のドメインモデルを組み立てて返すという部分まで実装されている事が多いので、それらのロジックが複雑なのであればテストを用意しておいたほうが良い。

以下はDBから取得した会話履歴を元に一定の長さのメッセージリストを作成するRepositoryのメソッドのテスト。

このようにRepositoryは場合によってはデータの加工でそこそこ複雑なロジックが書かれる事があるので、このようなRepositoryはテストを書いておいたほうが良い。

型チェック

https://zenn.dev/link/comments/506c6bded987ee の通り mypy を使っている。

ただ少し不便だなと感じる事もある。

AsyncGenerator を使った時に変な形で解釈されてしまい意図しない型エラーが出てしまったり、キャストしなければいけない部分が増えたりと実装コストが大きくなる事は確か。

mypy ではなく別の AsyncGenerator に対応した物に変更する可能性はあるが、実装コストが多少大きくなる代わりにバグを防いでくれたり、コードの読みやすさやIDEと組み合わせた時の生産性の向上等のメリットのほうが大きいので基本的には何らかの仕組みで型チェックを実行出来るようにしたほうが良い。

概要

LLMから応答を取得する部分をリポジトリパターンを使って分割

mypyによる型チェックを導入

HTTPのリクエスト処理、レスポンス生成ロジックをプレゼンテーション層に移動

ユースケース層の実装

全体方針まとめ

各層の大まかな解説

ブートストラップ src/main.py

プレゼンテーション層

ユースケース層

インフラストラクチャ層

ドメイン層

その他

リポジトリパターン

テスト戦略

型チェック

ブートストラップ `src/main.py`