🧑‍💻
関数型プログラミングをPythonで実践しよう - #1 まずは書いてみる -

naoto.a.kimura
2024/09/24に公開
change log2024/09/24 初版公開

 はじめに筆者の周りで ちらほら 関数型プログラミングの話を聞くようになりました。それも、ドメイン駆動開発と合わせた形で耳に入ることが多い気がします。
https://youtu.be/qBf910_YiqA?si=z-CQxLv-FpcB48j4
TypeScriptを使ったバックエンド開発を題材に、関数型プログラミングとドメインの考え方について説明されています。わかりやすく、ワクワクできるのでぜひ！
色々と調べていると、自分で書いてみたくなる...。
ということで、本記事では、Pythonを用いて関数型プログラミングの基本概念を解説し、実際にプログラムを作成しながらその実践方法を整理します。
関数型プログラミングは、値の変更を許容しない「イミュータブルなデータ」と「純粋関数」を中心としたアプローチで、コードの安全性や再利用性を高める手法です。

 対象読者本記事は、以下のような方々に向けて書かれています。
関数型プログラミングの基本に触れて、実際に試してみたい方
Pythonで関数型プログラミングを始めたいが、どこから手を付ければよいか迷っている方

 対象外の読者以下のような方々には、本記事の内容が適しておらず、「自分向けじゃなかった...」となる可能性があります。
数学的に厳密な関数型プログラミング理論を深く学びたい方
既に関数型プログラミングを習得しており、より高度な技術に挑戦したい方

 本記事を読むことで得られること本記事を読み終えると、以下のことができるようになります。
関数のシグネチャに全ての情報が表現されている「嘘をつかない関数」の重要性を説明できる
関数型プログラミングへの理解と興味が深まる
Pythonでの関数型プログラミングの基本的な実装への知見が得られる

 実装内容Pythonを使って、OpenAI APIを活用したコンソールベースのチャット機能を実装します。オブジェクト指向での実装と関数型プログラミングでの実装の違いを比較しながら、関数型プログラミングの特徴を学びましょう！

 オブジェクト指向での実装オブジェクト指向では、役割ごとにクラスを作成し、クラス内にその役割を果たすために必要な処理（メソッド）を実装します。本記事ではChatBotクラスを作成し、このクラス内にチャット機能を実装します。
oop_chat.py
import httpx


class ChatBot:
    def __init__(self, api_key: str):
        """
        ChatBotクラスのコンストラクタ。APIキーを設定し、メッセージ履歴を初期化します。

        Args:
            api_key (str): OpenAI APIキー。
        """
        self.api_key = api_key
        self.messages = []  # メッセージ履歴を保持するリスト

    def _create_and_send_message(self, user_message: str) -> str:
        """
        ユーザーからのメッセージを基にOpenAIとやり取りし、アシスタントからのレスポンスを取得します。

        Args:
            user_message (str): ユーザーからのメッセージ。

        Returns:
            Optional[str]: アシスタントからのメッセージ。失敗した場合はNone。
        """
        # メッセージ履歴にユーザーのメッセージを追加
        self.messages.append({"role": "user", "content": user_message})

        # OpenAI APIに送信するペイロードを作成
        payload = {
            "model": "gpt-4o",
            "messages": self.messages,
            "temperature": 0.7,
        }

        # OpenAI APIへのリクエスト
        url = "https://api.openai.com/v1/chat/completions"
        headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}

        try:
            response = httpx.post(url=url, headers=headers, json=payload, timeout=None)
            data = response.raise_for_status().json()
        except httpx.HTTPStatusError as exc:
            print(
                f"エラーが発生しました。ステータスコード: {exc.response.status_code}, メッセージ: {exc.response.text}"
            )

        # アシスタントのレスポンスを取得してメッセージ履歴に追加
        assistant_message = data["choices"][0]["message"]["content"]
        self.messages.append({"role": "assistant", "content": assistant_message})

        return assistant_message

    def start_chat(self) -> None:
        """
        LLMとのチャットを開始します。
        """
        while True:
            user_message = input("メッセージを入力してください (`/end` で終了): ")
            if user_message == "/end":
                break

            assistant_message = self._create_and_send_message(user_message)
            if assistant_message:
                print(f"LLM: {assistant_message}")


if __name__ == "__main__":
    import os
    from dotenv import load_dotenv

    load_dotenv()
    chatbot = ChatBot(api_key=os.getenv("OPENAI_API_KEY"))
    chatbot.start_chat()
!

