Spaces:
Paused
Paused
| """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() | |