Prompt工程进阶：Chain-of-Thought的变体与实践

# 对比：标准Prompt vs CoT Prompt

# 标准Prompt
standard_prompt = """
问：小明有5个苹果，小红给了他3个，
   然后小明吃掉了2个。小明现在有多少苹果？

答：
"""

# CoT Prompt
cot_prompt = """
问：小明有5个苹果，小红给了他3个，
   然后小明吃掉了2个。小明现在有多少苹果？

让我们逐步推理：

1. 小明开始有5个苹果
2. 小红给了他3个，所以：5 + 3 = 8个
3. 小明吃掉了2个，所以：8 - 2 = 6个

答：6个
"""

class CoTTriggerMethods:
    """
    2026年主流CoT触发方法
    """
    
    @staticmethod
    def few_shot_cot(examples: list[dict]) -> str:
        """
        Few-shot CoT：通过示例展示推理过程
        """
        prompt = "请在回答时展示完整的推理过程。\n\n"
        prompt += "示例：\n"
        
        for ex in examples:
            prompt += f"问题：{ex['question']}\n"
            prompt += "推理过程：\n"
            for step in ex['reasoning_steps']:
                prompt += f"  {step}\n"
            prompt += f"答案：{ex['answer']}\n\n"
        
        return prompt
    
    @staticmethod
    def zero_shot_cot(question: str) -> str:
        """
        Zero-shot CoT：使用触发词
        2026年最佳触发词组合
        """
        return f"""{question}

请逐步思考（Step by Step），展示完整的推理过程，最后给出答案。"""
    
    @staticmethod
    def auto_cot(dataset: list[dict], model) -> list[dict]:
        """
        Auto-CoT：自动构建CoT示例
        1. 使用聚类选择多样性问题
        2. 使用模型生成推理过程
        3. 验证生成的正确性
        """
        # 步骤1：问题聚类
        embeddings = model.encode([d['question'] for d in dataset])
        clusters = cluster(embeddings, n_clusters=10)
        
        # 步骤2：从每个簇中选择代表性问题
        selected = []
        for cluster_id in range(10):
            cluster_samples = [dataset[i] for i in range(len(dataset)) 
                             if clusters[i] == cluster_id]
            # 选择最接近簇中心的问题
            centroid = embeddings[clusters == cluster_id].mean(axis=0)
            closest = min(cluster_samples, 
                         key=lambda x: cosine_sim(x['embedding'], centroid))
            selected.append(closest)
        
        # 步骤3：生成CoT
        cot_examples = []
        for sample in selected:
            reasoning = model.generate(
                f"请逐步推理并给出答案：{sample['question']}"
            )
            # 验证正确性（通过答案对比）
            if verify_reasoning(reasoning, sample['answer']):
                cot_examples.append({
                    'question': sample['question'],
                    'reasoning': reasoning,
                    'answer': sample['answer']
                })
        
        return cot_examples

class SelfConsistency:
    """
    Self-Consistency with CoT
    核心：多路径推理 + 多数投票
    """
    
    def __init__(self, model, n_samples=10, temperature=0.7):
        self.model = model
        self.n_samples = n_samples
        self.temperature = temperature
    
    def solve(self, question: str) -> tuple[str, dict]:
        """
        使用Self-Consistency解决推理问题
        """
        # 多次采样推理路径
        reasoning_paths = []
        answers = []
        
        for _ in range(self.n_samples):
            response = self.model.generate(
                prompt=question,
                cot_trigger=True,
                temperature=self.temperature,
                max_tokens=500
            )
            
            # 提取推理过程和最终答案
            reasoning, answer = self._parse_response(response)
            reasoning_paths.append(reasoning)
            answers.append(answer)
        
        # 多数投票
        final_answer = self._majority_vote(answers)
        
        # 统计信息
        stats = {
            "total_samples": self.n_samples,
            "answer_distribution": Counter(answers),
            "consistency_rate": answers.count(final_answer) / self.n_samples,
            "winning_reasoning": reasoning_paths[answers.index(final_answer)]
        }
        
        return final_answer, stats
    
    def _majority_vote(self, answers: list[str]) -> str:
        """多数投票"""
        # 使用语义相似度聚合相似答案
        from collections import Counter
        
        # 简单投票
        return Counter(answers).most_common(1)[0][0]
    
    def _parse_response(self, response: str) -> tuple[str, str]:
        """从响应中提取推理过程和答案"""
        # 假设格式：推理过程...\n因此，答案是：XXX
        parts = response.split("因此，答案是：")
        if len(parts) == 2:
            return parts[0], parts[1].strip().rstrip('。')
        return response, response.split("\n")[-1]

