深入理解 Chain-of-Thought (CoT)：让大模型学会"思考"

1
2020: GPT-3 时代
2
  │  大模型展现强大的 few-shot 能力
3
  │  但推理能力仍然有限
4
  │
5
2022.1: Chain-of-Thought Prompting (Wei et al.)
6
  │  Google Brain 首次提出 CoT 概念
7
  │  在算术、常识、符号推理任务上取得突破
8
  │
9
2022.3: Zero-Shot CoT (Kojima et al.)
10
  │  仅用一句"让我们一步一步思考"就能激发推理
11
  │
12
2022.4: Self-Consistency (Wang et al.)
13
  │  多个推理路径投票，提高鲁棒性
14
  │
15
2022.9: Tree of Thoughts (Yao et al.)
16
  │  将推理组织成树形搜索
17
  │
18
2022.10: ReAct (Yao et al.)
19
  │  推理 + 行动，与工具结合
20
  │
21
2023: Auto-CoT, Active-Prompt
22
  │  自动选择最有信息量的演示样例
23
  │
24
2023-2024: CoT + 工具使用
25
  │  Toolformer, PAL, Program-aided Reasoning
26
  │
27
2024-2025: 多模态 CoT
28
  │  视觉 + 语言的协同推理
29
  │
30
2025+: 推理模型崛起
31
  │  OpenAI o1, DeepSeek-R1, Claude 3.7
32
  │  CoT 训练进模型权重

1
# Few-Shot CoT Prompt 示例
2

3
prompt = """
4
问题：小明有 5 个苹果。他又买了 3 袋苹果，每袋 4 个。现在小明一共有多少个苹果？
5

6
推理：小明原来有 5 个苹果。他买了 3 袋，每袋 4 个，所以新买了 3 × 4 = 12 个苹果。总共：5 + 12 = 17 个苹果。
7

8
答案：17 个苹果
9

10
问题：一本书有 240 页。小红第一天读了 1/3，第二天读了剩下的 1/4。第三天她需要读多少页才能读完？
11

12
推理：小红第一天读了 240 × 1/3 = 80 页。剩下 240 - 80 = 160 页。第二天她读了剩下 160 × 1/4 = 40 页。最后剩下 160 - 40 = 120 页。
13

14
答案：120 页
15

16
问题：{用户问题}
17

18
推理：
19
"""

1
class AutoCoT:
2
    """
3
    自动思维链：自动选择/生成多样化的演示
4
    """
5
    def __init__(self, llm, dataset):
6
        self.llm = llm
7
        self.dataset = dataset
8

9
    def sample_diverse_questions(self, n=8):
10
        """
11
        使用聚类选择多样化的代表性问题
12
        """
13
        # 1. 编码所有问题
14
        embeddings = self.embed_questions(self.dataset.questions)
15

16
        # 2. K-means 聚类
17
        clusters = self.kmeans(embeddings, n_clusters=n)
18

19
        # 3. 从每个聚类中心选择代表问题
20
        diverse_questions = []
21
        for cluster_id in range(n):
22
            center = clusters[cluster_id].center
23
            closest = self.find_closest_question(center, embeddings)
24
            diverse_questions.append(closest)
25

26
        return diverse_questions
27

28
    def generate_cot_for_question(self, question):
29
        """使用 Zero-Shot CoT 生成推理链"""
30
        prompt = f"{question}\n让我们一步一步思考。"
31
        reasoning = self.llm.generate(prompt)
32
        return reasoning
33

34
    def build_prompt(self, user_question):
35
        """构建 Auto-CoT prompt"""
36
        # 选择多样化问题
37
        diverse_qs = self.sample_diverse_questions()
38

39
        # 为每个问题生成 CoT
40
        demonstrations = []
41
        for q in diverse_qs:
42
            cot = self.generate_cot_for_question(q)
43
            demonstrations.append({
44
                'question': q,
45
                'reasoning': cot
46
            })
47

48
        # 构建最终 prompt
49
        prompt = self.format_demonstrations(demonstrations)
50
        prompt += f"\n问题：{user_question}\n推理："
51

52
        return prompt

1
class SelfConsistency:
2
    """
3
    自一致性：从多个推理路径中投票
4
    """
