WHOOSH/docs/TEAM_COMPOSER_SPEC.md

WHOOSH Team Composer Specification

LLM-Powered Autonomous Team Formation Engine

Overview

The Team Composer is the central intelligence of WHOOSH's Autonomous AI Development Teams architecture. It uses Large Language Models to analyze incoming tasks, determine optimal team compositions, and orchestrate the formation of self-organizing AI development teams through sophisticated reasoning and pattern matching.

🎯 Core Purpose

Analyze development tasks and automatically compose optimal AI development teams with the right mix of capabilities, experience levels, and collaborative dynamics to successfully complete complex software development challenges.

🧠 Architecture Components

1. Task Analysis Engine

Input Processing Pipeline

@dataclass
class TaskAnalysisInput:
    # Core task information
    title: str
    description: str
    requirements: List[str]
    repository: Optional[str] = None
    
    # Context and constraints
    priority: TaskPriority = TaskPriority.MEDIUM
    deadline: Optional[datetime] = None
    estimated_complexity: Optional[ComplexityLevel] = None
    budget_limit: Optional[int] = None
    
    # Technical context
    technology_stack: List[str] = field(default_factory=list)
    existing_codebase: Optional[CodebaseContext] = None
    integration_requirements: List[str] = field(default_factory=list)
    
    # Quality requirements
    quality_gates: QualityRequirements = field(default_factory=QualityRequirements)
    compliance_requirements: List[str] = field(default_factory=list)
    security_level: SecurityLevel = SecurityLevel.STANDARD

Multi-Stage Analysis Process

Stage 1: Task Classification

class TaskClassifier:
    """Classifies tasks into categories and complexity levels"""
    
    async def classify_task(self, task: TaskAnalysisInput) -> TaskClassification:
        # LLM-powered classification prompt
        classification_prompt = f"""
        Analyze this software development task and provide classification:
        
        Task: {task.title}
        Description: {task.description}
        Requirements: {', '.join(task.requirements)}
        
        Classify this task across multiple dimensions:
        
        1. TASK TYPE (select primary):
           - feature_development: New functionality implementation
           - bug_fix: Fixing existing issues
           - refactoring: Code improvement without feature changes  
           - migration: System/data migration
           - research: Investigation and proof-of-concept
           - optimization: Performance/efficiency improvements
           - security: Security enhancements
           - integration: System integration work
           - maintenance: Ongoing maintenance tasks
           
        2. COMPLEXITY ASSESSMENT (0.0-1.0 scale):
           - Technical complexity
           - Business logic complexity  
           - Integration complexity
           - Risk level
           
        3. DOMAIN ANALYSIS:
           - Primary technical domains required
           - Secondary supporting domains
           - Cross-cutting concerns
           
        4. EFFORT ESTIMATION:
           - Estimated hours for completion
           - Confidence level in estimate
           - Key uncertainty factors
           
        5. SUCCESS FACTORS:
           - Critical requirements for success
           - Major risk factors
           - Quality gates needed
           
        Respond in structured JSON format.
        """
        
        return await self.llm_client.analyze(
            prompt=classification_prompt,
            model="deepseek-r1:14b",  # Reasoning model for analysis
            response_schema=TaskClassification
        )

Stage 2: Skill Requirements Analysis