from dataclasses import dataclass, field
from typing import Optional
import heapq

@dataclass
class ThoughtNode:
    """思维树节点"""
    content: str
    value: float
    depth: int
    parent: Optional['ThoughtNode'] = None
    children: list['ThoughtNode'] = field(default_factory=list)
    
    def __lt__(self, other):
        return self.value > other.value  # 最大堆

class TreeOfThoughts:
    """
    Tree of Thoughts 实现
    2026年优化：结合LLM评估和BFS/DFS搜索
    """
    
    def __init__(self, model, k=5, max_depth=5):
        self.model = model
        self.k = k  # 每步探索的候选数
        self.max_depth = max_depth
    
    def solve(self, problem: str, strategy="DFS") -> str:
        """
        使用ToT解决问题
        
        策略：
        - DFS: 深度优先搜索
        - BFS: 广度优先搜索 + 状态评估
        - A*: A*搜索（评估 + 启发式）
        """
        if strategy == "DFS":
            return self._dfs_search(problem)
        elif strategy == "BFS":
            return self._bfs_search(problem)
        elif strategy == "A_star":
            return self._a_star_search(problem)
    
    def _generate_candidates(self, thought: str) -> list[str]:
        """生成候选下一步思维"""
        prompt = f"""基于以下思维，生成3-5个不同的下一步推理方向：

当前思维：{thought}

请列出不同的推理方向（每个一行）：
"""
        response = self.model.generate(prompt, temperature=0.8)
        # 解析候选...
        return self._parse_candidates(response)
    
    def _evaluate_thought(self, thought: str, goal: str) -> float:
        """评估当前思维的价值"""
        evaluation_prompt = f"""评估以下思维路径对于解决目标问题的价值：

目标：{goal}
当前思维：{thought}

评估标准：
1. 与目标的关联性（0-10）
2. 推理的正确性（0-10）
3. 解决方案的潜力（0-10）

请给出总体评分（0-100）并简要说明理由：
"""
        response = self.model.generate(evaluation_prompt)
        return self._parse_score(response)
    
    def _bfs_search(self, problem: str) -> str:
        """
        BFS搜索策略
        每一层保留top-k节点
        """
        # 初始化根节点
        root = ThoughtNode(content=problem, value=0, depth=0)
        frontier = [root]
        
        while frontier and root.depth < self.max_depth:
            # 生成下一层
            new_frontier = []
            
            for node in frontier:
                # 生成候选
                candidates = self._generate_candidates(node.content)
                
                # 评估并选择top-k
                scored = []
                for cand in candidates:
                    value = self._evaluate_thought(cand, problem)
                    scored.append((value, cand))
                
                # 取top-k
                top_k = heapq.nlargest(self.k, scored, key=lambda x: x[0])
                
                for value, cand in top_k:
                    child = ThoughtNode(
                        content=cand,
                        value=value,
                        depth=node.depth + 1,
                        parent=node
                    )
                    node.children.append(child)
                    new_frontier.append(child)
            
            frontier = new_frontier
        
        # 选择最终解：从叶子节点中选择价值最高的
        best_leaf = max(self._get_all_leaves(root), 
                       key=lambda x: x.value)
        
        # 回溯重建路径
        return self._reconstruct_path(best_leaf)

