详解下面代码背后的逻辑， 隐式的cot是如何实现的； 蕴含了什么数学原理：import torch
import torch.nn as nn
import torch.optim as optim
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch.nn.functional as F
import pandas as pd
import argparse
import yaml
import os
from tqdm import tqdm

# Ensure reproducibility
torch.manual_seed(42)


# Load configuration from a YAML file
def load_config(config_file=None):
    if config_file and os.path.exists(config_file):
        with open(config_file, 'r') as f:
            config = yaml.safe_load(f)
    else:
        # Default configuration
        config = {
            "model_name": "qwen/qwen-7b",  # Updated to Qwen model
            "learning_rate_stage1": 1e-5,
            "learning_rate_stage2": 1e-5,
            "epochs_stage_1": 2,
            "epochs_stage_2": 3,
            "beta_kl_stage1": 0.1,
            "beta_kl_stage2": 0.05,
            "alpha_reward_bonus": 2.0,
            "data_file": "SCoRe_dataset.csv",
            "max_length": 512,  # Maximum token length
            "batch_size": 1,  # For simplicity, keep it 1. Adjust as needed
            "device": "cuda" if torch.cuda.is_available() else "cpu"
        }
    return config


# Reward function for self-correction
def reward_function(original_answer, corrected_answer, correct_answer):
    if corrected_answer.strip() == correct_answer.strip():  # Fully correct answer
        return 1.0
    elif corrected_answer.strip() == original_answer.strip():  # No improvement from original answer
        return -1.0
    else:
        return 0.5  # Partial improvement, better than original but still incorrect


# Apply chat template to format the conversation
def apply_chat_template(conversation):
    prompt = ""
    for turn in conversation:
        role = turn['role']
        content = turn['content']
        if role == 'user':
            prompt += f"User: {content}\n"
        elif role == 'assistant':
            prompt += f"Assistant: {content}\n"
    return prompt


# Stage I: Training the Model to Generate Correct Answers Directly
def stage_one_initialization(train_model, reference_model, tokenizer, data, config):
    optimizer = optim.AdamW(train_model.parameters(), lr=config["learning_rate_stage1"])
    train_model.train()
    reference_model.eval()  # Ensure reference model is in eval mode

    for epoch in range(config["epochs_stage_1"]):
        total_loss = 0.0
        for example in tqdm(data, desc=f"Stage I - Epoch {epoch + 1}/{config['epochs_stage_1']}"):
            # Format input using updated chat format for implicit reasoning
            conversation = stage1_chat_format(example)

            # Apply chat template to get the prompt
            conversation_text = apply_chat_template(conversation)

            # Tokenize inputs
            inputs = tokenizer(conversation_text, return_tensors="pt",
                               padding=True, truncation=True, max_length=config["max_length"])
            inputs = {k: v.to(config["device"]) for k, v in inputs.items()}

            # Generate target_probs from the reference model
            with torch.no_grad():
                ref_outputs = reference_model(**inputs)
                logits_ref = ref_outputs.logits
                target_probs = F.softmax(logits_ref, dim=-1)

            # Forward pass with the train model
            outputs = train_model(**inputs, labels=inputs['input_ids'])

            # Cross-entropy loss
            cross_entropy_loss = outputs.loss

            # Get log_probs from the train model
            logits_train = outputs.logits
            log_probs_train = F.log_softmax(logits_train, dim=-1)

            # Compute KL divergence: KL(target_probs || q_train)
            kl_loss = F.kl_div(log_probs_train, target_probs, reduction='batchmean')

            # Total loss combines cross-entropy and scaled KL divergence
            total_loss_value = cross_entropy_loss + config["beta_kl_stage1"] * kl_loss

            # Backward pass and optimization
            optimizer.zero_grad()
            total_loss_value.backward()
            optimizer.step()

            total_loss += total_loss_value.item()
        avg_loss = total_loss / len(data)
        print(f"Stage I - Epoch {epoch + 1}/{config['epochs_stage_1']}, Avg Loss: {avg_loss:.4f}")