5
    def __init__(self, llm, num_samples=10, temperature=0.7):
6
        self.llm = llm
7
        self.num_samples = num_samples
8
        self.temperature = temperature
9

10
    def generate(self, question):
11
        """生成多个推理路径并投票"""
12
        # 1. 生成多个 CoT 推理
13
        reasoning_paths = []
14
        for i in range(self.num_samples):
15
            prompt = self._build_prompt(question)
16
            response = self.llm.generate(
17
                prompt,
18
                temperature=self.temperature,
19
                do_sample=True
20
            )
21
            reasoning_paths.append(self._extract_answer(response))
22

23
        # 2. 多数投票
24
        answer_counts = Counter(reasoning_paths)
25
        final_answer = answer_counts.most_common(1)[0][0]
26

27
        return {
28
            'answer': final_answer,
29
            'reasoning_paths': reasoning_paths,
30
            'confidence': answer_counts[final_answer] / self.num_samples
31
        }
32

33
    def _extract_answer(self, response):
34
        """从响应中提取最终答案"""
35
        # 多种答案提取策略
36
        patterns = [
37
            r'答案是[：:]\s*(.+?)(?:\n|$)',
38
            r'答案[：:]\s*(.+?)(?:\n|$)',
39
            r'####\s*(.+?)(?:\n|$)',
40
        ]
41
        for pattern in patterns:
42
            match = re.search(pattern, response)
43
            if match:
44
                return match.group(1).strip()
45
        return response.strip()

1
class TreeOfThoughts:
2
    """
3
    树形思维：探索多个推理路径
4
    """
5
    def __init__(self, llm, branching_factor=3, max_depth=5):
6
        self.llm = llm
7
        self.branching_factor = branching_factor
8
        self.max_depth = max_depth
9
        self.tree = ThoughtTree()
10

11
    def generate_thoughts(self, state):
12
        """在当前状态下生成多个可能的下一步"""
13
        prompt = f"""
14
        当前状态：{state}
15

16
        请提出 {self.branching_factor} 个可能的下一步推理：
17
        """
18
        response = self.llm.generate(prompt)
19
        thoughts = self.parse_thoughts(response)
20
        return thoughts
21

22
    def evaluate_state(self, state):
23
        """评估当前状态的优劣"""
24
        prompt = f"""
25
        状态：{state}
26

27
        这个状态距离最终答案有多远？评估（1-10）：
28
        """
29
        score = int(self.llm.generate(prompt).strip())
30
        return score
31

32
    def search(self, problem):
33
        """使用 BFS/DFS 搜索最佳推理路径"""
34
        # 初始化
35
        root = ThoughtNode(state=problem, depth=0)
36
        self.tree.add(root)
37

38
        for depth in range(self.max_depth):
39
            current_nodes = self.tree.get_nodes_at_depth(depth)
40
            new_nodes = []
41

42
            for node in current_nodes:
43
                # 生成多个下一步
44
                thoughts = self.generate_thoughts(node.state)
45

46
                for thought in thoughts:
47
                    new_state = node.state + " → " + thought
48
                    score = self.evaluate_state(new_state)
49

50
                    new_node = ThoughtNode(
51
                        state=new_state,
52
                        parent=node,
53
                        depth=depth + 1,
54
                        score=score
55
                    )
56
                    new_nodes.append(new_node)
57

58
            # 剪枝：保留 top-k 最佳节点
59
            new_nodes.sort(key=lambda x: x.score, reverse=True)
60
            top_nodes = new_nodes[:self.branching_factor]
61

62
            for node in top_nodes:
63
                self.tree.add(node)
64

65
        # 选择最佳路径
66
        best_path = self.tree.get_best_path()
67
        return best_path

1
class GraphOfThoughts:
2
    """
3
    图思维：在推理图中合并、重用
4
    """
5
    def __init__(self, llm):
6
        self.llm = llm
7
        self.graph = ThoughtGraph()
8

9
    def transform(self, nodes, operation):
10
        """
11
        图操作：生成、聚合、改进、验证
12
        """
13
        if operation == 'aggregate':
14
            # 合并多个想法
