Session 4 Module A - Test Solutions¶

class FlowState(BaseModel):
    # Core workflow data
    project_id: str
    current_phase: str
    completed_phases: List[str]

    # Task management
    task_queue: List[Dict[str, Any]]
    active_tasks: Dict[str, Dict[str, Any]]
    completed_tasks: Dict[str, Dict[str, Any]]

    # Team coordination
    team_assignments: Dict[str, List[str]]
    resource_allocation: Dict[str, float]
    performance_metrics: Dict[str, Any]

    # Flow control
    flow_status: str
    error_history: List[Dict[str, Any]]
    checkpoint_data: Dict[str, Any]

@listen(initiate_research_project)
def orchestrate_research_teams(self, state: FlowState) -> FlowState:
    # Dynamic team formation based on task requirements
    optimal_teams = self._form_optimal_research_teams(state.task_queue)

    # Assign tasks to teams with workload balancing
    team_assignments = {}
    for team_id, team_config in optimal_teams.items():
        assigned_tasks = self._assign_tasks_to_team(
            team_config, 
            state.task_queue,
            state.resource_allocation
        )
        team_assignments[team_id] = assigned_tasks

@router(execute_parallel_research)
def route_based_on_quality(self, state: FlowState) -> str:
    # Calculate average quality score
    average_quality = sum(
        task.get("quality_score", 0) for task in successful_tasks
    ) / len(successful_tasks)

    # Determine routing based on quality thresholds
    if average_quality >= 0.8:
        return "synthesize_research_findings"  # High quality - proceed to synthesis
    elif average_quality >= 0.6:
        return "enhance_research_quality"     # Medium quality - enhancement needed
    else:
        return "retry_research_phase"         # Low quality - retry needed

# Weighted composite score
effectiveness_score = (
    skill_coverage * 0.35 +      # 35% - how well team covers required skills
    performance_score * 0.25 +   # 25% - historical performance
    collaboration_score * 0.20 + # 20% - team compatibility
    availability_score * 0.15 +  # 15% - agent availability
    size_efficiency * 0.05       # 5% - prefer smaller effective teams
)

# Find highest skill level in team for this skill
for agent_id in team_members:
    agent_capability = self.agent_capabilities.get(agent_id)
    if agent_capability:
        # Check primary skills first
        if skill in agent_capability.primary_skills:
            agent_level = agent_capability.primary_skills[skill]
            if agent_level.value > best_team_level.value:
                best_team_level = agent_level
        # Check secondary skills
        elif skill in agent_capability.secondary_skills:
            agent_level = agent_capability.secondary_skills[skill]
            if agent_level.value > best_team_level.value:
                best_team_level = agent_level

if any(indicator in task_lower for indicator in ["complex", "advanced", "expert"]):
    required_skills[skill] = SkillLevel.EXPERT
elif any(indicator in task_lower for indicator in ["simple", "basic", "quick"]):
    required_skills[skill] = SkillLevel.INTERMEDIATE