class SkillRequirementsAnalyzer:
    """Analyzes required skills and proficiency levels"""
    
    async def analyze_skill_requirements(
        self, 
        task: TaskAnalysisInput,
        classification: TaskClassification
    ) -> SkillRequirements:
        
        analysis_prompt = f"""
        Based on this classified task, determine detailed skill requirements:
        
        Task Type: {classification.task_type}
        Complexity: {classification.complexity_score}
        Domains: {', '.join(classification.primary_domains)}
        
        For each required skill domain, specify:
        
        1. SKILL PROFICIENCY MAPPING:
           Domain -> Required Level -> Weight in Decision
           - novice (0.0-0.3): Basic understanding
           - intermediate (0.3-0.6): Working knowledge
           - advanced (0.6-0.8): Strong expertise  
           - expert (0.8-1.0): Deep specialization
           
        2. CRITICAL SKILLS (must-have):
           - Skills absolutely required for success
           - Minimum proficiency levels
           - Why these skills are critical
           
        3. DESIRABLE SKILLS (nice-to-have):
           - Skills that would improve outcomes
           - Preferred proficiency levels
           - Value-add they provide
           
        4. SKILL COMBINATIONS:
           - Which skills work well together
           - Complementary skill pairs
           - Potential skill conflicts
           
        5. EXPERIENCE REQUIREMENTS:
           - Years of experience per domain
           - Specific technology experience
           - Project type experience
           
        Analyze the task requirements and provide detailed skill mapping.
        """
        
        return await self.llm_client.analyze(
            prompt=analysis_prompt,
            model="qwen2.5-coder:32b",  # Code-focused analysis
            response_schema=SkillRequirements
        )

Stage 3: Risk Assessment

class RiskAssessmentEngine:
    """Identifies and evaluates project risks"""
    
    async def assess_risks(
        self, 
        task: TaskAnalysisInput,
        classification: TaskClassification,
        skill_requirements: SkillRequirements
    ) -> RiskAssessment:
        
        risk_prompt = f"""
        Conduct comprehensive risk analysis for this development task:
        
        Context:
        - Task: {task.title}
        - Type: {classification.task_type}
        - Complexity: {classification.complexity_score}
        - Deadline: {task.deadline}
        - Required Skills: {skill_requirements.critical_skills}
        
        Analyze risks across categories:
        
        1. TECHNICAL RISKS:
           - Implementation complexity risks
           - Technology/integration risks
           - Performance/scalability risks  
           - Dependency and compatibility risks
           
        2. TEAM RISKS:
           - Skill gap risks
           - Communication/coordination risks
           - Agent availability risks
           - Team chemistry/collaboration risks
           
        3. PROJECT RISKS:
           - Scope creep risks
           - Timeline/deadline risks
           - Quality/defect risks
           - Stakeholder/requirements risks
           
        4. EXTERNAL RISKS:
           - Dependency on external systems
           - Regulatory/compliance risks
           - Security/vulnerability risks
           - Market/business risks
           
        For each identified risk:
        - Probability (0.0-1.0)
        - Impact severity (minor/moderate/major/critical)
        - Mitigation strategies
        - Early warning indicators
        - Contingency plans
        
        Provide structured risk assessment with prioritized mitigation strategies.
        """
        
        return await self.llm_client.analyze(
            prompt=risk_prompt,
            model="deepseek-r1:14b",  # Reasoning for risk analysis
            response_schema=RiskAssessment
        )

2. Team Composition Engine

Composition Strategy Selection

class TeamCompositionStrategist:
    """Determines optimal team composition strategies"""
    
    COMPOSITION_TEMPLATES = {
        "minimal_viable": {
            "description": "Smallest team that can deliver",
            "max_size": 3,
            "focus": "efficiency",
            "risk_tolerance": "medium"
        },
        "balanced_standard": {
            "description": "Standard balanced team",
            "max_size": 5,
            "focus": "balance", 
            "risk_tolerance": "low"
        },
        "comprehensive_quality": {
            "description": "Quality-focused with redundancy",
            "max_size": 7,
            "focus": "quality",
            "risk_tolerance": "very_low"
        },
        "rapid_prototype": {
            "description": "Fast iteration and experimentation",
            "max_size": 4,
            "focus": "speed",
            "risk_tolerance": "high"
        },
        "security_critical": {
            "description": "Security-first with multiple reviews",
            "max_size": 6,
            "focus": "security",
            "risk_tolerance": "very_low"
        }
    }
    
    async def select_composition_strategy(
        self, 
        task: TaskAnalysisInput,
        classification: TaskClassification,
        risk_assessment: RiskAssessment
    ) -> CompositionStrategy:
        
        strategy_prompt = f"""
        Select optimal team composition strategy for this task:
        
        Task Context:
        - Complexity: {classification.complexity_score}
        - Priority: {task.priority.value}
        - Deadline Pressure: {self._assess_timeline_pressure(task.deadline)}
        - Risk Level: {risk_assessment.overall_risk_level}
        - Security Requirements: {task.security_level.value}
        
        Available Strategies: {list(self.COMPOSITION_TEMPLATES.keys())}
        
        Consider these factors:
        1. Task complexity vs team coordination overhead
        2. Quality requirements vs speed needs
        3. Risk tolerance vs resource efficiency
        4. Security needs vs development velocity
        5. Available agent pool vs ideal team size
        
        Select the best strategy and explain reasoning:
        - Which template best fits the requirements?
        - What customizations are needed?
        - What are the trade-offs?
        - How does this optimize for success?
        """
        
        return await self.llm_client.analyze(
            prompt=strategy_prompt,
            model="deepseek-r1:14b",
            response_schema=CompositionStrategy
        )

