github-starred/awesome-llm-apps
2
0
Fork 0
mirror of https://github.com/Shubhamsaboo/awesome-llm-apps.git synced 2026-04-30 15:20:47 -05:00
Code Issues 25 Packages Projects Releases Wiki Activity

Migrate ai_services_agency to agency-swarm v1.7.0

Browse Source
This commit is contained in:
Nick Bobrowski
2026-02-08 22:44:06 +00:00
parent 12d491c009
commit e7e2e1b298
2 changed files with 59 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
advanced_ai_agents/multi_agent_apps/agent_teams/ai_services_agency/agency.py
View File

@@ -1,9 +1,13 @@
from typing import List, Literal, Dict, Optional import os
from agency_swarm import Agent, Agency, set_openai_key, BaseTool from typing import Literal
from pydantic import Field, BaseModel
from agency_swarm import Agent, Agency, BaseTool, ModelSettings
from pydantic import Field
import streamlit as st import streamlit as st
class AnalyzeProjectRequirements(BaseTool): class AnalyzeProjectRequirements(BaseTool):
"""Analyze project requirements and feasibility."""
project_name: str = Field(..., description="Name of the project") project_name: str = Field(..., description="Name of the project")
project_description: str = Field(..., description="Project description and goals") project_description: str = Field(..., description="Project description and goals")
project_type: Literal["Web Application", "Mobile App", "API Development", project_type: Literal["Web Application", "Mobile App", "API Development",
@@ -19,7 +23,10 @@ class AnalyzeProjectRequirements(BaseTool):
def run(self) -> str: def run(self) -> str:
"""Analyzes project and stores results in shared state""" """Analyzes project and stores results in shared state"""
if self._shared_state.get("project_analysis", None) is not None: if self.context is None:
raise RuntimeError("Tool context is not available.")
if self.context.get("project_analysis", None) is not None:
raise ValueError("Project analysis already exists. Please proceed with technical specification.") raise ValueError("Project analysis already exists. Please proceed with technical specification.")
analysis = { analysis = {
@@ -31,10 +38,12 @@ class AnalyzeProjectRequirements(BaseTool):
"requirements": ["Scalable architecture", "Security", "API integration"] "requirements": ["Scalable architecture", "Security", "API integration"]
} }
self._shared_state.set("project_analysis", analysis) self.context.set("project_analysis", analysis)
return "Project analysis completed. Please proceed with technical specification." return "Project analysis completed. Please proceed with technical specification."
class CreateTechnicalSpecification(BaseTool): class CreateTechnicalSpecification(BaseTool):
"""Create a technical specification based on project analysis."""
architecture_type: Literal["monolithic", "microservices", "serverless", "hybrid"] = Field( architecture_type: Literal["monolithic", "microservices", "serverless", "hybrid"] = Field(
..., ...,
description="Proposed architecture type" description="Proposed architecture type"
@@ -55,7 +64,10 @@ class CreateTechnicalSpecification(BaseTool):
def run(self) -> str: def run(self) -> str:
"""Creates technical specification based on analysis""" """Creates technical specification based on analysis"""
project_analysis = self._shared_state.get("project_analysis", None) if self.context is None:
raise RuntimeError("Tool context is not available.")
project_analysis = self.context.get("project_analysis", None)
if project_analysis is None: if project_analysis is None:
raise ValueError("Please analyze project requirements first using AnalyzeProjectRequirements tool.") raise ValueError("Please analyze project requirements first using AnalyzeProjectRequirements tool.")
@@ -66,7 +78,7 @@ class CreateTechnicalSpecification(BaseTool):
"scalability": self.scalability_requirements "scalability": self.scalability_requirements
} }
self._shared_state.set("technical_specification", spec) self.context.set("technical_specification", spec)
return f"Technical specification created for {project_analysis['name']}." return f"Technical specification created for {project_analysis['name']}."
def init_session_state() -> None: def init_session_state() -> None:
@@ -99,9 +111,8 @@ def main() -> None:
st.markdown("[Get your API key here](https://platform.openai.com/api-keys)") st.markdown("[Get your API key here](https://platform.openai.com/api-keys)")
return return
# Initialize agents with the provided API key # Agency Swarm v1 uses standard OPENAI_API_KEY resolution.
set_openai_key(st.session_state.api_key) os.environ["OPENAI_API_KEY"] = st.session_state.api_key
api_headers = {"Authorization": f"Bearer {st.session_state.api_key}"}
# Project Input Form # Project Input Form
with st.form("project_form"): with st.form("project_form"):
@@ -149,9 +160,6 @@ def main() -> None:
if submitted and project_name and project_description: if submitted and project_name and project_description:
try: try:
# Set OpenAI key
set_openai_key(st.session_state.api_key)
# Create agents # Create agents
ceo = Agent( ceo = Agent(
name="Project Director", name="Project Director",
@@ -170,9 +178,7 @@ def main() -> None:
3. Review the analysis results and provide strategic recommendations. 3. Review the analysis results and provide strategic recommendations.
""", """,
tools=[AnalyzeProjectRequirements], tools=[AnalyzeProjectRequirements],
api_headers=api_headers, model_settings=ModelSettings(temperature=0.7, max_tokens=25000),
temperature=0.7,
max_prompt_tokens=25000
) )
cto = Agent( cto = Agent(
@@ -191,9 +197,7 @@ def main() -> None:
3. Review the technical specification and provide additional recommendations. 3. Review the technical specification and provide additional recommendations.
""", """,
tools=[CreateTechnicalSpecification], tools=[CreateTechnicalSpecification],
api_headers=api_headers, model_settings=ModelSettings(temperature=0.5, max_tokens=25000),
temperature=0.5,
max_prompt_tokens=25000
) )
product_manager = Agent( product_manager = Agent(
@@ -203,9 +207,7 @@ def main() -> None:
- Manage project scope and timeline giving the roadmap of the project - Manage project scope and timeline giving the roadmap of the project
- Define product requirements and you should give potential products and features that can be built for the startup - Define product requirements and you should give potential products and features that can be built for the startup
""", """,
api_headers=api_headers, model_settings=ModelSettings(temperature=0.4, max_tokens=25000),
temperature=0.4,
max_prompt_tokens=25000
) )
developer = Agent( developer = Agent(
@@ -216,9 +218,7 @@ def main() -> None:
- Provide effort estimates - Provide effort estimates
- Review technical feasibility - Review technical feasibility
""", """,
api_headers=api_headers, model_settings=ModelSettings(temperature=0.3, max_tokens=25000),
temperature=0.3,
max_prompt_tokens=25000
) )
client_manager = Agent( client_manager = Agent(
@@ -229,25 +229,26 @@ def main() -> None:
- Manage expectations - Manage expectations
- Handle feedback - Handle feedback
""", """,
api_headers=api_headers, model_settings=ModelSettings(temperature=0.6, max_tokens=25000),
temperature=0.6,
max_prompt_tokens=25000
) )
# Create agency # Create agency
agency = Agency( agency = Agency(
[ ceo,
ceo, cto, product_manager, developer, client_manager, cto,
[ceo, cto], product_manager,
[ceo, product_manager], developer,
[ceo, developer], client_manager,
[ceo, client_manager], communication_flows=[
[cto, developer], (ceo, cto),
[product_manager, developer], (ceo, product_manager),
[product_manager, client_manager] (ceo, developer),
(ceo, client_manager),
(cto, developer),
(product_manager, developer),
(product_manager, client_manager),
], ],
async_mode='threading', shared_files_folder="shared_files",
shared_files='shared_files'
) )
# Prepare project info # Prepare project info
@@ -266,8 +267,9 @@ def main() -> None:
# Create tabs and run analysis # Create tabs and run analysis
with st.spinner("AI Services Agency is analyzing your project..."): with st.spinner("AI Services Agency is analyzing your project..."):
try: try:
# Get analysis from each agent using agency.get_completion() # Get analysis from each agent using get_response_sync.
ceo_response = agency.get_completion( ceo_response = str(
agency.get_response_sync(
message=f"""Analyze this project using the AnalyzeProjectRequirements tool: message=f"""Analyze this project using the AnalyzeProjectRequirements tool:
Project Name: {project_name} Project Name: {project_name}
Project Description: {project_description} Project Description: {project_description}
@@ -276,9 +278,11 @@ def main() -> None:
Use these exact values with the tool and wait for the analysis results.""", Use these exact values with the tool and wait for the analysis results.""",
recipient_agent=ceo recipient_agent=ceo
).final_output
) )
cto_response = agency.get_completion( cto_response = str(
agency.get_response_sync(
message=f"""Review the project analysis and create technical specifications using the CreateTechnicalSpecification tool. message=f"""Review the project analysis and create technical specifications using the CreateTechnicalSpecification tool.
Choose the most appropriate: Choose the most appropriate:
- architecture_type (monolithic/microservices/serverless/hybrid) - architecture_type (monolithic/microservices/serverless/hybrid)
@@ -287,24 +291,31 @@ def main() -> None:
Base your choices on the project requirements and analysis.""", Base your choices on the project requirements and analysis.""",
recipient_agent=cto recipient_agent=cto
).final_output
) )
pm_response = agency.get_completion( pm_response = str(
agency.get_response_sync(
message=f"Analyze project management aspects: {str(project_info)}", message=f"Analyze project management aspects: {str(project_info)}",
recipient_agent=product_manager, recipient_agent=product_manager,
additional_instructions="Focus on product-market fit and roadmap development, and coordinate with technical and marketing teams." additional_instructions="Focus on product-market fit and roadmap development, and coordinate with technical and marketing teams."
).final_output
) )
developer_response = agency.get_completion( developer_response = str(
agency.get_response_sync(
message=f"Analyze technical implementation based on CTO's specifications: {str(project_info)}", message=f"Analyze technical implementation based on CTO's specifications: {str(project_info)}",
recipient_agent=developer, recipient_agent=developer,
additional_instructions="Provide technical implementation details, optimal tech stack you would be using including the costs of cloud services (if any) and feasibility feedback, and coordinate with product manager and CTO to build the required products for the startup." additional_instructions="Provide technical implementation details, optimal tech stack you would be using including the costs of cloud services (if any) and feasibility feedback, and coordinate with product manager and CTO to build the required products for the startup."
).final_output
) )
client_response = agency.get_completion( client_response = str(
agency.get_response_sync(
message=f"Analyze client success aspects: {str(project_info)}", message=f"Analyze client success aspects: {str(project_info)}",
recipient_agent=client_manager, recipient_agent=client_manager,
additional_instructions="Provide detailed go-to-market strategy and customer acquisition plan, and coordinate with product manager." additional_instructions="Provide detailed go-to-market strategy and customer acquisition plan, and coordinate with product manager."
).final_output
) )
# Create tabs for different analyses # Create tabs for different analyses
@@ -362,4 +373,4 @@ def main() -> None:
st.rerun() st.rerun()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

4
advanced_ai_agents/multi_agent_apps/agent_teams/ai_services_agency/requirements.txt
View File

@@ -1,3 +1,3 @@
python-dotenv==1.0.1 python-dotenv==1.0.1
agency-swarm==0.4.1 agency-swarm==1.7.0
streamlit streamlit