生成AIアプリのクリーンアーキテクチャを考える

はじめに

こんにちは、InsightEdgeの開発チームに参画させていただいています伊藤です。 InsightEdgeでは現在、LLM/生成AIを用いたアプリ開発を多く手掛けています。

私もOpenAI等の生成AIを使ったアプリケーションの実装に関わることが増えてきており、ある程度理解が進んできたところで、改めてアーキテクチャを洗練させたいと思うようになりました。

この記事では生成AIを組み込んだアプリを構築する際の、クリーンアーキテクチャを考えていきたいと思います。

クリーンアーキテクチャとは

まずは、クリーンアーキテクチャについて軽く振り返りましょう。

クリーンアーキテクチャではソフトウェアの理解・開発・デプロイ・運用・保守を容易にするために、フレームワークやDB・UI・外部サービスなどへの依存を最小にします。この時の下記のような多層構造に分割し、それぞれの層に役割に合わせたクラスを配置していきます。

ここで登場する層はそれぞれ下記のような役割を持っています

Entities

この層にはビジネスルールやビジネス上の概念を表すクラスを格納します。ドメイン駆動開発(DDD)やオニオンアーキテクチャ的にいうとドメインモデル/ドメインサービスのレイヤに該当します。

Use Cases

アプリケーション固有のビジネスが含まれます。 Use Case層のクラスはEntity層のクラスを使ってアプリケーションの処理フローを実行します。

Controllers/Presenters

Controllerは入力をUse Caseに合わせた形に変換した上で、Use Caseを実行します。 PresenterはUse CaseがUIへの出力をする時に呼び出されるクラスになります。

Gateways

GatewayはEntity層/Use Case層で扱うクラスをDBや外部サービスに合わせた形に変換し、永続化やAPIコールなどを行います。

その外側

DBそのものやUI、外部のサービスなどはすべてControler/Presenter/Gatewayを通じてやり取りするものになります。

これらの各層に含まれるコードは、必ずより内側の層のコードだけに依存するようにすることで、サービス/ビジネスの本質的なロジック部分を表現するコードが、フレームワークやUIといった部分の変更に影響を受けないように構成します。

詳細についてはオリジナルのWebサイトや書籍のほか、いろいろが出ていますのでここでは割愛します。

参考

例として考えるアプリケーション

アーキテクチャを考える上での想定アプリケーションとして、ここではRAGを加えたChatGPTのようなアプリケーションを作るものとします。

ユーザはブラウザでアクセスする
自然言語で質問を入力すると、自然言語で回答が表示される
回答作成時には、それまでのやり取りなどの文脈が考慮される
回答作成時には、あらかじめシステムに登録しておいた関連情報が加味される
回答は生成完了してから一度に表示されるのではなく、生成された文字から逐次表示される

生成AIアプリにおける難しさ

前述のようなアプリをクリーンアーキテクチャで作成しようとした際に、特に下記の点は検討が必要でした。

どの要素がEntityになるのか
ストリーミング形式のアウトプットはクリーンアーキテクチャで実現できるのか

それぞれ、どんな課題に対して、どのように考えていったのかを説明していきたいと思います。

どの要素がEntityになるのか

論点

クリーンアーキテクチャのEntity層にはアプリの核となるビジネスロジックが置かれます。一方で生成AIアプリ開発の中心となるのは、どうやってGPT等の生成AIに望むクオリティのアウトプットを生成してもらうかだと思います。しかし、生成AIそのものはAPI経由で呼び出すのでアプリ外にあり、これはクリーンアーキテクチャで言うところのGatewayの先の部分のはずです。

では、どんな部分をEntityとして依存関係の中心に置くべきなのでしょうか。

考え方

今回想定しているアプリが提供する機能は「質問を受けて、今の文脈に合わせて、関連情報を使って、回答を生成する」です。ここから考えるとEntityとして表現するべき概念は 質問 ・ 文脈 ・ 関連情報 ・ 回答 であるはずです。

