ZerdaZerda

BaseOutputParser & wrappers (base_parser.py)

On this page

Overview

base_parser.py defines the shared abstract interface for the single-format parser toolkit (JSONOutputParser, YAMLOutputParser, CSVOutputParser, ListOutputParser, StructuredOutputParser; PydanticOutputParser also conforms via inheritance from BaseOutputParser). It also provides two composable wrapper classes, AutoFixingParser and ChainedParser, that add retry/fallback behavior on top of any BaseOutputParser without modifying the underlying parser.

Depends only on the standard library (asyncio, logging, abc, typing).

BaseOutputParser

Python
class BaseOutputParser(ABC):

Abstract base class. Every subclass must implement parse() and get_format_instructions().

Abstract Methods

parse

Python · 2 lines
@abstractmethod
def parse(self, text: str) -> Any

Converts raw LLM text into structured output. Raises: ParsingError (by convention — the abstract method itself doesn't raise anything, this is the documented contract subclasses follow).

get_format_instructions

Python · 2 lines
@abstractmethod
def get_format_instructions(self) -> str

Returns a string of formatting instructions suitable for embedding in an LLM prompt.

Concrete Methods (inherited by all subclasses)

async_parse

Python
async def async_parse(self, text: str) -> Any

Runs self.parse(text) in a thread-pool executor: await asyncio.get_event_loop().run_in_executor(None, self.parse, text). Makes any synchronous parser awaitable from async code without blocking the event loop, but does not make parse() itself faster or non-blocking in absolute terms — the work still happens synchronously on a worker thread.

parse_with_prompt

Python
def parse_with_prompt(self, text: str) -> tuple[Any, str]

Convenience method returning (self.parse(text), self.get_format_instructions()).

parse_or_default

Python
def parse_or_default(self, text: str, default: Any = None) -> Any

Calls self.parse(text); on any exception (ParsingError or otherwise — the except clause catches (ParsingError, Exception), which in practice is equivalent to catching just Exception since ParsingError is itself an Exception subclass), logs a warning and returns default instead of propagating.

Python
result = parser.parse_or_default(possibly_bad_text, default={"answer": ""})

__repr__

Returns f"{self.__class__.__name__}()".

ParsingError

Python · 2 lines
class ParsingError(Exception):
    def __init__(self, message: str, original_text: Optional[str] = None, cause: Optional[Exception] = None)

The exception type raised by every single-format parser in the toolkit on parse failure.

Attributes: original_text (the text that failed to parse), cause (the underlying exception, if any).

__str__: returns the base message, plus "\nCaused by: {cause}" if cause is set, plus "\nOriginal text preview: {preview}" if original_text is set — the preview truncates to the first 200 characters with "..." appended if longer.

Python · 7 lines
try:
    parser.parse(bad_text)
except ParsingError as e:
    print(str(e))
    # Invalid JSON: Expecting value: line 1 column 1 (char 0)
    # Caused by: Expecting value: line 1 column 1 (char 0)
    # Original text preview: this is not json

AutoFixingParser

Python · 7 lines
class AutoFixingParser(BaseOutputParser):
    def __init__(
        self,
        parser: BaseOutputParser,
        fixer: Optional[Callable[[str, str], str]] = None,
        max_retries: int = 2,
    )

Wraps another BaseOutputParser with automatic retry + fix logic: on parse failure, calls a fixer(bad_text, format_instructions) -> fixed_text callback (typically another LLM call) and retries.

Parameters

  • parser: BaseOutputParser — the underlying parser to wrap.
  • fixer: Optional[Callable[[str, str], str]] — receives (current_text, format_instructions), returns corrected text. If None, no fixing occurs between retries (retries just re-attempt the same text, which is unlikely to succeed differently).
  • max_retries: int — default 2.

parse

Python
def parse(self, text: str) -> Any

Loops self.max_retries + 1 times total (i.e. the initial attempt plus max_retries retries). On each failure (before the last attempt), if a fixer is set, calls it to get corrected text for the next attempt; if the fixer itself raises, logs a warning and breaks out of the retry loop early. If all attempts are exhausted, raises ParsingError(f"Failed after {self.max_retries + 1} attempts.", original_text=text, cause=last_error).

Python · 6 lines
def simple_fixer(bad_text: str, instructions: str) -> str:
    return call_llm(f"Fix this to match:\n{instructions}\n\nBroken output:\n{bad_text}")

base_parser = JSONOutputParser()
parser = AutoFixingParser(base_parser, fixer=simple_fixer, max_retries=2)
result = parser.parse(raw_llm_output)

Raises: ParsingError if every attempt fails.

Side effects: calls self.fixer (potentially an LLM call) between attempts; logs at info/warning.

get_format_instructions

Delegates to self.parser.get_format_instructions().

ChainedParser

Python · 2 lines
class ChainedParser(BaseOutputParser):
    def __init__(self, parsers: list[BaseOutputParser])

Tries multiple parsers in order, returning the first successful result — useful when the LLM might return output in one of several formats.

Raises: ValueError in __init__ if parsers is empty.

parse

Python
def parse(self, text: str) -> Any

Tries each parser in self.parsers in order; returns the first result that doesn't raise. If all fail, raises ParsingError("All parsers in chain failed:\n" + joined per-parser error messages, original_text=text).

Python · 2 lines
parser = ChainedParser([JSONOutputParser(), ListOutputParser()])
result = parser.parse(text)  # tries JSON first, then list

get_format_instructions

Returns a combined instructions string listing every wrapped parser's instructions under an [Option N] heading:

code · 5 lines
Accept any of the following formats:

[Option 1] Return ONLY valid JSON...

[Option 2] Return the result as a numbered list...

Thread Safety

BaseOutputParser subclasses generally hold only immutable configuration (set at __init__) and are safe to share across threads for concurrent parse() calls, as long as the underlying parse() implementation itself doesn't mutate shared state — none of the toolkit parsers in this package do. AutoFixingParser/ChainedParser add no additional mutable state beyond what they wrap, so the same guarantee extends to them (again, contingent on any fixer callable you supply being thread-safe if used concurrently).

Async: async_parse() is safe to call concurrently from multiple coroutines — each call gets its own executor task.

Best practices

  • Use ChainedParser when a single prompt might reasonably produce one of several formats (e.g. either a JSON object or a numbered list) rather than trying to force one format via increasingly elaborate prompt engineering.
  • Use AutoFixingParser when you have a cheap, reliable fixer callback (even a rule-based one, not necessarily an LLM call) — it composes with any BaseOutputParser, including ones from this toolkit or your own custom subclasses.

Common mistakes

  • Constructing AutoFixingParser without a fixer and expecting retries to help — without a fixer, each retry re-parses the same unmodified text, which will fail identically every time (the loop still runs max_retries + 1 times, but every attempt after the first is guaranteed to fail the same way unless parse() itself has nondeterministic behavior, which none of the built-in parsers do).
  • Forgetting that ChainedParser.get_format_instructions() describes all options to the LLM — if you want the LLM to strongly prefer one format, don't rely on ChainedParser's combined instructions; put your preferred format first in your own prompt engineering and use ChainedParser purely as a parsing-side fallback mechanism.

Version History

Not determinable from source.

Related pages