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)