函数式编程：map/filter/reduce、lambda

一句话概述

函数式编程（Functional Programming）是一种编程范式，强调用函数组合而非命令式循环来处理数据。Python 不是纯函数式语言，但支持核心的函数式工具：map() 对每个元素应用函数并返回迭代器；filter() 按条件筛选元素；reduce() 将序列累积为单个值；lambda 创建匿名函数作为参数传入。这些工具配合生成器，可以构建简洁、可读、无副作用的数据处理管道。

💡 核心要点：①lambda 参数: 表达式 创建匿名函数——只能包含单个表达式，不能有语句 ②map(func, iterable) 对每个元素应用 func，返回惰性迭代器 ③filter(func, iterable) 保留 func 返回真值的元素 ④reduce(func, iterable) 将序列归约为单个值（需 from functools import reduce）⑤函数式编程风格适合数据转换管道——每一步变换清晰、无副作用

教学与演示

一、lambda 匿名函数——"用完即弃"的函数

是什么：lambda 是 Python 中创建匿名函数的关键字。语法：lambda 参数1, 参数2, ...: 表达式。它只能包含一个表达式（不能有 return 语句、循环、赋值等），表达式的计算结果自动作为返回值。lambda 通常作为参数传给 map()、filter()、sorted() 等高阶函数——用作"一次性"的小函数。

大白话 lambda 就像便利贴——你需要临时记个东西（做个小计算），撕一张便利贴写上就行，用完就可以扔掉（函数没有名字）。而 def 定义的函数就像一份正式文件——有标题、有存档、可以被多处引用。

为什么：在函数式编程中，经常需要传一个简单的函数作为参数——"每个元素平方"、"只保留大于 0 的"。如果每个小逻辑都写一个完整的 def 函数，代码会变得散乱。lambda 让逻辑和调用点保持在一起——代码自包含、可读性高。

怎么做：

import numpy as np

# ====== 1. lambda 的基本语法 ======

square = lambda x: x ** 2                       # 等价于 def square(x): return x ** 2
add = lambda a, b: a + b                        # 等价于 def add(a, b): return a + b

print("=== lambda 基本用法 ===")
print(f"square(5) = {square(5)}")               # 25
print(f"add(3, 7) = {add(3, 7)}")               # 10
print(f"匿名调用: {(lambda x, y: x * y)(6, 7)}")  # 42

# ====== 2. sorted() 的 key 参数——按自定义规则排序 ======

students = [
    {"name": "张三", "score": 92},
    {"name": "李四", "score": 78},
    {"name": "王五", "score": 85},
    {"name": "赵六", "score": 95},
]

ranked = sorted(students, key=lambda s: s["score"], reverse=True)
print(f"\n按分数排名:")
for i, s in enumerate(ranked):
    print(f"  {i+1}. {s['name']}: {s['score']}分")

# ====== 3. lambda 与 map/filter 配合 ======

nums = [1, 2, 3, 4, 5]
doubled = list(map(lambda x: x * 2, nums))
print(f"\nmap 加倍: {doubled}")                 # [2, 4, 6, 8, 10]

evens = list(filter(lambda x: x % 2 == 0, nums))
print(f"filter 偶数: {evens}")                  # [2, 4]

# 组合使用：先过滤后映射
result = list(map(lambda x: x ** 2, filter(lambda x: x > 2, nums)))
print(f"组合：>2 的平方: {result}")              # [9, 16, 25]

# ====== 4. AI 场景：数据预处理管道 ======

raw_data = [
    (np.array([1.5, 2.3, 0.8, 3.1]), 1),
    (np.array([0.5, 1.2, 1.9, 2.4]), 0),
    (np.array([2.1, 3.5, 1.2, 4.0]), 1),
    (np.array([0.9, 0.7, 2.8, 1.5]), 0),
]

cleaned = map(
    lambda item: ((np.clip(item[0], 0, 5) - 2.0) / 1.5, item[1]),
    raw_data
)