15
            prompt = f"合并这些想法：{nodes}"
16
            new_node = self.llm.generate(prompt)
17
            return [new_node]
18

19
        elif operation == 'improve':
20
            # 改进单个想法
21
            prompt = f"改进这个想法：{nodes[0]}"
22
            return [self.llm.generate(prompt)]
23

24
        elif operation == 'generate':
25
            # 生成新想法
26
            prompt = f"基于：{nodes}\n生成新想法："
27
            return [self.llm.generate(prompt) for _ in range(3)]
28

29
    def solve(self, problem):
30
        """通过图操作解决"""
31
        # 添加初始节点
32
        initial = self.graph.add(problem)
33

34
        # 生成多个思路
35
        thoughts = self.transform([initial], 'generate')
36

37
        # 聚合思路
38
        merged = self.transform(thoughts, 'aggregate')
39

40
        # 改进思路
41
        refined = self.transform(merged, 'improve')
42

43
        return refined[0]

1
class ReActAgent:
2
    """
3
    ReAct: 推理 + 行动
4
    """
5
    def __init__(self, llm, tools):
6
        self.llm = llm
7
        self.tools = tools
8
        self.memory = []
9

10
    def think(self, observation):
11
        """思考下一步行动"""
12
        prompt = self._build_prompt(observation)
13
        thought = self.llm.generate(prompt)
14
        return thought
15

16
    def act(self, thought):
17
        """根据思考执行行动"""
18
        # 解析行动
19
        action_name, action_input = self.parse_action(thought)
20

21
        if action_name in self.tools:
22
            result = self.tools[action_name](action_input)
23
            return f"Observation: {result}"
24
        else:
25
            return f"Unknown action: {action_name}"
26

27
    def step(self, question):
28
        """执行一步 ReAct"""
29
        # 思考
30
        thought = self.think(question)
31
        self.memory.append(f"Thought: {thought}")
32

33
        # 行动
34
        if "Final Answer" in thought:
35
            return self.extract_answer(thought)
36

37
        observation = self.act(thought)
38
        self.memory.append(observation)
39

40
        return None
41

42
    def solve(self, question, max_steps=10):
43
        for step in range(max_steps):
44
            result = self.step(question)
45
            if result:
46
                return result
47
        return "Failed to find answer"

1
class SelfRefine:
2
    """
3
    自我精炼：迭代评估和改进
4
    """
5
    def __init__(self, llm, max_iterations=3):
6
        self.llm = llm
7
        self.max_iterations = max_iterations
8

9
    def generate_initial(self, question):
10
        """初始生成"""
11
        return self.llm.generate(question + "\n让我们一步一步思考。")
12

13
    def critique(self, response):
14
        """自我评估"""
15
        prompt = f"""
16
        以下是对问题的回答：
17
        {response}
18

19
        请评估这个回答，指出：
20
        1. 推理中是否有错误？
21
        2. 是否遗漏了重要步骤？
22
        3. 如何改进？
23

24
        评估：
25
        """
26
        return self.llm.generate(prompt)
27

28
    def improve(self, response, critique):
29
        """根据评估改进"""
30
        prompt = f"""
31
        原回答：{response}
32
        评估：{critique}
33

34
        请根据评估改进原回答：
35
        """
36
        return self.llm.generate(prompt)
37

38
    def solve(self, question):
39
        """通过自我精炼求解"""
40
        response = self.generate_initial(question)
41

42
        for i in range(self.max_iterations):
43
            critique = self.critique(response)
44

45
            if self._is_satisfied(critique):
46
                break
47

48
            response = self.improve(response, critique)
49

50
        return response

1
class CoTPromptBuilder:
2
    """
3
    高质量的 CoT Prompt 构建器
4
    """
5
    def __init__(self):
6
        self.templates = {}
7

8
    def build_math_cot_prompt(self, question):
9
        """数学推理的 CoT prompt"""
10
        template = """
11
你是一位数学专家。请通过清晰的步骤解决问题。
12

13
问题: {question}
14

15
让我们一步一步思考:
16
- 第一步: 提取已知信息
17
- 第二步: 确定解题方法
18
- 第三步: 应用公式或定理
19
- 第四步: 计算中间结果
20
- 第五步: 验证答案
21

22
推理:
23
"""
24
        return template.format(question=question)
