07. 推理模型与思维链¶

class DeepSeekR1Architecture:
    """
    DeepSeek R1架构特点
    """

    def __init__(self):
        self.architecture = {
            # MoE架构
            "total_params": "671B",
            "activated_params": "37B",  # 每次前向只激活37B
            "num_experts": 256,
            "active_experts": 8,

            # 推理优化
            "attention": "Multi-Head Latent Attention (MLA)",
            "kv_cache": "压缩90%",

            # 训练创新
            "training": {
                "stage1": "冷启动：数千条高质量CoT数据",
                "stage2": "面向推理的强化学习",
                "stage3": "拒绝采样 + SFT",
                "stage4": "全场景强化学习"
            }
        }

    def moe_forward(self, x, expert_choice):
        """
        MoE前向传播

        Args:
            x: 输入 [batch, seq_len, hidden_dim]
            expert_choice: 选择的专家索引

        Returns:
            输出 [batch, seq_len, hidden_dim]
        """
        # 只激活选中的8个专家
        expert_outputs = []
        for expert_idx in expert_choice:
            expert = self.experts[expert_idx]
            expert_outputs.append(expert(x))

        # 加权聚合
        output = sum(w * out for w, out in zip(self.expert_weights, expert_outputs))
        return output

class R1TrainingPipeline:
    """
    DeepSeek R1训练流程
    """

    def __init__(self):
        self.stages = [
            "cold_start",      # 冷启动
            "rl_reasoning",    # 推理导向RL
            "rejection_sampling",  # 拒绝采样
            "rl_all_scenarios"     # 全场景RL
        ]

    def stage1_cold_start(self, model, reasoning_data):
        """
        阶段1：冷启动

        使用数千条高质量CoT数据进行SFT
        目的：让模型学会正确的推理格式
        """
        # 数据包含：问题 + 详细思维链 + 答案
        formatted_data = self.format_cot_data(reasoning_data)

        # 监督微调
        model = sft_train(model, formatted_data, epochs=2)
        return model

    def stage2_rl_for_reasoning(self, model, reward_model):
        """
        阶段2：面向推理的强化学习

        使用GRPO（Group Relative Policy Optimization）
        奖励信号：答案正确性 + 格式合规
        """
        for batch in reasoning_dataset:
            # 生成多个推理路径
            group_outputs = model.generate_group(batch, group_size=8)

            # 计算相对奖励
            rewards = reward_model.score_group(group_outputs)

            # GRPO更新
            advantage = rewards - rewards.mean()
            loss = -log_prob * advantage

            model.update(loss)

        return model

    def stage3_rejection_sampling(self, model):
        """
        阶段3：拒绝采样 + SFT

        生成大量推理路径，只保留正确答案的进行微调
        """
        collected_data = []

        for problem in dataset:
            # 生成多个候选
            candidates = model.generate_multiple(problem, n=16)

            # 筛选正确答案
            correct = [c for c in candidates if verify(c.answer)]  # 列表推导式过滤：只保留verify验证通过的候选结果

            # 加入训练集
            collected_data.extend(correct)

        # 用高质量数据再次SFT
        model = sft_train(model, collected_data)
        return model

    def stage4_rl_all_scenarios(self, model):
        """
        阶段4：全场景强化学习

        对齐人类偏好，提升有用性和无害性
        """
        # 结合奖励模型进行最终RL
        model = rl_train(model, reward_model, all_scenarios_data)
        return model

