hbllmutils.response.parsable

Parsable LLM task functionality with automatic retry mechanism for output parsing.

This module provides parsable LLM task functionality with an automatic retry mechanism for output parsing. It extends the base LLM task to support parsing of model outputs with configurable retry logic when parsing fails. The module includes exception handling for parse failures and tracking of all retry attempts.

The module is designed to handle scenarios where LLM outputs need to be parsed into specific formats (such as JSON, XML, or structured data), with automatic retry when parsing fails due to malformed or unexpected output. This is particularly useful when working with LLMs that may occasionally produce outputs that do not conform to the expected format.

The module contains the following main components:

• OutputParseWithException - Data class for storing failed parse attempts

• OutputParseFailed - Exception raised when all parsing attempts fail

• ParsableLLMTask - LLM task with automatic output parsing and retry logic

Architecture:

The module uses a template method pattern where subclasses implement the ParsableLLMTask._parse_and_validate() method to define their specific parsing logic. The base class handles the retry mechanism, exception tracking, and logging automatically.

Retry Mechanism:

When parsing fails, the task will:

1. Log the parsing error with attempt count

2. Store the failed output and exception for debugging

3. Request a new response from the model

4. Attempt to parse the new response

5. Repeat until success or max retries reached

If all retries are exhausted, an OutputParseFailed exception is raised containing all failed attempts for comprehensive debugging.

Note

The retry mechanism sends new requests to the LLM for each failed parse attempt, which may incur additional API costs and latency. Consider setting appropriate max_retries values based on your use case.

Warning

Ensure that your parsing logic in ParsableLLMTask._parse_and_validate() raises exceptions that match the types specified in ParsableLLMTask.__exceptions__. Other exception types will propagate immediately without retry.

Example:

>>> import json
>>> from hbllmutils.model import LLMModel
>>> from hbllmutils.response import ParsableLLMTask, extract_code, parse_json
>>>
>>> class JSONParsableTask(ParsableLLMTask):
...     '''A task that parses JSON responses from the model.'''
...     __exceptions__ = (json.JSONDecodeError, KeyError)
...
...     def _parse_and_validate(self, content: str):
...         # Extract code block and parse JSON
...         data = parse_json(extract_code(content))
...
...         # Validate required fields
...         if 'answer' not in data:
...             raise KeyError("Missing 'answer' field")
...
...         return data
>>>
>>> # Initialize model and task
>>> model = LLMModel(...)
>>> task = JSONParsableTask(model, default_max_retries=3)
>>>
>>> # Ask question with automatic parsing and retry
>>> result = task.ask_then_parse(
...     input_content="What is the capital of France? Answer in JSON with 'answer' key",
...     max_retries=5
... )
>>> print(result['answer'])
Paris
>>>
>>> # Handle parsing failures
>>> try:
...     result = task.ask_then_parse("Invalid request")
... except OutputParseFailed as e:
...     print(f"Failed after {len(e.tries)} attempts")
...     for i, attempt in enumerate(e.tries):
...         print(f"Attempt {i+1}: {attempt.exception}")

OutputParseWithException

class hbllmutils.response.parsable.OutputParseWithException(output: str, exception: Exception)

Data class to store a failed parse attempt with its output and exception.

This class encapsulates information about a single failed parsing attempt, including both the raw output that failed to parse and the exception that was raised during the parsing process. It is used by OutputParseFailed to provide comprehensive debugging information about all failed attempts.

The class is immutable (frozen dataclass) to ensure that stored failure information cannot be accidentally modified.

Variables:

• output (str) – The raw output string from the model that failed to parse.

• exception (Exception) – The exception that occurred during parsing attempt.

Example:

>>> import json
>>> attempt = OutputParseWithException(
...     output='{"invalid": json}',
...     exception=json.JSONDecodeError("Expecting ',' delimiter", "", 0)
... )
>>> print(attempt.output)
{"invalid": json}
>>> print(type(attempt.exception))
<class 'json.decoder.JSONDecodeError'>
>>>
>>> # Used in debugging failed parsing attempts
>>> for attempt in failed_attempts:
...     print(f"Output: {attempt.output[:50]}...")
...     print(f"Error: {attempt.exception}")

