✅博主简介：擅长数据搜集与处理、建模仿真、程序设计、仿真代码、论文写作与指导，毕业论文、期刊论文经验交流。
✅成品或者定制，扫描文章底部微信二维码。
（1）研发项目组合排程问题的特征分析与数学建模
企业研发活动是将创新理念转化为技术成果和市场产品的关键环节，研发项目的有效管理直接影响企业的核心竞争力和可持续发展能力。随着市场竞争加剧和技术迭代加速，企业同时开展的研发项目数量不断增加，这些项目在时间上相互交叉、在资源上相互竞争、在目标上相互关联，形成了复杂的研发项目组合。如何在有限的资源条件下合理安排多个研发项目的进度计划，使企业整体研发目标得到最优实现，是研发项目管理面临的核心挑战。传统的单项目进度管理方法难以处理项目间的资源争夺和优先级协调问题，需要从项目组合的整体视角进行统筹规划。
研发项目组合排程问题具有区别于一般项目排程的特殊性质。首先研发项目的持续时间和资源需求具有较大的不确定性，研发过程中可能出现技术难题导致任务延期，也可能因为技术突破而提前完成，这种不确定性给进度计划的制定和执行带来困难。其次研发项目之间可能存在技术关联和知识共享关系，某些项目的研发成果可以为其他项目提供技术支撑，合理安排这些项目的先后顺序可以提高整体研发效率。再次研发人员作为核心资源具有专业分工和能力差异，不同人员适合承担不同类型的研发任务，人员配置需要考虑专业匹配度和团队协作因素。最后研发项目往往承载着企业的战略目标，不同项目的战略重要性和紧迫程度不同，排程方案需要体现项目优先级的差异。
研发项目组合排程问题可以建立混合整数规划模型进行形式化描述。决策变量包括各任务的开始时间、各任务分配的资源数量、各时段各资源的使用情况等。目标函数通常以最小化项目组合的总完工时间或总延误惩罚为主，也可以包含资源使用成本、负载均衡程度等分项。约束条件包括任务先后顺序约束确保前置任务完成后后续任务才能开始，资源可用性约束确保任一时点的资源使用量不超过可用量，资源连续性约束确保某些任务一旦开始就不能中断。对于考虑不确定性的问题，还需要引入随机变量或模糊参数来刻画持续时间和资源需求的不确定性，建立鲁棒优化或随机规划模型。模型的可行性可以通过商业求解器在小规模实例上进行验证，但对于实际规模的问题，求解器往往难以在可接受时间内获得最优解。
（2）项目集启发式算法与分支定界精确算法设计
针对研发项目组合排程问题的求解，可以根据问题规模和求解精度要求设计不同类型的算法。对于大规模问题或实时决策场景，启发式算法能够快速生成可行解；对于小规模问题或需要精确解的场景，分支定界算法可以获得最优解并证明其最优性；对于中等规模问题，元启发式算法在解的质量和计算时间之间取得较好平衡。
基于项目集的启发式算法从项目组合的结构特征出发设计调度规则。项目集是项目组合中具有紧密关联关系的项目子集，同一项目集内的项目往往共享某些关键资源或存在技术依赖关系。算法首先识别项目组合中的项目集结构，然后采用资源导向的调度策略依次安排各项目集的进度。在安排某一项目集时，优先考虑资源约束最紧张的时段，将任务安排在资源可用的最早时间窗口内，尽量避免资源冲突的发生。当资源冲突不可避免时，根据项目优先级和任务松弛时间确定哪些任务延后执行。这种启发式方法的优点是计算速度快，能够在短时间内生成可行的排程方案，缺点是解的质量依赖于调度规则的设计，可能与最优解存在较大差距。
分支定界算法是求解整数规划问题的经典精确算法，通过系统地搜索解空间来找到最优解。算法将原问题不断分解为子问题构成搜索树，每个子问题对应原问题的一个子区域。通过计算子问题的下界来判断该子区域是否可能包含最优解，如果下界已经超过当前已知的最优解则剪枝不再搜索该区域。针对研发项目组合排程问题，可以设计基于最佳优先规则的节点选择策略，优先搜索下界最小的子问题，加快找到最优解的速度。下界的计算可以采用线性松弛或拉格朗日松弛的方法，松弛方法的选择影响下界的紧密程度和计算效率。分支策略的设计需要考虑问题的特殊结构，可以选择对目标函数影响较大的变量进行分支，以快速缩小解空间。
（3）改进人工免疫算法的设计与实验验证分析
人工免疫算法是模拟生物免疫系统工作机理的一类元启发式算法，具有良好的全局搜索能力和多样性保持机制，适合求解复杂的组合优化问题。标准人工免疫算法包含抗原识别、抗体生成、克隆选择、高频变异等操作，通过迭代进化不断改进抗体群体的质量。针对研发项目组合排程问题的特点，可以对标准算法进行改进以提高求解效率和解的质量。
在抗体编码方面，采用基于优先级的间接编码方式，每个抗体是一个实数向量，向量的每个分量对应一个任务的调度优先级。解码时根据优先级顺序依次安排任务的开始时间，同时考虑前置任务约束和资源约束。这种编码方式的优点是任何抗体都能解码为可行解，避免了处理不可行解的额外开销。在克隆选择方面，根据抗体的亲和度（即解的质量）确定克隆数量，亲和度越高的抗体获得越多的克隆机会，使搜索向优质解区域集中。在高频变异方面，变异程度与抗体亲和度成反比，高亲和度抗体进行小范围精细搜索，低亲和度抗体进行大范围探索搜索，在深度搜索和广度搜索之间取得平衡。
为加速算法收敛，可以引入局部搜索算子对优秀抗体进行邻域优化。邻域结构的设计需要考虑问题的特点，可以定义任务交换邻域和任务插入邻域，通过小范围调整任务顺序来改进解的质量。此外还可以引入精英记忆库保存搜索过程中发现的最优解集，防止优秀解在进化过程中丢失。算法的参数设置如种群规模、克隆系数、变异概率等对搜索性能有重要影响，可以通过参数敏感性分析确定合适的参数取值范围。
import numpy as np from typing import List, Dict, Tuple, Set from dataclasses import dataclass, field from heapq import heappush, heappop import random import copy @dataclass class RDTask: task_id: str project_id: str duration: int resource_requirements: Dict[str, int] predecessors: List[str] = field(default_factory=list) skill_requirements: List[str] = field(default_factory=list) @dataclass class RDProject: project_id: str tasks: List[RDTask] priority: float deadline: int strategic_value: float class ResourcePool: def __init__(self, capacities: Dict[str, int]): self.capacities = capacities self.usage_timeline = {r: {} for r in capacities} def check_availability(self, resources: Dict[str, int], start: int, duration: int) -> bool: for resource, amount in resources.items(): if resource not in self.capacities: return False for t in range(start, start + duration): current_usage = self.usage_timeline[resource].get(t, 0) if current_usage + amount > self.capacities[resource]: return False return True def allocate(self, resources: Dict[str, int], start: int, duration: int): for resource, amount in resources.items(): for t in range(start, start + duration): self.usage_timeline[resource][t] = self.usage_timeline[resource].get(t, 0) + amount def reset(self): self.usage_timeline = {r: {} for r in self.capacities} class ProjectPortfolioScheduler: def __init__(self, projects: List[RDProject], resource_capacities: Dict[str, int]): self.projects = projects self.resource_pool = ResourcePool(resource_capacities) self.all_tasks = [] self.task_dict = {} for p in projects: for t in p.tasks: self.all_tasks.append(t) self.task_dict[t.task_id] = t def calculate_earliest_start(self, task: RDTask, completed: Dict[str, int]) -> int: earliest = 0 for pred_id in task.predecessors: if pred_id in completed: earliest = max(earliest, completed[pred_id]) return earliest def schedule_with_priority(self, priority_vector: np.ndarray) -> Dict[str, Tuple[int, int]]: self.resource_pool.reset() task_order = sorted(range(len(self.all_tasks)), key=lambda i: -priority_vector[i]) schedule = {} completed = {} for idx in task_order: task = self.all_tasks[idx] preds_done = all(p in completed for p in task.predecessors) if not preds_done: continue earliest = self.calculate_earliest_start(task, completed) start = earliest while not self.resource_pool.check_availability(task.resource_requirements, start, task.duration): start += 1 if start > 1000: break self.resource_pool.allocate(task.resource_requirements, start, task.duration) finish = start + task.duration schedule[task.task_id] = (start, finish) completed[task.task_id] = finish return schedule def evaluate_schedule(self, schedule: Dict[str, Tuple[int, int]]) -> float: if not schedule: return float('inf') makespan = max(s[1] for s in schedule.values()) tardiness = 0 for project in self.projects: project_finish = 0 for task in project.tasks: if task.task_id in schedule: project_finish = max(project_finish, schedule[task.task_id][1]) if project_finish > project.deadline: tardiness += (project_finish - project.deadline) * project.priority * project.strategic_value return makespan + tardiness * 10 class BranchAndBound: def __init__(self, scheduler: ProjectPortfolioScheduler): self.scheduler = scheduler self.best_solution = None self.best_cost = float('inf') self.nodes_explored = 0 def solve(self, time_limit: int = 60) -> Tuple[Dict, float]: n_tasks = len(self.scheduler.all_tasks) initial_priority = np.random.rand(n_tasks) initial_schedule = self.scheduler.schedule_with_priority(initial_priority) self.best_cost = self.scheduler.evaluate_schedule(initial_schedule) self.best_solution = initial_schedule heap = [(0, 0, [], set())] import time start_time = time.time() while heap and (time.time() - start_time) < time_limit: _, level, fixed, remaining = heappop(heap) self.nodes_explored += 1 if level == n_tasks: continue unassigned = [i for i in range(n_tasks) if i not in remaining] for task_idx in unassigned[:3]: new_fixed = fixed + [task_idx] new_remaining = remaining | {task_idx} priority = np.zeros(n_tasks) for i, idx in enumerate(new_fixed): priority[idx] = n_tasks - i for i in range(n_tasks): if i not in new_remaining: priority[i] = random.random() schedule = self.scheduler.schedule_with_priority(priority) cost = self.scheduler.evaluate_schedule(schedule) if cost < self.best_cost: self.best_cost = cost self.best_solution = schedule lower_bound = cost * 0.9 if lower_bound < self.best_cost: heappush(heap, (lower_bound, level + 1, new_fixed, new_remaining)) return self.best_solution, self.best_cost class ImmuneAlgorithm: def __init__(self, scheduler: ProjectPortfolioScheduler, pop_size: int = 50): self.scheduler = scheduler self.pop_size = pop_size self.n_tasks = len(scheduler.all_tasks) self.population = [self.create_antibody() for _ in range(pop_size)] self.memory = [] self.memory_size = 10 def create_antibody(self) -> np.ndarray: return np.random.rand(self.n_tasks) def calculate_affinity(self, antibody: np.ndarray) -> float: schedule = self.scheduler.schedule_with_priority(antibody) cost = self.scheduler.evaluate_schedule(schedule) return 1.0 / (1.0 + cost) def clone_and_mutate(self, antibody: np.ndarray, affinity: float) -> List[np.ndarray]: n_clones = max(1, int(self.pop_size * affinity)) clones = [] for _ in range(n_clones): clone = antibody.copy() mutation_rate = 1.0 - affinity mutation_mask = np.random.rand(self.n_tasks) < mutation_rate clone[mutation_mask] = np.random.rand(np.sum(mutation_mask)) clones.append(clone) return clones def local_search(self, antibody: np.ndarray) -> np.ndarray: best = antibody.copy() best_affinity = self.calculate_affinity(best) for _ in range(10): neighbor = best.copy() i, j = random.sample(range(self.n_tasks), 2) neighbor[i], neighbor[j] = neighbor[j], neighbor[i] affinity = self.calculate_affinity(neighbor) if affinity > best_affinity: best = neighbor best_affinity = affinity return best def optimize(self, generations: int = 100) -> Tuple[Dict, float]: for gen in range(generations): affinities = [self.calculate_affinity(ab) for ab in self.population] sorted_indices = np.argsort(affinities)[::-1] elite = [self.population[i] for i in sorted_indices[:5]] all_clones = [] for i in sorted_indices[:self.pop_size // 2]: clones = self.clone_and_mutate(self.population[i], affinities[i]) all_clones.extend(clones) clone_affinities = [self.calculate_affinity(c) for c in all_clones] best_clones_idx = np.argsort(clone_affinities)[::-1][:self.pop_size - 5] best_clones = [all_clones[i] for i in best_clones_idx] if gen % 10 == 0: best_clones = [self.local_search(c) for c in best_clones[:5]] + best_clones[5:] self.population = elite + best_clones best_idx = np.argmax(affinities) if len(self.memory) < self.memory_size: self.memory.append(self.population[best_idx].copy()) else: mem_affinities = [self.calculate_affinity(m) for m in self.memory] worst_mem_idx = np.argmin(mem_affinities) if affinities[best_idx] > mem_affinities[worst_mem_idx]: self.memory[worst_mem_idx] = self.population[best_idx].copy() all_candidates = self.population + self.memory all_affinities = [self.calculate_affinity(c) for c in all_candidates] best_idx = np.argmax(all_affinities) best_antibody = all_candidates[best_idx] best_schedule = self.scheduler.schedule_with_priority(best_antibody) best_cost = self.scheduler.evaluate_schedule(best_schedule) return best_schedule, best_cost def main(): tasks_p1 = [RDTask("P1_T1", "P1", 5, {"engineer": 2, "lab": 1}, []), RDTask("P1_T2", "P1", 8, {"engineer": 3}, ["P1_T1"]), RDTask("P1_T3", "P1", 6, {"engineer": 2, "lab": 2}, ["P1_T1"]), RDTask("P1_T4", "P1", 4, {"engineer": 4}, ["P1_T2", "P1_T3"])] tasks_p2 = [RDTask("P2_T1", "P2", 6, {"engineer": 2}, []), RDTask("P2_T2", "P2", 5, {"engineer": 2, "lab": 1}, ["P2_T1"]), RDTask("P2_T3", "P2", 7, {"engineer": 3}, ["P2_T2"])] projects = [RDProject("P1", tasks_p1, 1.5, 30, 0.9), RDProject("P2", tasks_p2, 1.0, 25, 0.7)] resource_capacities = {"engineer": 6, "lab": 3} scheduler = ProjectPortfolioScheduler(projects, resource_capacities) print("Running Immune Algorithm...") ia = ImmuneAlgorithm(scheduler, pop_size=40) ia_schedule, ia_cost = ia.optimize(generations=50) print(f"Immune Algorithm Cost: {ia_cost:.2f}") print("Running Branch and Bound...") bb = BranchAndBound(scheduler) bb_schedule, bb_cost = bb.solve(time_limit=10) print(f"Branch and Bound Cost: {bb_cost:.2f}, Nodes Explored: {bb.nodes_explored}") print("\nBest Schedule (Immune Algorithm):") for task_id, (start, finish) in sorted(ia_schedule.items()): print(f" {task_id}: [{start}, {finish}]") if __name__ == "__main__": main()如有问题，可以直接沟通
👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇👇
