src/backend/base/langflow/components/retrievers/amazon_kendra.py

from typing import cast

from langchain_community.retrievers import AmazonKendraRetriever

from langflow.custom import CustomComponent
from langflow.field_typing import Retriever


class AmazonKendraRetrieverComponent(CustomComponent):
    display_name: str = "Amazon Kendra Retriever"
    description: str = "Retriever that uses the Amazon Kendra API."
    name = "AmazonKendra"
    icon = "Amazon"
    legacy: bool = True

    def build_config(self):
        return {
            "index_id": {"display_name": "Index ID"},
            "region_name": {"display_name": "Region Name"},
            "credentials_profile_name": {"display_name": "Credentials Profile Name"},
            "attribute_filter": {
                "display_name": "Attribute Filter",
                "field_type": "code",
            },
            "top_k": {"display_name": "Top K", "field_type": "int"},
            "user_context": {
                "display_name": "User Context",
                "field_type": "code",
            },
            "code": {"show": False},
        }

    def build(
        self,
        index_id: str,
        top_k: int = 3,
        region_name: str | None = None,
        credentials_profile_name: str | None = None,
        attribute_filter: dict | None = None,
        user_context: dict | None = None,
    ) -> Retriever:  # type: ignore[type-var]
        try:
            output = AmazonKendraRetriever(
                index_id=index_id,
                top_k=top_k,
                region_name=region_name,
                credentials_profile_name=credentials_profile_name,
                attribute_filter=attribute_filter,
                user_context=user_context,
            )
        except Exception as e:
            msg = "Could not connect to AmazonKendra API."
            raise ValueError(msg) from e
        return cast("Retriever", output)