Creating a Memory-enhanced AI Agent With Anthropic's Claude in Amazon Bedrock With MongoDB Atlas

1
       def mongodb_place_lookup_by_country(query_str: str) -> str:
2
       """Retrieve place by Country Name"""
3
       res = ""
4
       res = collection.aggregate(
5
           [
6
               {"$match": {"Country": {"$regex": query_str, "$options": "i"}}},
7
               {"$project": {"Place Name": 1}},
8
           ]
9
       )
10
       places = []
11
       for place in res:
12
           places.append(place["Place Name"])
13
       return str(places)

1
    # filter the data using the criteria and do a Schematic search
2
    def mongodb_search(query: str) -> str:
3
       """Retrieve results from MongoDB related to the user input by performing vector search. Pass text input only."""
4
       embeddings = BedrockEmbeddings(
5
           client=bedrock_runtime,
6
           model_id=embedding_model_id,
7
       )
8
       text_as_embeddings = embeddings.embed_documents([query])
9
       embedding_value = text_as_embeddings[0]
10
    
11
    
12
       # get the vector search results based on the filter conditions.
13
       response = collection.aggregate(
14
           [
15
               {
16
                   "$vectorSearch": {
17
                       "index": "awsanthropic_vector_index",
18
                       "path": "details_embedding",
19
                       "queryVector": embedding_value,
20
                       "numCandidates": 200,
21
                       "limit": 10,
22
                   }
23
               },
24
               {
25
                   "$project": {
26
                       "score": {"$meta": "searchScore"},
27
                       field_name_to_be_vectorized: 1,
28
                       "_id": 0,
29
                   }
30
               },
31
           ]
32
       )
33
    
34
    
35
       # Result is a list of docs with the array fields
36
       docs = list(response)
37
    
38
    
39
       # Extract an array field from the docs
40
       array_field = [doc[field_name_to_be_vectorized] for doc in docs]
41
    
42
    
43
       # Join array elements into a string
44
       llm_input_text = "\n \n".join(str(elem) for elem in array_field)
45
    
46
    
47
       # utility
48
       newline, bold, unbold = "\n", "\033[1m", "\033[0m"
49
       logging.info(
50
           newline
51
           + bold
52
           + "Given Input From MongoDB Vector Search: "
53
           + unbold
54
           + newline
55
           + llm_input_text
56
           + newline
57
       )
58
    
59
    
60
       return llm_input_text

1
    def get_session_history(session_id: str) -> MongoDBChatMessageHistory:
2
       return MongoDBChatMessageHistory(
3
           utils.get_MongoDB_Uri(),
4
           session_id,
5
           database_name="anthropic-travel-agency",
6
           collection_name="chat_history",
7
       )
8
    
9
    
10
    
11
    
12
    def interact_with_agent(sessionId, input_query, chat_history):
13
       """Interact with the agent and store chat history. Return the response."""
14
    
15
    
16
       # Initialize bedrock and llm
17
       bedrock_runtime = setup_bedrock()
18
       llm = initialize_llm(bedrock_runtime)
19
    
20
    
21
       # Initialize tools
22
       tool_mongodb_search = StructuredTool.from_function(mongodb_search)
23
       tool_mongodb_place_lookup_by_country = StructuredTool.from_function(
24
           mongodb_place_lookup_by_country
25
       )
26
       tool_mongodb_place_lookup_by_name = StructuredTool.from_function(
27
           mongodb_place_lookup_by_name
28
       )
29
       tool_mongodb_place_lookup_by_best_time_to_visit = StructuredTool.from_function(
30
           mongodb_place_lookup_by_best_time_to_visit
31
       )
32
    
33
    
34
       tools = [
35
           tool_mongodb_search,
36
           tool_mongodb_place_lookup_by_country,
37
           tool_mongodb_place_lookup_by_name,
38
           tool_mongodb_place_lookup_by_best_time_to_visit,
39
       ]
40
    
41
    
42
       chat_message_int = MessagesPlaceholder(variable_name="chat_history")
43
    
44
    
45
       memory = ConversationBufferMemory(
46
           memory_key="chat_history",
47
           chat_memory=get_session_history(sessionId),
48
           return_messages=True,
49
       )
50
    
51
    
52
       PREFIX = """You are a helpful and polite Travel recommendations assistant. Answer the following questions as best you can using only the provided tools and chat history. You have access to the following tools:"""
53
       FORMAT_INSTRUCTIONS = """Always return only the final answer to the original input question in human readable format as text only without any extra special characters. Also tell all the tools you used to reach to this answer in brief."""
54
       SUFFIX = """Begin!"""
55
    
56
    
57
       agent_executor = initialize_agent(
58
           tools,
59
           llm,
60
           agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
61
           agent_kwargs={
62
               "memory_prompts": [chat_message_int],
63
               "input_variables": ["input", "agent_scratchpad", "chat_history"],
64
               "prefix":PREFIX,
65
               "format_instructions":FORMAT_INSTRUCTIONS,
66
               "suffix":SUFFIX
67
           },
68
           memory=memory,
69
           verbose=True,
70
           max_iterations=3
71
       )
72
       result = agent_executor.invoke(
73
           {
74
               "input": input_query,
75
               "chat_history": chat_history,
76
           }
77
       )
78
       chat_history.append(HumanMessage(content=input_query))
79
       chat_history.append(AIMessage(content="Assistant: " + result["output"]))
80
       return result