class R1ZeroTraining:
    """
    DeepSeek-R1-Zero：无需SFT，纯RL训练

    关键发现：
    1. 模型自发学会延长思考时间
    2. 自发出现自我反思（"Wait, let me check..."）
    3. 重新评估和修正错误
    """

    def __init__(self):
        self.rl_algorithm = "GRPO"  # Group Relative Policy Optimization
        self.reward_signals = {
            "accuracy": "答案正确性",
            "format": "格式合规（如<think>标签）"
        }

    def grpo_update(self, model, batch):
        """
        GRPO：组相对策略优化

        不需要critic模型，使用组内相对奖励
        """
        group_size = 8

        for problem in batch:
            # 为同一问题生成多个输出
            outputs = [model.generate(problem) for _ in range(group_size)]

            # 计算奖励
            rewards = [self.compute_reward(out) for out in outputs]

            # 相对优势
            mean_reward = sum(rewards) / len(rewards)
            advantages = [r - mean_reward for r in rewards]

            # 策略更新
            for output, advantage in zip(outputs, advantages):  # zip按位置配对多个可迭代对象
                loss = -output.log_prob * advantage
                model.backward(loss)

        model.step()

    def analyze_emergent_behavior(self, training_logs):
        """
        分析训练中出现的新兴行为
        """
        behaviors = {
            "reflection": [],      # 自我反思
            "correction": [],      # 错误修正
            "verification": [],    # 验证步骤
            "exploration": []      # 探索不同方法
        }

        for log in training_logs:
            thought = log.reasoning_process

            # 检测反思行为
            if any(kw in thought for kw in ["wait", "actually", "let me check"]):  # any()任一为True则返回True
                behaviors["reflection"].append(log)

            # 检测修正行为
            if "correction" in thought or "修正" in thought:
                behaviors["correction"].append(log)

        return behaviors

class O1Training:
    """
    o1模型训练方法（基于公开信息推测）
    """

    def __init__(self):
        self.key_techniques = {
            "base_model": "大规模预训练模型",
            "rl_training": "强化学习优化推理",
            "prm": "Process Reward Model（过程奖励模型）",
            "cot_data": "海量思维链数据"
        }

    def process_reward_model(self, reasoning_steps, final_answer):
        """
        过程奖励模型（PRM）

        不仅奖励最终答案，还奖励正确的推理步骤
        """
        step_rewards = []

        for i, step in enumerate(reasoning_steps):  # enumerate同时获取索引和元素
            # 评估每一步的正确性
            step_reward = self.evaluate_step(step, i)
            step_rewards.append(step_reward)

        # 最终答案奖励
        final_reward = self.verify_answer(final_answer)

        # 总奖励 = 步骤奖励之和 + 最终奖励
        total_reward = sum(step_rewards) + final_reward

        return {
            "step_rewards": step_rewards,
            "final_reward": final_reward,
            "total": total_reward
        }

    def train_with_prm(self, model, prm, dataset):
        """
        使用过程奖励模型训练
        """
        for problem in dataset:
            # 生成推理路径
            reasoning_path = model.generate_cot(problem)

            # PRM评估
            reward_info = prm.score(reasoning_path)

            # 强化学习更新
            if reward_info["total"] > threshold:
                model.reinforce(positive=True)
            else:
                model.reinforce(positive=False)

    def test_time_compute(self, model, problem, compute_budget):
        """
        测试时计算扩展

        投入更多计算时间获得更好答案
        """
        best_answer = None
        best_score = -inf

        for _ in range(compute_budget):
            # 生成候选答案
            candidate = model.generate_with_cot(problem)

            # 自我评估
            score = model.self_evaluate(candidate)

            if score > best_score:
                best_score = score
                best_answer = candidate

        return best_answer

class GeminiMultimodalReasoning:
    """
    Gemini 2.0多模态推理
    """

    def __init__(self):
        self.modalities = ["text", "image", "video", "audio"]
        self.reasoning_mode = "step_by_step"

    def multimodal_cot(self, query, media_inputs):
        """
        多模态思维链

        Args:
            query: 文本查询
            media_inputs: 图像/视频输入
        """
        # 编码多模态输入
        embeddings = []
        for media in media_inputs:
            if media.type == "image":
                emb = self.encode_image(media.data)
            elif media.type == "video":
                emb = self.encode_video(media.data)
            embeddings.append(emb)

        # 融合多模态信息
        fused = self.fusion_layer(embeddings, query)

        # 生成思维链
        reasoning_steps = []
        context = fused

        for step in range(max_steps):
            # 观察
            observation = self.observe(context, media_inputs)

            # 思考
            thought = self.think(observation, reasoning_steps)
            reasoning_steps.append(thought)

            # 检查是否足够
            if self.is_sufficient(thought):
                break

            context = self.update_context(context, thought)

        # 生成最终答案
        answer = self.generate_answer(reasoning_steps)

        return {
            "reasoning": reasoning_steps,
            "answer": answer
        }

    def real_time_reasoning(self, video_stream, query):
        """
        实时视频推理
        """
        frame_buffer = []

        for frame in video_stream:
            frame_buffer.append(frame)

            # 每N帧进行一次推理
            if len(frame_buffer) >= self.frame_interval:
                # 分析当前场景
                scene_understanding = self.analyze_scene(frame_buffer)

                # 结合历史推理
                reasoning = self.reason_with_history(
                    scene_understanding,
                    self.reasoning_history
                )

                yield {  # yield产出值，函数变为生成器
                    "timestamp": frame.timestamp,
                    "reasoning": reasoning,
                    "action": self.decide_action(reasoning, query)
                }

                frame_buffer = []

