tessl/pypi-llama-index-core

Interface between LLMs and your data

—

Pending

Overview

Eval results

Files

Query Engines

Name: tessl/pypi-llama-index-core
Author: tessl

High-level interfaces for question-answering over indexed data. Query engines combine retrieval, processing, and response synthesis to provide comprehensive answers to user queries with support for various reasoning patterns and data sources.

Capabilities

Base Query Engine Interface

Foundation interface for all query engines, providing the core query method and response handling.

class BaseQueryEngine:
    """
    Base interface for query engines.
    
    Parameters:
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(self, callback_manager: Optional[CallbackManager] = None): ...
    
    def query(self, str_or_query_bundle: Union[str, QueryBundle]) -> RESPONSE_TYPE:
        """
        Query the engine with a string or QueryBundle.
        
        Parameters:
        - str_or_query_bundle: Union[str, QueryBundle], query input
        
        Returns:
        RESPONSE_TYPE: Query response
        """
        
    async def aquery(self, str_or_query_bundle: Union[str, QueryBundle]) -> RESPONSE_TYPE:
        """Async version of query method."""

Retriever Query Engine

Standard query engine that combines a retriever for finding relevant content with a response synthesizer for generating answers.

class RetrieverQueryEngine(BaseQueryEngine):
    """
    Query engine using retriever + response synthesizer pattern.
    
    Parameters:
    - retriever: BaseRetriever, retriever for finding relevant content
    - response_synthesizer: Optional[BaseSynthesizer], synthesizer for generating responses
    - node_postprocessors: Optional[List[BaseNodePostprocessor]], postprocessors for retrieved nodes
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        retriever: BaseRetriever,
        response_synthesizer: Optional[BaseSynthesizer] = None,
        node_postprocessors: Optional[List[BaseNodePostprocessor]] = None,
        callback_manager: Optional[CallbackManager] = None
    ): ...
    
    @classmethod
    def from_args(
        cls,
        retriever: BaseRetriever,
        response_mode: str = "compact",
        text_qa_template: Optional[BasePromptTemplate] = None,
        **kwargs
    ) -> "RetrieverQueryEngine":
        """Create query engine with specified configuration."""

Custom Query Engine

Base class for implementing custom query engine logic with full control over the query processing pipeline.

class CustomQueryEngine(BaseQueryEngine):
    """
    Base class for custom query engine implementations.
    
    Parameters:
    - query_fn: Callable, function that processes queries
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        query_fn: Callable[[Union[str, QueryBundle]], RESPONSE_TYPE],
        callback_manager: Optional[CallbackManager] = None
    ): ...

Router Query Engine

Query engine that routes queries to different sub-engines based on query content or metadata, enabling specialized handling for different query types.

class RouterQueryEngine(BaseQueryEngine):
    """
    Routes queries to different query engines based on selection criteria.
    
    Parameters:
    - selector: BaseSelector, selector for choosing query engine
    - query_engine_tools: Sequence[QueryEngineTool], available query engines
    - verbose: bool, whether to output selection reasoning
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        selector: BaseSelector,
        query_engine_tools: Sequence[QueryEngineTool],
        verbose: bool = True,
        callback_manager: Optional[CallbackManager] = None
    ): ...
    
    @classmethod
    def from_defaults(
        cls,
        query_engine_tools: Sequence[QueryEngineTool],
        selector: Optional[BaseSelector] = None,
        selector_type: str = "single",
        **kwargs
    ) -> "RouterQueryEngine":
        """Create router with default selector."""

Sub Question Query Engine

Query engine that decomposes complex queries into sub-questions, answers each independently, and synthesizes a final comprehensive response.