Role Definition Engine

class RoleDefinitionEngine:
    """Defines specific roles needed for the team"""
    
    CORE_ROLE_TEMPLATES = {
        "architect": {
            "responsibilities": ["System design", "Architecture decisions", "Technical leadership"],
            "required_skills": ["system_design", "architecture_patterns", "technical_leadership"],
            "min_proficiency": 0.8,
            "typical_effort_percentage": 20
        },
        "backend_developer": {
            "responsibilities": ["API development", "Business logic", "Data management"],
            "required_skills": ["backend_development", "api_design", "database_design"],
            "min_proficiency": 0.7,
            "typical_effort_percentage": 40
        },
        "frontend_developer": {
            "responsibilities": ["UI implementation", "User experience", "Client-side logic"],
            "required_skills": ["frontend_development", "ui_ux", "javascript"],
            "min_proficiency": 0.7,
            "typical_effort_percentage": 30
        },
        "qa_engineer": {
            "responsibilities": ["Test design", "Quality assurance", "Bug validation"],
            "required_skills": ["testing", "quality_assurance", "automation"],
            "min_proficiency": 0.6,
            "typical_effort_percentage": 15
        },
        "security_specialist": {
            "responsibilities": ["Security review", "Threat modeling", "Vulnerability assessment"],
            "required_skills": ["security", "threat_modeling", "vulnerability_assessment"],
            "min_proficiency": 0.8,
            "typical_effort_percentage": 10
        },
        "devops_engineer": {
            "responsibilities": ["Deployment", "Infrastructure", "CI/CD"],
            "required_skills": ["devops", "deployment", "infrastructure"],
            "min_proficiency": 0.7,
            "typical_effort_percentage": 15
        }
    }
    
    async def define_team_roles(
        self, 
        skill_requirements: SkillRequirements,
        composition_strategy: CompositionStrategy,
        estimated_hours: int
    ) -> List[TeamRole]:
        
        role_definition_prompt = f"""
        Define specific team roles for this development task:
        
        Requirements:
        - Required Skills: {skill_requirements.critical_skills}
        - Desirable Skills: {skill_requirements.desirable_skills}
        - Team Size Target: {composition_strategy.target_team_size}
        - Strategy Focus: {composition_strategy.focus}
        - Total Estimated Hours: {estimated_hours}
        
        Available Role Templates: {list(self.CORE_ROLE_TEMPLATES.keys())}
        
        For each role needed:
        1. ROLE SELECTION:
           - Which template roles are essential?
           - Which roles can be combined?
           - What custom roles are needed?
           
        2. ROLE CUSTOMIZATION:
           - Specific responsibilities for this task
           - Required skill levels and domains
           - Estimated effort hours and percentage
           - Dependencies on other roles
           
        3. ROLE PRIORITIZATION:
           - Which roles are absolutely required?
           - Which roles are desirable but optional?
           - Which roles can be filled by same agent?
           
        4. COLLABORATION PATTERNS:
           - How do roles interact?
           - What are the communication needs?
           - Where are the handoff points?
           
        5. QUALITY GATES:
           - What approvals does each role need?
           - What artifacts must each role produce?
           - How is role completion validated?
           
        Design optimal role structure that balances coverage, efficiency, and collaboration.
        """
        
        return await self.llm_client.analyze(
            prompt=role_definition_prompt,
            model="qwen2.5-coder:32b",
            response_schema=List[TeamRole]
        )

