1.6 小结和复习
恭喜！你已掌握函数式编程的精髓
通过 Module 1 的学习，你已经从"能写代码"进化到"会写优雅代码"。你现在掌握了：
- ✅ 函数基础：参数、返回值、类型注解
- ✅ 高阶函数：map/filter/reduce、Lambda、闭包
- ✅ 装饰器：Python 最强大的特性之一
- ✅ 模块化：组织大型项目的艺术
- ✅ LangChain Tool：真实世界的应用
这些技能让你能够阅读和理解 LangChain 的源码，构建可维护的 Agent 系统。
核心概念速查表
函数签名
python
def function_name(
required_param: type,
default_param: type = default_value,
*args,
**kwargs
) -> return_type:
"""文档字符串"""
pass高阶函数模式
python
# 函数作为参数
def apply(func, value):
return func(value)
# 函数作为返回值
def multiplier(n):
return lambda x: x * n
# 闭包
def counter():
count = 0
def increment():
nonlocal count
count += 1
return count
return increment装饰器模式
python
from functools import wraps
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 前置逻辑
result = func(*args, **kwargs)
# 后置逻辑
return result
return wrapper
# 带参数的装饰器
def decorator_with_args(param):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 使用 param
return func(*args, **kwargs)
return wrapper
return decorator模块导入
python
# 基本导入
import module
from module import function
from package.module import Class
# 相对导入（包内部）
from .module import function
from ..package import Class🔥 高难度编码挑战
挑战 1：装饰器链（难度：⭐⭐⭐⭐）
任务描述
实现一个装饰器管理系统，支持：
- 注册装饰器：将装饰器注册到系统中
- 按名称应用装饰器：通过配置文件应用装饰器
- 装饰器组合：支持多个装饰器的链式组合
- 条件装饰器：根据条件决定是否应用装饰器
- 装饰器优先级：支持装饰器排序
技术要求
- 使用高阶函数
- 使用装饰器
- 使用闭包
- 实现元编程
预期功能
python
# 注册装饰器
register_decorator("log", log_decorator)
register_decorator("retry", retry_decorator)
register_decorator("cache", cache_decorator)
# 通过配置应用装饰器
@apply_decorators(["log", "retry", "cache"])
def critical_function():
pass
# 条件装饰器
@conditional_decorator(lambda: DEBUG_MODE, debug_decorator)
def process_data():
pass
# 装饰器优先级
@apply_with_priority([
(log_decorator, 1), # 最外层
(retry_decorator, 2),
(cache_decorator, 3) # 最内层
])
def api_call():
pass你的解决方案
python
from functools import wraps
from typing import Callable, Dict, List, Tuple, Any
# 装饰器注册表
_decorator_registry: Dict[str, Callable] = {}
def register_decorator(name: str, decorator: Callable) -> None:
"""
注册装饰器到系统
Args:
name: 装饰器名称
decorator: 装饰器函数
"""
# TODO: 实现注册逻辑
pass
def apply_decorators(decorator_names: List[str]) -> Callable:
"""
应用多个装饰器
Args:
decorator_names: 装饰器名称列表
Returns:
组合后的装饰器
"""
# TODO: 实现
pass
def conditional_decorator(
condition: Callable[[], bool],
decorator: Callable
) -> Callable:
"""
条件装饰器：只在条件为真时应用
Args:
condition: 条件函数
decorator: 要应用的装饰器
Returns:
条件装饰器
"""
# TODO: 实现
pass
def apply_with_priority(
decorators_with_priority: List[Tuple[Callable, int]]
) -> Callable:
"""
按优先级应用装饰器
Args:
decorators_with_priority: (装饰器, 优先级) 元组列表
优先级数字越小越外层
Returns:
组合后的装饰器
"""
# TODO: 实现
pass
# 测试代码
def main():
# 定义一些测试装饰器
def log_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"[LOG] 调用 {func.__name__}")
return func(*args, **kwargs)
return wrapper
def timer_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
import time
start = time.time()
result = func(*args, **kwargs)
print(f"[TIMER] 耗时 {time.time() - start:.4f}秒")
return result
return wrapper
# 测试注册
register_decorator("log", log_decorator)
register_decorator("timer", timer_decorator)
# 测试应用
@apply_decorators(["log", "timer"])
def test_function():
print("执行函数")
test_function()
if __name__ == "__main__":
main()挑战 2：函数式数据处理管道（难度：⭐⭐⭐⭐⭐）
任务描述
实现一个函数式编程风格的数据处理管道，用于处理 Agent 的消息历史：
- 管道构建：链式调用多个处理函数
- Lazy Evaluation：延迟执行，只在需要时计算
- 并行处理：支持并行执行独立的处理步骤
- 错误恢复：处理失败时的回退机制
- 性能分析：统计每个步骤的执行时间
技术要求
- 使用高阶函数
- 使用闭包
- 使用生成器
- 实现链式调用
预期功能
python
# 创建处理管道
pipeline = (
Pipeline()
.filter(lambda msg: msg["role"] == "user")
.map(lambda msg: msg["content"].lower())
.filter(lambda text: len(text) > 10)
.map(lambda text: text.strip())
.distinct()
.take(5)
)
# 执行管道
messages = [
{"role": "user", "content": " Hello World! "},
{"role": "assistant", "content": "Hi there!"},
{"role": "user", "content": "How are you?"},
]
results = pipeline.execute(messages)
print(results)
# 性能分析
stats = pipeline.get_stats()
print(f"总耗时: {stats['total_time']}")
print(f"各步骤耗时: {stats['step_times']}")你的解决方案
python
from typing import Callable, List, Any, Iterator, Optional
import time
from functools import reduce
class Pipeline:
"""函数式数据处理管道"""
def __init__(self):
"""初始化管道"""
self.steps: List[Tuple[str, Callable]] = []
self.stats: Dict[str, Any] = {}
def filter(self, predicate: Callable[[Any], bool]) -> 'Pipeline':
"""
过滤元素
Args:
predicate: 过滤条件函数
Returns:
Pipeline 实例（支持链式调用）
"""
# TODO: 实现
pass
def map(self, transformer: Callable[[Any], Any]) -> 'Pipeline':
"""
映射转换
Args:
transformer: 转换函数
Returns:
Pipeline 实例
"""
# TODO: 实现
pass
def distinct(self) -> 'Pipeline':
"""去重"""
# TODO: 实现
pass
def take(self, n: int) -> 'Pipeline':
"""取前 n 个元素"""
# TODO: 实现
pass
def execute(self, data: List[Any]) -> List[Any]:
"""
执行管道
Args:
data: 输入数据
Returns:
处理后的数据
"""
# TODO: 实现
pass
def get_stats(self) -> Dict[str, Any]:
"""获取性能统计"""
return self.stats
# 测试代码
def main():
# 测试数据
messages = [
{"role": "user", "content": " Hello "},
{"role": "assistant", "content": "Hi!"},
{"role": "user", "content": "How are you?"},
{"role": "user", "content": " Hello "}, # 重复
{"role": "user", "content": "Python is great!"},
]
# 创建管道
pipeline = (
Pipeline()
.filter(lambda msg: msg["role"] == "user")
.map(lambda msg: msg["content"].strip())
.distinct()
.map(lambda text: text.lower())
.take(3)
)
# 执行
results = pipeline.execute(messages)
print("结果:", results)
# 统计
print("\n性能统计:")
print(pipeline.get_stats())
if __name__ == "__main__":
main()挑战答案提示
挑战 1 提示
点击查看提示
注册装饰器：
python
def register_decorator(name: str, decorator: Callable) -> None:
_decorator_registry[name] = decorator应用装饰器：
python
def apply_decorators(names: List[str]) -> Callable:
def decorator(func: Callable) -> Callable:
result = func
for name in reversed(names): # 反向应用
dec = _decorator_registry[name]
result = dec(result)
return result
return decorator挑战 2 提示
点击查看提示
链式调用：
python
def filter(self, predicate):
self.steps.append(("filter", predicate))
return self # 返回 self 支持链式调用执行管道：
python
def execute(self, data):
result = data
for step_name, step_func in self.steps:
if step_name == "filter":
result = [x for x in result if step_func(x)]
elif step_name == "map":
result = [step_func(x) for x in result]
return result自我评估清单
在进入 Module 2 之前，确保你能够：
- [ ] 独立编写带类型注解的函数
- [ ] 使用 map/filter/reduce 处理数据
- [ ] 理解并编写装饰器
- [ ] 使用闭包管理私有状态
- [ ] 组织模块化的项目结构
- [ ] 创建自定义 LangChain Tool
- [ ] 完成至少一个高难度挑战（60%+正确）
学习资源
推荐阅读
进阶主题
- 描述符（Descriptors）
- 元类（Metaclasses）
- 异步装饰器
- 类型系统高级用法
下一步
恭喜完成 Module 1！你已经掌握了函数式编程的核心技能。
在 Module 2: 数据结构与状态管理 中，你将学习：
- 列表、字典、集合的高级用法
- TypedDict 与 Pydantic：类型安全的状态管理
- 数据类（dataclass）
- Agent 状态的设计模式
- 实战：构建 LangGraph 的状态管理系统
继续前进，你离构建生产级 AI Agent 越来越近了！