# Stage II: Encouraging Implicit Self-Correction without Explicit Instructions
def stage_two_training(train_model, tokenizer, data, config):
    optimizer = optim.AdamW(train_model.parameters(), lr=config["learning_rate_stage2"])
    train_model.train()

    for epoch in range(config["epochs_stage_2"]):
        total_loss = 0.0
        for example in tqdm(data, desc=f"Stage II - Epoch {epoch + 1}/{config['epochs_stage_2']}"):
            # First attempt: Generate the initial answer
            conversation1 = stage2_chat_format_first_attempt(example)
            conversation_text1 = apply_chat_template(conversation1)
            inputs1 = tokenizer(conversation_text1, return_tensors="pt",
                                padding=True, truncation=True, max_length=config["max_length"])
            inputs1 = {k: v.to(config["device"]) for k, v in inputs1.items()}

            # Generate output for the first attempt
            with torch.no_grad():
                outputs1 = train_model.generate(**inputs1, max_new_tokens=150, do_sample=False, temperature=0.0)
                generated_text1 = tokenizer.decode(outputs1[0], skip_special_tokens=True).strip()

            # Second attempt: Present the same question again
            conversation2 = stage2_chat_format_second_attempt(example, generated_text1)
            conversation_text2 = apply_chat_template(conversation2)
            inputs2 = tokenizer(conversation_text2, return_tensors="pt",
                                padding=True, truncation=True, max_length=config["max_length"])
            inputs2 = {k: v.to(config["device"]) for k, v in inputs2.items()}

            # Forward pass with labels for loss calculation
            outputs2 = train_model(**inputs2, labels=inputs2['input_ids'])

            # Cross-entropy loss
            loss = outputs2.loss

            # Decode the second attempt
            generated_ids2 = outputs2.logits.argmax(dim=-1)
            generated_text2 = tokenizer.decode(generated_ids2[0], skip_special_tokens=True).strip()

            # Compute reward based on self-correction
            reward = reward_function(generated_text1, generated_text2, example.get('correct_answer', ''))

            # Apply reward shaping: Encourage implicit self-correction
            shaped_loss = loss * (config["alpha_reward_bonus"] * reward)

            # Backward pass and optimization
            optimizer.zero_grad()
            shaped_loss.backward()
            optimizer.step()

            total_loss += shaped_loss.item()
        avg_loss = total_loss / len(data)
        print(f"Stage II - Epoch {epoch + 1}/{config['epochs_stage_2']}, Avg Shaped Loss: {avg_loss:.4f}")


# Format for Stage I
def stage1_chat_format(example):
    return [
        {"role": "user", "content": example['question']},
        {"role": "assistant", "content": f"{example.get('correct_answer', '')}"},
    ]


# Format for Stage II (First Attempt)
def stage2_chat_format_first_attempt(example):
    return [
        {"role": "user", "content": example['question']},
    ]


# Format for Stage II (Second Attempt without explicit correction instruction)
def stage2_chat_format_second_attempt(example, first_attempt):
    return [
        {"role": "user", "content": example['question']},
        {"role": "assistant", "content": f"{first_attempt}"},
        {"role": "user", "content": example['question']},  # Presenting the same question again
    ]


# Evaluation functions
def evaluate_chat_format_first_attempt(example):
    return [
        {"role": "user", "content": example['question']},
    ]


def evaluate_chat_format_second_attempt(example, first_attempt):
    return [
        {"role": "user", "content": example['question']},
        {"role": "assistant", "content": f"{first_attempt}"},
        {"role": "user", "content": example['question']},  # Presenting the question again
    ]


# Evaluation function
def evaluate_model(model, tokenizer, data, config):
    model.eval()
    correct_t1 = 0
    correct_t2 = 0
    total = 0
    generated_answers_t1 = []
    generated_answers_t2 = []
    reference_answers = []

    incorrect_to_correct = 0
    correct_to_incorrect = 0

    with torch.no_grad():
        for example in tqdm(data, desc="Evaluating"):
            # First attempt
            conversation1 = evaluate_chat_format_first_attempt(example)
            conversation_text1 = apply_chat_template(conversation1)
            inputs1 = tokenizer(conversation_text1, return_tensors="pt",
                                padding=True, truncation=True, max_length=config["max_length"])
            inputs1 = {k: v.to(config["device"]) for k, v in inputs1.items()}

            outputs1 = model.generate(**inputs1, max_new_tokens=150, do_sample=False, temperature=0.0)
            generated_text1 = tokenizer.decode(outputs1[0], skip_special_tokens=True).strip()

            reference = example.get('correct_answer', '').strip()
            generated_answers_t1.append(generated_text1)
            reference_answers.append(reference)
            first_correct = generated_text1 == reference
            if first_correct:
                correct_t1 += 1

            # Second attempt
            conversation2 = evaluate_chat_format_second_attempt(example, generated_text1)
            conversation_text2 = apply_chat_template(conversation2)
            inputs2 = tokenizer(conversation_text2, return_tensors="pt",
                                padding=True, truncation=True, max_length=config["max_length"])
            inputs2 = {k: v.to(config["device"]) for k, v in inputs2.items()}

            outputs2 = model.generate(**inputs2, max_new_tokens=150, do_sample=False, temperature=0.0)
            generated_text2 = tokenizer.decode(outputs2[0], skip_special_tokens=True).strip()

            generated_answers_t2.append(generated_text2)
            second_correct = generated_text2 == reference
            if second_correct:
                correct_t2 += 1

            # Compute Δi→c and Δc→i
            if not first_correct and second_correct:
                incorrect_to_correct += 1
            if first_correct and not second_correct:
                correct_to_incorrect += 1

            total += 1

    accuracy_t1 = correct_t1 / total if total > 0 else 0
    accuracy_t2 = correct_t2 / total if total > 0 else 0
    delta = accuracy_t2 - accuracy_t1

    print(f"Evaluation Metrics:")
    print(f"Accuracy@t1: {accuracy_t1 * 100:.2f}%")
    print(f"Accuracy@t2: {accuracy_t2 * 100:.2f}%")
    print(f"Δ(t1, t2): {delta * 100:.2f}%")
    print(f"Δi→c(t1, t2): {incorrect_to_correct * 100 / total:.2f}%")
    print(f"Δc→i(t1, t2): {correct_to_incorrect * 100 / total:.2f}%")

    return generated_answers_t1, generated_answers_t2, reference_answers, accuracy_t1, accuracy_t2, delta, incorrect_to_correct, correct_to_incorrect