3. Agent Matching Engine

Capability Assessment

class AgentCapabilityMatcher:
    """Matches available agents to role requirements"""
    
    async def find_suitable_agents(
        self, 
        role: TeamRole,
        available_agents: List[Agent],
        team_context: TeamContext
    ) -> List[AgentMatch]:
        
        matches = []
        for agent in available_agents:
            match_score = await self._calculate_match_score(
                agent, role, team_context
            )
            if match_score.overall_score >= 0.6:  # Minimum threshold
                matches.append(match_score)
        
        # Sort by overall score, then by availability
        return sorted(
            matches, 
            key=lambda m: (m.overall_score, m.availability_score),
            reverse=True
        )
    
    async def _calculate_match_score(
        self, 
        agent: Agent,
        role: TeamRole, 
        team_context: TeamContext
    ) -> AgentMatch:
        
        matching_prompt = f"""
        Evaluate agent fit for team role:
        
        AGENT PROFILE:
        - ID: {agent.agent_id}
        - Specialization: {agent.specialization}
        - Capabilities: {agent.capabilities}
        - Experience: {agent.completed_teams} teams, {agent.success_rate} success rate
        - Current Load: {agent.current_load}
        - Reputation: {agent.reputation_score}
        
        ROLE REQUIREMENTS:
        - Role: {role.role_name}
        - Required Skills: {role.required_skills}
        - Min Proficiency: {role.minimum_proficiency}
        - Responsibilities: {role.responsibilities}
        - Estimated Hours: {role.estimated_effort_hours}
        
        TEAM CONTEXT:
        - Team Size: {team_context.target_size}
        - Timeline: {team_context.deadline}
        - Complexity: {team_context.complexity}
        - Other Members: {team_context.confirmed_members}
        
        Evaluate match across dimensions:
        
        1. SKILL MATCH (0.0-1.0):
           - How well do agent's skills align with role requirements?
           - Proficiency levels in required domains
           - Breadth vs depth considerations
           - Gap analysis for missing skills
           
        2. EXPERIENCE MATCH (0.0-1.0):
           - Relevant project experience
           - Similar task completion history
           - Track record in this role type
           - Success rate in comparable complexity
           
        3. AVAILABILITY MATCH (0.0-1.0):
           - Current workload vs capacity
           - Timeline alignment
           - Scheduling conflicts
           - Commitment level feasibility
           
        4. TEAM CHEMISTRY (0.0-1.0):
           - Collaboration history with confirmed members
           - Communication style compatibility
           - Working style alignment
           - Previous team feedback
           
        5. VALUE-ADD POTENTIAL (0.0-1.0):
           - Unique capabilities beyond minimum requirements
           - Innovation potential
           - Mentorship/leadership qualities
           - Cross-functional contributions
           
        Provide detailed scoring with explanations and overall recommendation.
        """
        
        return await self.llm_client.analyze(
            prompt=matching_prompt,
            model="phi4:14b",  # Good at structured analysis
            response_schema=AgentMatch
        )

Team Chemistry Analysis

