《Python AI 应用开发入门》第 3.7 节。
使用推导式表达数据转换,理解迭代协议并用生成器按需处理数据。
本节目标
学完本节后,你应当能够:
- 4. 使用
iter() 和 next() 观察迭代过程。
1. 推导式解决什么问题
把所有消息内容转为列表:
contents = []for message in messages: contents.append(message["content"])
列表推导式:
contents = [ message["content"] for message in messages]
它表达“从每条消息取出内容,组成新列表”。
推导式适合单一、清楚的数据转换,不是为了让代码尽可能短。
2. 列表推导式
基本形式:
[结果表达式 for 临时变量 in 可迭代对象]
字符串清理:
raw_tags = [" Python ", " AI ", " RAG "]tags = [tag.strip() for tag in raw_tags]
带过滤条件:
user_messages = [ message for message in messages if message["role"] == "user"]
先过滤再转换:
user_contents = [ message["content"] for message in messages if message["role"] == "user"]
这里的条件放在末尾,表示只保留满足条件的项目。
3. 推导式中的条件表达式
需要对每项二选一时:
labels = [ "用户" if message["role"] == "user" else "其他" for message in messages]
区别:
[value for value in values if condition] # 过滤[a if condition else b for value in values] # 每项转换
两者位置和含义不同。
4. 字典推导式
统计结果转换为显示标签:
role_counts = { "user": 3, "assistant": 2,}display_counts = { role: f"{count} 条" for role, count in role_counts.items()}
从配置中选择允许字段:
allowed_keys = {"model", "style", "max_history"}filtered_config = { key: value for key, value in config.items() if key in allowed_keys}
如果多个项目产生相同键,后面的值会覆盖前面的值。设计键时要保证符合业务唯一性。
5. 集合推导式
得到不重复角色:
roles = { message["role"] for message in messages}
标准化标签并去除空值:
normalized_tags = { tag.strip().lower() for tag in raw_tags if tag.strip()}
集合不保证业务显示顺序,需要展示时使用 sorted()。
6. 何时不要使用推导式
下面的逻辑虽然可以塞进推导式,但不易读:
valid_messages = []for index, message in enumerate(messages, start=1): try: valid_messages.append(validate_message(message)) except ValueError as error: print(f"跳过第 {index} 条:{error}")
它包含:
保留普通循环更清楚。
经验规则:
7. 可迭代对象
可以逐项遍历的对象叫可迭代对象,例如:
它们可以放在 for ... in ... 中:
for message in messages: print(message)
“可迭代”不等于“支持索引”。集合和生成器可以迭代,但不能使用 values[0]。
8. 迭代器
iter() 从可迭代对象获得迭代器:
messages = ["A", "B", "C"]iterator = iter(messages)print(next(iterator))print(next(iterator))print(next(iterator))
继续调用:
# next(iterator)
会触发 StopIteration,表示没有下一项。
可以提供默认值:
print(next(iterator, "没有更多消息"))
迭代器会记录当前位置。已经取出的项目不会自动重新出现。
9. for 循环与迭代协议
下面代码:
for message in messages: print(message)
可以近似理解为:
iterator = iter(messages)while True: try: message = next(iterator) except StopIteration: break print(message)
实际使用中让 for 自动处理 StopIteration。理解协议有助于理解文件、生成器和自定义迭代对象。
10. 迭代器只能继续向前
iterator = iter(["A", "B"])print(list(iterator))print(list(iterator))
第一次得到 ["A", "B"],第二次得到空列表,因为迭代器已经耗尽。
列表可以反复创建新迭代器:
messages = ["A", "B"]print(list(messages))print(list(messages))
收到迭代器或生成器时,不要在调试输出中先完整消费一遍,又期待后续逻辑还能读取。
11. 生成器表达式
把列表推导式的方括号换成圆括号:
lengths = ( len(message["content"]) for message in messages)
这不会立即创建包含全部长度的列表,而是返回生成器,使用时按需计算:
for length in lengths: print(length)
对比:
length_list = [len(text) for text in contents]length_generator = (len(text) for text in contents)
- • 生成器按需产生结果,占用内存更少,但通常只能消费一次。
数据量很小、需要反复使用时,列表更简单。
12. 生成器函数与 yield
函数中出现 yield,调用时会返回生成器:
def iter_user_messages(messages): for message in messages: if message["role"] == "user": yield message
调用函数不会立即执行完整函数体:
user_messages = iter_user_messages(messages)
开始迭代时才运行。每次执行到 yield:
for message in user_messages: print(message["content"])
13. 生成器中的 return
生成器执行 return 或到达函数末尾时停止:
def take_messages(messages, limit): if limit <= 0: return for index, message in enumerate(messages): if index >= limit: return yield message
这里的 return 不会像普通函数那样成为循环中的下一项,它表示生成结束。
14. 按需搜索消息
def search_messages( messages: list[dict[str, str]], keyword: str,): normalized_keyword = keyword.strip().lower() if not normalized_keyword: return for message in messages: if normalized_keyword in message["content"].lower(): yield message
调用者可以只取第一个匹配:
matches = search_messages(messages, "Python")first_match = next(matches, None)
也可以全部遍历:
for message in search_messages(messages, "Python"): print(message["content"])
类型注解可以写成 Iterator[dict[str, str]],需要从 collections.abc 导入:
from collections.abc import Iteratordef search_messages( messages: list[dict[str, str]], keyword: str,) -> Iterator[dict[str, str]]: ...
15. 逐块读取大文件
from collections.abc import Iteratorfrom pathlib import Pathdef read_chunks( path: Path, chunk_size: int = 1024,) -> Iterator[str]: if chunk_size <= 0: raise ValueError("chunk_size 必须大于 0") with path.open(mode="r", encoding="utf-8") as file: while True: chunk = file.read(chunk_size) if not chunk: break yield chunk
使用:
for chunk in read_chunks(Path("large_notes.txt"), chunk_size=4096): print(f"本块字符数:{len(chunk)}")
文件会在生成器迭代结束或生成器被关闭时离开 with。调用者不应创建后完全不消费并长期保存生成器。
16. 生成器的优势与限制
优势:
限制:
不要看到“大文件”就自动使用生成器。先确认是否真的需要按需处理。
17. 完成章节项目
最终结构:
stage2/└── chat_app/ ├── __init__.py ├── main.py ├── config.py ├── prompts.py ├── storage.py └── data/ ├── config.json ├── prompt_template.txt └── history.json
职责:
- •
config.py:加载并验证 JSON 配置。 - •
prompts.py:读取模板并构造 Prompt。 - •
storage.py:验证、加载和安全保存历史。
main.py 的主流程:
def main() -> None: try: config = load_config(CONFIG_PATH) messages = load_history(HISTORY_PATH) except ValueError as error: print(f"启动失败:{error}") return while True: show_menu() command = input("请选择:").strip() if command == "1": add_message_flow(messages, config) save_history(HISTORY_PATH, messages) elif command == "2": display_messages(messages) elif command == "3": search_flow(messages) elif command == "0": print("再见") break else: print("无效选项")
从 stage2 目录启动:
python -m chat_app.main
至少验证:
动手练习
练习 1:三种推导式
从聊天历史中:
练习 2:观察迭代器耗尽
对同一个迭代器连续调用 next(),再两次转换为列表。记录每一步结果并解释原因。
练习 3:完成章节项目
整合 7 节内容完成模块化聊天程序,并记录至少 10 个测试场景。
随堂小测
- 2. 过滤条件和条件表达式在推导式中的位置有何不同?
参考答案
- 2. 过滤条件放在
for 之后;二选一表达式写在结果位置并包含 else。 - 3. 可迭代对象能创建迭代器;迭代器记录当前位置并通过
next() 逐项返回。 - 4. 默认触发
StopIteration,也可以给 next() 提供默认值。 - 5. 它会保存并推进当前位置,耗尽后没有项目可返回。
- 6. 列表立即保存全部结果;生成器按需计算,通常只能消费一次。
- 7. 调用返回生成器,迭代到
yield 时产生值并暂停,下次继续。 - 9. 创建生成器时函数体尚未完整执行,相关代码在请求下一项时才运行。