class ChainOfThought:
    """
    思维链（Chain of Thought）技术

    核心思想：让模型展示中间推理步骤，而非直接给出答案
    """

    def __init__(self):
        self.cot_templates = {
            "standard": "Let's think step by step.",
            "structured": """Step 1: Understand the problem
Step 2: Identify key information
Step 3: Apply relevant knowledge
Step 4: Derive the answer""",
            "self_consistency": "Generate multiple reasoning paths and vote"
        }

    def generate_cot_prompt(self, problem, style="standard"):
        """
        生成思维链提示
        """
        template = self.cot_templates[style]

        prompt = f"""Problem: {problem}

{template}

Solution:"""

        return prompt

    def parse_cot_output(self, output):
        """
        解析思维链输出
        """
        # 提取推理步骤
        steps = []
        current_step = ""

        for line in output.split('\n'):
            if line.strip().startswith(("Step", "1.", "2.", "-")):  # 链式调用：strip去除空白
                if current_step:
                    steps.append(current_step.strip())
                current_step = line
            else:
                current_step += " " + line

        if current_step:
            steps.append(current_step.strip())

        # 提取最终答案
        answer = self.extract_final_answer(output)

        return {
            "reasoning_steps": steps,
            "answer": answer
        }

class SelfConsistency:
    """
    自我一致性：生成多条推理路径，选择最一致的答案
    """

    def __init__(self, num_paths=10):
        self.num_paths = num_paths

    def solve_with_consistency(self, model, problem):
        """
        使用自我一致性解决问题
        """
        # 生成多条推理路径
        reasoning_paths = []
        answers = []

        for _ in range(self.num_paths):
            # 使用不同温度采样
            temperature = random.uniform(0.5, 1.0)

            output = model.generate(
                problem,
                temperature=temperature,
                cot=True
            )

            parsed = self.parse_output(output)
            reasoning_paths.append(parsed["reasoning"])
            answers.append(parsed["answer"])

        # 投票选择最一致的答案
        answer_counts = Counter(answers)  # Counter统计元素出现次数
        best_answer = answer_counts.most_common(1)[0][0]

        # 找到支持该答案的推理路径
        supporting_paths = [
            path for path, ans in zip(reasoning_paths, answers)
            if ans == best_answer
        ]

        return {
            "answer": best_answer,
            "confidence": answer_counts[best_answer] / len(answers),
            "reasoning": supporting_paths[0],  # 选择一条代表性路径
            "all_paths": reasoning_paths
        }

class TreeOfThoughts:
    """
    树状思维：探索多个推理分支
    """

    def __init__(self, branching_factor=3, max_depth=5):
        self.branching_factor = branching_factor
        self.max_depth = max_depth

    class ThoughtNode:
        def __init__(self, state, parent=None):
            self.state = state
            self.parent = parent
            self.children = []
            self.value = None
            self.visits = 0

    def solve(self, model, initial_state):
        """
        使用ToT解决问题
        """
        root = self.ThoughtNode(initial_state)

        for iteration in range(self.max_iterations):
            # 选择：使用UCT选择最有希望的节点
            node = self.select(root)

            # 扩展：生成子节点
            if not self.is_terminal(node):
                self.expand(model, node)

            # 评估：评估新节点的价值
            self.evaluate(model, node)

            # 回溯：更新路径上的节点价值
            self.backpropagate(node)

        # 返回最佳路径
        best_path = self.get_best_path(root)
        return best_path

    def expand(self, model, node):
        """
        扩展节点，生成可能的下一步
        """
        prompt = f"Given: {node.state}\nGenerate {self.branching_factor} possible next steps:"

        outputs = model.generate_multiple(prompt, n=self.branching_factor)

        for output in outputs:
            child = self.ThoughtNode(output, parent=node)
            node.children.append(child)

    def evaluate(self, model, node):
        """
        评估节点的价值
        """
        # 使用价值模型或启发式函数
        value_prompt = f"Evaluate the quality of this reasoning step:\n{node.state}"

        value = model.score(value_prompt)
        node.value = value
        node.visits += 1

