批次處理與連續批次：提升吞吐的關鍵
從靜態批次、動態批次到連續批次，拆解調度策略、批次大小取捨，並用 Python 實作一個連續批次系統。
批次處理與連續批次：提升吞吐的關鍵
上一篇我們談了 vLLM 與 PagedAttention，理解了它們如何解決 KV Cache 的記憶體碎片問題，讓記憶體利用率從 20-40% 提升到 90% 以上。
但 PagedAttention 只是解決了「記憶體怎麼放」的問題，還沒有解決「請求怎麼排」的問題。
如果調度策略不對，即使記憶體利用率再高，GPU 仍然會閒置，吞吐仍然上不去。
這一篇，我們要深入批次處理的核心：靜態批次、動態批次、連續批次，理解它們的差異、設計高效的調度策略、平衡吞吐與延遲，並用 Python 實作一個連續批次系統。
一、為什麼需要批次處理？
單一請求的 GPU 利用率很低
當 GPU 只處理一個請求時，大部分算力都在閒置。
以一個 7B 模型為例：
| 指標 | 單一請求 | 批次 32 |
|---|---|---|
| GPU 算力利用率 | 15% | 85% |
| 記憶體頻寬利用率 | 30% | 90% |
| 每秒生成 token 數 | 30 | 600 |
| 每 token 成本 | $0.001 | $0.00005 |
單一請求時，GPU 大部分時間在等記憶體讀取，而不是在計算。
批次處理讓多個請求同時計算，把 GPU 餵飽。
批次處理的核心權衡
吞吐 ↑ → 批次 ↑ → 延遲 ↑批次越大，吞吐越高，但每個請求的等待時間也越長。
這是因為：
- 排隊延遲：請求要等批次湊滿才能開始。
- 計算延遲：批次越大，每個步驟的計算時間越長。
- 同步延遲：批次內所有請求要一起完成。
什麼時候該用批次？
| 場景 | 適合批次？ | 原因 |
|---|---|---|
| 離線批次處理 | 非常適合 | 沒有延遲要求 |
| 大量相似請求 | 適合 | 批次效率高 |
| 即時對話 | 需謹慎 | 延遲敏感 |
| 單一互動 | 不適合 | 沒有其他請求可批次 |
二、靜態批次：最簡單的做法
運作方式
靜態批次是最直覺的做法：
1. 收集 N 個請求
2. 一起送入模型
3. 等待所有請求完成
4. 收集下一批 N 個請求用程式碼表示：
def static_batching(requests: list, batch_size: int, model):
"""靜態批次處理"""
results = []
for i in range(0, len(requests), batch_size):
batch = requests[i:i + batch_size]
# 把批次內所有請求一起送入模型
batch_results = model.generate(batch)
results.extend(batch_results)
return results靜態批次的問題
問題一：短請求等長請求
假設批次內有兩個請求：
- 請求 A：生成 10 個 token 就完成
- 請求 B：生成 500 個 token 才完成
請求 A 完成後，它的 GPU 資源被閒置，但要等請求 B 完成才能開始下一批。
時間軸：
[請求 A: 10 tokens][閒置 490 tokens 的時間]
[請求 B: 500 tokens]問題二：新請求要等
新請求進來時，必須等當前批次全部完成才能加入。
時間軸：
[批次 1: 500 tokens 的時間]
新請求只能在這裡加入
[批次 2: ...]問題三：批次大小難以決定
- 批次太小：GPU 利用率低。
- 批次太大：延遲高，且可能 OOM。
- 請求長度不一：很難找到最佳批次大小。
靜態批次的適用場景
靜態批次適合：
- 離線批次處理：沒有延遲要求。
- 請求長度相近：如固定格式的摘要任務。
- 簡單實作：不需要複雜的調度邏輯。
三、動態批次：按需調整
運作方式
動態批次改善了靜態批次的問題：
1. 維護一個請求佇列
2. 當 GPU 空閒時，從佇列取出請求
3. 動態決定批次大小（根據請求數量和長度）
4. 請求完成就移出，新請求可以加入用程式碼表示：
class DynamicBatcher:
"""動態批次處理器"""
def __init__(self, model, max_batch_size: int = 32,
max_wait_time: float = 0.1):
self.model = model
self.max_batch_size = max_batch_size
self.max_wait_time = max_wait_time
self.queue = []
self.running = True
def add_request(self, request):
"""加入請求到佇列"""
self.queue.append(request)
def process(self):
"""處理佇列中的請求"""
while self.running:
if not self.queue:
time.sleep(0.001)
continue
# 取出最多 max_batch_size 個請求
batch = self.queue[:self.max_batch_size]
self.queue = self.queue[self.max_batch_size:]
# 處理批次
self._process_batch(batch)
def _process_batch(self, batch):
"""處理一個批次"""
# 可以根據請求長度動態調整
results = self.model.generate(batch)
# 回傳結果...動態批次的改進
| 面向 | 靜態批次 | 動態批次 |
|---|---|---|
| 批次大小 | 固定 | 動態 |
| 新請求加入 | 等批次完成 | 隨時可加入 |
| 短請求 | 等長請求 | 完成就移出 |
| GPU 利用率 | 中 | 較高 |
| 實作複雜度 | 低 | 中 |
動態批次仍然存在的問題
動態批次雖然改善了靜態批次的問題，但仍有局限：
- 等待批次湊滿：如果請求來得慢，要等一段時間才能湊成批次。
- 批次內的同步：批次內所有請求要一起完成，短請求仍要等長請求。
- 沒有真正解決根本問題：只是把「等整個批次」變成「等批次內最長的請求」。
四、連續批次：真正的解法
核心思想
連續批次（Continuous Batching）的核心思想是：
不再以「批次」為單位，而是以「步驟」為單位。每個步驟都可以加入新請求、移出完成的請求。
打個比方：
- 靜態批次像一台巴士，要等所有人上車才發車，到了終點才讓所有人下車。
- 動態批次像一台巴士，可以隨時上車，但還是要等所有人到終點。
- 連續批次像一台巴士，每個站都可以上下車，乘客到站就下車，新乘客隨時上車。
具體運作方式
步驟 1：請求 A、B、C、D 同時處理
步驟 2：A 完成，移出；請求 E 加入
步驟 3：B 完成，移出；請求 F 加入
步驟 4：C 完成，移出；請求 G 加入
...每個步驟只生成一個 token，所以：
- 完成的請求可以立即移出，釋放資源。
- 新請求可以立即加入，不需要等待。
連續批次的關鍵
連續批次能運作，需要兩個關鍵技術：
- PagedAttention：讓 KV Cache 可以動態分配和釋放。
- 迭代層級的調度：在每個解碼步驟之間，重新決定批次組成。
用程式碼理解
class ContinuousBatcher:
"""連續批次處理器"""
def __init__(self, model, max_batch_size: int = 32):
self.model = model
self.max_batch_size = max_batch_size
self.active_requests = [] # 當前處理中的請求
self.waiting_queue = [] # 等待加入的請求
def add_request(self, request):
"""加入請求"""
self.waiting_queue.append(request)
def step(self):
"""執行一個解碼步驟"""
# 1. 移出完成的請求
self.active_requests = [
r for r in self.active_requests
if not r.is_finished()
]
# 2. 加入新請求（如果還有空間）
while (len(self.active_requests) < self.max_batch_size
and self.waiting_queue):
new_request = self.waiting_queue.pop(0)
self.active_requests.append(new_request)
# 3. 為所有活躍請求生成一個 token
if self.active_requests:
self.model.decode_step(self.active_requests)
def run(self):
"""持續執行直到所有請求完成"""
while self.active_requests or self.waiting_queue:
self.step()連續批次 vs 靜態批次
| 面向 | 靜態批次 | 動態批次 | 連續批次 |
|---|---|---|---|
| 批次單位 | 整個批次 | 整個批次 | 單一步驟 |
| 新請求加入 | 等批次完成 | 等批次完成 | 隨時加入 |
| 完成請求移出 | 等批次完成 | 等批次完成 | 立即移出 |
| GPU 閒置 | 多 | 中 | 極少 |
| 吞吐 | 低 | 中 | 高 |
| 延遲 | 高 | 中 | 低 |
| 實作複雜度 | 低 | 中 | 高 |
五、連續批次的調度策略
連續批次的核心是調度策略：每個步驟要選擇哪些請求進入批次？
策略一：先來先服務（FIFO）
最簡單的策略：按照請求到達的順序處理。
def fifo_scheduling(waiting_queue, max_batch_size):
"""先來先服務"""
return waiting_queue[:max_batch_size]優點：公平、簡單。
缺點：長請求會阻塞短請求。
策略二：最短作業優先（SJF）
優先處理預期輸出較短的請求。
def sjf_scheduling(waiting_queue, max_batch_size):
"""最短作業優先"""
# 根據預期輸出長度排序
sorted_queue = sorted(
waiting_queue,
key=lambda r: r.expected_output_length,
)
return sorted_queue[:max_batch_size]優點：降低平均延遲。
缺點：長請求可能永遠等不到。
策略三：優先級調度
根據請求的優先級決定。
def priority_scheduling(waiting_queue, max_batch_size):
"""優先級調度"""
sorted_queue = sorted(
waiting_queue,
key=lambda r: -r.priority, # 高優先級在前
)
return sorted_queue[:max_batch_size]優點：可以區分不同等級的服務。
缺點：低優先級請求可能餓死。
策略四：公平分享（Fair Sharing）
確保每個使用者或每個請求都獲得公平的資源。
def fair_scheduling(waiting_queue, max_batch_size, active_requests):
"""公平分享"""
# 統計每個使用者的活躍請求數
user_counts = {}
for r in active_requests:
user_counts[r.user_id] = user_counts.get(r.user_id, 0) + 1
# 優先選擇活躍請求少的使用者
sorted_queue = sorted(
waiting_queue,
key=lambda r: user_counts.get(r.user_id, 0),
)
return sorted_queue[:max_batch_size]優點：避免單一使用者獨占資源。
缺點：實作較複雜。
策略五：混合策略
實務上，通常會組合多種策略：
def hybrid_scheduling(waiting_queue, max_batch_size, active_requests):
"""混合策略：先按優先級，再按等待時間"""
sorted_queue = sorted(
waiting_queue,
key=lambda r: (-r.priority, r.arrival_time),
)
return sorted_queue[:max_batch_size]調度策略的權衡
| 策略 | 吞吐 | 延遲 | 公平性 | 實作複雜度 |
|---|---|---|---|---|
| FIFO | 中 | 中 | 高 | 低 |
| SJF | 高 | 低（平均） | 低 | 中 |
| 優先級 | 中 | 中 | 低 | 中 |
| 公平分享 | 中 | 中 | 高 | 高 |
| 混合 | 高 | 低 | 中 | 高 |
六、批次大小的取捨
批次大小是連續批次中最重要的參數之一。
批次大小的影響
| 批次大小 | 吞吐 | 延遲 | GPU 利用率 | 記憶體 |
|---|---|---|---|---|
| 1 | 低 | 低 | 低 | 低 |
| 8 | 中 | 中 | 中 | 中 |
| 32 | 高 | 中高 | 高 | 中高 |
| 64 | 很高 | 高 | 很高 | 高 |
| 128 | 最高 | 很高 | 最高 | 很高 |
如何找到最佳批次大小？
def find_optimal_batch_size(model, test_requests):
"""找到最佳批次大小"""
results = []
for batch_size in [1, 2, 4, 8, 16, 32, 64, 128]:
# 測試這個批次大小
throughput, latency = benchmark(
model, test_requests, batch_size
)
results.append({
"batch_size": batch_size,
"throughput": throughput,
"latency": latency,
"efficiency": throughput / latency,
})
# 選擇效率最高的
best = max(results, key=lambda r: r["efficiency"])
return best經驗法則
| 場景 | 建議批次大小 |
|---|---|
| 延遲敏感（即時對話） | 1-8 |
| 平衡（一般服務） | 8-32 |
| 吞吐優先（批次處理） | 32-128 |
| 記憶體受限 | 根據可用記憶體計算 |
記憶體限制下的批次大小
def calculate_max_batch_size(
gpu_memory_gb: float,
model_size_gb: float,
kv_cache_per_token_mb: float,
avg_seq_length: int,
) -> int:
"""計算最大批次大小"""
# 可用於 KV Cache 的記憶體
available_for_kv = gpu_memory_gb - model_size_gb
# 每個請求的 KV Cache 大小
kv_per_request_mb = kv_cache_per_token_mb * avg_seq_length
# 最大批次大小
max_batch = int(available_for_kv * 1024 / kv_per_request_mb)
return max(1, max_batch)
# 範例
max_batch = calculate_max_batch_size(
gpu_memory_gb=80,
model_size_gb=14, # 7B FP16
kv_cache_per_token_mb=0.5,
avg_seq_length=512,
)
print(f"最大批次大小：{max_batch}") # 約 264七、實作：完整的連續批次系統
讓我們實作一個完整的連續批次系統。
import time
import asyncio
from dataclasses import dataclass, field
from enum import Enum
from collections import deque
class RequestStatus(Enum):
WAITING = "waiting"
RUNNING = "running"
FINISHED = "finished"
@dataclass
class Request:
"""請求"""
id: str
prompt: str
max_tokens: int = 100
temperature: float = 0.7
priority: int = 0
arrival_time: float = field(default_factory=time.time)
# 狀態
status: RequestStatus = RequestStatus.WAITING
generated_tokens: list = field(default_factory=list)
output_text: str = ""
def is_finished(self) -> bool:
"""是否完成"""
return (
self.status == RequestStatus.FINISHED or
len(self.generated_tokens) >= self.max_tokens
)
@property
def output_length(self) -> int:
"""已生成的長度"""
return len(self.generated_tokens)
class ContinuousBatchEngine:
"""連續批次引擎"""
def __init__(
self,
model,
tokenizer,
max_batch_size: int = 32,
max_waiting: int = 256,
scheduling_policy: str = "fifo",
):
self.model = model
self.tokenizer = tokenizer
self.max_batch_size = max_batch_size
self.max_waiting = max_waiting
self.scheduling_policy = scheduling_policy
# 請求佇列
self.waiting = deque()
self.running = []
self.finished = []
# 統計
self.stats = {
"total_requests": 0,
"total_tokens": 0,
"total_steps": 0,
"start_time": None,
}
self._running = False
def add_request(self, request: Request):
"""加入請求"""
if len(self.waiting) >= self.max_waiting:
raise RuntimeError("等待佇列已滿")
self.waiting.append(request)
self.stats["total_requests"] += 1
def _schedule(self) -> list:
"""調度：選擇要進入批次的請求"""
available_slots = self.max_batch_size - len(self.running)
if available_slots <= 0 or not self.waiting:
return []
# 根據策略排序
if self.scheduling_policy == "fifo":
selected = list(self.waiting)[:available_slots]
elif self.scheduling_policy == "sjf":
sorted_waiting = sorted(
self.waiting,
key=lambda r: r.max_tokens,
)
selected = sorted_waiting[:available_slots]
elif self.scheduling_policy == "priority":
sorted_waiting = sorted(
self.waiting,
key=lambda r: (-r.priority, r.arrival_time),
)
selected = sorted_waiting[:available_slots]
else:
selected = list(self.waiting)[:available_slots]
# 從等待佇列移除
for req in selected:
self.waiting.remove(req)
req.status = RequestStatus.RUNNING
return selected
def step(self) -> dict:
"""執行一個解碼步驟"""
# 1. 移出完成的請求
completed = []
self.running = [
r for r in self.running
if not r.is_finished()
]
for req in list(self.running):
if req.is_finished():
req.status = RequestStatus.FINISHED
completed.append(req)
self.finished.append(req)
self.running.remove(req)
# 2. 調度新請求
new_requests = self._schedule()
self.running.extend(new_requests)
# 3. 為所有活躍請求生成一個 token
if self.running:
self._decode_step()
self.stats["total_steps"] += 1
return {
"completed": [r.id for r in completed],
"running": len(self.running),
"waiting": len(self.waiting),
"new": [r.id for r in new_requests],
}
def _decode_step(self):
"""執行一步解碼"""
# 這裡是簡化的實作
# 實際應該呼叫模型的 decode 方法
for req in self.running:
# 模擬生成一個 token
token = self._generate_token(req)
req.generated_tokens.append(token)
self.stats["total_tokens"] += 1
def _generate_token(self, req: Request) -> str:
"""生成一個 token（模擬）"""
# 實際應該呼叫模型
return f"token_{len(req.generated_tokens)}"
def run_until_complete(self, timeout: float = None) -> dict:
"""執行直到所有請求完成"""
self.stats["start_time"] = time.time()
self._running = True
start = time.time()
while self._running:
# 檢查是否全部完成
if not self.running and not self.waiting:
break
# 檢查超時
if timeout and (time.time() - start) > timeout:
break
# 執行一步
self.step()
self._running = False
return self.get_stats()
def get_stats(self) -> dict:
"""取得統計"""
elapsed = time.time() - self.stats["start_time"] if self.stats["start_time"] else 0
return {
"total_requests": self.stats["total_requests"],
"total_tokens": self.stats["total_tokens"],
"total_steps": self.stats["total_steps"],
"elapsed_seconds": round(elapsed, 3),
"throughput_tokens_per_sec": (
round(self.stats["total_tokens"] / elapsed, 2)
if elapsed > 0 else 0
),
"completed": len(self.finished),
"avg_tokens_per_request": (
round(self.stats["total_tokens"] / len(self.finished), 2)
if self.finished else 0
),
}
# ========== 使用範例 ==========
def demo_continuous_batching():
"""示範連續批次"""
engine = ContinuousBatchEngine(
model=None, # 模擬
tokenizer=None,
max_batch_size=4,
scheduling_policy="fifo",
)
# 加入不同長度的請求
requests = [
Request(id=f"req_{i}", prompt=f"問題 {i}", max_tokens=10 + i * 5)
for i in range(10)
]
for req in requests:
engine.add_request(req)
# 執行
stats = engine.run_until_complete()
print("連續批次統計：")
for key, value in stats.items():
print(f" {key}: {value}")
if __name__ == "__main__":
demo_continuous_batching()與靜態批次的比較
def compare_batching_strategies():
"""比較不同批次策略"""
import matplotlib.pyplot as plt
strategies = ["static", "dynamic", "continuous"]
throughputs = []
latencies = []
for strategy in strategies:
if strategy == "static":
throughput, latency = benchmark_static()
elif strategy == "dynamic":
throughput, latency = benchmark_dynamic()
else:
throughput, latency = benchmark_continuous()
throughputs.append(throughput)
latencies.append(latency)
# 繪圖
fig, ax1 = plt.subplots(figsize=(10, 6))
x = range(len(strategies))
ax1.bar(x, throughputs, color="blue", alpha=0.7, label="吞吐")
ax1.set_ylabel("吞吐 (tokens/s)", color="blue")
ax2 = ax1.twinx()
ax2.plot(x, latencies, color="red", marker="o", label="延遲")
ax2.set_ylabel("延遲 (ms)", color="red")
ax1.set_xticks(x)
ax1.set_xticklabels(strategies)
plt.title("批次策略比較")
plt.tight_layout()
plt.savefig("batching_comparison.png")八、與 vLLM 的整合
vLLM 已經內建了連續批次，我們只需要正確設定參數。
關鍵參數
from vllm import LLM, SamplingParams
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
# 批次相關
max_num_seqs=256, # 最大同時處理的請求數
max_num_batched_tokens=4096, # 每個批次的最大 token 數
# 記憶體相關
gpu_memory_utilization=0.9, # GPU 記憶體使用率
swap_space=4, # CPU swap 空間（GB）
# 調度相關
scheduling_policy="fcfs", # 調度策略
enable_prefix_caching=True, # 啟用前綴快取
)參數的意義
| 參數 | 說明 | 建議值 |
|---|---|---|
| max_num_seqs | 最大批次大小 | 根據 GPU 記憶體調整 |
| max_num_batched_tokens | 每個批次的最大 token 數 | 4096-16384 |
| gpu_memory_utilization | GPU 記憶體使用率 | 0.85-0.95 |
| swap_space | CPU swap 空間 | 4-16 GB |
| scheduling_policy | 調度策略 | fcfs / priority |
| enable_prefix_caching | 前綴快取 | True |
監控連續批次的行為
def monitor_vllm_batching(llm):
"""監控 vLLM 的批次行為"""
# 取得引擎統計
engine = llm.llm_engine
stats = engine.statistics
return {
"running_requests": stats.num_running,
"pending_requests": stats.num_pending,
"swapped_requests": stats.num_swapped,
"gpu_cache_usage": stats.gpu_cache_usage,
"cpu_cache_usage": stats.cpu_cache_usage,
"avg_prompt_throughput": stats.avg_prompt_throughput,
"avg_generation_throughput": stats.avg_generation_throughput,
}調整批次大小
# 延遲敏感
llm_low_latency = LLM(
model="meta-llama/Llama-2-7b-hf",
max_num_seqs=8, # 小批次
)
# 吞吐優先
llm_high_throughput = LLM(
model="meta-llama/Llama-2-7b-hf",
max_num_seqs=256, # 大批次
)九、最佳實踐
1. 選擇合適的調度策略
# 一般服務：FIFO
scheduling_policy = "fifo"
# 需要區分優先級：priority
scheduling_policy = "priority"
# 平均延遲優先：sjf
scheduling_policy = "sjf"2. 動態調整批次大小
def adaptive_batch_size(current_queue_size: int, gpu_usage: float) -> int:
"""動態調整批次大小"""
if gpu_usage > 0.9:
return 16 # GPU 快滿了，降低批次
elif current_queue_size > 100:
return 64 # 佇列很長，提高批次
elif current_queue_size > 20:
return 32
else:
return 83. 設定合理的 max_num_seqs
# 根據 GPU 記憶體計算
def calculate_max_seqs(
gpu_memory_gb: float,
model_size_gb: float,
avg_seq_length: int,
kv_per_token_mb: float = 0.5,
) -> int:
"""計算 max_num_seqs"""
available = gpu_memory_gb - model_size_gb
kv_per_seq = kv_per_token_mb * avg_seq_length
return int(available * 1024 / kv_per_seq)4. 監控關鍵指標
def monitor_batching(llm):
"""監控批次指標"""
stats = llm.llm_engine.statistics
return {
"batch_size": stats.num_running,
"queue_length": stats.num_pending,
"gpu_utilization": stats.gpu_cache_usage,
"throughput": stats.avg_generation_throughput,
}5. 處理突發流量
# 用佇列吸收突發流量
class RequestQueue:
def __init__(self, max_size: int = 1000):
self.queue = deque(maxlen=max_size)
self.rejected = 0
def add(self, request):
if len(self.queue) >= self.queue.maxlen:
self.rejected += 1
return False
self.queue.append(request)
return True6. 設定超時
# 避免請求永遠等待
def process_with_timeout(engine, timeout: float = 30.0):
"""帶超時的處理"""
start = time.time()
while not engine.is_complete():
if time.time() - start > timeout:
engine.cancel_all()
break
engine.step()7. 使用前綴快取
# 多個請求有相同前綴時，大幅節省計算
llm = LLM(
model="meta-llama/Llama-2-7b-hf",
enable_prefix_caching=True,
)8. 測試不同配置
def benchmark_configs(model, configs: list):
"""測試不同配置"""
results = []
for config in configs:
llm = LLM(model=model, **config)
throughput, latency = benchmark(llm)
results.append({
"config": config,
"throughput": throughput,
"latency": latency,
})
return sorted(results, key=lambda r: -r["throughput"])十、常見的陷阱
1. 批次太大
# 錯誤的例子：批次 512，延遲很高
llm = LLM(model="...", max_num_seqs=512)
# 正確的例子：根據需求調整
llm = LLM(model="...", max_num_seqs=64)2. 批次太小
# 不好的例子：批次 1，GPU 利用率低
llm = LLM(model="...", max_num_seqs=1)
# 好的例子：至少 8-16
llm = LLM(model="...", max_num_seqs=16)3. 沒有監控
# 錯誤的例子：部署後就不管了
# 正確的例子：持續監控批次大小、佇列長度、GPU 使用率4. 忽略調度策略
# 錯誤的例子：用預設值，不考慮場景
# 正確的例子：根據場景選擇合適的策略5. 沒有超時機制
# 錯誤的例子：請求可能永遠等待
# 正確的例子：設定超時，避免請求卡死6. 記憶體不足
# 錯誤的例子：批次太大導致 OOM
# 正確的例子：根據 GPU 記憶體計算最大批次7. 忽略前綴快取
# 錯誤的例子：多個請求有相同前綴，重複計算
# 正確的例子：啟用前綴快取8. 沒有壓力測試
# 錯誤的例子：上線後才發現效能問題
# 正確的例子：先做壓力測試，找出最佳配置十一、總結：從批次到連續，從等待到流動
讓我們回顧這一篇的核心：
- 為什麼需要批次：單一請求的 GPU 利用率很低。
- 靜態批次：簡單但效率低，短請求要等長請求。
- 動態批次：改善靜態批次，但仍有同步問題。
- 連續批次：以步驟為單位，請求隨時加入、完成立即移出。
- 調度策略：FIFO、SJF、優先級、公平分享、混合。
- 批次大小的取捨：吞吐、延遲、GPU 利用率、記憶體。
- 實作：完整的連續批次系統。
- 與 vLLM 的整合：關鍵參數、監控、調整。
- 最佳實踐：選擇策略、動態調整、設定 max_num_seqs、監控、處理突發、設定超時、用前綴快取、測試配置。
- 常見陷阱：批次太大或太小、沒有監控、忽略策略、沒有超時、記憶體不足、忽略前綴快取、沒有壓力測試。
連續批次是 LLM 推理的核心優化。
它讓 GPU 幾乎不會閒置，讓吞吐提升數倍，讓延遲降到最低。
沒有它，vLLM 的 PagedAttention 也無法充分發揮。
理解了批次處理與連續批次，你就掌握了高吞吐推理的另一半。
接下來，我們要談一個更進一步的優化：推測解碼。
下一篇預告
《推測解碼：用小模型加速大模型》
我們會解釋推測解碼的原理、草稿模型與目標模型的配合、如何保證輸出品質、常見的推測解碼方法（Draft Model、Medusa、Lookahead），以及如何在實際部署中應用這項技術。