Dynamic metadata filtering for Amazon Bedrock Data Bases with LangChain

{
    "andAll": [
        {
            "equals": {
                "key": "desired_destination",
                "value": ""  # This will be overwritten with appropriate values at runtime
            }
        },
        {
            "equals": {
                "key": "travelling_with_children",
                "value": ""  # This will be overwritten with appropriate values at runtime
            }
        }
    ]
}

# Check AWS connection
# Create a session utilizing your AWS credentials
session = boto3.Session()

# Create an STS shopper
sts_client = session.shopper('sts')

# Get the caller id
response = sts_client.get_caller_identity()

# Print the response
print(response)

knowledge_base_id = 'XXXXXXXXXX'

retrieval_config = {
    "vectorSearchConfiguration": {
        "numberOfResults": 4,
        "overrideSearchType": "HYBRID"
    }
}

# Check bedrock information bases connection
shopper = boto3.shopper('bedrock-agent-runtime')

response = shopper.retrieve(
    knowledgeBaseId=knowledge_base_id,
    retrievalConfiguration=retrieval_config,
    retrievalQuery={"textual content": "Howdy world"}
)

print(response)

def setup_retrieval_config(inputs):

    # Make a duplicate as a result of the filter is up to date dynamically based mostly on the user_data, this lets you begin from the default every time
    local_retrieval_config = copy.deepcopy(retrieval_config)

    updated_vector_search_config = replace_values(local_retrieval_config["vectorSearchConfiguration"], inputs["user_data"])
    local_retrieval_config["vectorSearchConfiguration"] = updated_vector_search_config

    return local_retrieval_config

def replace_values(vector_search_config: Dict, user_data: Dict):
    # Change the worth fields within the filter with the proper worth in response to the user_data
    # Makes use of breadth first search to seek out the entire worth fields

    # Filter shouldn't be a required key, if you do not need any filters eliminate the important thing
    if "filter" in vector_search_config and never vector_search_config["filter"]:
        del vector_search_config["filter"]

    # Recursively traverse from the foundation
    elif 'filter' in vector_search_config:
        vector_search_config['filter'] = replace_values(vector_search_config['filter'], user_data)

    # At a node that's not the foundation
    else:
        for key, worth in vector_search_config.objects():
            if isinstance(worth, dict):

                # At a leaf e.g. {"key": "age", "worth": ""}}
                if 'key' in worth and 'worth' in worth:

                    # Solely overwrite worth['value'] that aren't unknown
                    if worth['key'] in user_data and never (worth["value"] == "unknown" or worth["value"] == ["unknown"]):

                        # Primitive information sort
                        if sort(worth["value"]) in [str, int, float, bool]:
                            worth['value'] = user_data[value['key']]

                        # Listing information sort
                        elif isinstance(worth["value"], checklist):
                            worth['value'] = [user_data[value['key']]]
                        else:
                            increase ValueError(f"Unsupported worth['value'] sort {sort(worth['value'])}")
                else:
                    vector_search_config[key] = replace_values(worth, user_data)

            # Recurse on every merchandise within the checklist
            elif isinstance(worth, checklist):
                vector_search_config[key] = [replace_values(item, user_data) for item in value]
            else:
                increase ValueError(f"Unsupported worth sort {sort(worth)}")

    return vector_search_config

def create_retrieval_chain() -> Runnable:
        """
        Creates a retrieval chain for the retriever.

        Returns:
            Runnable: The retrieval chain.
        """

        question = create_query_for_retrieval()

        def create_retriever(inputs):
            # This wrapper is important as a result of if you happen to return a callable object LangChain will robotically name it instantly, which isn't the specified habits
            # as an alternative we wish to name the retriever on the subsequent step of the chain
            retriever_wrapper = {"retriever": AmazonKnowledgeBasesRetriever(knowledge_base_id=knowledge_base_id, retrieval_config=inputs["retrieval_config"])}
            return retriever_wrapper

        # Retrieval chain has three steps: (1) create the filter based mostly off of the person information, (2) create the retriever, and (3) invoke the retriever
        retrieval_chain = (
            {
                "user_data" : itemgetter("user_data"),
                "retrieval_config" : lambda inputs: setup_retrieval_config(inputs)
            } |
            {
                "question" : question,
                "retriever" : lambda inputs: create_retriever(inputs)
            } |
            RunnableLambda(lambda inputs: inputs["retriever"]["retriever"].invoke(inputs["query"]))
        )
        return retrieval_chain

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=knowledge_base_id,
    retrieval_config=retrieval_config
)