OpenAI APIの利用にはPython用クライアントモジュールopenaiを利用するのが一般的ですが、HTTP通信に関わる例外処理について解説するため、あえてhttpxを使用しています。

実行結果の例
メッセージを入力してください (`/end` で終了): 今東京にいるんだけど、めちゃ暑いっす
LLM: 東京の夏は確かに暑いですね！湿度も高いので、体感温度がさらに上がります。水分をこまめに摂ること、日陰や冷房の効いた場所で休むことが大切です。もし外出するなら、帽子や日傘を使って日差しを避けると少し楽になりますよ。また、冷たい飲み物やアイスクリームを楽しむのもいいですね。本当に暑い日は、無理せず涼しい場所で過ごすことをお勧めします。体調には気をつけてくださいね！
メッセージを入力してください (`/end` で終了): 今僕はどこにいるんだっけ？
LLM: あなたは「今東京にいる」とおっしゃっていましたね。東京のどのエリアにいるかまではわかりませんが、東京の暑さには気をつけて過ごしてください。もし具体的な場所やおすすめのスポットについて知りたいことがあれば、教えてください！
メッセージを入力してください (`/end` で終了): /end

 関数型プログラミングの観点で見直す先ほどのオブジェクト指向による実装を、関数型プログラミングの観点から見直してみましょう。

 本記事における関数型プログラミングとは本記事では「なっとく！関数型プログラミング」から、下記の用語と定義を拝借します。
関数型プログラミング
関数型プログラミングとは イミュータブルな値を操作する純粋関数を利用するプログラミングである (P.70)
純粋関数
戻り値は1つだけ
引数のみに基づいて戻り値を計算する
既存の値を変更しない
(P.46)


また同書では、純粋関数の定義「引数にのみ基づいて戻り値を計算する」を

関数のシグネチャは嘘をついてはいけない
と言い直しています。

関数のシグネチャとは関数名、引数、戻り値を表す部分を指します。
def add_two_numbers(num1: int, num2: int) -> int: # <- シグネチャ
    return num1 + num2 # <- 本体（関数の実体）
「関数のシグネチャが嘘をつかない」とは、関数がシグネチャに書いてある引数以外を利用した計算はせず、またシグネチャに示されていない副作用や例外を発生させないことを指しています。

言い換えると、その関数が何をするのかはシグネチャに全て書いてあるべきであり、シグネチャから読み取れない処理を関数は行ってはいけない、ということを意味しています。
例えば上記関数add_two_numbersのシグネチャからは、「引数にある2つの整数（int型）を足した結果（1つのint型）を返す」ことが明確に読み取れます。したがって、この関数のシグネチャは嘘をついていないと言えます。
本記事で関数型プログラミングを実践するにあたっては、この指摘を関数型プログラミング定義に加えます。
関心の分離
あともう1つ、関数型プログラミングを実践する上で重要な考え方である「関心の分離」も本記事における関数型プログラミングの定義に加えます。



関心の分離とは

それぞれのコードがそれぞれの責任を負い、そのことにのみ関心を持つ (P.29)
ようにコードを作成することを言います。

1つの関数が複数の責務や関心領域を持たないように、責務や関心領域の単位で関数を作成し、それらを組み合わせて機能全体を成り立たせることを目指します。

 ツッコミの観点以上を踏まえて、下記の観点からコードを見直します。
関数のシグネチャが嘘をついていないか
イミュータブルな値のみを操作しているか
関心が分離されているか

 def _create_and_send_message(self, user_message: str) -> str:

 1. 関数のシグネチャが嘘をついていないか ---> Noシグネチャから読み取れることは下記のとおりです。
関数名の先頭に_があることから、クラス内でのみ使うことを開発者は意図していた[1]
関数名から、この関数はメッセージを作成して送信する
引数と戻り値から、この関数は（インスタンス自身と）str型の「ユーザメッセージ」を受け取って、str型の値を返す
しかし、この関数が例外httpx.HTTPStatusErrorをスローする可能性があること、例外発生時にはコンソールへのメッセージ出力があること、そしてこの時、str型の値は返却されないことは、シグネチャからは読み取れません。