class TestTimeCompute:
    """
    测试时计算扩展

    核心思想：在推理阶段投入更多计算资源，提升输出质量
    """

    def __init__(self, model):
        self.model = model
        self.strategies = {
            "sampling": "多次采样选择最佳",
            "verifier": "使用验证模型筛选",
            "mcts": "蒙特卡洛树搜索",
            "iterative": "迭代优化"
        }

    def best_of_n_sampling(self, problem, n=16):
        """
        Best-of-N采样

        生成N个候选，选择验证分数最高的
        """
        candidates = []

        for _ in range(n):
            # 生成候选
            candidate = self.model.generate(problem, temperature=0.8)

            # 验证评分
            score = self.verifier.score(candidate)

            candidates.append({
                "answer": candidate,
                "score": score
            })

        # 选择最佳
        best = max(candidates, key=lambda x: x["score"])  # lambda匿名函数
        return best["answer"]

    def process_based_verification(self, problem, max_steps=10):
        """
        基于过程的验证

        每生成一步就验证，及时纠正错误
        """
        reasoning_steps = []
        current_state = problem

        for step in range(max_steps):
            # 生成下一步
            next_step = self.model.generate_next_step(current_state)

            # 验证这一步的正确性
            step_score = self.verifier.verify_step(next_step, reasoning_steps)

            if step_score < 0.5:
                # 这一步可能有问题，尝试修正
                correction = self.model.correct_step(next_step, reasoning_steps)
                next_step = correction

            reasoning_steps.append(next_step)
            current_state = self.update_state(current_state, next_step)

            # 检查是否完成
            if self.is_complete(current_state):
                break

        return self.extract_answer(reasoning_steps)

    def monte_carlo_tree_search(self, problem, num_simulations=100):
        """
        使用MCTS进行推理
        """
        root = MCTSNode(problem)

        for _ in range(num_simulations):
            # 选择
            node = self.mcts_select(root)

            # 扩展
            if not node.is_terminal():
                self.mcts_expand(node)

            # 模拟
            reward = self.mcts_simulate(node)

            # 回溯
            self.mcts_backpropagate(node, reward)

        # 选择访问次数最多的路径
        best_child = max(root.children, key=lambda c: c.visits)
        return best_child.answer

class ComputePerformanceTradeoff:
    """
    分析计算扩展与性能提升的关系
    """

    def analyze_scaling_law(self, model, test_set):
        """
        分析测试时计算扩展的规律
        """
        results = []

        compute_budgets = [1, 2, 4, 8, 16, 32, 64]

        for budget in compute_budgets:
            correct = 0
            total_time = 0

            for problem in test_set:
                start = time.time()

                # 使用budget进行推理
                answer = self.model.solve_with_budget(problem, budget)

                elapsed = time.time() - start
                total_time += elapsed

                if self.verify(answer, problem.ground_truth):
                    correct += 1

            accuracy = correct / len(test_set)
            avg_time = total_time / len(test_set)

            results.append({
                "compute_budget": budget,
                "accuracy": accuracy,
                "avg_time": avg_time
            })

        # 拟合扩展规律
        # 通常遵循：Accuracy = a - b * exp(-c * budget)
        return results

    def plot_tradeoff(self, results):
        """
        绘制计算-性能权衡曲线
        """
        import matplotlib.pyplot as plt

        budgets = [r["compute_budget"] for r in results]
        accuracies = [r["accuracy"] for r in results]
        times = [r["avg_time"] for r in results]

        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))

        # 计算-准确率曲线
        ax1.plot(budgets, accuracies, 'o-')
        ax1.set_xlabel('Compute Budget')
        ax1.set_ylabel('Accuracy')
        ax1.set_title('Test-Time Compute Scaling')
        ax1.set_xscale('log')

        # 时间-准确率曲线
        ax2.plot(times, accuracies, 'o-')
        ax2.set_xlabel('Average Time (s)')
        ax2.set_ylabel('Accuracy')
        ax2.set_title('Latency vs Accuracy')

        plt.tight_layout()
        plt.show()

