Tree summarize

Init file.

Accumulate

Bases: BaseSynthesizer

Accumulate responses from multiple text chunks.

Source code in llama-index-core/llama_index/core/response_synthesizers/accumulate.py

class Accumulate(BaseSynthesizer):
    """Accumulate responses from multiple text chunks."""

    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        text_qa_template: Optional[BasePromptTemplate] = None,
        chat_content_qa_template: Optional[BasePromptTemplate] = None,
        output_cls: Optional[Type[BaseModel]] = None,
        streaming: bool = False,
        use_async: bool = False,
        multimodal: bool = False,
    ) -> None:
        super().__init__(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            streaming=streaming,
            output_cls=output_cls,
            multimodal=multimodal,
        )
        self._text_qa_template = text_qa_template or DEFAULT_TEXT_QA_PROMPT_SEL
        self._chat_content_qa_template = (
            chat_content_qa_template or CHAT_CONTENT_QA_PROMPT
        )
        self._use_async = use_async

    def _get_prompts(self) -> PromptDictType:
        return {
            "text_qa_template": self._text_qa_template,
            "chat_content_qa_template": self._chat_content_qa_template,
        }

    def _update_prompts(self, prompts: PromptDictType) -> None:
        if "text_qa_template" in prompts:
            self._text_qa_template = prompts["text_qa_template"]
        if "chat_content_qa_template" in prompts:
            self._chat_content_qa_template = prompts["chat_content_qa_template"]

    def flatten_list(self, md_array: List[List[Any]]) -> List[Any]:
        return [item for sublist in md_array for item in sublist]

    def _format_response(self, outputs: List[Any], separator: str) -> str:
        responses: List[str] = []
        for response in outputs:
            responses.append(response or "Empty Response")

        return separator.join(
            [f"Response {index + 1}: {item}" for index, item in enumerate(responses)]
        )

    def _make_prompt_kwargs(self, chunk: str | ChatMessage) -> dict[str, Any]:
        if self._multimodal:
            template = self._chat_content_qa_template.partial_format(
                query_str="{query_str}"
            )
            return {"context_messages": [chunk], "template": template}
        else:
            template = self._text_qa_template.partial_format(query_str="{query_str}")
            return {"context_str": chunk, "template": template}

    def _give_responses(
        self,
        query_str: str,
        chunk: str | ChatMessage,
        use_async: bool = False,
        **response_kwargs: Any,
    ) -> List[Any]:
        repacked: list[str] | list[ChatMessage]
        if self._multimodal:
            assert isinstance(chunk, ChatMessage)
            template = self._chat_content_qa_template.partial_format(
                query_str=query_str
            )
            repacked = self._chat_prompt_helper.repack(template, [chunk], llm=self._llm)
        else:
            assert isinstance(chunk, str)
            template = self._text_qa_template.partial_format(query_str=query_str)
            repacked = self._prompt_helper.repack(template, [chunk], llm=self._llm)

        predictor: Callable
        if self._output_cls is None:
            predictor = self._llm.apredict if use_async else self._llm.predict
            return [
                predictor(template, **self._make_prompt_kwargs(c), **response_kwargs)
                for c in repacked
            ]
        else:
            predictor = (
                self._llm.astructured_predict
                if use_async
                else self._llm.structured_predict
            )
            return [
                predictor(
                    self._output_cls,
                    template,
                    **self._make_prompt_kwargs(c),
                    **response_kwargs,
                )
                for c in repacked
            ]

    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        separator: str = "\n---------------------\n",
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        if self._streaming:
            raise ValueError("Unable to stream in Accumulate response mode")

        tasks = [
            self._give_responses(
                query_str, text_chunk, use_async=True, **response_kwargs
            )
            for text_chunk in text_chunks
        ]

        flattened_tasks = self.flatten_list(tasks)
        outputs = await asyncio.gather(*flattened_tasks)

        return self._format_response(outputs, separator)

    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        separator: str = "\n---------------------\n",
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        if self._streaming:
            raise ValueError("Unable to stream in Accumulate response mode")

        tasks = [
            self._give_responses(
                query_str, text_chunk, use_async=self._use_async, **response_kwargs
            )
            for text_chunk in text_chunks
        ]

        outputs = self.flatten_list(tasks)

        if self._use_async:
            outputs = run_async_tasks(outputs)

        return self._format_response(outputs, separator)

    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        separator: str = "\n---------------------\n",
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        if self._streaming:
            raise ValueError("Unable to stream in Accumulate response mode")

        tasks = [
            self._give_responses(query_str, chunk, use_async=True, **response_kwargs)
            for chunk in message_chunks
        ]

        flattened_tasks = self.flatten_list(tasks)
        outputs = await asyncio.gather(*flattened_tasks)

        return self._format_response(outputs, separator)

    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        separator: str = "\n---------------------\n",
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        if self._streaming:
            raise ValueError("Unable to stream in Accumulate response mode")

        tasks = [
            self._give_responses(
                query_str, chunk, use_async=self._use_async, **response_kwargs
            )
            for chunk in message_chunks
        ]

        outputs = self.flatten_list(tasks)

        if self._use_async:
            outputs = run_async_tasks(outputs)

        return self._format_response(outputs, separator)

BaseSynthesizer

Bases: PromptMixin, DispatcherSpanMixin

Response builder class.

Source code in llama-index-core/llama_index/core/response_synthesizers/base.py

