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 non-obligatory 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 function is to:
1. Collect and analyze data on given matters
2. Establish key developments, patterns, and insights
3. Present complete analysis summaries
4. Cite sources and preserve objectivity
All the time 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 function 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.
All the time clarify your analytical strategy."""
)
author = self.create_assistant_agent(
identify="Author",
system_message="""You're a Technical Author and Content material Strategist. Your function is to:
1. Rework analysis and evaluation into clear, partaking content material
2. Create well-structured experiences and articles
3. Guarantee content material is accessible to the audience
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 workforce"""
strategist = self.create_assistant_agent(
identify="BusinessStrategist",
system_message="""You're a Senior Enterprise Technique Marketing consultant. Your function 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 contemplate a number of views.
All the time present actionable suggestions.""",
use_pro_model=True
)
financial_analyst = self.create_assistant_agent(
identify="FinancialAnalyst",
system_message="""You're a Monetary Evaluation Professional. Your function 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 function is to:
1. Analyze market developments and shopper habits
2. Analysis aggressive panorama and positioning
3. Establish goal markets and buyer segments
4. Present market sizing and alternative evaluation
Deal with 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 improvement workforce"""
developer = self.create_assistant_agent(
identify="SeniorDeveloper",
system_message="""You're a Senior Software program Developer. Your function 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 finest practices and coding requirements
All the time clarify your code and design choices.
Deal with clear, maintainable options."""
)
devops = self.create_assistant_agent(
identify="DevOpsEngineer",
system_message="""You're a DevOps Engineer. Your function 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 finest practices
Deal with automation, reliability, and scalability."""
)
qa_engineer = self.create_assistant_agent(
identify="QAEngineer",
system_message="""You're a High quality Assurance Engineer. Your function is to:
1. Design complete check methods and instances
2. Establish potential bugs and edge instances
3. Guarantee code high quality and efficiency requirements
4. Validate necessities and person 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, matter: str, max_rounds: int = 8) -> str:
"""Run a complete analysis venture"""
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: {matter}
Please collaborate to supply a complete analysis report following this workflow:
1. Researcher: Collect data and key insights about {matter}
2. DataAnalyst: Analyze any quantitative points and create visualizations if wanted
3. Author: Create a well-structured ultimate 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:
- Govt abstract
- Key findings and insights
- Knowledge evaluation (if relevant)
- Conclusions and proposals
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 venture"""
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 offer complete enterprise evaluation following this strategy:
1. BusinessStrategist: Analyze the enterprise downside 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 ultimate suggestions
Closing deliverable ought to embrace:
- Downside 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 improvement venture"""
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"""
Growth Venture: {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 checks 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 improvement 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 ultimate end result 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 a 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()
}
