你有同一個函式的五種實作。三個回傳相同的結果。一個稍有不同。一個拋出例外。哪一個才是對的?
大多數團隊預設採用多數決。當輸出完全一致且錯誤顯而易見時,這麼做沒問題。但只要你的實作在細微處出現分歧,或每個變體都回傳不同答案時,這套方法就會崩解。N-version programming 只解決了問題的前半:同時執行多個版本。更困難的後半是決定該信任哪個輸出。
什麼是 N-version programming?
N-version programming 是一種容錯技術,你會針對同一份規格執行多個獨立實作,再合併它們的結果。經典形式是 triple modular redundancy:三個系統投票,多數者勝出。這可以追溯到安全關鍵硬體——例如航空電子——在那裡單一錯誤就可能致命。
同樣的概念在 AI engineering 中重新浮現。當你要求 LLM 產生程式碼時,你可能會取樣五個不同的 completion。當你同時擁有 legacy parser 和重寫版本時,你可能會平行執行兩者並比較。硬體的根源清晰可見。我們借用了架構,卻不總是連同讓它運作的 selection logic 一起借用。
在硬體中,輸出是位元。在軟體中,輸出是結構化資料、字串、排序結果或side effect。多數決假設比較相等是便宜的,且相等是常見的。對大多數軟體問題來說,這兩個假設都不成立。
投票陷阱:為什麼「最常見」不等於「最正確」
去年我在一次搜尋排序重構中遇到這個問題。我們有五種排序演算法:legacy system、兩種 model-based 方法,以及兩種 heuristic baseline。在一份範例查詢上,四個回傳了不同的前 10 名清單。沒有任何兩個完全相符。根本沒有多數可以投。
這是正常情況,不是邊界案例。不同的實作針對不同目標進行最佳化。一個可能偏好新鮮度。另一個可能加權熱門度。第三個可能有個只在星期二出現的 bug。如果你用精確字串匹配來投票,你會把自己綁在最平庸的實作上——那個回傳最無聊、最不出奇輸出的版本。
精確匹配投票也會默默失效。兩個實作可能因為共享錯誤假設或複製的 bug 而回傳相同的錯誤答案。correlated failure 會完全破壞 N-version redundancy。如果你的五個 parser 中有三個是在同一個有偏誤的資料集上訓練的,它們的共識毫無意義。
Differential testing 實際上如何運作
更好的方法是 differential testing 搭配結構化比較。不要問「哪些輸出完全相同」,而是問「根據我能定義且衡量的準則,哪個輸出最好」。
首先為你的領域定義一個 equivalence function。對於搜尋排序,你可能使用 normalized discounted cumulative gain(NDCG)來比較。對於 JSON parser,你可能比較產生的物件圖。對於字串輸出,你可能使用語義相似度或下游任務分數。關鍵在於,相等變成一個光譜,而非二元開關。
接著,定義一個 scoring function,將每個輸出映射到一個純量。這就是領域知識進場的地方。對於程式碼生成任務,scoring function 可能結合編譯成功與否、測試通過率、執行時效能和輸出長度。確切的權重不如「它們是明確的」這件事來得重要。
有了分數之後,選擇就變成簡單的最佳化。挑選分數最高的輸出。如果平手,使用 fallback heuristic,例如偏好歷史錯誤率最低的實作。
這裡有一個你可以調整的 selector。它執行 N 個變體,為每個輸出打分,並回傳最佳結果以及診斷用的 metadata:
from dataclasses import dataclass
from typing import Callable, List, Optional, TypeVar
import statistics
T = TypeVar("T")
@dataclass
class VariantResult:
variant_id: str
output: Optional[T]
error: Optional[Exception]
score: float = 0.0
class NVersionSelector:
def __init__(
self,
score_fn: Callable[[T], float],
equivalence_fn: Optional[Callable[[T, T], float]] = None,
tie_breaker: Optional[Callable[[List[VariantResult]], VariantResult]] = None,
):
self.score_fn = score_fn
self.equivalence_fn = equivalence_fn or (lambda a, b: 1.0 if a == b else 0.0)
self.tie_breaker = tie_breaker
def select(self, variants: List[Callable[..., T]], *args, **kwargs) -> VariantResult:
results: List[VariantResult] = []
for variant in variants:
try:
output = variant(*args, **kwargs)
results.append(VariantResult(
variant_id=variant.__name__,
output=output,
error=None,
))
except Exception as exc:
results.append(VariantResult(
variant_id=variant.__name__,
output=None,
error=exc,
))
# Score only successful runs
for r in results:
if r.output is not None:
r.score = self.score_fn(r.output)
# Filter to successful, scored results
valid = [r for r in results if r.error is None and r.output is not None]
if not valid:
raise RuntimeError("All variants failed")
best_score = max(r.score for r in valid)
candidates = [r for r in valid if r.score == best_score]
if len(candidates) == 1:
return candidates[0]
if self.tie_breaker:
return self.tie_breaker(candidates)
return self._default_tie_break(candidates, valid)
def _default_tie_break(
self, candidates: List[VariantResult], all_valid: List[VariantResult]
) -> VariantResult:
# Prefer the candidate whose output is most similar to other outputs
def consensus_score(candidate: VariantResult) -> float:
similarities = [
self.equivalence_fn(candidate.output, other.output)
for other in all_valid
if other.variant_id != candidate.variant_id
]
return statistics.mean(similarities) if similarities else 0.0
return max(candidates, key=consensus_score)
score_fn 是你為問題編碼「最好」意味什麼的地方。equivalence_fn 處理分數平手的情況,方法是衡量某個輸出與其他輸出的相似程度。這將多數決一般化為即使在沒有兩個輸出完全匹配時也能運作的方法。
使用多準則評估為實作打分
單一純量分數很乾淨,但如果你把太多維度壓縮成一個數字,就會變得危險。我偏好兩層式作法。
第一層是 hard filter。編譯失敗、違反不變條件或當機的輸出會被立刻丟棄。沒有 partial credit。
第二層是針對存活者的 soft score。這可以是加權總和,但要讓權重明確且可調整。如果你發現 selector 總是偏好快但錯的答案,你可以調高正確性權重,而不需要重寫整個架構。
def score_generated_code(output: str) -> float:
if not compiles(output):
return -1.0 # Hard filter
tests_passed = run_test_suite(output)
execution_time = benchmark(output)
line_count = len(output.splitlines())
# Weighted sum on survivors only
return (
0.6 * tests_passed
+ 0.3 * (1.0 / (1.0 + execution_time))
+ 0.1 * (1.0 / (1.0 + line_count))
)
上面的權重是任意設定的。針對 held-out validation set 進行調校,並在需求改變時更新它們。不要把它們埋藏在 class 內部,讓它們變得看不見。
何時會失效:correlated failure 與執行時成本
N-version selection 不是免費的。執行五個變體意味著五倍的計算量、五倍的延遲,以及五倍的維護負擔。如果你的實作是 LLM 呼叫,成本以美元和秒數衡量。如果是 microservices,則以queue深度和執行緒數衡量。
更大的風險是 correlated failure。五個實作並不代表你從 bug 分佈中獨立抽取五次。它們共享語言、函式庫、訓練資料和人類作者。如果五個實作都使用同一個正規表示式來解析日期,它們會在同一個格式錯誤的輸入上全部崩潰。多樣性很難強制執行。它需要主動稽核共享的依賴和複製的邏輯。
還有一個問題:當 selector 本身是錯的時候該怎麼辦。如果你的 scoring function 有盲點,你會持續選到壞的輸出,而且永遠不會發現。監控 selection 的分佈。如果某個變體從未被選中,它不是沒用,就是你的 score function 有偏誤。無論哪一種,你都應該調查。
整合起來:今天就能上線的 selector
你不需要分散式系統才能開始。把你現有的函式包進 selector,加入一個替代實作,並定義一個對你來說重要的單一評分準則。平行執行它們。比較輸出。記錄分數。
記錄一週後,回顧變體出現分歧的案例。這種分歧是一份禮物。它告訴你規格在哪裡含糊不清、score function 在哪裡有錯,或者哪個實作有別人沒有的 bug。
逐步擴大。兩個變體搭配好的 selector,勝過五個變體搭配多數決。
常見問題:correlated failure、具狀態系統與執行時開銷
如果五個實作都回傳不同結果怎麼辦?
這是非平凡問題的預期情況。使用 scoring function 來為輸出排名,而不是尋找完全匹配。如果你無法定義好的 score function,問題不在 selector。而是你還不知道對你的領域來說「正確」意味著什麼。
如何防止變體之間的 correlated failures?
稽核共享的依賴、複製的程式碼和共同的訓練資料。強制至少有一個變體使用不同的語言或框架。目標是讓錯誤分佈不相關,但這比聽起來更難。
N-version programming 適用於具狀態系統嗎?
它對有side effect的系統效果很差。如果你的函式會寫入資料庫或刷信用卡,執行五次是具破壞性的。N-version selection 最適合 pure function、唯讀查詢,或你可以在選擇後才執行side effect的操作。
執行五個變體會增加多少開銷?
除非你平行執行,否則延遲由最慢的變體決定。CPU 和記憶體線性增長。先從兩個變體開始,在承諾使用五個之前先測量。操作成本往往比 selection logic 更重要。