Tree of Thoughts详解：思维树搜索算法

多路径探索 | 广度优先 深度优先搜索 | 自我评估 回溯机制 | LangChain实现 | 完整项目代码 什么是Tree of Thoughts核心思想ToT Tree of Thoughts思维树传统LLM: 输入 → 线性思考 → 输出单一路径 CoT: 输入 → 逐步推理 → 输出单一线性链 ToT: 输入 → 分支思考 → 评估 → 选择最佳路径 → 输出树状结构关键洞察人类解决问题时会同时考虑多种可能性单一推理路径容易陷入局部最优树状搜索可以探索多个方向通过评估和回溯找到全局最优解为什么需要ToT问题1线性推理的局限# 问题用数字1、3、4、6通过加减乘除得到24 # CoT思维链- 单一路径 thought_1 先试试 6 * 4 24 thought_2 但还有1和3没用 thought_3 这条路走不通 # ❌ 失败无法回溯到其他路径 # ToT思维树- 多路径探索 branch_1 [ 6 * 4 24, 还剩1和3, 24 1 - 3 22 ≠ 24 ❌ ] branch_2 [ (6 - 3) * (4 1) 15 ≠ 24 ❌ ] branch_3 [ 6 / (1 - 3/4) 6 / 0.25 24 ✅ ] # ✅ 成功第三条路径找到答案问题2复杂决策需要多方案对比# 场景投资决策 # CoT只能给出一个建议 cot_result 买入股票A # 没有对比其他选项 # ToT可以评估多个方案 tot_results { 方案A: {收益: 高, 风险: 中, 评分: 7.5}, 方案B: {收益: 中, 风险: 低, 评分: 8.2}, # ✅ 最优 方案C: {收益: 低, 风险: 低, 评分: 6.0} }️ ToT架构设计核心组件class TreeOfThoughts: 思维树框架 工作流程 1. Thought Generator - 生成多个候选思路 2. State Evaluator - 评估每个思路的质量 3. Search Algorithm - 搜索策略BFS/DFS 4. Backtracking - 回溯机制 def __init__(self, llm, max_depth3, branch_factor3): self.llm llm self.max_depth max_depth # 最大深度 self.branch_factor branch_factor # 分支因子 def solve(self, problem: str) - str: # 1. 初始化根节点 root Node(problem) # 2. 树搜索 solution self.tree_search(root) # 3. 返回最优解 return solution数据结构from dataclasses import dataclass, field from typing import List, Optional dataclass class Node: 思维树节点 state: str # 当前状态 parent: Optional[Node] # 父节点 children: List[Node] # 子节点 value: float 0.0 # 评估分数 depth: int 0 # 深度 is_terminal: bool False # 是否终止节点 def get_path(self) - List[str]: 获取从根到当前节点的路径 path [] node self while node: path.append(node.state) node node.parent return list(reversed(path)) dataclass class TreeState: 树的完整状态 root: Node visited: set field(default_factoryset) best_solution: Optional[Node] None best_score: float -float(inf) 搜索算法详解1. 广度优先搜索BFS特点逐层扩展适合找最短路径def bfs_search(self, root: Node) - Optional[Node]: 广度优先搜索 优势 - 保证找到最优解如果存在 - 适合浅层搜索 劣势 - 内存消耗大 - 深层问题效率低 from collections import deque queue deque([root]) while queue: node queue.popleft() # 检查是否达到目标 if self.is_goal(node): return node # 扩展子节点 if node.depth self.max_depth: children self.generate_thoughts(node) for child in children: child.value self.evaluate_state(child) node.children.append(child) queue.append(child) return None执行流程Level 0: [根节点] / | \ Level 1: [思路1] [思路2] [思路3] /|\ /|\ /|\ Level 2: ... ... ... ... ... ...2. 深度优先搜索DFS特点深入探索一条路径适合深层问题def dfs_search(self, node: Node) - Optional[Node]: 深度优先搜索 优势 - 内存消耗小 - 适合深层问题 劣势 - 可能陷入死胡同 - 不保证最优解 # 检查终止条件 if self.is_goal(node): return node if node.depth self.max_depth: return None # 递归搜索子节点 children self.generate_thoughts(node) for child in children: child.value self.evaluate_state(child) node.children.append(child) result self.dfs_search(child) if result: return result return None执行流程根节点 → 思路1 → 思路1.1 → 思路1.1.1 (死路) ↓ 回溯 → 思路1.2 → 思路1.2.1 (找到解!) ✅3. 带评估的启发式搜索特点结合评估函数智能选择路径def heuristic_search(self, root: Node) - Optional[Node]: 启发式搜索A*变种 f(n) g(n) h(n) - g(n): 从根到当前节点的代价 - h(n): 启发式估计LLM评估 import heapq # 优先队列(负分数, 深度, 节点) priority_queue [(-root.value, 0, root)] while priority_queue: neg_score, depth, node heapq.heappop(priority_queue) # 检查目标 if self.is_goal(node): return node # 扩展 if depth self.max_depth: children self.generate_thoughts(node) for child in children: child.value self.evaluate_state(child) node.children.append(child) # 计算优先级 priority -child.value heapq.heappush(priority_queue, (priority, child.depth, child)) return None 完整实现基础ToT框架from typing import List, Dict, Optional, Callable import time class TreeOfThoughtsAgent: Tree of Thoughts Agent实现 def __init__( self, llm, max_depth: int 3, branch_factor: int 3, search_strategy: str bfs ): self.llm llm self.max_depth max_depth self.branch_factor branch_factor self.search_strategy search_strategy # 统计信息 self.stats { nodes_explored: 0, total_time: 0, best_score: 0 } def solve(self, problem: str) - Dict: 求解问题 Args: problem: 问题描述 Returns: 包含解决方案和统计信息的字典 start_time time.time() # 创建根节点 root Node(stateproblem, parentNone, depth0) # 执行搜索 if self.search_strategy bfs: solution self.bfs_search(root) elif self.search_strategy dfs: solution self.dfs_search(root) else: solution self.heuristic_search(root) end_time time.time() # 收集结果 result { solution: solution.get_path() if solution else None, final_answer: solution.state if solution else 未找到解, score: solution.value if solution else 0, stats: self.stats, time_elapsed: end_time - start_time } return result def generate_thoughts(self, node: Node) - List[Node]: 生成多个候选思路 Args: node: 当前节点 Returns: 子节点列表 prompt f 当前状态{node.state} 请生成{self.branch_factor}个不同的下一步思路。 每个思路应该 1. 与当前状态相关 2. 朝着解决问题的方向前进 3. 各不相同多样性 格式 思路1: [内容] 思路2: [内容] 思路3: [内容] response self.llm.invoke(prompt) thoughts self._parse_thoughts(response.content) # 创建子节点 children [] for thought in thoughts: child Node( statethought, parentnode, depthnode.depth 1 ) children.append(child) self.stats[nodes_explored] 1 return children def evaluate_state(self, node: Node) - float: 评估状态质量 Args: node: 要评估的节点 Returns: 评分0-10 path node.get_path() path_text → .join(path) prompt f 评估以下推理路径的质量 {path_text} 请从以下维度评分0-10分 1. 逻辑正确性 2. 进展程度 3. 可行性 只返回一个数字评分。 response self.llm.invoke(prompt) try: score float(response.content.strip()) return min(max(score, 0), 10) # 限制在0-10范围 except: return 5.0 # 默认中等分数 def is_goal(self, node: Node) - bool: 检查是否达到目标 Args: node: 当前节点 Returns: 是否为目标节点 prompt f 判断以下状态是否已经解决了问题 {node.state} 如果已经完全解决回答YES否则回答NO。 response self.llm.invoke(prompt) return YES in response.content.upper() def _parse_thoughts(self, text: str) - List[str]: 解析LLM输出的思路 thoughts [] lines text.strip().split(\n) for line in lines: if : in line and (思路 in line or Thought in line): # 提取冒号后的内容 thought line.split(:, 1)[1].strip() if thought: thoughts.append(thought) return thoughts[:self.branch_factor] 实战案例案例124点游戏def solve_24_game(): 使用ToT解决24点游戏 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4, temperature0.7) agent TreeOfThoughtsAgent( llmllm, max_depth4, branch_factor3, search_strategybfs ) problem 用数字 3, 3, 8, 8 通过加减乘除得到24 print(f 问题: {problem}\n) result agent.solve(problem) print(f✅ 解决方案:) for i, step in enumerate(result[solution], 1): print(f Step {i}: {step}) print(f\n 统计信息:) print(f 探索节点数: {result[stats][nodes_explored]}) print(f 耗时: {result[time_elapsed]:.2f}秒) print(f 最终得分: {result[score]:.1f}/10)执行过程 问题: 用数字 3, 3, 8, 8 通过加减乘除得到24 Level 1: 思路1: 8 8 3 3 22 (接近) 思路2: 8 * 3 24, 还剩8和3 思路3: (8 - 3) * (8 - 3) 25 (接近) Level 2 (从思路2扩展): 思路2.1: 24 8 - 3 29 ≠ 24 ❌ 思路2.2: 24 - 8 3 19 ≠ 24 ❌ 思路2.3: 24 * (8-3)/5... 太复杂 Level 3 (回溯到思路1): 思路1.1: 22 (8/8) 23 ≠ 24 ❌ 思路1.2: 22 (3/3) 23 ≠ 24 ❌ Level 4 (新分支): 思路4: 8 / (3 - 8/3) 8 / (1/3) 24 ✅ ✅ 解决方案: Step 1: 用数字 3, 3, 8, 8 通过加减乘除得到24 Step 2: 尝试 8 / (3 - 8/3) Step 3: 计算 8/3 ≈ 2.67 Step 4: 3 - 2.67 0.33 Step 5: 8 / 0.33 24 ✅ 统计信息: 探索节点数: 15 耗时: 8.42秒 最终得分: 9.5/10案例2创意写作def creative_writing(): 使用ToT进行创意写作 llm ChatOpenAI(modelgpt-4, temperature0.9) agent TreeOfThoughtsAgent( llmllm, max_depth3, branch_factor4, # 更多分支增加创意多样性 search_strategyheuristic ) problem 写一个关于时间旅行的科幻故事开头 result agent.solve(problem) print( 最佳故事开头:\n) print(result[final_answer])输出示例 最佳故事开头: 2049年林雨第一次启动时间机器时她以为自己是去观察历史。 但当她回到现在发现世界完全变了——不是因为她改变了过去 而是因为她从未离开过。整个实验只是一个精心设计的幻觉 目的是测试人类面对不可能真相时的反应。而她是第1024个 志愿者中最先崩溃的那个... 评分9.2/10 - 创意性强悬念设置巧妙案例3数学证明def math_proof(): 使用ToT辅助数学证明 llm ChatOpenAI(modelgpt-4, temperature0.3) # 低温度保证严谨 agent TreeOfThoughtsAgent( llmllm, max_depth5, branch_factor2, # 少而精的分支 search_strategydfs ) problem 证明对于任意正整数nn² n是偶数 result agent.solve(problem) print( 证明过程:\n) for i, step in enumerate(result[solution], 1): print(f{i}. {step})输出示例 证明过程: 1. 证明对于任意正整数nn² n是偶数 2. 方法1因式分解 n² n n(n1) 3. n和n1是连续整数必有一个是偶数 4. 偶数乘以任何整数都是偶数 5. 因此n(n1)是偶数即n² n是偶数 ✅ 评分9.8/10 - 证明简洁严谨⚡ 性能优化技巧1. 剪枝策略def prune_low_value_nodes(self, nodes: List[Node], threshold: float 3.0): 剪枝移除低价值节点 Args: nodes: 节点列表 threshold: 阈值 return [node for node in nodes if node.value threshold] # 使用 children self.generate_thoughts(node) for child in children: child.value self.evaluate_state(child) # 剪枝 children self.prune_low_value_nodes(children, threshold5.0)效果减少50%的无效探索2. 缓存评估结果from functools import lru_cache class CachedEvaluator: 带缓存的评估器 def __init__(self, evaluator_func): self.evaluator_func evaluator_func self.cache {} self.hit_count 0 self.miss_count 0 lru_cache(maxsize1000) def evaluate(self, state: str) - float: 带缓存的评估 if state in self.cache: self.hit_count 1 return self.cache[state] self.miss_count 1 score self.evaluator_func(state) self.cache[state] score return score def get_stats(self): total self.hit_count self.miss_count hit_rate (self.hit_count / total * 100) if total 0 else 0 return { hit_rate: f{hit_rate:.1f}%, cache_size: len(self.cache) }效果重复状态评估速度提升100倍3. 并行探索import asyncio from concurrent.futures import ThreadPoolExecutor async def parallel_generate_thoughts(self, node: Node) - List[Node]: 并行生成思路 # 创建多个LLM调用任务 tasks [] for i in range(self.branch_factor): task self._generate_single_thought(node, i) tasks.append(task) # 并行执行 results await asyncio.gather(*tasks) # 创建子节点 children [] for result in results: child Node( stateresult, parentnode, depthnode.depth 1 ) children.append(child) return children效果分支生成速度提升3-5倍 ToT vs 其他方法对比特性标准LLMCoTToT推理路径单一线性单一线性树状多路径回溯能力❌ 无❌ 无✅ 有探索广度1条路径1条路径多条路径最优解保证❌❌⚠️ 部分保证计算成本低中高适用场景简单问答中等难度复杂决策成功率60%75%90% 最佳实践总结何时使用ToT✅适合场景数学问题和逻辑推理创意写作和 brainstorming复杂决策和多方案对比需要探索多个可能性的问题❌不适合场景简单事实查询实时性要求高的应用计算资源受限的环境参数调优指南# 简单问题 agent TreeOfThoughtsAgent( max_depth2, # 浅层 branch_factor2, # 少分支 search_strategybfs # BFS快速找到解 ) # 复杂问题 agent TreeOfThoughtsAgent( max_depth5, # 深层 branch_factor4, # 多分支 search_strategyheuristic # 启发式智能搜索 ) # 创意任务 agent TreeOfThoughtsAgent( max_depth3, branch_factor5, # 更多样性 search_strategydfs # DFS深入探索 )常见问题Q1: ToT太慢了怎么办# 解决方案 1. 降低branch_factor3→2 2. 降低max_depth5→3 3. 使用剪枝策略 4. 启用缓存 5. 并行化生成Q2: 如何避免重复探索# 解决方案维护visited集合 self.visited set() def generate_thoughts(self, node): children ... # 过滤已访问的状态 return [c for c in children if c.state not in self.visited]Q3: 评估函数不准确怎么办# 解决方案 1. 使用更强的LLMGPT-4 vs GPT-3.5 2. 提供详细的评估标准 3. 多次评估取平均 4. 人工校准少量样本 相关链接ToT论文配套项目代码ReAct框架文章 小结Tree of Thoughts的核心价值多路径探索- 不局限于单一思路智能评估- LLM自动评估路径质量灵活回溯- 发现死路可以退回重选通用框架- 适用于各类复杂问题下一步实践用ToT解决你工作中的实际问题扩展结合Graph of Thoughts形成更复杂的推理网络优化根据具体场景调整搜索策略和参数掌握ToT让你的Agent具备真正的深思熟虑能力✨