class TeamChemistryAnalyzer:
    """Analyzes team dynamics and compatibility"""
    
    async def analyze_team_compatibility(
        self, 
        proposed_team: List[AgentMatch],
        team_context: TeamContext
    ) -> TeamCompatibilityReport:
        
        chemistry_prompt = f"""
        Analyze team chemistry for proposed team composition:
        
        PROPOSED TEAM:
        {self._format_team_summary(proposed_team)}
        
        TEAM CONTEXT:
        - Task Complexity: {team_context.complexity}
        - Timeline Pressure: {team_context.timeline_pressure}
        - Quality Requirements: {team_context.quality_requirements}
        - Communication Needs: {team_context.communication_intensity}
        
        Analyze team dynamics across dimensions:
        
        1. COMMUNICATION COMPATIBILITY:
           - Communication style alignment
           - Language/cultural considerations  
           - Previous collaboration patterns
           - Conflict resolution approaches
           
        2. WORKING STYLE HARMONY:
           - Work pace compatibility
           - Decision-making preferences
           - Planning vs improvisation styles
           - Detail orientation levels
           
        3. SKILL COMPLEMENTARITY:
           - Skill overlap vs gaps
           - Knowledge sharing potential
           - Learning/mentoring opportunities
           - Redundancy for risk mitigation
           
        4. LEADERSHIP DYNAMICS:
           - Natural leadership tendencies
           - Authority/hierarchy preferences
           - Decision-making distribution
           - Conflict resolution capabilities
           
        5. INNOVATION POTENTIAL:
           - Creative collaboration likelihood
           - Diverse perspective benefits
           - Innovation catalyst combinations
           - Risk-taking alignment
           
        6. STRESS RESPONSE COMPATIBILITY:
           - Performance under pressure
           - Support vs independence needs
           - Deadline management approaches
           - Quality vs speed trade-off handling
           
        Provide compatibility assessment with:
        - Overall team chemistry score
        - Potential friction points
        - Mitigation strategies
        - Optimization recommendations
        """
        
        return await self.llm_client.analyze(
            prompt=chemistry_prompt,
            model="deepseek-r1:14b",
            response_schema=TeamCompatibilityReport
        )

4. Formation Orchestration Engine

Formation Timeline Planning

class FormationTimelinePlanner:
    """Plans optimal team formation timeline"""
    
    async def plan_formation_timeline(
        self, 
        team_composition: TeamComposition,
        agent_availability: Dict[str, AgentAvailability]
    ) -> FormationTimeline:
        
        timeline_prompt = f"""
        Plan optimal team formation timeline:
        
        TEAM COMPOSITION:
        - Required Roles: {len(team_composition.required_roles)}
        - Optional Roles: {len(team_composition.optional_roles)}
        - Priority Order: {team_composition.formation_priority}
        
        AGENT AVAILABILITY:
        {self._format_availability_summary(agent_availability)}
        
        Plan formation considering:
        
        1. ROLE PRIORITY SEQUENCING:
           - Which roles must be filled first?
           - Which roles can wait?
           - What are the dependencies?
           - How to minimize formation time?
           
        2. AGENT RECRUITMENT STRATEGY:
           - Simultaneous vs sequential recruitment
           - Fallback options for each role
           - Wait time vs compromise trade-offs
           - Backup agent identification
           
        3. FORMATION MILESTONES:
           - Core team formation (minimum viable)
           - Full team formation (all required roles)
           - Enhanced team formation (optional roles)
           - Team integration and kickoff
           
        4. RISK MITIGATION:
           - What if key agents decline?
           - How to handle scheduling conflicts?
           - When to start with partial team?
           - Escalation procedures
           
        5. TIMELINE OPTIMIZATION:
           - Fastest formation path
           - Most reliable formation path  
           - Best quality formation path
           - Resource-efficient formation path
           
        Provide detailed timeline with milestones, dependencies, and contingencies.
        """
        
        return await self.llm_client.analyze(
            prompt=timeline_prompt,
            model="qwen2.5-coder:32b",
            response_schema=FormationTimeline
        )

GITEA Integration