例外への処理がシグネチャに反映されていないため、嘘をついていると言えます。

 2. イミュータブルな値のみを操作しているか ---> Noこの関数では、self.messagesというミュータブルなリストを直接操作しています。したがって、イミュータブルな値のみを操作していません。

 3. 関心が分離されているか？ ---> Noこの関数は、メッセージ履歴の更新、APIリクエストの作成と送信、エラーハンドリングなど、複数の責務を持っており、関心が分離されていないと言えます。

 def start_chat(self) -> None:

 1. 関数のシグネチャが嘘をついていないか？ ---> Noユーザーが/endを入力するとループを抜けますが、その動作はシグネチャからは読み取れません。また、ユーザー入力や出力といった副作用もシグネチャに反映されていません。

 2. イミュータブルな値のみを操作しているか？ ---> Yesこの関数自体では、特にミュータブルな値の操作は行っていません。

 3. 関心が分離されているか？ ---> Yesこの関数はユーザーとの対話全体を管理しており、関心が適切に分離されていると言えます。

 関数型プログラミングでの実装
 書き直しの方針
メッセージ履歴のイミュータブル化
メッセージ履歴をイミュータブルなデータ構造（タプル）に置き換え、メッセージを追加するたびに新しい履歴を生成する


例外処理の明示
関数のシグネチャに例外が発生する可能性を反映させるため、戻り値の型をOptionalにし、例外発生時にはNoneを返す


関数の分割
各関数が単一の責任を持つように、処理を細かく分割する


 実践
 1. メッセージ履歴のイミュータブル化メッセージ履歴の扱いでは、一度宣言したコレクションを使い回すのではなく、メッセージを追加するたびに、新しいコレクションを生成するように書き換えます。

また、メッセージ履歴をイミュータブルなデータ構造にするため、タプルを採用しましょう。

!5.3. タプルとシーケンスには

タプルはリストと似ていますが、たいてい異なる場面と異なる目的で利用されます。タプルは 不変 で、複数の型の要素からなることもあり、要素はアンパック(この節の後半に出てきます)操作やインデックス (あるいは namedtuples の場合は属性)でアクセスすることが多いです。一方、リストは 可変 で、要素はたいてい同じ型のオブジェクトであり、たいていイテレートによってアクセスします。（太字は筆者による）
とあり、「同じ型の要素」に対して「イテレートを行う」場合にはリストを使うのが一般的であるように読み取れます。ただしここでは、本記事の主旨に合わせるため、タプルを採用します。

# メッセージ履歴を保持するためのユーザ定義型
Message = tuple[str, str]  # (role, content)

def create_message(role: str, content: str) -> Message:
    """
    新しいメッセージを作成する純粋関数。

    Args:
        role (str): メッセージの役割（例: 'user', 'assistant'）
        content (str): メッセージの内容

    Returns:
        Message: 新しいメッセージ
    """
    return (role, content)

def update_messages(messages: tuple[Message, ...], new_message: Message) -> tuple[Message, ...]:
    return messages + (new_message,)

 2. 例外処理の明示Pythonは他の言語（JavaやC#）とは違い、関数のシグネチャで例外をスローする可能性があることを明示できません。
Javaの例
// Javaで検査例外を投げる場合、`throws`を付記することで、検査例外をスローすることを明示できる
public static void performOperation(int value) throws CustomCheckedException {
    if (value < 0) {
        throw new CustomCheckedException("負の値は許可されていません: " + value);
    }
    System.out.println("処理成功: " + value);
}
!

もちろん「Javaであれば、検査例外を使えばよい」と言いたいわけではありません。

そこで、PythontypingモジュールのOptional型を利用します。Optionalは指定した型とNoneとの直和を表現する型です。例えばOptional[str]とすると、この型は「strまたはNoneである」ことを表現しています。



これを使うと、例外が発生する可能性がある処理において

正常終了であれば、指定した型を返す
例外発生であれば、Noneを返す
という戻り値の「出し分け」が可能となり、それを関数のシグネチャで表現できるのです。
例えば戻り値がOptional[str]である関数の利用者は、「この関数は正常終了時にはstrを、異常終了時にはNoneが返ってくる」と読み取れます。

くわえてこの際、関数をtry-exceptで囲んで、どんな例外がスローされるのかを事前に確認する必要はなく、ただ戻り値がNoneか否かを確認すればよいため、利用者にとっても使いやすい関数となります。
ということで、OpenAI APIとHTTP通信をする関数を作成しましょう。
def send_message(api_key: str, messages: list[Message]) -> Optional[str]:
    """
    OpenAI APIにメッセージを送信する純粋関数。

    Args:
        api_key (str): OpenAI APIキー。
        messages (list[Message]): メッセージ履歴のリスト。

    Returns:
        Optional[str]: LLMの返答。通信に失敗した場合はNone。
    """
    url = "https://api.openai.com/v1/chat/completions"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    payload = {
        "model": "gpt-4o",
        "messages": [{"role": role, "content": content} for role, content in messages],
        "temperature": 0.7,
    }

    try:
        response = httpx.post(url, headers=headers, json=payload)
        data = response.raise_for_status().json()
        return data["choices"][0]["message"]["content"]
    except httpx.HTTPStatusError:
        # 例外をスローせず、`None`を返す
        return None

 3. 関心の分離最後に、残りのチャット機能を構成する関数を処理単位に分割しましょう。

