class TransformQueryEngine(BaseQueryEngine):
"""Transform query engine.
Applies a query transform to a query bundle before passing
it to a query engine.
Args:
query_engine (BaseQueryEngine): A query engine object.
query_transform (BaseQueryTransform): A query transform object.
transform_metadata (Optional[dict]): metadata to pass to the
query transform.
callback_manager (Optional[CallbackManager]): A callback manager.
"""
def __init__(
self,
query_engine: BaseQueryEngine,
query_transform: BaseQueryTransform,
transform_metadata: Optional[dict] = None,
callback_manager: Optional[CallbackManager] = None,
) -> None:
self._query_engine = query_engine
self._query_transform = query_transform
self._transform_metadata = transform_metadata
super().__init__(callback_manager)
def _get_prompt_modules(self) -> PromptMixinType:
"""Get prompt sub-modules."""
return {
"query_transform": self._query_transform,
"query_engine": self._query_engine,
}
def retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
query_bundle = self._query_transform.run(
query_bundle, metadata=self._transform_metadata
)
return self._query_engine.retrieve(query_bundle)
def synthesize(
self,
query_bundle: QueryBundle,
nodes: List[NodeWithScore],
additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
) -> RESPONSE_TYPE:
query_bundle = self._query_transform.run(
query_bundle, metadata=self._transform_metadata
)
return self._query_engine.synthesize(
query_bundle=query_bundle,
nodes=nodes,
additional_source_nodes=additional_source_nodes,
)
async def asynthesize(
self,
query_bundle: QueryBundle,
nodes: List[NodeWithScore],
additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
) -> RESPONSE_TYPE:
query_bundle = self._query_transform.run(
query_bundle, metadata=self._transform_metadata
)
return await self._query_engine.asynthesize(
query_bundle=query_bundle,
nodes=nodes,
additional_source_nodes=additional_source_nodes,
)
def _query(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
"""Answer a query."""
query_bundle = self._query_transform.run(
query_bundle, metadata=self._transform_metadata
)
return self._query_engine.query(query_bundle)
async def _aquery(self, query_bundle: QueryBundle) -> RESPONSE_TYPE:
"""Answer a query."""
query_bundle = self._query_transform.run(
query_bundle, metadata=self._transform_metadata
)
return await self._query_engine.aquery(query_bundle)