@dataclass
class TaskConfig:
description: str
expected_output: str
precedence: TaskPriority
max_execution_time: int = 300
requires_human_input: bool = False
class SupervisorFramework:
"""
Superior Supervisor Agent Framework utilizing CrewAI
Manages a number of specialised brokers with hierarchical coordination
"""
def __init__(self, gemini_api_key: str, serper_api_key: str = None):
"""
Initialize the supervisor framework
Args:
gemini_api_key: Google Gemini API key (free tier)
serper_api_key: Serper API key for internet search (elective, has free tier)
"""
os.environ["GOOGLE_API_KEY"] = gemini_api_key
if serper_api_key:
os.environ["SERPER_API_KEY"] = serper_api_key
self.llm = ChatGoogleGenerativeAI(
mannequin="gemini-1.5-flash",
temperature=0.7,
max_tokens=2048
)
self.instruments = []
if serper_api_key:
self.instruments.append(SerperDevTool())
self.brokers = {}
self.supervisor = None
self.crew = None
print("🚀 SupervisorFramework initialized efficiently!")
print(f"📡 LLM Mannequin: {self.llm.mannequin}")
print(f"🛠️ Accessible instruments: {len(self.instruments)}")
def create_research_agent(self) -> Agent:
"""Create a specialised analysis agent"""
return Agent(
position="Senior Analysis Analyst",
objective="Conduct complete analysis and collect correct info on any given subject",
backstory="""You might be an professional analysis analyst with years of expertise in
info gathering, fact-checking, and synthesizing advanced knowledge from a number of sources.
You excel at discovering dependable sources and presenting well-structured analysis findings.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
instruments=self.instruments,
max_iter=3,
reminiscence=True
)
def create_analyst_agent(self) -> Agent:
"""Create a specialised knowledge analyst agent"""
return Agent(
position="Strategic Information Analyst",
objective="Analyze knowledge, determine patterns, and supply actionable insights",
backstory="""You're a strategic knowledge analyst with experience in statistical evaluation,
sample recognition, and enterprise intelligence. You rework uncooked knowledge and analysis
into significant insights that drive decision-making.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
max_iter=3,
reminiscence=True
)
def create_writer_agent(self) -> Agent:
"""Create a specialised content material author agent"""
return Agent(
position="Professional Technical Author",
objective="Create clear, partaking, and well-structured written content material",
backstory="""You might be an professional technical author with a expertise for making advanced
info accessible and fascinating. You concentrate on creating documentation,
stories, and content material that successfully communicates insights to various audiences.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
max_iter=3,
reminiscence=True
)
def create_reviewer_agent(self) -> Agent:
"""Create a high quality assurance reviewer agent"""
return Agent(
position="High quality Assurance Reviewer",
objective="Overview, validate, and enhance the standard of all deliverables",
backstory="""You're a meticulous high quality assurance professional with a watch for element
and a dedication to excellence. You guarantee all work meets excessive requirements of accuracy,
completeness, and readability earlier than closing supply.""",
verbose=True,
allow_delegation=False,
llm=self.llm,
max_iter=2,
reminiscence=True
)
def create_supervisor_agent(self) -> Agent:
"""Create the primary supervisor agent"""
return Agent(
position="Challenge Supervisor & Coordinator",
objective="Coordinate crew efforts, handle workflows, and guarantee challenge success",
backstory="""You might be an skilled challenge supervisor with experience in crew
coordination, workflow optimization, and high quality administration. You make sure that all
crew members work effectively in the direction of frequent objectives and keep excessive requirements
all through the challenge lifecycle.""",
verbose=True,
allow_delegation=True,
llm=self.llm,
max_iter=2,
reminiscence=True
)
def setup_agents(self):
"""Initialize all brokers within the framework"""
print("🤖 Organising specialised brokers...")
self.brokers = {
'researcher': self.create_research_agent(),
'analyst': self.create_analyst_agent(),
'author': self.create_writer_agent(),
'reviewer': self.create_reviewer_agent()
}
self.supervisor = self.create_supervisor_agent()
print(f"✅ Created {len(self.brokers)} specialised brokers + 1 supervisor")
for position, agent in self.brokers.objects():
print(f" └── {position.title()}: {agent.position}")
def create_task_workflow(self, subject: str, task_configs: Dict[str, TaskConfig]) -> Listing[Task]:
"""
Create a complete job workflow
Args:
subject: Primary subject/challenge focus
task_configs: Dictionary of job configurations
Returns:
Listing of CrewAI Activity objects
"""
duties = []
if 'analysis' in task_configs:
config = task_configs['research']
research_task = Activity(
description=f"{config.description} Deal with: {subject}",
expected_output=config.expected_output,
agent=self.brokers['researcher']
)
duties.append(research_task)
if 'evaluation' in task_configs:
config = task_configs['analysis']
analysis_task = Activity(
description=f"{config.description} Analyze the analysis findings about: {subject}",
expected_output=config.expected_output,
agent=self.brokers['analyst'],
context=duties
)
duties.append(analysis_task)
if 'writing' in task_configs:
config = task_configs['writing']
writing_task = Activity(
description=f"{config.description} Create content material about: {subject}",
expected_output=config.expected_output,
agent=self.brokers['writer'],
context=duties
)
duties.append(writing_task)
if 'assessment' in task_configs:
config = task_configs['review']
review_task = Activity(
description=f"{config.description} Overview all work associated to: {subject}",
expected_output=config.expected_output,
agent=self.brokers['reviewer'],
context=duties
)
duties.append(review_task)
supervisor_task = Activity(
description=f"""Because the challenge supervisor, coordinate your complete workflow for: {subject}.
Monitor progress, guarantee high quality requirements, resolve any conflicts between brokers,
and supply closing challenge oversight. Guarantee all deliverables meet necessities.""",
expected_output="""A complete challenge abstract together with:
- Government abstract of all accomplished work
- High quality evaluation of deliverables
- Suggestions for enhancements or subsequent steps
- Closing challenge standing report""",
agent=self.supervisor,
context=duties
)
duties.append(supervisor_task)
return duties
def execute_project(self,
subject: str,
task_configs: Dict[str, TaskConfig],
process_type: Course of = Course of.hierarchical) -> Dict[str, Any]:
"""
Execute an entire challenge utilizing the supervisor framework
Args:
subject: Primary challenge subject
task_configs: Activity configurations
process_type: CrewAI course of kind (hierarchical beneficial for supervisor)
Returns:
Dictionary containing execution outcomes
"""
print(f"🚀 Beginning challenge execution: {subject}")
print(f"📋 Course of kind: {process_type.worth}")
if not self.brokers or not self.supervisor:
self.setup_agents()
duties = self.create_task_workflow(subject, task_configs)
print(f"📝 Created {len(duties)} duties in workflow")
crew_agents = listing(self.brokers.values()) + [self.supervisor]
self.crew = Crew(
brokers=crew_agents,
duties=duties,
course of=process_type,
manager_llm=self.llm,
verbose=True,
reminiscence=True
)
print("🎯 Executing challenge...")
attempt:
end result = self.crew.kickoff()
return {
'standing': 'success',
'end result': end result,
'subject': subject,
'tasks_completed': len(duties),
'agents_involved': len(crew_agents)
}
besides Exception as e:
print(f"❌ Error throughout execution: {str(e)}")
return {
'standing': 'error',
'error': str(e),
'subject': subject
}
def get_crew_usage_metrics(self) -> Dict[str, Any]:
"""Get utilization metrics from the crew"""
if not self.crew:
return {'error': 'No crew execution discovered'}
attempt:
return {
'total_tokens_used': getattr(self.crew, 'total_tokens_used', 'Not obtainable'),
'total_cost': getattr(self.crew, 'total_cost', 'Not obtainable'),
'execution_time': getattr(self.crew, 'execution_time', 'Not obtainable')
}
besides:
return {'observe': 'Metrics not obtainable for this execution'}