関数の分割にあたっては、なるべく純粋関数を作るようにしていきます。
print()は戻り値がなく、純粋関数の定義「引数のみに基づいて戻り値を計算する」に反してしまいます。そのため、この処理は外に切り出すことで、純粋でない部分、すなわち「副作用がある部分」はそれだけで1つの関数として独立させる方針とします。
def get_user_input() -> str:
    """
    ユーザーからの入力を取得します。

    Returns:
        str: ユーザーが入力したメッセージ。
    """
    return input("メッセージを入力してください (`/end` で終了): ")


def display_message(message: str) -> None:
    """
    アシスタントからのメッセージを表示します。

    Args:
        message (str): 表示するメッセージ。
    """
    print(f"LLM: {message}")


def chat_loop(api_key: str, messages: tuple[Message, ...]) -> None:
    """
    ユーザーとのチャットを継続的に処理します。

    Args:
        api_key (str): OpenAI APIキー。
        messages (tuple[Message, ...]): 現在のメッセージ履歴。
    """
    user_message = get_user_input()

    if user_message == "/end":
        # ここの`print()`は処理終了時に呼ばれることから、関数として切り出さないことにします。
        print("チャットを終了します。")
        return

    new_messages_with_user = update_messages(messages, create_message("user", user_message))
    assistant_message = send_message(api_key, new_messages_with_user)

    if assistant_message is not None:
        new_messages_with_assistant = update_messages(
            new_messages_with_user, create_message("assistant", assistant_message)
        )
        display_message(assistant_message)
        chat_loop(api_key, new_messages_with_assistant)
    else:
        print("アシスタントからの応答がありませんでした。")
        chat_loop(api_key, new_messages_with_user)


def chat(api_key: str) -> None:
    """
    チャットセッションを開始します。

    Args:
        api_key (str): OpenAI APIキー。
    """
    print("チャットを開始します。`/end`と入力すると終了します。")
    initial_messages: tuple[Message, ...] = ()
    chat_loop(api_key, initial_messages)
コード全体
fp_chat.py
from typing import Optional
import httpx

# メッセージ履歴を保持するためのデータ構造をイミュータブルに定義
Message = tuple[str, str]  # (role, content)


def create_message(role: str, content: str) -> Message:
    """
    新しいメッセージを作成します。

    Args:
        role (str): メッセージの役割（例: 'user', 'assistant')
        content (str): メッセージの内容

    Returns:
        Message: 新しいメッセージ
    """
    return (role, content)


def update_messages(messages: tuple[Message, ...], new_message: Message) -> Tuple[Message, ...]:
    """
    メッセージ履歴に新しいメッセージを追加します。

    Args:
        messages (tuple[Message, ...]): 現在のメッセージ履歴。
        new_message (Message): 追加する新しいメッセージ。

    Returns:
        tuple[Message, ...]: 新しいメッセージを追加したメッセージ履歴。
    """
    return messages + (new_message,)


def send_message(api_key: str, messages: list[Message]) -> Optional[str]:
    """
    OpenAI APIにメッセージを送信します。

    Args:
        api_key (str): OpenAI APIキー。
        messages (list[Message]): メッセージ履歴のリスト。

    Returns:
        Optional[str]: LLMの返答。通信に失敗した場合はNone。
    """
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": "gpt-4o",
        "messages": [{"role": role, "content": content} for role, content in messages],
        "temperature": 0.7,
    }

    try:
        response = httpx.post(url, headers=headers, json=payload)
        data = response.raise_for_status().json()
        return data["choices"][0]["message"]["content"]
    except httpx.HTTPStatusError:
        # 例外をスローせず、`None`を返す
        return None


def get_user_input() -> str:
    """
    ユーザーからの入力を取得します。

    Returns:
        str: ユーザーが入力したメッセージ。
    """
    return input("メッセージを入力してください (`/end` で終了): ")


def display_message(message: str) -> None:
    """
    アシスタントからのメッセージを表示します。

    Args:
        message (str): 表示するメッセージ。
    """
    print(f"LLM: {message}")


