Self-consistency: sample several paths, take the majority

"""
Self-consistency: sample several reasoning paths, take the majority

Ask the same question N times at a temperature high enough to get genuinely
different reasoning, then take the answer the paths agree on. Errors tend to
be idiosyncratic and scatter; correct answers tend to converge.

    Question -> N independent paths -> normalise -> majority answer

The vote is plain Python, deliberately. Asking a model to pick the best of N
answers reintroduces exactly the single-point-of-failure the sampling was
meant to remove, and it lets a confidently-worded wrong answer win.

The agreement count is the useful output, not a decoration. Five of five is a
different situation from three of five, and a caller that only reads the
answer has thrown away the reason to have run this at all.

Run: python self_consistency.py
"""

from collections import Counter

from dotenv import load_dotenv
from openai import OpenAI
from pydantic import BaseModel, Field

load_dotenv()

client = OpenAI()

MODEL = "gpt-5.4-mini"
SAMPLES = 5
#: Enough spread to get different reasoning. At 0 every path is the same path
#: and the vote is theatre.
TEMPERATURE = 1.0


class Attempt(BaseModel):
    reasoning: str = Field(description="Work the problem through, step by step.")
    answer: str = Field(
        description="The final answer alone. No units, no restatement, no prose."
    )


def attempt(question: str) -> Attempt:
    completion = client.beta.chat.completions.parse(
        model=MODEL,
        temperature=TEMPERATURE,
        messages=[
            {
                "role": "system",
                "content": (
                    "Work the problem through and give the final answer. Put "
                    "only the answer itself in `answer`, nothing else."
                ),
            },
            {"role": "user", "content": question},
        ],
        response_format=Attempt,
    )
    return completion.choices[0].message.parsed


def normalise(answer: str) -> str:
    """Group answers that are the same fact written differently.

    Without this the vote splits on formatting and everything ties at one
    each, which looks like disagreement when the paths actually agreed.
    """
    return answer.strip().lower().rstrip(".").replace(",", "").replace("$", "")


def run(question: str) -> tuple[str, int, int]:
    attempts: list[Attempt] = []
    for index in range(1, SAMPLES + 1):
        print(f"  [{index}/{SAMPLES}] sampling...")
        current = attempt(question)
        print(f"      answer: {current.answer}")
        attempts.append(current)

    counts = Counter(normalise(a.answer) for a in attempts)
    winner, votes = counts.most_common(1)[0]

    # Surface the spread, not just the winner. Agreement IS the signal.
    print(f"  distribution: {dict(counts)}")
    print(f"  majority    : {winner} ({votes}/{SAMPLES})")
    if votes == 1:
        print("  WARNING: every path disagreed. Treat this as no answer.")
    elif votes <= SAMPLES // 2:
        print("  WARNING: the majority is thin. Escalate rather than rely on it.")

    return winner, votes, SAMPLES


if __name__ == "__main__":
    question = (
        "A team of 4 engineers takes 6 days to migrate 120 services. Two more "
        "engineers join at the same rate. How many days for the next 120?"
    )
    print(f"\n> {question}")
    run(question)