Trip Planner Agent using CrewAI

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Sourangshu Pal
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,3 +1,19 @@
+![CrewAI](https://miro.medium.com/v2/resize:fit:1400/0*-7HC-GJCxjn-Dm7i.png)
+
+# 🏖️ Trip Planner: Streamlit with CrewAI
+
+![Streamlit App](images/trip_planner.jpg)
+
+## Introduction
+
+Trip Planner leverages the CrewAI framework to automate and enhance the trip planning experience, a user-friendly Streamlit interface.
+
+
+## CrewAI Framework
+
+CrewAI simplifies the orchestration of role-playing AI agents. In VacAIgent, these agents collaboratively decide on cities and craft a complete itinerary for your trip based on specified preferences, all accessible via a streamlined Streamlit user interface.
+
+
 ---
 title: Trip Planner
 emoji: 🏃

agents/trip_agents.py

@@ -0,0 +1,125 @@
+
+from crewai import Agent
+import re
+import streamlit as st
+from langchain_core.language_models.chat_models import BaseChatModel
+from crewai import LLM
+from tools.browser_tools import BrowserTools
+from tools.calculator_tools import CalculatorTools
+from tools.search_tools import SearchTools
+from typing import Any
+
+class TripAgents():
+    def __init__(self, llm: BaseChatModel = None, 
+                 gemini_api_key: str = None, 
+                 serper_api_key :str = None, 
+                 browserless_api_key: str = None):
+        
+        if llm is None:
+            self.llm = LLM(model="gemini/gemini-2.0-flash", api_key=gemini_api_key)
+        else:
+            self.llm = llm
+
+
+        # Initialize tools once
+        self.search_tool = SearchTools(serper_api_key = serper_api_key)
+        self.browser_tool = BrowserTools(browserless_api_key = browserless_api_key)
+        self.calculator_tool = CalculatorTools()
+
+    def city_selection_agent(self):
+        return Agent(
+            role='City Selection Expert',
+            goal='Select the best city based on weather, season, and prices',
+            backstory='An expert in analyzing travel data to pick ideal destinations',
+            tools=[self.search_tool, self.browser_tool],
+            allow_delegation=False,
+            llm=self.llm,
+            verbose=True
+        )
+
+    def local_expert(self):
+        return Agent(
+            role='Local Expert at this city',
+            goal='Provide the BEST insights about the selected city',
+            backstory="""A knowledgeable local guide with extensive information
+        about the city, it's attractions and customs""",
+            tools=[self.search_tool, self.browser_tool],
+            allow_delegation=False,
+            llm=self.llm,
+            verbose=True
+        )
+
+    def travel_concierge(self):
+        return Agent(
+            role='Amazing Travel Concierge',
+            goal="""Create the most amazing travel itineraries with budget and 
+        packing suggestions for the city""",
+            backstory="""Specialist in travel planning and logistics with 
+        decades of experience""",
+            tools=[self.search_tool, self.browser_tool, self.calculator_tool],
+            allow_delegation=False,
+            llm=self.llm,
+            verbose=True
+        )
+
+###########################################################################################
+# Print agent process to Streamlit app container                                          #
+# This portion of the code is adapted from @AbubakrChan; thank you!                       #
+# https://github.com/AbubakrChan/crewai-UI-business-product-launch/blob/main/main.py#L210 #
+###########################################################################################
+class StreamToExpander:
+    def __init__(self, expander):
+        self.expander = expander
+        self.buffer = []
+        self.colors = ['red', 'green', 'blue', 'orange']
+        self.color_index = 0
+
+    def write(self, data):
+        # Filter out ANSI escape codes using a regular expression
+        cleaned_data = re.sub(r'\x1B\[[0-9;]*[mK]', '', data)
+
+        # Check if the data contains 'task' information
+        task_match_object = re.search(r'\"task\"\s*:\s*\"(.*?)\"', cleaned_data, re.IGNORECASE)
+        task_match_input = re.search(r'task\s*:\s*([^\n]*)', cleaned_data, re.IGNORECASE)
+        task_value = None
+        if task_match_object:
+            task_value = task_match_object.group(1)
+        elif task_match_input:
+            task_value = task_match_input.group(1).strip()
+
+        if task_value:
+            st.toast(":robot_face: " + task_value)
+
+        # Check if the text contains the specified phrase and apply color
+        if "Entering new CrewAgentExecutor chain" in cleaned_data:
+            self.color_index = (self.color_index + 1) % len(self.colors)
+            cleaned_data = cleaned_data.replace("Entering new CrewAgentExecutor chain", 
+                                              f":{self.colors[self.color_index]}[Entering new CrewAgentExecutor chain]")
+
+        if "City Selection Expert" in cleaned_data:
+            cleaned_data = cleaned_data.replace("City Selection Expert", 
+                                              f":{self.colors[self.color_index]}[City Selection Expert]")
+        if "Local Expert at this city" in cleaned_data:
+            cleaned_data = cleaned_data.replace("Local Expert at this city", 
+                                              f":{self.colors[self.color_index]}[Local Expert at this city]")
+        if "Amazing Travel Concierge" in cleaned_data:
+            cleaned_data = cleaned_data.replace("Amazing Travel Concierge", 
+                                              f":{self.colors[self.color_index]}[Amazing Travel Concierge]")
+        if "Finished chain." in cleaned_data:
+            cleaned_data = cleaned_data.replace("Finished chain.", 
+                                              f":{self.colors[self.color_index]}[Finished chain.]")
+
+        self.buffer.append(cleaned_data)
+        if "\n" in data:
+            self.expander.markdown(''.join(self.buffer), unsafe_allow_html=True)
+            self.buffer = []
+
+    def flush(self):
+        """Flush the buffer to the expander"""
+        if self.buffer:
+            self.expander.markdown(''.join(self.buffer), unsafe_allow_html=True)
+            self.buffer = []
+
+    def close(self):
+        """Close the stream"""
+        self.flush()

app.py

@@ -0,0 +1,152 @@
+from crewai import Crew, LLM
+from agents.trip_agents import TripAgents, StreamToExpander
+from tasks.trip_tasks import TripTasks
+import streamlit as st
+import datetime
+import sys
+
+
+st.set_page_config(page_icon="✈️", layout="wide")
+global gemini_api_key
+global serper_api_key
+global browserless_api_key
+
+
+def icon(emoji: str):
+    """Shows an emoji as a Notion-style page icon."""
+    st.write(
+        f'<span style="font-size: 78px; line-height: 1">{emoji}</span>',
+        unsafe_allow_html=True,
+    )
+
+
+class TripCrew:
+
+    def __init__(self, origin, cities, date_range, interests):
+        self.cities = cities
+        self.origin = origin
+        self.interests = interests
+        # Convert date_range to string format for better handling
+        self.date_range = f"{date_range[0].strftime('%Y-%m-%d')} to {date_range[1].strftime('%Y-%m-%d')}"
+        self.output_placeholder = st.empty()
+        self.llm = LLM(model="gemini/gemini-2.0-flash")
+
+    def run(self):
+        try:
+            agents = TripAgents(llm=self.llm,
+                                gemini_api_key=gemini_api_key,
+                                serper_api_key=serper_api_key,
+                                browserless_api_key=browserless_api_key)
+            tasks = TripTasks()
+
+            city_selector_agent = agents.city_selection_agent()
+            local_expert_agent = agents.local_expert()
+            travel_concierge_agent = agents.travel_concierge()
+
+            identify_task = tasks.identify_task(
+                city_selector_agent,
+                self.origin,
+                self.cities,
+                self.interests,
+                self.date_range
+            )
+
+            gather_task = tasks.gather_task(
+                local_expert_agent,
+                self.origin,
+                self.interests,
+                self.date_range
+            )
+
+            plan_task = tasks.plan_task(
+                travel_concierge_agent,
+                self.origin,
+                self.interests,
+                self.date_range
+            )
+
+            crew = Crew(
+                agents=[
+                    city_selector_agent, local_expert_agent, travel_concierge_agent
+                ],
+                tasks=[identify_task, gather_task, plan_task],
+                verbose=True
+            )
+
+            result = crew.kickoff()
+            self.output_placeholder.markdown(result)
+            return result
+        except Exception as e:
+            st.error(f"An error occurred: {str(e)}")
+            return None
+
+
+if __name__ == "__main__":
+    icon("🏖️ Trekki")
+
+    st.subheader("Let AI agents plan your next vacation!",
+                 divider="rainbow", anchor=False)
+
+    import datetime
+
+    today = datetime.datetime.now().date()
+    next_year = today.year + 1
+    jan_16_next_year = datetime.date(next_year, 1, 10)
+
+    with st.sidebar:
+        st.header("👇 Enter API Keys")
+        gemini_api_key  = st.text_input("GEMINI_API_KEY", type="password")
+        serper_api_key  = st.text_input("SERPER_API_KEY", type="password")
+        browserless_api_key = st.text_input("BROWSERLESS_API_KEY", type="password")
+        st.divider()
+        
+        st.header("👇 Enter your trip details")
+        with st.form("my_form"):
+            location = st.text_input(
+                "Where are you currently located?", placeholder="San Mateo, CA")
+            cities = st.text_input(
+                "City and country are you interested in vacationing at?", placeholder="Bali, Indonesia")
+            date_range = st.date_input(
+                "Date range you are interested in traveling?",
+                min_value=today,
+                value=(today, jan_16_next_year + datetime.timedelta(days=6)),
+                format="MM/DD/YYYY",
+            )
+            interests = st.text_area("High level interests and hobbies or extra details about your trip?",
+                                     placeholder="2 adults who love swimming, dancing, hiking, and eating")
+
+            submitted = st.form_submit_button("Submit")
+
+        st.divider()
+
+        # Credits to joaomdmoura/CrewAI for the code: https://github.com/joaomdmoura/crewAI
+        st.sidebar.markdown(
+        """
+        Credits to [**@joaomdmoura**](https://twitter.com/joaomdmoura)
+        for creating **crewAI** 🚀
+        """,
+            unsafe_allow_html=True
+        )
+
+        st.sidebar.info("Click the logo to visit GitHub repo", icon="👇")
+        st.sidebar.markdown(
+            """
+        <a href="https://github.com/joaomdmoura/crewAI" target="_blank">
+            <img src="https://raw.githubusercontent.com/joaomdmoura/crewAI/main/docs/crewai_logo.png" alt="CrewAI Logo" style="width:100px;"/>
+        </a>
+        """,
+            unsafe_allow_html=True
+        )
+
+
+if submitted:
+    with st.status("🤖 **Agents at work...**", state="running", expanded=True) as status:
+        with st.container(height=500, border=False):
+            sys.stdout = StreamToExpander(st)
+            trip_crew = TripCrew(location, cities, date_range, interests)
+            result = trip_crew.run()
+        status.update(label="✅ Trip Plan Ready!",
+                      state="complete", expanded=False)
+
+    st.subheader("Here is your Trip Plan", anchor=False, divider="rainbow")
+    st.markdown(result)

pyproject.toml

@@ -0,0 +1,18 @@
+[project]
+name = "trip-planner-using-crewai"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "crewai>=0.114.0",
+    "langchain>=0.3.24",
+    "langchain-community>=0.3.22",
+    "langchain-core>=0.3.56",
+    "langchain-groq>=0.3.2",
+    "requests>=2.32.3",
+    "streamlit>=1.44.1",
+    "tools>=0.1.9",
+    "unstructured>=0.17.2",
+    "uvicorn>=0.34.2",
+]

tasks/trip_tasks.py

@@ -0,0 +1,91 @@
+from crewai import Task
+from textwrap import dedent
+from datetime import date
+import streamlit as st
+
+class TripTasks():
+    def __validate_inputs(self, origin, cities, interests, date_range):
+        if not origin or not cities or not interests or not date_range:
+            raise ValueError("All input parameters must be provided")
+        return True
+
+    def identify_task(self, agent, origin, cities, interests, range):
+        self.__validate_inputs(origin, cities, interests, range)
+        return Task(description=dedent(f"""
+            Analyze and select the best city for the trip based
+            on specific criteria such as weather patterns, seasonal
+            events, and travel costs. This task involves comparing
+            multiple cities, considering factors like current weather
+            conditions, upcoming cultural or seasonal events, and
+            overall travel expenses.
+
+            Your final answer must be a detailed
+            report on the chosen city, and everything you found out
+            about it, including the actual flight costs, weather
+            forecast and attractions.
+            {self.__tip_section()}
+
+            Traveling from: {origin}
+            City Options: {cities}
+            Trip Date: {range}
+            Traveler Interests: {interests}
+          """),
+            expected_output="A detailed report on the chosen city with flight costs, weather forecast, and attractions.",
+            agent=agent)
+
+    def gather_task(self, agent, origin, interests, range):
+        return Task(description=dedent(f"""
+            As a local expert on this city you must compile an
+            in-depth guide for someone traveling there and wanting
+            to have THE BEST trip ever!
+            Gather information about  key attractions, local customs,
+            special events, and daily activity recommendations.
+            Find the best spots to go to, the kind of place only a
+            local would know.
+            This guide should provide a thorough overview of what
+            the city has to offer, including hidden gems, cultural
+            hotspots, must-visit landmarks, weather forecasts, and
+            high level costs.
+
+            The final answer must be a comprehensive city guide,
+            rich in cultural insights and practical tips,
+            tailored to enhance the travel experience.
+            {self.__tip_section()}
+
+            Trip Date: {range}
+            Traveling from: {origin}
+            Traveler Interests: {interests}
+          """),
+            expected_output="A comprehensive city guide with cultural insights and practical tips.",
+            agent=agent)
+
+    def plan_task(self, agent, origin, interests, range):
+        return Task(description=dedent(f"""
+            Expand this guide into a full travel
+            itinerary for this time {range} with detailed per-day plans, including
+            weather forecasts, places to eat, packing suggestions,
+            and a budget breakdown.
+
+            You MUST suggest actual places to visit, actual hotels
+            to stay and actual restaurants to go to.
+
+            This itinerary should cover all aspects of the trip,
+            from arrival to departure, integrating the city guide
+            information with practical travel logistics.
+
+            Your final answer MUST be a complete expanded travel plan,
+            formatted as markdown, encompassing a daily schedule,
+            anticipated weather conditions, recommended clothing and
+            items to pack, and a detailed budget, ensuring THE BEST
+            TRIP EVER, Be specific and give it a reason why you picked
+            # up each place, what make them special! {self.__tip_section()}
+
+            Trip Date: {range}
+            Traveling from: {origin}
+            Traveler Interests: {interests}
+          """),
+            expected_output="A complete 7-day travel plan, formatted as markdown, with a daily schedule and budget.",
+            agent=agent)
+
+    def __tip_section(self):
+        return "If you do your BEST WORK, I'll tip you $100 and grant you any wish you want!"

tools/browser_tools.py

@@ -0,0 +1,59 @@
+import json
+import requests
+import streamlit as st
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+from unstructured.partition.html import partition_html
+from crewai import Agent, Task
+#from langchain_openai import ChatOpenAI
+from langchain_groq import ChatGroq
+from crewai import LLM
+
+class WebsiteInput(BaseModel):
+    website: str = Field(..., description="The website URL to scrape")
+
+class BrowserTools(BaseTool):
+        
+    name: str = "Scrape website content"
+    description: str = "Useful to scrape and summarize a website content"
+    args_schema: type[BaseModel] = WebsiteInput
+    browserless_api_key : str = Field(..., description="Browserless API Key")
+
+    def _run(self, website: str) -> str:
+        try:
+            url = f"https://chrome.browserless.io/content?token={self.browserless_api_key}"
+            payload = json.dumps({"url": website})
+            headers = {'cache-control': 'no-cache', 'content-type': 'application/json'}
+            response = requests.request("POST", url, headers=headers, data=payload)
+            
+            if response.status_code != 200:
+                return f"Error: Failed to fetch website content. Status code: {response.status_code}"
+            
+            elements = partition_html(text=response.text)
+            content = "\n\n".join([str(el) for el in elements])
+            content = [content[i:i + 8000] for i in range(0, len(content), 8000)]
+            summaries = []
+            
+            #llm = LLM(model="groq/deepseek-r1-distill-llama-70b")
+            llm = LLM(model="gemini/gemini-2.0-flash")
+            
+            for chunk in content:
+                agent = Agent(
+                    role='Principal Researcher',
+                    goal='Do amazing researches and summaries based on the content you are working with',
+                    backstory="You're a Principal Researcher at a big company and you need to do a research about a given topic.",
+                    allow_delegation=False,
+                    llm=llm
+                )
+                task = Task(
+                    description=f'Analyze and summarize the content below, make sure to include the most relevant information in the summary, return only the summary nothing else.\n\nCONTENT\n----------\n{chunk}',
+                    agent=agent
+                )
+                summary = task.execute()
+                summaries.append(summary)
+            return "\n\n".join(summaries)
+        except Exception as e:
+            return f"Error while processing website: {str(e)}"
+
+    async def _arun(self, website: str) -> str:
+        raise NotImplementedError("Async not implemented")

tools/calculator_tools.py

@@ -0,0 +1,18 @@
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+class CalculationInput(BaseModel):
+    operation: str = Field(..., description="The mathematical expression to evaluate")
+
+class CalculatorTools(BaseTool):
+    name: str = "Make a calculation"
+    description: str = """Useful to perform any mathematical calculations, 
+    like sum, minus, multiplication, division, etc.
+    The input should be a mathematical expression, e.g. '200*7' or '5000/2*10'"""
+    args_schema: type[BaseModel] = CalculationInput
+
+    def _run(self, operation: str) -> float:
+        return eval(operation)
+
+    async def _arun(self, operation: str) -> float:
+        raise NotImplementedError("Async not implemented")

tools/search_tools.py

@@ -0,0 +1,52 @@
+import json
+import requests
+import streamlit as st
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+class SearchQuery(BaseModel):
+    query: str = Field(..., description="The search query to look up")
+
+class SearchTools(BaseTool):
+    
+    name: str = "Search the internet"
+    description: str = "Useful to search the internet about a given topic and return relevant results"
+    args_schema: type[BaseModel] = SearchQuery
+    serper_api_key: str = Field(..., description="Serper API Key")
+
+    def _run(self, query: str) -> str:
+        try:
+            top_result_to_return = 4
+            url = "https://google.serper.dev/search"
+            payload = json.dumps({"q": query})
+            headers = {
+                'X-API-KEY': self.serper_api_key,
+                'content-type': 'application/json'
+            }
+            response = requests.request("POST", url, headers=headers, data=payload)
+            
+            if response.status_code != 200:
+                return f"Error: Search API request failed. Status code: {response.status_code}"
+            
+            data = response.json()
+            if 'organic' not in data:
+                return "No results found or API error occurred."
+            
+            results = data['organic']
+            string = []
+            for result in results[:top_result_to_return]:
+                try:
+                    string.append('\n'.join([
+                        f"Title: {result['title']}", 
+                        f"Link: {result['link']}",
+                        f"Snippet: {result['snippet']}", 
+                        "\n-----------------"
+                    ]))
+                except KeyError:
+                    continue
+            return '\n'.join(string) if string else "No valid results found"
+        except Exception as e:
+            return f"Error during search: {str(e)}"
+
+    async def _arun(self, query: str) -> str:
+        raise NotImplementedError("Async not implemented")