25

26
    def build_logic_cot_prompt(self, question):
27
        """逻辑推理的 CoT prompt"""
28
        template = """
29
请分析这个逻辑问题，展示完整推理过程。
30

31
问题: {question}
32

33
推理框架:
34
1. 识别前提
35
2. 列出约束条件
36
3. 推导可能结论
37
4. 排除矛盾选项
38
5. 得出最终答案
39

40
推理:
41
"""
42
        return template.format(question=question)
43

44
    def build_code_cot_prompt(self, question):
45
        """代码生成的 CoT prompt"""
46
        template = """
47
请编写解决以下问题的代码，并解释思路。
48

49
问题: {question}
50

51
步骤:
52
1. 理解问题需求
53
2. 设计算法
54
3. 编写代码
55
4. 测试边界条件
56

57
代码与解释:
58
"""
59
        return template.format(question=question)

1
class AnswerExtractor:
2
    """
3
    从 CoT 响应中提取结构化答案
4
    """
5
    def __init__(self):
6
        self.patterns = [
7
            # 中文模式
8
            r'答案是[：:]?\s*(.+?)(?:\n|$)',
9
            r'最终答案[：:]?\s*(.+?)(?:\n|$)',
10
            r'结论[：:]?\s*(.+?)(?:\n|$)',
11
            r'所以[，,]?\s*(.+?)(?:\n|$)',
12

13
            # 英文模式
14
            r'[Tt]he answer is\s*[:：]?\s*(.+?)(?:\n|$)',
15
            r'[Ff]inal [Aa]nswer\s*[:：]?\s*(.+?)(?:\n|$)',
16
            r'[Tt]herefore[，,]?\s*(.+?)(?:\n|$)',
17
            r'####\s*(.+?)(?:\n|$)',
18

19
            # LaTeX boxed
20
            r'\\boxed\{(.+?)\}',
21
        ]
22

23
    def extract(self, response):
24
        """从响应中提取答案"""
25
        for pattern in self.patterns:
26
            match = re.search(pattern, response)
27
            if match:
28
                return match.group(1).strip()
29

30
        # 如果没有匹配，尝试使用 LLM 提取
31
        return self._extract_with_llm(response)
32

33
    def extract_with_confidence(self, response):
34
        """提取答案并给出置信度"""
35
        candidates = []
36
        for pattern in self.patterns:
37
            matches = re.findall(pattern, response)
38
            for match in matches:
39
                candidates.append(match.strip())
40

41
        if not candidates:
42
            return None, 0.0
43

44
        # 选择最后一个或最常出现的
45
        from collections import Counter
46
        counts = Counter(candidates)
47
        most_common = counts.most_common(1)[0]
48

49
        confidence = most_common[1] / len(candidates) if candidates else 0
50
        return most_common[0], confidence

1
class CoTEvaluator:
2
    """
3
    CoT 推理质量评估器
4
    """
5
    def __init__(self, llm):
6
        self.llm = llm
7

8
    def evaluate_correctness(self, question, reasoning, answer, expected):
9
        """评估答案正确性"""
10
        # 1. 精确匹配
11
        if str(answer).strip() == str(expected).strip():
12
            return 1.0
13

14
        # 2. 数值比较（对数学题）
15
        try:
16
            if abs(float(answer) - float(expected)) < 1e-6:
17
                return 1.0
18
        except (ValueError, TypeError):
19
            pass
20

21
        # 3. 语义相似度（使用 LLM 判断）
22
        prompt = f"""
23
        预期答案：{expected}
24
        实际答案：{answer}
25
        这两个答案在语义上是否等价？
26
        回答：是/否
27
        """
28
        response = self.llm.generate(prompt).strip()
29
        return 1.0 if "是" in response or "yes" in response.lower() else 0.0
30

31
    def evaluate_reasoning_quality(self, reasoning):
32
        """评估推理质量"""
33
        criteria = {
34
            'logical_consistency': '推理步骤是否逻辑一致？',
35
            'step_clarity': '每一步是否清晰？',
36
            'completeness': '是否包含所有必要步骤？',
37
            'efficiency': '是否高效，没有冗余步骤？'
38
        }