OutputParseFailed

class hbllmutils.response.parsable.OutputParseFailed(message: str, tries: List[OutputParseWithException])

Exception raised when output parsing fails after all retry attempts.

This exception is raised by ParsableLLMTask.ask_then_parse() when the task exhausts all retry attempts without successfully parsing the model’s output. It contains comprehensive information about all failed attempts, including the raw outputs and exceptions, which is invaluable for debugging parsing issues.

The exception message includes the total number of failed attempts, and the tries attribute provides detailed information about each individual failure for in-depth analysis.

Variables:

tries (List[OutputParseWithException]) – List of all failed parse attempts with their outputs and exceptions. Each element is an OutputParseWithException instance containing the raw output and the exception that occurred.

Example:

>>> import json
>>> tries = [
...     OutputParseWithException(
...         output='{"incomplete": ',
...         exception=json.JSONDecodeError("Expecting value", "", 15)
...     ),
...     OutputParseWithException(
...         output='not json at all',
...         exception=json.JSONDecodeError("Expecting value", "", 0)
...     )
... ]
>>> exc = OutputParseFailed("Parsing failed after 2 tries", tries)
>>> print(str(exc))
Parsing failed after 2 tries
>>> print(len(exc.tries))
2
>>>
>>> # Accessing detailed failure information
>>> for i, attempt in enumerate(exc.tries, 1):
...     print(f"Attempt {i}:")
...     print(f"  Output: {attempt.output}")
...     print(f"  Error: {attempt.exception}")
Attempt 1:
  Output: {"incomplete":
  Error: Expecting value: line 1 column 15 (char 15)
Attempt 2:
  Output: not json at all
  Error: Expecting value: line 1 column 0 (char 0)
>>>
>>> # Handling in try-except block
>>> try:
...     result = task.ask_then_parse("some input")
... except OutputParseFailed as e:
...     print(f"All {len(e.tries)} parsing attempts failed")
...     # Log or analyze failures
...     for attempt in e.tries:
...         logger.error(f"Failed output: {attempt.output}")

__init__(message: str, tries: List[OutputParseWithException])

Initialize the OutputParseFailed exception.

Parameters:

• message (str) – The error message describing the failure. Typically includes the total number of failed attempts.

• tries (List[OutputParseWithException]) – List of all failed parse attempts. Each element should be an OutputParseWithException instance containing the output and exception from that attempt.

Example:

>>> tries = [
...     OutputParseWithException("bad output 1", ValueError("error 1")),
...     OutputParseWithException("bad output 2", ValueError("error 2"))
... ]
>>> exc = OutputParseFailed("Parsing failed after 2 tries", tries)
>>> raise exc
Traceback (most recent call last):
    ...
OutputParseFailed: Parsing failed after 2 tries

ParsableLLMTask

class hbllmutils.response.parsable.ParsableLLMTask(model: str | LLMModel, history: LLMHistory | None = None, default_max_retries: int = 5)

An LLM task that supports automatic parsing of model outputs with retry mechanism.

This class extends LLMTask to provide automatic parsing of model outputs with configurable retry logic. When parsing fails, it will retry up to a maximum number of times before raising an OutputParseFailed exception. This is useful when the model’s output needs to be parsed into a specific format (e.g., JSON, XML, structured data) and the model may occasionally produce malformed output.

The class uses a template method pattern where subclasses implement the _parse_and_validate() method to define their specific parsing logic. The base class handles the retry mechanism, exception tracking, and logging automatically.

Workflow:

1. Send request to model (optionally with new input content)

2. Receive raw text response from model

3. Attempt to parse response using _parse_and_validate()

4. If parsing succeeds, return parsed result

5. If parsing fails with an exception in __exceptions__:

   • Log the failure with attempt count

   • Store the failed output and exception

   • Send a new request to the model

   • Repeat from step 2

6. If max retries reached, raise OutputParseFailed with all attempts

Variables:

• __exceptions__ (Union[Type[Exception], Tuple[Type[Exception], ...]]) – Exception types to catch during parsing attempts. Can be a single exception type or a tuple of exception types. Only exceptions matching these types will trigger a retry; other exceptions will propagate immediately. Defaults to Exception (catches all).

• default_max_retries (int) – Default maximum number of retry attempts for parsing. Used when max_retries is not specified in ask_then_parse().

Note

Each retry sends a new request to the LLM, which may incur additional API costs and increase response time. Set appropriate max_retries values based on your use case and budget constraints.

Warning

Ensure that _parse_and_validate() raises exceptions that match the types specified in __exceptions__. Other exception types will not trigger retries and will propagate immediately.

Example:

>>> import json
>>> from hbllmutils.model import LLMModel
>>> from hbllmutils.response import ParsableLLMTask, extract_code, parse_json
>>>
>>> class JSONParsableTask(ParsableLLMTask):
...     '''Task that parses JSON responses with validation.'''
...     __exceptions__ = (json.JSONDecodeError, KeyError, ValueError)
...
...     def _parse_and_validate(self, content: str):
...         # Extract code block if present
...         data = parse_json(extract_code(content))
...
...         # Validate structure
...         if 'result' not in data:
...             raise KeyError("Missing 'result' field")
...         if not isinstance(data['result'], (int, float)):
...             raise ValueError("Result must be numeric")
...
...         return data['result']
>>>
>>> # Initialize with custom default retries
>>> model = LLMModel(...)
>>> task = JSONParsableTask(model, default_max_retries=3)
>>>
>>> # Simple usage with default retries
>>> result = task.ask_then_parse(input_content="Calculate 2+2")
>>> print(result)
4
>>>
>>> # Usage with custom max_retries for specific request
>>> result = task.ask_then_parse(
...     input_content="What is 10*5?",
...     max_retries=10,
...     temperature=0.7
... )
>>> print(result)
50
>>>
>>> # Handling parse failures
>>> try:
...     result = task.ask_then_parse("Invalid request")
... except OutputParseFailed as e:
...     print(f"Failed after {len(e.tries)} attempts")
...     for i, attempt in enumerate(e.tries, 1):
...         print(f"Attempt {i}: {attempt.exception}")

__exceptions__

alias of Exception

__init__(model: str | LLMModel, history: LLMHistory | None = None, default_max_retries: int = 5)

Initialize the ParsableLLMTask.

Parameters:

• model (LLMModelTyping) – The LLM model to use for generating responses. Can be a model name string, an LLMModel instance, or None to load the default model from configuration.

• history (Optional[LLMHistory]) – Optional conversation history. If None, a new empty history will be created. The history tracks the conversation context across multiple interactions.

• default_max_retries (int) – Default maximum number of retry attempts for parsing. Must be a positive integer. This value is used when max_retries is not specified in ask_then_parse(). Defaults to 5.

Raises:

ValueError – If default_max_retries is not a positive integer.

Example:

>>> from hbllmutils.model import LLMModel
>>> from hbllmutils.history import LLMHistory
>>>
>>> # Simple initialization with defaults
>>> model = LLMModel(...)
>>> task = ParsableLLMTask(model)
>>> print(task.default_max_retries)
5
>>>
>>> # Initialize with custom default retries
>>> task = ParsableLLMTask(model, default_max_retries=10)
>>> print(task.default_max_retries)
10
>>>
>>> # Initialize with existing history
>>> history = LLMHistory().with_system_prompt("You are a helpful assistant.")
>>> task = ParsableLLMTask(model, history=history, default_max_retries=3)
>>> len(task.history)
1

ask_then_parse(input_content: str | None = None, max_retries: int | None = None, **params: Any) → Any

Ask the model a question and parse the response with automatic retry on parse failure.

This method will repeatedly ask the model and attempt to parse the output until either parsing succeeds or the maximum number of retries is reached. Each failed attempt is logged and tracked. If all retries fail, an OutputParseFailed exception is raised containing all failed attempts for debugging.

The method uses _parse_and_validate() to parse outputs and will catch exceptions specified in __exceptions__. Other exceptions will propagate immediately without retry.

Workflow:

1. Preprocess input content using _preprocess_input_content()

2. Send request to model using ask()

3. Attempt to parse response using _parse_and_validate()

4. On success: return parsed result

5. On failure (matching __exceptions__):

   • Log warning with attempt count

   • Store failed output and exception

   • Increment retry counter

   • Repeat from step 2 if retries remain

6. If max retries exhausted: raise OutputParseFailed

Parameters:

• input_content (Optional[str]) – Optional user input content to add to the history before asking. If None, uses the existing history without modification. The content will be preprocessed by _preprocess_input_content().

• max_retries (Optional[int]) – Maximum number of retry attempts. If None, uses default_max_retries. Must be a positive integer if provided. Each retry sends a new request to the model.

• params (dict) – Additional parameters to pass to the ask() method. Common parameters include temperature, max_tokens, top_p, etc., depending on the model.

Returns:

The successfully parsed output from the model. The return type depends on the implementation of _parse_and_validate().

Return type:

Any

Raises:

• ValueError – If max_retries is provided and is not a positive integer.

• OutputParseFailed – If parsing fails after all retry attempts. The exception contains all failed attempts in its tries attribute, each with the raw output and exception for debugging.

• Exception – Any exception not matching __exceptions__ will propagate immediately without retry.

Note

Each retry sends a new request to the LLM, which may incur additional API costs and increase total response time. Monitor your usage and set appropriate retry limits.

Warning

The method does not modify the conversation history with failed attempts. Only successful interactions are recorded in the history.

Example:

>>> import json
>>> class NumberTask(ParsableLLMTask):
...     '''Task that parses numeric responses.'''
...     __exceptions__ = (ValueError, TypeError)
...
...     def _parse_and_validate(self, content: str):
...         value = float(content.strip())
...         if value < 0:
...             raise ValueError("Value must be non-negative")
...         return value
>>>
>>> task = NumberTask(model)
>>>
>>> # Simple usage with default retries
>>> result = task.ask_then_parse(input_content="What is 2+2?")
>>> print(result)
4.0
>>>
>>> # Usage with custom max_retries and model parameters
>>> result = task.ask_then_parse(
...     input_content="Calculate 10*5",
...     max_retries=3,
...     temperature=0.7,
...     max_tokens=100
... )
>>> print(result)
50.0
>>>
>>> # Handling parse failures
>>> try:
...     result = task.ask_then_parse(
...         input_content="Invalid request",
...         max_retries=2
...     )
... except OutputParseFailed as e:
...     print(f"Failed after {len(e.tries)} attempts")
...     for i, attempt in enumerate(e.tries, 1):
...         print(f"Attempt {i}:")
...         print(f"  Output: {attempt.output[:50]}...")
...         print(f"  Error: {attempt.exception}")
Failed after 2 attempts
Attempt 1:
  Output: I'm not sure what you're asking...
  Error: could not convert string to float: "I'm not sure what you're asking"
Attempt 2:
  Output: Please clarify your question...
  Error: could not convert string to float: "Please clarify your question"
>>>
>>> # Example with JSON parsing
>>> class JSONTask(ParsableLLMTask):
...     __exceptions__ = (json.JSONDecodeError, KeyError)
...
...     def _parse_and_validate(self, content: str):
...         data = json.loads(content)
...         if 'answer' not in data:
...             raise KeyError("Missing 'answer' field")
...         return data['answer']
>>>
>>> task = JSONTask(model)
>>> result = task.ask_then_parse(
...     input_content="What is the capital of France? Answer in JSON",
...     max_retries=5
... )
>>> print(result)
Paris