1.2 高阶函数与函数式编程

引言：函数作为一等公民

在 Python 中，函数不仅仅是代码块——它们是一等公民（First-Class Citizens）。这意味着函数可以：

赋值给变量
作为参数传递给其他函数
作为返回值从函数中返回
存储在数据结构中（如列表、字典）

这种特性使得 Python 能够支持函数式编程范式，而这正是 LangChain 和 LangGraph 大量使用的编程风格。

通俗理解：想象函数就像一张食谱卡片。在其他编程语言中，食谱只能被厨师使用；但在 Python 中，这张食谱卡片可以：
交给别人（赋值给变量）
放进一个"食谱盒"里（存储在列表中）
告诉另一个厨师"按这个做"（作为参数传递）
甚至可以"制造"出新的食谱（作为返回值）

学习目标

✅ 理解函数作为一等公民的概念
✅ 掌握 map(), filter(), reduce() 等高阶函数
✅ 使用 Lambda 表达式简化代码
✅ 理解闭包（Closure）机制
✅ 实战：构建可配置的 Agent 工厂

第一部分：函数作为一等公民

为什么需要理解这个？

小白解读：如果你不理解"函数是一等公民"，你会觉得很多高级代码"看不懂"。比如 LangGraph 中经常出现：
python
workflow.add_conditional_edges("agent", some_function)
1
为什么 some_function 后面没有括号 () ？因为我们在传递函数本身，而不是调用它！

函数可以赋值给变量

python

def greet(name: str) -> str:
    return f"Hello, {name}!"

# 将函数赋值给变量
say_hello = greet

# 通过变量调用函数
message = say_hello("Claude")
print(message)  # Hello, Claude!

# 函数名本身就是对函数对象的引用
print(type(greet))  # <class 'function'>
print(greet)        # <function greet at 0x...>

通俗理解：
greet 是函数的"名字"，就像你的名字
say_hello = greet 就像给你起一个"外号"
用名字或外号都能找到你（调用你）
关键：greet（没有括号）= 函数本身；greet()（有括号）= 调用函数

函数可以作为参数传递

python

def apply_operation(
    value: int,
    operation: callable
) -> int:
    """
    对值应用操作函数

    Args:
        value: 输入值
        operation: 要应用的函数

    Returns:
        操作后的结果
    """
    return operation(value)

# 定义几个操作函数
def double(x: int) -> int:
    return x * 2

def square(x: int) -> int:
    return x ** 2

# 传递不同的函数
result1 = apply_operation(5, double)  # 10
result2 = apply_operation(5, square)  # 25

print(f"Double: {result1}, Square: {result2}")

通俗理解：想象你是一个万能计算器（apply_operation）：
用户给你一个数字（value）
用户还给你一张"操作说明"（operation）
你按照说明来处理这个数字
这样设计的好处是：计算器不需要提前知道所有可能的操作。用户可以随时"发明"新操作，然后传给你！

🔗 与 Agent 的联系：LangGraph 的条件路由就是传递函数作为参数：
python
workflow.add_conditional_edges(
    "agent",
    lambda x: "tools" if x["needs_tool"] else "end"
)
1
2
3
4

函数可以作为返回值

python

def create_multiplier(factor: int) -> callable:
    """
    创建一个乘法函数工厂

    Args:
        factor: 乘数因子

    Returns:
        一个执行乘法的函数
    """
    def multiplier(x: int) -> int:
        return x * factor

    return multiplier

# 创建不同的乘法器
double = create_multiplier(2)
triple = create_multiplier(3)

print(double(5))  # 10
print(triple(5))  # 15

通俗理解 - 函数工厂：
想象 create_multiplier 是一个机器制造厂：
你告诉工厂："我要一个×2的机器"
工厂给你造出 double 这台机器
你告诉工厂："我要一个×3的机器"
工厂给你造出 triple 这台机器
每台机器都是独立的，互不影响。这就是"函数返回函数"的威力！

第二部分：内置高阶函数

什么是高阶函数？
接受函数作为参数，或者返回函数的函数，就叫高阶函数。
类比：普通函数像是"工人"，只能干活；高阶函数像是"包工头"，可以指挥工人干活。

map() - 映射转换

