code: """Response builder class.
  
  This class provides general functions for taking in a set of text
  and generating a response.
  
  Will support different modes, from 1) stuffing chunks into prompt,
  2) create and refine separately over each chunk, 3) tree summarization.
  
  """
  
  import logging
  from abc import abstractmethod
  from typing import Any, Dict, Generator, List, Optional, Sequence, AsyncGenerator
  
  from llama_index.core.base.query_pipeline.query import (
      ChainableMixin,
      InputKeys,
      OutputKeys,
      QueryComponent,
      validate_and_convert_stringable,
  )
  from llama_index.core.base.response.schema import (
      RESPONSE_TYPE,
      PydanticResponse,
      Response,
      StreamingResponse,
      AsyncStreamingResponse,
  )
  from llama_index.core.bridge.pydantic import BaseModel, Field
  from llama_index.core.callbacks.base import CallbackManager
  from llama_index.core.callbacks.schema import CBEventType, EventPayload
  from llama_index.core.indices.prompt_helper import PromptHelper
  from llama_index.core.prompts.mixin import PromptMixin
  from llama_index.core.schema import (
      BaseNode,
      MetadataMode,
      NodeWithScore,
      QueryBundle,
      QueryType,
  )
  from llama_index.core.service_context import ServiceContext
  from llama_index.core.service_context_elements.llm_predictor import LLMPredictorType
  from llama_index.core.settings import (
      Settings,
      callback_manager_from_settings_or_context,
      llm_from_settings_or_context,
  )
  from llama_index.core.types import RESPONSE_TEXT_TYPE
  from llama_index.core.instrumentation.events.synthesis import (
      SynthesizeStartEvent,
      SynthesizeEndEvent,
  )
  import llama_index.core.instrumentation as instrument
  
  dispatcher = instrument.get_dispatcher(__name__)
  
  logger = logging.getLogger(__name__)
  
  QueryTextType = QueryType
  
  
  def empty_response_generator() -> Generator[str, None, None]:
      yield "Empty Response"
  
  
  async def empty_response_agenerator() -> AsyncGenerator[str, None]:
      yield "Empty Response"
  
  
  class BaseSynthesizer(ChainableMixin, PromptMixin):
      """Response builder class."""
  
      def __init__(
          self,
          llm: Optional[LLMPredictorType] = None,
          callback_manager: Optional[CallbackManager] = None,
          prompt_helper: Optional[PromptHelper] = None,
          streaming: bool = False,
          output_cls: BaseModel = None,
          # deprecated
          service_context: Optional[ServiceContext] = None,
      ) -> None:
          """Init params."""
          self._llm = llm or llm_from_settings_or_context(Settings, service_context)
  
          if callback_manager:
              self._llm.callback_manager = callback_manager
  
          self._callback_manager = (
              callback_manager
              or callback_manager_from_settings_or_context(Settings, service_context)
          )
  
          self._prompt_helper = (
              prompt_helper
              or Settings._prompt_helper
              or PromptHelper.from_llm_metadata(
                  self._llm.metadata,
              )
          )
  
          self._streaming = streaming
          self._output_cls = output_cls
  
      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 _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(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 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 = StreamingResponse(
                      response_gen=empty_response_generator()
                  )
                  dispatcher.event(
                      SynthesizeEndEvent(query=query, response=str(empty_response))
                  )
                  return empty_response
              else:
                  empty_response = Response("Empty Response")
                  dispatcher.event(
                      SynthesizeEndEvent(query=query, response=str(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:
              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=str(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 = AsyncStreamingResponse(
                      response_gen=empty_response_agenerator()
                  )
                  dispatcher.event(
                      SynthesizeEndEvent(query=query, response=str(empty_response))
                  )
                  return empty_response
              else:
                  empty_response = Response("Empty Response")
                  dispatcher.event(
                      SynthesizeEndEvent(query=query, response=str(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:
              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=str(response)))
          return response
  
      def _as_query_component(self, **kwargs: Any) -> QueryComponent:
          """As query component."""
          return SynthesizerComponent(synthesizer=self)
  
  
  class SynthesizerComponent(QueryComponent):
      """Synthesizer component."""
  
      synthesizer: BaseSynthesizer = Field(..., description="Synthesizer")
  
      class Config:
          arbitrary_types_allowed = True
  
      def set_callback_manager(self, callback_manager: CallbackManager) -> None:
          """Set callback manager."""
          self.synthesizer.callback_manager = callback_manager
  
      def _validate_component_inputs(self, input: Dict[str, Any]) -> Dict[str, Any]:
          """Validate component inputs during run_component."""
          # make sure both query_str and nodes are there
          if "query_str" not in input:
              raise ValueError("Input must have key 'query_str'")
          input["query_str"] = validate_and_convert_stringable(input["query_str"])
  
          if "nodes" not in input:
              raise ValueError("Input must have key 'nodes'")
          nodes = input["nodes"]
          if not isinstance(nodes, list):
              raise ValueError("Input nodes must be a list")
          for node in nodes:
              if not isinstance(node, NodeWithScore):
                  raise ValueError("Input nodes must be a list of NodeWithScore")
          return input
  
      def _run_component(self, **kwargs: Any) -> Dict[str, Any]:
          """Run component."""
          output = self.synthesizer.synthesize(kwargs["query_str"], kwargs["nodes"])
          return {"output": output}
  
      async def _arun_component(self, **kwargs: Any) -> Dict[str, Any]:
          """Run component."""
          output = await self.synthesizer.asynthesize(
              kwargs["query_str"], kwargs["nodes"]
          )
          return {"output": output}
  
      @property
      def input_keys(self) -> InputKeys:
          """Input keys."""
          return InputKeys.from_keys({"query_str", "nodes"})
  
      @property
      def output_keys(self) -> OutputKeys:
          """Output keys."""
          return OutputKeys.from_keys({"output"})
  
apis: llama_index.core.instrumentation.events.synthesis.SynthesizeStartEvent llama_index.core.settings.llm_from_settings_or_context llama_index.core.base.query_pipeline.query.InputKeys.from_keys llama_index.core.settings.callback_manager_from_settings_or_context llama_index.core.base.response.schema.AsyncStreamingResponse llama_index.core.bridge.pydantic.Field llama_index.core.instrumentation.get_dispatcher llama_index.core.base.query_pipeline.query.OutputKeys.from_keys llama_index.core.base.response.schema.PydanticResponse llama_index.core.indices.prompt_helper.PromptHelper.from_llm_metadata llama_index.core.base.response.schema.Response llama_index.core.schema.QueryBundle llama_index.core.base.response.schema.StreamingResponse llama_index.core.base.query_pipeline.query.validate_and_convert_stringable
extract_api: [((1679, 1714), 'llama_index.core.instrumentation.get_dispatcher', 'instrument.get_dispatcher', (['__name__'], {}), '(__name__)\n', (1704, 1714), True, 'import llama_index.core.instrumentation as instrument\n'), ((1725, 1752), 'logging.getLogger', 'logging.getLogger', (['__name__'], {}), '(__name__)\n', (1742, 1752), False, 'import logging\n'), ((9986, 10023), 'llama_index.core.bridge.pydantic.Field', 'Field', (['...'], {'description': '"""Synthesizer"""'}), "(..., description='Synthesizer')\n", (9991, 10023), False, 'from llama_index.core.bridge.pydantic import BaseModel, Field\n'), ((10590, 10641), 'llama_index.core.base.query_pipeline.query.validate_and_convert_stringable', 'validate_and_convert_stringable', (["input['query_str']"], {}), "(input['query_str'])\n", (10621, 10641), False, 'from llama_index.core.base.query_pipeline.query import ChainableMixin, InputKeys, OutputKeys, QueryComponent, validate_and_convert_stringable\n'), ((11597, 11640), 'llama_index.core.base.query_pipeline.query.InputKeys.from_keys', 'InputKeys.from_keys', (["{'query_str', 'nodes'}"], {}), "({'query_str', 'nodes'})\n", (11616, 11640), False, 'from llama_index.core.base.query_pipeline.query import ChainableMixin, InputKeys, OutputKeys, QueryComponent, validate_and_convert_stringable\n'), ((11739, 11771), 'llama_index.core.base.query_pipeline.query.OutputKeys.from_keys', 'OutputKeys.from_keys', (["{'output'}"], {}), "({'output'})\n", (11759, 11771), False, 'from llama_index.core.base.query_pipeline.query import ChainableMixin, InputKeys, OutputKeys, QueryComponent, validate_and_convert_stringable\n'), ((2470, 2525), 'llama_index.core.settings.llm_from_settings_or_context', 'llm_from_settings_or_context', (['Settings', 'service_context'], {}), '(Settings, service_context)\n', (2498, 2525), False, 'from llama_index.core.settings import Settings, callback_manager_from_settings_or_context, llm_from_settings_or_context\n'), ((2694, 2762), 'llama_index.core.settings.callback_manager_from_settings_or_context', 'callback_manager_from_settings_or_context', (['Settings', 'service_context'], {}), '(Settings, service_context)\n', (2735, 2762), False, 'from llama_index.core.settings import Settings, callback_manager_from_settings_or_context, llm_from_settings_or_context\n'), ((2886, 2936), 'llama_index.core.indices.prompt_helper.PromptHelper.from_llm_metadata', 'PromptHelper.from_llm_metadata', (['self._llm.metadata'], {}), '(self._llm.metadata)\n', (2916, 2936), False, 'from llama_index.core.indices.prompt_helper import PromptHelper\n'), ((5135, 5212), 'llama_index.core.base.response.schema.Response', 'Response', (['response_str'], {'source_nodes': 'source_nodes', 'metadata': 'response_metadata'}), '(response_str, source_nodes=source_nodes, metadata=response_metadata)\n', (5143, 5212), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n'), ((5343, 5434), 'llama_index.core.base.response.schema.StreamingResponse', 'StreamingResponse', (['response_str'], {'source_nodes': 'source_nodes', 'metadata': 'response_metadata'}), '(response_str, source_nodes=source_nodes, metadata=\n    response_metadata)\n', (5360, 5434), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n'), ((5565, 5661), 'llama_index.core.base.response.schema.AsyncStreamingResponse', 'AsyncStreamingResponse', (['response_str'], {'source_nodes': 'source_nodes', 'metadata': 'response_metadata'}), '(response_str, source_nodes=source_nodes, metadata=\n    response_metadata)\n', (5587, 5661), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n'), ((5794, 5884), 'llama_index.core.base.response.schema.PydanticResponse', 'PydanticResponse', (['response_str'], {'source_nodes': 'source_nodes', 'metadata': 'response_metadata'}), '(response_str, source_nodes=source_nodes, metadata=\n    response_metadata)\n', (5810, 5884), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n'), ((6309, 6342), 'llama_index.core.instrumentation.events.synthesis.SynthesizeStartEvent', 'SynthesizeStartEvent', ([], {'query': 'query'}), '(query=query)\n', (6329, 6342), False, 'from llama_index.core.instrumentation.events.synthesis import SynthesizeStartEvent, SynthesizeEndEvent\n'), ((7013, 7041), 'llama_index.core.schema.QueryBundle', 'QueryBundle', ([], {'query_str': 'query'}), '(query_str=query)\n', (7024, 7041), False, 'from llama_index.core.schema import BaseNode, MetadataMode, NodeWithScore, QueryBundle, QueryType\n'), ((8148, 8181), 'llama_index.core.instrumentation.events.synthesis.SynthesizeStartEvent', 'SynthesizeStartEvent', ([], {'query': 'query'}), '(query=query)\n', (8168, 8181), False, 'from llama_index.core.instrumentation.events.synthesis import SynthesizeStartEvent, SynthesizeEndEvent\n'), ((8857, 8885), 'llama_index.core.schema.QueryBundle', 'QueryBundle', ([], {'query_str': 'query'}), '(query_str=query)\n', (8868, 8885), False, 'from llama_index.core.schema import BaseNode, MetadataMode, NodeWithScore, QueryBundle, QueryType\n'), ((6758, 6784), 'llama_index.core.base.response.schema.Response', 'Response', (['"""Empty Response"""'], {}), "('Empty Response')\n", (6766, 6784), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n'), ((8602, 8628), 'llama_index.core.base.response.schema.Response', 'Response', (['"""Empty Response"""'], {}), "('Empty Response')\n", (8610, 8628), False, 'from llama_index.core.base.response.schema import RESPONSE_TYPE, PydanticResponse, Response, StreamingResponse, AsyncStreamingResponse\n')]