Implementation:Spcl Graph of thoughts ValidateAndImprove 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_Validation_And_Improvement

Overview

Implementation of the iterative validation-and-improvement pattern that validates thought states and repeatedly prompts the language model for improvements until validation passes or a retry limit is reached.

Description

The ValidateAndImprove class is a concrete operation in the Graph of Thoughts framework that combines validation and iterative improvement in a single operation. It is implemented as a subclass of Operation with operation type OperationType.validate_and_improve.

The execution flow for each predecessor thought is:

1. Clone the thought via Thought.from_thought()
2. Enter a while loop:
   1. Validate: If a validate_function is provided, call it with the thought's state. Otherwise, generate a validation prompt via prompter.validation_prompt(**state), query the LM, and parse the response via parser.parse_validation_answer().
   2. Set current_thought.valid to the validation result and append to the history list.
   3. Check termination: Exit if improvement is disabled (improve=False), thought is valid, or retry limit reached (current_try >= num_tries).
   4. Improve: Generate an improve prompt via prompter.improve_prompt(**state), query the LM for one response, parse the state update via parser.parse_improve_answer(), and create a new thought with merged state.
   5. Increment current_try.
3. Append the complete history list for this thought to self.thoughts.

The internal thoughts attribute is a List[List[Thought]] -- each inner list represents the improvement history for one input thought. The get_thoughts() method returns only the last element from each inner list.

Usage

from graph_of_thoughts.operations import ValidateAndImprove

# Validate with a custom function, improve up to 3 times
vai = ValidateAndImprove(
    num_samples=1,
    improve=True,
    num_tries=3,
    validate_function=lambda state: state.get("count") == expected_count,
)

# Validate using LM (no validate_function), no improvement
vai_no_improve = ValidateAndImprove(
    improve=False,
    validate_function=None,
)

# Wire into graph
vai.add_predecessor(generate_op)

Code Reference

Source Location

• File: graph_of_thoughts/operations/operations.py, Lines 269-388
• Import: from graph_of_thoughts.operations import ValidateAndImprove

Class Signature

class ValidateAndImprove(Operation):
    operation_type: OperationType = OperationType.validate_and_improve

    def __init__(
        self,
        num_samples: int = 1,
        improve: bool = True,
        num_tries: int = 3,
        validate_function: Callable[[Dict], bool] = None,
    ) -> None:
        """
        Initializes a new ValidateAndImprove operation.

        :param num_samples: Number of samples to use for validation. Defaults to 1.
        :param improve: Whether to improve the thought if it is not valid. Defaults to True.
        :param num_tries: Number of tries to improve before giving up. Defaults to 3.
        :param validate_function: A function to validate thoughts. Defaults to None (use LM).
        """

Key Methods

• __init__(self, num_samples, improve, num_tries, validate_function) -> None -- Initializes the operation with all configuration parameters and an empty thoughts: List[List[Thought]].
• get_thoughts(self) -> List[Thought] -- Returns [thought_list[-1] for thought_list in self.thoughts], i.e., the last (most improved) thought from each history.
• _execute(self, lm, prompter, parser, **kwargs) -> None -- Core execution logic: iterates over predecessor thoughts, runs the validate-improve loop for each, and stores the full history.

Internal State

• self.num_samples: int -- Number of LM responses for validation queries.
• self.improve: bool -- Whether to attempt improvement on invalid thoughts.
• self.num_tries: int -- Maximum number of improvement attempts per thought.
• self.validate_function: Callable[[Dict], bool] -- Optional programmatic validation function.
• self.thoughts: List[List[Thought]] -- Improvement history for each input thought. Each inner list contains the sequence of thoughts from initial to final state.

I/O Contract

Input	Output	Side Effects
Predecessor thoughts from one or more predecessor operations. Each thought carries a state dictionary that will be validated and optionally improved.	Validated (and optionally improved) thoughts -- one output thought per input thought, representing the final state after the validate-improve loop. Accessed via get_thoughts() which returns the last thought from each history list. Each output thought has validated == True and valid set to the final validation result.	May query the language model multiple times per thought (once per validation if using LM, once per improvement attempt). Logs prompts, responses, and final valid/invalid counts.

Data structure detail:

# Internal storage: List[List[Thought]]
# Example for 2 input thoughts, first needing 2 improvement attempts:
self.thoughts = [
    [thought_0_v0, thought_0_v1, thought_0_v2],  # history for input 0
    [thought_1_v0],                                 # input 1 was valid immediately
]

# get_thoughts() returns:
[thought_0_v2, thought_1_v0]  # last from each history

Assertions:

• At least one predecessor must exist (len(self.predecessors) > 0)

Loop termination conditions (any one triggers exit):

• improve == False (no improvement attempted)
• current_thought.valid == True (validation passed)
• current_try >= num_tries (retry limit reached)

Usage Examples

Keyword Counting: Validate and Improve Count

from graph_of_thoughts.operations import Generate, ValidateAndImprove

# Generate a keyword count, then validate and improve
gen = Generate(num_branches_prompt=1, num_branches_response=1)

def check_count(state):
    """Programmatic validation: check if count matches actual."""
    text = state.get("text", "")
    keyword = state.get("keyword", "")
    expected = text.lower().count(keyword.lower())
    return state.get("count", -1) == expected

vai = ValidateAndImprove(
    improve=True,
    num_tries=3,
    validate_function=check_count,
)
vai.add_predecessor(gen)

LM-Based Validation Without Improvement

from graph_of_thoughts.operations import ValidateAndImprove

# Use LM to validate, but do not attempt improvement
vai = ValidateAndImprove(
    num_samples=1,
    improve=False,
    validate_function=None,  # will use prompter.validation_prompt
)
vai.add_predecessor(some_predecessor)

Related Pages

• Principle:Spcl_Graph_of_thoughts_Validation_And_Improvement - The principle this implementation realizes
• Implementation:Spcl_Graph_of_thoughts_Improve_Operation - Simpler single-pass improvement without validation
• Implementation:Spcl_Graph_of_thoughts_KeepValid_Operation - Filters thoughts based on validation results set by ValidateAndImprove
• Workflow:Spcl_Graph_of_thoughts_GoT_Keyword_Counting_Pipeline - Keyword counting workflow using ValidateAndImprove