class RLForReasoning:
    """
    用于推理的强化学习
    """

    def __init__(self, model, reward_model):
        self.policy = model
        self.reward_model = reward_model
        self.rl_algorithm = "PPO"  # 或 GRPO

    def compute_reasoning_reward(self, output, ground_truth):
        """
        计算推理任务的奖励
        """
        rewards = {}

        # 1. 最终答案正确性
        rewards["accuracy"] = 1.0 if self.match(output.answer, ground_truth) else 0.0

        # 2. 推理过程质量
        if hasattr(output, "reasoning_steps"):  # hasattr检查对象是否有某属性
            # 步骤数量适中（不太短也不太长）
            step_count = len(output.reasoning_steps)
            rewards["step_count"] = 1.0 - abs(step_count - self.optimal_steps) / self.optimal_steps

            # 格式合规
            rewards["format"] = 1.0 if self.check_format(output.reasoning_steps) else 0.0

        # 3. 综合奖励
        total_reward = (
            0.6 * rewards["accuracy"] +
            0.2 * rewards.get("step_count", 0) +  # get(key, 0)：键不存在时返回0，保证可选指标缺失时计算不报错
            0.2 * rewards.get("format", 0)
        )

        return total_reward

    def grpo_training_step(self, batch):
        """
        GRPO (Group Relative Policy Optimization) 训练步骤

        DeepSeek R1使用的算法
        """
        group_size = 8

        for problem in batch:
            # 为同一问题生成多个输出
            group_outputs = []
            for _ in range(group_size):
                output = self.policy.generate(problem)
                reward = self.compute_reasoning_reward(output, problem.answer)
                group_outputs.append((output, reward))

            # 计算组内相对奖励
            rewards = [r for _, r in group_outputs]
            mean_reward = sum(rewards) / len(rewards)
            std_reward = (sum((r - mean_reward)**2 for r in rewards) / len(rewards)) ** 0.5

            # 归一化优势
            advantages = [(r - mean_reward) / (std_reward + 1e-8) for r in rewards]

            # 策略更新
            for (output, _), advantage in zip(group_outputs, advantages):
                loss = -output.log_prob * advantage
                self.policy.backward(loss)

        self.policy.step()

class ReasoningDistillation:
    """
    将大推理模型的能力蒸馏到小模型
    """

    def __init__(self, teacher_model, student_model):
        self.teacher = teacher_model
        self.student = student_model

    def generate_teacher_reasoning(self, dataset):
        """
        使用教师模型生成高质量推理数据
        """
        reasoning_data = []

        for problem in dataset:
            # 教师模型生成详细推理
            teacher_output = self.teacher.generate_with_cot(
                problem,
                temperature=0.7,
                max_tokens=4096
            )

            # 验证答案正确性
            if self.verify(teacher_output.answer, problem.answer):
                reasoning_data.append({
                    "problem": problem,
                    "reasoning": teacher_output.reasoning,
                    "answer": teacher_output.answer
                })

        return reasoning_data

    def distill(self, reasoning_data, epochs=3):
        """
        蒸馏训练
        """
        for epoch in range(epochs):
            for batch in self.get_batches(reasoning_data):
                # 学生模型生成
                student_output = self.student.generate(batch.problem)

                # 计算蒸馏损失
                # 1. 输出分布匹配
                distill_loss = self.kl_divergence(
                    student_output.logits,
                    batch.teacher_logits
                )

                # 2. 推理过程匹配
                reasoning_loss = self.sequence_loss(
                    student_output.reasoning_tokens,
                    batch.teacher_reasoning_tokens
                )

                # 3. 答案正确性
                answer_loss = self.cross_entropy(
                    student_output.answer_logits,
                    batch.answer
                )

                total_loss = distill_loss + reasoning_loss + answer_loss

                self.student.backward(total_loss)
                self.student.step()

import torch
import torch.nn as nn
from transformers import AutoModelForCausalLM, AutoTokenizer

class SimpleReasoningModel:
    """
    简化的推理模型实现
    """

    def __init__(self, model_name="Qwen/QwQ-32B"):
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.bfloat16,
            device_map="auto"
        )
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

    def solve_with_cot(self, problem, max_length=4096):
        """
        使用思维链解决问题
        """
        # 构建提示
        prompt = f"""请逐步思考以下问题，展示你的推理过程。

问题：{problem}

思考过程："""

        # 生成
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)

        outputs = self.model.generate(
            **inputs,
            max_length=max_length,
            temperature=0.7,
            do_sample=True,
            pad_token_id=self.tokenizer.eos_token_id
        )

        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)

        # 解析推理过程和答案
        return self.parse_response(response)

    def parse_response(self, response):
        """
        解析模型输出
        """
        # 提取推理过程
        if "答案：" in response or "Answer:" in response:
            parts = response.split("答案：") if "答案：" in response else response.split("Answer:")
            reasoning = parts[0].strip()
            answer = parts[1].strip()
        else:
            reasoning = response
            answer = "未明确给出答案"

        return {
            "reasoning": reasoning,
            "answer": answer
        }

# 使用示例
def example_usage():
    model = SimpleReasoningModel()

    problem = """
    一个农场有鸡和兔子，共有35个头，94只脚。
    问：鸡和兔子各有多少只？
    """

    result = model.solve_with_cot(problem)

    print("推理过程：")
    print(result["reasoning"])
    print("\n答案：")
    print(result["answer"])

# example_usage()

class ReasoningEvaluator:
    """
    推理能力评估器
    """

    def __init__(self):
        self.benchmarks = {
            "gsm8k": "数学推理",
            "math": "竞赛级数学",
            "humaneval": "代码生成",
            "mmlu": "多学科知识",
            "arc": "科学推理"
        }

    def evaluate_gsm8k(self, model, test_set):
        """
        评估GSM8K数学推理
        """
        correct = 0
        total = len(test_set)

        for problem in test_set:
            # 生成答案
            result = model.solve_with_cot(problem.question)

            # 提取数值答案
            predicted = self.extract_number(result["answer"])
            ground_truth = self.extract_number(problem.answer)

            if abs(predicted - ground_truth) < 1e-6:
                correct += 1

        accuracy = correct / total
        return {
            "benchmark": "GSM8K",
            "accuracy": accuracy,
            "correct": correct,
            "total": total
        }

    def evaluate_reasoning_quality(self, model, test_set):
        """
        评估推理过程质量
        """
        metrics = {
            "step_count": [],
            "reasoning_length": [],
            "self_correction": 0,
            "verification_steps": 0
        }

        for problem in test_set:
            result = model.solve_with_cot(problem.question)
            reasoning = result["reasoning"]

            # 统计步骤数
            steps = reasoning.count("Step") + reasoning.count("步骤")
            metrics["step_count"].append(steps)

            # 检测自我修正
            if any(kw in reasoning.lower() for kw in ["wait", "actually", "correction", "修正"]):
                metrics["self_correction"] += 1

            # 检测验证步骤
            if any(kw in reasoning.lower() for kw in ["verify", "check", "验证", "检查"]):
                metrics["verification_steps"] += 1

        return {
            "avg_steps": sum(metrics["step_count"]) / len(metrics["step_count"]),
            "self_correction_rate": metrics["self_correction"] / len(test_set),
            "verification_rate": metrics["verification_steps"] / len(test_set)
        }

import matplotlib.pyplot as plt
import networkx as nx