class GITEATeamIntegrator:
    """Integrates team formation with GITEA issues"""
    
    async def create_team_formation_issue(
        self, 
        team_composition: TeamComposition,
        formation_timeline: FormationTimeline,
        repository_url: str
    ) -> GITEAIssue:
        
        issue_content = await self._generate_issue_content(
            team_composition, formation_timeline
        )
        
        # Create GITEA issue with structured metadata
        issue_data = {
            "title": f"Team Formation: {team_composition.team_name}",
            "body": issue_content,
            "labels": self._generate_labels(team_composition),
            "assignees": [],  # Will be populated as agents join
            "milestone": None,
            "metadata": {
                "team_id": team_composition.team_id,
                "formation_strategy": team_composition.strategy,
                "required_roles": [role.role_name for role in team_composition.required_roles],
                "optional_roles": [role.role_name for role in team_composition.optional_roles],
                "ucxl_address": team_composition.ucxl_address,
                "p2p_channel": team_composition.p2p_channel_id
            }
        }
        
        return await self.gitea_client.create_issue(repository_url, issue_data)
    
    async def _generate_issue_content(
        self, 
        composition: TeamComposition,
        timeline: FormationTimeline
    ) -> str:
        
        content_prompt = f"""
        Generate GITEA issue content for team formation:
        
        TEAM: {composition.team_name}
        TASK: {composition.task_title}
        
        Create structured issue content including:
        
        ## Team Formation Overview
        - Task description and objectives
        - Success criteria and deliverables
        - Timeline and milestones
        
        ## Team Composition
        - Required roles with checkboxes
        - Optional roles with checkboxes
        - Skills and proficiency requirements
        
        ## Application Process
        - How agents can apply
        - Required information for applications
        - Review and approval process
        
        ## Team Coordination
        - P2P channel information
        - Communication protocols
        - Collaboration tools and processes
        
        ## Quality Gates
        - Consensus requirements
        - Review processes
        - Completion criteria
        
        ## Timeline & Milestones
        - Formation deadlines
        - Project milestones
        - Key deliverable dates
        
        Use clear markdown formatting with task lists, tables, and sections.
        Make it actionable for agents to self-organize and join the team.
        """
        
        return await self.llm_client.generate(
            prompt=content_prompt,
            model="phi4:14b",
            max_tokens=2000
        )

🚀 Implementation Architecture

Core Service Structure

# Team Composer Service
class WHOOSHTeamComposer:
    """Main Team Composer service orchestrating all components"""
    
    def __init__(self, config: ComposerConfig):
        # Core analysis engines
        self.task_classifier = TaskClassifier(config.llm_config)
        self.skill_analyzer = SkillRequirementsAnalyzer(config.llm_config)
        self.risk_assessor = RiskAssessmentEngine(config.llm_config)
        
        # Composition engines
        self.composition_strategist = TeamCompositionStrategist(config.llm_config)
        self.role_definer = RoleDefinitionEngine(config.llm_config)
        
        # Matching engines
        self.agent_matcher = AgentCapabilityMatcher(config.llm_config)
        self.chemistry_analyzer = TeamChemistryAnalyzer(config.llm_config)
        
        # Formation orchestration
        self.timeline_planner = FormationTimelinePlanner(config.llm_config)
        self.gitea_integrator = GITEATeamIntegrator(config.gitea_config)
        
        # Data layer
        self.database = DatabaseManager(config.db_config)
        self.agent_registry = AgentRegistry(self.database)
    
    async def analyze_and_compose_team(
        self, 
        task_input: TaskAnalysisInput,
        constraints: TeamConstraints = None
    ) -> TeamCompositionResult:
        """Complete analysis and team composition pipeline"""
        
        try:
            # Stage 1: Analyze the task
            classification = await self.task_classifier.classify_task(task_input)
            skill_requirements = await self.skill_analyzer.analyze_skill_requirements(
                task_input, classification
            )
            risk_assessment = await self.risk_assessor.assess_risks(
                task_input, classification, skill_requirements
            )
            
            # Store analysis results
            analysis_id = await self.database.store_task_analysis(
                classification, skill_requirements, risk_assessment
            )
            
            # Stage 2: Determine composition strategy
            composition_strategy = await self.composition_strategist.select_composition_strategy(
                task_input, classification, risk_assessment
            )
            
            # Stage 3: Define team roles
            team_roles = await self.role_definer.define_team_roles(
                skill_requirements, composition_strategy, classification.estimated_duration_hours
            )
            
            # Stage 4: Find and match agents
            available_agents = await self.agent_registry.get_available_agents(
                required_skills=skill_requirements.critical_skills,
                constraints=constraints
            )
            
            role_matches = {}
            for role in team_roles:
                matches = await self.agent_matcher.find_suitable_agents(
                    role, available_agents, TeamContext.from_analysis(classification)
                )
                role_matches[role.role_name] = matches[:5]  # Top 5 candidates
            
            # Stage 5: Optimize team composition
            proposed_teams = self._generate_team_combinations(role_matches, composition_strategy)
            
            best_team = None
            best_chemistry_score = 0.0
            
            for team_option in proposed_teams[:3]:  # Analyze top 3 combinations
                compatibility = await self.chemistry_analyzer.analyze_team_compatibility(
                    team_option, TeamContext.from_analysis(classification)
                )
                
                if compatibility.overall_score > best_chemistry_score:
                    best_chemistry_score = compatibility.overall_score
                    best_team = team_option
                    best_compatibility = compatibility
            
            # Stage 6: Plan formation timeline
            agent_availability = await self._get_agent_availability(best_team)
            formation_timeline = await self.timeline_planner.plan_formation_timeline(
                TeamComposition.from_matches(best_team), agent_availability
            )
            
            # Stage 7: Create team composition result
            result = TeamCompositionResult(
                analysis_id=analysis_id,
                team_composition=TeamComposition.from_matches(best_team),
                formation_timeline=formation_timeline,
                compatibility_report=best_compatibility,
                alternative_options=proposed_teams[1:3] if len(proposed_teams) > 1 else []
            )
            
            return result
            
        except Exception as e:
            logger.error(f"Team composition failed: {str(e)}")
            raise TeamCompositionError(f"Failed to compose team: {str(e)}")