ここにはプロンプトなどの生成AI的な要素は出てきませんが、そもそもAIは実装/実現手段の詳細なので、適切な回答を返してくれるのであれば、背後にあるのはGPTでもGeminiでも、ただのデータベースでも良いということです。

これまで開発のたびに色々と工夫してきたプロンプトは本質的な部分ではないのかと一瞬ショックを受けましたがよく考えるとパフォーマンス重視のサービスでSQL部分のチューニングをしているようなものですね。どんなにチューニングが大切なアプリであってもSQLそのものがアーキの中心とはならない、というのと同じです。

ストリーミング形式のアウトプットはクリーンアーキテクチャで実現できるのか

論点

GPTを使ったアプリケーションでは、UX向上のためにGPTからの回答を逐次画面に表示するUIが一般的だと思います。

これを行うためにはアプリのフロントエンドとバックエンドがHTTPストリーミングのような形で通信しなければなりません。このとき、GPTの出力からアプリのフロントエンドまでの間の、回答 に関連するすべてのクラスがストリーミング形式のデータに対応できている必要があります。

しかし、UIがストリーミング出力だからといって、そこに合わせてUse Case層の仕様を変更することはクリーンアーキテクチャの重視している依存関係が逆転してしまっている気がします。

クリーンアーキテクチャの考え方に沿ったままストリーミング型のUIに対応するにはどうしたらよいのでしょうか。

考え方

上記について考えている中で、自分の捉え方に思い込みがあることに気づきました。変数の値が決まってから返却するのが「普通」で、ストリーミング出力は「UIに合わせた特殊な形」だと考えていたのです。

pythonは言語自体にGeneratorやPromise等の「後で値が決まる仕様」の型を持っています。使うプログラミング言語で表現できるのであれば 回答 を「逐次生成されるもの」として定義してしまってもよいのではないでしょうか。 回答 という概念が、そもそも逐次生成される性質を持っている、という捉え方です。

今回は回答(Answer)というEntityをGeneratorの一種として定義しました。

Answer: TypeAlias = Generator[str, None, None]

この書き方もEntityを Generator[str, None, None] 型なんかにしてしまってよいのかという違和感があったのですが、 list[str] 等で表現するのと違いはないはず、と思い納得することにしました。 回答 の持つ性質をコードとして表現した時に、たまたまGeneratorという型が一番適切だったのだと解釈します。

実際に作ってみる

上記を踏まえ、実装をしてみます。 Entity・Use Case・Gateway・Controller/Presenterそれぞれを見ていきましょう。

全体の構成は以下のようになっています。

ドメイン駆動開発(DDD)をする場合だと domain や repository といった表現の方がよく使われるかもしれませんが、今回はクリーンアーキテクチャとの対応がわかりやすいように entity や gateway といった表現に統一しています。

src
├── entity: エンティティ層
│   ├── answer: 回答モデル
│   ├── context: 文脈モデル
│   ├── knowledge: 背景知識モデル
│   └── question: 質問モデル
├── usecase: ユースケース層
│   ├── knowledge: 背景知識モデル
│   └── chat: 質問回答ユースケース
├── adapter: アダプタ層
│   ├── controller: Use Caseを起動するためのController実装
│   ├── gateway: 各種Gateway/Repositoryの実装
│   │   ├── answer: AnswerGateway実装
│   │   ├── context: ContextGateway実装
│   │   └── knowledge: KnowledgeGateway実装
│   └── presenter: 出力をハンドリングするPresenter実装
└── main_*.py: 起動のためのエントリポイント

Entity定義

まずコアとなるEntitiy層には、前述した下記の4要素と、その取得を担当するGatewayの抽象クラスを作成します。

質問(Question)
文脈(Context)
背景知識(Knowledge)
回答(Answer)

質問(Question)関連

question.py

Question: TypeAlias = str
""" 質問文を表す型。ただの文字列 """

文脈(Context)関連

context.py

ContextId: TypeAlias = int
"""
これまで文脈を取得するためのキーにする値を表す型。今回は整数として定義する。
実際はユーザやセッションに関する値をキーする想定。
"""

