関数型プログラミングのエッセンスをTypeScript・Pythonの実務コードで身につける
関数型プログラミング(FP)は「HaskellやScalaの世界」と思われがちですが、2025年以降、TypeScriptやPythonの実務コードベースに自然と浸透しています。Reactのコンポーネントモデル、TypeScriptのEffect統合、Pythonのitertools/functoolsなど、私たちが日常的に使っている道具の中にFPのエッセンスは既に存在します。
この記事では、MLエンジニアが日常的に触れるTypeScriptとPythonを題材に、FPの核心概念を「使える形」で整理します。理論的な深堀りではなく、「このパターンをどう実務に適用するか」にフォーカスします。
この記事でわかること
- 純粋関数・イミュータビリティ・関数合成の3原則を実務コードに適用する方法
- TypeScriptでの
pipe/Option/Eitherパターンの具体的な実装 - PythonのMLパイプラインに関数型パターンを導入してテスタビリティを向上させる手法
- Railway Oriented Programming(ROP)でエラーハンドリングを型安全に設計するアプローチ
- FPを段階的に導入する戦略と、導入すべきでないケースの判断基準
対象読者
- 想定読者: TypeScriptまたはPythonを日常的に使うソフトウェアエンジニア・MLエンジニア
-
必要な前提知識:
- TypeScript 5.x / Python 3.11+ の基本文法
- 関数、クラス、型の基礎概念
-
map、filter、reduceの基本的な使い方
結論・成果
関数型プログラミングの原則を部分的に導入することで、以下のような効果が報告されています。
- 並行処理性能: FPの不変データ構造を活用したシステムで、OOP版と比較して約3倍のスループットと20%の低レイテンシが報告されている(OOP vs. functional programming: My experience using both in production)
- テスト容易性: 純粋関数はモック不要でテスト可能。テストコードの記述量が大幅に減少
- バグ密度の低減: イミュータブルなデータ構造により、状態変更に起因するバグが構造的に排除される
ただし、完全なFP化は現実的でないケースも多く、段階的な部分導入が実務での推奨アプローチです。
FPの3原則を実務コードで理解する
関数型プログラミングの核心は3つの原則に集約されます。これらはHaskellの教科書のためではなく、日々のTypeScript・Pythonコードの品質を向上させるための実践的な道具です。
原則1: 純粋関数で副作用を分離する
純粋関数(Pure Function)とは、同じ入力に対して常に同じ出力を返し、外部の状態を変更しない関数です。Pythonの__init__に相当するTypeScriptのconstructorと違い、純粋関数はオブジェクトの状態を持ちません。
// 不純な関数: 外部状態に依存
let discount = 0.1;
function calcPrice(price: number): number {
return price * (1 - discount); // discountが変わると結果が変わる
}
// 純粋関数: 入力のみに依存
function calcPricePure(price: number, discountRate: number): number {
return price * (1 - discountRate);
}
# 不純な関数: グローバル状態を変更
results = []
def process_item(item):
result = transform(item)
results.append(result) # 副作用: リストを変更
return result
# 純粋関数: 新しい値を返す
def process_item_pure(item):
return transform(item)
# 呼び出し側で集約
processed = [process_item_pure(item) for item in items]
なぜこの実装を選んだか:
- 純粋関数はモック不要でテストできる。入力と出力だけ検証すればよい
- 副作用がないため、並列実行しても安全(MLのデータ前処理パイプラインで特に有効)
注意点:
副作用を完全に排除することは非現実的です。DB書き込み、ログ出力、API呼び出しは避けられません。重要なのは「純粋な計算ロジック」と「副作用を伴う入出力」を明確に分離することです。
原則2: イミュータビリティでバグの温床を断つ
データを変更せず、新しいデータを作る。これがイミュータビリティ(不変性)の原則です。
// ミュータブル: オブジェクトを直接変更
interface User {
name: string;
scores: number[];
}
function addScore(user: User, score: number): void {
user.scores.push(score); // 元のオブジェクトを変更
}
// イミュータブル: 新しいオブジェクトを返す
function addScoreImmutable(user: Readonly<User>, score: number): User {
return {
...user,
scores: [...user.scores, score],
};
}
from dataclasses import dataclass, field
from typing import Tuple
# ミュータブル: デフォルトのdataclass
@dataclass
class ModelConfig:
learning_rate: float
layers: list[int]
# イミュータブル: frozen=True
@dataclass(frozen=True)
class ModelConfigImmutable:
learning_rate: float
layers: tuple[int, ...] # listではなくtupleで不変に
def with_learning_rate(self, lr: float) -> "ModelConfigImmutable":
"""新しい設定を返す(元のオブジェクトは変更しない)"""
return ModelConfigImmutable(
learning_rate=lr,
layers=self.layers,
)
# 使用例
config = ModelConfigImmutable(learning_rate=0.001, layers=(128, 64, 32))
new_config = config.with_learning_rate(0.0001)
# config.learning_rate は 0.001 のまま
MLエンジニアにとって、ハイパーパラメータの設定が意図せず変更されるバグは致命的です。frozen=Trueのdataclassを使えば、設定変更は明示的に新しいオブジェクトとして行われるため、実験の再現性が保証されます。
よくある間違い:
最初は「毎回コピーするのはメモリの無駄」と考えがちですが、現代のランタイムは構造共有(Structural Sharing)によりコピーコストを最小化しています。実測で問題になるケースは稀です。ただし、数百万行のデータフレーム操作では、Pandasのcopy()は実際にメモリを倍消費するため、この原則が適用外になります。
原則3: 関数合成で処理をパイプラインにする
小さな関数を組み合わせて複雑な処理を構築する。これが関数合成(Function Composition)です。Pythonのsklearn.pipeline.Pipelineはまさにこの概念の具現化です。
// fp-ts/Effectのpipeを使った関数合成
// npm install effect
import { pipe } from "effect/Function";
const processUser = (rawInput: string) =>
pipe(
rawInput,
parseInput, // string -> RawData
validateAge, // RawData -> ValidatedData
enrichWithDefaults, // ValidatedData -> EnrichedData
formatForDB // EnrichedData -> DBRecord
);
// 各関数は単一責任で、独立してテスト可能
function parseInput(raw: string): RawData {
return JSON.parse(raw);
}
function validateAge(data: RawData): ValidatedData {
if (data.age < 0 || data.age > 150) {
throw new Error("Invalid age");
}
return { ...data, validated: true };
}
from functools import reduce
from typing import Callable, TypeVar
T = TypeVar("T")
def pipe(value: T, *functions: Callable) -> T:
"""シンプルなpipe関数の実装"""
return reduce(lambda v, f: f(v), functions, value)
# MLデータ前処理パイプライン
def remove_nulls(df):
return df.dropna()
def normalize_features(df):
return (df - df.mean()) / df.std()
def add_feature_interactions(df):
df_new = df.copy()
for i, col1 in enumerate(df.columns):
for col2 in df.columns[i+1:]:
df_new[f"{col1}_x_{col2}"] = df[col1] * df[col2]
return df_new
# パイプラインとして合成
processed = pipe(
raw_dataframe,
remove_nulls,
normalize_features,
add_feature_interactions,
)
Railway Oriented Programmingでエラーハンドリングを設計する
ここまでの3原則を踏まえた上で、実務で最も効果を発揮するFPパターンがRailway Oriented Programming(ROP)です。Scott WlaschinがF# for fun and profitで提唱したこのパターンは、エラーハンドリングを「成功トラック」と「失敗トラック」の2線路モデルで扱います。
Either型: 成功と失敗を型で表現する
try/catchの問題点は、「この関数がどんなエラーを返しうるか」が型レベルで表現できないことです。Either<Error, Success>を使えば、エラーの可能性を型システムに組み込めます。
// Either型の簡易実装(実務ではEffect/neverthrowを使用)
type Either<E, A> = { tag: "left"; error: E } | { tag: "right"; value: A };
const left = <E>(error: E): Either<E, never> => ({ tag: "left", error });
const right = <A>(value: A): Either<never, A> => ({ tag: "right", value });
function map<E, A, B>(
either: Either<E, A>,
f: (a: A) => B
): Either<E, B> {
return either.tag === "right" ? right(f(either.value)) : either;
}
function flatMap<E, A, B>(
either: Either<E, A>,
f: (a: A) => Either<E, B>
): Either<E, B> {
return either.tag === "right" ? f(either.value) : either;
}
実務例: ユーザー登録のバリデーションパイプライン
// バリデーションエラーの型定義
type ValidationError =
| { type: "INVALID_EMAIL"; message: string }
| { type: "WEAK_PASSWORD"; message: string }
| { type: "DUPLICATE_USER"; message: string };
interface RegisterInput {
email: string;
password: string;
}
interface ValidatedInput {
email: string;
passwordHash: string;
}
// 各バリデーションは Either を返す
function validateEmail(
input: RegisterInput
): Either<ValidationError, RegisterInput> {
const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
return emailRegex.test(input.email)
? right(input)
: left({ type: "INVALID_EMAIL", message: "メール形式が不正です" });
}
function validatePassword(
input: RegisterInput
): Either<ValidationError, RegisterInput> {
return input.password.length >= 8
? right(input)
: left({
type: "WEAK_PASSWORD",
message: "パスワードは8文字以上必要です",
});
}
function hashPassword(
input: RegisterInput
): Either<ValidationError, ValidatedInput> {
// 実際にはbcryptなどを使用
return right({
email: input.email,
passwordHash: `hashed_${input.password}`,
});
}
// Railway: 成功トラックを進むか、失敗トラックに切り替わる
function registerUser(
input: RegisterInput
): Either<ValidationError, ValidatedInput> {
return pipe(
right(input) as Either<ValidationError, RegisterInput>,
(e) => flatMap(e, validateEmail),
(e) => flatMap(e, validatePassword),
(e) => flatMap(e, hashPassword)
);
}
PythonでのResult型パターン
Pythonでも同様のパターンを実装できます。returnsライブラリを使う方法もありますが、ここではシンプルに自作します。
from dataclasses import dataclass
from typing import Generic, TypeVar, Union, Callable
T = TypeVar("T")
E = TypeVar("E")
U = TypeVar("U")
@dataclass(frozen=True)
class Ok(Generic[T]):
value: T
@dataclass(frozen=True)
class Err(Generic[E]):
error: E
Result = Union[Ok[T], Err[E]]
def bind(result: Result, f: Callable) -> Result:
"""成功時のみfを適用、失敗時はそのまま返す"""
if isinstance(result, Ok):
return f(result.value)
return result
# MLモデル推論パイプラインでの使用例
def load_model(path: str) -> Result:
try:
# model = torch.load(path) のような処理
return Ok({"model": "loaded", "path": path})
except Exception as e:
return Err(f"モデル読み込み失敗: {e}")
def validate_input(model_data: dict) -> Result:
if "model" not in model_data:
return Err("モデルデータが不正です")
return Ok(model_data)
def run_inference(model_data: dict) -> Result:
# 推論実行
return Ok({"prediction": 0.95, "model": model_data["model"]})
# パイプライン実行
result = bind(
bind(
load_model("model.pt"),
validate_input
),
run_inference
)
match result:
case Ok(value):
print(f"推論成功: {value}")
case Err(error):
print(f"エラー: {error}")
なぜtry/catchではなくResult型を使うのか:
-
try/catchはどの例外が飛んでくるか型レベルでわからない - Result型なら呼び出し側がエラー処理を忘れられない(型チェッカーが警告する)
- テスト時にエラーケースを網羅的に検証しやすい
制約条件:
このアプローチは、既存のPythonエコシステム(特にPandas、scikit-learn等)と完全に整合しません。これらのライブラリは例外ベースのエラーハンドリングを前提としているため、境界部分でResult型と例外の変換が必要になります。
Effect: TypeScript FPの次世代標準ライブラリ
2025年、fp-tsの作成者Giulio CantiがEffect organizationに参加し、Effectがfp-ts v3として位置づけられました(Effect vs fp-ts)。これはTypeScriptの関数型プログラミングにおける大きな転換点です。
Effectの基本: 型安全な副作用管理
EffectはEffect<Success, Error, Requirements>の3つの型パラメータで、成功値・エラー型・必要な依存をすべて型レベルで表現します。
import { Effect, pipe } from "effect";
// Effect<User, HttpError | ParseError, HttpClient>
// 成功時: User、エラー: HttpError | ParseError、依存: HttpClient
const fetchUser = (id: string) =>
pipe(
Effect.tryPromise({
try: () => fetch(`/api/users/${id}`),
catch: () => new HttpError("fetch failed"),
}),
Effect.flatMap((response) =>
Effect.tryPromise({
try: () => response.json(),
catch: () => new ParseError("JSON parse failed"),
})
),
Effect.map((data) => data as User)
);
段階的導入: Effectを全面採用しない選択肢
Harbor社の事例(Why We Love FP but Don't Use Effect-TS)は、Effectの全面採用が現実的でないケースを示しています。
同社はEffect-TSの代わりに、以下のライブラリを組み合わせて「部分的FP」を実現しています。
| ライブラリ | 用途 | FPの原則 |
|---|---|---|
| ts-pattern | パターンマッチング | 代数的データ型の分解 |
| Zod | スキーマバリデーション | 型安全なデータ変換 |
| Jotai | 状態管理 | アトミックな状態合成 |
トレードオフ:
- Effectの全面採用は型安全性を最大化するが、チームの学習コストが高い
- Node.jsエコシステムは「Promiseがrejectする」「関数がthrowする」前提で設計されているため、Effectとの橋渡しが必要になる
- データベーストランザクションのロールバックは例外に依存しており、Result型でラップするとロールバックが発火しない問題がある
// Harbor社のアプローチ: ts-patternでパターンマッチング
import { match, P } from "ts-pattern";
type ApiResponse =
| { status: "success"; data: User }
| { status: "not_found" }
| { status: "error"; message: string };
const handleResponse = (response: ApiResponse) =>
match(response)
.with({ status: "success" }, ({ data }) => renderUser(data))
.with({ status: "not_found" }, () => renderNotFound())
.with({ status: "error", message: P.select() }, (msg) =>
renderError(msg)
)
.exhaustive(); // すべてのケースを網羅していないとコンパイルエラー
PythonのMLパイプラインにFPを適用する
MLエンジニアにとって、関数型パターンが最も効果を発揮するのはデータ前処理・特徴量エンジニアリングのパイプラインです。
functoolsとitertoolsの実戦的な使い方
Python標準ライブラリのfunctoolsとitertoolsは、関数型プログラミングの基本ツールです(Python公式 関数型プログラミング HOWTO)。
from functools import partial, reduce
from itertools import chain, starmap
from typing import Callable, Iterator
# partial: 引数の部分適用
# MLの前処理関数を設定値で部分適用する例
def clip_feature(value: float, min_val: float, max_val: float) -> float:
return max(min_val, min(max_val, value))
# 特定の範囲にクリップする関数を作成
clip_to_unit = partial(clip_feature, min_val=0.0, max_val=1.0)
clip_to_score = partial(clip_feature, min_val=0.0, max_val=100.0)
# リスト内包表記より map + partial が意図を明確にするケース
raw_scores = [0.5, 1.2, -0.3, 0.8, 2.1]
clipped = list(map(clip_to_unit, raw_scores))
# [0.5, 1.0, 0.0, 0.8, 1.0]
# reduce: 複数の変換を合成
from functools import reduce
from typing import Callable
import pandas as pd
# 変換関数のリスト
transforms: list[Callable[[pd.DataFrame], pd.DataFrame]] = [
lambda df: df.dropna(),
lambda df: df[df["age"] > 0],
lambda df: df.assign(age_bucket=pd.cut(df["age"], bins=5)),
]
# reduceで順番に適用(sklearnのPipelineと同じ発想)
def apply_transforms(
df: pd.DataFrame,
transforms: list[Callable[[pd.DataFrame], pd.DataFrame]]
) -> pd.DataFrame:
return reduce(lambda d, f: f(d), transforms, df)
# 使用
processed_df = apply_transforms(raw_df, transforms)
ジェネレータでメモリ効率的なデータ処理
大規模データの処理では、ジェネレータ(遅延評価)がFPの強力な武器になります。
from itertools import islice
from typing import Iterator, TypeVar
T = TypeVar("T")
def batched(iterable: Iterator[T], batch_size: int) -> Iterator[list[T]]:
"""イテレータをバッチに分割(Python 3.12+ではitertools.batchedを使用)"""
it = iter(iterable)
while True:
batch = list(islice(it, batch_size))
if not batch:
break
yield batch
def process_large_dataset(file_path: str, batch_size: int = 1000):
"""大規模CSVをメモリ効率的に処理"""
# ジェネレータチェーン: 各ステップが遅延評価される
lines = (line.strip() for line in open(file_path)) # 遅延読み込み
records = (parse_csv_line(line) for line in lines) # 遅延パース
valid = (r for r in records if r is not None) # 遅延フィルタ
# バッチ単位で処理(メモリは1バッチ分のみ使用)
for batch in batched(valid, batch_size):
yield process_batch(batch)
ハマりポイント:
ジェネレータは一度しか消費できません。
list()でリスト化するか、itertools.tee()で複製しないと、2回目のイテレーションで空の結果になります。MLの訓練ループで「1エポック目だけ動いて2エポック目でデータが空」というバグはこれが原因であることが多いです。
よくある問題と解決方法
| 問題 | 原因 | 解決方法 |
|---|---|---|
| 「FP化したらチームの理解度が下がった」 | Either/Optionの概念が馴染みない | まず純粋関数の分離から始める。モナドは後回し |
| 「既存ライブラリがFPに対応していない」 | Node.js/Pythonは例外ベースが前提 | 境界部分にアダプタ層を設け、内部のみFP化 |
| 「イミュータブルでパフォーマンスが悪化」 | 大規模データのコピーコスト | Immer(TS)やPersistent Data Structureを使用 |
| 「型が複雑になりすぎる」 | Effect等の型パラメータが多い | 型エイリアスで抽象化、段階的導入 |
| 「ジェネレータが2回目のループで空になる」 | ジェネレータの一回消費特性 |
list()で実体化、またはitertools.tee()で複製 |
FPを導入すべきケースと避けるべきケース
すべてのコードを関数型にする必要はありません。以下の判断基準を参考にしてください。
| 観点 | FPが適するケース | OOP/手続き型が適するケース |
|---|---|---|
| データ変換 | ETL、前処理パイプライン、バリデーション | 複雑なGUIの状態管理 |
| 並行処理 | 不変データの並列処理、MapReduce | リソースの排他制御が必要な場合 |
| エラーハンドリング | バリデーションチェーン、APIレスポンス処理 | DBトランザクションのロールバック |
| テスト | ビジネスロジックの単体テスト | E2Eテスト、統合テスト |
| チーム | FP経験者がいる、教育体制がある | 短納期、FP経験者が不在 |
まとめと次のステップ
まとめ:
- FPの3原則(純粋関数・イミュータビリティ・関数合成)は、TypeScript・Pythonの実務コードに段階的に適用できる
- Railway Oriented ProgrammingとEither/Result型は、エラーハンドリングの型安全性を大幅に向上させる
- Effect(TypeScript)は2025年にfp-tsと統合され、次世代の標準ライブラリとして位置づけられたが、全面採用には学習コストとエコシステムの互換性課題がある
- Pythonでは
functools/itertools/ジェネレータが実戦的なFPツールとして機能する - 完全なFP化ではなく、ts-pattern + Zod等の部分的導入が現実的なアプローチ
次にやるべきこと:
- まず既存コードから「純粋関数に切り出せるロジック」を1つ特定し、リファクタリングする
- TypeScript: Effect公式ドキュメントでpipe/Either/Optionのチュートリアルを試す
- Python:
functools.partialを使って、設定値のハードコードを部分適用に置き換える
参考
- Railway Oriented Programming | F# for fun and profit
- Effect vs fp-ts | Effect Documentation
- Why We Love Functional Programming but Don't Use Effect-TS | Harbor
- 関数型プログラミング HOWTO | Python 3 公式ドキュメント
- TypeScript x 関数型プログラミング 2025年最新実装ガイド | ラーゲイト
- OOP vs. functional programming: My experience using both in production | Medium
- Functional Programming in 2025: The Comeback of Pure Functions | Dev Tech Insights
注意: この記事はAI(Claude Code)により自動生成されました。内容の正確性については複数の情報源で検証していますが、実際の利用時は公式ドキュメントもご確認ください。