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 person proxy agent with elective 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 workforce"""
researcher = self.create_assistant_agent(
identify="Researcher",
system_message="""You're a Senior Analysis Analyst. Your position is to:
1. Collect and analyze data on given subjects
2. Determine key traits, 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 Knowledge Evaluation Professional. 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 method."""
)
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, participating 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 workforce"""
strategist = self.create_assistant_agent(
identify="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 think about 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 Professional. 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 traits and shopper habits
2. Analysis aggressive panorama and positioning
3. Determine goal markets and buyer segments
4. Present market sizing and alternative evaluation
Give attention to actionable market insights and suggestions."""
)
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 workforce"""
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 present code
4. Observe greatest practices and coding requirements
At all times clarify your code and design choices.
Give attention to 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
Give attention to 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 circumstances
2. Determine potential bugs and edge circumstances
3. Guarantee code high quality and efficiency requirements
4. Validate necessities and person acceptance standards
Be thorough and take into consideration edge circumstances 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: 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 challenge"""
workforce = self.create_research_team()
agents_list = [team["researcher"], workforce["analyst"], workforce["writer"], workforce["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 Venture: {subject}
Please collaborate to supply a complete analysis report following this workflow:
1. Researcher: Collect data and key insights about {subject}
2. DataAnalyst: Analyze any quantitative facets and create visualizations if wanted
3. Author: Create a well-structured last report primarily based on the analysis and evaluation
4. CodeExecutor: Execute any code wanted for evaluation or visualization
The ultimate deliverable needs to be an expert analysis report with:
- Government abstract
- Key findings and insights
- Knowledge evaluation (if relevant)
- Conclusions and suggestions
Start the analysis course of now.
"""
chat_result = workforce["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 challenge"""
workforce = self.create_business_team()
agents_list = [team["strategist"], workforce["financial_analyst"],
workforce["market_researcher"], workforce["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 Venture: {business_problem}
Please collaborate to supply complete enterprise evaluation following this method:
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 last suggestions
Closing deliverable ought to embrace:
- Drawback evaluation and root causes
- Strategic suggestions
- Monetary impression evaluation
- Market alternative evaluation
- Implementation roadmap
Start the evaluation now.
"""
chat_result = workforce["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 challenge"""
workforce = self.create_development_team()
agents_list = [team["developer"], workforce["devops"], workforce["qa_engineer"], workforce["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 Venture: {project_description}
Please collaborate to ship a whole software program resolution:
1. SeniorDeveloper: Design structure and write core code
2. DevOpsEngineer: Plan deployment and infrastructure
3. QAEngineer: Design assessments and high quality assurance method
4. DevExecutor: Execute code and coordinate implementation
Deliverables ought to embrace:
- System structure and design
- Working code implementation
- Deployment configuration
- Take a look at circumstances and QA plan
- Documentation
Begin growth now.
"""
chat_result = workforce["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 last 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": 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()
}