Implementation:Spcl Graph of thoughts DocMergeParser

Knowledge Sources	Graph of Thoughts Graph of Thoughts: Solving Elaborate Problems with Large Language Models
Domains	Response_Parsing, Document_Merging
Source File	examples/doc_merge/doc_merge.py, Lines 301-483
Superclass	graph_of_thoughts.parser.Parser (ABC)
External Dependencies	re (regex), statistics.fmean (float mean)
Implements Principle	Principle:Spcl_Graph_of_thoughts_Document_Merging_Response_Parsing
Last Updated	2026-02-14

Overview

DocMergeParser is a domain-specific parser that extracts merged NDA documents from XML-tagged LLM responses and computes F1 quality scores from LLM-provided redundancy and retention ratings. It is the only parser in the GoT framework that implements parse_score_answer with actual logic (other parsers leave it unimplemented because they use programmatic scoring). The class subclasses the abstract Parser base class and is defined in the document merge example file.

Description

The DocMergeParser manages a response cache (self.cache) and provides a versatile tag-extraction helper method strip_answer_helper. It handles three distinct output formats: merged document text (free-form), aggregated documents with parts tracking, and numeric quality scores from XML tags.

Code Reference

Helper Method

def strip_answer_helper(self, text: str, tag: str = "") -> str:
    """
    Extracts content from between XML-style tags in LLM response.
    1. Strip whitespace.
    2. If "Output:" present, take text after it.
    3. If tag is specified, find last <tag> and </tag> occurrences.
    4. Extract content between them.
    5. Handle edge cases: only start tag, only end tag, or neither.
    Returns extracted text string.
    """
    text = text.strip()
    if "Output:" in text:
        text = text[text.index("Output:") + len("Output:"):].strip()
    if tag != "":
        start = text.rfind(f"<{tag}>")
        end = text.rfind(f"</{tag}>")
        if start != -1 and end != -1:
            text = text[start + len(f"<{tag}>") : end].strip()
        elif start != -1:
            text = text[start + len(f"<{tag}>"):].strip()
        elif end != -1:
            text = text[:end].strip()
    return text

Key Methods

class DocMergeParser(parser.Parser):
    def __init__(self) -> None:
        self.cache = {}

    def parse_generate_answer(self, state: Dict, texts: List[str]) -> List[Dict]:
        """
        Extracts merged NDA text from <Merged> tags.
        Creates one new state per LLM response text with 'current' set to extracted text.
        """

    def parse_aggregation_answer(self, states: List[Dict], texts: List[str]) -> Union[Dict, List[Dict]]:
        """
        Two aggregation modes:
        1. Subpart: Extracts from <Merged> tags, unions 'parts' sets from all input states.
        2. Full: Extracts from <Merged> tags, copies state as-is.
        Determines mode by comparing len(parts) to len(documents).
        """

    def parse_score_answer(self, states: List[Dict], texts: List[str]) -> List[float]:
        """
        THE UNIQUE SCORING PARSER. Processes multiple LLM response texts:
        1. Extract <Redundancy> tag content, find float via regex r'\d+\.?\d*'
        2. Extract <Retained> tag content, find float via regex
        3. Compute mean of all redundancy scores and mean of all retention scores
        4. Calculate F1 = 2 * mean_redundancy * mean_retain / (mean_redundancy + mean_retain)
        5. Return [f1] as a single-element list
        Asserts exactly 1 input state. Returns [0.0] if no valid scores found.
        """

    def parse_improve_answer(self, state: Dict, texts: List[str]) -> Dict:
        """Not implemented (returns None)."""

    def parse_validation_answer(self, state: Dict, texts: List[str]) -> bool:
        """Not implemented (returns None)."""

Detailed parse_score_answer Logic