一句话解释：map() 就是"对列表中的每一个元素执行相同的操作"。

python

# 传统方式
messages = ["hello", "world", "ai"]
uppercased = []
for msg in messages:
    uppercased.append(msg.upper())

# 使用 map()
messages = ["hello", "world", "ai"]
uppercased = list(map(str.upper, messages))
print(uppercased)  # ['HELLO', 'WORLD', 'AI']

# 使用自定义函数
def format_message(msg: str) -> str:
    return f"[Agent] {msg}"

formatted = list(map(format_message, messages))
print(formatted)  # ['[Agent] hello', '[Agent] world', '[Agent] ai']

通俗理解 - 流水线工厂：
想象你有一条流水线：
原材料 → [工人1] → [工人2] → [工人3] → 成品
hello     ↓         ↓          ↓
          HELLO     HELLO      HELLO
world     ↓         ↓          ↓
          WORLD     WORLD      WORLD
ai        ↓         ↓          ↓
          AI        AI         AI
1
2
3
4
5
6
7
map(str.upper, messages) 就是：
列表中每个元素依次进入流水线
每个元素都被 str.upper 这个"工人"处理
处理完的结果组成新列表

AI Agent 应用：

python

# 批量格式化提示词
prompts = ["Explain AI", "What is ML", "Define NLP"]

def create_full_prompt(query: str) -> str:
    return f"You are a helpful assistant. User asks: {query}"

full_prompts = list(map(create_full_prompt, prompts))

filter() - 过滤筛选

一句话解释：filter() 就是"只保留符合条件的元素"。

python

# 过滤出有效的消息（非空）
messages = ["Hello", "", "World", "", "AI"]
valid_messages = list(filter(lambda x: x != "", messages))
print(valid_messages)  # ['Hello', 'World', 'AI']

# 过滤出高置信度的结果
results = [
    {"text": "Result 1", "confidence": 0.9},
    {"text": "Result 2", "confidence": 0.3},
    {"text": "Result 3", "confidence": 0.8},
]

def is_high_confidence(result: dict) -> bool:
    return result["confidence"] >= 0.7

high_conf_results = list(filter(is_high_confidence, results))
print(len(high_conf_results))  # 2

通俗理解 - 安检门：
想象 filter() 是一道安检门：
所有乘客 → [安检门：有票吗？] → 通过的乘客
   张三 ✅    →    通过
   李四 ❌    →    拦下
   王五 ✅    →    通过
1
2
3
4
filter(is_high_confidence, results) 就是：
每个结果都要过"安检"
安检员（is_high_confidence）检查是否符合条件
只有返回 True 的才能通过

reduce() - 累积聚合

一句话解释：reduce() 就是"把列表中的所有元素逐个累积成一个结果"。

python

from functools import reduce

# 计算总 token 数
token_counts = [100, 250, 180, 320]
total = reduce(lambda a, b: a + b, token_counts)
print(f"Total tokens: {total}")  # 850

# 更复杂的例子：合并多个字典
configs = [
    {"model": "gpt-3.5-turbo"},
    {"temperature": 0.7},
    {"max_tokens": 2000},
]

def merge_dicts(dict1: dict, dict2: dict) -> dict:
    return {**dict1, **dict2}

merged_config = reduce(merge_dicts, configs)
print(merged_config)
# {'model': 'gpt-3.5-turbo', 'temperature': 0.7, 'max_tokens': 2000}

通俗理解 - 滚雪球：
想象你在滚雪球：
起点     第1次      第2次       第3次       最终
 0   →  0+100  →  100+250  →  350+180  →  530+320 = 850
        =100      =350        =530
1
2
3
reduce(lambda a, b: a + b, [100, 250, 180, 320]) 就是：
从左到右，两两累积
a 是之前累积的结果（雪球大小）
b 是当前要加入的元素（新雪）
最后雪球包含了所有的雪

第三部分：Lambda 表达式

为什么需要 Lambda？

小白解读：有时候你只需要一个一次性的小函数，专门为它写 def 太麻烦了。Lambda 就是"临时工"函数！

Lambda 基础

python

# 普通函数
def add(x: int, y: int) -> int:
    return x + y

# 等价的 Lambda 表达式
add_lambda = lambda x, y: x + y

