Repository: pathwaycom/pathway
File: python/pathway/xpacks/llm/question_answering.py (Python)

Code before:
            raise ValueError(
                "HTTP server is not built, initialize it with `build_server`"
            )
        self.server.run(*args, **kwargs)


class RerankedRAGQuestionAnswerer(BaseRAGQuestionAnswerer):
    """
    Enhanced RAG Question Answerer with document reranking support.

    This implementation extends the base RAG question answerer by incorporating a reranking step
    to improve the relevance and quality of retrieved documents. After initial retrieval based on
    vector similarity, a reranker model reassesses each document's relevance to the query,
    ensuring the most contextually appropriate documents are used for answer generation.

    Reranking addresses limitations of initial cosine similarity-based retrieval by using more
    sophisticated relevance scoring methods (such as cross-encoders or LLM-based evaluation)
    that jointly evaluate query-document pairs [web:4][web:5][web:14].

    Args:
        llm: LLM instance for question answering.
            See https://pathway.com/developers/api-docs/pathway-xpacks-llm/llms for available models.
        indexer: Indexing object for search & retrieval to be used for context augmentation.
        reranker: Reranker instance to evaluate document relevance. Can be ``CrossEncoderReranker``,
            ``EncoderReranker``, or ``LLMReranker`` from ``pathway.xpacks.llm.rerankers``.
            If ``None``, behaves as standard RAG without reranking. Defaults to ``None``.
        rerank_topk: Number of top-scoring documents to retain after reranking.
            If ``None``, defaults to half of ``search_topk``. Defaults to ``None``.
        **kwargs: Additional arguments passed to ``BaseRAGQuestionAnswerer`` (e.g., ``search_topk``,
            ``default_llm_name``, ``prompt_template``, ``context_processor``).

    Example:

    >>> import pathway as pw  # doctest: +SKIP
    >>> from pathway.xpacks.llm import embedders, splitters, llms, parsers, rerankers  # doctest: +SKIP
    >>> from pathway.xpacks.llm.vector_store import VectorStoreServer  # doctest: +SKIP
    >>> from pathway.xpacks.llm.question_answering import RerankedRAGQuestionAnswerer  # doctest: +SKIP
    >>> from pathway.udfs import DiskCache, ExponentialBackoffRetryStrategy  # doctest: +SKIP
    >>> my_folder = pw.io.fs.read(
    ...     path="/PATH/TO/MY/DATA/*",  # replace with your folder
    ...     format="binary",
    ...     with_metadata=True)  # doctest: +SKIP
    >>> sources = [my_folder]  # doctest: +SKIP
    >>> app_host = "0.0.0.0"  # doctest: +SKIP
    >>> app_port = 8000  # doctest: +SKIP
    >>> parser = parsers.UnstructuredParser()  # doctest: +SKIP
    >>> text_splitter = splitters.TokenCountSplitter(max_tokens=400)  # doctest: +SKIP
    >>> embedder = embedders.OpenAIEmbedder(cache_strategy=DiskCache())  # doctest: +SKIP
    >>> vector_server = VectorStoreServer(  # doctest: +SKIP
    ...     *sources,
    ...     embedder=embedder,

Review comment (suggestion):
This doesn't seem correct - AFAIU if `reranker` is `None` the line 1098 will error.  

Code after:
        class PWAIQuerySchema(pw.Schema):
            prompt: str
            filters: str | None = pw.column_definition(default_value=None)

        self.AnswerQuerySchema = PWAIQuerySchema
        self.RetrieveQuerySchema = self.indexer.RetrieveQuerySchema
        self.StatisticsQuerySchema = self.indexer.StatisticsQuerySchema
        self.InputsQuerySchema = self.indexer.InputsQuerySchema

    @pw.table_transformer
    def answer_query(self, pw_ai_queries: pw.Table) -> pw.Table:
        """Return slides similar to the given query."""

        pw_ai_results = pw_ai_queries + self.indexer.retrieve_query(
            pw_ai_queries.select(
                metadata_filter=pw.this.filters,
                filepath_globpattern=None,
                query=pw.this.prompt,
                k=self.search_topk,
            )
        ).select(
            docs=pw.this.result,
        )

        @pw.udf
        def _format_results(docs: pw.Json) -> pw.Json:
            docs_ls = docs.as_list()

            for docs_dc in docs_ls:
                metadata: dict = docs_dc["metadata"]

                for metadata_key in self.excluded_response_metadata:
                    metadata.pop(metadata_key, None)

                docs_dc["metadata"] = metadata

            return pw.Json(docs_ls)

        pw_ai_results += pw_ai_results.select(result=_format_results(pw.this.docs))

        return pw_ai_results

    @pw.table_transformer
    def retrieve(self, retrieve_queries: pw.Table) -> pw.Table:
        return self.indexer.retrieve_query(retrieve_queries)

    @pw.table_transformer
    def statistics(self, statistics_queries: pw.Table) -> pw.Table:
        return self.indexer.statistics_query(statistics_queries)

    @pw.table_transformer