Python函数式编程实践总结
2025/7/302671 字
最近实习，要写很多数据处理的代码。正好最近对函数式编程（fp）很感兴趣，找了个课程学了点皮毛，结合自己的实践，把目前总结的 Python 函数式编程方式经验总结一下。
函数式编程虽然比较"学院派"，看着有些多此一举，不过我认为真的是有其独特优势和魅力的：
- 步骤清晰，重用性好：比起面向过程各个步骤依赖的参数和执行代码混在一起，函数式编程的各个处理步骤各自的参数和执行代码分界清晰，使得可以轻易地修改，调换，删除其中某个步骤，提高代码可修改性。
- 纯函数的"无副作用"性质：使得你可以并发启动任意多个纯函数的流水线而不用担心传统并发编程需要考虑的竞态问题，心智负担低。
不过好多人说 fp 难以理解，脱裤子放屁。我认为是他们是在用面向过程或者是面向对象的方式在看待 fp 的代码。一旦不符合他们习惯的认知，就认为这种方式是难以理解的、怪异的。
但是我认为，从 lambda 演算法 以及 范畴论的角度看待函数式编程方法，fp 是十分自然且一致的。总之是习惯和视角的问题。
函数式编程基础工具在python中的实现
递归代替循环，filter 和 map 的实现
不用循环只用递归实现 filter 和 map 的方法，第一次看到这个，非常震撼。
有点说动我支持像去掉 goto 一样去掉 循环语法的观点了。
递归 filter 函数实现
pythonfrom typing import Callable, TypeVar, Iterable
T = TypeVar('T')
def filter(sequence: Iterable[T], condition: Callable[[T], bool]) -> Iterable[T]:
"""使用递归方式实现的filter函数，支持任意可迭代对象"""
iterator = iter(sequence)
try:
current = next(iterator)
if condition(current):
yield current
# 继续处理剩余元素
yield from filter(iterator, condition)
except StopIteration:
# 迭代器耗尽，停止递归
return
# test
arr = [1, 2, 3, 4, 5, 6]
print(list(filter(arr, lambda x: x % 2 == 1)))
# 测试其他可迭代对象
print(list(filter(range(10), lambda x: x % 3 == 0)))
print(list(filter("hello world", lambda x: x in "aeiou")))
输出结果：
plaintext[1, 3, 5]
[0, 3, 6, 9]
['e', 'o', 'o']
递归 map 函数实现
pythonfrom typing import Callable, TypeVar, Iterable
T = TypeVar('T')
U = TypeVar('U')
def map(sequence: Iterable[T], function: Callable[[T], U]) -> Iterable[U]:
"""使用递归方式实现的 map 函数，支持任意可迭代对象"""
iterator = iter(sequence)
try:
current = next(iterator)
yield function(current)
# 继续处理剩余元素
yield from map(iterator, function)
except StopIteration:
# 迭代器耗尽，停止递归
return
# test
arr = [1, 2, 3, 4, 5, 6]
print(list(map(arr, lambda x: x**2)))
输出结果：
plaintext[1, 4, 9, 16, 25, 36]
reduce 工具的理解和使用
把 reduce 操作当成流水线操作
输入一个 初始值 作为流水线的输入
然后一个序列，其中每个元素分别为这个流水线上各个环节的工人可以拿到的数据
还要输入一个函数，是各个工人如何使用的数据的方法
reduce 实现 map
写的比较早，感觉不如上面的高级和通用，不过思路挺好，可以学习一下：
按照上面的 reduce 的理解，map 函数就可以理解为，输入一个空盒子
每个工人都拿到一个序列中的数据，然后使用函数处理这个数据，再把数据放进盒子里，最后把这个盒子传给下一个工人
pythonfrom functools import reduce
def map(func, iterable):
return reduce(lambda x, y: x + [func(y)], iterable, [])
# test
arr = range(1, 6)
print(map(lambda x: x**2, arr))
输出结果：
plaintext[1, 4, 9, 16, 25]
pipe 管道实现
pipe 管道的功能被定义为接受一系列函数，返回一个函数，返回的函数是输入的函数的链式组合（即把上一个函数的输出作为下一个函数的输入）
也按照上面的 reduce 工具的理解：
初始值是最初输入到流水线的数据
每个工人拿到的是一个函数
所做的工作是，接受上个工人传递下来的处理完成的数据，使用手上的函数处理这个数据，将处理完成的结果传递给下一个工人
pythonfrom functools import reduce
from typing import Iterable, Callable
def pipe(funcs: Iterable[Callable]) -> Callable:
def process(input_val):
return reduce(lambda pre, cur: cur(pre), funcs, input_val)
return process
# test
w = 8.8
b = 1.5
test_pipe = pipe([
lambda x: x * w,
lambda x: x + b,
])
input_val = 10
print(test_pipe(input_val))
输出结果：
plaintext89.5
柯里化 实现
python# 🌟 递归实现通用柯里化函数
# 该函数可以将任意多参数函数柯里化，支持递归调用
from typing import Callable
from functools import wraps
def curry(func: Callable) -> Callable:
@wraps(func)
def curried(*args, **kwargs):
# 如果传入的参数数量足够，则直接调用
if len(args) + len(kwargs) >= func.__code__.co_argcount:
return func(*args, **kwargs)
# 否则返回一个新的函数，继续收集参数
return lambda *a, **k: curried(*(args + a), **{**kwargs, **k})
return curried
def greet(name: str, age: int) -> str:
return f"Hello, {name}! You are {age} years old."
curried_greet = curry(greet)
def test_curry(curried_greet):
print(curried_greet("Alice", 25)) # Hello, Alice! You are 25 years old.
print(curried_greet("Charlie")("35")) # Hello, Charlie! You are 35 years old.
print(curried_greet(age=45)("David")) # Hello, David! You are 45 years old.
print(curried_greet(name="Corn")(age=21)) # Hello, Corn! You are 21 years old.
test_curry(curried_greet)
输出结果：
plaintextHello, Alice! You are 25 years old.
Hello, Charlie! You are 35 years old.
Hello, David! You are 45 years old.
Hello, Corn! You are 21 years old.
现在我发现实际上 python 自带了一个 partial 函数，可以用来实现部分参数的函数调用，不如自己实现的好用，因为只能分成两个函数，还不能按照顺序省略某些参数，不过也够了
不过优点是可以把可调用对象，不仅仅是函数给柯里化
pythonfrom functools import partial
try:
test_curry(partial(greet))
except Exception as e:
print(e)
输出结果：
plaintextHello, Alice! You are 25 years old.
greet() missing 1 required positional argument: 'age'
pythonfrom dataclasses import dataclass
@dataclass
class Point:
x: float
y: float
try:
print(curry(Point)(1)(2)) # Point(x=1, y=2)
except Exception as e:
print(e)
print(partial(Point, 1)(2))
输出结果：
plaintexttype object 'Point' has no attribute '__code__'
Point(x=1, y=2)
多元函子
pythonfrom typing import Callable, Iterable
def functor(func: Callable) -> Callable:
def wrapper(*iterables: Iterable) -> Iterable:
for args in zip(*iterables):
yield func(*args)
return wrapper
# 三元函数：拼接字符串
concatenate = functor(lambda a, b, c: f"{a}{b}{c}")
str_list = ["a", "b", "c"]
num_list = [1, 2, 3]
char_list = ["X", "Y", "Z"]
print("三元函子:", list(concatenate(str_list, num_list, char_list)))
# 输出: ['a1X', 'b2Y', 'c3Z']
输出结果：
plaintext三元函子: ['a1X', 'b2Y', 'c3Z']
带固定参数的函子
实际上就是两者组合，先柯里化，再函子即可
pythonfrom typing import Iterable
exp = lambda a, b: a**b
print(exp(2, 3))
curried_exp = curry(exp)
print(curried_exp(4)(3))
arr = range(1, 6)
pow_2 = curried_exp(b = 2) # equals with "lambda x: x**2"
pow_2_for_list = functor(pow_2)
print(list(pow_2_for_list(arr)))
# both them all
print(list(
functor(
curry(lambda a, b: a**b)(b=4)
)(arr)
))
输出结果：
plaintext8
64
[1, 4, 9, 16, 25]
[1, 16, 81, 256, 625]
变体类型和模式匹配
变体类型
变体类型就是可以是多种类型的其中一种的复合类型，也叫做和类型（or）
与之相对的是结构体，包含了多种类型的复合类型，也叫做积类型（and）
例子——表达式定义
pythonfrom dataclasses import dataclass
# 定义表达式类型
# 变量
@dataclass
class Var:
index: int
# 常数
@dataclass
class Con:
value: float
# 幂函数
@dataclass
class Pow:
base: 'Expr'
exp: float
# 加法复合
@dataclass
class Add:
expr1: 'Expr'
expr2: 'Expr'
# 乘法复合
@dataclass
class Mul:
expr1: 'Expr'
expr2: 'Expr'
type Expr = Var | Con | Pow | Add | Mul # 使用 Union 类型的语法糖，定义指定的和类型
python# 构建表达式
test_expr = Add(
Pow(Var(0), 2), # v0^2
Add(
Mul(Con(2), Var(0)), # 2*v0
Con(1) # 1
)
)
print(test_expr)
输出结果：
plaintextAdd(expr1=Pow(base=Var(index=0), exp=2), expr2=Add(expr1=Mul(expr1=Con(value=2), expr2=Var(index=0)), expr2=Con(value=1)))
模式匹配
模式匹配就是一种解析输入不同值以及不同类型的情况 的写法，优势在于可以一遍匹配判断，一边把值取出来，直观清晰
例子——表达式求值
pythonfrom typing import List
def eval(expr: Expr, values: List[float]) -> float:
match expr:
case Var(index): return values[index]
case Con(value): return value
case Pow(base, exp): return eval(base, values) ** exp # 递归调用
case Add(expr1, expr2): return eval(expr1, values) + eval(expr2, values)
case Mul(expr1, expr2): return eval(expr1, values) * eval(expr2, values)
values = [2.]
result = eval(test_expr, values)
print(result)
输出结果：
plaintext9.0
实践技巧
直接串起来感觉很漂亮
pythonpipeline = [
functor(curry(lambda x, w: w * x)(w = 10)), # * 10
functor(curry(lambda x, b: x + b)(b = 2)), # + 2
functor(curry(lambda x, m: x ** m)(m = 2)) # ^ 2
]
arr = [1, 2, 3, 4, 5]
list(pipe(pipeline)(arr))
输出结果：
plaintext[144, 484, 1024, 1764, 2704]
但是这适用于返回值只有一个变量，且处理流程严格线性的情况，即某一层函数的返回值会且只会被下一层函数使用
当一个函数有多个返回值的时候，如果这多个返回值都会作为参数传递给下一个函数，说明这多个返回值应该包装为一个类。
不过如果是分别传递给不同的函数，这个时候就没有办法直接串起来，就需要把这两条分叉的管线打包成一个函数，如下面的例子。
python# 多元函数返回值分别传递给不同函数的例子
from typing import Tuple, List
# 一个函数返回多个值
def analyze_data(numbers: List[int]) -> Tuple[float, int]:
"""分析数据，返回平均值和最大值"""
if not numbers:
return 0.0, 0
avg = sum(numbers) / len(numbers)
max_val = max(numbers)
return avg, max_val
# 分别处理这两个返回值的函数
def scale_average(avg: float) -> float:
"""将平均值放大10倍"""
return avg * 10
def check_threshold(max_val: int) -> str:
"""检查最大值是否超过阈值"""
return "high" if max_val > 50 else "low"
# 传统的非函数式写法
data = [10, 20, 30, 40, 50]
avg, max_val = analyze_data(data)
scaled_avg = scale_average(avg)
threshold = check_threshold(max_val)
print(f"传统写法: scaled_avg={scaled_avg}, threshold={threshold}")
# 函数式写法：将分叉管线打包成一个函数
# 使用函数式写法
pipeline = [
analyze_data,
lambda input: [
avg := input[0],
max_val := input[1],
{
'scaled_average': scale_average(avg),
'threshold': check_threshold(max_val)
}
][-1],
print,
]
pipe(pipeline)(data)
输出结果：
plaintext传统写法: scaled_avg=300.0, threshold=low
{'scaled_average': 300.0, 'threshold': 'low'}
这样是用了，不过还是不太美观，目前我还没有探索出更好的方案。
可能完全使用函数式编程实现功能还是有些苛刻的，目前在恰当的地方使用 fp 发挥奇效即可。
💬 评论区
暂无评论，来发表第一条吧。