def parse_score_answer(self, states, texts):
    assert len(states) == 1, "Only one state is allowed for scoring."
    redundancy_scores = []
    retain_scores = []
    for text in texts:
        # Extract redundancy
        answer = self.strip_answer_helper(text, "Redundancy")
        res = re.findall(r"\d+\.?\d*", answer)
        if len(res) == 1:
            redundancy_scores.append(float(res[0]))
        elif len(res) > 1:
            redundancy_scores.append(float(res[-1]))  # use last

        # Extract retention
        answer = self.strip_answer_helper(text, "Retained")
        res = re.findall(r"\d+\.?\d*", answer)
        if len(res) == 1:
            retain_scores.append(float(res[0]))
        elif len(res) > 1:
            retain_scores.append(float(res[-1]))  # use last

    if len(redundancy_scores) == 0 or len(retain_scores) == 0:
        return [0.0]
    mean_redundancy = fmean(redundancy_scores)
    mean_retain = fmean(retain_scores)
    f1 = 2 * mean_redundancy * mean_retain / (mean_redundancy + mean_retain)
    return [f1]

I/O Contract

Input

Parameter	Type	Description
state	Dict	Current thought state with keys: documents, parts, current, method
states	List[Dict]	For aggregation: multiple states with partial merges. For scoring: exactly 1 state.
texts	List[str]	Raw LLM response strings with XML tags

Output

Method	Return Type	Description
parse_generate_answer	List[Dict]	States with current set to extracted merged NDA text
parse_aggregation_answer	List[Dict]	States with merged text and unioned parts sets
parse_score_answer	List[float]	Single-element list with F1 score (0.0 to 10.0 scale)
parse_improve_answer	Dict	Not implemented (returns None)
parse_validation_answer	bool	Not implemented (returns None)

Usage Examples

Parsing a Merged Document Response

parser = DocMergeParser()
state = {
    "documents": ["NDA 1...", "NDA 2...", "NDA 3...", "NDA 4..."],
    "parts": set(),
    "current": "",
    "method": "io",
}
texts = ["Here is the merged NDA:\n<Merged>\nThis Non-Disclosure Agreement...\n</Merged>"]
new_states = parser.parse_generate_answer(state, texts)
# Returns [{"current": "This Non-Disclosure Agreement...", ...}]

Parsing LLM Quality Scores

parser = DocMergeParser()
states = [{
    "documents": ["NDA 1...", "NDA 2..."],
    "parts": set(),
    "current": "Merged NDA text...",
}]
texts = [
    "The merged NDA retains most information but has some redundancy.\n<Redundancy>7</Redundancy>\n<Retained>8</Retained>",
    "Good coverage with minor overlap.\n<Redundancy>6</Redundancy>\n<Retained>9</Retained>",
    "Well merged.\n<Redundancy>8</Redundancy>\n<Retained>7</Retained>",
]
scores = parser.parse_score_answer(states, texts)
# mean_redundancy = fmean([7, 6, 8]) = 7.0
# mean_retain = fmean([8, 9, 7]) = 8.0
# F1 = 2 * 7.0 * 8.0 / (7.0 + 8.0) = 7.467
# Returns [7.467]

Parsing a Subpart Aggregation Response

parser = DocMergeParser()
states = [
    {"documents": ["NDA1", "NDA2", "NDA3", "NDA4"], "parts": {0, 1}, "current": "Merge of 1+2..."},
    {"documents": ["NDA1", "NDA2", "NDA3", "NDA4"], "parts": {2, 3}, "current": "Merge of 3+4..."},
]
texts = ["<Merged>Combined NDA covering all four agreements...</Merged>"]
new_states = parser.parse_aggregation_answer(states, texts)
# Returns [{"current": "Combined NDA covering all four agreements...", "parts": {0, 1, 2, 3}, ...}]

Related Pages

• Principle:Spcl_Graph_of_thoughts_Document_Merging_Response_Parsing -- Design principle behind this parser
• Implementation:Spcl_Graph_of_thoughts_DocMergePrompter -- Companion prompter for generating prompts parsed here
• Workflow:Spcl_Graph_of_thoughts_GoT_Document_Merging_Pipeline -- End-to-end pipeline using this parser