class SubQuestionQueryEngine(BaseQueryEngine):
    """
    Decomposes complex queries into sub-questions for comprehensive answers.
    
    Parameters:
    - question_gen: SubQuestionGenerator, generator for creating sub-questions
    - query_engine_tools: Sequence[QueryEngineTool], tools for answering sub-questions
    - response_synthesizer: Optional[BaseSynthesizer], synthesizer for final response
    - verbose: bool, whether to show sub-question processing
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        question_gen: SubQuestionGenerator,
        query_engine_tools: Sequence[QueryEngineTool],
        response_synthesizer: Optional[BaseSynthesizer] = None,
        verbose: bool = True,
        callback_manager: Optional[CallbackManager] = None
    ): ...
    
    @classmethod
    def from_defaults(
        cls,
        query_engine_tools: Sequence[QueryEngineTool],
        question_gen: Optional[SubQuestionGenerator] = None,
        **kwargs
    ) -> "SubQuestionQueryEngine":
        """Create sub-question query engine with defaults."""

class SubQuestionAnswerPair:
    """Container for sub-question and its answer."""
    def __init__(self, sub_q: SubQuestion, answer: str): ...

Multi-Step Query Engine

Query engine implementing multi-step reasoning, where each step builds upon previous results to handle complex analytical queries.

class MultiStepQueryEngine(BaseQueryEngine):
    """
    Implements multi-step reasoning for complex queries.
    
    Parameters:
    - query_engine: BaseQueryEngine, base query engine for individual steps
    - query_transform: BaseQueryTransform, transformation for step queries
    - index_summary: str, summary of indexed content
    - num_steps: Optional[int], maximum number of reasoning steps
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        query_transform: BaseQueryTransform,
        index_summary: str,
        num_steps: Optional[int] = None,
        callback_manager: Optional[CallbackManager] = None
    ): ...

Transform Query Engine

Query engine that applies transformations to queries before processing, enabling query enhancement, reformulation, or filtering.

class TransformQueryEngine(BaseQueryEngine):
    """
    Applies transformations to queries before processing.
    
    Parameters:
    - query_engine: BaseQueryEngine, underlying query engine
    - query_transform: BaseQueryTransform, transformation to apply
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        query_transform: BaseQueryTransform,
        callback_manager: Optional[CallbackManager] = None
    ): ...

Citation Query Engine

Query engine that provides detailed citations and source attribution for generated responses, enabling transparency and verification.

class CitationQueryEngine(BaseQueryEngine):
    """
    Query engine that provides citations for generated responses.
    
    Parameters:
    - retriever: BaseRetriever, retriever for finding relevant content
    - response_synthesizer: Optional[BaseSynthesizer], synthesizer with citation support
    - citation_chunk_size: int, size of citation chunks
    - citation_chunk_overlap: int, overlap between citation chunks
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        retriever: BaseRetriever,
        response_synthesizer: Optional[BaseSynthesizer] = None,
        citation_chunk_size: int = 512,
        citation_chunk_overlap: int = 20,
        callback_manager: Optional[CallbackManager] = None
    ): ...

Retry Query Engines

Query engines with built-in retry mechanisms for handling failures, source validation, and guideline adherence.

class RetryQueryEngine(BaseQueryEngine):
    """
    Query engine with retry logic for handling failures.
    
    Parameters:
    - query_engine: BaseQueryEngine, underlying query engine
    - evaluator: BaseEvaluator, evaluator for response quality
    - max_retries: int, maximum number of retry attempts
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        evaluator: BaseEvaluator,
        max_retries: int = 3,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class RetrySourceQueryEngine(BaseQueryEngine):
    """Query engine with retry based on source validation."""
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        evaluator: BaseEvaluator,
        max_retries: int = 3,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class RetryGuidelineQueryEngine(BaseQueryEngine):
    """Query engine with retry based on guideline adherence."""
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        guideline_evaluator: GuidelineEvaluator,
        max_retries: int = 3,
        callback_manager: Optional[CallbackManager] = None
    ): ...

SQL Query Engines

Specialized query engines for natural language querying of SQL databases and structured data.

class NLSQLTableQueryEngine(BaseQueryEngine):
    """
    Natural language to SQL query engine for database tables.
    
    Parameters:
    - sql_database: SQLDatabase, database connection
    - tables: Optional[List[str]], specific tables to query
    - context_query_kwargs: Optional[dict], context retrieval configuration
    - synthesize_response: bool, whether to synthesize natural language response
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        sql_database: SQLDatabase,
        tables: Optional[List[str]] = None,
        context_query_kwargs: Optional[dict] = None,
        synthesize_response: bool = True,
        callback_manager: Optional[CallbackManager] = None,
        **kwargs
    ): ...