print(add(3, 5))        # 8
print(add_lambda(3, 5)) # 8

通俗理解 - Lambda 语法解析：

python

lambda x, y: x + y
│      │  │  └── 返回什么（表达式）
│      │  └── 冒号分隔
│      └── 参数列表
└── 关键字"lambda"，告诉 Python 这是个匿名函数

相当于：

python

def ???(x, y):    # 没有名字
    return x + y

Lambda 的适用场景

python

# 1. 作为 sort 的 key 参数
agents = [
    {"name": "Agent A", "score": 85},
    {"name": "Agent B", "score": 92},
    {"name": "Agent C", "score": 78},
]

# 按分数排序
sorted_agents = sorted(agents, key=lambda x: x["score"], reverse=True)
print([a["name"] for a in sorted_agents])  # ['Agent B', 'Agent A', 'Agent C']

# 2. 在 map/filter 中使用
numbers = [1, 2, 3, 4, 5]
squares = list(map(lambda x: x ** 2, numbers))
evens = list(filter(lambda x: x % 2 == 0, numbers))

# 3. LangGraph 条件路由
# workflow.add_conditional_edges(
#     "agent",
#     lambda x: "continue" if x["iteration"] < 5 else "end"
# )

通俗理解 - 排序的 key 参数：
想象你要给学生按成绩排名：
学生数据：[{"name": "小明", "score": 85}, ...]
sorted() 问你："怎么比较两个学生的大小？"
你说："lambda x: x["score"]"（看分数）
于是 sorted() 就按分数来排序
key 参数就像是给 sorted() 一个"比较说明书"

⚠️ Lambda 的限制：
只能包含单个表达式
不能包含语句（如 if/for/while）
不能有类型注解
难以调试

最佳实践：

✅ 简单的单行操作使用 Lambda
❌ 复杂逻辑使用命名函数

第四部分：闭包（Closure）

什么是闭包？

闭包是一个函数，它"记住"了定义时的外部作用域的变量。

小白解读：闭包是这门课最难理解的概念之一，但也是最强大的！
通俗比喻 - 记忆胶囊：
想象你给儿子一个"记忆胶囊"（闭包）：
胶囊里装着你家的地址（外部变量）
儿子带着胶囊去了外地（函数被返回）
多年后儿子打开胶囊，还能找到回家的路（访问外部变量）
即使外部函数已经执行完毕，闭包仍然"记得"那些变量！

python

def create_counter(start: int = 0):
    """
    创建一个计数器闭包

    Args:
        start: 起始值

    Returns:
        计数器函数
    """
    count = start  # 外部作用域的变量

    def counter() -> int:
        nonlocal count  # 声明使用外部变量
        count += 1
        return count

    return counter

# 创建两个独立的计数器
counter1 = create_counter(0)
counter2 = create_counter(100)

print(counter1())  # 1
print(counter1())  # 2
print(counter2())  # 101
print(counter1())  # 3

图解闭包原理：

create_counter(0) 被调用
┌─────────────────────────────┐
│  count = 0  ← 外部变量       │
│                             │
│  def counter():             │
│      nonlocal count         │
│      count += 1   ← 使用外部变量
│      return count           │
│                             │
│  return counter ← 返回内部函数 │
└─────────────────────────────┘
        │
        ▼
counter1 = <counter函数 + 它记住的count=0>

当调用 counter1() 时：
- 它找到自己"记住"的 count（现在是0）
- 把 count 加 1（变成1）
- 返回 1

再次调用 counter1()：
- 它找到自己"记住"的 count（现在是1）
- 把 count 加 1（变成2）
- 返回 2

关键点：

counter1 和 counter2 各自"记住"了自己的 count
它们互不影响，就像两个独立的计数器

`nonlocal` 关键字是什么？

python

def outer():
    x = 10

    def inner():
        nonlocal x  # 告诉 Python：我要用外面的 x，不是创建新的
        x = 20      # 修改外部的 x

    inner()
    print(x)  # 20

outer()

通俗理解：
如果不写 nonlocal x，x = 20 会创建一个新的局部变量 x
写了 nonlocal x，就是告诉 Python："我说的是外面那个 x"

AI Agent 中的闭包应用

python

