"""Tree-of-Thoughts reasoning pattern implementation.""" from __future__ import annotations import logging from typing import Any from hermes.core.types import AgentStrategy logger = logging.getLogger(__name__) class TreeNode: """A node in the thought tree.""" def __init__( self, thought: str, state: str = "", score: float = 0.0, parent: TreeNode | None = None, ) -> None: self.thought = thought self.state = state self.score = score self.parent = parent self.children: list[TreeNode] = [] self.depth = (parent.depth + 1) if parent else 0 def add_child(self, child: TreeNode) -> None: """Add a child node.""" child.parent = self child.depth = self.depth + 1 self.children.append(child) def get_path(self) -> list[str]: """Get the path from root to this node.""" path = [] node: TreeNode | None = self while node: path.append(node.thought) node = node.parent return list(reversed(path)) class TreeOfThoughtsReasoner: """Implements the Tree-of-Thoughts reasoning pattern.""" def __init__(self, max_depth: int = 3, max_branches: int = 3) -> None: self.strategy = AgentStrategy.TREE_OF_THOUGHTS self.max_depth = max_depth self.max_branches = max_branches def create_generation_prompt(self, task: str, state: str = "", num_thoughts: int = 3) -> str: """Create prompt for generating thought candidates.""" state_section = f"\nCurrent state:\n{state}\n" if state else "" return f"""You are an AI agent exploring multiple reasoning paths for a task. Task: {task} {state_section} Generate {num_thoughts} different possible next thoughts/reasoning steps. Each thought should take a different approach or consider different aspects. Format your response as: Thought 1: [First possible reasoning step] Approach: [brief description of the approach taken] Thought 2: [Second possible reasoning step] Approach: [brief description] Thought 3: [Third possible reasoning step] Approach: [brief description] Important: - Each thought should be distinct and explore a different angle - Consider both promising and unconventional approaches - Be specific about what each approach entails""" def create_evaluation_prompt(self, task: str, thoughts: list[str]) -> str: """Create prompt for evaluating thought candidates.""" thoughts_text = "\n".join( f"Thought {i + 1}: {t}" for i, t in enumerate(thoughts) ) return f"""Evaluate the following reasoning thoughts for a task. Task: {task} Thoughts to evaluate: {thoughts_text} Rate each thought on a scale of 1-10 based on: 1. Relevance to the task 2. Potential to lead to a solution 3. Logical soundness 4. Completeness Format your response as: Evaluation 1: [thought number] Score: [1-10] Reasoning: [why this score] Evaluation 2: [thought number] Score: [1-10] Reasoning: [why this score] ... Best Thought: [number of the best thought]""" def create_state_prompt(self, task: str, path: list[str]) -> str: """Create prompt to generate state from a thought path.""" path_text = "\n".join(f"Step {i + 1}: {p}" for i, p in enumerate(path)) return f"""Based on the following reasoning steps, summarize the current state. Task: {task} Reasoning steps taken: {path_text} Provide a concise summary of: 1. What has been determined so far 2. What information has been gathered 3. What remains to be resolved Format: Current State: [concise summary] Key Findings: [list of important findings] Open Questions: [what still needs to be addressed]""" def parse_generation_response(self, response: str) -> list[str]: """Parse thought generation response.""" thoughts = [] lines = response.strip().split("\n") current_thought: list[str] = [] for line in lines: stripped = line.strip() if stripped.startswith("Thought ") and ":" in stripped: if current_thought: thoughts.append(" ".join(current_thought)) current_thought = [stripped[stripped.index(":") + 1 :].strip()] elif stripped.startswith("Approach:"): if current_thought: current_thought.append(f"({stripped})") elif current_thought and stripped: current_thought.append(stripped) if current_thought: thoughts.append(" ".join(current_thought)) return thoughts[: self.max_branches] def parse_evaluation_response(self, response: str) -> list[dict[str, Any]]: """Parse evaluation response.""" evaluations = [] lines = response.strip().split("\n") current_eval: dict[str, Any] = {} for line in lines: stripped = line.strip() if stripped.startswith("Evaluation ") and ":" in stripped: if current_eval: evaluations.append(current_eval) current_eval = {"thought_number": stripped} elif stripped.startswith("Score:"): if current_eval: try: current_eval["score"] = float(stripped.split(":")[1].strip()) except ValueError: current_eval["score"] = 5.0 elif stripped.startswith("Reasoning:"): if current_eval: current_eval["reasoning"] = stripped[len("Reasoning:"):].strip() if current_eval: evaluations.append(current_eval) return evaluations def select_best(self, evaluations: list[dict[str, Any]]) -> int: """Select the best thought index based on evaluations.""" if not evaluations: return 0 best_idx = 0 best_score = -1 for i, eval_data in enumerate(evaluations): score = eval_data.get("score", 0) if score > best_score: best_score = score best_idx = i return best_idx def build_tree( self, task: str, generations: list[list[str]], evaluations: list[list[dict[str, Any]]] ) -> TreeNode: """Build a thought tree from generations and evaluations.""" root = TreeNode(thought=task, state="Initial task") current_nodes = [root] for _gen_idx, (gen, evals) in enumerate(zip(generations, evaluations, strict=False)): next_nodes: list[TreeNode] = [] for node in current_nodes: scores = {e.get("thought_number", ""): e.get("score", 5.0) for e in evals} for i, thought in enumerate(gen): thought_key = f"Thought {i + 1}" score = scores.get(thought_key, 5.0) child = TreeNode(thought=thought, score=score, parent=node) node.add_child(child) next_nodes.append(child) current_nodes = next_nodes return root def get_best_path(self, root: TreeNode) -> list[str]: """Get the best path through the tree.""" best_leaf = root best_score = float("-inf") def _find_best(node: TreeNode) -> None: nonlocal best_leaf, best_score if not node.children and node.score > best_score: best_score = node.score best_leaf = node for child in node.children: _find_best(child) _find_best(root) return best_leaf.get_path()