Implementation:Spcl Graph of thoughts Aggregate 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_Thought_Aggregation

Overview

Implementation of the thought aggregation pattern that merges multiple thought states into a unified state by prompting a language model to combine results.

Description

The Aggregate class is a concrete operation in the Graph of Thoughts framework that collects thoughts from multiple predecessor operations, builds an aggregation prompt containing all predecessor states, and produces one or more merged thoughts. It is implemented as a subclass of Operation with operation type OperationType.aggregate.

The execution flow is:

1. Collect all predecessor thoughts via get_previous_thoughts()
2. Build a base state by merging all predecessor thought states in ascending score order (so higher-scored states overwrite lower-scored ones)
3. Generate an aggregation prompt by passing the list of predecessor states to prompter.aggregation_prompt()
4. Query the language model for num_responses responses
5. Parse the responses via parser.parse_aggregation_answer(), which may return a single dict or a list of dicts
6. For each parsed state, create a new Thought by merging the base state with the parsed update

Usage

from graph_of_thoughts.operations import Aggregate

# Create an Aggregate operation with default single response
agg = Aggregate()

# Create an Aggregate operation requesting multiple merged outputs
agg_multi = Aggregate(num_responses=3)

# Wire into graph: aggregate takes thoughts from two predecessor branches
agg.add_predecessor(branch_1_keep_best)
agg.add_predecessor(branch_2_keep_best)

Code Reference

Source Location

• File: graph_of_thoughts/operations/operations.py, Lines 536-610
• Import: from graph_of_thoughts.operations import Aggregate

Class Signature

class Aggregate(Operation):
    operation_type: OperationType = OperationType.aggregate

    def __init__(self, num_responses: int = 1) -> None:
        """
        Initializes a new Aggregate operation.

        :param num_responses: Number of responses to use for aggregation. Defaults to 1.
        :type num_responses: int
        """

Key Methods

• __init__(self, num_responses: int = 1) -> None -- Initializes the operation with the desired number of LM responses and an empty thoughts list.
• get_thoughts(self) -> List[Thought] -- Returns the list of aggregated thoughts produced by execution.
• _execute(self, lm, prompter, parser, **kwargs) -> None -- Core execution logic: collects predecessors, builds base state, prompts LM, parses response, creates merged thoughts.

Internal State

• self.thoughts: List[Thought] -- Stores the resulting merged thought(s) after execution.
• self.num_responses: int -- Number of responses requested from the language model.

I/O Contract

Input	Output	Side Effects
Predecessor thoughts from one or more predecessor operations. Each thought carries a state dictionary. All predecessor thoughts must exist (if predecessors have no thoughts, the operation returns early without producing output).	Merged thought(s) -- one or more new Thought objects whose state is the merge of the base state (all predecessor states layered by ascending score) with the LM-generated aggregation result.	Queries the language model via lm.query(). Logs the prompt and responses at DEBUG level.

Base state construction:

base_state: Dict = {}
for thought in sorted(previous_thoughts, key=lambda thought: thought.score):
    base_state = {**base_state, **thought.state}

Assertions:

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

Usage Examples

Sorting: Merge Sorted Sublists

from graph_of_thoughts.operations import Generate, Score, KeepBestN, Aggregate, Selector

# After splitting and sorting sublists in parallel branches:
# branch_1: Selector -> Generate(5) -> Score -> KeepBestN(1)
# branch_2: Selector -> Generate(5) -> Score -> KeepBestN(1)

# Merge the best sorted sublists back together
agg = Aggregate(num_responses=1)
agg.add_predecessor(branch_1_keep_best)
agg.add_predecessor(branch_2_keep_best)

# The aggregation prompt (defined in SortingPrompter) instructs the LM
# to merge two sorted sublists into a single sorted list.

Document Merging: Combine Summaries

from graph_of_thoughts.operations import Aggregate

# Merge independently generated document summaries
agg = Aggregate(num_responses=1)
agg.add_predecessor(summary_branch_1)
agg.add_predecessor(summary_branch_2)
agg.add_predecessor(summary_branch_3)

# The aggregation prompt (defined in DocMergingPrompter) instructs
# the LM to synthesize the individual summaries.

Related Pages

• Principle:Spcl_Graph_of_thoughts_Thought_Aggregation - The principle this implementation realizes
• Implementation:Spcl_Graph_of_thoughts_Selector_Operation - Selector used to split thoughts before aggregation
• Implementation:Spcl_Graph_of_thoughts_KeepBestN_Operation - KeepBestN often precedes Aggregate to filter candidates
• Workflow:Spcl_Graph_of_thoughts_GoT_Sorting_Pipeline - Sorting workflow demonstrating the Aggregate operation