Query transformer

Defines a base class for query transformers used in Gen AI applications.

Authors

Dimitrij Ray (dimitrij.ray@gdplabs.id) Henry Wicaksono (henry.wicaksono@gdplabs.id)

References

NONE

BaseQueryTransformer(lm_request_processor, extract_func=None, on_error=ErrorHandling.RAISE)

Bases: Component, UsesLM, ABC

An abstract base class for the query transformers used in Gen AI applications.

Using the implementations of this class, users can transform a query into a list of strings. Each query transformer comes with a default extractor function that extracts the query from the LLM output. Users can also supply their own extractor function to customize the extraction process. A JSON extractor function is also provided for convenience.

See the usage examples below for more details.

Attributes:

Name	Type	Description
lm_request_processor	LMRequestProcessor | None	The LMRequestProcessor instance to be used for transforming queries. Can be None if the transformer does not require an LMRequestProcessor, but it has to be supplied.
extract_func	Callable[[str | list[str] | dict[str, str | list[str]]], str | list[str]]	A function to extract the transformed query from the LM output.
on_error	ErrorHandling	The error handling strategy to use when an exception occurs during query transformation. Defaults to ErrorHandling.RAISE.

Usage Examples

Using the original constructor

transformer = ConcreteQueryTransformer(lm_request_processor=None)

Using the from_lm_components constructor with a custom JSON extractor

transformer = ConcreteQueryTransformer.from_lm_components( prompt_builder, lm_invoker, output_parser, extract_func=BaseQueryTransformer.json_extractor("query") )

Initialize the BaseQueryTransformer.

Parameters:

Name	Type	Description	Default
lm_request_processor	LMRequestProcessor | None	The LMRequestProcessor instance to be used for transforming queries. Can be None if the transformer does not require an LMRequestProcessor, but it has to be supplied.	required
extract_func	Callable[[str | list[str] | dict[str, str | list[str]]], str | list[str]]	A function to extract the transformed query from the output. Defaults to None, in which case a default extractor will be used.	None
on_error	ErrorHandling	The error handling strategy to use when an exception occurs during query transformation. Defaults to ErrorHandling.RAISE.	RAISE

json_extractor(key) staticmethod

Creates a JSON extractor function for a given key.

Parameters:

Name	Type	Description	Default
key	str	The key to extract from the JSON result.	required

Returns:

Type	Description
Callable[[dict[str, Any]], str | list[str]]	Callable[[dict[str, Any]], str | list[str]]: A function that extracts the specified key from a JSON object.

Raises:

Type	Description
KeyError	If the specified key is not found in the JSON object.

transform(query) async

Transforms the given query into a list of strings, with configurable error handling.

This method wraps the core transformation logic (_transform) with error handling based on the on_error attribute.

Parameters:

Name	Type	Description	Default
query	str | list[str] | dict[str, str | list[str]]	The query, list of queries, or dictionary of queries to be transformed.	required

Returns:

Type	Description
list[str]	list[str]: A list of transformed query strings. Behavior on error depends on self.on_error: 1. "keep": Returns the original query, converted to list[str] format. 2. "empty": Returns an empty list. 3. "raise": Re-raises the exception.

Raises:

Type	Description
Exception	If self.on_error is "raise" and an error occurs during transformation.

ErrorHandling

Bases: StrEnum

Enum for error handling options in query transformation.

Attributes:

Name	Type	Description
KEEP	str	Keep the original query on error.
EMPTY	str	Return an empty list on error.
RAISE	str	Raise an exception on error.