print(f"\n=== AI 数据预处理管道 ===")
for features, label in cleaned:
    print(f"  特征: {np.round(features, 3)}, 标签: {label}")

什么用：lambda 在 AI 数据处理中无处不在。PyTorch 的 Dataset 类的 transform 参数经常用 lambda 做简单变换。Pandas 的 apply() 常用 lambda 做列操作。sorted() 配合 lambda 按自定义规则排序数据。

二、map——对每个元素做同样的操作

是什么：map(function, iterable, ...) 是 Python 内置的高阶函数。它接收一个函数和一个（或多个）可迭代对象，对可迭代对象中的每个元素应用这个函数，返回一个迭代器（不是列表）。多个可迭代对象时，函数并行接收各迭代器的对应元素。

大白话 map 就像一个流水线上的机器人——传送带（可迭代对象）上的每个零件（元素）经过机器人时，机器人对它做同样的操作（函数）。零件流进来，加工好的零件流出去。

为什么：map 比手动 for 循环有三个优势：①意图明确——一眼就知道"这是一个逐元素变换" ②惰性求值——只在需要时才计算 ③可组合——可以无缝接入下一个 map/filter，形成数据处理管道。

怎么做：

import numpy as np

# ====== 1. map 基本用法 ======

def normalize(x):
    """线性归一化：将 [0, 255] 映射到 [0, 1]"""
    return x / 255.0

pixels = [0, 64, 128, 192, 255]
normalized = map(normalize, pixels)
print("=== map 基本用法 ===")
print(f"原始像素: {pixels}")
print(f"归一化后: {list(normalized)}")

# ====== 2. map 多迭代器——并行处理 ======

img1_pixels = [100, 150, 200, 180, 90]
img2_pixels = [200, 100, 150, 220, 110]