class GraphOfThoughts:
    """
    Graph of Thoughts
    支持更复杂的思维关系：
    - 聚合（多个思维合并为一个）
    - 精炼（一个思维细化为多个）
    - 生成（从任意思维生成新思维）
    """
    
    def __init__(self, model):
        self.model = model
        self.nodes: dict[str, dict] = {}  # node_id -> {content, type, score}
        self.edges: list[tuple[str, str, str]] = []  # (from, to, relation)
    
    def solve(self, problem: str) -> str:
        # 初始化
        self._add_node(problem, "initial")
        
        # 思维演化循环
        for step in range(10):
            # 生成新思维
            new_thoughts = self._generate_thoughts()
            
            for thought in new_thoughts:
                self._add_node(thought, "generated")
                # 连接到源节点
                self._add_edge(thought, "derived_from", source_nodes)
            
            # 聚合相似的思维
            self._aggregate_similar()
            
            # 精炼不完整的思维
            self._refine_incomplete()
            
            # 评估并剪枝
            self._prune_low_value()
            
            # 检查是否完成
            if self._is_solved():
                break
        
        return self._extract_solution()
    
    def _aggregate_similar(self):
        """聚合相似的思维节点"""
        # 找出语义相似的节点对
        similar_pairs = self._find_similar_pairs(threshold=0.85)
        
        for node_a, node_b in similar_pairs:
            # 创建一个聚合节点
            aggregation_prompt = f"""综合以下两个思维：

思维A：{self.nodes[node_a]['content']}
思维B：{self.nodes[node_b]['content']}

请给出一个综合的、更完整的思维：
"""
            aggregated = self.model.generate(aggregation_prompt)
            new_node = self._add_node(aggregated, "aggregated")
            
            # 连接
            self._add_edge(new_node, "aggregates", [node_a, node_b])
            
            # 标记原节点
            self.nodes[node_a]["superseded"] = True
            self.nodes[node_b]["superseded"] = True

class ProgramOfThoughts:
    """
    Program of Thoughts
    用代码替代自然语言进行推理
    """
    
    def __init__(self, model):
        self.model = model
    
    def solve(self, problem: str) -> str:
        # 步骤1：生成程序
        program = self._generate_program(problem)
        
        # 步骤2：执行程序
        result = self._execute_program(program)
        
        # 步骤3：验证并输出
        return self._format_result(program, result)
    
    def _generate_program(self, problem: str) -> str:
        """生成Python程序"""
        prompt = f"""将以下问题转化为Python程序并执行：

问题：{problem}

要求：
1. 程序应该精确计算答案
2. 包含必要的注释
3. 变量命名清晰
4. 使用print()输出最终答案

请生成代码：
"""
        response = self.model.generate(prompt)
        return self._extract_code(response)
    
    def _execute_program(self, program: str) -> str:
        """安全执行程序"""
        # 沙箱执行
        import subprocess
        import tempfile
        
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py') as f:
            f.write(program)
            f.flush()
            
            try:
                result = subprocess.run(
                    ['python3', f.name],
                    capture_output=True,
                    text=True,
                    timeout=10
                )
                return result.stdout if result.returncode == 0 else f"Error: {result.stderr}"
            except subprocess.TimeoutExpired:
                return "Execution timeout"
    
    def _extract_code(self, response: str) -> str:
        """从LLM响应中提取代码"""
        # 查找 ```python ... ``` 块
        import re
        matches = re.findall(r'```python\n(.*?)```', response, re.DOTALL)
        if matches:
            return matches[0]
        return response

COT_BEST_PRACTICES = """
=== Chain-of-Thought 最佳实践 ===

1. 触发词选择
   - 中文：让我们逐步分析、首先、因此、所以、接下来
   - 英文：Let's think step by step、First、Therefore、Thus、Next

2. 格式规范
   - 使用编号步骤（如 1. 2. 3.）
   - 每步尽量简短
   - 最终答案用"答案是："明确标出

3. Few-shot示例设计
   - 选择多样性的示例
   - 包含典型错误案例及其修正
   - 示例数量：3-5个最佳

4. 与其他技术结合
   - CoT + Retrieval：推理时检索相关知识
   - CoT + Self-Correction：推理后自我修正
   - CoT + Verification：推理后验证结果

5. 常见错误
   - 步骤过多（>10步可能降低准确性）
   - 步骤过于简略（缺乏解释）
   - 缺少最终答案确认
"""