Documentation Index
Fetch the complete documentation index at: https://docs.praison.ai/llms.txt
Use this file to discover all available pages before exploring further.
Query Engine Module
The query engine module provides different strategies for answering questions using retrieved knowledge.
Quick Start
from praisonaiagents.knowledge.query_engine import (
QueryMode,
QueryResult,
QueryEngineProtocol,
get_query_engine_registry,
decompose_question,
SimpleQueryEngine,
SubQuestionEngine
)
# Decompose complex questions
sub_questions = decompose_question(
"What is Python and how do I install it?"
)
# Returns: ["What is Python?", "How do I install Python?"]
# Use sub-question engine
engine = SubQuestionEngine()
result = engine.query(
"Compare Python and Java for web development",
retriever=my_retriever
)
Query Modes
QueryMode Enum
from praisonaiagents.knowledge.query_engine import QueryMode
class QueryMode(Enum):
DEFAULT = "default" # Direct retrieval + synthesis
SUB_QUESTION = "sub_question" # Decompose into sub-questions
SUMMARIZE = "summarize" # Summarize retrieved content
SQL = "sql" # SQL query generation
ROUTER = "router" # Route to appropriate engine
Mode Comparison
| Mode | Description | Best For |
|---|
default | Simple retrieve + answer | Direct questions |
sub_question | Decompose complex queries | Multi-part questions |
summarize | Summarize all results | Overview queries |
sql | Generate SQL queries | Structured data |
router | Auto-select best mode | Mixed workloads |
Classes
QueryResult
Dataclass for query results.
@dataclass
class QueryResult:
answer: str
sources: List[Dict[str, Any]] = field(default_factory=list)
sub_questions: Optional[List[str]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
QueryEngineProtocol
Protocol for query engine implementations.
class QueryEngineProtocol(Protocol):
name: str
mode: QueryMode
def query(
self,
question: str,
retriever: Any,
top_k: int = 10,
**kwargs
) -> QueryResult:
"""Answer a question using retrieved knowledge."""
...
SimpleQueryEngine
Basic retrieve-and-answer engine.
from praisonaiagents.knowledge.query_engine import SimpleQueryEngine
engine = SimpleQueryEngine()
result = engine.query(
question="What is machine learning?",
retriever=my_retriever,
top_k=5
)
print(result.answer)
print(f"Sources: {len(result.sources)}")
SubQuestionEngine
Decomposes complex questions into sub-questions.
from praisonaiagents.knowledge.query_engine import SubQuestionEngine
engine = SubQuestionEngine()
result = engine.query(
question="Compare the performance and ease of use of Python vs Java",
retriever=my_retriever
)
print(f"Sub-questions: {result.sub_questions}")
print(f"Answer: {result.answer}")
Utility Functions
decompose_question
Break a complex question into simpler sub-questions.
from praisonaiagents.knowledge.query_engine import decompose_question
# Simple decomposition (keyword-based)
questions = decompose_question(
"What is Python and how do I install it on Windows?"
)
# Returns: ["What is Python?", "How do I install Python on Windows?"]
# With LLM decomposition
questions = decompose_question(
"Compare the pros and cons of microservices vs monolith",
use_llm=True
)
QueryEngineRegistry
Registry for managing query engines.
from praisonaiagents.knowledge.query_engine import get_query_engine_registry
registry = get_query_engine_registry()
# List available engines
engines = registry.list_engines() # ['default', 'sub_question', ...]
# Get engine by name
engine = registry.get("sub_question")
# Register custom engine
registry.register("custom", MyEngine)
Using with Knowledge
from praisonaiagents import Agent, Knowledge
# Configure query mode
agent = Agent(
instructions="You are a helpful assistant",
knowledge={
"sources": ["docs/"],
"query_mode": "sub_question" # or "default", "summarize"
}
"query_mode": "sub_question", # or "default", "summarize"
}
)
response = agent.chat("What are the benefits and drawbacks of this approach?")
Creating Custom Query Engines
from praisonaiagents.knowledge.query_engine import (
QueryMode,
QueryResult,
get_query_engine_registry
)
from typing import Any
class MyQueryEngine:
name = "my_engine"
mode = QueryMode.DEFAULT
def query(
self,
question: str,
retriever: Any,
top_k: int = 10,
**kwargs
) -> QueryResult:
# Retrieve relevant documents
docs = retriever.retrieve(question, top_k=top_k)
# Custom answer synthesis
answer = self._synthesize(question, docs)
return QueryResult(
answer=answer,
sources=[{"text": d.text} for d in docs]
)
def _synthesize(self, question: str, docs: list) -> str:
# Your synthesis logic
...
# Register
registry = get_query_engine_registry()
registry.register("my_engine", MyQueryEngine)
- decompose_question has a fast keyword-based mode (no API calls)
- LLM-based decomposition requires API calls
- Sub-question engine parallelizes sub-query retrieval