CoT 없이도 정확도↑ - Self-Consistency 구현법

import json
from dataclasses import dataclass
from typing import Any, Dict, List, Tuple

# 예시: OpenAI 호환 API를 가정한 간단 래퍼(의사 코드)
# 실제론 openai / azure / vLLM 등 환경에 맞게 교체하세요.

def call_llm(prompt: str, temperature: float, seed: int | None = None) -> Dict[str, Any]:
    """모델 호출 결과를 dict로 반환한다고 가정."""
    raise NotImplementedError


ANSWER_SCHEMA = {
    "type": "object",
    "properties": {
        "answer": {"type": "string"},
        "confidence": {"type": "number"}
    },
    "required": ["answer"],
    "additionalProperties": False
}


def build_prompt(question: str) -> str:
    # 부등호는 MDX 이슈가 있어 코드로 감쌌습니다.
    return (
        "You are a careful assistant. "
        "Return ONLY valid JSON matching this schema: "
        f"{json.dumps(ANSWER_SCHEMA)}\n"
        "Do not include any explanation.\n"
        f"Question: {question}\n"
    )

import re
from collections import Counter


def normalize_answer(text: str) -> str:
    # 케이스/공백/구두점/통화기호 등 간단 정규화
    t = text.strip().lower()
    t = re.sub(r"\s+", " ", t)
    t = re.sub(r"[\.,!\?\"']", "", t)
    return t


def self_consistency_vote(question: str, n: int = 7, temperature: float = 0.8) -> Tuple[str, List[str]]:
    prompt = build_prompt(question)

    candidates: List[str] = []
    raw: List[str] = []

    for i in range(n):
        resp = call_llm(prompt, temperature=temperature, seed=i)
        # resp["content"]가 JSON 문자열이라고 가정
        obj = json.loads(resp["content"])
        ans = obj["answer"]
        raw.append(ans)
        candidates.append(normalize_answer(ans))

    counts = Counter(candidates)
    best_norm, _ = counts.most_common(1)[0]

    # 정규화된 best_norm에 대응하는 원문 답 하나를 대표로 선택
    for a in raw:
        if normalize_answer(a) == best_norm:
            return a, raw

    return raw[0], raw

from collections import defaultdict


def self_consistency_early_stop(
    question: str,
    max_n: int = 9,
    min_n: int = 3,
    temperature: float = 0.8,
    margin: int = 2,
) -> Tuple[str, List[str]]:
    prompt = build_prompt(question)

    counts = defaultdict(int)
    raw: List[str] = []

    for i in range(max_n):
        resp = call_llm(prompt, temperature=temperature, seed=i)
        obj = json.loads(resp["content"])
        ans = obj["answer"]
        raw.append(ans)

        norm = normalize_answer(ans)
        counts[norm] += 1

        # 최소 샘플 수 충족 후 조기 종료 판단
        if i + 1 >= min_n:
            top2 = sorted(counts.items(), key=lambda x: x[1], reverse=True)[:2]
            top1_key, top1_cnt = top2[0]
            top2_cnt = top2[1][1] if len(top2) > 1 else 0

            # margin 이상 벌어지면 종료
            if top1_cnt - top2_cnt >= margin:
                # 대표 원문 반환
                for a in raw:
                    if normalize_answer(a) == top1_key:
                        return a, raw

    # 끝까지 가면 최다 득표
    best_norm = max(counts.items(), key=lambda x: x[1])[0]
    for a in raw:
        if normalize_answer(a) == best_norm:
            return a, raw
    return raw[0], raw

def validate_answer(question: str, answer: str) -> bool:
    # 도메인별 룰을 넣으세요. 예: 숫자만 허용, 특정 포맷 강제 등
    return len(answer.strip()) > 0


def answer_with_fallback(question: str) -> str:
    prompt = build_prompt(question)

    first = json.loads(call_llm(prompt, temperature=0.2, seed=0)["content"])["answer"]
    if validate_answer(question, first):
        return first

    # 애매할 때만 더 비싼 전략 수행
    best, _raws = self_consistency_early_stop(
        question,
        max_n=9,
        min_n=3,
        temperature=0.9,
        margin=2,
    )
    return best

from datetime import datetime


def normalize_choice(text: str, mapping: dict[str, str]) -> str:
    t = normalize_answer(text)
    return mapping.get(t, t)

import decimal


def parse_price_krw(text: str) -> int | None:
    t = text.strip().lower().replace(",", "")
    t = t.replace("원", "").replace("krw", "").strip()
    if not t:
        return None
    try:
        # 정수 원 단위로 가정
        return int(decimal.Decimal(t))
    except decimal.InvalidOperation:
        return None


def self_consistency_price(question: str, n: int = 7) -> int | None:
    prompt = build_prompt(question)
    prices: list[int] = []

    for i in range(n):
        obj = json.loads(call_llm(prompt, temperature=0.9, seed=i)["content"])
        p = parse_price_krw(obj["answer"])
        if p is not None:
            prices.append(p)

    if not prices:
        return None

    # 최빈값
    return Counter(prices).most_common(1)[0][0]