39

40
        scores = {}
41
        for criterion, question in criteria.items():
42
            prompt = f"推理过程：{reasoning}\n\n{question}（评分 1-10）"
43
            score = self.llm.generate(prompt)
44
            try:
45
                scores[criterion] = float(score.strip().split()[0]) / 10
46
            except (ValueError, IndexError):
47
                scores[criterion] = 0.5
48

49
        return sum(scores.values()) / len(scores)

1
class PAL:
2
    """
3
    Program-Aided Reasoning
4
    """
5
    def __init__(self, llm, code_executor):
6
        self.llm = llm
7
        self.executor = code_executor
8

9
    def solve(self, question):
10
        """通过生成代码来求解"""
11
        prompt = f"""
12
        问题：{question}
13

14
        请生成 Python 代码来解决这个问题。代码应该包含推理过程。
15

16
        代码：
17
        ```python
18
        """
19
        code = self.llm.generate(prompt)
20

21
        # 执行代码
22
        result = self.executor.execute(code)
23
        return result

1
# PAL 解决数学问题
2

3
question = "一家餐厅周一卖出 15 个汉堡，周二比周一多卖出 1/3，周三是周二的 80%。三天共卖出多少汉堡？"
4

5
# LLM 生成代码
6
generated_code = """
7
# 周一
8
monday = 15
9
# 周二: 周一 + 周一的 1/3
10
tuesday = monday + monday * (1/3)
11
# 周三: 周二的 80%
12
wednesday = tuesday * 0.8
13
# 总数
14
total = monday + tuesday + wednesday
15
print(f"周一: {monday}, 周二: {tuesday}, 周三: {wednesday}, 总数: {total}")
16
"""
17

18
# 执行得到答案
19
# 输出: 周一: 15, 周二: 20.0, 周三: 16.0, 总数: 51.0

1
class ToolformerAgent:
2
    """
3
    自主工具调用
4
    """
5
    def __init__(self, llm, tools):
6
        self.llm = llm
7
        self.tools = tools
8
        self.tool_descriptions = self._format_tools(tools)
9

10
    def decide_tool_use(self, context, question):
11
        """决定是否使用工具"""
12
        prompt = f"""
13
        可用工具:
14
        {self.tool_descriptions}
15

16
        当前问题: {question}
17

18
        请判断是否需要使用工具。如果需要，调用工具；如果不需要，直接回答。
19

20
        决定：
21
        """
22
        decision = self.llm.generate(prompt)
23

24
        if "[Tool:" in decision:
25
            tool_call = self.parse_tool_call(decision)
26
            return self.execute_tool(tool_call)
27
        else:
28
            return decision
29

30
    def execute_tool(self, tool_call):
31
        """执行工具调用"""
32
        tool_name = tool_call['name']
33
        tool_input = tool_call['input']
34

35
        if tool_name in self.tools:
36
            return self.tools[tool_name](tool_input)
37
        return None

1
class RAGWithCoT:
2
    """
3
    检索增强的思维链
4
    """
5
    def __init__(self, llm, retriever):
6
        self.llm = llm
7
        self.retriever = retriever
8

9
    def solve(self, question):
10
        # 1. 检索相关文档
11
        docs = self.retriever.retrieve(question, top_k=5)
12
        context = "\n".join([d.content for d in docs])
13

14
        # 2. 基于检索的 CoT
15
        prompt = f"""
16
        参考资料：
17
        {context}
18

19
        问题：{question}
20

21
        请基于参考资料，一步一步推理：
22
        1. 提取关键信息
23
        2. 整合多个资料
24
        3. 推导答案
25
        4. 给出最终答案
26

27
        推理：
28
        """
29
        return self.llm.generate(prompt)

1
class CoTDataGenerator:
2
    """
3
    自动生成 CoT 训练数据
4
    """
5
    def __init__(self, llm, dataset):
6
        self.llm = llm
7
        self.dataset = dataset
8

9
    def generate_cot_for_problem(self, problem, solution):
10
        """为现有问题生成 CoT"""
