Source code for langchain.retrievers.aws_kendra_index_retriever
"""Retriever wrapper for AWS Kendra."""
import re
from typing import Any, Dict, List
from langchain.schema import BaseRetriever, Document
[docs]class AwsKendraIndexRetriever(BaseRetriever):
    """Wrapper around AWS Kendra."""
    kendraindex: str
    """Kendra index id"""
    k: int
    """Number of documents to query for."""
    languagecode: str
    """Languagecode used for querying."""
    kclient: Any
    """ boto3 client for Kendra. """
    def __init__(
        self, kclient: Any, kendraindex: str, k: int = 3, languagecode: str = "en"
    ):
        self.kendraindex = kendraindex
        self.k = k
        self.languagecode = languagecode
        self.kclient = kclient
    def _clean_result(self, res_text: str) -> str:
        return re.sub("\s+", " ", res_text).replace("...", "")
    def _get_top_n_results(self, resp: Dict, count: int) -> Document:
        r = resp["ResultItems"][count]
        doc_title = r["DocumentTitle"]["Text"]
        doc_uri = r["DocumentURI"]
        r_type = r["Type"]
        if (
            r["AdditionalAttributes"]
            and r["AdditionalAttributes"][0]["Key"] == "AnswerText"
        ):
            res_text = r["AdditionalAttributes"][0]["Value"]["TextWithHighlightsValue"][
                "Text"
            ]
        else:
            res_text = r["DocumentExcerpt"]["Text"]
        doc_excerpt = self._clean_result(res_text)
rtdocs_stable/api.python.langchain.com/en/stable/_modules/langchain/retrievers/aws_kendra_index_retriever.html