LLM Integration Layer

class LLMClient:
    """Unified LLM client supporting multiple models and providers"""
    
    def __init__(self, config: LLMConfig):
        self.config = config
        self.ollama_client = OllamaClient(config.ollama_endpoints)
        self.cloud_clients = self._init_cloud_clients(config.cloud_providers)
    
    async def analyze(
        self, 
        prompt: str,
        model: str,
        response_schema: Type[BaseModel] = None,
        temperature: float = 0.1,
        max_tokens: int = 4000
    ) -> Any:
        """Analyze using specified model with structured output"""
        
        # Route to appropriate provider based on model
        if model.startswith("gpt-"):
            client = self.cloud_clients["openai"]
        elif model.startswith("claude-"):
            client = self.cloud_clients["anthropic"] 
        else:
            # Use local Ollama models
            client = self.ollama_client
        
        response = await client.complete(
            model=model,
            prompt=prompt,
            temperature=temperature,
            max_tokens=max_tokens,
            structured_output=response_schema is not None
        )
        
        if response_schema:
            return response_schema.parse_raw(response.content)
        
        return response.content
    
    async def _select_optimal_model(
        self, 
        task_type: str,
        complexity: float,
        context_length: int
    ) -> str:
        """Select optimal model based on task characteristics"""
        
        # Model selection logic based on task requirements
        if task_type in ["reasoning", "risk_assessment"]:
            if complexity > 0.8:
                return "deepseek-r1:14b"  # Best reasoning model
            else:
                return "deepseek-r1:7b"   # Lighter reasoning
                
        elif task_type in ["code_analysis", "skill_matching"]:
            if context_length > 8000:
                return "qwen2.5-coder:32b"  # Large context code model
            else:
                return "qwen2.5-coder:14b"
                
        elif task_type in ["structured_analysis", "classification"]:
            return "phi4:14b"  # Good at structured tasks
            
        else:
            # Default to general purpose model
            return "llama3.1:8b"

Performance Optimization

