233
class GeminiAutoGenFramework:
"""
Full AutoGen framework utilizing free Gemini API
Helps multi-agent conversations, code execution, and retrieval
"""
def __init__(self, gemini_api_key: str):
"""Initialize with Gemini API key"""
self.gemini_api_key = gemini_api_key
self.setup_gemini_config()
self.brokers: Dict[str, autogen.Agent] = {}
self.group_chats: Dict[str, GroupChat] = {}
def setup_gemini_config(self):
"""Configure Gemini for AutoGen"""
os.environ["GOOGLE_API_KEY"] = self.gemini_api_key
self.llm_config = {
"config_list": [
{
"model": "gemini/gemini-1.5-flash",
"api_key": self.gemini_api_key,
"api_type": "google",
"temperature": 0.7,
"max_tokens": 4096,
}
],
"timeout": 120,
"cache_seed": 42,
}
self.llm_config_pro = {
"config_list": [
{
"model": "gemini/gemini-1.5-pro",
"api_key": self.gemini_api_key,
"api_type": "google",
"temperature": 0.5,
"max_tokens": 8192,
}
],
"timeout": 180,
"cache_seed": 42,
}
def create_assistant_agent(self, title: str, system_message: str,
use_pro_model: bool = False) -> AssistantAgent:
"""Create a specialised assistant agent"""
config = self.llm_config_pro if use_pro_model else self.llm_config
agent = AssistantAgent(
title=title,
system_message=system_message,
llm_config=config,
human_input_mode="NEVER",
max_consecutive_auto_reply=10,
code_execution_config=False,
)
self.brokers[name] = agent
return agent
def create_user_proxy(self, title: str = "UserProxy",
enable_code_execution: bool = True) -> UserProxyAgent:
"""Create consumer proxy agent with non-compulsory code execution"""
code_config = {
"work_dir": "autogen_workspace",
"use_docker": False,
"timeout": 60,
"last_n_messages": 3,
} if enable_code_execution else False
agent = UserProxyAgent(
title=title,
human_input_mode="TERMINATE",
max_consecutive_auto_reply=0,
is_termination_msg=lambda x: x.get("content material", "").rstrip().endswith("TERMINATE"),
code_execution_config=code_config,
system_message="""A human admin. Work together with the brokers to unravel duties.
Reply TERMINATE when the duty is solved."""
)
self.brokers[name] = agent
return agent
def create_research_team(self) -> Dict[str, autogen.Agent]:
"""Create a research-focused agent group"""
researcher = self.create_assistant_agent(
title="Researcher",
system_message="""You're a Senior Analysis Analyst. Your position is to:
1. Collect and analyze info on given subjects
2. Establish key tendencies, patterns, and insights
3. Present complete analysis summaries
4. Cite sources and preserve objectivity
At all times construction your analysis with clear sections and bullet factors.
Be thorough however concise."""
)
analyst = self.create_assistant_agent(
title="DataAnalyst",
system_message="""You're a Information Evaluation Knowledgeable. Your position is to:
1. Analyze quantitative information and statistics
2. Create information visualizations and charts
3. Establish patterns and correlations
4. Present statistical insights and interpretations
Use Python code when wanted for calculations and visualizations.
At all times clarify your analytical strategy."""
)
author = self.create_assistant_agent(
title="Author",
system_message="""You're a Technical Author and Content material Strategist. Your position is to:
1. Remodel analysis and evaluation into clear, partaking content material
2. Create well-structured studies and articles
3. Guarantee content material is accessible to the audience
4. Preserve skilled tone and accuracy
Construction content material with clear headings, bullet factors, and conclusions."""
)
executor = self.create_user_proxy("CodeExecutor", enable_code_execution=True)
return {
"researcher": researcher,
"analyst": analyst,
"author": author,
"executor": executor
}
def create_business_team(self) -> Dict[str, autogen.Agent]:
"""Create enterprise evaluation group"""
strategist = self.create_assistant_agent(
title="BusinessStrategist",
system_message="""You're a Senior Enterprise Technique Marketing consultant. Your position is to:
1. Analyze enterprise issues and alternatives
2. Develop strategic suggestions and motion plans
3. Assess market dynamics and aggressive panorama
4. Present implementation roadmaps
Assume systematically and take into account a number of views.
At all times present actionable suggestions.""",
use_pro_model=True
)
financial_analyst = self.create_assistant_agent(
title="FinancialAnalyst",
system_message="""You're a Monetary Evaluation Knowledgeable. Your position is to:
1. Carry out monetary modeling and evaluation
2. Assess monetary dangers and alternatives
3. Calculate ROI, NPV, and different monetary metrics
4. Present price range and funding suggestions
Use quantitative evaluation and supply clear monetary insights."""
)
market_researcher = self.create_assistant_agent(
title="MarketResearcher",
system_message="""You're a Market Analysis Specialist. Your position is to:
1. Analyze market tendencies and shopper habits
2. Analysis aggressive panorama and positioning
3. Establish goal markets and buyer segments
4. Present market sizing and alternative evaluation
Give attention to actionable market insights and proposals."""
)
return {
"strategist": strategist,
"financial_analyst": financial_analyst,
"market_researcher": market_researcher,
"executor": self.create_user_proxy("BusinessExecutor")
}
def create_development_team(self) -> Dict[str, autogen.Agent]:
"""Create software program growth group"""
developer = self.create_assistant_agent(
title="SeniorDeveloper",
system_message="""You're a Senior Software program Developer. Your position is to:
1. Write high-quality, environment friendly code
2. Design software program structure and options
3. Debug and optimize current code
4. Observe finest practices and coding requirements
At all times clarify your code and design choices.
Give attention to clear, maintainable options."""
)
devops = self.create_assistant_agent(
title="DevOpsEngineer",
system_message="""You're a DevOps Engineer. Your position is to:
1. Design deployment and infrastructure options
2. Automate construct, take a look at, and deployment processes
3. Monitor system efficiency and reliability
4. Implement safety and scalability finest practices
Give attention to automation, reliability, and scalability."""
)
qa_engineer = self.create_assistant_agent(
title="QAEngineer",
system_message="""You're a High quality Assurance Engineer. Your position is to:
1. Design complete take a look at methods and instances
2. Establish potential bugs and edge instances
3. Guarantee code high quality and efficiency requirements
4. Validate necessities and consumer acceptance standards
Be thorough and take into consideration edge instances and failure situations."""
)
return {
"developer": developer,
"devops": devops,
"qa_engineer": qa_engineer,
"executor": self.create_user_proxy("DevExecutor", enable_code_execution=True)
}
def create_group_chat(self, brokers: Record[autogen.Agent], chat_name: str,
max_round: int = 10) -> GroupChat:
"""Create group chat with specified brokers"""
group_chat = GroupChat(
brokers=brokers,
messages=[],
max_round=max_round,
speaker_selection_method="round_robin",
allow_repeat_speaker=False,
)
self.group_chats[chat_name] = group_chat
return group_chat
def run_research_project(self, subject: str, max_rounds: int = 8) -> str:
"""Run a complete analysis undertaking"""
group = self.create_research_team()
agents_list = [team["researcher"], group["analyst"], group["writer"], group["executor"]]
group_chat = self.create_group_chat(agents_list, "research_chat", max_rounds)
supervisor = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config
)
initial_message = f"""
Analysis Challenge: {subject}
Please collaborate to supply a complete analysis report following this workflow:
1. Researcher: Collect info and key insights about {subject}
2. DataAnalyst: Analyze any quantitative features and create visualizations if wanted
3. Author: Create a well-structured remaining report primarily based on the analysis and evaluation
4. CodeExecutor: Execute any code wanted for evaluation or visualization
The ultimate deliverable must be knowledgeable analysis report with:
- Government abstract
- Key findings and insights
- Information evaluation (if relevant)
- Conclusions and proposals
Start the analysis course of now.
"""
chat_result = group["executor"].initiate_chat(
supervisor,
message=initial_message,
max_consecutive_auto_reply=0
)
return self._extract_final_result(chat_result)
def run_business_analysis(self, business_problem: str, max_rounds: int = 8) -> str:
"""Run enterprise evaluation undertaking"""
group = self.create_business_team()
agents_list = [team["strategist"], group["financial_analyst"],
group["market_researcher"], group["executor"]]
group_chat = self.create_group_chat(agents_list, "business_chat", max_rounds)
supervisor = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config_pro
)
initial_message = f"""
Enterprise Evaluation Challenge: {business_problem}
Please collaborate to offer complete enterprise evaluation following this strategy:
1. BusinessStrategist: Analyze the enterprise drawback and develop strategic framework
2. FinancialAnalyst: Assess monetary implications and create monetary fashions
3. MarketResearcher: Analysis market context and aggressive panorama
4. BusinessExecutor: Coordinate and compile remaining suggestions
Last deliverable ought to embrace:
- Downside evaluation and root causes
- Strategic suggestions
- Monetary influence evaluation
- Market alternative evaluation
- Implementation roadmap
Start the evaluation now.
"""
chat_result = group["executor"].initiate_chat(
supervisor,
message=initial_message,
max_consecutive_auto_reply=0
)
return self._extract_final_result(chat_result)
def run_development_project(self, project_description: str, max_rounds: int = 10) -> str:
"""Run software program growth undertaking"""
group = self.create_development_team()
agents_list = [team["developer"], group["devops"], group["qa_engineer"], group["executor"]]
group_chat = self.create_group_chat(agents_list, "dev_chat", max_rounds)
supervisor = GroupChatManager(
groupchat=group_chat,
llm_config=self.llm_config
)
initial_message = f"""
Improvement Challenge: {project_description}
Please collaborate to ship a whole software program answer:
1. SeniorDeveloper: Design structure and write core code
2. DevOpsEngineer: Plan deployment and infrastructure
3. QAEngineer: Design exams and high quality assurance strategy
4. DevExecutor: Execute code and coordinate implementation
Deliverables ought to embrace:
- System structure and design
- Working code implementation
- Deployment configuration
- Take a look at instances and QA plan
- Documentation
Begin growth now.
"""
chat_result = group["executor"].initiate_chat(
supervisor,
message=initial_message,
max_consecutive_auto_reply=0
)
return self._extract_final_result(chat_result)
def _extract_final_result(self, chat_result) -> str:
"""Extract and format remaining outcome from chat"""
if hasattr(chat_result, 'chat_history'):
messages = chat_result.chat_history
else:
messages = chat_result
final_messages = []
for msg in messages[-5:]:
if isinstance(msg, dict) and 'content material' in msg:
final_messages.append(f"{msg.get('title', 'Agent')}: {msg['content']}")
return "nn".be a part of(final_messages)
def get_framework_stats(self) -> Dict[str, Any]:
"""Get framework statistics"""
return {
"brokers": listing(self.brokers.keys()),
"group_chats": listing(self.group_chats.keys()),
"llm_config": {
"mannequin": self.llm_config["config_list"][0]["model"],
"temperature": self.llm_config["config_list"][0]["temperature"]
},
"timestamp": datetime.now().isoformat()
}
