doc_excerpt = self._clean_result(res_text)
        combined_text = f"""Document Title: {doc_title}
Document Excerpt: {doc_excerpt}
"""
        return Document(
            page_content=combined_text,
            metadata={
                "source": doc_uri,
                "title": doc_title,
                "excerpt": doc_excerpt,
                "type": r_type,
            },
        )
    def _kendra_query(self, kquery: str) -> List[Document]:
        response = self.kclient.query(
            IndexId=self.kendraindex,
            QueryText=kquery.strip(),
            AttributeFilter={
                "AndAllFilters": [
                    {
                        "EqualsTo": {
                            "Key": "_language_code",
                            "Value": {
                                "StringValue": self.languagecode,
                            },
                        }
                    }
                ]
            },
        )
        if len(response["ResultItems"]) > self.k:
            r_count = self.k
        else:
            r_count = len(response["ResultItems"])
        return [self._get_top_n_results(response, i) for i in range(0, r_count)]
[docs]    def get_relevant_documents(self, query: str) -> List[Document]:
        """Run search on Kendra index and get top k documents
        docs = get_relevant_documents('This is my query')
        """
        return self._kendra_query(query)
[docs]    async def aget_relevant_documents(self, query: str) -> List[Document]:
        raise NotImplementedError("AwsKendraIndexRetriever does not support async")
By Harrison Chase
    
      © Copyright 2023, Harrison Chase.
      
  Last updated on Jun 16, 2023.
rtdocs_stable/api.python.langchain.com/en/stable/_modules/langchain/retrievers/aws_kendra_index_retriever.html