-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathl5tasks.py
More file actions
163 lines (131 loc) · 4.52 KB
/
Copy pathl5tasks.py
File metadata and controls
163 lines (131 loc) · 4.52 KB
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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
# Warning control
import warnings
warnings.filterwarnings('ignore')
from crewai import Agent, Crew, Task
import os
#from utils import get_openai_api_key,get_serper_api_key
from dotenv import load_dotenv
def get_openai_api_key():
"""
Loads the OpenAI API key from the environment or a .env file.
Raises an error if not found.
"""
load_dotenv() # Load environment variables from .env if present
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise ValueError("OPENAI_API_KEY not found. Please set it in your environment or .env file.")
return api_key
openai_api_key = get_openai_api_key()
os.environ["OPENAI_MODEL_NAME"] = 'gpt-3.5-turbo'
os.environ["SERPER_API_KEY"] = get_openai_api_key()
from crewai_tools import ScrapeWebsiteTool, SerperDevTool
# Initialize the tools
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
# Agent 1: Venue Coordinator
venue_coordinator = Agent(
role="Venue Coordinator",
goal="Identify and book an appropriate venue "
"based on event requirements",
tools=[search_tool, scrape_tool],
verbose=True,
backstory=(
"With a keen sense of space and "
"understanding of event logistics, "
"you excel at finding and securing "
"the perfect venue that fits the event's theme, "
"size, and budget constraints."
)
)
# Agent 2: Logistics Manager
logistics_manager = Agent(
role='Logistics Manager',
goal=(
"Manage all logistics for the event "
"including catering and equipmen"
),
tools=[search_tool, scrape_tool],
verbose=True,
backstory=(
"Organized and detail-oriented, "
"you ensure that every logistical aspect of the event "
"from catering to equipment setup "
"is flawlessly executed to create a seamless experience."
)
)
# Agent 3: Marketing and Communications Agent
marketing_communications_agent = Agent(
role="Marketing and Communications Agent",
goal="Effectively market the event and "
"communicate with participants",
tools=[search_tool, scrape_tool],
verbose=True,
backstory=(
"Creative and communicative, "
"you craft compelling messages and "
"engage with potential attendees "
"to maximize event exposure and participation."
)
)
from pydantic import BaseModel
# Define a Pydantic model for venue details
# (demonstrating Output as Pydantic)
class VenueDetails(BaseModel):
name: str
address: str
capacity: int
booking_status: str
venue_task = Task(
description="Find a venue in {event_city} "
"that meets criteria for {event_topic}.",
expected_output="All the details of a specifically chosen"
"venue you found to accommodate the event.",
human_input=True,
output_json=VenueDetails,
output_file="venue_details.json",
# Outputs the venue details as a JSON file
agent=venue_coordinator
)
logistics_task = Task(
description="Coordinate catering and "
"equipment for an event "
"with {expected_participants} participants "
"on {tentative_date}.",
expected_output="Confirmation of all logistics arrangements "
"including catering and equipment setup.",
human_input=True,
async_execution=True,
agent=logistics_manager
)
marketing_task = Task(
description="Promote the {event_topic} "
"aiming to engage at least"
"{expected_participants} potential attendees.",
expected_output="Report on marketing activities "
"and attendee engagement formatted as markdown.",
async_execution=True,
output_file="marketing_report.md", # Outputs the report as a text file
agent=marketing_communications_agent
)
# Define the crew with agents and tasks
event_management_crew = Crew(
agents=[venue_coordinator,
logistics_manager,
marketing_communications_agent],
tasks=[venue_task,
logistics_task,
marketing_task],
verbose=True
)
event_details = {
'event_topic': "Tech Innovation Conference",
'event_description': "A gathering of tech innovators "
"and industry leaders "
"to explore future technologies.",
'event_city': "San Francisco",
'tentative_date': "2024-09-15",
'expected_participants': 500,
'budget': 20000,
'venue_type': "Conference Hall"
}
result = event_management_crew.kickoff(inputs=event_details)