Context: TypeAlias = str
"""
文脈を表す型。今回はただの文字列として定義する。
実際は過去の会話の履歴などを含むことが考えられる。
""

context_gateway.py

class ContextGateway(metaclass=ABCMeta):
    """文脈(Context)を取得するためのGateway"""

    @abstractmethod
    def get_context(self, context_id: ContextId) -> Context:
        """ContextIdに基づくContextを取得する"""
        pass

背景知識(Knowledge)関連

knowledge.py

Knowledge: TypeAlias = str
"""
背景知識１つを表す型。今回はただの文字列として定義する。
実際は参照先や、関連度のスコアなどを含むクラスにすることが考えられる。
"""

Knowledges: TypeAlias = list[Knowledge]
""" 複数の背景知識を表す型 """

knowledge_gateway.py

class KnowledgeGateway(metaclass=ABCMeta):
    """背景知識(Knowledge)を取得するためのGateway"""

    @abstractmethod
    def find_knowledges(self, question: Question) -> Knowledges:
        """Questionに関連する背景知識(Knowledge)を取得する"""
        pass

回答(Answer)関連

answer.py

Answer: TypeAlias = Generator[str, None, None]
""" 回答を表す型。文字列が逐次生成されるものとして定義する。 """

answer_gateway.py

class AnswerGateway(metaclass=ABCMeta):
    """回答(Answer)を取得するためのGateway"""

    @abstractmethod
    def get_answer(
        self, question: Question, context: Context, knowledges: Knowledges
    ) -> Answer:
        """Question/Context/Knowledgeに基づくAnswerを取得する"""
        pass

Gateway実装 (ダミー版)

まずは動作確認するためにダミー版のGateway実装を作成します。それぞれ固定値を返すようになっています。

今回は以下の例でダミー文を作成しました。

文脈:
- 「組織のデータを活用するためのOpenAIアプリの開発で困っている」
背景知識：
- 「InsightEdge社はOpenAIを使ったアプリ開発ができる」
- 「InsightEdge社にはデータサイエンスの専門家がいる」
回答:
- 「InsightEdgeを検討してみてはいかがでしょうか？」

dummy_context_gateway.py

class DummyContextGateway(ContextGateway):
    """ ダミーの文脈情報を返すContextGateway """

    def get_context(self, context_id: ContextId) -> str:
        return Context("組織のデータを活用するためのOpenAIアプリの開発で困っている")

dummy_knowledge_gateway.py

class DummyKnowledgeGateway(KnowledgeGateway):
    """ ダミーの背景知識を返すKnowledgeGatway """

    def find_knowledges(self, question: Question) -> Knowledges:
        return [
            Knowledge("InsightEdge社はOpenAIを使ったアプリ開発ができる"),
            Knowledge("InsightEdge社にはデータサイエンスの専門家がいる"),
        ]

dummy_answer_gateway.py

class DummyAnswerGateway(AnswerGateway):
    """ ダミーの回答を返すAnswerGateway """

    def get_answer(
        self, question: Question, context: Context, knowledges: Knowledges
    ) -> Generator[str, None, None]:
        yield f"「{context}」という文脈において、\n"
        sleep(1)

        yield f"「{question}」という質問に対して、\n"
        sleep(1)

        for knowledge in knowledges:
            yield f"「{knowledge}」"
        sleep(1)

        yield "という背景知識をもとに、\n"
        sleep(1)

        yield "ダミーの回答をします。\n"
        sleep(1)

        yield "InsightEdgeを検討してみてはいかがでしょうか？\n"

get_answer() の戻り値は Generator である Answer として親クラスで定義されていますが、こちらの実装は str 型を yield することで定義を満たしています。なお、今回はストリーミングの動きがわかるように sleep() を挟んでいます。

Use Case/Port実装

続いて、Use Case層を作ります。

ここではEntityを使用してアプリとしての処理フローを定義するUse Caseと、その出力の仕方を定義するOutputPortを作成します。

後ほど出てくるControllerはこのUse Caseを呼び出す形となり、 PresenterはOutputPortを継承したクラスとして、Use Case層の定義に従って実装することになります。

これはクリーンアーキテクチャの図では右下の部分で表現されている部分となります。

図ではUseCaseの親となるInput Portも記載されていますが、今回は省略します。

図に厳密に従ってUse Case Input Portを作成しControllerからの依存がそちらに向かう構造にすることで、 Use Caseの変更をした場合にControllerに影響しにくくできますが、把握しておくクラス/依存関係を減らすことの方がメリットが高いと考えており自分の場合はInput Portは作らない、という選択をすることが多いです。

では、実装の中身です。

まずは「質問を受けて回答を生成する」というユースケースを実現するためのクラスを作ります。

chat_usecase.py

class ChatUsecase:
    """
    質問に対する回答を返すユースケース
    """

    def __init__(
        self,
        context_gateway: ContextGateway,
        knowledge_gateway: KnowledgeGateway,
        answer_generator: AnswerGateway,
    ) -> None:
        """
        使用するGatewayをセットする
        """
        self.context_gateway = context_gateway
        self.knowledge_gateway = knowledge_gateway
        self.answer_generator = answer_generator

    def execute(
        self,
        question: Question,
        context_id: ContextId,
        output_port: ChatUsecaseOutputPort,
    ) -> None:
        """
        ユースケースを実行する
        """

        # 文脈を把握する
        context: Context = self.context_gateway.get_context(context_id)

        # 背景知識を取得する
        knowledges: Knowledges = self.knowledge_gateway.find_knowledges(question)

        # 回答を作成する
        answer = self.answer_generator.get_answer(question, context, knowledges)

        # 回答を出力する
        output_port.emit(answer)

そしてUse Caseのアウトプットを受け取るOutput Portを定義します。このOutputPortを実装したPresenterであれば、アウトプットの仕方はなんでも良いということになります。

chat_usecase_output_port.py

class ChatUsecaseOutputPort(metaclass=ABCMeta):
    """ ChatUsecaseの出力口の定義"""

    @abstractmethod
    def emit(self, answer: Answer) -> None:
        """Answerを出力用に受け取る"""
        pass

Presenter/Controller（コマンドライン実行用）

最後にPresenterとControllerを作成します。まずはシンプルにコマンドラインで呼び出す想定で作ってみます。

Controllerは標準入力から質問を受け付け、Entityである Question 型に変換してから、ユースケースを起動します。

cli_chat_controller.py

class CliChatController:
    # CLI上で質問を受け取り、ユースケースを実行するコントローラ

    def __init__(self, usecase: ChatUsecase, presenter: ChatUsecaseOutputPort):

        # 実行するユースケース
        self.usecase = usecase

        # ユースケースのOutputPortの実装
        self.presenter = presenter

    def handle(self):

        # コンテクストIDを取得する（今回はサンプルなので固定）
        contextId = 1

        # 質問文を受け取る
        question_str = input("質問を入力してください: ")
        question: Question = Question(question_str)

        # ユースケースを実行する
        self.usecase.execute(question, contextId, self.presenter)

        # 出力はPresenterが担当するので、ここでは何もしない

続いてPresenterを作ります。 CLIでの操作を想定し標準出力への表示をするものを用意します。ChatUsecaseOutputPort の実装(の1つ)になります。

stdout_chat_presenter.py

class StdoutChatPresenter(ChatUsecaseOutputPort):
    """標準出力にAnswerを出力するPresenter"""

    def emit(self, answer: str) -> None:
        """Answerを出力用に受け取る"""

        # Answerの内容を逐次出力する
        for answer_chunk in answer:
            sys.stdout.write(answer_chunk)
            sys.stdout.flush()

実行

さてここまでの実装を繋げて動かしてみます。

エントリポイント

エントリポイントとして、手動でDependency Injection(以下DI)した上で、コントローラを呼び出す処理を作ります。

main_cli.py

"""
CLI上で実行するためのエントリーポイント
"""

# DI済みのUse Caseを作成する
context_gateway = DummyContextGateway()
knowledge_gateway = DummyKnowledgeGateway()
answer_gateway = DummyAnswerGateway()
usecase = ChatUsecase(context_gateway, knowledge_gateway, answer_gateway)

# DI済みのコントローラを作成する
presenter = StdoutChatPresenter()
controller = CliChatController(usecase, presenter)

# コントローラを呼び出す
controller.handle()

実行結果

上記のエントリポイントからアプリを動かすとこのようになります。

DummyAnswerGateway によって生成された結果が、標準出力に逐次出力されています。

さて、これでアプリの全体像は完成しました。いわゆる生成AIアプリであっても、アーキテクチャの観点ではAIそのものからは切り離されているということが改めて確認できたと思います。

実装を差し替えてみる

ここからはGateway/Controller/Presenterの「詳細」を変えていきましょう。クリーンアーキテクチャを目指したことで変更に強いものになっているでしょうか。

AnswerGatewayをOpenAI利用版にしてみる

まずは回答の生成を実際にAzure OpenAIを使って行うようにしてみます。

DummyAnswerGateway を置き換えるための別クラスを作ります。親クラスは AnswerGateway です。

openai_answer_gateway.py

class OpenAIAnswerGateway(AnswerGateway):
    """ OpenAIを用いて回答を生成するAnswerGateway"""

    def __init__(self):

        api_key = os.getenv("AZURE_OPENAI_API_KEY", "")
        api_base = os.getenv("AZURE_OPENAI_API_BASE", "")
        api_version = os.getenv("AZURE_OPENAI_API_VERSION", "")
        self.deployment_name = os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME", "")

        # AzureOpenAIクライアントを用意する
        self.client = AzureOpenAI(
            api_key=api_key,
            api_version=api_version,
            azure_endpoint=api_base,
        )

    def get_answer(
        self, question: Question, context: Context, knowledges: Knowledges
    ) -> Answer:

        # OpenAIに質問を投げて回答を取得する
        response = self.client.chat.completions.create(
            model=self.deployment_name,
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": f"{context}という文脈において、"
                            + f"{knowledges}という背景知識をもとに、"
                            + f"{question}という質問に対して回答してください。",
                        }
                    ],
                },
            ],
            stream=True,
        )

        # OpenAIからの回答を逐次返す
        for chunk in response:
            choices = chunk.choices
            if choices and choices[0].delta and choices[0].delta.content is not None:
                yield str(chunk.choices[0].delta.content)
                # strをyieldすることで、Answer型つまりGenerator[str, None, None]型を満たす

こちらの get_answer() も DummyAnswerGateway の時と同じく str 型を yield しているので、戻り値の型 Answer 型に合致しています。

エントリーポイント

エントリーポイントは、先ほどのCLI入力/表示用の実装から、DummyAnswerGateway の代わりに上記の OpenAIAnswerGateway を使って実行するよう、DIしている1行だけを変更します。

今回は簡易化のため ContextGateway と KnowledgeGateway はダミーのままとします。実際にRAGを行う場合は KnowledgeGateway の部分にEmbeddingやVectorStoreを実装することになると思います。

main_cli_openai.py

"""
CLI上で実行するためのエントリーポイント
"""

# DI済みのUseCaseを作成する
context_gateway = DummyContextGateway()
knowledge_gateway = DummyKnowledgeGateway()
answer_gateway = OpenAIAnswerGateway()          # <= OpenAIAnswerGatewayを使う
usecase = ChatUsecase(context_gateway, knowledge_gateway, answer_gateway)

# DI済みのコントローラを作成する
presenter = StdoutChatPresenter()
controller = CliChatController(usecase, presenter)

# コントローラを呼び出す
controller.handle()

実行結果

上記のエントリポイントから実行してみます。

はい、これだけで実際にAzure OpenAIを組み込んだアプリに変わりました。

以下のポイントのおかげで修正箇所が非常に限定されています。

各Gatewayの実装はEntity層で定義した抽象Gatewayに従うようになっている
Use Caseも抽象のGatewayの定義に依存している
実際にどの実装を使うかはDIによって決まる

Presenter/ControllerをHTTPストリーム版にしてみる

次はControllerとPresenterを差し替えて、APIとして動作する形にしてみましょう。

ControllerはFastAPIを使ってHTTPリクエストを受け取るものを作成します。

http_chat_controller.py

class HttpChatController:
    # Httpリクエストで質問を受け取り、ユースケースを実行するコントローラ

    def __init__(self, usecase: ChatUsecase, presenter: AbstractHttpChatPresenter):

        # 実行するユースケース
        self.usecase = usecase

        # ユースケースのOutputPortの実装
        self.presenter = presenter

        # FastAPIのルーター設定
        self.router = APIRouter()
        self.router.add_api_route("/{context_id}", self.handle, methods=["GET"])

    def handle(self, context_id: int, q: str) -> Response:

        # 質問文を取得する
        question = Question(q)

        # ユースケースを実行する
        self.usecase.execute(question, context_id, self.presenter)

        # レスポンスを返す
        return self.presenter.get_response()

CliChatControllerとほぼ一緒ですね。

Presenterとしては ChatUsecaseOutputPort を継承しつつ Response の形で取り出すための get_response() を持たせたいので、その点を表現した抽象クラスをもう一段階挟みます。

class AbstractHttpChatPresenter(ChatUsecaseOutputPort):
    """HTTP経由でAnswerを出力するPresenterの抽象クラス"""

    @abstractmethod
    def emit(self, answer: Answer) -> None:
        """Answerを出力用に受け取る"""
        pass

    @abstractmethod
    def get_response(self) -> Response:
        """HTTPレスポンスの形で回答を返す"""
        pass

Presenterの実装は emit() で設定された Answer を HttpStream で返却するものを用意します。

http_stream_chat_presenter.py

class HttpStreamChatPresenter(AbstractHttpChatPresenter):
    """HTTPストリーミングでAnswerを出力するPresenter"""

    def emit(self, answer: Answer) -> None:
        """Answerを出力用に受け取る"""
        self.answer = answer

    def get_response(self) -> Response:
        """HTTPストリームの形で回答を返す"""
        response = StreamingResponse(self.answer, media_type="text/event-stream")
        return response

こちらはFastAPIの定義する StreamingResponse を使っています。 Entity層で定義した Answer が Generator 型なので、簡単にストリーミング対応ができます。

エントリーポイント

上記のControllerをFastAPIで実行するようなエントリポイントを用意します。変更箇所は、Controler/Presenterの差し替えとサーバの起動処理の追加です。

main_http_openai.py

"""
Httpサーバ上で実行するためのエントリーポイント
"""

# DI済みのUseCaseを作成する
context_gateway = DummyContextGateway()
knowledge_gateway = DummyKnowledgeGateway()
answer_gateway = OpenAIAnswerGateway()  # OpenAIを使って回答を行うAnswerGateway
usecase = ChatUsecase(context_gateway, knowledge_gateway, answer_gateway)

# DI済みのコントローラを作成する
presenter = HttpStreamChatPresenter()
controller = HttpChatController(usecase, presenter)

# コントローラをサーバ上で呼び出し待ち状態にする
app = FastAPI()
app.include_router(controller.router)
uvicorn.run(app, host="127.0.0.1", port=8000)

実行結果

上記を起動した上で、ブラウザでアクセスしてみます。

無事にAPI化もできました。

アプリの処理そのものはUse Case層がEntity層のクラスを用いて行う
Controller/PresenterはUse Case層のクラスに従う

この2点のおかげで、入出力をCLIからHTTP形式の変更する際に、Use Case層/Entity層に影響を与えることなく行えています。

ストリーミングではないHTTPレスポンスにしたい場合はどうするのか

ここまでは内部のEntityも出力のPresenterも回答を逐次処理するようにしていました。では「やっぱりストリーミングではなく一度のHTTPレスポンスで回答を返却したくなった」という場合はどのような影響があるでしょうか。

出力形式が変わるので、新たにPresenterを用意します。 HTTP Responseを返す get_response() を持っていて欲しいので、AbstractHttpChatPresenter の子クラスとして作成します。

http_response_chat_presenter.py

class HttpResponseChatPresenter(AbstractHttpChatPresenter):
    """HTTPレスポンスでAnswerを出力するPresenter"""

    def emit(self, answer: Answer) -> None:
        """Answerを出力用に受け取る"""
        self.answer = answer

    def get_response(self) -> Response:
        """HTTPレスポンスの形で回答を返す"""

        result = ""
        for chunk in self.answer:
            result += chunk
            print(result)

        response = Response(content=result, media_type="text/plain")
        return response

Entity層のAnswer は Generater 型のまま、生成される回答をすべてPresenter内で受け取ったあとに1回のHTTPレスポンスを返すようにしています。

エントリーポイント

呼び出しはHTTP経由で変わらないので、HttpChatController のまま、PresenterだけDIで変更します。

"""
Httpサーバ上で実行するためのエントリーポイント
"""

# DI済みのUseCaseを作成する
context_gateway = DummyContextGateway()
knowledge_gateway = DummyKnowledgeGateway()
answer_gateway = OpenAIAnswerGateway()
usecase = ChatUsecase(context_gateway, knowledge_gateway, answer_gateway)

# DI済みのコントローラを作成する
presenter = HttpResponseChatPresenter()  # <= 通常レスポンスを返すPresenterに変更
controller = HttpChatController(usecase, presenter)

# コントローラをサーバ上で呼び出し待ち状態にする
app = FastAPI()
app.include_router(controller.router)
uvicorn.run(app, host="127.0.0.1", port=8000)

これでストリーミングではなく、一括で回答が返却される形になりました。 Entityである Answer が Generater として定義されていても、Presenterが適切に処理することで外からはただの str であるかのように表示に使うことができます。

余談

今回は実施しませんでしたが、バッチ処理で回答を生成して何かに格納したい場合も、バッチ処理用のControllerとPresenterを用意することで対応できるはずです。ただしその場合には、Use Caseとして表現するべきものが本当にユーザ操作とバッチ処理で同じなのかを検討する必要があります。

まとめ

今回は生成AIアプリの内部をクリーンアーキテクチャで構成する方法について検討しました。

生成AIを使ったアプリであっても、クリーンアーキテクチャの考え方に沿って層を分け、アプリが扱う要素/概念を生成AI自体の使い方と切り離して捉えて抽象化し、依存関係が一方向に向かうように構成することで

回答生成方法の変更
アプリ起動方法の変更
出力形式の変更

などを、影響範囲を最小にした上で実施できることが確認できました。

また回答(Answer)をGeneratorとしての性質を持ったものとしてモデル化するという工夫をすることで、生成AIアプリケーションでよく使う逐次表示形式と、Request/Response形式の両方の出力に対応できることが分かりました。

感想

今回クリーンアーキテクチャを目指す設計をしたことで生成AIアプリケーションについて自分が持っていたメンタルモデルが、よりシンプルなものに更新されたと感じます。

新たにアプリを作るときよりも、既存のアプリの方が概念やフローが分かっていて検討しやすいと思いますので既存のアプリについてクリーンアーキテクチャで作ったらこうなるのでは、というのを一度考えてみるのも面白いかもしれません。

目次

はじめに

クリーンアーキテクチャとは

Entities

Use Cases

Controllers/Presenters

Gateways

その外側

参考

例として考えるアプリケーション

生成AIアプリにおける難しさ

どの要素がEntityになるのか

論点

考え方

ストリーミング形式のアウトプットはクリーンアーキテクチャで実現できるのか

論点

考え方

実際に作ってみる

Entity定義

質問(Question)関連

文脈(Context)関連

背景知識(Knowledge)関連

回答(Answer)関連

Gateway実装 (ダミー版)

Use Case/Port実装

Presenter/Controller（コマンドライン実行用）

実行

エントリポイント

実行結果

実装を差し替えてみる

AnswerGatewayをOpenAI利用版にしてみる

エントリーポイント

実行結果

Presenter/ControllerをHTTPストリーム版にしてみる

エントリーポイント

main_http_openai.py

実行結果

ストリーミングではないHTTPレスポンスにしたい場合はどうするのか

エントリーポイント

余談

まとめ

感想