-
Notifications
You must be signed in to change notification settings - Fork 47
/
trip_agents.py
143 lines (127 loc) · 6.45 KB
/
trip_agents.py
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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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
## My initial parsing code using callback handler to print to app
# def streamlit_callback(step_output):
# # This function will be called after each step of the agent's execution
# st.markdown("---")
# for step in step_output:
# if isinstance(step, tuple) and len(step) == 2:
# action, observation = step
# if isinstance(action, dict) and "tool" in action and "tool_input" in action and "log" in action:
# st.markdown(f"# Action")
# st.markdown(f"**Tool:** {action['tool']}")
# st.markdown(f"**Tool Input** {action['tool_input']}")
# st.markdown(f"**Log:** {action['log']}")
# st.markdown(f"**Action:** {action['Action']}")
# st.markdown(
# f"**Action Input:** ```json\n{action['tool_input']}\n```")
# elif isinstance(action, str):
# st.markdown(f"**Action:** {action}")
# else:
# st.markdown(f"**Action:** {str(action)}")
# st.markdown(f"**Observation**")
# if isinstance(observation, str):
# observation_lines = observation.split('\n')
# for line in observation_lines:
# if line.startswith('Title: '):
# st.markdown(f"**Title:** {line[7:]}")
# elif line.startswith('Link: '):
# st.markdown(f"**Link:** {line[6:]}")
# elif line.startswith('Snippet: '):
# st.markdown(f"**Snippet:** {line[9:]}")
# elif line.startswith('-'):
# st.markdown(line)
# else:
# st.markdown(line)
# else:
# st.markdown(str(observation))
# else:
# st.markdown(step)
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,
# 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,
# 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,
# step_callback=streamlit_callback,
)
###########################################################################################
# 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'] # 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 = []