快轉到主要內容
  1. 教學文章/

Python Generators/Yield 完全攻略:惰性運算的藝術

·7 分鐘· loading · loading · ·
Python Generator Yield Lazy Evaluation Iterator
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 38: 本文

一、前言
#

你有沒有遇過這種場景:要處理一個幾 GB 的 log 檔,結果 Python 一口氣把整個檔案讀進記憶體,然後……記憶體爆了 💥

或者你寫了一個函式回傳一個超大的 list,但其實呼叫端只需要前 10 個元素?白白浪費了計算和記憶體。

這就是 generator(生成器) 大顯身手的時候了!Generator 讓你用「惰性運算(lazy evaluation)」的方式,一次只產生一個值,需要時才計算下一個,記憶體用量從 O(n) 降到 O(1)。

今天拍拍君就帶大家從基礎到進階,徹底搞懂 Python 的 generator 與 yield 關鍵字!


二、從 Iterator 說起
#

在深入 generator 之前,先快速回顧 Python 的 iterator protocol

# 任何實作了 __iter__ 和 __next__ 的物件就是 iterator
class Countdown:
    def __init__(self, start):
        self.current = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.current <= 0:
            raise StopIteration
        val = self.current
        self.current -= 1
        return val

for n in Countdown(5):
    print(n)  # 5, 4, 3, 2, 1

這段程式碼可以跑,但寫起來有點囉嗦——要定義一個 class、維護狀態、處理 StopIteration……

Generator 就是 Python 給你的語法糖,讓你用超簡潔的方式寫出同樣的東西 🍬


三、Generator Function 基礎
#

3.1 第一個 Generator
#

只要函式裡出現 yield,它就自動變成一個 generator function

def countdown(start):
    while start > 0:
        yield start
        start -= 1

# 呼叫 generator function 不會執行函式內容
# 而是回傳一個 generator object
gen = countdown(5)
print(type(gen))  # <class 'generator'>

# 用 next() 逐一取值
print(next(gen))  # 5
print(next(gen))  # 4
print(next(gen))  # 3

# 當然也能用 for 迴圈
for n in countdown(3):
    print(n)  # 3, 2, 1

3.2 yield vs return 的差異
#

特性 return yield
函式行為 回傳值後結束 暫停並產出值,下次從暫停處繼續
呼叫結果 直接得到回傳值 得到 generator object
狀態保存 不保存 保存所有區域變數
可執行次數 一次 可以多次 yield
def simple_return():
    return 1
    return 2  # 永遠不會執行

def simple_yield():
    yield 1
    yield 2  # 第二次 next() 時會執行到這裡
    yield 3  # 第三次 next() 時會執行到這裡

gen = simple_yield()
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3
print(next(gen))  # StopIteration!

3.3 Generator 的執行流程
#

這是很多初學者容易搞混的地方,讓拍拍君用圖解說明:

def demo():
    print("A: 開始執行")
    yield 1
    print("B: 繼續執行")
    yield 2
    print("C: 結束前")

gen = demo()           # 什麼都不會印!只是建立 generator
val1 = next(gen)       # 印出 "A: 開始執行",val1 = 1,暫停在第一個 yield
val2 = next(gen)       # 印出 "B: 繼續執行",val2 = 2,暫停在第二個 yield
next(gen)              # 印出 "C: 結束前",然後 raise StopIteration

重點:generator 在 yield 處暫停,保存所有狀態,下次 next() 從暫停處繼續執行。


四、Generator Expression(生成器表達式)
#

就像 list comprehension 的懶惰版:

# List comprehension - 立即計算所有值,存在記憶體
squares_list = [x**2 for x in range(1_000_000)]  # 占用大量記憶體

# Generator expression - 惰性計算,幾乎不占記憶體
squares_gen = (x**2 for x in range(1_000_000))    # 只是一個 generator

import sys
print(sys.getsizeof(squares_list))  # ~8,000,056 bytes (約 8 MB)
print(sys.getsizeof(squares_gen))   #        200 bytes (固定大小!)

4.1 常見用法
#

# 搭配 sum、max、min 等內建函式
total = sum(x**2 for x in range(100))  # 不需要額外括號
largest = max(len(line) for line in open("data.txt"))

# 搭配 any / all 做條件檢查
has_negative = any(x < 0 for x in numbers)
all_positive = all(x > 0 for x in numbers)

# 鏈式過濾
result = sum(
    x**2
    for x in range(1000)
    if x % 3 == 0
    if x % 5 != 0
)

4.2 什麼時候用 list comp vs generator expr?
#

  • 需要多次迭代 → list comprehension(generator 只能迭代一次!)
  • 只需要迭代一次 → generator expression
  • 資料量很大 → generator expression(省記憶體)
  • 需要 indexing / slicing → list comprehension

五、實戰應用
#

5.1 處理大型檔案
#