# Main function to run the training and evaluation process
def main(config_file=None):
    config = load_config(config_file)

    # Check for device
    device = torch.device(config["device"])
    print(f"Using device: {device}")

    # Load tokenizer
    model_name = config["model_name"]
    print(f"Loading tokenizer for {model_name}...")
    tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

    # Load reference model (pre-trained, not fine-tuned)
    print(f"Loading reference model {model_name} for Stage I KL loss...")
    reference_model = AutoModelForCausalLM.from_pretrained(model_name, trust_remote_code=True)
    reference_model.to(device)
    reference_model.eval()  # Freeze reference model

    # Load trainable model
    print(f"Loading trainable model {model_name}...")
    train_model = AutoModelForCausalLM.from_pretrained(model_name, trust_remote_code=True)
    train_model.to(device)

    # Load the dataset
    data_file_path = config["data_file"]
    if not os.path.exists(data_file_path):
        raise FileNotFoundError(f"Data file {data_file_path} not found.")

    df = pd.read_csv(data_file_path)

    # Prepare the data for Stage I and Stage II
    data_stage_one = df[["question", "correct_answer"]].dropna().to_dict(orient="records")
    data_stage_two = df[["question", "correct_answer"]].dropna().to_dict(orient="records")

    # Stage I training (Initialization)
    print("Starting Stage I training (Initialization)...")
    stage_one_initialization(
        train_model, reference_model, tokenizer, data_stage_one, config
    )

    # Stage II training (Implicit Self-Correction)
    print("Starting Stage II training (Implicit Self-Correction)...")
    stage_two_training(
        train_model, tokenizer, data_stage_two, config
    )

    # Save the trained model
    save_path = "./trained_socre_model"
    os.makedirs(save_path, exist_ok=True)
    print(f"Saving the trained model to {save_path}...")
    train_model.save_pretrained(save_path)
    tokenizer.save_pretrained(save_path)

    # Evaluation
    print("Starting evaluation...")
    # For evaluation, we can use the same data or separate validation data
    # Here, we'll use the same Stage Two data for simplicity
    (generated_answers_t1, generated_answers_t2, reference_answers, accuracy_t1,
     accuracy_t2, delta, incorrect_to_correct, correct_to_incorrect) = evaluate_model(
        train_model, tokenizer, data_stage_two, config
    )

    # Optionally, save evaluation results
    eval_results = pd.DataFrame({
        "question": [ex['question'] for ex in data_stage_two],
        "generated_answer_t1": generated_answers_t1,
        "generated_answer_t2": generated_answers_t2,
        "reference_answer": reference_answers
    })
    eval_results.to_csv(os.path.join(save_path, "evaluation_results.csv"), index=False)
    print(f"Saved evaluation results to {os.path.join(save_path, 'evaluation_results.csv')}")

    # Save additional metrics
    metrics = {
        "Accuracy@t1": accuracy_t1,
        "Accuracy@t2": accuracy_t2,
        "Δ(t1, t2)": delta,
        "Δi→c(t1, t2)": incorrect_to_correct / total if total > 0 else 0,
        "Δc→i(t1, t2)": correct_to_incorrect / total if total > 0 else 0
    }
    with open(os.path.join(save_path, "evaluation_metrics.yaml"), 'w') as f:
        yaml.dump(metrics, f)
    print(f"Saved evaluation metrics to {os.path.join(save_path, 'evaluation_metrics.yaml')}")


# Run the main function
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--config', type=str, help='Path to config file', default=None)
    args = parser.parse_args()

    main(args.config)