你有同一个函数的五种实现。三种返回相同结果。一种略有不同。一种抛出异常。哪个才是对的？

大多数团队默认采用多数投票。当输出完全一致、错误显而易见时，这招确实管用。但一旦各实现在细微之处产生分歧，或者每个变体都给出不同答案时，这套机制就会崩解。N-version programming 只解决了问题的前半段：运行多个版本。更困难的后半段是判断该信任哪个输出。

什么是 N-version programming？

N-version programming 是一种容错技术，即运行同一规范的多个独立实现，然后综合它们的结果。经典形式是 triple modular redundancy：三个系统投票，多数获胜。它最早源于安全关键型硬件——比如航空电子——在那里单个 bug 就可能致命。

同样的思路在 AI 工程中又 resurgence 了。当你让 LLM 生成代码时，你可能会采样五个不同的 completion。当你有一个 legacy parser 和一个 rewrite 时，你可能会并行运行两者并比较。其硬件根源显而易见。我们借用了架构，却不一定总是借用了让它有效运作的 selection logic。

在硬件中，输出是比特。在软件中，输出是结构化数据、字符串、排序结果或副作用。多数投票假设「判断是否相等」计算开销低且常见。对大多数软件问题来说，这两个假设都不成立。

投票陷阱：为什么「最常见的」不等于「最正确的」

去年我在一次搜索排序重构中遇到了这个问题。我们有五种排序算法：legacy 系统、两种基于模型的方法，以及两种 heuristic baseline。在一个样本查询上，其中四种返回了不同的 top-10 列表。没有一种完全匹配。根本不存在可以投票的多数派。

这是常态，而非边缘情况。不同的实现针对不同的目标进行优化。一种可能偏好 recency。另一种可能加权 popularity。第三种可能有个只在周二才会触发的 bug。如果你按 exact string match 来投票，你就会把自己绑定到最平庸的实现上——也就是那个返回最平淡、最不出人意料输出的实现。

Exact-match voting 还会静默失效。两种实现可能因共享一个错误假设或复制的 bug 而返回同样的错误答案。Correlated failures 会彻底破坏 N-version redundancy。如果你的五个 parser 中有三个是在同一个有偏数据集上训练的，它们的共识毫无意义。

Differential testing 实际如何运作

更好的方法是结合 structured comparison 的 differential testing。不问「哪些输出完全相同」，而是问「根据我可以定义和衡量的标准，哪个输出最好」。

首先，为你的领域定义一个 equivalence function。对于搜索排序，你可能会用 normalized discounted cumulative gain（NDCG）来比较。对于 JSON parser，你可能会比较生成的对象图。对于字符串输出，你可能会用 semantic similarity 或下游任务得分。关键在于，相等变成一个光谱，而不是一个二元开关。

接下来，定义一个 scoring function，将每个输出映射为一个标量。这是领域知识介入的地方。对于代码生成任务，scoring function 可能综合编译成功、测试通过率、运行时性能和输出长度。具体权重并不重要，重要的是它们被显式定义出来。

有了分数，选择就变成了一个简单的优化问题：挑出得分最高的输出。如果平局，使用 fallback heuristic，比如优先选择历史错误率最低的实现。

下面是一个你可以改用的 selector。它运行 N 个变体，为每个输出打分，并返回最佳结果及诊断元数据：

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 处理分数平局的情况，通过衡量某个输出与其他输出的相似度。这将 majority vote 泛化为一种即使在没有任何两个输出完全匹配时也能工作的机制。

用多准则评估为实现打分

单一的标量分数很简洁，但如果你把太多维度压缩成一个数字，就会很危险。我更倾向于双层方法。

第一层是 hard filter。编译失败、违反不变量或崩溃的输出会被立即丢弃。没有 partial credit。

第二层是给幸存者的 soft score。这可以是一个 weighted sum，但权重要保持显式且可调。如果你发现 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))
    )

上面的权重是 arbitrary 的。在 held-out validation set 上调参，并在需求变化时更新它们。不要把它们埋在一个类里，让它们变得不可见。

何时会失效：correlated failures 与运行时开销

N-version selection 并非无成本。运行五个变体意味着五倍的计算、五倍的延迟和五倍的维护负担。如果你的实现是 LLM 调用，成本以美元和秒计算。如果它们是 microservices，成本以队列深度和线程数计算。

更大的风险是 correlated failure。五个实现并不代表从 bug 分布中独立抽取五次。它们共享语言、库、训练数据和人类作者。如果五个实现都用同一个正则表达式来解析日期，它们都会在同一 malformed input 上崩溃。Diversity 很难强制执行。它需要主动审计共享依赖和复制的逻辑。

还有一个问题是 selector 本身出错该怎么办。如果你的 scoring function 存在盲点，你会持续选出 bad outputs 却毫无察觉。监控 selection distribution。如果某个变体从未被选中，它要么没用，要么你的 score function 有偏。无论哪种情况，你都应该调查。

整合起来：今天就能上线的 selector

你不需要一个分布式系统才能起步。把你现有的函数包装进一个 selector，添加一个替代实现，并定义一个对你真正重要的单一评分标准。并行运行它们。比较输出。记录分数。

记录一周后，回顾那些变体出现分歧的 case。这种分歧是一份礼物。它告诉你规范哪里模糊、score function 哪里出错、或者哪个实现有 bug 而其他实现没有。

逐步扩展。两个变体配一个好的 selector，胜过五个变体配 majority vote。

FAQ：correlated failures、stateful systems 与运行时开销

如果五种实现返回五种不同结果怎么办？

对于非平凡问题，这是预期中的情况。使用 scoring function 对输出进行排序，而不是寻找 exact match。如果你无法定义一个好的 score function，问题不在 selector，而在于你尚未弄清楚「正确」对你的领域意味着什么。

如何防止跨变体的 correlated failures？

审计共享依赖、复制的代码和共同的训练数据。强制至少一个变体使用不同的语言或框架。目标是 uncorrelated error distributions，这比你想象的要难。

N-version programming 适用于 stateful systems 吗？

对于带副作用的系统，它表现很差。如果你的函数写入数据库或刷信用卡，运行五次是具有破坏性的。N-version selection 最适合 pure functions、只读查询，或者你可以在 selection 之后再执行副作用的操作。

运行五个变体会带来多少开销？

除非你并行运行，否则延迟由最慢的变体决定。CPU 和内存线性增长。从两个变体开始，在承诺使用五个之前先测量。运营成本往往比 selection logic 更重要。