def create_rate_limiter(max_calls: int, window_seconds: int):
    """
    创建一个速率限制器（闭包实现）

    Args:
        max_calls: 时间窗口内最大调用次数
        window_seconds: 时间窗口（秒）

    Returns:
        检查函数
    """
    import time
    from collections import deque

    call_times = deque(maxlen=max_calls)

    def can_call() -> bool:
        """检查是否可以发起调用"""
        now = time.time()

        # 移除窗口外的调用记录
        while call_times and call_times[0] < now - window_seconds:
            call_times.popleft()

        if len(call_times) < max_calls:
            call_times.append(now)
            return True

        return False

    return can_call

# 使用速率限制器
rate_limiter = create_rate_limiter(max_calls=3, window_seconds=10)

for i in range(5):
    if rate_limiter():
        print(f"调用 #{i+1} 允许")
    else:
        print(f"调用 #{i+1} 被限制")

为什么用闭包实现速率限制器？
因为 call_times（调用记录）需要在多次调用之间持续存在：
如果用普通函数，每次调用都会重新创建 call_times
用闭包，call_times 被"记住"了，可以累积调用记录

第五部分：实战 - 可配置的 Agent 工厂

python

"""
Agent 工厂：使用高阶函数和闭包构建可配置的 Agent
"""

from typing import Callable, Optional


def create_agent_factory(
    default_model: str = "gpt-3.5-turbo",
    default_temperature: float = 0.7
) -> Callable:
    """
    创建 Agent 工厂函数

    这是一个高阶函数，返回一个配置好的 Agent 创建器

    Args:
        default_model: 默认模型
        default_temperature: 默认温度

    Returns:
        Agent 创建函数
    """
    def create_agent(
        name: str,
        system_prompt: str,
        tools: Optional[list] = None,
        model: Optional[str] = None,
        temperature: Optional[float] = None
    ) -> dict:
        """
        创建 Agent 实例

        Args:
            name: Agent 名称
            system_prompt: 系统提示词
            tools: 可用工具列表
            model: 模型（可选，使用默认值）
            temperature: 温度（可选，使用默认值）

        Returns:
            Agent 配置字典
        """
        return {
            "name": name,
            "model": model or default_model,
            "temperature": temperature or default_temperature,
            "system_prompt": system_prompt,
            "tools": tools or [],
            "created_with_defaults": {
                "model": default_model,
                "temperature": default_temperature
            }
        }

    return create_agent


def create_tool_registry():
    """
    创建工具注册表（使用闭包）

    Returns:
        包含注册和获取函数的字典
    """
    tools: dict[str, Callable] = {}

    def register(name: str, func: Callable) -> None:
        """注册工具"""
        tools[name] = func
        print(f"工具 '{name}' 已注册")

    def get(name: str) -> Optional[Callable]:
        """获取工具"""
        return tools.get(name)

    def list_tools() -> list[str]:
        """列出所有工具"""
        return list(tools.keys())

    return {
        "register": register,
        "get": get,
        "list": list_tools
    }


def compose(*functions: Callable) -> Callable:
    """
    函数组合：将多个函数组合成一个

    Args:
        *functions: 要组合的函数列表

    Returns:
        组合后的函数

    Example:
        >>> def add_one(x): return x + 1
        >>> def double(x): return x * 2
        >>> f = compose(add_one, double)
        >>> f(3)  # double(add_one(3)) = double(4) = 8
        8
    """
    from functools import reduce

    def compose_two(f: Callable, g: Callable) -> Callable:
        return lambda x: g(f(x))

    return reduce(compose_two, functions, lambda x: x)