class AnalysisCache:
    """Caches analysis results for performance optimization"""
    
    def __init__(self, redis_client: Redis):
        self.redis = redis_client
        self.cache_ttl = 3600  # 1 hour
    
    async def get_cached_analysis(
        self, 
        task_hash: str,
        analysis_type: str
    ) -> Optional[Any]:
        """Get cached analysis result"""
        
        cache_key = f"analysis:{analysis_type}:{task_hash}"
        cached = await self.redis.get(cache_key)
        
        if cached:
            return json.loads(cached)
        
        return None
    
    async def cache_analysis(
        self, 
        task_hash: str,
        analysis_type: str,
        result: Any
    ) -> None:
        """Cache analysis result"""
        
        cache_key = f"analysis:{analysis_type}:{task_hash}"
        serialized = json.dumps(result, default=str)
        
        await self.redis.setex(cache_key, self.cache_ttl, serialized)
    
    def _generate_task_hash(self, task: TaskAnalysisInput) -> str:
        """Generate hash for task to enable caching"""
        
        content = f"{task.title}|{task.description}|{','.join(task.requirements)}"
        return hashlib.sha256(content.encode()).hexdigest()

📊 Monitoring & Analytics

Team Composition Metrics

class ComposerMetrics:
    """Tracks Team Composer performance and effectiveness"""
    
    async def track_analysis_performance(
        self, 
        analysis_id: str,
        stage: str,
        duration_ms: int,
        success: bool,
        model_used: str
    ) -> None:
        """Track analysis stage performance"""
        
        metrics = {
            "analysis_id": analysis_id,
            "stage": stage,
            "duration_ms": duration_ms,
            "success": success,
            "model": model_used,
            "timestamp": datetime.utcnow()
        }
        
        await self.influx_client.write_point("composer_analysis", metrics)
    
    async def track_team_formation_outcome(
        self, 
        team_id: str,
        formation_success: bool,
        formation_time_minutes: int,
        team_performance_score: float = None
    ) -> None:
        """Track team formation success and performance"""
        
        metrics = {
            "team_id": team_id,
            "formation_success": formation_success,
            "formation_time_minutes": formation_time_minutes,
            "team_performance_score": team_performance_score,
            "timestamp": datetime.utcnow()
        }
        
        await self.influx_client.write_point("team_formation", metrics)

Quality Feedback Loop

class ComposerFeedbackLoop:
    """Learns from team outcomes to improve composition"""
    
    async def record_team_outcome(
        self, 
        team_id: str,
        analysis_id: str,
        final_outcome: TeamOutcome
    ) -> None:
        """Record team completion outcome for learning"""
        
        feedback = TeamCompositionFeedback(
            analysis_id=analysis_id,
            team_id=team_id,
            success_score=final_outcome.success_score,
            quality_score=final_outcome.quality_score,
            timeline_accuracy=final_outcome.timeline_accuracy,
            team_satisfaction=final_outcome.team_satisfaction,
            lessons_learned=final_outcome.lessons_learned
        )
        
        await self.database.store_composition_feedback(feedback)
        
        # Trigger model retraining if enough feedback accumulated
        await self._check_retraining_trigger()
    
    async def _improve_composition_models(self) -> None:
        """Use feedback to improve composition accuracy"""
        
        # Analyze patterns in successful vs unsuccessful teams
        feedback_data = await self.database.get_recent_feedback(days=30)
        
        # Identify improvement opportunities
        analysis_prompt = f"""
        Analyze team composition feedback to identify improvement patterns:
        
        {self._format_feedback_data(feedback_data)}
        
        Identify:
        1. What patterns correlate with successful teams?
        2. What composition mistakes are repeated?
        3. Which role combinations work best?
        4. What risk factors are underestimated?
        5. How can agent matching be improved?
        
        Provide actionable insights for improving team composition logic.
        """
        
        insights = await self.llm_client.analyze(
            prompt=analysis_prompt,
            model="deepseek-r1:14b",
            response_schema=CompositionInsights
        )
        
        # Update composition rules and heuristics
        await self._update_composition_rules(insights)

This Team Composer specification provides the foundation for WHOOSH's intelligent team formation capabilities, enabling sophisticated analysis of development tasks and automatic composition of optimal AI development teams through advanced LLM reasoning and pattern matching.