class CoTVisualizer:
    """
    思维链可视化工具
    """

    def visualize_tree(self, reasoning_steps):
        """
        将思维链可视化为树状图
        """
        G = nx.DiGraph()

        # 添加节点
        for i, step in enumerate(reasoning_steps):
            G.add_node(i, label=f"Step {i+1}")

        # 添加边（线性连接）
        for i in range(len(reasoning_steps) - 1):
            G.add_edge(i, i+1)

        # 绘制
        pos = nx.spring_layout(G)
        plt.figure(figsize=(12, 8))

        nx.draw(G, pos, with_labels=True, node_color='lightblue',
                node_size=3000, font_size=10, arrows=True)

        # 添加步骤内容
        for i, step in enumerate(reasoning_steps):
            plt.text(pos[i][0], pos[i][1]-0.1, step[:50]+"...",
                    ha='center', fontsize=8)

        plt.title("Chain of Thought Visualization")
        plt.show()

    def plot_confidence_over_steps(self, step_confidences):
        """
        绘制每步的置信度变化
        """
        plt.figure(figsize=(10, 6))
        plt.plot(range(1, len(step_confidences)+1), step_confidences, 'o-')
        plt.xlabel('Step')
        plt.ylabel('Confidence')
        plt.title('Confidence Over Reasoning Steps')
        plt.grid(True)
        plt.show()

from openai import OpenAI

client = OpenAI()

# 官方文档当前推荐优先使用 Responses API
response = client.responses.create(
    model="gpt-5.4",
    reasoning={"effort": "high"},
    input="分析这段代码的潜在竞态条件。"
)

print(response.output_text)

class ReasoningEffortSelector:
    """
    根据任务类型自动选择推理等级
    """

    def select_effort(self, task_description: str) -> str:
        """分析任务描述，返回推荐的推理等级"""
        task_lower = task_description.lower()

        # 简单任务 → none/low
        if any(kw in task_lower for kw in
               ["格式化", "提取", "转换", "format", "extract"]):
            return "low"

        # 中等复杂度 → medium
        if any(kw in task_lower for kw in
               ["实现", "编写", "debug", "implement", "fix"]):
            return "medium"

        # 复杂推理 → high
        if any(kw in task_lower for kw in
               ["架构", "设计", "优化", "architect", "design", "optimize"]):
            return "high"

        # 最复杂任务 → xhigh
        if any(kw in task_lower for kw in
               ["安全审计", "重构", "研究", "audit", "refactor", "research"]):
            return "xhigh"

        return "medium"  # 默认

# 使用示例
selector = ReasoningEffortSelector()

effort = selector.select_effort("""
    重构这个异步函数以处理所有边缘情况，
    包括网络故障、超时和部分响应
""")

task = """
重构这个异步函数以处理所有边缘情况，
包括网络故障、超时和部分响应
"""

response = client.responses.create(
    model="gpt-5.4",
    reasoning={"effort": effort},
    input=task
)

print(response.output_text)

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    messages=[{
        "role": "user",
        "content": "分析这个复杂的分布式系统架构..."
    }]
)

def select_reasoning_effort(task: str, context_size: int, budget: float) -> str:
    """
    推理等级选择决策树

    Args:
        task: 任务描述
        context_size: 上下文大小（tokens）
        budget: 预算约束（0-1，1表示无限制）
    """
    # 预算约束
    if budget < 0.3:
        return "low"

    # 上下文大小影响
    if context_size > 500000:  # 超长上下文
        return "medium"  # 避免成本爆炸

    # 任务复杂度分析
    complexity_keywords = {
        "xhigh": ["安全审计", "架构重构", "研究分析", "数学证明"],
        "high": ["调试", "优化", "设计", "分析", "debug", "optimize"],
        "medium": ["实现", "编写", "修改", "implement", "write"],
        "low": ["格式化", "提取", "转换", "format", "extract"]
    }

    for level, keywords in complexity_keywords.items():
        if any(kw in task.lower() for kw in keywords):
            return level

    return "medium"

