Query transformer

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

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
supports_single_output	bool	Whether this transformer supports return_str=True in transform. Defaults to True. Set to False on transformers that produce multiple outputs by design (e.g. OneToManyQueryTransformer).
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, *, return_str=False) async

transform(query: str | list[str] | dict[str, str | list[str]], *, return_str: Literal[True]) -> str | None

transform(query: str | list[str] | dict[str, str | list[str]], *, return_str: Literal[False] = ...) -> list[str]

Transforms the given query, with configurable error handling and output format.

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
return_str	bool	If True, returns the first transformed result as a plain str (or None if the result is empty) instead of a list[str]. Only supported on X-to-One transformers (i.e. those with supports_single_output = True, e.g. OneToOneQueryTransformer, ManyToOneQueryTransformer). Defaults to False.	False

Returns:

Type	Description
list[str] | str | None	list[str]: A list of transformed query strings when return_str=False. 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.
list[str] | str | None	str | None: The first transformed query string, or None if the result is empty, when return_str=True.

Raises:

Type	Description
NotImplementedError	If return_str=True is used on a transformer where supports_single_output is False.
ValueError	If query is None or not a valid type.
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.