這是 generator 最經典的應用場景:

def read_large_file(filepath):
    """逐行讀取大檔案,不需一次載入全部"""
    with open(filepath, "r") as f:
        for line in f:
            yield line.strip()

def filter_errors(lines):
    """過濾出 ERROR 開頭的行"""
    for line in lines:
        if line.startswith("ERROR"):
            yield line

def parse_timestamp(lines):
    """擷取時間戳記"""
    for line in lines:
        parts = line.split(" ", 3)
        if len(parts) >= 4:
            yield {"time": parts[1], "message": parts[3]}

# 管道式處理 - 超級優雅!
lines = read_large_file("server.log")
errors = filter_errors(lines)
parsed = parse_timestamp(errors)

# 整個管道在這裡才開始執行
for entry in parsed:
    print(f"[{entry['time']}] {entry['message']}")

這就是所謂的 generator pipeline(生成器管道)!每一步都只在需要時才計算,記憶體用量極低。

5.2 無限序列
#

Generator 可以產生無限多個值(只要你不要全部收集起來):

def fibonacci():
    """無限費波那契數列"""
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 取前 10 個
from itertools import islice
first_10 = list(islice(fibonacci(), 10))
print(first_10)  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

# 找第一個超過 1000 的費波那契數
for n in fibonacci():
    if n > 1000:
        print(n)  # 1597
        break
def infinite_id():
    """無限 ID 產生器"""
    n = 1
    while True:
        yield f"ITEM-{n:06d}"
        n += 1

id_gen = infinite_id()
print(next(id_gen))  # ITEM-000001
print(next(id_gen))  # ITEM-000002

5.3 滑動視窗(Sliding Window)
#

from collections import deque

def sliding_window(iterable, size):
    """產生固定大小的滑動視窗"""
    it = iter(iterable)
    window = deque(maxlen=size)

    # 先填滿視窗
    for _ in range(size):
        window.append(next(it))
    yield tuple(window)

    # 之後每次進一個、出一個
    for item in it:
        window.append(item)
        yield tuple(window)

data = [1, 2, 3, 4, 5, 6, 7]
for window in sliding_window(data, 3):
    print(window)
# (1, 2, 3)
# (2, 3, 4)
# (3, 4, 5)
# (4, 5, 6)
# (5, 6, 7)

六、進階:send、throw 與 close
#

6.1 send() — 向 Generator 傳值
#

yield 不只能往外送值,還能接收外部傳進來的值!

def accumulator():
    """累加器:每次 send 一個數字,yield 目前的總和"""
    total = 0
    while True:
        value = yield total  # yield 同時送出 total,並接收 send 的值
        if value is None:
            break
        total += value

gen = accumulator()
next(gen)          # 啟動 generator(第一次必須用 next 或 send(None))
print(gen.send(10))  # 10 (total = 0 + 10)
print(gen.send(20))  # 30 (total = 10 + 20)
print(gen.send(5))   # 35 (total = 30 + 5)

send() 的執行流程:

  1. 將值傳入 generator,作為 yield 表達式的結果
  2. Generator 從 yield 處繼續執行
  3. 執行到下一個 yield,產出新的值

6.2 throw() — 向 Generator 拋出例外
#

def careful_generator():
    try:
        while True:
            value = yield "等待中..."
            print(f"收到: {value}")
    except ValueError as e:
        yield f"處理錯誤: {e}"
    finally:
        print("清理資源")

gen = careful_generator()
print(next(gen))                        # "等待中..."
print(gen.throw(ValueError, "壞資料"))  # "處理錯誤: 壞資料"

6.3 close() — 優雅關閉 Generator
#

def resource_generator():
    print("開啟連線")
    try:
        while True:
            yield "資料"
    except GeneratorExit:
        print("收到關閉訊號,清理中...")
    finally:
        print("關閉連線")

gen = resource_generator()
next(gen)    # "開啟連線"
gen.close()  # "收到關閉訊號,清理中..." → "關閉連線"

七、yield from — 委派生成器
#

Python 3.3 引入的 yield from 讓你可以把一個 generator 委派給另一個:

# 沒有 yield from
def chain_old(*iterables):
    for it in iterables:
        for item in it:
            yield item

# 用 yield from(更簡潔!)
def chain_new(*iterables):
    for it in iterables:
        yield from it

result = list(chain_new([1, 2], [3, 4], [5]))
print(result)  # [1, 2, 3, 4, 5]

7.1 遞迴結構攤平
#

yield from 在處理巢狀結構時特別好用:

def flatten(nested):
    """遞迴攤平任意深度的巢狀 list"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)
        else:
            yield item

data = [1, [2, [3, 4]], [5, [6, [7]]]]
print(list(flatten(data)))  # [1, 2, 3, 4, 5, 6, 7]

7.2 樹的走訪
#

class TreeNode:
    def __init__(self, val, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

def inorder(node):
    """中序走訪(左 → 根 → 右)"""
    if node is None:
        return
    yield from inorder(node.left)
    yield node.val
    yield from inorder(node.right)

#       4
#      / \
#     2   6
#    / \
#   1   3
tree = TreeNode(4,
    TreeNode(2, TreeNode(1), TreeNode(3)),
    TreeNode(6)
)

print(list(inorder(tree)))  # [1, 2, 3, 4, 6]

八、搭配 itertools 更強大
#

Python 的 itertools 模組跟 generator 是天生一對(之前拍拍君也寫過 itertools 專文):

from itertools import chain, islice, count, takewhile, dropwhile

# chain - 串接多個 iterable
combined = chain(range(3), range(10, 13))
print(list(combined))  # [0, 1, 2, 10, 11, 12]

# islice - 切片(不需要轉成 list)
first_5_fib = list(islice(fibonacci(), 5))

# takewhile - 取到條件不成立為止
small_fibs = list(takewhile(lambda x: x < 100, fibonacci()))
print(small_fibs)  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89]

# count - 無限計數器
for i, char in zip(count(1), "abc"):
    print(f"{i}: {char}")  # 1: a, 2: b, 3: c

九、常見陷阱與最佳實踐
#

9.1 Generator 只能迭代一次!
#

gen = (x**2 for x in range(5))
print(list(gen))  # [0, 1, 4, 9, 16]
print(list(gen))  # [] ← 空的!已經耗盡了

如果需要多次迭代,用 generator function(每次呼叫產生新的 generator):

def squares(n):
    return (x**2 for x in range(n))

print(list(squares(5)))  # [0, 1, 4, 9, 16]
print(list(squares(5)))  # [0, 1, 4, 9, 16] ← 每次都是新的!

9.2 不要在 generator 裡偷改外部狀態
#

# ❌ 不好的做法
results = []
def bad_generator(data):
    for item in data:
        results.append(item)  # side effect!
        yield item * 2

# ✅ 好的做法:保持 generator 純淨
def good_generator(data):
    for item in data:
        yield item * 2

9.3 注意 generator 中的例外處理
#

def safe_parse(lines):
    """安全地解析每一行,跳過格式錯誤的"""
    for i, line in enumerate(lines, 1):
        try:
            yield int(line.strip())
        except ValueError:
            print(f"警告:第 {i} 行格式錯誤,已跳過")

data = ["10", "abc", "30", "", "50"]
print(list(safe_parse(data)))
# 警告:第 2 行格式錯誤,已跳過
# 警告:第 4 行格式錯誤,已跳過
# [10, 30, 50]

9.4 效能比較
#

import time
import sys

N = 10_000_000

# List approach
start = time.time()
total_list = sum([i * 2 for i in range(N)])
t_list = time.time() - start
mem_list = sys.getsizeof([i * 2 for i in range(N)])

# Generator approach
start = time.time()
total_gen = sum(i * 2 for i in range(N))
t_gen = time.time() - start

print(f"List:      {t_list:.3f}s, ~{mem_list / 1024 / 1024:.0f} MB")
print(f"Generator: {t_gen:.3f}s, ~0 MB (固定)")
# 典型結果:
# List:      1.2s, ~80 MB
# Generator: 0.8s, ~0 MB (固定)

結語
#

Generator 是 Python 裡最優雅的特性之一。掌握了它,你就能:

  • 🧠 省記憶體 — 處理 GB 級資料不再怕 OOM
  • 省計算 — 需要多少算多少,不做白工
  • 🔗 建立 pipeline — 像水管一樣串接資料處理步驟
  • ♾️ 表達無限 — 無限序列、串流資料都能優雅處理

從今天開始,看到 for 迴圈裡 append 到 list 的程式碼,先想一下:「這裡是不是可以用 yield?」

記得:能 yield 的地方就 yield,你的記憶體會感謝你的! 🎉


延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 38: 本文

相關文章

Python click:比 argparse 更優雅的 CLI 框架
·10 分鐘· loading · loading
Python Click Cli 命令列 開發工具
Python pytest:fixture + parametrize + mock 完整指南
·8 分鐘· loading · loading
Python Pytest Testing Fixture Mock Parametrize TDD
Python threading:多執行緒並行的正確打開方式
·8 分鐘· loading · loading
Python Threading Concurrency 並行 GIL
Python Profiling:cProfile + line_profiler 效能分析完全指南
·8 分鐘· loading · loading
Python Profiling CProfile Line_profiler Performance Optimization
Python subprocess:外部命令執行與管道串接完全指南
·8 分鐘· loading · loading
Python Subprocess Shell Automation Cli
Python 裝飾器:讓你的函式穿上超能力外套
·7 分鐘· loading · loading
Python Decorator 裝飾器 進階語法 設計模式