Google

GoogleTextSynthesizer

基类: BaseSynthesizer

Google 的归因式问答服务。

给定用户的查询和通道列表，Google 的服务器将返回一个基于提供的通道列表的响应。它不会基于参数记忆生成响应。

源码位于 llama-index-integrations/response_synthesizers/llama-index-response-synthesizers-google/llama_index/response_synthesizers/google/base.py

class GoogleTextSynthesizer(BaseSynthesizer):
    """
    Google's Attributed Question and Answering service.

    Given a user's query and a list of passages, Google's server will return
    a response that is grounded to the provided list of passages. It will not
    base the response on parametric memory.
    """

    _client: Any
    _temperature: float
    _answer_style: Any
    _safety_setting: List[Any]

    def __init__(
        self,
        *,
        temperature: float,
        answer_style: Any,
        safety_setting: List[Any],
        **kwargs: Any,
    ):
        """
        Create a new Google AQA.

        Prefer to use the factory `from_defaults` instead for type safety.
        See `from_defaults` for more documentation.
        """
        try:
            import llama_index.vector_stores.google.genai_extension as genaix
        except ImportError:
            raise ImportError(_import_err_msg)

        super().__init__(
            llm=MockLLM(),
            output_cls=SynthesizedResponse,
            **kwargs,
        )

        self._client = genaix.build_generative_service()
        self._temperature = temperature
        self._answer_style = answer_style
        self._safety_setting = safety_setting

    # Type safe factory that is only available if Google is installed.
    @classmethod
    def from_defaults(
        cls,
        temperature: float = 0.7,
        answer_style: int = 1,
        safety_setting: List["genai.SafetySetting"] = [],
    ) -> "GoogleTextSynthesizer":
        """
        Create a new Google AQA.

        Example:
          responder = GoogleTextSynthesizer.create(
              temperature=0.7,
              answer_style=AnswerStyle.ABSTRACTIVE,
              safety_setting=[
                  SafetySetting(
                      category=HARM_CATEGORY_SEXUALLY_EXPLICIT,
                      threshold=HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
                  ),
              ]
          )

        Args:
          temperature: 0.0 to 1.0.
          answer_style: See `google.ai.generativelanguage.GenerateAnswerRequest.AnswerStyle`
            The default is ABSTRACTIVE (1).
          safety_setting: See `google.ai.generativelanguage.SafetySetting`.

        Returns:
          an instance of GoogleTextSynthesizer.

        """
        return cls(
            temperature=temperature,
            answer_style=answer_style,
            safety_setting=safety_setting,
        )

    def get_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> SynthesizedResponse:
        """
        Generate a grounded response on provided passages.

        Args:
            query_str: The user's question.
            text_chunks: A list of passages that should be used to answer the
                question.

        Returns:
            A `SynthesizedResponse` object.

        """
        try:
            import llama_index.vector_stores.google.genai_extension as genaix

            import google.ai.generativelanguage as genai
        except ImportError:
            raise ImportError(_import_err_msg)

        client = cast(genai.GenerativeServiceClient, self._client)
        response = genaix.generate_answer(
            prompt=query_str,
            passages=list(text_chunks),
            answer_style=self._answer_style,
            safety_settings=self._safety_setting,
            temperature=self._temperature,
            client=client,
        )

        return SynthesizedResponse(
            answer=response.answer,
            attributed_passages=[
                passage.text for passage in response.attributed_passages
            ],
            answerable_probability=response.answerable_probability,
        )

    async def aget_response(
        self,
        query_str: str,
        text_chunks: Sequence[str],
        **response_kwargs: Any,
    ) -> RESPONSE_TEXT_TYPE:
        # TODO: Implement a true async version.
        return self.get_response(query_str, text_chunks, **response_kwargs)

    def synthesize(
        self,
        query: QueryTextType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> Response:
        """
        Returns a grounded response based on provided passages.

        Returns:
            Response's `source_nodes` will begin with a list of attributed
            passages. These passages are the ones that were used to construct
            the grounded response. These passages will always have no score,
            the only way to mark them as attributed passages. Then, the list
            will follow with the originally provided passages, which will have
            a score from the retrieval.

            Response's `metadata` may also have have an entry with key
            `answerable_probability`, which is the model's estimate of the
            probability that its answer is correct and grounded in the input
            passages.

        """
        if len(nodes) == 0:
            return Response("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:
            internal_response = 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 = list(additional_source_nodes or [])

            external_response = self._prepare_external_response(
                internal_response, nodes + additional_source_nodes
            )

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

        return external_response

    async def asynthesize(
        self,
        query: QueryTextType,
        nodes: List[NodeWithScore],
        additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
        **response_kwargs: Any,
    ) -> Response:
        # TODO: Implement a true async version.
        return self.synthesize(query, nodes, additional_source_nodes, **response_kwargs)

    def _prepare_external_response(
        self,
        response: SynthesizedResponse,
        source_nodes: List[NodeWithScore],
    ) -> Response:
        return Response(
            response=response.answer,
            source_nodes=[
                NodeWithScore(node=TextNode(text=passage))
                for passage in response.attributed_passages
            ]
            + source_nodes,
            metadata={
                "answerable_probability": response.answerable_probability,
            },
        )

    def _get_prompts(self) -> PromptDictType:
        # Not used.
        return {}

    def _update_prompts(self, prompts_dict: PromptDictType) -> None:
        # Not used.
        ...

from_defaults classmethod

from_defaults(temperature: float = 0.7, answer_style: int = 1, safety_setting: List[SafetySetting] = []) -> GoogleTextSynthesizer

创建一个新的 Google AQA。

示例

responder = GoogleTextSynthesizer.create( temperature=0.7, answer_style=AnswerStyle.ABSTRACTIVE, safety_setting=[ SafetySetting( category=HARM_CATEGORY_SEXUALLY_EXPLICIT, threshold=HarmBlockThreshold.BLOCK_LOW_AND_ABOVE, ), ] )

参数

名称	类型	描述	默认值
temperature	float	0.0 到 1.0。	0.7
answer_style	int	参见 google.ai.generativelanguage.GenerateAnswerRequest.AnswerStyle 默认值为 ABSTRACTIVE (1)。	1
safety_setting	List[SafetySetting]	参见 google.ai.generativelanguage.SafetySetting。	[]

返回

类型	描述
GoogleTextSynthesizer	一个 GoogleTextSynthesizer 实例。

源码位于 llama-index-integrations/response_synthesizers/llama-index-response-synthesizers-google/llama_index/response_synthesizers/google/base.py

@classmethod
def from_defaults(
    cls,
    temperature: float = 0.7,
    answer_style: int = 1,
    safety_setting: List["genai.SafetySetting"] = [],
) -> "GoogleTextSynthesizer":
    """
    Create a new Google AQA.

    Example:
      responder = GoogleTextSynthesizer.create(
          temperature=0.7,
          answer_style=AnswerStyle.ABSTRACTIVE,
          safety_setting=[
              SafetySetting(
                  category=HARM_CATEGORY_SEXUALLY_EXPLICIT,
                  threshold=HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
              ),
          ]
      )

    Args:
      temperature: 0.0 to 1.0.
      answer_style: See `google.ai.generativelanguage.GenerateAnswerRequest.AnswerStyle`
        The default is ABSTRACTIVE (1).
      safety_setting: See `google.ai.generativelanguage.SafetySetting`.

    Returns:
      an instance of GoogleTextSynthesizer.

    """
    return cls(
        temperature=temperature,
        answer_style=answer_style,
        safety_setting=safety_setting,
    )

get_response

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

基于提供的通道生成一个 grounded (有根据的) 响应。

参数

名称	类型	描述	默认值
query_str	str	用户的提问。	required
text_chunks	Sequence[str]	应用于回答问题的通道列表。	required

返回

类型	描述
SynthesizedResponse	一个 SynthesizedResponse 对象。

源码位于 llama-index-integrations/response_synthesizers/llama-index-response-synthesizers-google/llama_index/response_synthesizers/google/base.py

def get_response(
    self,
    query_str: str,
    text_chunks: Sequence[str],
    **response_kwargs: Any,
) -> SynthesizedResponse:
    """
    Generate a grounded response on provided passages.

    Args:
        query_str: The user's question.
        text_chunks: A list of passages that should be used to answer the
            question.

    Returns:
        A `SynthesizedResponse` object.

    """
    try:
        import llama_index.vector_stores.google.genai_extension as genaix

        import google.ai.generativelanguage as genai
    except ImportError:
        raise ImportError(_import_err_msg)

    client = cast(genai.GenerativeServiceClient, self._client)
    response = genaix.generate_answer(
        prompt=query_str,
        passages=list(text_chunks),
        answer_style=self._answer_style,
        safety_settings=self._safety_setting,
        temperature=self._temperature,
        client=client,
    )

    return SynthesizedResponse(
        answer=response.answer,
        attributed_passages=[
            passage.text for passage in response.attributed_passages
        ],
        answerable_probability=response.answerable_probability,
    )

synthesize

synthesize(query: QueryTextType, nodes: List[NodeWithScore], additional_source_nodes: Optional[Sequence[NodeWithScore]] = None, **response_kwargs: Any) -> Response

基于提供的通道返回一个 grounded (有根据的) 响应。

返回

类型	描述
响应	响应的 source_nodes 将以归因通道列表开头。
响应	这些通道是用于构建 grounded 响应的通道。
响应	这些通道将始终没有得分，这是标记它们为归因通道的唯一方法。
响应	然后，列表将跟随原始提供的通道，这些通道将
响应	具有检索得出的得分。
响应	检索得出的得分。
响应	响应的 metadata 也可能有一个带有键 answerable_probability 的条目，
响应	这是模型对其答案在输入通道中是正确且有根据的
响应	概率的估计。
响应	概率的估计。

源码位于 llama-index-integrations/response_synthesizers/llama-index-response-synthesizers-google/llama_index/response_synthesizers/google/base.py

def synthesize(
    self,
    query: QueryTextType,
    nodes: List[NodeWithScore],
    additional_source_nodes: Optional[Sequence[NodeWithScore]] = None,
    **response_kwargs: Any,
) -> Response:
    """
    Returns a grounded response based on provided passages.

    Returns:
        Response's `source_nodes` will begin with a list of attributed
        passages. These passages are the ones that were used to construct
        the grounded response. These passages will always have no score,
        the only way to mark them as attributed passages. Then, the list
        will follow with the originally provided passages, which will have
        a score from the retrieval.

        Response's `metadata` may also have have an entry with key
        `answerable_probability`, which is the model's estimate of the
        probability that its answer is correct and grounded in the input
        passages.

    """
    if len(nodes) == 0:
        return Response("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:
        internal_response = 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 = list(additional_source_nodes or [])

        external_response = self._prepare_external_response(
            internal_response, nodes + additional_source_nodes
        )

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

    return external_response