avg_pixels = list(map(lambda a, b: (a + b) // 2, img1_pixels, img2_pixels))
print(f"\n图像融合（逐像素平均）:")
print(f"  图A: {img1_pixels}")
print(f"  图B: {img2_pixels}")
print(f"  融合: {avg_pixels}")

# ====== 3. map 返回迭代器——惰性求值 ======

big_map = map(lambda x: x ** 2, range(10_000_000))
print(f"\n创建了 1 千万个元素的 map，但没有消耗内存！")
print(f"取前 5 个: {[next(big_map) for _ in range(5)]}")

# ====== 4. AI 场景：批量数据特征工程 ======

raw_features = np.array([
    [5.1, 3.5, 1.4, 0.2],
    [4.9, 3.0, 1.4, 0.2],
    [7.0, 3.2, 4.7, 1.4],
    [6.4, 3.2, 4.5, 1.5],
])

def feature_engineering(sample):
    """对单个样本做特征工程：标准化 + 多项式特征 + 交互特征"""
    mean_vec = np.array([5.8, 3.1, 3.0, 0.8])
    std_vec = np.array([1.0, 0.3, 1.8, 0.8])
    normalized = (sample - mean_vec) / (std_vec + 1e-8)
    squared = normalized ** 2
    interactions = np.array([normalized[0]*normalized[1],
                             normalized[1]*normalized[2],
                             normalized[2]*normalized[3]])
    return np.concatenate([normalized, squared, interactions])

print(f"\n=== 特征工程管道 ===")
processed = map(feature_engineering, raw_features)
for i, features in enumerate(processed):
    print(f"  样本{i+1}: {len(features)}个特征 | 前4个: {np.round(features[:4], 3)}")

# ====== 5. map 与列表推导式的选择 ======

# 有现成函数时：map 更简洁
words = ["hello", "world", "python"]
uppercase = list(map(str.upper, words))
print(f"\nmap(str.upper): {uppercase}")
print("小结：有现成函数用 map；简单表达式用列表推导式")

什么用：map 在 AI 数据预处理中是主力工具。批量图像归一化、文本向量化、特征标准化。PyTorch 的 DataLoader 不直接使用 map，但理念相同——对每个 batch 应用 collate_fn。TensorFlow 的 tf.data.Dataset.map() 直接使用了 map 的命名和思想。

三、filter——从数据流中"筛金子"

是什么：filter(function, iterable) 对可迭代对象的每个元素调用 function，保留那些**函数返回 True（或真值）**的元素。返回一个迭代器。如果 function 是 None，则保留所有真值元素。

大白话 filter 就像一个筛子——你把一堆石头（数据）倒进去，筛子的孔（过滤函数）决定了哪些石头能通过（True），哪些被拦住（False）。filter 不改变石头本身——只决定"过"还是"不过"。

为什么：数据清洗中最常见的操作就是"过滤"——去掉缺失值、去掉异常值、保留特定类别的样本。filter 让过滤逻辑声明式——"保留损失值 < 2.0 的样本"一目了然。配合 map 形成"过滤 → 变换"管道，是数据处理的标准模式。

怎么做：

import numpy as np

# ====== 1. filter 基本用法 ======

numbers = range(-5, 6)
positives = filter(lambda x: x > 0, numbers)
print("=== filter 基本用法 ===")
print(f"原始: {list(numbers)}")
print(f"正数: {list(positives)}")               # [1, 2, 3, 4, 5]

# ====== 2. AI 场景：数据清洗——过滤异常样本 ======

training_samples = [
    {"id": 1, "loss": 0.23, "label": "cat"},
    {"id": 2, "loss": 0.45, "label": "dog"},
    {"id": 3, "loss": 5.67, "label": "cat"},     # 异常！
    {"id": 4, "loss": 0.31, "label": "bird"},
    {"id": 5, "loss": 8.90, "label": "dog"},     # 异常！
    {"id": 6, "loss": 0.52, "label": "cat"},
    {"id": 7, "loss": 0.28, "label": "bird"},
]

clean_samples = list(filter(lambda s: s["loss"] < 2.0, training_samples))
print(f"\n=== 数据清洗 ===")
print(f"原始样本数: {len(training_samples)}, 清洗后: {len(clean_samples)}")
for s in clean_samples:
    print(f"  ID={s['id']}, loss={s['loss']}, label={s['label']}")

# 保留特定类别
cat_dog_samples = list(filter(
    lambda s: s["label"] in ("cat", "dog"), training_samples
))
print(f"\n猫狗样本数: {len(cat_dog_samples)}")

# ====== 3. filter + map 管道——数据流水线 ======

raw_vectors = [
    np.array([1.5, 2.3, np.nan]),                # 含 NaN
    np.array([0.5, 1.2, 3.4]),
    np.array([np.inf, 2.1, 0.8]),                # 含 inf
    np.array([2.1, 3.5, 1.2]),
]

print(f"\n=== 数据管道：过滤 + 变换 ===")
pipeline = map(
    lambda vec: (vec - np.mean(vec)) / (np.std(vec) + 1e-8),
    filter(
        lambda vec: not (np.any(np.isnan(vec)) or np.any(np.isinf(vec))),
        raw_vectors
    )
)
print("清洗+标准化后的结果:")
for vec in pipeline:
    print(f"  {np.round(vec, 3)}")

什么用：filter 在 AI 数据处理中是数据质量保证的第一道关口。过滤缺失值、过滤低置信度预测、过滤无效数据。在主动学习（Active Learning）中，用 filter 筛选模型不确定的样本进行人工标注。在数据清洗管道中，filter → map 是标准模式。

四、reduce——把序列"压缩"成一个值

是什么：reduce(function, iterable[, initializer]) 来自 functools 模块。它从左到右对序列的元素累积应用一个二元函数：先用前两个元素调用函数得到结果，再用结果和第三个元素调用函数，以此类推，最终将所有元素"归约"为一个值。

大白话 reduce 就像一个揉面团的厨师——他拿一团面（初始值或第一个元素），然后一个个加入后续的原料（元素），每加一个就揉一次（应用函数）。最后所有原料被揉成一团面（一个最终值）。

为什么：reduce 适用于"将序列聚合为单个值"的场景——求和、求积、自定义聚合。但注意——能用内置函数时优先用内置（sum 比 reduce 更清晰），reduce 适合自定义聚合逻辑和函数组合。

怎么做：

import numpy as np
from functools import reduce

# ====== 1. reduce 基本用法 ======

nums = [1, 2, 3, 4, 5]
total = reduce(lambda a, b: a + b, nums)
product = reduce(lambda a, b: a * b, nums)
print("=== reduce 基本用法 ===")
print(f"累加: {total}")                          # 15
print(f"累乘: {product}")                        # 120

# ====== 2. reduce 执行过程可视化 ======

print(f"\n=== reduce 执行过程 ===")
def show_reduce(acc, x):
    result = acc + x
    print(f"  累积值={acc}, 新元素={x} → {result}")
    return result

result = reduce(show_reduce, [10, 20, 30, 40])
print(f"  最终结果: {result}")

# ====== 3. AI 场景：计算加权平均 ======

data_points = [
    {"value": 0.85, "weight": 3},
    {"value": 0.72, "weight": 5},
    {"value": 0.91, "weight": 2},
    {"value": 0.68, "weight": 4},
]

def accumulate_weighted(acc, item):
    return (acc[0] + item["value"] * item["weight"],
            acc[1] + item["weight"])

weighted_sum, total_weight = reduce(accumulate_weighted, data_points, (0.0, 0))
weighted_avg = weighted_sum / total_weight
print(f"\n=== 加权平均 ===")
print(f"加权平均: {weighted_avg:.4f}")

# 场景2：函数组合——数据预处理管道
def normalize_data(data):
    return (data - np.mean(data)) / (np.std(data) + 1e-8)

def clip_outliers(data, lower=-3, upper=3):
    return np.clip(data, lower, upper)

def add_noise(data, scale=0.01):
    return data + np.random.normal(0, scale, data.shape)

# 用 reduce 组合多个函数
def compose(*functions):
    return lambda x: reduce(lambda acc, f: f(acc), functions, x)

preprocessing = compose(normalize_data, clip_outliers, add_noise)
raw = np.array([-10, 0, 0.5, 1.0, 5.0, 20.0])
processed = preprocessing(raw)
print(f"\n=== 函数组合管道 ===")
print(f"原始数据: {raw}")
print(f"处理后: {np.round(processed, 4)}")

什么用：reduce 在 AI 中用于累积聚合的场景——多折交叉验证汇总混淆矩阵、合并多个 batch 的损失值。函数组合（compose）是函数式编程的精髓——用 reduce 把多个小函数串联成管道。scikit-learn 的 Pipeline 和 PyTorch 的 nn.Sequential 的模块串联思想与 reduce 函组合同出一源。

概念关系图谱

重点答疑

Q1: map(func, data) 和 [func(x) for x in data] 有什么区别？

返回类型不同。 map() 返回迭代器（惰性求值），列表推导式返回立即计算的列表。对大数据，map() 内存友好。对已存在的小列表，列表推导式通常更可读。函数不同。 map() 可以接受现成的函数，列表推导式需要写表达式。

Q2: filter 和 map 返回的是迭代器，怎么快速转列表？

用 list() 包装即可：list(map(...))、list(filter(...))。如果要立即消费并转换成其他类型，也有 tuple(map(...))、set(filter(...))。注意：消费后迭代器就耗尽了——不能再次遍历。

Q3: 如何使用 reduce 实现一个管道操作符？

用 reduce 实现函数组合：def pipe(data, *funcs): return reduce(lambda d, f: f(d), funcs, data)。调用 pipe(data, f1, f2, f3) 等价于 f3(f2(f1(data)))。scikit-learn 的 Pipeline 和这个思想一致。

Q4: lambda 函数有什么限制？什么时候不应该用？

限制：(1) 只能包含单个表达式，不能有语句 (2) 不能有类型注解 (3) 不能有文档字符串。不该用的场景：(1) 逻辑超过一行 (2) 需要复用的逻辑 (3) 需要调试断点的逻辑。规则：lambda 写出来读不懂的内容就不要用 lambda。

章节单词汇总