def create_retrieval_chain() -> Runnable:
        """
        Creates a retrieval chain for the retriever.

        Returns:
            Runnable: The retrieval chain.
        """

        question = create_query_for_retrieval()
        
        def retrieve_and_format(inputs):
            outcomes = retriever.shopper.retrieve(
                retrievalQuery={"textual content": inputs["query"]}, 
                knowledgeBaseId=knowledge_base_id, 
                retrievalConfiguration=inputs["retrieval_config"]
            )
        
            paperwork = []
            for end in outcomes["retrievalResults"]:
                metadata = {
                    "location": outcome["location"],
                    "source_metadata": outcome["metadata"],
                    "rating": outcome["score"],
                }

                doc = Doc(
                    page_content=outcome["content"]["text"],
                    metadata=metadata
                )
                paperwork.append(doc)
            
            return paperwork

        retrieval_chain = (
            {
                "question" : question,
                "retrieval_config" : lambda inputs: setup_retrieval_config(inputs)
            } |
            RunnableLambda(lambda inputs: retrieve_and_format(inputs))
            # RunnableLambda(lambda inputs: retriever.shopper.retrieve(retrievalQuery={"textual content": inputs["query"]}, knowledgeBaseId=knowledge_base_id, retrievalConfiguration=inputs["retrieval_config"]))
        )
        return retrieval_chain

retrieval_chain_2 = create_retrieval_chain()

with open("information/user_data.json", "r") as file:
user_data = json.load(file)

print(json.dumps(user_data[:2], indent=2))

{
        "trip_id": 1,
        "desired_destination": "Bali, Indonesia",
        "stay_duration": 7,
        "age": 35,
        "gender": "male",
        "companion": "solo",
	"travelling_with_children": "no",
        "travelling_with_pets": "no"
    },
    {
        "trip_id": 2,
        "desired_destination": "Paris, France",
        "stay_duration": 5,
        "age": 28,
        "gender": "feminine",
        "companion": "solo",
	"travelling_with_children": "no",
        "travelling_with_pets": "sure"
    },

filters_to_test: Listing = [
    {
        "andAll": [
            {
                "equals": {
                    "key": "desired_destination",
                    "value": ""  # This will be overwritten with appropriate values at runtime
                }
            },
            {
                "equals": {
                    "key": "travelling_with_children",
                    "value": ""  # This will be overwritten with appropriate values at runtime
                }
            }
        ]
    },
    None
]

retrieval_chains = [retrieval_chain_1, retrieval_chain_2]

outcomes = []

for retrieval_chain_id, retrieval_chain in enumerate(retrieval_chains):
    logger.data(retrieval_chain_id)
    # Loop by way of every filter choices
    for filter in filters_to_test:
        retrieval_config["vectorSearchConfiguration"]["filter"] = filter
        # Loop by way of every person information entry
        for user_entry in user_data:
            inputs = {
                    "user_data": user_entry,
                    "retrieval_config": retrieval_config
                }

            # Run the retrieval chain with the present person entry
            strive:
                outcome = retrieval_chain.invoke(inputs)
                # print(f"End result for person entry {user_entry['trip_id']}: {outcome}")
                outcomes.append(({'retrieval_chain_id': retrieval_chain_id, 'person': user_entry, 'paperwork': outcome}))

            besides Exception as e:
                print(f"Error throughout retrieval for person entry {user_entry['trip_id']}: {e}")