Created: 2025-01-02
Purpose: Design clean, modular application for agent swarm management, MCP configuration, RAG integration, and agent logging
- Monolithic Structure: Single 1070+ line file with mixed concerns
- Fake Functionality: Mock implementations and placeholder features
- Import Hell: Complex dependency chains with fallback mechanisms
- Mixed Responsibilities: Agent building, system monitoring, API configuration all in one
- Poor Modularity: Difficult to maintain, test, and extend
- Session State Bloat: Complex state management across unrelated features
- UI Coupling: Business logic tightly coupled with UI components
# Examples of problematic patterns in universal_composition_app.py:
# 1. Fake availability flags
AGENTS_AVAILABLE = False
VIBE_AGILE_AVAILABLE = False
AGILE_CEREMONIES_AVAILABLE = False
# 2. Complex fallback mechanisms
try:
from agents.development.requirements_analyst import RequirementsAnalyst
AGENTS_AVAILABLE = True
except ImportError:
RequirementsAnalyst = None # Fake fallback
AGENTS_AVAILABLE = False
# 3. Mixed concerns in single functions
def display_composition_dashboard():
display_framework_selector() # Framework concern
display_enterprise_modules() # Enterprise concern
display_architecture_designer() # Architecture concern
display_api_configuration() # API concern
display_system_status() # Monitoring concernapps/agent_swarm_manager/
├── main.py # Application entry point
├── config/
│ ├── __init__.py
│ ├── app_config.py # Application configuration
│ └── logging_config.py # Logging configuration
├── core/
│ ├── __init__.py
│ ├── app_state.py # Centralized state management
│ ├── session_manager.py # Session lifecycle management
│ └── event_bus.py # Inter-module communication
├── modules/
│ ├── __init__.py
│ ├── agent_builder/ # Agent creation and configuration
│ │ ├── __init__.py
│ │ ├── ui.py # Agent builder UI components
│ │ ├── logic.py # Agent creation logic
│ │ └── templates.py # Agent templates and presets
│ ├── swarm_manager/ # Swarm orchestration and management
│ │ ├── __init__.py
│ │ ├── ui.py # Swarm management UI
│ │ ├── coordinator.py # Swarm coordination logic
│ │ └── workflows.py # Predefined swarm workflows
│ ├── mcp_manager/ # MCP server and tool management
│ │ ├── __init__.py
│ │ ├── ui.py # MCP management UI
│ │ ├── server_control.py # MCP server operations
│ │ └── tool_registry.py # Tool management logic
│ ├── rag_manager/ # RAG system configuration
│ │ ├── __init__.py
│ │ ├── ui.py # RAG configuration UI
│ │ ├── document_processor.py # Document processing logic
│ │ └── knowledge_base.py # Knowledge base management
│ ├── chat_interface/ # Agent swarm chat interface
│ │ ├── __init__.py
│ │ ├── ui.py # Chat UI components
│ │ ├── message_handler.py # Message processing
│ │ └── conversation_manager.py # Conversation state management
│ └── monitoring/ # Agent logging and monitoring
│ ├── __init__.py
│ ├── ui.py # Monitoring dashboard UI
│ ├── logger.py # Agent logging system
│ └── analytics.py # Usage analytics and metrics
├── shared/
│ ├── __init__.py
│ ├── components/ # Reusable UI components
│ │ ├── __init__.py
│ │ ├── forms.py # Common form components
│ │ ├── charts.py # Chart and visualization components
│ │ └── layouts.py # Layout components
│ ├── utils/
│ │ ├── __init__.py
│ │ ├── validation.py # Input validation utilities
│ │ ├── formatting.py # Data formatting utilities
│ │ └── helpers.py # General helper functions
│ └── constants.py # Application constants
└── tests/
├── __init__.py
├── test_agent_builder.py
├── test_swarm_manager.py
├── test_mcp_manager.py
├── test_rag_manager.py
├── test_chat_interface.py
└── test_monitoring.py
Each module has one clear purpose:
- Agent Builder: Create and configure individual agents
- Swarm Manager: Orchestrate multiple agents in workflows
- MCP Manager: Manage MCP server, tools, and prompts
- RAG Manager: Configure RAG system and knowledge base
- Chat Interface: Provide real-time agent interaction
- Monitoring: Track agent performance and logging
Modules communicate through:
- Event Bus: Publish/subscribe pattern for inter-module communication
- Shared State: Centralized state management with clear interfaces
- Service Interfaces: Abstract interfaces for module interactions
Related functionality grouped together:
- UI components with their business logic
- Configuration with validation
- Data processing with error handling
# modules/agent_builder/ui.py
class AgentBuilderUI:
"""UI for creating and configuring individual agents."""
def render_agent_builder(self):
"""Render the main agent builder interface."""
with st.container():
self.render_agent_templates()
self.render_agent_configuration()
self.render_capability_selection()
self.render_agent_preview()
def render_agent_templates(self):
"""Display available agent templates."""
templates = [
"Requirements Analyst",
"Architecture Designer",
"Code Reviewer",
"Test Generator",
"Documentation Writer",
"Custom Agent"
]
# Template selection UI
def render_agent_configuration(self):
"""Agent-specific configuration options."""
# Name, description, personality, constraints
def render_capability_selection(self):
"""Select MCP tools and RAG capabilities for agent."""
# Tool selection, RAG integration options# modules/swarm_manager/ui.py
class SwarmManagerUI:
"""UI for orchestrating agent swarms."""
def render_swarm_manager(self):
"""Render the main swarm management interface."""
with st.container():
self.render_swarm_templates()
self.render_agent_selection()
self.render_workflow_designer()
self.render_swarm_controls()
def render_swarm_templates(self):
"""Display predefined swarm templates."""
templates = [
"Software Development Sprint",
"Code Review Process",
"Documentation Generation",
"Testing Pipeline",
"Research & Analysis",
"Custom Workflow"
]
def render_workflow_designer(self):
"""Visual workflow designer for agent coordination."""
# Drag-and-drop workflow builder
# Agent role assignments
# Task dependencies# modules/mcp_manager/ui.py
class MCPManagerUI:
"""UI for MCP server and tool management."""
def render_mcp_manager(self):
"""Render the main MCP management interface."""
with st.container():
self.render_server_status()
self.render_tool_registry()
self.render_prompt_database()
self.render_security_settings()
def render_tool_registry(self):
"""Display and manage available MCP tools."""
# Tool list with status
# Tool configuration
# Tool testing interface
def render_prompt_database(self):
"""Manage prompt templates and versions."""
# Prompt editor
# Version control
# A/B testing setup# modules/rag_manager/ui.py
class RAGManagerUI:
"""UI for RAG system configuration."""
def render_rag_manager(self):
"""Render the main RAG management interface."""
with st.container():
self.render_document_upload()
self.render_knowledge_base_status()
self.render_search_configuration()
self.render_embedding_settings()
def render_document_upload(self):
"""Document upload and processing interface."""
# Multi-format file upload
# Batch processing status
# Processing queue management
def render_knowledge_base_status(self):
"""Display knowledge base statistics and health."""
# Index statistics
# Search performance metrics
# Content freshness indicators# modules/chat_interface/ui.py
class ChatInterfaceUI:
"""UI for real-time agent swarm interaction."""
def render_chat_interface(self):
"""Render the main chat interface."""
with st.container():
self.render_swarm_selector()
self.render_chat_window()
self.render_agent_status_panel()
self.render_task_progress()
def render_chat_window(self):
"""Main chat interface with agent interactions."""
# Multi-agent conversation display
# Message input with context awareness
# Real-time agent responses
def render_agent_status_panel(self):
"""Display real-time agent status and activity."""
# Active agents list
# Current tasks
# Performance metrics# modules/monitoring/ui.py
class MonitoringUI:
"""UI for agent logging and performance monitoring."""
def render_monitoring_dashboard(self):
"""Render the main monitoring dashboard."""
with st.container():
self.render_real_time_metrics()
self.render_agent_logs()
self.render_performance_charts()
self.render_error_tracking()
def render_agent_logs(self):
"""Display comprehensive agent logging."""
# Real-time log streaming
# Log filtering and search
# Log level configuration
def render_performance_charts(self):
"""Display performance metrics and analytics."""
# Response time trends
# Success rate metrics
# Resource usage charts# core/event_bus.py
class EventBus:
"""Centralized event bus for inter-module communication."""
def __init__(self):
self.subscribers = {}
def subscribe(self, event_type: str, callback: callable):
"""Subscribe to specific event types."""
if event_type not in self.subscribers:
self.subscribers[event_type] = []
self.subscribers[event_type].append(callback)
def publish(self, event_type: str, data: dict):
"""Publish events to all subscribers."""
if event_type in self.subscribers:
for callback in self.subscribers[event_type]:
callback(data)
# Example usage:
# Agent Builder publishes agent creation
event_bus.publish("agent_created", {
"agent_id": "agent_123",
"agent_type": "requirements_analyst",
"capabilities": ["mcp_tools", "rag_search"]
})
# Swarm Manager subscribes to agent events
event_bus.subscribe("agent_created", swarm_manager.on_agent_created)# core/app_state.py
class AppState:
"""Centralized application state management."""
def __init__(self):
self.agents = {}
self.swarms = {}
self.mcp_status = {}
self.rag_config = {}
self.chat_sessions = {}
def get_agents(self) -> dict:
"""Get all configured agents."""
return self.agents.copy()
def add_agent(self, agent_id: str, agent_config: dict):
"""Add new agent to state."""
self.agents[agent_id] = agent_config
self.notify_observers("agents_updated")
def get_active_swarms(self) -> dict:
"""Get all active swarms."""
return {k: v for k, v in self.swarms.items() if v.get("status") == "active"}- Color Scheme: Professional blue/gray palette
- Typography: Clear, readable fonts with proper hierarchy
- Icons: Consistent icon library (Streamlit icons + custom)
- Spacing: Consistent margins and padding throughout
- Multi-Column Layouts: Efficient use of screen space
- Collapsible Sections: Reduce cognitive load
- Progressive Disclosure: Show details on demand
- Mobile-Friendly: Responsive design for different screen sizes
- Live Status Indicators: Real-time agent and system status
- Progress Bars: Visual feedback for long-running operations
- Auto-Refresh: Automatic updates for dynamic content
- WebSocket Integration: Real-time communication where needed
Frontend:
Framework: Streamlit with custom components
Components: streamlit-aggrid, plotly, streamlit-chat
Styling: Custom CSS with professional theme
State: Streamlit session state + custom state management
Backend:
MCP_Integration: Existing utils/mcp/ infrastructure
RAG_System: Enhanced context/context_engine.py
Agent_Framework: Existing agents/ infrastructure
Logging: Universal agent tracker integration
Data_Storage:
Configuration: YAML files for persistence
Logs: SQLite database with structured logging
Session_Data: Streamlit session state
Knowledge_Base: FAISS vector store# main.py
class ModularApp:
"""Main application with modular architecture."""
def __init__(self):
self.modules = {}
self.event_bus = EventBus()
self.app_state = AppState()
self.load_modules()
def load_modules(self):
"""Dynamically load all modules."""
module_configs = [
("agent_builder", AgentBuilderModule),
("swarm_manager", SwarmManagerModule),
("mcp_manager", MCPManagerModule),
("rag_manager", RAGManagerModule),
("chat_interface", ChatInterfaceModule),
("monitoring", MonitoringModule)
]
for name, module_class in module_configs:
try:
self.modules[name] = module_class(
event_bus=self.event_bus,
app_state=self.app_state
)
st.success(f"✅ {name} module loaded")
except Exception as e:
st.error(f"❌ Failed to load {name}: {e}")
def render_app(self):
"""Render the main application."""
st.sidebar.title("🤖 Agent Swarm Manager")
# Module navigation
selected_module = st.sidebar.selectbox(
"Select Module",
options=list(self.modules.keys()),
format_func=lambda x: x.replace("_", " ").title()
)
# Render selected module
if selected_module in self.modules:
self.modules[selected_module].render()- Lines of Code per Module: < 500 lines per module
- Cyclomatic Complexity: < 10 per function
- Test Coverage: > 90% for all modules
- Import Dependencies: < 5 external dependencies per module
- Page Load Time: < 2 seconds for module switching
- UI Responsiveness: < 100ms for user interactions
- Error Rate: < 1% for normal operations
- User Task Completion: > 95% success rate
- Module Coupling: Low coupling between modules
- Code Duplication: < 5% duplicate code
- Documentation Coverage: 100% public API documented
- Refactoring Safety: All modules independently testable
- Day 1-2: Core architecture (event bus, state management, module loader)
- Day 3-4: Shared components and utilities
- Day 5: Basic UI framework and navigation
- Day 1-2: Agent Builder and Swarm Manager modules
- Day 3-4: MCP Manager and RAG Manager modules
- Day 5: Module integration and testing
- Day 1-2: Chat Interface and Monitoring modules
- Day 3-4: Real-time features and WebSocket integration
- Day 5: Performance optimization and comprehensive testing
This modular architecture provides a clean, maintainable, and extensible foundation for agent swarm management while eliminating the bloat and fake functionality of the current universal composition app.