def chat_loop(api_key: str, messages: tuple[Message, ...]) -> None:
    """
    ユーザーとのチャットを継続的に処理します。

    Args:
        api_key (str): OpenAI APIキー。
        messages (tuple[Message, ...]): 現在のメッセージ履歴。
    """
    user_message = get_user_input()

    if user_message == "/end":
        print("チャットを終了します。")
        return

    new_messages_with_user = update_messages(messages, create_message("user", user_message))
    assistant_message = send_message(api_key, new_messages_with_user)

    if assistant_message is not None:
        new_messages_with_assistant = update_messages(
            new_messages_with_user, create_message("assistant", assistant_message)
        )
        display_message(assistant_message)
        chat_loop(api_key, new_messages_with_assistant)
    else:
        print("アシスタントからの応答がありませんでした。")
        chat_loop(api_key, new_messages_with_user)


def chat(api_key: str) -> None:
    """
    チャットセッションを開始します。

    Args:
        api_key (str): OpenAI APIキー。
    """
    print("チャットを開始します。`/end`と入力すると終了します。")
    initial_messages: tuple[Message, ...] = ()
    chat_loop(api_key, initial_messages)


if __name__ == "__main__":
    from dotenv import load_dotenv
    import os

    # 環境変数からAPIキーをロード
    load_dotenv()
    api_key = os.getenv("OPENAI_API_KEY")

    # チャットセッションを開始
    chat(api_key)

 関数型プログラミングのまとめ関数型プログラミングの観点からコードを見直し、以下の改善をしました。
イミュータブルなデータの活用: メッセージ履歴をタプルで管理し、メッセージ追加時に新しいタプルを生成するように変更しました。
純粋関数の作成: 各関数が副作用を持たず、引数に基づいて戻り値を計算する純粋関数になるように設計しました。
関心の分離: 入力取得、メッセージ送信、表示などの機能をそれぞれ独立した関数に分割し、単一の責任を持つようにしました。
例外処理の明示: 関数のシグネチャに例外の可能性を反映させ、戻り値の型をOptionalに設定しました。
これらの改善により、コードの可読性と再利用性が向上し、関数型プログラミングのメリットを実感できる実装となりました。

 まとめ本記事では、Pythonを用いて関数型プログラミングの基本概念を整理しました。

具体的には、イミュータブルなデータ構造の採用、純粋関数の作成、関心の分離、そして例外処理の明示を通じて、コードの安全性と再利用性を高めました。

関数型プログラミングを適用することで、コードがより予測可能でテストしやすくなり、バグ発生の抑制が期待できます。

 本記事の次ステップ関数型プログラミングに限らず、関数のシグネチャに「その関数がどのような型（ドメイン）の引数を受け取り、どのような型を返すのか」を明確に示すことは重要です。今回の実装では、Pythonの組み込み型とtypingモジュールを使用しましたが、ユーザー定義型を活用することで、関数のシグネチャをより豊かに表現できます。

次のステップとして、Pythonの型ヒントやデータクラス、TypedDictなどを活用して、型安全性を高めた実装に取り組んでみましょう。

 参考文献https://www.shoeisha.co.jp/book/detail/9784798179803
https://peps.python.org/pep-0008/#descriptive-naming-styles

脚注
PEP 8 – Style Guide for Python Code / Descriptive: Naming Styles ↩︎
GitHubで編集を提案
Accenture Japan (有志)
アクセンチュア株式会社所属の社員有志による運営です。アクセンチュアの社員による様々な発信をまとめています。なお、投稿内容は社員個人の見解であり、所属する組織を代表するものではありません。
関数型プログラミングをPythonで実践しよう - #1 まずは書いてみる -

はじめに

対象読者

対象外の読者

本記事を読むことで得られること

実装内容

オブジェクト指向での実装

関数型プログラミングの観点で見直す

本記事における関数型プログラミングとは

ツッコミの観点

`def _create_and_send_message(self, user_message: str) -> str:`

1. 関数のシグネチャが嘘をついていないか ---> No

2. イミュータブルな値のみを操作しているか ---> No

3. 関心が分離されているか？ ---> No

`def start_chat(self) -> None:`

1. 関数のシグネチャが嘘をついていないか？ ---> No

2. イミュータブルな値のみを操作しているか？ ---> Yes

3. 関心が分離されているか？ ---> Yes

関数型プログラミングでの実装

書き直しの方針

実践

1. メッセージ履歴のイミュータブル化

2. 例外処理の明示

3. 関心の分離

関数型プログラミングのまとめ

まとめ

本記事の次ステップ

参考文献

Discussion
