Implementation:Spcl Graph of thoughts Controller Output Graph

Field	Value
Pattern Name	Controller_Output_Graph
Type	Implementation
Repository	spcl/graph-of-thoughts
Source File	graph_of_thoughts/controller/controller.py
Lines	L101-152
Import	from graph_of_thoughts.controller import Controller
Domains	Serialization, Evaluation
Related Principle	Principle:Spcl_Graph_of_thoughts_Result_Serialization

Overview

Documents the Controller.output_graph(path: str) method, which serializes the complete state and results of an executed Graph of Operations to a JSON file.

Signature

def output_graph(self, path: str) -> None

Parameters

Parameter	Type	Description
path	str	The file path where the JSON output will be written.

Returns

None. The method writes a JSON file to the specified path as a side effect.

I/O

• Input: An executed Controller instance (i.e., Controller.run() has been called), containing a fully-traversed GraphOfOperations with populated thought states, scores, validity flags, and ground truth evaluations.
• Output: A JSON file at the given path containing an array of operation dictionaries plus a final cost dictionary.

Source Code

def output_graph(self, path: str) -> None:
    """
    Serialize the state and results of the operations graph to a JSON file.

    :param path: The path to the output file.
    :type path: str
    """
    output = []
    for operation in self.graph.operations:
        operation_serialized = {
            "operation": operation.operation_type.name,
            "thoughts": [thought.state for thought in operation.get_thoughts()],
        }
        if any([thought.scored for thought in operation.get_thoughts()]):
            operation_serialized["scored"] = [
                thought.scored for thought in operation.get_thoughts()
            ]
            operation_serialized["scores"] = [
                thought.score for thought in operation.get_thoughts()
            ]
        if any([thought.validated for thought in operation.get_thoughts()]):
            operation_serialized["validated"] = [
                thought.validated for thought in operation.get_thoughts()
            ]
            operation_serialized["validity"] = [
                thought.valid for thought in operation.get_thoughts()
            ]
        if any(
            [
                thought.compared_to_ground_truth
                for thought in operation.get_thoughts()
            ]
        ):
            operation_serialized["compared_to_ground_truth"] = [
                thought.compared_to_ground_truth
                for thought in operation.get_thoughts()
            ]
            operation_serialized["problem_solved"] = [
                thought.solved for thought in operation.get_thoughts()
            ]
        output.append(operation_serialized)

    output.append(
        {
            "prompt_tokens": self.lm.prompt_tokens,
            "completion_tokens": self.lm.completion_tokens,
            "cost": self.lm.cost,
        }
    )

    with open(path, "w") as file:
        file.write(json.dumps(output, indent=2))

JSON Output Structure

The method produces a JSON array. Each element (except the last) corresponds to one operation in self.graph.operations, serialized in the order they were added to the graph.

Per-Operation Dictionary

Every operation dictionary contains at minimum:

Key	Type	Description
"operation"	string	The operation type name from the OperationType enum (e.g., "generate", "score", "keep_best_n", "aggregate", "selector", "ground_truth_evaluator", "validate_and_improve").
"thoughts"	array of dict	The state dictionaries of all thoughts produced by this operation.

Conditional Fields

Additional fields are included only when at least one thought in the operation has the corresponding attribute set:

Condition	Keys Added	Types	Description
Any thought has scored == True	"scored", "scores"	array of bool, array of float	Whether each thought was scored and the score values.
Any thought has validated == True	"validated", "validity"	array of bool, array of bool	Whether each thought was validated and whether it passed validation.
Any thought has compared_to_ground_truth == True	"compared_to_ground_truth", "problem_solved"	array of bool, array of bool	Whether each thought was compared to ground truth and whether it solved the problem correctly.

Cost Dictionary (Final Element)

The last element in the output array is always a cost summary:

Key	Type	Description
"prompt_tokens"	int	Total number of prompt tokens used across all LLM calls during execution.
"completion_tokens"	int	Total number of completion tokens generated across all LLM calls.
"cost"	float	Total monetary cost (in dollars) of all LLM calls during execution.

Usage Example

from graph_of_thoughts.controller import Controller

# Build and run the controller
ctrl = Controller(lm, operations_graph, prompter, parser, problem_params)
ctrl.run()

# Serialize to JSON
ctrl.output_graph("results/got/sample_0.json")

# The resulting JSON can be loaded for analysis:
import json
with open("results/got/sample_0.json", "r") as f:
    results = json.load(f)

# Access the final cost
cost_info = results[-1]
print(f"Total cost: ${cost_info['cost']:.4f}")

# Check if the problem was solved
for entry in results:
    if "problem_solved" in entry:
        print(f"Solved: {entry['problem_solved']}")

Related

• Principle:Spcl_Graph_of_thoughts_Result_Serialization