File size: 4,547 Bytes
78797ac |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
from crewai import Agent
import re
import streamlit as st
from langchain_community.llms import OpenAI
from tools.browser_tools import BrowserTools
from tools.calculator_tools import CalculatorTools
from tools.search_tools import SearchTools
from crewai.llm import LLM
model_kwargs = {
"model": f"openai/{st.secrets['MODEL_ID']}",
"base_url": st.secrets["MODEL_BASE_URL"],
"api_key": st.secrets["OPENAI_API_KEY"]
}
llm = LLM(**model_kwargs) # Loading Llama
class TripAgents():
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=[
SearchTools.search_internet,
BrowserTools.scrape_and_summarize_website,
],
verbose=True,
llm=llm
# step_callback=streamlit_callback,
)
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=[
SearchTools.search_internet,
BrowserTools.scrape_and_summarize_website,
],
verbose=True,
llm=llm
# step_callback=streamlit_callback,
)
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=[
SearchTools.search_internet,
BrowserTools.scrape_and_summarize_website,
CalculatorTools.calculate,
],
verbose=True,
llm=llm
# step_callback=streamlit_callback,
)
class StreamToExpander:
def __init__(self, expander):
self.expander = expander
self.buffer = []
self.colors = ['red', 'green', 'blue', 'orange'] # Define a list of colors
self.color_index = 0 # Initialize color index
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:
# Apply different color and switch color index
self.color_index = (self.color_index + 1) % len(self.colors) # Increment color index and wrap around if necessary
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:
# Apply different color
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 = []
|