Overview
SortingParser is a domain-specific parser that extracts sorted list results from LLM text responses and updates thought state dictionaries. It subclasses the abstract Parser base class and implements its five abstract methods. The class is defined directly in the sorting example file.
Description
The SortingParser manages a response cache (self.cache) and implements the full parser interface. Its primary responsibility is converting free-form LLM text into structured state dictionaries with updated current fields and appropriate phase transitions.
Code Reference
Key Methods
class SortingParser(parser.Parser):
def __init__(self) -> None:
self.cache = {}
def parse_generate_answer(self, state: Dict, texts: List[str]) -> List[Dict]:
"""
Two code paths based on method and phase:
1. GoT phase 0 (split): Parse JSON with "List 1"/"List 2" keys,
create one state per sublist with phase=1, part, unsorted_sublist.
2. All other: Extract sorted list from text lines containing brackets,
prefer lines with "Output", set phase=2.
Returns list of new states (one per text for non-split, multiple per text for split).
"""
def parse_aggregation_answer(self, states: List[Dict], texts: List[str]) -> Union[Dict, List[Dict]]:
"""
Parses merge results from two sorted sublists.
- Extracts list from lines containing brackets
- Merges unsorted_sublist fields from both input states
- Sorts input states by 'part' for deterministic ordering
- Returns list of new states with merged current and unsorted_sublist
Asserts exactly 2 input states.
"""
def parse_improve_answer(self, state: Dict, texts: List[str]) -> Dict:
"""Not implemented for sorting (returns None)."""
def parse_validation_answer(self, state: Dict, texts: List[str]) -> bool:
"""Not implemented for sorting (returns None)."""
def parse_score_answer(self, states: List[Dict], texts: List[str]) -> List[float]:
"""Not implemented for sorting (scoring is programmatic via utils.num_errors)."""
Detailed parse_generate_answer Logic
# GoT split path (phase 0):
text = text[text.index("{") : text.index("}") + 1]
json_dict = json.loads(text)
for key, value in json_dict.items():
new_state["current"] = str(value)
new_state["unsorted_sublist"] = str(value)
new_state["phase"] = 1
new_state["part"] = key # e.g. "List 1"
# Standard sort path:
answers = text.strip().split("\n")
answers = [a for a in answers if "[" in a and "]" in a]
# Prefer last line containing "Output"
answer = answer[answer.index("[") : answer.index("]") + 1]
new_state["current"] = answer
new_state["phase"] = 2
I/O Contract
Input
| Parameter |
Type |
Description
|
state |
Dict |
Current thought state with keys: original, current, method, phase
|
states |
List[Dict] |
For aggregation: exactly 2 states from sublists being merged
|
texts |
List[str] |
Raw LLM response strings
|
Output
| Method |
Return Type |
Description
|
parse_generate_answer |
List[Dict] |
New states with current updated to sorted list or sublists
|
parse_aggregation_answer |
List[Dict] |
New states with current updated to merged sorted list
|
parse_improve_answer |
Dict |
Not implemented (returns None)
|
parse_validation_answer |
bool |
Not implemented (returns None)
|
parse_score_answer |
List[float] |
Not implemented (scoring uses programmatic num_errors)
|
Usage Examples
Parsing a GoT Split Response
parser = SortingParser()
state = {"original": "[3, 7, 0, 2, ...]", "current": "", "method": "got", "phase": 0}
texts = ['{\n "List 1": [3, 7, 0, 2, 8, 1, 2, 2, 2, 4, 7, 8, 5, 5, 3, 9],\n "List 2": [4, 3, 5, 6, 6, 4, 4, 5, 2, 0, 9, 3, 3, 9, 2, 1]\n}']
new_states = parser.parse_generate_answer(state, texts)
# Returns 2 states, one per sublist, each with phase=1
Parsing a Sorted List Response
parser = SortingParser()
state = {"original": "[3, 7, 0, 2]", "current": "[3, 7, 0, 2]", "method": "io", "phase": 0}
texts = ["Output: [0, 2, 3, 7]"]
new_states = parser.parse_generate_answer(state, texts)
# Returns [{"current": "[0, 2, 3, 7]", "phase": 2, ...}]
Related Pages