class BaseSynthesizer(PromptMixin, DispatcherSpanMixin):
    """Response builder class."""

    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        streaming: bool = False,
        output_cls: Optional[Type[BaseModel]] = None,
        empty_response: Optional[str] = None,
        multimodal: bool = False,
    ) -> None:
        """Init params."""
        self._llm = llm or Settings.llm

        if callback_manager:
            self._llm.callback_manager = callback_manager

        self._callback_manager = callback_manager or Settings.callback_manager
        self._streaming = streaming
        self._output_cls = output_cls
        self._empty_response = empty_response or "Empty Response"
        self._multimodal = multimodal
        self._prompt_helper: PromptHelper
        if multimodal:
            if not is_chat_model(self._llm):
                raise ValueError("Multimodal synthesis requires a chat LLM.")
        self._prompt_helper = (
            prompt_helper
            or Settings._prompt_helper
            or PromptHelper.from_llm_metadata(
                self._llm.metadata,
            )
        )
        self._chat_prompt_helper = (
            chat_prompt_helper
            or Settings._chat_prompt_helper
            or ChatPromptHelper.from_llm_metadata(
                self._llm.metadata,
            )
        )

    def _empty_response_generator(self) -> Generator[str, None, None]:
        yield self._empty_response

    async def _empty_response_agenerator(self) -> AsyncGenerator[str, None]:
        yield self._empty_response

    def _get_prompt_modules(self) -> Dict[str, Any]:
        """Get prompt modules."""
        # TODO: keep this for now since response synthesizers don't generally have sub-modules
        return {}

    @property
    def callback_manager(self) -> CallbackManager:
        return self._callback_manager

    @callback_manager.setter
    def callback_manager(self, callback_manager: CallbackManager) -> None:
        """Set callback manager."""
        self._callback_manager = callback_manager
        # TODO: please fix this later
        self._callback_manager = callback_manager
        self._llm.callback_manager = callback_manager

    @abstractmethod
    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get response."""
        ...

    @abstractmethod
    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get response."""
        ...

    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        raise NotImplementedError

    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        raise NotImplementedError

    def _log_prompt_and_response(
        self,
        formatted_prompt: str,
        response: RESPONSE_TEXT_TYPE,
        log_prefix: str = "",
    ) -> None:
        """Log prompt and response from LLM."""
        logger.debug(f"> {log_prefix} prompt template: {formatted_prompt}")
        logger.debug(f"> {log_prefix} response: {response}")

    def _get_metadata_for_response(
        self,
        nodes: List[BaseNode],
    ) -> Optional[Dict[str, Any]]:
        """Get metadata for response."""
        return {node.node_id: node.metadata for node in nodes}

    def _prepare_response_output(
        self,
        response_str: Optional[RESPONSE_TEXT_TYPE],
        source_nodes: List[NodeWithScore],
    ) -> RESPONSE_TYPE:
        """Prepare response object from response string."""
        response_metadata = self._get_metadata_for_response(
            [node_with_score.node for node_with_score in source_nodes]
        )

        if isinstance(self._llm, StructuredLLM):
            # convert string to output_cls
            output = self._llm.output_cls.model_validate_json(str(response_str))
            return PydanticResponse(
                output,
                source_nodes=source_nodes,
                metadata=response_metadata,
            )

        if isinstance(response_str, str):
            return Response(
                response_str,
                source_nodes=source_nodes,
                metadata=response_metadata,
            )
        if isinstance(response_str, Generator):
            return StreamingResponse(
                response_str,
                source_nodes=source_nodes,
                metadata=response_metadata,
            )
        if isinstance(response_str, AsyncGenerator):
            return AsyncStreamingResponse(
                response_str,
                source_nodes=source_nodes,
                metadata=response_metadata,
            )

        if self._output_cls is not None and isinstance(response_str, self._output_cls):
            return PydanticResponse(
                response_str, source_nodes=source_nodes, metadata=response_metadata
            )

        raise ValueError(
            f"Response must be a string or a generator. Found {type(response_str)}"
        )

    @dispatcher.span
    def synthesize(
        self,
        query: QueryTextType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TYPE:
        dispatcher.event(
            SynthesizeStartEvent(
                query=query,
            )
        )

        if len(nodes) == 0:
            if self._streaming:
                empty_response_stream = StreamingResponse(
                    response_gen=self._empty_response_generator()
                )
                dispatcher.event(
                    SynthesizeEndEvent(
                        query=query,
                        response=empty_response_stream,
                    )
                )
                return empty_response_stream
            else:
                empty_response = Response(self._empty_response)
                dispatcher.event(
                    SynthesizeEndEvent(
                        query=query,
                        response=empty_response,
                    )
                )
                return empty_response

        if isinstance(query, str):
            query = QueryBundle(query_str=query)

        with self._callback_manager.event(
            CBEventType.SYNTHESIZE,
            payload={EventPayload.QUERY_STR: query.query_str},
        ) as event:
            if self._multimodal:
                response_str = self.get_response_from_messages(
                    query_str=query.query_str,
                    message_chunks=[
                        ChatMessage(
                            blocks=n.node.get_content_blocks(
                                metadata_mode=MetadataMode.LLM
                            )
                        )
                        for n in nodes
                    ],
                    **response_kwargs,
                )
            else:
                response_str = self.get_response(
                    query_str=query.query_str,
                    text_chunks=[
                        n.node.get_content(metadata_mode=MetadataMode.LLM)
                        for n in nodes
                    ],
                    **response_kwargs,
                )

            additional_source_nodes = additional_source_nodes or []
            source_nodes = list(nodes) + list(additional_source_nodes)

            response = self._prepare_response_output(response_str, source_nodes)

            event.on_end(payload={EventPayload.RESPONSE: response})

        dispatcher.event(
            SynthesizeEndEvent(
                query=query,
                response=response,
            )
        )
        return response

    @dispatcher.span
    async def asynthesize(
        self,
        query: QueryTextType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TYPE:
        dispatcher.event(
            SynthesizeStartEvent(
                query=query,
            )
        )
        if len(nodes) == 0:
            if self._streaming:
                empty_response_stream = AsyncStreamingResponse(
                    response_gen=self._empty_response_agenerator()
                )
                dispatcher.event(
                    SynthesizeEndEvent(
                        query=query,
                        response=empty_response_stream,
                    )
                )
                return empty_response_stream
            else:
                empty_response = Response(self._empty_response)
                dispatcher.event(
                    SynthesizeEndEvent(
                        query=query,
                        response=empty_response,
                    )
                )
                return empty_response

        if isinstance(query, str):
            query = QueryBundle(query_str=query)

        with self._callback_manager.event(
            CBEventType.SYNTHESIZE,
            payload={EventPayload.QUERY_STR: query.query_str},
        ) as event:
            if self._multimodal:
                response_str = await self.aget_response_from_messages(
                    query_str=query.query_str,
                    message_chunks=[
                        ChatMessage(
                            blocks=n.node.get_content_blocks(
                                metadata_mode=MetadataMode.LLM
                            )
                        )
                        for n in nodes
                    ],
                    **response_kwargs,
                )
            else:
                response_str = await self.aget_response(
                    query_str=query.query_str,
                    text_chunks=[
                        n.node.get_content(metadata_mode=MetadataMode.LLM)
                        for n in nodes
                    ],
                    **response_kwargs,
                )

            additional_source_nodes = additional_source_nodes or []
            source_nodes = list(nodes) + list(additional_source_nodes)

            response = self._prepare_response_output(response_str, source_nodes)

            event.on_end(payload={EventPayload.RESPONSE: response})

        dispatcher.event(
            SynthesizeEndEvent(
                query=query,
                response=response,
            )
        )
        return response

get_response abstractmethod

get_response(
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Get response.

Source code in llama-index-core/llama_index/core/response_synthesizers/base.py

@abstractmethod
def get_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Get response."""
    ...

aget_response abstractmethod async

aget_response(
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Get response.

Source code in llama-index-core/llama_index/core/response_synthesizers/base.py

@abstractmethod
async def aget_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Get response."""
    ...

CompactAndRefine

Bases: Refine

Refine responses across compact text chunks.

Source code in llama-index-core/llama_index/core/response_synthesizers/compact_and_refine.py

class CompactAndRefine(Refine):
    """Refine responses across compact text chunks."""

    @dispatcher.span
    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        compact_texts = self._make_compact_text_chunks(query_str, text_chunks)
        return await super().aget_response(
            query_str=query_str,
            text_chunks=compact_texts,
            prev_response=prev_response,
            **response_kwargs,
        )

    @dispatcher.span
    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get compact response."""
        # use prompt helper to fix compact text_chunks under the prompt limitation
        new_texts = self._make_compact_text_chunks(query_str, text_chunks)
        return super().get_response(
            query_str=query_str,
            text_chunks=new_texts,
            prev_response=prev_response,
            **response_kwargs,
        )

    def _make_compact_text_chunks(
        self, query_str: str, text_chunks: Sequence[str]
    ) -> List[str]:
        text_qa_template = self._text_qa_template.partial_format(query_str=query_str)
        refine_template = self._refine_template.partial_format(query_str=query_str)

        max_prompt = get_biggest_prompt([text_qa_template, refine_template])
        return self._prompt_helper.repack(
            max_prompt, text_chunks, llm=self._llm, padding=self._response_padding_size
        )

    def _make_compact_message_chunks(
        self, query_str: str, message_chunks: Sequence[ChatMessage]
    ) -> List[ChatMessage]:
        qa_template = self._chat_content_qa_template.partial_format(query_str=query_str)
        refine_template = self._chat_content_refine_template.partial_format(
            query_str=query_str
        )

        max_prompt = get_biggest_prompt([qa_template, refine_template])
        return self._chat_prompt_helper.repack(
            max_prompt,
            list(message_chunks),
            llm=self._llm,
            padding=self._response_padding_size,
        )

    @dispatcher.span
    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        compact_chunks = self._make_compact_message_chunks(query_str, message_chunks)
        return super().get_response_from_messages(
            query_str=query_str,
            message_chunks=compact_chunks,
            prev_response=prev_response,
            **response_kwargs,
        )

    @dispatcher.span
    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        compact_chunks = self._make_compact_message_chunks(query_str, message_chunks)
        return await super().aget_response_from_messages(
            query_str=query_str,
            message_chunks=compact_chunks,
            prev_response=prev_response,
            **response_kwargs,
        )

get_response

get_response(
    query_str: str,
    text_chunks: Sequence[str],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Get compact response.

Source code in llama-index-core/llama_index/core/response_synthesizers/compact_and_refine.py

@dispatcher.span
def get_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Get compact response."""
    # use prompt helper to fix compact text_chunks under the prompt limitation
    new_texts = self._make_compact_text_chunks(query_str, text_chunks)
    return super().get_response(
        query_str=query_str,
        text_chunks=new_texts,
        prev_response=prev_response,
        **response_kwargs,
    )

Generation

Bases: BaseSynthesizer

Source code in llama-index-core/llama_index/core/response_synthesizers/generation.py

class Generation(BaseSynthesizer):
    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        simple_template: Optional[BasePromptTemplate] = None,
        streaming: bool = False,
        multimodal: bool = False,
    ) -> None:
        super().__init__(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            streaming=streaming,
            multimodal=multimodal,
        )
        self._input_prompt = simple_template or (
            CHAT_SIMPLE_INPUT_PROMPT if multimodal else DEFAULT_SIMPLE_INPUT_PROMPT
        )

    def _get_prompts(self) -> PromptDictType:
        """Get prompts."""
        return {"simple_template": self._input_prompt}

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """Update prompts."""
        if "simple_template" in prompts:
            self._input_prompt = prompts["simple_template"]

    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        del text_chunks

        if not self._streaming:
            return self._llm.predict(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )
        else:
            return self._llm.stream(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )

    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        del text_chunks

        if not self._streaming:
            return await self._llm.apredict(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )
        else:
            return await self._llm.astream(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )

    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        del message_chunks

        if not self._streaming:
            return self._llm.predict(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )
        else:
            return self._llm.stream(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )

    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        del message_chunks

        if not self._streaming:
            return await self._llm.apredict(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )
        else:
            return await self._llm.astream(
                self._input_prompt,
                query_str=query_str,
                **response_kwargs,
            )

    # NOTE: synthesize and asynthesize bypass the base class empty-node early
    #       return so that Generation always calls the LLM regardless of nodes.

    @dispatcher.span
    def synthesize(
        self,
        query: QueryType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TYPE:
        dispatcher.event(
            SynthesizeStartEvent(
                query=query,
            )
        )

        if isinstance(query, str):
            query = QueryBundle(query_str=query)

        with self._callback_manager.event(
            CBEventType.SYNTHESIZE,
            payload={EventPayload.QUERY_STR: query.query_str},
        ) as event:
            if self._multimodal:
                response_str = self.get_response_from_messages(
                    query_str=query.query_str,
                    message_chunks=[
                        ChatMessage(
                            blocks=n.node.get_content_blocks(
                                metadata_mode=MetadataMode.LLM
                            )
                        )
                        for n in nodes
                    ],
                    **response_kwargs,
                )
            else:
                response_str = self.get_response(
                    query_str=query.query_str,
                    text_chunks=[
                        n.node.get_content(metadata_mode=MetadataMode.LLM)
                        for n in nodes
                    ],
                    **response_kwargs,
                )

            additional_source_nodes = additional_source_nodes or []
            source_nodes = list(nodes) + list(additional_source_nodes)

            response = self._prepare_response_output(response_str, source_nodes)

            event.on_end(payload={EventPayload.RESPONSE: response})

        dispatcher.event(
            SynthesizeEndEvent(
                query=query,
                response=response,
            )
        )
        return response

    @dispatcher.span
    async def asynthesize(
        self,
        query: QueryType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TYPE:
        dispatcher.event(
            SynthesizeStartEvent(
                query=query,
            )
        )

        if isinstance(query, str):
            query = QueryBundle(query_str=query)

        with self._callback_manager.event(
            CBEventType.SYNTHESIZE,
            payload={EventPayload.QUERY_STR: query.query_str},
        ) as event:
            if self._multimodal:
                response_str = await self.aget_response_from_messages(
                    query_str=query.query_str,
                    message_chunks=[
                        ChatMessage(
                            blocks=n.node.get_content_blocks(
                                metadata_mode=MetadataMode.LLM
                            )
                        )
                        for n in nodes
                    ],
                    **response_kwargs,
                )
            else:
                response_str = await self.aget_response(
                    query_str=query.query_str,
                    text_chunks=[
                        n.node.get_content(metadata_mode=MetadataMode.LLM)
                        for n in nodes
                    ],
                    **response_kwargs,
                )

            additional_source_nodes = additional_source_nodes or []
            source_nodes = list(nodes) + list(additional_source_nodes)

            response = self._prepare_response_output(response_str, source_nodes)

            event.on_end(payload={EventPayload.RESPONSE: response})

        dispatcher.event(
            SynthesizeEndEvent(
                query=query,
                response=response,
            )
        )
        return response

Refine

Bases: BaseSynthesizer

Refine a response to a query across text chunks.

Source code in llama-index-core/llama_index/core/response_synthesizers/refine.py

class Refine(BaseSynthesizer):
    """Refine a response to a query across text chunks."""

    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        text_qa_template: Optional[BasePromptTemplate] = None,
        refine_template: Optional[BasePromptTemplate] = None,
        chat_content_qa_template: Optional[BasePromptTemplate] = None,
        chat_content_refine_template: Optional[BasePromptTemplate] = None,
        output_cls: Optional[Type[BaseModel]] = None,
        response_padding_size: int = DEFAULT_RESPONSE_PADDING_SIZE,
        streaming: bool = False,
        verbose: bool = False,
        structured_answer_filtering: bool = False,
        program_factory: Optional[
            Callable[[BasePromptTemplate], BasePydanticProgram]
        ] = None,
        multimodal: bool = False,
    ) -> None:
        super().__init__(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            streaming=streaming,
            multimodal=multimodal,
        )
        self._text_qa_template = text_qa_template or DEFAULT_TEXT_QA_PROMPT_SEL
        self._refine_template = refine_template or DEFAULT_REFINE_PROMPT_SEL
        self._chat_content_qa_template = (
            chat_content_qa_template or CHAT_CONTENT_QA_PROMPT
        )
        self._chat_content_refine_template = (
            chat_content_refine_template or CHAT_CONTENT_REFINE_PROMPT
        )
        self._verbose = verbose
        self._structured_answer_filtering = structured_answer_filtering
        self._output_cls = output_cls
        self._response_padding_size = response_padding_size

        if not self._structured_answer_filtering and program_factory is not None:
            raise ValueError(
                "Program factory not supported without structured answer filtering."
            )
        self._program_factory = program_factory or self._default_program_factory

    def _get_prompts(self) -> PromptDictType:
        """Get prompts."""
        return {
            "text_qa_template": self._text_qa_template,
            "refine_template": self._refine_template,
            "chat_content_qa_template": self._chat_content_qa_template,
            "chat_content_refine_template": self._chat_content_refine_template,
        }

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """Update prompts."""
        if "text_qa_template" in prompts:
            self._text_qa_template = prompts["text_qa_template"]
        if "refine_template" in prompts:
            self._refine_template = prompts["refine_template"]
        if "chat_content_qa_template" in prompts:
            self._chat_content_qa_template = prompts["chat_content_qa_template"]
        if "chat_content_refine_template" in prompts:
            self._chat_content_refine_template = prompts["chat_content_refine_template"]

    @staticmethod
    def _get_attribute_from_object_generator(
        generator: Generator, structured_response: BaseModel | None, attribute: str
    ) -> Generator:
        """
        Object generators like those returned by the DefaultRefineProgram or FunctionCallingProgram
        stream_call may yield multiple objects, but because we cannot guarantee the order of object attribute generation
        we need to wait until it's fully generated to be sure that the attribute is both present and complete
        """
        for obj in generator:
            structured_response = obj
        yield getattr(structured_response, attribute)

    @staticmethod
    async def _get_attribute_from_object_async_generator(
        generator: AsyncGenerator, structured_response: BaseModel | None, attribute: str
    ) -> AsyncGenerator:
        """
        Object generators like those returned by the DefaultRefineProgram or FunctionCallingProgram
        stream_call may yield multiple objects, but because we cannot guarantee the order of object attribute generation
        we need to wait until it's fully generated to be sure that the attribute is both present and complete
        """
        async for obj in generator:
            structured_response = obj
        yield getattr(structured_response, attribute)
        return

    def _default_program_factory(
        self, prompt: BasePromptTemplate
    ) -> BasePydanticProgram:
        if self._structured_answer_filtering:
            from llama_index.core.program.utils import get_program_for_llm

            return get_program_for_llm(
                StructuredRefineResponse,
                prompt,
                self._llm,
                verbose=self._verbose,
            )
        else:
            return DefaultRefineProgram(
                prompt=prompt,
                llm=self._llm,
                output_cls=self._output_cls,
            )

    def _update_response(
        self, program: BasePydanticProgram, program_kwargs: dict, response_kwargs: dict
    ) -> Optional[RESPONSE_TEXT_TYPE]:
        """Update response."""
        query_satisfied: bool | None = False
        structured_response: Union[
            StructuredRefineResponse,
            Any,
            list[Any],
            FlexibleModel,
            list[FlexibleModel],
            None,
        ]
        if not self._streaming:
            try:
                structured_response = cast(
                    StructuredRefineResponse,
                    program(
                        **program_kwargs,
                        **response_kwargs,
                    ),
                )
                query_satisfied = structured_response.query_satisfied
                if query_satisfied:
                    return structured_response.answer
            except (ValidationError, ValueError, TypeError) as e:
                logger.warning(f"Structured response error: {e}", exc_info=True)
        elif self._streaming:
            try:
                structured_response_gen = program.stream_call(
                    **program_kwargs,
                    **response_kwargs,
                )
                structured_response = None
                for sr in structured_response_gen:
                    assert not isinstance(sr, list)
                    structured_response = sr
                    if sr is not None:
                        query_satisfied = getattr(sr, "query_satisfied", None)
                        if query_satisfied is not None:
                            break
                if query_satisfied:
                    return self._get_attribute_from_object_generator(
                        structured_response_gen,
                        structured_response,
                        "answer",
                    )
            except (ValidationError, ValueError, TypeError) as e:
                logger.warning(f"Structured response error: {e}", exc_info=True)
        return None

    async def _aupdate_response(
        self, program: BasePydanticProgram, program_kwargs: dict, response_kwargs: dict
    ) -> Optional[RESPONSE_TEXT_TYPE]:
        """Update response."""
        query_satisfied: bool | None = False
        structured_response: Union[
            StructuredRefineResponse,
            Any,
            list[Any],
            FlexibleModel,
            list[FlexibleModel],
            None,
        ]
        if not self._streaming:
            try:
                structured_response = cast(
                    StructuredRefineResponse,
                    await program.acall(
                        **program_kwargs,
                        **response_kwargs,
                    ),
                )
                if structured_response.query_satisfied:
                    return structured_response.answer
            except (ValidationError, ValueError, TypeError) as e:
                logger.warning(f"Structured response error: {e}", exc_info=True)
        elif self._streaming:
            try:
                structured_response_gen = await program.astream_call(
                    **program_kwargs,
                    **response_kwargs,
                )
                structured_response = None
                async for sr in structured_response_gen:
                    assert not isinstance(sr, list)
                    structured_response = sr
                    if sr is not None:
                        query_satisfied = getattr(sr, "query_satisfied", None)
                        if query_satisfied is not None:
                            break
                if query_satisfied:
                    return self._get_attribute_from_object_async_generator(
                        structured_response_gen,
                        structured_response,
                        "answer",
                    )
            except (ValidationError, ValueError, TypeError) as e:
                logger.warning(f"Structured response error: {e}", exc_info=True)
        return None

    def _run_refine_loop(
        self,
        query_str: str,
        chunks: Sequence[Any],
        qa_template: BasePromptTemplate,
        refine_template: BasePromptTemplate,
        make_qa_prompt_kwargs: Callable[[Any], dict],
        make_refine_prompt_kwargs: Callable[[Any], dict],
        start_event: Any,
        end_event: Any,
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        dispatcher.event(start_event)
        max_prompt = get_biggest_prompt([qa_template, refine_template])
        # Make first best guess at how many chunks we can fit in the prompt at once. Increase padding
        # to give more room for the response
        prompt_helper: PromptHelper | ChatPromptHelper = (
            self._prompt_helper if not self._multimodal else self._chat_prompt_helper
        )
        chunks_deque: deque = deque(
            [
                tc
                for chunk in chunks
                for tc in prompt_helper.repack(
                    max_prompt,
                    [chunk],
                    llm=self._llm,
                    padding=self._response_padding_size,
                )
            ]
        )
        response = prev_response
        while chunks_deque:
            if isinstance(response, Generator):
                response = get_response_text(response)

            chunk = chunks_deque.popleft()

            if response is None:
                prompt_template = qa_template.partial_format(query_str=query_str)
                prompt_kwargs = make_qa_prompt_kwargs(chunk)
            else:
                prompt_template = refine_template.partial_format(
                    query_str=query_str, existing_answer=response
                )
                # Because the existing answer portion is constantly being updated, it may be necessary to
                # repack the chunk with the new prompt template to ensure it fits. Since the template has been
                # partially formatted with the actual response, there's no need for the extra padding.
                repacked = prompt_helper.repack(
                    prompt_template,
                    [chunk],
                    llm=self._llm,
                )
                # If chunk is too big to be packed into a single chunk, push new chunks into the front of the deque
                if len(repacked) > 1:
                    chunks_deque.extendleft(repacked)
                    continue
                chunk = repacked[0]
                prompt_kwargs = make_refine_prompt_kwargs(chunk)

            program = self._program_factory(prompt_template)
            if resp := self._update_response(program, prompt_kwargs, response_kwargs):
                response = resp

        if isinstance(response, str):
            if self._output_cls is not None:
                try:
                    response = self._output_cls.model_validate_json(response)
                except (ValidationError, ValueError, TypeError):
                    pass
            else:
                response = response or "Empty Response"
        elif response is None:
            response = "Empty Response"
        else:
            response = cast(Generator, response)
        dispatcher.event(end_event)
        return response

    async def _arun_refine_loop(
        self,
        query_str: str,
        chunks: Sequence[Any],
        qa_template: BasePromptTemplate,
        refine_template: BasePromptTemplate,
        make_qa_prompt_kwargs: Callable[[Any], dict],
        make_refine_prompt_kwargs: Callable[[Any], dict],
        start_event: Any,
        end_event: Any,
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        dispatcher.event(start_event)
        max_prompt = get_biggest_prompt([qa_template, refine_template])
        prompt_helper: PromptHelper | ChatPromptHelper = (
            self._prompt_helper if not self._multimodal else self._chat_prompt_helper
        )
        # Make first best guess at how many chunks we can fit in the prompt at once. Increase padding
        # to give more room for the response
        chunks_deque: deque = deque(
            [
                tc
                for chunk in chunks
                for tc in prompt_helper.repack(
                    max_prompt,
                    [chunk],
                    llm=self._llm,
                    padding=self._response_padding_size,
                )
            ]
        )
        response = prev_response
        while chunks_deque:
            if isinstance(response, AsyncGenerator):
                response = await aget_response_text(response)

            chunk = chunks_deque.popleft()

            if response is None:
                prompt_template = qa_template.partial_format(query_str=query_str)
                prompt_kwargs = make_qa_prompt_kwargs(chunk)
            else:
                prompt_template = refine_template.partial_format(
                    query_str=query_str, existing_answer=response
                )
                # Because the existing answer portion is constantly being updated, it may be necessary to
                # repack the chunk with the new prompt template to ensure it fits. Since the template has been
                # partially formatted with the actual response, there's no need for the extra padding.
                repacked = prompt_helper.repack(
                    prompt_template,
                    [chunk],
                    llm=self._llm,
                )
                # If chunk is too big to be packed into a single chunk, push new chunks into the front of the deque
                if len(repacked) > 1:
                    chunks_deque.extendleft(repacked)
                    continue
                chunk = repacked[0]
                prompt_kwargs = make_refine_prompt_kwargs(chunk)

            program = self._program_factory(prompt_template)
            if resp := await self._aupdate_response(
                program, prompt_kwargs, response_kwargs
            ):
                response = resp

        if isinstance(response, str):
            if self._output_cls is not None:
                try:
                    response = self._output_cls.model_validate_json(response)
                except (ValidationError, ValueError, TypeError):
                    pass
            else:
                response = response or "Empty Response"
        elif response is None:
            response = "Empty Response"
        else:
            response = cast(AsyncGenerator, response)
        dispatcher.event(end_event)
        return response

    # TODO: Why does this class call dispatcher.span on this method when other classes only call it on synthesize
    @dispatcher.span
    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Give response over chunks."""
        return self._run_refine_loop(
            query_str=query_str,
            chunks=text_chunks,
            qa_template=self._text_qa_template,
            refine_template=self._refine_template,
            make_qa_prompt_kwargs=lambda chunk: {"context_str": chunk},
            make_refine_prompt_kwargs=lambda chunk: {"context_msg": chunk},
            start_event=GetResponseStartEvent(
                query_str=query_str, text_chunks=list(text_chunks)
            ),
            end_event=GetResponseEndEvent(),
            prev_response=prev_response,
            **response_kwargs,
        )

    @dispatcher.span
    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        return await self._arun_refine_loop(
            query_str=query_str,
            chunks=text_chunks,
            qa_template=self._text_qa_template,
            refine_template=self._refine_template,
            make_qa_prompt_kwargs=lambda chunk: {"context_str": chunk},
            make_refine_prompt_kwargs=lambda chunk: {"context_msg": chunk},
            start_event=GetResponseStartEvent(
                query_str=query_str, text_chunks=list(text_chunks)
            ),
            end_event=GetResponseEndEvent(),
            prev_response=prev_response,
            **response_kwargs,
        )

    @dispatcher.span
    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Give response over message chunks."""
        return self._run_refine_loop(
            query_str=query_str,
            chunks=message_chunks,
            qa_template=self._chat_content_qa_template,
            refine_template=self._chat_content_refine_template,
            make_qa_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
            make_refine_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
            start_event=GetMessageResponseStartEvent(
                query_str=query_str, message_chunks=list(message_chunks)
            ),
            end_event=GetMessageResponseEndEvent(),
            prev_response=prev_response,
            **response_kwargs,
        )

    @dispatcher.span
    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        return await self._arun_refine_loop(
            query_str=query_str,
            chunks=message_chunks,
            qa_template=self._chat_content_qa_template,
            refine_template=self._chat_content_refine_template,
            make_qa_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
            make_refine_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
            start_event=GetMessageResponseStartEvent(
                query_str=query_str, message_chunks=list(message_chunks)
            ),
            end_event=GetMessageResponseEndEvent(),
            prev_response=prev_response,
            **response_kwargs,
        )

get_response

get_response(
    query_str: str,
    text_chunks: Sequence[str],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Give response over chunks.

Source code in llama-index-core/llama_index/core/response_synthesizers/refine.py

@dispatcher.span
def get_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Give response over chunks."""
    return self._run_refine_loop(
        query_str=query_str,
        chunks=text_chunks,
        qa_template=self._text_qa_template,
        refine_template=self._refine_template,
        make_qa_prompt_kwargs=lambda chunk: {"context_str": chunk},
        make_refine_prompt_kwargs=lambda chunk: {"context_msg": chunk},
        start_event=GetResponseStartEvent(
            query_str=query_str, text_chunks=list(text_chunks)
        ),
        end_event=GetResponseEndEvent(),
        prev_response=prev_response,
        **response_kwargs,
    )

get_response_from_messages

get_response_from_messages(
    query_str: str,
    message_chunks: Sequence[ChatMessage],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Give response over message chunks.

Source code in llama-index-core/llama_index/core/response_synthesizers/refine.py

@dispatcher.span
def get_response_from_messages(
    self,
    query_str: str,
    message_chunks: Sequence[ChatMessage],
    prev_response: Optional[RESPONSE_TEXT_TYPE] = None,
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Give response over message chunks."""
    return self._run_refine_loop(
        query_str=query_str,
        chunks=message_chunks,
        qa_template=self._chat_content_qa_template,
        refine_template=self._chat_content_refine_template,
        make_qa_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
        make_refine_prompt_kwargs=lambda chunk: {"context_messages": [chunk]},
        start_event=GetMessageResponseStartEvent(
            query_str=query_str, message_chunks=list(message_chunks)
        ),
        end_event=GetMessageResponseEndEvent(),
        prev_response=prev_response,
        **response_kwargs,
    )

SimpleSummarize

Bases: BaseSynthesizer

Source code in llama-index-core/llama_index/core/response_synthesizers/simple_summarize.py

class SimpleSummarize(BaseSynthesizer):
    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        text_qa_template: Optional[BasePromptTemplate] = None,
        chat_content_qa_template: Optional[BasePromptTemplate] = None,
        streaming: bool = False,
        multimodal: bool = False,
        strict_truncation: bool = False,
    ) -> None:
        super().__init__(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            streaming=streaming,
            multimodal=multimodal,
        )
        self._text_qa_template = text_qa_template or DEFAULT_TEXT_QA_PROMPT_SEL
        self._chat_content_qa_template = (
            chat_content_qa_template or CHAT_CONTENT_QA_PROMPT
        )
        self._strict_truncation = strict_truncation

    def _get_prompts(self) -> PromptDictType:
        """Get prompts."""
        return {
            "text_qa_template": self._text_qa_template,
            "chat_content_qa_template": self._chat_content_qa_template,
        }

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """Update prompts."""
        if "text_qa_template" in prompts:
            self._text_qa_template = prompts["text_qa_template"]
        if "chat_content_qa_template" in prompts:
            self._chat_content_qa_template = prompts["chat_content_qa_template"]

    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        text_qa_template = self._text_qa_template.partial_format(query_str=query_str)
        single_text_chunk = "\n".join(text_chunks)
        truncated_chunks = self._prompt_helper.truncate(
            prompt=text_qa_template,
            text_chunks=[single_text_chunk],
            llm=self._llm,
        )

        response: RESPONSE_TEXT_TYPE
        if not self._streaming:
            response = self._llm.predict(
                text_qa_template,
                context_str=truncated_chunks,
                **kwargs,
            )
        else:
            response = self._llm.stream(
                text_qa_template,
                context_str=truncated_chunks,
                **kwargs,
            )

        if isinstance(response, str):
            response = response or "Empty Response"
        else:
            response = cast(Generator, response)

        return response

    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        text_qa_template = self._text_qa_template.partial_format(query_str=query_str)
        single_text_chunk = "\n".join(text_chunks)
        truncated_chunks = self._prompt_helper.truncate(
            prompt=text_qa_template,
            text_chunks=[single_text_chunk],
            llm=self._llm,
        )

        response: RESPONSE_TEXT_TYPE
        if not self._streaming:
            response = await self._llm.apredict(
                text_qa_template,
                context_str=truncated_chunks,
                **response_kwargs,
            )
        else:
            response = await self._llm.astream(
                text_qa_template,
                context_str=truncated_chunks,
                **response_kwargs,
            )

        if isinstance(response, str):
            response = response or "Empty Response"
        else:
            response = cast(Generator, response)

        return response

    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        chat_content_qa_template = self._chat_content_qa_template.partial_format(
            query_str=query_str
        )
        single_message_chunk = ChatMessage.merge(
            splits=message_chunks,
            chunk_size=sum(m.estimate_tokens() for m in message_chunks),
        )[0]
        truncated_chunks = self._chat_prompt_helper.truncate(
            prompt=chat_content_qa_template,
            messages=[single_message_chunk],
            llm=self._llm,
        )

        response: RESPONSE_TEXT_TYPE
        if not self._streaming:
            response = self._llm.predict(
                chat_content_qa_template,
                context_messages=truncated_chunks,
                **kwargs,
            )
        else:
            response = self._llm.stream(
                chat_content_qa_template,
                context_messages=truncated_chunks,
                **kwargs,
            )

        if isinstance(response, str):
            response = response or "Empty Response"
        else:
            response = cast(Generator, response)

        return response

    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        chat_content_qa_template = self._chat_content_qa_template.partial_format(
            query_str=query_str
        )
        single_message_chunk = ChatMessage.merge(
            splits=message_chunks,
            chunk_size=sum(m.estimate_tokens() for m in message_chunks),
        )[0]
        truncated_chunks = await self._chat_prompt_helper.atruncate(
            prompt=chat_content_qa_template,
            messages=[single_message_chunk],
            llm=self._llm,
        )

        response: RESPONSE_TEXT_TYPE
        if not self._streaming:
            response = await self._llm.apredict(
                chat_content_qa_template,
                context_messages=truncated_chunks,
                **response_kwargs,
            )
        else:
            response = await self._llm.astream(
                chat_content_qa_template,
                context_messages=truncated_chunks,
                **response_kwargs,
            )

        if isinstance(response, str):
            response = response or "Empty Response"
        else:
            response = cast(Generator, response)

        return response

TreeSummarize

Bases: BaseSynthesizer

Tree summarize response builder.

This response builder recursively merges text chunks and summarizes them in a bottom-up fashion (i.e. building a tree from leaves to root).

More concretely, at each recursively step: 1. we repack the text chunks so that each chunk fills the context window of the LLM 2. if there is only one chunk, we give the final response 3. otherwise, we summarize each chunk and recursively summarize the summaries.

Source code in llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

class TreeSummarize(BaseSynthesizer):
    """
    Tree summarize response builder.

    This response builder recursively merges text chunks and summarizes them
    in a bottom-up fashion (i.e. building a tree from leaves to root).

    More concretely, at each recursively step:
    1. we repack the text chunks so that each chunk fills the context window of the LLM
    2. if there is only one chunk, we give the final response
    3. otherwise, we summarize each chunk and recursively summarize the summaries.
    """

    def __init__(
        self,
        llm: Optional[LLM] = None,
        callback_manager: Optional[CallbackManager] = None,
        prompt_helper: Optional[PromptHelper] = None,
        chat_prompt_helper: Optional[ChatPromptHelper] = None,
        summary_template: Optional[BasePromptTemplate] = None,
        chat_summary_template: Optional[BasePromptTemplate] = None,
        output_cls: Optional[Type[BaseModel]] = None,
        streaming: bool = False,
        use_async: bool = False,
        verbose: bool = False,
        multimodal: bool = False,
    ) -> None:
        super().__init__(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            streaming=streaming,
            output_cls=output_cls,
            multimodal=multimodal,
        )
        self._summary_template = summary_template or DEFAULT_TREE_SUMMARIZE_PROMPT_SEL
        self._chat_summary_template = (
            chat_summary_template or CHAT_CONTENT_TREE_SUMMARIZE_PROMPT
        )
        self._use_async = use_async
        self._verbose = verbose

    def _get_prompts(self) -> PromptDictType:
        """Get prompts."""
        return {
            "summary_template": self._summary_template,
            "chat_summary_template": self._chat_summary_template,
        }

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """Update prompts."""
        if "summary_template" in prompts:
            self._summary_template = prompts["summary_template"]
        if "chat_summary_template" in prompts:
            self._chat_summary_template = prompts["chat_summary_template"]

    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get tree summarize response."""
        summary_template = self._summary_template.partial_format(query_str=query_str)
        # repack text_chunks so that each chunk fills the context window
        text_chunks = self._prompt_helper.repack(
            summary_template, text_chunks=text_chunks, llm=self._llm
        )

        if self._verbose:
            print(f"{len(text_chunks)} text chunks after repacking")

        # give final response if there is only one chunk
        if len(text_chunks) == 1:
            response: RESPONSE_TEXT_TYPE
            if self._streaming:
                response = await self._llm.astream(
                    summary_template, context_str=text_chunks[0], **response_kwargs
                )
            else:
                if self._output_cls is None:
                    response = await self._llm.apredict(
                        summary_template,
                        context_str=text_chunks[0],
                        **response_kwargs,
                    )
                else:
                    response = await self._llm.astructured_predict(
                        self._output_cls,
                        summary_template,
                        context_str=text_chunks[0],
                        **response_kwargs,
                    )

            # return pydantic object if output_cls is specified
            return response

        else:
            # summarize each chunk
            if self._output_cls is None:
                str_tasks = [
                    self._llm.apredict(
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]
                summaries = await asyncio.gather(*str_tasks)
            else:
                model_tasks = [
                    self._llm.astructured_predict(
                        self._output_cls,
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]
                summary_models = await asyncio.gather(*model_tasks)
                summaries = [
                    summary.model_dump_json()
                    if isinstance(summary, BaseModel)
                    else str(summary)
                    for summary in summary_models
                ]

            # recursively summarize the summaries
            return await self.aget_response(
                query_str=query_str,
                text_chunks=summaries,
                **response_kwargs,
            )

    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        """Get tree summarize response."""
        summary_template = self._summary_template.partial_format(query_str=query_str)
        # repack text_chunks so that each chunk fills the context window
        text_chunks = self._prompt_helper.repack(
            summary_template, text_chunks=text_chunks, llm=self._llm
        )

        if self._verbose:
            print(f"{len(text_chunks)} text chunks after repacking")

        # give final response if there is only one chunk
        if len(text_chunks) == 1:
            response: RESPONSE_TEXT_TYPE
            if self._streaming:
                response = self._llm.stream(
                    summary_template, context_str=text_chunks[0], **response_kwargs
                )
            else:
                if self._output_cls is None:
                    response = self._llm.predict(
                        summary_template,
                        context_str=text_chunks[0],
                        **response_kwargs,
                    )
                else:
                    response = self._llm.structured_predict(
                        self._output_cls,
                        summary_template,
                        context_str=text_chunks[0],
                        **response_kwargs,
                    )

            return response

        else:
            # summarize each chunk
            if self._use_async:
                if self._output_cls is None:
                    tasks = [
                        self._llm.apredict(
                            summary_template,
                            context_str=text_chunk,
                            **response_kwargs,
                        )
                        for text_chunk in text_chunks
                    ]
                else:
                    tasks = [
                        self._llm.astructured_predict(
                            self._output_cls,
                            summary_template,
                            context_str=text_chunk,
                            **response_kwargs,
                        )
                        for text_chunk in text_chunks
                    ]

                summary_responses = run_async_tasks(tasks)

                if self._output_cls is not None:
                    summaries = [
                        summary.model_dump_json()
                        if isinstance(summary, BaseModel)
                        else str(summary)
                        for summary in summary_responses
                    ]
                else:
                    summaries = summary_responses
            else:
                if self._output_cls is None:
                    summaries = [
                        self._llm.predict(
                            summary_template,
                            context_str=text_chunk,
                            **response_kwargs,
                        )
                        for text_chunk in text_chunks
                    ]
                else:
                    summary_models = [
                        self._llm.structured_predict(
                            self._output_cls,
                            summary_template,
                            context_str=text_chunk,
                            **response_kwargs,
                        )
                        for text_chunk in text_chunks
                    ]
                    summaries = [
                        summary.model_dump_json()
                        if isinstance(summary, BaseModel)
                        else str(summary)
                        for summary in summary_models
                    ]

            # recursively summarize the summaries
            return self.get_response(
                query_str=query_str, text_chunks=summaries, **response_kwargs
            )

    async def aget_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        summary_template = self._chat_summary_template.partial_format(
            query_str=query_str
        )
        message_chunks = self._chat_prompt_helper.repack(
            summary_template, messages=list(message_chunks), llm=self._llm
        )

        if self._verbose:
            print(f"{len(message_chunks)} message chunks after repacking")

        if len(message_chunks) == 1:
            response: RESPONSE_TEXT_TYPE
            if self._streaming:
                response = await self._llm.astream(
                    summary_template,
                    context_messages=[message_chunks[0]],
                    **response_kwargs,
                )
            else:
                if self._output_cls is None:
                    response = await self._llm.apredict(
                        summary_template,
                        context_messages=[message_chunks[0]],
                        **response_kwargs,
                    )
                else:
                    response = await self._llm.astructured_predict(
                        self._output_cls,
                        summary_template,
                        context_messages=[message_chunks[0]],
                        **response_kwargs,
                    )
            return response

        else:
            if self._output_cls is None:
                str_tasks = [
                    self._llm.apredict(
                        summary_template,
                        context_messages=[chunk],
                        **response_kwargs,
                    )
                    for chunk in message_chunks
                ]
                summaries = await asyncio.gather(*str_tasks)
            else:
                model_tasks = [
                    self._llm.astructured_predict(
                        self._output_cls,
                        summary_template,
                        context_messages=[chunk],
                        **response_kwargs,
                    )
                    for chunk in message_chunks
                ]
                summary_models = await asyncio.gather(*model_tasks)
                summaries = [summary.model_dump_json() for summary in summary_models]

            return await self.aget_response_from_messages(
                query_str=query_str,
                message_chunks=[ChatMessage(content=s) for s in summaries],
                **response_kwargs,
            )

    def get_response_from_messages(
        self,
        query_str: str,
        message_chunks: Sequence[ChatMessage],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        summary_template = self._chat_summary_template.partial_format(
            query_str=query_str
        )
        message_chunks = self._chat_prompt_helper.repack(
            summary_template, messages=list(message_chunks), llm=self._llm
        )

        if self._verbose:
            print(f"{len(message_chunks)} message chunks after repacking")

        if len(message_chunks) == 1:
            response: RESPONSE_TEXT_TYPE
            if self._streaming:
                response = self._llm.stream(
                    summary_template,
                    context_messages=[message_chunks[0]],
                    **response_kwargs,
                )
            else:
                if self._output_cls is None:
                    response = self._llm.predict(
                        summary_template,
                        context_messages=[message_chunks[0]],
                        **response_kwargs,
                    )
                else:
                    response = self._llm.structured_predict(
                        self._output_cls,
                        summary_template,
                        context_messages=[message_chunks[0]],
                        **response_kwargs,
                    )
            return response

        else:
            if self._use_async:
                if self._output_cls is None:
                    tasks = [
                        self._llm.apredict(
                            summary_template,
                            context_messages=[chunk],
                            **response_kwargs,
                        )
                        for chunk in message_chunks
                    ]
                else:
                    tasks = [
                        self._llm.astructured_predict(
                            self._output_cls,
                            summary_template,
                            context_messages=[chunk],
                            **response_kwargs,
                        )
                        for chunk in message_chunks
                    ]

                summary_responses = run_async_tasks(tasks)

                if self._output_cls is not None:
                    summaries = [s.model_dump_json() for s in summary_responses]
                else:
                    summaries = summary_responses
            else:
                if self._output_cls is None:
                    summaries = [
                        self._llm.predict(
                            summary_template,
                            context_messages=[chunk],
                            **response_kwargs,
                        )
                        for chunk in message_chunks
                    ]
                else:
                    summaries = [
                        self._llm.structured_predict(
                            self._output_cls,
                            summary_template,
                            context_messages=[chunk],
                            **response_kwargs,
                        )
                        for chunk in message_chunks
                    ]
                    summaries = [s.model_dump_json() for s in summaries]

            return self.get_response_from_messages(
                query_str=query_str,
                message_chunks=[ChatMessage(content=s) for s in summaries],
                **response_kwargs,
            )

aget_response async

aget_response(
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Get tree summarize response.

Source code in llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

async def aget_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Get tree summarize response."""
    summary_template = self._summary_template.partial_format(query_str=query_str)
    # repack text_chunks so that each chunk fills the context window
    text_chunks = self._prompt_helper.repack(
        summary_template, text_chunks=text_chunks, llm=self._llm
    )

    if self._verbose:
        print(f"{len(text_chunks)} text chunks after repacking")

    # give final response if there is only one chunk
    if len(text_chunks) == 1:
        response: RESPONSE_TEXT_TYPE
        if self._streaming:
            response = await self._llm.astream(
                summary_template, context_str=text_chunks[0], **response_kwargs
            )
        else:
            if self._output_cls is None:
                response = await self._llm.apredict(
                    summary_template,
                    context_str=text_chunks[0],
                    **response_kwargs,
                )
            else:
                response = await self._llm.astructured_predict(
                    self._output_cls,
                    summary_template,
                    context_str=text_chunks[0],
                    **response_kwargs,
                )

        # return pydantic object if output_cls is specified
        return response

    else:
        # summarize each chunk
        if self._output_cls is None:
            str_tasks = [
                self._llm.apredict(
                    summary_template,
                    context_str=text_chunk,
                    **response_kwargs,
                )
                for text_chunk in text_chunks
            ]
            summaries = await asyncio.gather(*str_tasks)
        else:
            model_tasks = [
                self._llm.astructured_predict(
                    self._output_cls,
                    summary_template,
                    context_str=text_chunk,
                    **response_kwargs,
                )
                for text_chunk in text_chunks
            ]
            summary_models = await asyncio.gather(*model_tasks)
            summaries = [
                summary.model_dump_json()
                if isinstance(summary, BaseModel)
                else str(summary)
                for summary in summary_models
            ]

        # recursively summarize the summaries
        return await self.aget_response(
            query_str=query_str,
            text_chunks=summaries,
            **response_kwargs,
        )

get_response

get_response(
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any
) -> RESPONSE_TEXT_TYPE

Get tree summarize response.

Source code in llama-index-core/llama_index/core/response_synthesizers/tree_summarize.py

def get_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any,
) -> RESPONSE_TEXT_TYPE:
    """Get tree summarize response."""
    summary_template = self._summary_template.partial_format(query_str=query_str)
    # repack text_chunks so that each chunk fills the context window
    text_chunks = self._prompt_helper.repack(
        summary_template, text_chunks=text_chunks, llm=self._llm
    )

    if self._verbose:
        print(f"{len(text_chunks)} text chunks after repacking")

    # give final response if there is only one chunk
    if len(text_chunks) == 1:
        response: RESPONSE_TEXT_TYPE
        if self._streaming:
            response = self._llm.stream(
                summary_template, context_str=text_chunks[0], **response_kwargs
            )
        else:
            if self._output_cls is None:
                response = self._llm.predict(
                    summary_template,
                    context_str=text_chunks[0],
                    **response_kwargs,
                )
            else:
                response = self._llm.structured_predict(
                    self._output_cls,
                    summary_template,
                    context_str=text_chunks[0],
                    **response_kwargs,
                )

        return response

    else:
        # summarize each chunk
        if self._use_async:
            if self._output_cls is None:
                tasks = [
                    self._llm.apredict(
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]
            else:
                tasks = [
                    self._llm.astructured_predict(
                        self._output_cls,
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]

            summary_responses = run_async_tasks(tasks)

            if self._output_cls is not None:
                summaries = [
                    summary.model_dump_json()
                    if isinstance(summary, BaseModel)
                    else str(summary)
                    for summary in summary_responses
                ]
            else:
                summaries = summary_responses
        else:
            if self._output_cls is None:
                summaries = [
                    self._llm.predict(
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]
            else:
                summary_models = [
                    self._llm.structured_predict(
                        self._output_cls,
                        summary_template,
                        context_str=text_chunk,
                        **response_kwargs,
                    )
                    for text_chunk in text_chunks
                ]
                summaries = [
                    summary.model_dump_json()
                    if isinstance(summary, BaseModel)
                    else str(summary)
                    for summary in summary_models
                ]

        # recursively summarize the summaries
        return self.get_response(
            query_str=query_str, text_chunks=summaries, **response_kwargs
        )

ResponseMode

Bases: str, Enum

Response modes of the response builder (and synthesizer).

Source code in llama-index-core/llama_index/core/response_synthesizers/type.py

class ResponseMode(str, Enum):
    """Response modes of the response builder (and synthesizer)."""

    REFINE = "refine"
    """
    Refine is an iterative way of generating a response.
    We first use the context in the first node, along with the query, to generate an \
    initial answer.
    We then pass this answer, the query, and the context of the second node as input \
    into a “refine prompt” to generate a refined answer. We refine through N-1 nodes, \
    where N is the total number of nodes.
    """

    COMPACT = "compact"
    """
    Compact and refine mode first combine text chunks into larger consolidated chunks \
    that more fully utilize the available context window, then refine answers \
    across them.
    This mode is faster than refine since we make fewer calls to the LLM.
    """

    SIMPLE_SUMMARIZE = "simple_summarize"
    """
    Merge all text chunks into one, and make a LLM call.
    This will fail if the merged text chunk exceeds the context window size.
    """

    TREE_SUMMARIZE = "tree_summarize"
    """
    Build a tree index over the set of candidate nodes, with a summary prompt seeded \
    with the query.
    The tree is built in a bottoms-up fashion, and in the end the root node is \
    returned as the response
    """

    GENERATION = "generation"
    """Ignore context, just use LLM to generate a response."""

    NO_TEXT = "no_text"
    """Return the retrieved context nodes, without synthesizing a final response."""

    CONTEXT_ONLY = "context_only"
    """Returns a concatenated string of all text chunks."""

    ACCUMULATE = "accumulate"
    """Synthesize a response for each text chunk, and then return the concatenation."""

    COMPACT_ACCUMULATE = "compact_accumulate"
    """
    Compact and accumulate mode first combine text chunks into larger consolidated \
    chunks that more fully utilize the available context window, then accumulate \
    answers for each of them and finally return the concatenation.
    This mode is faster than accumulate since we make fewer calls to the LLM.
    """

REFINE class-attribute instance-attribute

REFINE = 'refine'

Refine is an iterative way of generating a response. We first use the context in the first node, along with the query, to generate an initial answer. We then pass this answer, the query, and the context of the second node as input into a “refine prompt” to generate a refined answer. We refine through N-1 nodes, where N is the total number of nodes.

COMPACT class-attribute instance-attribute

COMPACT = 'compact'

Compact and refine mode first combine text chunks into larger consolidated chunks that more fully utilize the available context window, then refine answers across them. This mode is faster than refine since we make fewer calls to the LLM.

SIMPLE_SUMMARIZE class-attribute instance-attribute

SIMPLE_SUMMARIZE = 'simple_summarize'

Merge all text chunks into one, and make a LLM call. This will fail if the merged text chunk exceeds the context window size.

TREE_SUMMARIZE class-attribute instance-attribute

TREE_SUMMARIZE = 'tree_summarize'

Build a tree index over the set of candidate nodes, with a summary prompt seeded with the query. The tree is built in a bottoms-up fashion, and in the end the root node is returned as the response

GENERATION class-attribute instance-attribute

GENERATION = 'generation'

Ignore context, just use LLM to generate a response.

NO_TEXT class-attribute instance-attribute

NO_TEXT = 'no_text'

Return the retrieved context nodes, without synthesizing a final response.

CONTEXT_ONLY class-attribute instance-attribute

CONTEXT_ONLY = 'context_only'

Returns a concatenated string of all text chunks.

ACCUMULATE class-attribute instance-attribute

ACCUMULATE = 'accumulate'

Synthesize a response for each text chunk, and then return the concatenation.

COMPACT_ACCUMULATE class-attribute instance-attribute

COMPACT_ACCUMULATE = 'compact_accumulate'

Compact and accumulate mode first combine text chunks into larger consolidated chunks that more fully utilize the available context window, then accumulate answers for each of them and finally return the concatenation. This mode is faster than accumulate since we make fewer calls to the LLM.

get_response_synthesizer

get_response_synthesizer(
    llm: Optional[LLM] = None,
    prompt_helper: Optional[PromptHelper] = None,
    chat_prompt_helper: Optional[ChatPromptHelper] = None,
    text_qa_template: Optional[BasePromptTemplate] = None,
    refine_template: Optional[BasePromptTemplate] = None,
    summary_template: Optional[BasePromptTemplate] = None,
    simple_template: Optional[BasePromptTemplate] = None,
    chat_content_qa_template: Optional[
        BasePromptTemplate
    ] = None,
    chat_content_refine_template: Optional[
        BasePromptTemplate
    ] = None,
    chat_summary_template: Optional[
        BasePromptTemplate
    ] = None,
    response_mode: ResponseMode = COMPACT,
    callback_manager: Optional[CallbackManager] = None,
    use_async: bool = False,
    streaming: bool = False,
    structured_answer_filtering: bool = False,
    output_cls: Optional[Type[BaseModel]] = None,
    program_factory: Optional[
        Callable[[BasePromptTemplate], BasePydanticProgram]
    ] = None,
    verbose: bool = False,
    multimodal: bool = False,
) -> BaseSynthesizer

Get a response synthesizer.

Source code in llama-index-core/llama_index/core/response_synthesizers/factory.py

def get_response_synthesizer(
    llm: Optional[LLM] = None,
    prompt_helper: Optional[PromptHelper] = None,
    chat_prompt_helper: Optional[ChatPromptHelper] = None,
    text_qa_template: Optional[BasePromptTemplate] = None,
    refine_template: Optional[BasePromptTemplate] = None,
    summary_template: Optional[BasePromptTemplate] = None,
    simple_template: Optional[BasePromptTemplate] = None,
    chat_content_qa_template: Optional[BasePromptTemplate] = None,
    chat_content_refine_template: Optional[BasePromptTemplate] = None,
    chat_summary_template: Optional[BasePromptTemplate] = None,
    response_mode: ResponseMode = ResponseMode.COMPACT,
    callback_manager: Optional[CallbackManager] = None,
    use_async: bool = False,
    streaming: bool = False,
    structured_answer_filtering: bool = False,
    output_cls: Optional[Type[BaseModel]] = None,
    program_factory: Optional[
        Callable[[BasePromptTemplate], BasePydanticProgram]
    ] = None,
    verbose: bool = False,
    multimodal: bool = False,
) -> BaseSynthesizer:
    """Get a response synthesizer."""
    text_qa_template = text_qa_template or DEFAULT_TEXT_QA_PROMPT_SEL
    refine_template = refine_template or DEFAULT_REFINE_PROMPT_SEL
    simple_template = simple_template or DEFAULT_SIMPLE_INPUT_PROMPT
    summary_template = summary_template or DEFAULT_TREE_SUMMARIZE_PROMPT_SEL

    chat_content_qa_template = chat_content_qa_template or CHAT_CONTENT_QA_PROMPT
    chat_content_refine_template = (
        chat_content_refine_template or CHAT_CONTENT_REFINE_PROMPT
    )
    chat_summary_template = chat_summary_template or CHAT_CONTENT_TREE_SUMMARIZE_PROMPT

    callback_manager = callback_manager or Settings.callback_manager
    llm = llm or Settings.llm
    prompt_helper = (
        prompt_helper
        or Settings._prompt_helper
        or PromptHelper.from_llm_metadata(
            llm.metadata,
        )
    )
    chat_prompt_helper = (
        chat_prompt_helper
        or Settings._chat_prompt_helper
        or ChatPromptHelper.from_llm_metadata(
            llm.metadata,
        )
    )

    if response_mode == ResponseMode.REFINE:
        return Refine(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            text_qa_template=text_qa_template,
            refine_template=refine_template,
            chat_content_qa_template=chat_content_qa_template,
            chat_content_refine_template=chat_content_refine_template,
            output_cls=output_cls,
            streaming=streaming,
            structured_answer_filtering=structured_answer_filtering,
            program_factory=program_factory,
            verbose=verbose,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.COMPACT:
        return CompactAndRefine(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            text_qa_template=text_qa_template,
            refine_template=refine_template,
            chat_content_qa_template=chat_content_qa_template,
            chat_content_refine_template=chat_content_refine_template,
            output_cls=output_cls,
            streaming=streaming,
            structured_answer_filtering=structured_answer_filtering,
            program_factory=program_factory,
            verbose=verbose,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.TREE_SUMMARIZE:
        return TreeSummarize(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            summary_template=summary_template,
            chat_summary_template=chat_summary_template,
            output_cls=output_cls,
            streaming=streaming,
            use_async=use_async,
            verbose=verbose,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.SIMPLE_SUMMARIZE:
        return SimpleSummarize(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            text_qa_template=text_qa_template,
            chat_content_qa_template=chat_content_qa_template,
            streaming=streaming,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.GENERATION:
        return Generation(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            simple_template=simple_template,
            streaming=streaming,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.ACCUMULATE:
        return Accumulate(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            text_qa_template=text_qa_template,
            chat_content_qa_template=chat_content_qa_template,
            output_cls=output_cls,
            streaming=streaming,
            use_async=use_async,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.COMPACT_ACCUMULATE:
        return CompactAndAccumulate(
            llm=llm,
            callback_manager=callback_manager,
            prompt_helper=prompt_helper,
            chat_prompt_helper=chat_prompt_helper,
            text_qa_template=text_qa_template,
            chat_content_qa_template=chat_content_qa_template,
            output_cls=output_cls,
            streaming=streaming,
            use_async=use_async,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.NO_TEXT:
        return NoText(
            callback_manager=callback_manager,
            streaming=streaming,
            multimodal=multimodal,
        )
    elif response_mode == ResponseMode.CONTEXT_ONLY:
        return ContextOnly(
            callback_manager=callback_manager,
            streaming=streaming,
            multimodal=multimodal,
        )
    else:
        raise ValueError(f"Unknown mode: {response_mode}")

options: members: - TreeSummarize