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, identify: 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(
identify=identify,
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, identify: 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(
identify=identify,
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 resolve 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 staff"""
researcher = self.create_assistant_agent(
identify="Researcher",
system_message="""You're a Senior Analysis Analyst. Your position is to:
1. Collect and analyze info on given matters
2. Determine key developments, 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(
identify="DataAnalyst",
system_message="""You're a Information Evaluation Knowledgeable. Your position is to:
1. Analyze quantitative knowledge and statistics
2. Create knowledge visualizations and charts
3. Determine 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(
identify="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 target market
4. Keep 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 staff"""
strategist = self.create_assistant_agent(
identify="BusinessStrategist",
system_message="""You're a Senior Enterprise Technique Advisor. 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
Suppose 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(
identify="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 funds and funding suggestions
Use quantitative evaluation and supply clear monetary insights."""
)
market_researcher = self.create_assistant_agent(
identify="MarketResearcher",
system_message="""You're a Market Analysis Specialist. Your position is to:
1. Analyze market developments and client habits
2. Analysis aggressive panorama and positioning
3. Determine goal markets and buyer segments
4. Present market sizing and alternative evaluation
Concentrate on 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 staff"""
developer = self.create_assistant_agent(
identify="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. Comply with greatest practices and coding requirements
At all times clarify your code and design choices.
Concentrate on clear, maintainable options."""
)
devops = self.create_assistant_agent(
identify="DevOpsEngineer",
system_message="""You're a DevOps Engineer. Your position is to:
1. Design deployment and infrastructure options
2. Automate construct, check, and deployment processes
3. Monitor system efficiency and reliability
4. Implement safety and scalability greatest practices
Concentrate on automation, reliability, and scalability."""
)
qa_engineer = self.create_assistant_agent(
identify="QAEngineer",
system_message="""You're a High quality Assurance Engineer. Your position is to:
1. Design complete check methods and instances
2. Determine 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 eventualities."""
)
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: Listing[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"""
staff = self.create_research_team()
agents_list = [team["researcher"], staff["analyst"], staff["writer"], staff["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 provide a complete analysis report following this workflow:
1. Researcher: Collect info and key insights about {subject}
2. DataAnalyst: Analyze any quantitative elements 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 ought to be an expert analysis report with:
- Government abstract
- Key findings and insights
- Information evaluation (if relevant)
- Conclusions and proposals
Start the analysis course of now.
"""
chat_result = staff["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"""
staff = self.create_business_team()
agents_list = [team["strategist"], staff["financial_analyst"],
staff["market_researcher"], staff["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
Ultimate deliverable ought to embody:
- Drawback evaluation and root causes
- Strategic suggestions
- Monetary influence evaluation
- Market alternative evaluation
- Implementation roadmap
Start the evaluation now.
"""
chat_result = staff["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"""
staff = self.create_development_team()
agents_list = [team["developer"], staff["devops"], staff["qa_engineer"], staff["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 an entire 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 embody:
- System structure and design
- Working code implementation
- Deployment configuration
- Take a look at instances and QA plan
- Documentation
Begin growth now.
"""
chat_result = staff["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('identify', 'Agent')}: {msg['content']}")
return "nn".be part of(final_messages)
def get_framework_stats(self) -> Dict[str, Any]:
"""Get framework statistics"""
return {
"brokers": checklist(self.brokers.keys()),
"group_chats": checklist(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()
}
