同じ関数の5つの実装がある。3つは同じ結果を返す。1つは少し違う。1つは例外を投げる。どれが正しい？

ほとんどのチームは多数決をデフォルトにする。出力が同一でエラーが明らかな場合はうまくいく。しかし、実装が微妙に異なる場合や、すべてのバリアントが異なる答えを返す場合、それは崩壊する。N-version programmingは問題の前半、つまり複数のバージョンを実行することしか解決しない。より難しい後半は、どの出力を信頼するかを決定することだ。

N-version programmingとは何か

N-version programmingは、同じ仕様の複数の独立した実装を実行し、その結果を組み合わせるfault-tolerance技法である。古典的な形態はtriple modular redundancyである。3つのシステムが投票し、多数派が勝つ。これはsafety-criticalなハードウェア、例えばavionicsに遡る。そこでは1つのバグが人命を奪う可能性があった。

同じ考え方がAI engineeringに再浮上している。LLMにコード生成を依頼する際、5つの異なるcompletionをサンプリングすることがある。レガシーのパーサーとリライト版を持っている場合、両方を並列に実行して比較することもある。ハードウェアのルーツが見える。私たちはアーキテクチャを借用したが、それを機能させるselection logicを常に借用したわけではない。

ハードウェアでは出力はビットだ。ソフトウェアでは、出力はstructured data、strings、rankings、side effectsである。Majority voteは、equalityの計算が安価で、見つかりやすいことを前提とする。ほとんどのソフトウェア問題では、どちらの前提も成り立たない。

投票の罠：なぜ「最も一般的」が「最も正しい」ではないのか

昨年、search rankingのリファクタリングでこれに遭遇した。私たちは5つのranking algorithmを持っていた。レガシーシステム、2つのmodel-basedアプローチ、そして2つのheuristic baselineだ。サンプルクエリでは、そのうち4つが異なるtop-10リストを返した。どれも完全には一致しなかった。投票できる多数派は存在しなかった。

これは通常のケースであり、エッジケースではない。異なる実装は異なるものを最適化する。1つはrecencyを重視するかもしれない。もう1つはpopularityを重視するかもしれない。3つ目は火曜日だけ現れるバグを持っているかもしれない。exact string matchで投票すれば、最も平凡な実装、つまり最も平凡で驚きのない出力を返すものに縛られてしまう。

Exact-match votingは静かに失敗することもある。2つの実装が同じ間違った答えを返すことがある。それは悪い仮定やコピーされたバグを共有しているためだ。Correlated failuresはN-version redundancyを完全に破壊する。5つのパーサーのうち3つが同じbiased datasetで訓練されていたら、それらの合意は何の意味も持たない。

Differential testingが実際にどう機能するか

より良いアプローチは、structured comparisonを伴うdifferential testingだ。「どの出力が同一か」を問うのではなく、「私が定義し測定できる基準に照らしてどの出力が最善か」を問う。

まず、自分のドメインに対するequivalence functionを定義することから始める。search rankingsでは、normalized discounted cumulative gain（NDCG）を使って比較できる。JSON parsersでは、結果のobject graphsを比較できる。string outputsでは、semantic similarityやdownstream task scoreを使える。重要なのは、equalityがバイナリスイッチではなくスペクトラムになることだ。

次に、各出力をscalarにマッピングするscoring functionを定義する。ここでdomain knowledgeが入ってくる。code-generationタスクのscoring functionは、compilation success、test pass rate、runtime performance、output lengthを組み合わせるかもしれない。正確な重みよりも、それらが明示的であるという事実の方が重要だ。

スコアを手に入れれば、selectionは単純な最適化になる。最も高いスコアの出力を選ぶ。タイブレークの場合は、例えば歴史的エラー率が最も低い実装を優先するような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は、出力が他の出力群とどれだけ類似しているかを測定することで、スコアがタイになった場合を処理する。これは、2つの出力が完全に一致しない場合にも機能するものへとmajority voteを一般化する。

Multi-criteria evaluationによる実装のスコアリング

単一のscalar scoreはシンプルだが、あまりにも多くの次元を1つの数値に押し込めると危険だ。私はtwo-tierアプローチを好む。

Tier 1はhard filterだ。compilationに失敗したり、invariantを違反したり、クラッシュしたりする出力は、即座に破棄される。部分点はなしだ。

Tier 2は生き残りに対するsoft scoreだ。これはweighted sumにできるが、重みは明示的で調整可能に保つ。selectorが一貫して速いが間違った答えを優先していると気づいた場合、アーキテクチャを書き換えずにcorrectnessの重みを上げられる。

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 failuresとruntime cost

N-version selectionはタダではない。5つのバリアントを実行することは、5倍のcompute、5倍のlatency、5倍のメンテナンス負担を意味する。実装がLLM callsの場合、そのコストはドルと秒で測定される。microservicesの場合は、queue depthとthread countで測定される。

より大きなリスクはcorrelated failureだ。5つの実装は、bug distributionからの5つの独立したサンプリングを与えてくれるわけではない。それらは言語、ライブラリ、訓練データ、人間の作成者を共有している。5つすべてが日付を解析するために同じregular expressionを使っていれば、同じmalformed inputですべて壊れる。Diversityを強制することは難しい。shared dependenciesとcopied logicを積極的に監査する必要がある。

selector自体が間違っている場合どうするかという問題もある。scoring functionにblind spotがあれば、一貫して悪い出力を選び、気づくことはない。selection distributionを監視する。1つのバリアントが決して選ばれない場合、それは無用か、score functionにbiasがあるかのどちらかだ。どちらにせよ、調査すべきだ。

まとめ：今日からリリースできるselector

始めるのにdistributed systemは必要ない。既存の関数をselectorでラップし、1つの代替実装を追加し、自分にとって重要な単一のscoring criterionを定義する。並列に実行する。出力を比較する。スコアをログに記録する。

1週間のログ記録後、バリアントが意見を異にしたケースをレビューする。その不一致は贈り物だ。仕様が曖昧な場所、score functionが間違っている場所、他の実装にはないバグが1つの実装にある場所を教えてくれる。

徐々にスケールアップする。良いselectorを持つ2つのバリアントは、majority voteを持つ5つのバリアントより優れている。

FAQ：correlated failures、stateful systems、runtime overhead

5つの実装すべてが異なる結果を返したらどうするか？

これは非自明な問題に対する期待されるケースだ。exact matchesを探すのではなく、scoring functionを使って出力をランク付けする。良いscore functionを定義できない場合、問題はselectorではない。あなたのドメインにとって「正しい」とは何かをまだ知らないということだ。

バリアント間のcorrelated failuresをどう防ぐか？

shared dependencies、copied code、common training dataを監査する。少なくとも1つのバリアントに異なる言語やframeworkを使わせる。目標はuncorrelated error distributionsだが、聞こえるよりも難しい。

N-version programmingはstateful systemsで機能するか？

side effectsを持つシステムではうまく機能しない。関数がデータベースに書き込んだり、クレジットカードに請求したりする場合、5回実行することは破壊的だ。N-version selectionはpure functions、read-only queries、またはselection後にside effectを実行できる操作で最も機能する。

5つのバリアントを実行することでどれだけのoverheadが増えるか？

Latencyは、並列に実行しない限り最も遅いバリアントで制限される。CPUとメモリは線形にスケールする。2つのバリアントから始めて、5つにコミットする前に測定する。運用コストはselection logicよりもしばしば重要だ。