class ClosedSourceCoTRetrieval:
    """
    闭源模型思维链获取策略

    通过API调用获取闭源大模型的思维链推理能力
    适用于GPT-4、Claude、DeepSeek等闭源模型的蒸馏
    """

    def __init__(self, api_config):
        self.api_config = api_config
        self.client = self._init_client(api_config)

    def _init_client(self, api_config):
        """
        初始化API客户端
        """
        if api_config["provider"] == "openai":
            import openai
            return openai.OpenAI(api_key=api_config["api_key"])
        elif api_config["provider"] == "anthropic":
            import anthropic
            return anthropic.Anthropic(api_key=api_config["api_key"])
        elif api_config["provider"] == "deepseek":
            import openai
            return openai.OpenAI(
                api_key=api_config["api_key"],
                base_url="https://api.deepseek.com"
            )

    def generate_cot_with_api(self, prompt, provider="openai"):
        """
        通过API生成思维链

        Args:
            prompt: 输入问题
            provider: API提供商

        Returns:
            包含推理过程和答案的响应
        """
        if provider == "openai":
            return self._generate_with_openai(prompt)
        elif provider == "anthropic":
            return self._generate_with_anthropic(prompt)
        elif provider == "deepseek":
            return self._generate_with_deepseek(prompt)

    def _generate_with_openai(self, prompt):
        """
        使用OpenAI API生成思维链
        """
        # 添加思维链提示
        cot_prompt = f"{prompt}\n\n请逐步思考并展示你的推理过程。"

        response = self.client.chat.completions.create(
            model="gpt-5.4",
            messages=[{"role": "user", "content": cot_prompt}],
            temperature=0.7,
            max_tokens=4096
        )

        return self._parse_cot_response(response.choices[0].message.content)

    def _generate_with_anthropic(self, prompt):
        """
        使用Anthropic API生成思维链
        """
        cot_prompt = f"{prompt}\n\n请逐步思考并展示你的推理过程。"

        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": cot_prompt}]
        )

        return self._parse_cot_response(response.content[0].text)

    def _generate_with_deepseek(self, prompt):
        """
        使用DeepSeek API生成思维链
        """
        cot_prompt = f"{prompt}\n\n请逐步思考并展示你的推理过程。"

        response = self.client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": cot_prompt}],
            temperature=0.7,
            max_tokens=4096
        )

        return self._parse_cot_response(response.choices[0].message.content)

    def _parse_cot_response(self, response):
        """
        解析思维链响应

        提取推理步骤和最终答案
        """
        lines = response.split("\n")

        reasoning_steps = []
        answer = ""

        for line in lines:
            if "答案：" in line or "Answer:" in line or "因此" in line or "所以" in line:
                # 这行可能包含最终答案
                if not answer:
                    answer = line.split("答案：")[-1].split("Answer:")[-1].strip()
            else:
                reasoning_steps.append(line)

        return {
            "reasoning": "\n".join(reasoning_steps),
            "answer": answer,
            "full_response": response
        }

    def batch_generate_cot_data(self, prompts, provider="openai", max_workers=10):
        """
        批量生成思维链数据

        Args:
            prompts: 问题列表
            provider: API提供商
            max_workers: 最大并发数

        Returns:
            思维链数据列表
        """
        from concurrent.futures import ThreadPoolExecutor, as_completed

        cot_data = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.generate_cot_with_api, prompt, provider): prompt
                for prompt in prompts
            }

            for future in as_completed(futures):
                try:
                    result = future.result(timeout=60)
                    cot_data.append(result)
                except Exception as e:
                    print(f"Error generating CoT: {e}")
                    continue

        return cot_data

    def generate_verified_cot_data(self, prompts, ground_truths, provider="openai"):
        """
        生成并验证思维链数据

        只保留推理正确的样本
        """
        verified_data = []

        for prompt, ground_truth in zip(prompts, ground_truths):
            result = self.generate_cot_with_api(prompt, provider)

            # 验证答案正确性
            if self._verify_answer(result["answer"], ground_truth):
                verified_data.append({
                    "prompt": prompt,
                    "reasoning": result["reasoning"],
                    "answer": result["answer"],
                    "ground_truth": ground_truth
                })

        return verified_data

    def _verify_answer(self, predicted, ground_truth):
        """
        验证答案是否正确

        简单实现，实际需要根据任务类型定制
        """
        # 数值容差
        try:
            pred_num = float(predicted)
            gt_num = float(ground_truth)
            return abs(pred_num - gt_num) < 1e-6
        except:
            pass

        # 字符串匹配
        return predicted.strip().lower() == ground_truth.strip().lower()