Improve UI | Add chains for itinerary with map JSON

agents/agent.py

@@ -1,9 +1,14 @@
 import openai
 import logging
 import time
-from templates.validation import ValidationTemplate  # Adjust the import path as necessary
+from templates.validation import ValidationTemplate 
+from templates.itinerary import ItineraryTemplate
+from templates.mapping import MappingTemplate 
 from langchain.chat_models  import ChatOpenAI
 from langchain.chains import LLMChain, SequentialChain
+from datetime import datetime
+from datetime import date
+from dateutil.relativedelta import relativedelta
 
 logging.basicConfig(level=logging.INFO)
 
@@ -25,10 +30,16 @@ class Agent:
 
         # Initialize ChatOpenAI with the provided OpenAI API key and model details
         self.chat_model = ChatOpenAI(model=model, temperature=temperature, openai_api_key=self._openai_key)
-        # Initialize the ValidationTemplate
+
+        # Initialize the ValidationTemplate and ItineraryTemplate
         self.validation_prompt = ValidationTemplate()
+        self.itinerary_prompt = ItineraryTemplate()
+        self.mapping_prompt = MappingTemplate()
+
+
         # Setup the validation chain using the LLMChain and SequentialChain
         self.validation_chain = self._set_up_validation_chain(debug)
+        self.itinerary_chain = self._set_up_itinerary_chain(debug)
 
     def _set_up_validation_chain(self, debug=True):
         # Make validation agent chain using LLMChain
@@ -49,6 +60,38 @@ class Agent:
         )
 
         return overall_chain
+    
+    def _set_up_itinerary_chain(self, debug=True):
+
+        # set up LLMChain to get the itinerary as a string
+        itinerary_agent  = LLMChain(
+            llm=self.chat_model,
+            prompt=self.itinerary_prompt.chat_prompt,
+            verbose=debug,
+            output_key="itinerary_suggestion",
+        )
+
+        # set up LLMChain to extract the waypoints as a JSON object
+        mapping_agent = LLMChain(
+            llm=self.chat_model,
+            prompt=self.mapping_prompt.chat_prompt,
+            output_parser=self.mapping_prompt.parser,
+            verbose=debug,
+            output_key="mapping_list",
+        )
+
+        # overall chain allows us to call the travel_agent and parser in
+        # sequence, with labelled outputs.
+        overall_chain = SequentialChain(
+            chains=[itinerary_agent, mapping_agent],
+            input_variables=["start_location", "end_location", "start_date", "end_date",
+                             "attractions", "budget", "transportation", "accommodation",
+                             "schedule", "format_instructions"],
+            output_variables=["itinerary_suggestion","mapping_list"],
+            verbose=debug,
+        )
+
+        return overall_chain
 
     def validate_travel(self, query):
         self.logger.info("Validating query: %s", query)
@@ -60,7 +103,6 @@ class Agent:
             )
         )
 
-
         # Call the validation chain with the query and format instructions
         validation_result = self.validation_chain.run(
             {
@@ -69,9 +111,126 @@ class Agent:
             }
         )
 
-        # Extract the result from the validation output
-        validation_output = validation_result["validation_output"].dict()
+        self.logger.info("Datatype of validation_result: %s", type(validation_result))
+
+        # Convert the validation result into a dictionary if it's not one already
+        if isinstance(validation_result, dict):
+            validation_dict = validation_result
+        else:  # assuming validation_result is an instance of the Validation class
+            validation_dict = validation_result.dict()
+
+        # Log the datatype and content of the validation output
+        self.logger.info("Datatype of validation_dict: %s", type(validation_dict))
+        self.logger.info("Content of validation_dict: %s", validation_dict)
+
         t2 = time.time()
         self.logger.debug("Time to validate request: %.2f seconds", t2 - t1)
 
-        return validation_output
+        return validation_dict
+
+
+    def calculate_duration(self, start_date, end_date):
+        if not isinstance(start_date, date) or not isinstance(end_date, date):
+            raise ValueError("start_date and end_date must be datetime.date objects")
+
+        if end_date < start_date:
+            raise ValueError("End date must be after or equal to start date")
+
+        # Calculate the duration using relativedelta
+        delta = relativedelta(end_date, start_date)
+        
+        years = delta.years
+        months = delta.months
+        days = delta.days + 1 # We'll calculate weeks from days
+
+        duration_parts = []
+
+        if years > 0:
+            duration_parts.append(f"{years} year{'s' if years > 1 else ''}")
+
+        if months > 0:
+            duration_parts.append(f"{months} month{'s' if months > 1 else ''}")
+
+        weeks = days // 7
+        days = days % 7
+
+        if weeks > 0:
+            duration_parts.append(f"{weeks} week{'s' if weeks > 1 else ''}")
+
+        if days > 0:
+            duration_parts.append(f"{days} day{'s' if days > 1 else ''}")
+
+        return ', '.join(duration_parts)
+
+
+    def generate_itinerary(self, user_details):
+        self.logger.info("Generating itinerary for user details: %s", user_details)
+        
+        # Validate the user details dictionary keys match the expected input variables
+        expected_keys = ["start_location", "end_location", "start_date", "end_date",
+                        "attractions", "budget", "transportation", "accommodation",
+                        "schedule"]
+        for key in expected_keys:
+            if key not in user_details:
+                self.logger.error("Missing '%s' in user details.", key)
+                return None  # or handle the missing key appropriately
+        
+        try:
+            # Calculate trip duration
+            trip_duration = self.calculate_duration(user_details['start_date'], user_details['end_date'])
+            # Construct the query phrase
+            query_phrase = "{} trip from {} to {}".format(trip_duration, user_details['start_location'], user_details['end_location'])
+        except KeyError as e:
+            self.logger.error("Missing key in user details: %s", e)
+            return None  # or handle the missing key appropriately
+        except ValueError as e:
+            self.logger.error(e)
+            return None  # or handle the error appropriately
+
+        t1 = time.time()
+        
+        self.logger.info("Calling itinerary chain to validate user query")
+        validation_dict = self.validate_travel(query_phrase)
+        is_plan_valid = validation_dict["plan_is_valid"]
+
+        
+        if is_plan_valid.lower() == "no":
+            self.logger.warning("User request was not valid!")
+            print("\n######\n Travel plan is not valid \n######\n")
+            print(validation_result["updated_request"])
+
+            # Create a dictionary with variable names as keys
+            result_dict = {
+                "itinerary_suggestion": None,
+                "list_of_places": None,
+                "validation_dict": validation_dict
+            }
+            
+            return result_dict
+
+        self.logger.info("User query is valid. Calling itinerary chain on user details")
+
+        itinerary_details = user_details.copy()
+        itinerary_details["format_instructions"] = self.mapping_prompt.parser.get_format_instructions()
+
+        # Call the itinerary chain with the itinerary details
+        itinerary_result = self.itinerary_chain(itinerary_details)
+
+        itinerary_suggestion = itinerary_result["itinerary_suggestion"]
+        list_of_places = itinerary_result["mapping_list"].dict()
+
+        # Log the datatype and content of the list_of_places output
+        self.logger.info("Datatype of validation_dict: %s", type(list_of_places))
+        self.logger.info("Content of validation_dict: %s", list_of_places)
+
+        t2 = time.time()
+        self.logger.debug("Time to generate itinerary: %.2f seconds", t2 - t1)
+        
+        # Create a dictionary with variable names as keys
+        result_dict = {
+            "itinerary_suggestion": itinerary_suggestion,
+            "list_of_places": list_of_places,
+            "validation_dict": validation_dict
+        }
+
+        return result_dict

app.py

@@ -87,16 +87,59 @@ with st.sidebar:
 # Main page layout
 st.header('Your Itinerary')
 if submit:
-    if start_location and end_location and attractions and start_date and end_date:
-        # The function to generate the itinerary would go here.
-        # The following lines are placeholders to show the captured inputs.
-        st.write('From:', start_location)
-        st.write('To:', end_location)
-        st.write('Travel Dates:', st.session_state['start_date'], 'to', st.session_state['end_date'])
-        st.write('Attractions:', attractions)
-        st.write('Budget:', budget)
-        st.write('Transportation:', transportation)
-        st.write('Accommodation:', accommodation)
-        st.write('Daily Schedule:', schedule)
+    missing_fields = []
+
+    if not start_location:
+        missing_fields.append("Start Location")
+
+    if not end_location:
+        missing_fields.append("End Location")
+
+    if not attractions:
+        missing_fields.append("Attractions")
+
+    if not start_date:
+        missing_fields.append("Start Date")
+
+    if not end_date:
+        missing_fields.append("End Date")
+
+    if not accommodation:
+        missing_fields.append("Accommodation")
+
+    if not schedule:
+        missing_fields.append("Schedule")
+
+    if not transportation:
+        missing_fields.append("Transportation")
+
+    if not missing_fields:
+        user_details = {
+            'start_location': start_location,
+            'end_location': end_location,
+            'start_date': start_date,
+            'end_date': end_date,
+            'attractions': attractions,
+            'budget': budget,
+            'transportation': transportation,
+            'accommodation': accommodation,
+            'schedule': schedule
+        }
+
+        # Display user details in the Streamlit console
+        st.write("Start Location:", user_details['start_location'])
+        st.write("End Location:", user_details['end_location'])
+        st.write("Start Date:", user_details['start_date'])
+        st.write("End Date:", user_details['end_date'])
+        st.write("Attractions:", user_details['attractions'])
+        st.write("Budget:", user_details['budget'])
+        st.write("Transportation:", user_details['transportation'])
+        st.write("Accommodation:", user_details['accommodation'])
+        st.write("Schedule:", user_details['schedule'])
+
+        note = """Here is your personalized travel itinerary covering all the major locations you want to visit. 
+        This map gives you a general overview of your travel route from start to finish. 
+        For daily plans, please review the sequence of waypoints to ensure the best experience, 
+        as the route may vary based on daily activities and traffic conditions."""
     else:
-        st.error('Please fill out all required fields.')
+        st.error(f'Please fill out the following required fields: {", ".join(missing_fields)}')

templates/itinerary.py

@@ -0,0 +1,66 @@
+from langchain.prompts import (
+    ChatPromptTemplate,
+    SystemMessagePromptTemplate,
+    HumanMessagePromptTemplate,
+)
+
+class ItineraryTemplate:
+    def __init__(self):
+        self.system_template = """
+      You are a sophisticated AI travel guide. Your job is to create engaging and practical travel plans for users.
+
+      The user's travel details and preferences will be presented in a structured format, starting with four hashtags. Your task is to convert these details into a detailed travel itinerary that includes waypoints, activities, and suggestions tailored to the user's interests and constraints.
+
+      Focus on providing specific addresses or locations for each suggested activity.
+
+      Take into account the user's available time, budget, preferred transportation mode, accommodation type, and desired daily schedule intensity. Aim to construct an itinerary that is enjoyable, feasible, and aligns with the user's expectations.
+
+      Present the itinerary in a bulleted list format with clear indications of the start and end points, as well as the recommended mode of transit between locations.
+
+      If the user has not specified certain details, use your judgment to select appropriate options, ensuring to provide specific addresses. Your response should be the itinerary list exclusively.
+    """
+
+        self.human_template = """
+      #### User's Travel Details ####
+      - Starting Location: {start_location}
+      - Destination: {end_location}
+      - Travel Dates: {start_date} to {end_date}
+      - Attractions of Interest: {attractions}
+      - Travel Budget Range: {budget}
+      - Preferred Transportation: {transportation}
+      - Accommodation Type: {accommodation}
+      - Desired Daily Schedule Intensity: {schedule}
+    """
+
+        # Assuming SystemMessagePromptTemplate and HumanMessagePromptTemplate are classes
+        # defined elsewhere in your codebase that format these messages.
+        self.system_message_prompt = SystemMessagePromptTemplate.from_template(
+            self.system_template
+        )
+        self.human_message_prompt = HumanMessagePromptTemplate.from_template(
+            self.human_template,
+            input_variables=["start_location", "end_location", "start_date", "end_date",
+                             "attractions", "budget", "transportation", "accommodation",
+                             "schedule"]
+        )
+
+        self.chat_prompt = ChatPromptTemplate.from_messages(
+            [self.system_message_prompt, self.human_message_prompt]
+        )
+
+    def generate_prompt(self, user_details):
+        """
+        Fill in the human template with the actual details from the user's input.
+        """
+        filled_human_template = self.human_message_prompt.format(
+            start_location=user_details['start_location'],
+            end_location=user_details['end_location'],
+            start_date=user_details['start_date'].strftime('%Y-%m-%d'),
+            end_date=user_details['end_date'].strftime('%Y-%m-%d'),
+            attractions=", ".join(user_details['attractions']),
+            budget=f"{user_details['budget'][0]} to {user_details['budget'][1]}",
+            transportation=user_details['transportation'],
+            accommodation=user_details['accommodation'],
+            schedule=user_details['schedule']
+        )
+        return f"{self.system_message_prompt}\n{filled_human_template}"

templates/mapping.py