# 演示使用
def main() -> None:
    """主函数：演示高阶函数和闭包"""

    # 1. Agent 工厂
    print("=== Agent 工厂 ===")
    factory = create_agent_factory(
        default_model="gpt-4",
        default_temperature=0.5
    )

    agent1 = factory(
        name="ResearchBot",
        system_prompt="You are a research assistant"
    )

    agent2 = factory(
        name="ChatBot",
        system_prompt="You are a friendly chatbot",
        model="gpt-3.5-turbo",
        temperature=0.9
    )

    print(f"Agent1: {agent1['name']}, 模型: {agent1['model']}")
    print(f"Agent2: {agent2['name']}, 模型: {agent2['model']}")

    # 2. 工具注册表
    print("\n=== 工具注册表 ===")
    registry = create_tool_registry()

    def weather_tool(location: str) -> str:
        return f"Weather in {location}: Sunny"

    def calculator_tool(expression: str) -> str:
        return f"Calculating: {expression}"

    registry["register"]("weather", weather_tool)
    registry["register"]("calculator", calculator_tool)

    print(f"已注册工具: {registry['list']()}")

    tool = registry["get"]("weather")
    if tool:
        print(tool("Beijing"))

    # 3. 函数组合
    print("\n=== 函数组合 ===")

    def clean_text(text: str) -> str:
        return text.strip().lower()

    def remove_punctuation(text: str) -> str:
        import string
        return text.translate(str.maketrans("", "", string.punctuation))

    def tokenize(text: str) -> list[str]:
        return text.split()

    # 组合处理流程
    process_text = compose(clean_text, remove_punctuation)

    input_text = "  Hello, World!  "
    processed = process_text(input_text)
    print(f"原始: '{input_text}'")
    print(f"处理后: '{processed}'")


if __name__ == "__main__":
    main()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

本节总结

核心概念一览表

概念	一句话解释	生活比喻
一等公民	函数可以像变量一样传递	食谱卡片可以送人、收藏、复制
高阶函数	接受或返回函数的函数	包工头指挥工人干活
map()	对每个元素执行相同操作	流水线批量加工
filter()	保留符合条件的元素	安检门筛选乘客
reduce()	累积所有元素成一个结果	滚雪球越滚越大
Lambda	一次性的匿名小函数	临时工
闭包	记住外部变量的函数	带着家乡地址的游子

与 AI Agent 的联系

概念	在 Agent 中的应用
高阶函数	Agent 工厂、配置生成器
Lambda	LangGraph 条件路由
闭包	速率限制器、状态管理
map	批量处理提示词、结果
filter	筛选高质量响应
compose	构建处理管道

新手常见问题

Q1：greet 和 greet() 有什么区别？

greet = 函数对象本身（可以传递）
greet() = 调用函数，执行它的代码

Q2：Lambda 和普通函数有什么区别？

Lambda 是匿名的，一次性使用
普通函数有名字，可以复用
Lambda 只能写一行表达式

Q3：闭包有什么实际用途？

创建带"记忆"的函数（如计数器）
实现数据隐藏（外部无法直接访问内部变量）
工厂模式（批量创建配置好的函数）

下一节：1.3 装饰器：LangChain 的核心模式

1.2 高阶函数与函数式编程 ​

引言：函数作为一等公民 ​

学习目标 ​

第一部分：函数作为一等公民 ​

为什么需要理解这个？ ​

函数可以赋值给变量 ​

函数可以作为参数传递 ​

函数可以作为返回值 ​

第二部分：内置高阶函数 ​

map() - 映射转换 ​

filter() - 过滤筛选 ​

reduce() - 累积聚合 ​

第三部分：Lambda 表达式 ​

为什么需要 Lambda？ ​

Lambda 基础 ​

Lambda 的适用场景 ​

第四部分：闭包（Closure） ​

什么是闭包？ ​

nonlocal 关键字是什么？ ​

AI Agent 中的闭包应用 ​

第五部分：实战 - 可配置的 Agent 工厂 ​

本节总结 ​

核心概念一览表 ​

与 AI Agent 的联系 ​

新手常见问题 ​

1.2 高阶函数与函数式编程

引言：函数作为一等公民

学习目标

第一部分：函数作为一等公民

为什么需要理解这个？

函数可以赋值给变量

函数可以作为参数传递

函数可以作为返回值

第二部分：内置高阶函数

map() - 映射转换

filter() - 过滤筛选

reduce() - 累积聚合

第三部分：Lambda 表达式

为什么需要 Lambda？

Lambda 基础

Lambda 的适用场景

第四部分：闭包（Closure）

什么是闭包？

`nonlocal` 关键字是什么？

AI Agent 中的闭包应用

第五部分：实战 - 可配置的 Agent 工厂

本节总结

核心概念一览表

与 AI Agent 的联系

新手常见问题