Tourism_Agent / crew.py
geekphoenix's picture
Update crew.py
57c65bd verified
raw
history blame
3.42 kB
from crewai import Agent, Crew, Process, Task, LLM
from crewai.project import CrewBase, agent, crew, task
from tools.custom_tool import web_search_tool
from dotenv import load_dotenv
import os
# Load environment variables
load_dotenv()
@CrewBase
class tourismAgent():
"""Tourism Agent crew for providing personalized travel recommendations"""
agents_config = 'config/agents.yaml'
tasks_config = 'config/tasks.yaml'
def __init__(self, api_key=None):
"""
Initialize the tourism agent with an optional API key.
Args:
api_key (str, optional): Anthropic API key. If not provided,
will try to load from environment variables.
"""
# Use provided API key or fall back to environment variable
self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
if not self.api_key:
raise ValueError(
"No API key provided. Please either:\n"
"1. Pass an API key when initializing tourismAgent(api_key='your-key')\n"
"2. Set ANTHROPIC_API_KEY environment variable\n"
"3. Add ANTHROPIC_API_KEY to your .env file"
)
# Initialize LLM with the API key
self.llm = LLM(
model="claude-3-haiku-20240307",
api_key=self.api_key,
temperature=0.1, # Slightly higher for more creative responses
)
@agent
def tourism_activities_suggestor(self) -> Agent:
return Agent(
config=self.agents_config['tourism_activities_suggestor'],
verbose=True,
tools=[web_search_tool],
llm=self.llm,
max_retry_limit=3, # Reduced retry limit for faster responses
allow_delegation=False, # Prevent delegation loops
)
@task
def activities_suggestion(self) -> Task:
return Task(
config=self.tasks_config['activities_suggestion'],
agent=self.tourism_activities_suggestor(),
)
@crew
def crew(self) -> Crew:
"""Creates the tourism Agent crew"""
return Crew(
agents=self.agents, # Automatically created by the @agent decorator
tasks=self.tasks, # Automatically created by the @task decorator
process=Process.sequential,
verbose=True,
max_rpm=20, # Rate limiting
)
def validate_api_key(self):
"""
Validate the API key format and test connection.
Returns:
tuple: (is_valid: bool, message: str)
"""
if not self.api_key:
return False, "No API key provided"
if not self.api_key.startswith("sk-ant-"):
return False, "Invalid API key format. Should start with 'sk-ant-'"
if len(self.api_key) < 50:
return False, "API key appears to be too short"
# You could add a simple test call here if needed
return True, "API key format is valid"
# Helper function for backward compatibility
def create_tourism_agent_with_key(api_key):
"""
Helper function to create a tourism agent with a specific API key.
Args:
api_key (str): Anthropic API key
Returns:
tourismAgent: Configured tourism agent instance
"""
return tourismAgent(api_key=api_key)