@@ -0,0 +1,77 @@
+from langchain.prompts import (
+    ChatPromptTemplate,
+    SystemMessagePromptTemplate,
+    HumanMessagePromptTemplate,
+)
+from langchain.output_parsers import PydanticOutputParser
+from pydantic import BaseModel, Field
+from typing import List
+
+class Path(BaseModel):
+    start_location: str = Field(description="start location of trip")
+    end_location: str = Field(description="end location of trip")
+    waypoints: List[str] = Field(description="list of waypoints")
+    transit: str = Field(description="mode of transportation")
+
+class MappingTemplate(object):
+
+    MAX_WAYPOINTS = 20  # This can be adjusted or made configurable.
+
+    def __init__(self):
+        self.system_template = """
+      You an agent who converts detailed travel plans into a simple list of locations.
+
+      The itinerary will be denoted by four hashtags. Convert it into
+      list of places that they should visit. Try to include the specific address of each location.
+
+      Your output should always contain the start and end point of the trip, and may also include a list
+      of waypoints. It should also include a mode of transit. The number of waypoints cannot exceed 20.
+      If you can't infer the mode of transit, make a best guess given the trip location.
+
+      For example:
+
+      ####
+      Itinerary for a 2-day driving trip within London:
+      - Day 1:
+        - Start at Buckingham Palace (The Mall, London SW1A 1AA)
+        - Visit the Tower of London (Tower Hill, London EC3N 4AB)
+        - Explore the British Museum (Great Russell St, Bloomsbury, London WC1B 3DG)
+        - Enjoy shopping at Oxford Street (Oxford St, London W1C 1JN)
+        - End the day at Covent Garden (Covent Garden, London WC2E 8RF)
+      - Day 2:
+        - Start at Westminster Abbey (20 Deans Yd, Westminster, London SW1P 3PA)
+        - Visit the Churchill War Rooms (Clive Steps, King Charles St, London SW1A 2AQ)
+        - Explore the Natural History Museum (Cromwell Rd, Kensington, London SW7 5BD)
+        - End the trip at the Tower Bridge (Tower Bridge Rd, London SE1 2UP)
+      #####
+
+      Output:
+      Start: Buckingham Palace, The Mall, London SW1A 1AA
+      End: Tower Bridge, Tower Bridge Rd, London SE1 2UP
+      Waypoints: ["Tower of London, Tower Hill, London EC3N 4AB", "British Museum, Great Russell St, Bloomsbury, London WC1B 3DG", "Oxford St, London W1C 1JN", "Covent Garden, London WC2E 8RF","Westminster, London SW1A 0AA", "St. James's Park, London", "Natural History Museum, Cromwell Rd, Kensington, London SW7 5BD"]
+      Transit: driving
+
+      Transit can be only one of the following options: "driving", "train", "bus" or "flight".
+
+      {format_instructions}
+    """
+
+        self.human_template = """
+      ####{itinerary_suggestion}####
+    """
+
+        self.parser = PydanticOutputParser(pydantic_object=Path)
+
+        self.system_message_prompt = SystemMessagePromptTemplate.from_template(
+            self.system_template,
+            partial_variables={
+                "format_instructions": self.parser.get_format_instructions()
+            },
+        )
+        self.human_message_prompt = HumanMessagePromptTemplate.from_template(
+            self.human_template, input_variables=["itinerary_suggestion"]
+        )
+
+        self.chat_prompt = ChatPromptTemplate.from_messages(
+            [self.system_message_prompt, self.human_message_prompt]
+        )

tests/agent/test_agent.py

@@ -1,6 +1,7 @@
 import unittest
-from config import load_secrets # Update this path to match the actual location
-from agents.agent import Agent  # Update this path to match the actual location
+from config import load_secrets 
+from agents.agent import Agent  
+import datetime
 
 DEBUG = True
 
@@ -23,7 +24,7 @@ class TestAgentMethods(unittest.TestCase):
             debug=self.debug,
         )
 
-    # @unittest.skipIf(DEBUG, "Skipping this test while debugging other tests")
+    @unittest.skipIf(DEBUG, "Skipping this test while debugging other tests")
     def test_validation_chain(self):
         validation_chain = self.agent._set_up_validation_chain(debug=self.debug)
 
@@ -60,6 +61,34 @@ class TestAgentMethods(unittest.TestCase):
         q3_out = q3_res["validation_output"].dict()
         self.assertEqual(q3_out["plan_is_valid"], "yes")
 
+    
+    def test_generate_itinerary(self):
+
+        user_details = {
+            "start_location": "Berkeley, CA",
+            "end_location": "Seattle, WA",
+            "start_date": datetime.date(2023, 12, 10),
+            "end_date": datetime.date(2023, 12, 15),
+            "attractions": ["museums", "parks"],
+            "budget": "1500-3000 USD",
+            "transportation": "rental car, public Transport",
+            "accommodation": "hotels",
+            "schedule": "relaxed"
+        }
+
+        # Call the generate_itinerary method with the user_details
+        itinerary_result = self.agent.generate_itinerary(user_details)
+
+        itinerary_suggestion = itinerary_result["itinerary_suggestion"] 
+        list_of_places = itinerary_result["list_of_places"] 
+        validation_dict = itinerary_result["validation_dict"] 
+
+        print("\nItinerary Suggestion Returned:\n", itinerary_suggestion)
+        print("\nList of Places Returned:\n", list_of_places)
+
+        # Assert that the itinerary contains expected keys or values.
+        # This depends on what `generate_itinerary` returns. For example:
+        self.assertIsNotNone(itinerary_suggestion)
 
 if __name__ == "__main__":
     unittest.main()