Implementation:Spcl Graph of thoughts Parser ABC

Knowledge Sources	Graph of Thoughts graph-of-thoughts
Source File	graph_of_thoughts/parser/parser.py, Lines L14-90
Import	from graph_of_thoughts.parser import Parser
Domains	LLM_Orchestration, Output_Parsing
Last Updated	2026-02-14 12:00 GMT

Overview

The Parser class is the abstract base class (ABC) that defines the interface for all parsers in the Graph of Thoughts framework. Parsers are responsible for transforming raw text responses from language models into structured thought state dictionaries (or booleans/floats for validation and scoring).

Interface

Class Definition

from abc import ABC, abstractmethod
from typing import Dict, List, Union

class Parser(ABC):
    """
    Abstract base class that defines the interface for all parsers.
    Parsers are used to parse the responses from the language models.
    """

Abstract Methods

The Parser defines 5 abstract methods, one for each operation type in the Graph of Operations:

parse_aggregation_answer

@abstractmethod
def parse_aggregation_answer(
    self, states: List[Dict], texts: List[str]
) -> Union[Dict, List[Dict]]:
    """
    Parse the response from the language model for a aggregation prompt.

    :param states: The thought states used to generate the prompt.
    :type states: List[Dict]
    :param texts: The responses to the prompt from the language model.
    :type texts: List[str]
    :return: The new thought states after parsing the response from the language model.
    :rtype: Union[Dict, List[Dict]]
    """
    pass

parse_improve_answer

@abstractmethod
def parse_improve_answer(self, state: Dict, texts: List[str]) -> Dict:
    """
    Parse the response from the language model for an improve prompt.

    :param state: The thought state used to generate the prompt.
    :type state: Dict
    :param texts: The responses to the prompt from the language model.
    :type texts: List[str]
    :return: The new thought state after parsing the response from the language model.
    :rtype: Dict
    """
    pass

parse_generate_answer

@abstractmethod
def parse_generate_answer(self, state: Dict, texts: List[str]) -> List[Dict]:
    """
    Parse the response from the language model for a generate prompt.

    :param state: The thought state used to generate the prompt.
    :type state: Dict
    :param texts: The responses to the prompt from the language model.
    :type texts: List[str]
    :return: The new thought states after parsing the response from the language model.
    :rtype: List[Dict]
    """
    pass

parse_validation_answer

@abstractmethod
def parse_validation_answer(self, state: Dict, texts: List[str]) -> bool:
    """
    Parse the response from the language model for a validation prompt.

    :param state: The thought state used to generate the prompt.
    :type state: Dict
    :param texts: The responses to the prompt from the language model.
    :type texts: List[str]
    :return: Whether the thought state is valid or not.
    :rtype: bool
    """
    pass

parse_score_answer

@abstractmethod
def parse_score_answer(self, states: List[Dict], texts: List[str]) -> List[float]:
    """
    Parse the response from the language model for a score prompt.

    :param states: The thought states used to generate the prompt.
    :type states: List[Dict]
    :param texts: The responses to the prompt from the language model.
    :type texts: List[str]
    :return: The scores for the thought states.
    :rtype: List[float]
    """
    pass

Input / Output

Method	Input	Output
parse_aggregation_answer	states: List[Dict] -- original thought states; texts: List[str] -- LLM responses	Union[Dict, List[Dict]] -- merged thought state(s)
parse_improve_answer	state: Dict -- original thought state; texts: List[str] -- LLM responses	Dict -- refined thought state
parse_generate_answer	state: Dict -- original thought state; texts: List[str] -- LLM responses	List[Dict] -- new thought states (one per branch)
parse_validation_answer	state: Dict -- original thought state; texts: List[str] -- LLM responses	bool -- whether the thought is valid
parse_score_answer	states: List[Dict] -- thought states being scored; texts: List[str] -- LLM responses	List[float] -- scores for each state

Design Notes

• Every parse method receives both the original state(s) and the LLM response texts. This allows parsers to compare the output against the input when needed (e.g., verifying that a sorted list contains the same elements as the original).
• The texts parameter is always a List[str] because the LLM may return multiple responses per query (controlled by num_responses in the language model).
• parse_aggregation_answer returns Union[Dict, List[Dict]] to support both single-output and multi-output aggregation strategies.
• parse_validation_answer returns a simple bool, making it suitable for gating operations in the graph (e.g., keep-or-discard decisions).
• parse_score_answer returns List[float], providing one score per thought state, which can be used by KeepBestN or other selection operations.

Example: Subclassing Parser

from graph_of_thoughts.parser import Parser
from typing import Dict, List, Union
import json

class SortingParser(Parser):
    """Concrete parser for the sorting task."""

    def parse_generate_answer(self, state: Dict, texts: List[str]) -> List[Dict]:
        new_states = []
        for text in texts:
            # Extract the list from the LLM response
            sorted_list = json.loads(text.strip())
            new_states.append({
                "current": sorted_list,
                "original": state.get("original", state.get("current")),
            })
        return new_states

    def parse_aggregation_answer(
        self, states: List[Dict], texts: List[str]
    ) -> Union[Dict, List[Dict]]:
        merged = json.loads(texts[0].strip())
        return {"current": merged, "original": states[0].get("original")}

    def parse_improve_answer(self, state: Dict, texts: List[str]) -> Dict:
        improved = json.loads(texts[0].strip())
        return {"current": improved, "original": state.get("original")}

    def parse_validation_answer(self, state: Dict, texts: List[str]) -> bool:
        return texts[0].strip().lower().startswith("yes")

    def parse_score_answer(self, states: List[Dict], texts: List[str]) -> List[float]:
        return [float(t.strip()) for t in texts]

Related Pages

Implements

• Principle:Spcl_Graph_of_thoughts_LLM_Response_Parsing

Complementary Implementation

• Implementation:Spcl_Graph_of_thoughts_Prompter_ABC

Concrete Implementations (Examples)

• SortingParser in examples/sorting/sorting_032.py
• KeywordCountingParser in examples/keyword_counting/
• DocMergeParser in examples/doc_merge/
• SetIntersectionParser in examples/set_intersection/