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": (
{
"mannequin": "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": (
{
"mannequin": "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(title) = agent
return agent
def create_user_proxy(self, title: str = "UserProxy",
enable_code_execution: bool = True) -> UserProxyAgent:
"""Create consumer 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(
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 resolve duties.
Reply TERMINATE when the duty is solved."""
)
self.brokers(title) = agent
return agent
def create_research_team(self) -> Dict(str, autogen.Agent):
"""Create a research-focused agent workforce"""
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 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(
title="DataAnalyst",
system_message="""You're a Knowledge Evaluation Skilled. 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.
All the time 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. Rework analysis and evaluation into clear, participating content material
2. Create well-structured stories 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 workforce"""
strategist = self.create_assistant_agent(
title="BusinessStrategist",
system_message="""You're a Senior Enterprise Technique Guide. 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.
All the time present actionable suggestions.""",
use_pro_model=True
)
financial_analyst = self.create_assistant_agent(
title="FinancialAnalyst",
system_message="""You're a Monetary Evaluation Skilled. 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 developments 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 workforce"""
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 present code
4. Comply with finest practices and coding requirements
All the time clarify your code and design selections.
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 circumstances
2. Establish potential bugs and edge circumstances
3. Guarantee code high quality and efficiency requirements
4. Validate necessities and consumer acceptance standards
Be thorough and take into consideration edge circumstances 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 undertaking"""
workforce = self.create_research_team()
agents_list = (workforce("researcher"), workforce("analyst"), workforce("author"), 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 Undertaking: {matter}
Please collaborate to provide a complete analysis report following this workflow:
1. Researcher: Collect info and key insights about {matter}
2. DataAnalyst: Analyze any quantitative facets and create visualizations if wanted
3. Author: Create a well-structured closing 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 undertaking"""
workforce = self.create_business_team()
agents_list = (workforce("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 Undertaking: {business_problem}
Please collaborate to supply 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 closing suggestions
Last deliverable ought to embody:
- Downside evaluation and root causes
- Strategic suggestions
- Monetary influence 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 undertaking"""
workforce = self.create_development_team()
agents_list = (workforce("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 Undertaking: {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 assessments 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 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 closing 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('title', 'Agent')}: {msg('content material')}")
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)("mannequin"),
"temperature": self.llm_config("config_list")(0)("temperature")
},
"timestamp": datetime.now().isoformat()
}