Tree of Thoughts (ToT) is a systematic problem-solving framework that enhances language models' reasoning capabilities by exploring and maintaining multiple paths of thought simultaneously.
The core concept of ToT breaks down problem-solving processes into multiple stages, where various "thoughts" are generated and evaluated at each stage, similar to how a chess player thinks several moves ahead.
Each thought represents an independent solution path that the system explores and evaluates in parallel.
In this tutorial, we run a "Make 15" game that combines four numbers to reach 15 using an LLM agent search algorithm with Tree of Thoughts (ToT).
Main steps
Expand: Creative generation phase using LLM to explore all possible equation combinations
Score: Validation phase that evaluates how accurately generated equations reach the target value of 15
Prune: Optimization phase that selects only the best candidates to determine the starting point for the next exploration
Key Components
Reverse-Polish Notation(RPN) Calculator for equation evaluation
from dotenv import load_dotenv
from langchain_opentutorial import set_env
# Attempt to load environment variables from a .env file; if unsuccessful, set them manually.
if not load_dotenv():
set_env(
{
"OPENAI_API_KEY": "",
"LANGCHAIN_API_KEY": "",
"LANGCHAIN_TRACING_V2": "true",
"LANGCHAIN_ENDPOINT": "https://api.smith.langchain.com",
"LANGCHAIN_PROJECT": "Tree of Thoughts (ToT)",
}
)
Environment variables have been set successfully.
import operator
from typing import List, Union, Literal, NamedTuple, Optional
from pydantic import BaseModel, Field, field_validator
# Define the available operators and token type
OperatorType = Literal["+", "-", "*", "/"]
TokenType = Union[float, OperatorType]
class Equation(BaseModel):
tokens: List[TokenType] = Field(
description="The stack of tokens and operators in reverse-polish notation(RPN). Example: [2, 3, '+', 5, '*'] would evaluate to (2 + 3) * 5"
)
@field_validator('tokens')
def clean_tokens(cls, v):
return [float(t) if isinstance(t, (int, float)) else t.strip() for t in v]
# Validate operators
def validate_tokens(cls, v):
cleaned_tokens = []
for token in v:
if isinstance(token, str):
token = token.strip()
if token not in {'+', '-', '*', '/'}:
raise ValueError(f"Invalid operator: {token}")
cleaned_tokens.append(token)
return cleaned_tokens
def compute(self) -> float:
"""Calculate RPN formula"""
op_funcs = {
"+": operator.add,
"-": operator.sub,
"*": operator.mul,
"/": operator.truediv,
}
stack = []
for token in self.tokens:
if isinstance(token, float):
stack.append(token)
else:
b, a = stack.pop(), stack.pop()
stack.append(op_funcs[token](a, b))
return stack[0]
class GuessEquations(BaseModel):
reasoning: str = Field(
description="Detailed explanation of how the formula was created and the process"
)
equations: List[Equation] = Field(
description="List of generated equations"
)
class Candidate(NamedTuple):
candidate: Equation # Equation to be evaluate
score: Optional[float] = None # Evaluation score of equation
feedback: Optional[str] = None # Evaluation opinion
# Representation of Output
def __str__(self):
try:
computed = self.candidate.compute()
return f"Equation({self.candidate.tokens}) = {computed:.2f} (Score: {self.score:.2f if self.score is not None else 'N/A'})"
except Exception as e:
return f"Invalid equation: {self.candidate.tokens}; Error: {str(e)}"
class ScoredCandidate(Candidate):
candidate: Equation
score: float
feedback: str
from typing_extensions import Annotated, TypedDict
from langchain_core.runnables import RunnableConfig
def update_candidates(
existing: Optional[list] = None,
updates: Optional[Union[list, Literal["clear"]]] = None,
) -> List[str]:
if existing is None:
existing = []
if updates is None:
return existing
if updates == "clear":
return []
# Concatenate the lists
return existing + updates
class ToTState(TypedDict):
problem: str
candidates: Annotated[List[Candidate], update_candidates]
scored_candidates: Annotated[List[ScoredCandidate], update_candidates]
depth: Annotated[int, operator.add]
class Configuration(TypedDict, total=False):
max_depth: int
threshold: float
k: int
beam_size: int
def _ensure_configurable(config: RunnableConfig) -> Configuration:
configurable = config.get("configurable", {})
return {
**configurable,
"max_depth": configurable.get("max_depth", 10),
"threshold": config.get("threshold", 1.0), # success threshold
"k": configurable.get("k", 5), # Number of candidates to generate
"beam_size": configurable.get("beam_size", 3), # Number of candidates to keep
}
class ExpansionState(ToTState):
seed: Optional[Candidate]
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
# Define the Prompt Template
prompt = ChatPromptTemplate.from_messages(
[
( # System prompt: Define the rules of the "Make 15" game
"system",
"""You are playing Make 15. Create equations that evaluate to exactly 15 using the provided numbers.
Rules:
1. Use each number exactly once
2. You can use +, -, *, / operators
3. The result must be exactly 15
4. Submit {k} different valid guesses
Evaluate your guesses carefully before submitting."""
),
( # User prompt: Includes the number to use and a history of previous attempts
"user",
"Numbers to use: {problem}\nPrevious attempt (if any): {candidate}"
),
],
).partial(candidate="")
llm = ChatOpenAI(model="gpt-4o-mini")
bound_llm = llm.with_structured_output(GuessEquations) # Output setting
solver = prompt | bound_llm # Create pipeline
from langchain_core.runnables import RunnableConfig
from langgraph.constants import Send
from typing import Dict, Any
def compute_score(problem: str, candidate: Candidate) -> ScoredCandidate:
numbers = list(map(float, problem.split()))
used_numbers = [
token for token in candidate.candidate.tokens if isinstance(token, (int, float))
]
# Check if every number is used exactly once
if sorted(used_numbers) != sorted(numbers):
return ScoredCandidate(
candidate=candidate.candidate,
score=0,
feedback="The equation must use all 4 numbers exactly once."
)
# Score between 0 and 1 based on difference from 15
try:
result = candidate.candidate.compute()
score = max(0.0, 1.0 - abs(result - 15) / 15)
feedback = f"Result: {result}"
except Exception as e:
return ScoredCandidate(
candidate=candidate.candidate,
score=0,
feedback=f"Invalid equation. Error: {repr(e)}"
)
return ScoredCandidate(
candidate=candidate.candidate,
score=score,
feedback=feedback
)
# Create new candidates using LLM
def expand(state: ExpansionState, *, config: RunnableConfig) -> Dict[str, List[str]]:
configurable = _ensure_configurable(config)
# Ensure problem is a string
problem = state["problem"]
if isinstance(problem, list):
problem = problem[0]
try:
equation_submission = solver.invoke(
{
"problem": problem,
"candidate": str(state.get("seed", "")),
"k": configurable["k"],
},
config=config,
)
return {
"candidates": [
Candidate(candidate=eq)
for eq in equation_submission.equations
if all(t in {'+', '-', '*', '/'} or isinstance(t, (int, float))
for t in eq.tokens)
]
}
except Exception as e:
return {"candidates": []}
# Calculate the score for each candidate equation
def score(state: ToTState) -> Dict[str, List[ScoredCandidate]]:
candidates = state.get("candidates", [])
if not isinstance(candidates, list):
return {"scored_candidates": [], "candidates": "clear"}
scored = [
compute_score(state["problem"], candidate)
for candidate in candidates
if hasattr(candidate, 'candidate')
]
return {"scored_candidates": scored, "candidates": "clear"}
# Choose the candidates with the best results
# Keep only top candidates as big as beam_size
def prune(state: ToTState, *, config: RunnableConfig) -> Dict[str, List[Dict[str, Any]]]:
scored_candidates = state["scored_candidates"]
beam_size = _ensure_configurable(config)["beam_size"]
organized = sorted(
scored_candidates,
key=lambda x: x.score if isinstance(x, ScoredCandidate) else -float('inf'),
reverse=True
)
pruned = organized[:beam_size]
return {
"candidates": pruned, # Update the starting point for the next iteration
"scored_candidates": "clear", # Clear the old memory
"depth": 1, # Increment the depth by 1
}
def should_terminate(state: ToTState, config: RunnableConfig):
configurable = _ensure_configurable(config)
if not state["candidates"]:
return "__end__"
best_candidate = state["candidates"][0]
solved = isinstance(best_candidate, ScoredCandidate) and abs(best_candidate.score - 1.0) < 1e-6
if solved or state["depth"] >= configurable["max_depth"]:
return "__end__"
return [Send("expand", {**state, "seed": candidate}) for candidate in state["candidates"]]
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
# Create the graph
def create_graph(config: Configuration = None) -> StateGraph:
if config is None:
config = {
"max_depth": 10,
"threshold": 1.0,
"k": 5,
"beam_size": 3
}
builder = StateGraph(state_schema=ToTState, config_schema=Configuration)
# Add nodes : expand, score, prune
builder.add_node(expand)
builder.add_node(score)
builder.add_node(prune)
# Add edges : Define the execution order between nodes
builder.add_edge("expand", "score")
builder.add_edge("score", "prune")
builder.add_conditional_edges("prune", should_terminate, path_map=["expand", "__end__"])
# Set entry point
builder.add_edge("__start__", "expand")
return builder.compile(checkpointer=MemorySaver())
from langchain_opentutorial.graphs import visualize_graph
visualize_graph(create_graph())
import itertools
import random
def generate_make_puzzles(count=10, seed=None):
if seed is not None:
random.seed(seed)
numbers = list(range(1, 10))
all_combinations = list(itertools.combinations(numbers, 4))
selected_combinations = random.sample(all_combinations, min(count, len(all_combinations)))
return [" ".join(map(str, combo)) for combo in selected_combinations]
# Example of generated number puzzles
puzzles = generate_make_puzzles()
print(f"Example puzzles: {puzzles[:3]}")