11
        prompt = f"""
12
        问题：{problem}
13
        答案：{solution}
14

15
        请生成到达这个答案的详细推理步骤：
16
        """
17
        cot = self.llm.generate(prompt)
18
        return cot
19

20
    def bootstrap_cot_data(self, num_samples=1000):
21
        """通过 bootstrapping 生成 CoT 数据"""
22
        cot_dataset = []
23

24
        for item in self.dataset.sample(num_samples):
25
            # 1. 使用 Zero-Shot CoT 生成初始推理
26
            initial_cot = self.generate_cot_for_problem(
27
                item.question, item.answer
28
            )
29

30
            # 2. 验证推理是否正确
31
            is_correct = self.verify_cot(
32
                item.question, initial_cot, item.answer
33
            )
34

35
            if is_correct:
36
                cot_dataset.append({
37
                    'question': item.question,
38
                    'cot': initial_cot,
39
                    'answer': item.answer
40
                })
41

42
        return cot_dataset
43

44
    def verify_cot(self, question, cot, expected_answer):
45
        """验证 CoT 是否正确"""
46
        # 执行 CoT 中的推理步骤
47
        # 或使用 LLM 验证
48
        prompt = f"""
49
        问题：{question}
50
        推理：{cot}
51
        期望答案：{expected_answer}
52

53
        这个推理是否正确地推导出期望答案？
54
        """
55
        response = self.llm.generate(prompt)
56
        return "正确" in response or "correct" in response.lower()

1
class CoTFinetune:
2
    """
3
    CoT 微调训练
4
    """
5
    def __init__(self, model, tokenizer, cot_dataset):
6
        self.model = model
7
        self.tokenizer = tokenizer
8
        self.dataset = cot_dataset
9

10
    def prepare_training_data(self):
11
        """准备训练数据"""
12
        formatted_data = []
13
        for item in self.dataset:
14
            # 格式: 问题 + CoT + 答案
15
            text = f"""
16
            问题：{item['question']}
17
            推理：{item['cot']}
18
            答案：{item['answer']}
19
            """
20
            formatted_data.append({
21
                'text': text,
22
                'input_ids': self.tokenizer.encode(text)
23
            })
24
        return formatted_data
25

26
    def train(self, epochs=3, batch_size=8, lr=2e-5):
27
        """训练循环"""
28
        optimizer = torch.optim.AdamW(
29
            self.model.parameters(),
30
            lr=lr,
31
            weight_decay=0.01
32
        )
33

34
        for epoch in range(epochs):
35
            for batch in self.dataloader(batch_size):
36
                outputs = self.model(
37
                    input_ids=batch['input_ids'],
38
                    labels=batch['input_ids']
39
                )
40
                loss = outputs.loss
41

42
                loss.backward()
43
                optimizer.step()
44
                optimizer.zero_grad()

1
class CoTRLHF:
2
    """
3
    使用 RL 优化 CoT 推理
4
    """
5
    def __init__(self, model, reward_model):
6
        self.model = model
7
        self.reward_model = reward_model
8

9
    def compute_reward(self, question, cot, answer):
10
        """计算 CoT 推理的奖励"""
11
        # 1. 答案正确性奖励
12
        correctness_reward = self.reward_model.score(question, answer)
13

14
        # 2. 推理质量奖励
15
        quality_reward = self.evaluate_reasoning(cot)
16

17
        # 3. 格式奖励
18
        format_reward = self.evaluate_format(cot)
19

20
        return (
21
            0.6 * correctness_reward +
22
            0.3 * quality_reward +
23
            0.1 * format_reward
24
        )
25

26
    def train_step(self, batch):
27
        """单步训练"""
28
        # 1. 模型生成 CoT
29
        generated_cots = self.model.generate(batch['questions'])
30

31
        # 2. 计算奖励
32
        rewards = []
33
        for q, cot in zip(batch['questions'], generated_cots):
34
            answer = self.extract_answer(cot)
35
            reward = self.compute_reward(q, cot, answer)
36
            rewards.append(reward)
37

38
        # 3. PPO 更新
39
        loss = self.ppo_loss(batch, generated_cots, rewards)
40
        loss.backward()
41
        # ... optimizer step