class SQLTableRetrieverQueryEngine(BaseQueryEngine):
    """SQL query engine with retriever for table selection."""
    def __init__(
        self,
        sql_database: SQLDatabase,
        table_retriever: ObjectRetriever,
        context_query_kwargs: Optional[dict] = None,
        synthesize_response: bool = True,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class PGVectorSQLQueryEngine(BaseQueryEngine):
    """PostgreSQL vector query engine for hybrid search."""
    def __init__(
        self,
        sql_database: SQLDatabase,
        query_str: str,
        embed_model: Optional[BaseEmbedding] = None,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class SQLJoinQueryEngine(BaseQueryEngine):
    """SQL query engine supporting joins across multiple tables."""
    def __init__(
        self,
        sql_database: SQLDatabase,
        tables: List[str],
        callback_manager: Optional[CallbackManager] = None
    ): ...

class SQLAutoVectorQueryEngine(BaseQueryEngine):
    """Automatic SQL vector query engine with intelligent query planning."""
    def __init__(
        self,
        sql_database: SQLDatabase,
        vector_store_info: VectorStoreInfo,
        callback_manager: Optional[CallbackManager] = None
    ): ...

Pandas Query Engine

Query engine for natural language querying of pandas DataFrames and structured tabular data.

class PandasQueryEngine(BaseQueryEngine):
    """
    Natural language query engine for pandas DataFrames.
    
    Parameters:
    - df: pd.DataFrame, DataFrame to query
    - instruction_str: Optional[str], custom instructions for querying
    - output_processor: Optional[Callable], processor for query results
    - pandas_config: Optional[dict], pandas configuration options
    - head: int, number of DataFrame rows to show in context
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        df: "pd.DataFrame",
        instruction_str: Optional[str] = None,
        output_processor: Optional[Callable] = None,
        pandas_config: Optional[dict] = None,
        head: int = 5,
        callback_manager: Optional[CallbackManager] = None,
        **kwargs
    ): ...

Specialized Query Engines

Additional query engines for specific use cases and integrations.

class JSONalyzeQueryEngine(BaseQueryEngine):
    """Query engine for analyzing JSON data structures."""
    def __init__(
        self,
        json_value: Union[str, dict, List[dict]],
        callback_manager: Optional[CallbackManager] = None,
        **kwargs
    ): ...

class KnowledgeGraphQueryEngine(BaseQueryEngine):
    """Query engine for knowledge graph traversal and reasoning."""
    def __init__(
        self,
        storage_context: StorageContext,
        refresh_schema: bool = False,
        verbose: bool = True,
        callback_manager: Optional[CallbackManager] = None,
        **kwargs
    ): ...

class SimpleMultiModalQueryEngine(BaseQueryEngine):
    """Simple multi-modal query engine for text and image content."""
    def __init__(
        self,
        retriever: BaseRetriever,
        multi_modal_llm: Optional[MultiModalLLM] = None,
        text_qa_template: Optional[BasePromptTemplate] = None,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class CogniswitchQueryEngine(BaseQueryEngine):
    """Query engine integration with Cogniswitch platform."""
    def __init__(
        self,
        cs_token: str,
        OAI_token: str,
        apiKey: str,
        callback_manager: Optional[CallbackManager] = None
    ): ...

class ComposableGraphQueryEngine(BaseQueryEngine):
    """Query engine for composable graph structures."""
    def __init__(
        self,
        graph: ComposableGraph,
        custom_query_engines: Optional[Dict[str, BaseQueryEngine]] = None,
        recursive: bool = True,
        callback_manager: Optional[CallbackManager] = None
    ): ...

FLARE Query Engine

Implementation of the FLARE (Forward-Looking Active REtrieval) pattern for iterative query processing with active retrieval.

class FLAREInstructQueryEngine(BaseQueryEngine):
    """
    FLARE (Forward-Looking Active REtrieval) query engine.
    
    Parameters:
    - query_engine: BaseQueryEngine, base query engine for retrieval
    - lookahead_answer_inserter: BaseLookaheadAnswerInserter, answer insertion logic
    - callback_manager: Optional[CallbackManager], callback management
    """
    def __init__(
        self,
        query_engine: BaseQueryEngine,
        lookahead_answer_inserter: BaseLookaheadAnswerInserter,
        callback_manager: Optional[CallbackManager] = None
    ): ...

Response Synthesis Functions

Factory function for creating response synthesizers used by query engines.

def get_response_synthesizer(
    response_mode: str = "compact",
    text_qa_template: Optional[BasePromptTemplate] = None,
    refine_template: Optional[BasePromptTemplate] = None,
    summary_template: Optional[BasePromptTemplate] = None,
    simple_template: Optional[BasePromptTemplate] = None,
    use_async: bool = False,
    streaming: bool = False,
    structured_answer_filtering: bool = False,
    **kwargs
) -> BaseSynthesizer:
    """
    Factory function for creating response synthesizers.
    
    Parameters:
    - response_mode: str, synthesis mode ("compact", "refine", "tree_summarize", etc.)
    - text_qa_template: Optional[BasePromptTemplate], template for Q&A
    - refine_template: Optional[BasePromptTemplate], template for refinement
    - summary_template: Optional[BasePromptTemplate], template for summarization
    - simple_template: Optional[BasePromptTemplate], template for simple responses
    - use_async: bool, whether to use async processing
    - streaming: bool, whether to enable streaming responses
    - structured_answer_filtering: bool, whether to filter structured answers
    
    Returns:
    BaseSynthesizer: Configured response synthesizer
    """

Usage Examples

Basic Query Engine Usage

from llama_index.core import VectorStoreIndex, Document
from llama_index.core.query_engine import RetrieverQueryEngine

# Create index and documents
documents = [
    Document(text="Machine learning is a subset of artificial intelligence."),
    Document(text="Deep learning uses neural networks with multiple layers."),
    Document(text="Natural language processing enables computers to understand text.")
]
index = VectorStoreIndex.from_documents(documents)

# Create and use query engine
query_engine = index.as_query_engine()
response = query_engine.query("What is machine learning?")
print(response.response)
print(f"Sources: {len(response.source_nodes)}")

Router Query Engine for Specialized Handling

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.tools import QueryEngineTool

# Create specialized query engines
vector_engine = vector_index.as_query_engine()
keyword_engine = keyword_index.as_query_engine()

# Define query engine tools
query_engine_tools = [
    QueryEngineTool.from_defaults(
        query_engine=vector_engine,
        description="Useful for semantic similarity queries"
    ),
    QueryEngineTool.from_defaults(
        query_engine=keyword_engine,
        description="Useful for exact keyword matching"
    )
]

# Create router query engine
router_engine = RouterQueryEngine.from_defaults(
    query_engine_tools=query_engine_tools,
    verbose=True
)

response = router_engine.query("Find documents about machine learning")

Sub-Question Decomposition

from llama_index.core.query_engine import SubQuestionQueryEngine

# Create sub-question query engine
sub_question_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=query_engine_tools,
    verbose=True
)

# Query with complex question
response = sub_question_engine.query(
    "Compare machine learning and deep learning approaches, "
    "and explain their applications in natural language processing"
)

SQL Database Querying

from llama_index.core.query_engine import NLSQLTableQueryEngine
from llama_index.core import SQLDatabase

# Connect to database
sql_database = SQLDatabase.from_uri("sqlite:///example.db")

# Create SQL query engine
sql_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    tables=["customers", "orders", "products"],
    synthesize_response=True
)

# Query in natural language
response = sql_engine.query("How many orders were placed last month?")

Custom Query Engine Implementation

from llama_index.core.query_engine import CustomQueryEngine

def custom_query_fn(query_bundle):
    """Custom query processing logic."""
    query_str = query_bundle.query_str
    # Implement custom logic here
    return f"Custom response for: {query_str}"

# Create custom query engine
custom_engine = CustomQueryEngine(query_fn=custom_query_fn)
response = custom_engine.query("Custom query")

Types & Enums

RESPONSE_TYPE = Union[str, Response, StreamingResponse]

class ResponseMode(str, Enum):
    """Response synthesis modes."""
    REFINE = "refine"
    COMPACT = "compact"
    TREE_SUMMARIZE = "tree_summarize"
    SIMPLE_SUMMARIZE = "simple_summarize"
    GENERATION = "generation"
    ACCUMULATE = "accumulate"
    COMPACT_ACCUMULATE = "compact_accumulate"

Install with Tessl CLI