Self-Consistency 技巧：多次采样提升推理质量

# 单次 CoT：一条路径
问题 → [推理路径 A] → 答案 A

# Self-Consistency：多条路径 + 投票
问题 → [推理路径 A] → 答案 A ─┐
     → [推理路径 B] → 答案 B ─┼→ 多数投票 → 最终答案
     → [推理路径 C] → 答案 A ─┘
     → [推理路径 D] → 答案 A ─┘

import openai

def self_consistency_query(prompt, n=10, temperature=0.7, top_p=0.95):
    """Self-Consistency 多次采样"""
    responses = []
    for i in range(n):
        resp = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            top_p=top_p,
            max_tokens=1024,
        )
        responses.append(resp.choices[0].message.content)
    return responses

from collections import Counter

def majority_vote(answers):
    """简单多数投票"""
    counter = Counter(answers)
    return counter.most_common(1)[0][0]

import re
from collections import defaultdict

def weighted_vote(responses_and_answers):
    """
    加权投票：推理路径越短，权重越高（奥卡姆剃刀）
    responses_and_answers: [(reasoning_text, answer), ...]
    """
    scores = defaultdict(float)
    for reasoning, answer in responses_and_answers:
        # 路径越短权重越高
        length = len(reasoning.split())
        weight = 1.0 / (1.0 + length / 100)
        scores[answer] += weight
    return max(scores, key=scores.get)

def extract_answer(response_text):
    """从推理文本中提取最终答案"""
    # 匹配 "答案是 X" 或 "The answer is X"
    match = re.search(r'(?:答案|answer)\s*[:：]??\s*([^\n。]+)', 
                      response_text, re.IGNORECASE)
    return match.group(1).strip() if match else response_text.strip()

import math

def logprob_weighted_vote(paths_with_logprobs):
    """
    paths_with_logprobs: [(answer, avg_logprob), ...]
    avg_logprob: 该路径 token 的平均 log probability
    """
    scores = defaultdict(float)
    for answer, logprob in paths_with_logprobs:
        # 转换为概率权重
        scores[answer] += math.exp(logprob)
    return max(scores, key=scores.get)

import openai
import re
from collections import Counter

class SelfConsistency:
    def __init__(self, model="gpt-4", n_samples=10, 
                 temperature=0.7, top_p=0.95):
        self.model = model
        self.n_samples = n_samples
        self.temperature = temperature
        self.top_p = top_p
    
    def _build_prompt(self, question):
        return f"""请逐步推理以下问题，并在最后给出答案。

问题：{question}

请按以下格式回答：
推理过程：...
答案：[最终答案]

解题步骤："""
    
    def _sample(self, prompt):
        resp = openai.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=self.temperature,
            top_p=self.top_p,
            max_tokens=1024,
        )
        return resp.choices[0].message.content
    
    def _extract_answer(self, text):
        match = re.search(r'答案[：:]\s*(.+?)(?:\n|$)', text)
        return match.group(1).strip() if match else text.strip()
    
    def solve(self, question):
        prompt = self._build_prompt(question)
        
        # 并行采样（实际生产环境用 asyncio）
        responses = [self._sample(prompt) for _ in range(self.n_samples)]
        
        # 提取答案
        answers = [self._extract_answer(r) for r in responses]
        
        # 多数投票
        counter = Counter(answers)
        best_answer, vote_count = counter.most_common(1)[0]
        confidence = vote_count / len(answers)
        
        return {
            "answer": best_answer,
            "confidence": confidence,
            "all_answers": answers,
            "responses": responses,
        }


# 使用示例
sc = SelfConsistency(n_samples=10)
result = sc.solve("一个水池有两个进水管A和B，A单独20小时注满，B单独30小时注满。同时开3小时后关闭A，B还需几小时注满？")
print(f"答案: {result['answer']}")
print(f"置信度: {result['confidence']:.0%}")

def adaptive_self_consistency(question, solver, 
                                initial_n=3, max_n=10, 
                                confidence_threshold=0.8):
    """自适应采样：达到置信阈值即停止"""
    responses = []
    for i in range(max_n):
        resp = solver._sample(solver._build_prompt(question))
        responses.append(resp)
        
        if i + 1 >= initial_n:
            answers = [solver._extract_answer(r) for r in responses]
            counter = Counter(answers)
            top_answer, votes = counter.most_common(1)[0]
            confidence = votes / len(answers)
            
            if confidence >= confidence_threshold:
                return {
                    "answer": top_answer,
                    "confidence": confidence,
                    "samples_used": i + 1,
                }
    
    return {
        "answer": top_answer,
        "confidence": confidence,
        "samples_used": max_n,
    }