Implementation:Spcl Graph of thoughts KeepBestN Operation

Knowledge Sources	Graph of Thoughts Graph of Thoughts: Solving Elaborate Problems with Large Language Models
Domains	Graph_Reasoning, Thought_Operations
Last Updated	2026-02-14
Implements	Principle:Spcl_Graph_of_thoughts_Best_N_Selection

Overview

Implementation of the best-N selection pattern that retains only the N highest-scoring (or lowest-scoring) thoughts from predecessors, enabling quality-based pruning in reasoning graphs.

Description

The KeepBestN class is a concrete operation in the Graph of Thoughts framework that filters predecessor thoughts based on their scores, keeping only the top N. It is implemented as a subclass of Operation with operation type OperationType.keep_best_n.

The execution flow is:

1. Assert at least one predecessor exists
2. Retrieve all predecessor thoughts via get_previous_thoughts()
3. Assert all thoughts have been scored (thought.scored == True for all)
4. Sort thoughts by score in descending order (if higher_is_better=True) or ascending order (if higher_is_better=False)
5. Keep the first N thoughts from the sorted list
6. Clone the selected thoughts via Thought.from_thought()
7. Store the cloned thoughts as the operation's output

The class also includes error handling: if sorting fails (e.g., due to non-float scores), it filters out thoughts with non-float scores, logs an error, and retries the sort on the remaining valid thoughts.

Usage

from graph_of_thoughts.operations import KeepBestN

# Keep the single best thought (highest score)
keep1 = KeepBestN(n=1, higher_is_better=True)

# Keep the 3 best thoughts (lowest score, e.g., error count)
keep3_low = KeepBestN(n=3, higher_is_better=False)

# Wire into graph after a Score operation
keep1.add_predecessor(score_op)

Code Reference

Source Location

• File: graph_of_thoughts/operations/operations.py, Lines 612-709
• Import: from graph_of_thoughts.operations import KeepBestN

Class Signature

class KeepBestN(Operation):
    operation_type: OperationType = OperationType.keep_best_n

    def __init__(self, n: int, higher_is_better: bool = True) -> None:
        """
        Initializes a new KeepBestN operation.

        :param n: Maximum number of thoughts to keep.
        :type n: int
        :param higher_is_better: Whether higher scores are better. Defaults to True.
        :type higher_is_better: bool
        :raises AssertionError: If n is not greater than zero.
        """

Key Methods

• __init__(self, n: int, higher_is_better: bool = True) -> None -- Initializes the operation. Asserts n > 0.
• get_best_n(self) -> List[Thought] -- Internal method that retrieves predecessor thoughts, asserts all are scored, sorts by score, and returns the top N. Includes error handling for non-float scores.
• get_thoughts(self) -> List[Thought] -- Returns the list of kept thoughts after execution.
• _execute(self, lm, prompter, parser, **kwargs) -> None -- Calls get_best_n(), clones the selected thoughts, and stores them.

Internal State

• self.n: int -- Maximum number of thoughts to retain.
• self.higher_is_better: bool -- Sort direction flag.
• self.thoughts: List[Thought] -- Stores the selected thoughts after execution.

I/O Contract

Input	Output	Side Effects
Scored predecessor thoughts from one or more predecessor operations. Every thought must have scored == True (enforced by assertion in get_best_n()).	Top N thoughts (cloned) -- up to N new Thought objects cloned from the highest-scoring (or lowest-scoring) predecessors. If fewer than N thoughts exist, all are kept.	No language model interaction. Logs kept thoughts at DEBUG level and count at INFO level. Logs errors if non-float scores are encountered.

Selection logic:

sorted(
    previous_thoughts,
    key=lambda thought: thought.score,
    reverse=self.higher_is_better,
)[:self.n]

Assertions:

• At least one predecessor must exist (len(self.predecessors) >= 1)
• n > 0 (checked at initialization)
• All predecessor thoughts must be scored (thought.scored == True)

Usage Examples

Sorting: Keep Best Sorted Sublist

from graph_of_thoughts.operations import Generate, Score, KeepBestN

# Generate 5 candidate sorted sublists, score them, keep the best
gen = Generate(num_branches_prompt=5, num_branches_response=1)
score = Score(scoring_function=num_errors)  # lower error count = better
keep = KeepBestN(n=1, higher_is_better=False)  # keep lowest error count

score.add_predecessor(gen)
keep.add_predecessor(score)

Iterative Refinement: Narrowing Beam

from graph_of_thoughts.operations import Generate, Score, KeepBestN, Improve

# Generate candidates -> Score -> Keep top 3 -> Improve -> Score -> Keep top 1
gen = Generate(num_branches_prompt=5, num_branches_response=1)
score1 = Score(scoring_function=eval_fn)
keep3 = KeepBestN(n=3, higher_is_better=True)
improve = Improve()
score2 = Score(scoring_function=eval_fn)
keep1 = KeepBestN(n=1, higher_is_better=True)

score1.add_predecessor(gen)
keep3.add_predecessor(score1)
improve.add_predecessor(keep3)
score2.add_predecessor(improve)
keep1.add_predecessor(score2)

Related Pages

• Principle:Spcl_Graph_of_thoughts_Best_N_Selection - The principle this implementation realizes
• Environment:Spcl_Graph_of_thoughts_Python_3_8_Runtime
• Heuristic:Spcl_Graph_of_thoughts_Scoring_With_Error_Counting
• Implementation:Spcl_Graph_of_thoughts_Aggregate_Operation - Aggregate often follows KeepBestN in merge pipelines
• Implementation:Spcl_Graph_of_thoughts_KeepValid_Operation - Boolean-based filtering (complementary to score-based KeepBestN)
• Workflow:Spcl_Graph_of_thoughts_GoT_Sorting_Pipeline - Sorting workflow demonstrating Score-KeepBestN pattern