Implementation:Spcl Graph of thoughts GoT Sorting Topology

Field	Value
Pattern Name	GoT_Sorting_Topology
Type	Implementation (Pattern Doc)
Repository	spcl/graph-of-thoughts
Source File	`examples/sorting/sorting_032.py`
Lines	L554-598
Domains	Graph_Reasoning, Topology_Design
Related Principle	Principle:Spcl_Graph_of_thoughts_GoT_Graph_Topology_Design

Overview

This is a Pattern Doc that documents the canonical got() function for the sorting example. It is not a class or method in the GoT framework library itself, but rather a user-defined topology builder that demonstrates the full GoT decompose-solve-merge pattern applied to sorting a list of 32 numbers.

Source

File: examples/sorting/sorting_032.py, Lines 554-598

Type: User-defined function (not a framework API). Each GoT example defines its own topology builder functions.

Topology Construction Pattern

The got() function constructs a GraphOfOperations that implements the following DAG:

Generate (split): A single Generate(1, 1) operation prompts the LLM to split the input list of 32 numbers into 2 sublists of 16 numbers each.
Selector (per sublist): For each of the 2 sublists, a Selector operation filters the thoughts to isolate the sublist (by matching thought.state["part"]).
Generate (sort sublist): Each branch independently sorts its sublist via Generate(1, 5), producing 5 candidate sorted sublists.
Score (per branch): Each branch's candidates are Scored using utils.num_errors (counts sorting errors relative to the correct answer).
KeepBestN (per branch): KeepBestN(1, False) retains only the best-scoring candidate per branch (lower score is better, so higher_is_better=False).
Aggregate (merge): An Aggregate(10) operation merges the two sorted sublists into a single sorted list via an LLM merge prompt (merge sort style), producing up to 10 candidate merges.
Score (merged): The merged candidates are Scored.
KeepBestN (merged): KeepBestN(1, False) retains the best merged candidate.
Generate (refine): A Generate(1, 10) operation refines the merged result (using a "fix the incorrectly sorted list" prompt), producing 10 refinement candidates.
Score (refined): The refinements are Scored, with a predecessor link back to the previous KeepBestN to ensure proper DAG structure.
KeepBestN (final): KeepBestN(1, False) retains the best final candidate.
GroundTruth: GroundTruth(utils.test_sorting) evaluates whether the final sorted list matches the correct answer.

Full Source Code

def got() -> operations.GraphOfOperations:
    """
    Generates the Graph of Operations for the GoT method.

    :return: Graph of Operations
    :rtype: GraphOfOperations
    """
    operations_graph = operations.GraphOfOperations()

    plans = operations.Generate(1, 1)
    operations_graph.append_operation(plans)  # generate the sublists
    for i in range(1, 3):
        list_id = f"List {i}"
        sub_list = operations.Selector(
            lambda thoughts, list_id=list_id: [
                thought for thought in thoughts if thought.state["part"] == list_id
            ]
        )
        sub_list.add_predecessor(plans)
        operations_graph.add_operation(sub_list)
        sort_sub_list = operations.Generate(1, 5)
        sort_sub_list.add_predecessor(sub_list)
        operations_graph.add_operation(sort_sub_list)
        score_sub_list = operations.Score(1, False, utils.num_errors)
        score_sub_list.add_predecessor(sort_sub_list)
        operations_graph.add_operation(score_sub_list)
        keep_best_sub_list = operations.KeepBestN(1, False)
        keep_best_sub_list.add_predecessor(score_sub_list)
        operations_graph.add_operation(keep_best_sub_list)

    final_aggregate = operations.Aggregate(10)
    operations_graph.append_operation(final_aggregate)
    operations_graph.append_operation(operations.Score(1, False, utils.num_errors))
    keep_best_aggregate_final = operations.KeepBestN(1, False)
    operations_graph.append_operation(keep_best_aggregate_final)

    operations_graph.append_operation(operations.Generate(1, 10))
    score_aggr_3 = operations.Score(1, False, utils.num_errors)
    score_aggr_3.add_predecessor(keep_best_aggregate_final)
    operations_graph.append_operation(score_aggr_3)
    operations_graph.append_operation(operations.KeepBestN(1, False))

    operations_graph.append_operation(operations.GroundTruth(utils.test_sorting))

    return operations_graph

Graph Structure (DAG)

The topology forms the following DAG:

                    Generate(1,1) [split into 2 sublists]
                     /                    \
            Selector("List 1")     Selector("List 2")
                |                         |
          Generate(1,5)            Generate(1,5)
           [sort sublist]           [sort sublist]
                |                         |
            Score                      Score
                |                         |
          KeepBestN(1)             KeepBestN(1)
                 \                       /
                  \                     /
                   Aggregate(10) [merge sorted sublists]
                        |
                      Score
                        |
                   KeepBestN(1)
                        |
                   Generate(1,10) [refine merged list]
                        |
                      Score
                        |
                   KeepBestN(1)
                        |
                   GroundTruth

Key Implementation Details

append_operation vs add_operation

The function uses two different methods to add operations to the graph:

append_operation(op) -- Appends the operation to all current leaves of the graph. This is used for linear sequences.
add_operation(op) -- Adds the operation considering its explicitly set predecessors. This is used when building the parallel branches, where each operation's predecessor is manually set via add_predecessor().

Selector Lambda Closure

The Selector uses a lambda with a default argument to capture the loop variable correctly:

sub_list = operations.Selector(
    lambda thoughts, list_id=list_id: [
        thought for thought in thoughts if thought.state["part"] == list_id
    ]
)

The list_id=list_id default argument ensures each Selector captures the correct sublist identifier, avoiding the common Python closure pitfall.

Scoring Direction

All KeepBestN operations use higher_is_better=False because the scoring function (utils.num_errors) counts errors -- lower is better.

Usage Example

from graph_of_thoughts import controller, language_models, operations

# 1. Build the GoT topology
operations_graph = got()

# 2. Create the Controller with an LM, prompter, parser, and initial state
executor = controller.Controller(
    lm,                    # language model instance
    operations_graph,      # the GoT topology
    SortingPrompter(),     # domain-specific prompter
    SortingParser(),       # domain-specific parser
    {
        "original": "[3, 7, 0, 2, 8, 1, ...]",  # input list as string
        "current": "",
        "phase": 0,
        "method": "got",
    },
)

# 3. Execute the topology
executor.run()

# 4. Save results
executor.output_graph("results/got/sample_0.json")

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment