Consulting Staff Planning#
Introduction#
In this example, we will explore how to model a consulting staff planning problem using the DecisionAI interface. This problem addresses the planning challenge of assigning consultants with heterogeneous capacities to projects with time-varying demand while respecting various operational constraints.
To try out this base model in the CLI, you can run the following command:
export QUANTAGONIA_API_KEY=<YOUR_API_KEY>
python -m decision_ai.examples.consulting_staff_planning.chat_example
Through this example, you will learn how to define the problem’s inputs, variables, objective, and constraints within the DecisionAI framework, including more advanced rules like lock-in periods and all-or-nothing project fulfillment.
We start with the mathematical definition of the problem and then show how to implement it using the DecisionAI interface.
Mathematical Model Description#
The consulting staff planning model is designed to optimize consultant assignments to projects across multiple time periods. The goal is to minimize unfulfilled demand while respecting consultant capacities, project lifecycles, and operational constraints.
Inputs#
\(C\): The set of consultants.
\(P\): The set of projects.
\(T\): The set of time periods (months) in the planning horizon.
\(D_{p,t}\): Demand for project \(p\) in month \(t\) (person-days).
\(Q_c\): Monthly capacity of consultant \(c\) (working days per month).
\(u_{\max}\): Maximum utilization rate (fraction of capacity that can be allocated).
\(L\): Maximum number of parallel projects per consultant.
\(k\): Minimum working days per assignment.
\(M\): Lock-in window duration (months).
\([s_p, e_p]\): Start and end months for project \(p\).
\(m_{c,p}\): Number of months consultant \(c\) has already worked on project \(p\) before the planning horizon.
Variables#
\(x_{c,p,t} \in \{0, 1, \ldots, Q_c\}\): Integer variable representing the number of days consultant \(c\) works on project \(p\) in month \(t\).
\(z_{c,p,t} \in \{0, 1\}\): Binary variable indicating whether consultant \(c\) is assigned to project \(p\) in month \(t\).
\(y_p \in \{0, 1\}\): Binary variable for all-or-nothing project fulfillment (project \(p\) is either fully staffed or not at all).
Objective#
The objective is to minimize the total demand that cannot be fulfilled by maximizing the number of fully staffed projects:
\[\min \sum_{p \in P} (1 - y_p) \ \sum_{t \in T} D_{p,t}\]
Constraints#
All-or-Nothing Demand Fulfillment: Projects are either fully staffed or not at all:
\[\sum_{c \in C} x_{c,p,t} = D_{p,t} \ y_p \quad \forall p \in P,\ t \in T\]Capacity and Utilization Limits: Consultants cannot exceed their monthly capacity:
\[\sum_{p \in P} x_{c,p,t} \leq u_{\max} \ Q_c \quad \forall c \in C, t \in T\]Assignment-Work Linkage: Consultants can only work if assigned to a project:
\[x_{c,p,t} \leq u_{\max} \ Q_c \ z_{c,p,t} \quad \forall c \in C,\ p \in P,\ t \in T\]Project Lifecycle: No staffing outside of project active periods:
\[z_{c,p,t} = 0 \quad \forall c \in C,\ p \in P,\ t \notin [s_p, e_p]\]Maximum Parallel Projects: Consultants have a limit on concurrent projects:
\[\sum_{p \in P} z_{c,p,t} \leq L \quad \forall c \in C, t \in T\]Minimum Days per Assignment: Consultants must work at least \(k\) days if assigned:
\[x_{c,p,t} \geq k \ z_{c,p,t} \quad \forall c \in C, p \in P, t \in T\]Lock-in Constraints: If consultants are assigned for \(M\) consecutive months, they must continue working for the entire project horizon:
\[z_{c,p,t+M} \geq \sum_{i=0}^{M-1} z_{c,p,t+i} - M + 1 \quad \forall c \in C,\ p \in P,\ t \in T \text{ where } t+M \leq |T|\]Initial Lock-in Constraints: If a consultant has already worked on a project for \(m_{c,p}\) consecutive months before the planning horizon starts:
If \(m_{c,p} \geq M\), the consultant must be assigned to the project for the entire planning horizon (if the project is active):
\[z_{c,p,t} = 1 \quad \forall c \in C,\ p \in P,\ t \in T \text{ where } t \in [s_p, e_p]\]If \(m_{c,p} < M\), the consultants must continue working for the entire project horizon if they have been working for enough months (including pre-horizon months) to trigger the lock-in:
\[z_{c,p,t+1} \geq \left(\sum_{i=1}^{t} z_{c,p,i} + (m_{c,p} - t + 1)\right) - M + 1 \quad \forall c \in C,\ p \in P,\ t \in \{1, \ldots, M-1\} \text{ where } m_{c,p} - t + 1 \geq 1\]
Model Implementation#
In order to implement the model, we first need to define the input data, variables, and constraints.
Input Data#
We start by defining the input data class. Note that the data is stored in flattened structures for easier serialization and handling.
class ConsultantStaffingInput(InputData):
"""Input data for the consultant staffing optimization problem"""
# Consultant data
consultants: list[str] = Field(..., description="List of consultant IDs")
consultant_capacities: dict[str, int] = Field(
..., description="Monthly capacity for each consultant keyed by consultant ID"
)
# Project data
projects: dict[str, dict[str, int]] = Field(
..., description="Projects keyed by project ID with start_month and end_month"
)
# Demand data
demand: dict[str, dict[int, float]] = Field(..., description="Demand keyed by project ID and month")
# Planning parameters
max_utilization_rate: float = Field(
default=0.85,
description="Maximal fraction of consultant capacity that can be used for work on any project",
)
max_parallel_projects_per_consultant: int = Field(default=4, description="Max parallel projects per consultant")
min_working_days_per_assignment: int = Field(
default=1,
description="Minimum days a consultant must work in assigned project",
)
lock_in_window_months: int = Field(default=3, description="Lock-in window in months")
# Initial staffing data
initial_staffing: dict[str, dict[str, int]] = Field(
default_factory=dict,
description="Initial staffing keyed by consultant ID, then project ID, with months already worked as value",
)
# Planning configuration
planning_start_date: str = Field(description="The YYYY-MM start date of the planning horizon")
@property
def project_ids(self) -> list[str]:
"""Get list of project IDs for convenience"""
return list(self.projects.keys())
@property
def horizon(self) -> int:
"""Planning horizon in number of months from planning start date"""
if not self.projects:
return 1 # Default minimum horizon if no projects
# The planning start date is used as the base for month_string_to_int conversion,
# so it always becomes month 1, and project months are relative to this base.
# Therefore: horizon = latest_end_month - 1 + 1 = latest_end_month
return max(project["end_month"] for project in self.projects.values())
@property
def months_in_planning_horizon(self) -> list[int]:
"""Get list of months in the planning horizon (1 to horizon), where 1 corresponds to the planning start date"""
return list(range(1, self.horizon + 1))
def active_months_of_project(self, project_id: str) -> range:
"""Get the active months range for a project"""
project = self.projects[project_id]
return range(project["start_month"], project["end_month"] + 1)
@staticmethod
def _month_string_to_int(month_str: str, base_year: int, base_month: int) -> int:
"""Convert YYYY-MM string to month number relative to base period"""
year, month = map(int, month_str.split("-"))
return (year - base_year) * 12 + (month - base_month) + 1
@classmethod
def from_string_dates(
cls,
consultants: list[str],
consultant_capacities: dict[str, int],
demand: dict[str, dict[str, float]], # month keys as strings
planning_start_date: str,
initial_staffing: dict[str, dict[str, str]] = None, # since_month as strings
max_utilization_rate: float = 0.85,
max_parallel_projects_per_consultant: int = 4,
min_working_days_per_assignment: int = 1,
lock_in_window_months: int = 3,
) -> "ConsultantStaffingInput":
"""Create ConsultantStaffingInput from string date representations.
This method accepts demand and initial_staffing with string dates (YYYY-MM format)
and converts them to the internal integer representation relative to the planning_start_date.
Project start and end months are automatically derived from the demand data.
Args:
consultants: List of consultant IDs
consultant_capacities: Monthly capacity for each consultant
demand: Demand with string month keys, e.g., {"p1": {"2024-01": 15.0, "2024-02": 20.0}}
planning_start_date: Base date in YYYY-MM format (becomes month 1)
initial_staffing: Past staffing with since_month strings, e.g., {"c1": {"p1": "2023-12"}}
max_utilization_rate: Maximum utilization rate
max_parallel_projects_per_consultant: Maximum parallel projects per consultant
min_working_days_per_assignment: Minimum working days per assignment
lock_in_window_months: Lock-in window in months
Returns:
ConsultantStaffingInput instance with converted integer dates
"""
# Parse base year and month from planning start date
base_year, base_month = map(int, planning_start_date.split("-"))
# Convert demand dates from strings to integers and derive project dates
converted_demand = {}
converted_projects = {}
for project_id, project_demand in demand.items():
converted_demand[project_id] = {}
month_ints = []
for month_str, demand_value in project_demand.items():
month_int = cls._month_string_to_int(month_str, base_year, base_month)
converted_demand[project_id][month_int] = demand_value
month_ints.append(month_int)
# Derive project start and end months from demand data
if month_ints:
converted_projects[project_id] = {
"start_month": min(month_ints),
"end_month": max(month_ints),
}
# Convert initial staffing from since_month strings to months_already_worked integers
converted_initial_staffing = {}
if initial_staffing:
for consultant_id, consultant_projects in initial_staffing.items():
converted_initial_staffing[consultant_id] = {}
for project_id, since_month_str in consultant_projects.items():
# Calculate months already worked as difference from planning start
since_month_int = cls._month_string_to_int(since_month_str, base_year, base_month)
months_already_worked = abs(since_month_int - 1) # 1 is the planning start month
converted_initial_staffing[consultant_id][project_id] = months_already_worked
# Create instance with converted data
return cls(
consultants=consultants,
consultant_capacities=consultant_capacities,
projects=converted_projects,
demand=converted_demand,
planning_start_date=planning_start_date,
initial_staffing=converted_initial_staffing,
max_utilization_rate=max_utilization_rate,
max_parallel_projects_per_consultant=max_parallel_projects_per_consultant,
min_working_days_per_assignment=min_working_days_per_assignment,
lock_in_window_months=lock_in_window_months,
)
@staticmethod
def create_sample_input() -> "ConsultantStaffingInput":
"""Create a simple sample input for testing the model"""
consultants = ["c1", "c2", "c3"]
# Sample consultant capacities (different capacities to test individual handling)
consultant_capacities = {
"c1": 20, # full-time
"c2": 15, # part-time
"c3": 18, # near full-time
}
# Demand data - using string dates
# Project timelines are automatically derived from demand data
demand = {
"p1": {"2024-01": 15.0, "2024-02": 20.0, "2024-03": 10.0},
"p2": {"2024-02": 25.0, "2024-03": 15.0, "2024-04": 20.0},
"p3": {"2024-01": 10.0, "2024-02": 15.0},
}
# Initial staffing data - using string dates
initial_staffing = {
"c1": {"p1": "2023-12"} # consultant c1 has been on project p1 since 2023-12
}
return ConsultantStaffingInput.from_string_dates(
consultants=consultants,
consultant_capacities=consultant_capacities,
demand=demand,
initial_staffing=initial_staffing,
max_utilization_rate=0.85,
max_parallel_projects_per_consultant=3,
min_working_days_per_assignment=1,
lock_in_window_months=2,
planning_start_date="2024-01",
)
@staticmethod
def parse_planning_start_date(params_dict: dict) -> str:
"""Parse planning start date from parameters."""
console = Console()
try:
start_date_str = str(params_dict["PlanningStartDate"])
# Validate format
map(int, start_date_str.split("-")) # This will raise ValueError if invalid
except (KeyError, ValueError) as e:
console.print(f"[bold red]❌ Error parsing planning start date: {e}[/bold red]")
console.print("[yellow]💡 Using sample data instead.[/yellow]")
return "2024-01"
else:
return start_date_str
@staticmethod
def create_input_from_excel(filename: str) -> "ConsultantStaffingInput":
"""Create ConsultantStaffingInput object from Excel file data"""
try:
# 1. Read Consultants
consultants_df = pd.read_excel(filename, sheet_name="Consultants")
consultants = consultants_df["ConsultantID"].tolist()
consultant_capacities = dict(
zip(
consultants_df["ConsultantID"],
consultants_df["MonthlyCapacity_Q"],
strict=False,
)
)
# 2. Read Parameters to get planning start date
parameters_df = pd.read_excel(filename, sheet_name="Parameters")
params_dict = dict(zip(parameters_df["Parameter"], parameters_df["Value"], strict=False))
# Get planning start date
start_date_str = ConsultantStaffingInput.parse_planning_start_date(params_dict)
# 3. Read Demand - keep string dates, model will convert them
# Project timelines are automatically derived from demand data
demand_df = pd.read_excel(filename, sheet_name="Demand")
demand = {}
for _, row in demand_df.iterrows():
month_str = str(row["Month"])
project_id = row["ProjectID"]
demand_days = float(row["DemandDays"])
if project_id not in demand:
demand[project_id] = {}
demand[project_id][month_str] = demand_days
# 4. Read Current Staffing - use string dates, model will convert them
try:
current_staffing_df = pd.read_excel(filename, sheet_name="PastStaffing")
initial_staffing = {}
for _, row in current_staffing_df.iterrows():
consultant_id = row["ConsultantID"]
project_id = row["ProjectID"]
since_month = str(row["SinceMonth"])
if consultant_id not in initial_staffing:
initial_staffing[consultant_id] = {}
# Store the since_month string, model will convert to months_worked
initial_staffing[consultant_id][project_id] = since_month
except (KeyError, ValueError):
initial_staffing = {} # Empty if sheet doesn't exist or is empty
return ConsultantStaffingInput.from_string_dates(
consultants=consultants,
consultant_capacities=consultant_capacities,
demand=demand,
initial_staffing=initial_staffing,
max_utilization_rate=float(params_dict["UtilizationTarget"]),
max_parallel_projects_per_consultant=int(params_dict["MaxProjectsPerConsultant"]),
min_working_days_per_assignment=int(params_dict["MinWorkingDaysPerAssignment"]),
lock_in_window_months=int(params_dict["LockInWindow_Months"]),
planning_start_date=start_date_str,
)
except Exception as e:
error_msg = f"Error reading Excel file {filename}: {str(e)}"
raise ValueError(error_msg) from e
Variables#
We then define the variables pydantic model. The model includes different types of variable definitions:
A dictionary of dictionaries for working days (consultant × project × month)
A dictionary of dictionaries for staffing assignments (binary variables)
A dictionary for project fully staffed variables (all-or-nothing)
For each variable, we define a static method init_{variable_name} that returns the initialized variable attribute given the input data.
class ConsultantStaffingVariables(PulpVariables):
"""Decision variables for the consultant staffing optimization problem"""
working_days: dict = Field(
...,
description="Integer variable representing the number of days a consultant is assigned "
"to a project in a given month",
)
staffing_assignment: dict[str, dict[str, dict[int, pulp.LpVariable]]] = Field(
..., description="1 if consultant is staffed on project in month"
)
project_fully_staffed: dict[str, pulp.LpVariable] = Field(
..., description="1 if project is fully staffed (all-or-nothing)"
)
@staticmethod
def init_working_days(input_: ConsultantStaffingInput) -> dict:
"""Initialize working days variables: how many days each consultant works on each project per month"""
working_days = {}
for consultant_id in input_.consultants:
capacity = input_.consultant_capacities[consultant_id]
working_days[consultant_id] = {}
for project_id in input_.project_ids:
working_days[consultant_id][project_id] = {}
for month in input_.months_in_planning_horizon:
working_days[consultant_id][project_id][month] = pulp.LpVariable(
f"working_days_{consultant_id}_{project_id}_month_{month}",
lowBound=0,
upBound=capacity,
cat="Integer",
)
return working_days
@staticmethod
def init_staffing_assignment(input_: ConsultantStaffingInput) -> dict[str, dict[str, dict[int, pulp.LpVariable]]]:
"""Initialize staffing assignment variables: whether consultant is assigned to project in month"""
staffing_assignment = {}
for consultant_id in input_.consultants:
staffing_assignment[consultant_id] = {}
for project_id in input_.project_ids:
staffing_assignment[consultant_id][project_id] = {}
for month in input_.months_in_planning_horizon:
staffing_assignment[consultant_id][project_id][month] = pulp.LpVariable(
f"staffing_assignment_{consultant_id}_{project_id}_month_{month}",
cat="Binary",
)
return staffing_assignment
@staticmethod
def init_project_fully_staffed(input_: ConsultantStaffingInput) -> dict[str, pulp.LpVariable]:
"""Initialize project fully staffed variables: whether each project is fully staffed"""
project_fully_staffed = {}
for project_id in input_.project_ids:
project_fully_staffed[project_id] = pulp.LpVariable(
f"project_fully_staffed_{project_id}",
cat="Binary",
)
return project_fully_staffed
Model Class#
Finally, we define the model class. Note that we attach the variables class to the model class by assigning it to the variables_class attribute.
class ConsultantStaffingOptimizationModel(PulpDecisionAIModel[ConsultantStaffingInput, ConsultantStaffingVariables]):
"""Optimization model for consultant staffing and project assignment"""
variables_class = ConsultantStaffingVariables
@constraint
def consultant_monthly_utilization_limits(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Consultant monthly utilization cannot exceed capacity limits"""
for consultant_id in input_.consultants:
consultant_capacity = input_.consultant_capacities[consultant_id]
for month in input_.months_in_planning_horizon:
# Max utilization constraint: total work across all projects
yield (
pulp.lpSum(
variables.working_days[consultant_id][project_id][month] for project_id in input_.project_ids
)
<= input_.max_utilization_rate * consultant_capacity,
f"consultant_monthly_utilization_limits_{consultant_id}_month_{month}",
)
@constraint
def only_assigned_consultants_work(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Consultants can only work on projects they are assigned to"""
for consultant_id in input_.consultants:
consultant_capacity = input_.consultant_capacities[consultant_id]
for month in input_.months_in_planning_horizon:
# Assignment-work linkage: can only work if assigned to project
for project_id in input_.project_ids:
yield (
variables.working_days[consultant_id][project_id][month]
<= input_.max_utilization_rate
* consultant_capacity
* variables.staffing_assignment[consultant_id][project_id][month],
f"only_assigned_consultants_work_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def project_lifecycle(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""No staffing outside of project lifecycle"""
for project_id in input_.project_ids:
active_months = input_.active_months_of_project(project_id)
for month in input_.months_in_planning_horizon:
if month not in active_months:
for consultant_id in input_.consultants:
yield (
variables.staffing_assignment[consultant_id][project_id][month] == 0,
f"project_lifecycle_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def max_parallel_projects(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Maximum parallel projects per consultant constraint"""
for consultant_id in input_.consultants:
for month in input_.months_in_planning_horizon:
yield (
pulp.lpSum(
variables.staffing_assignment[consultant_id][project_id][month]
for project_id in input_.project_ids
)
<= input_.max_parallel_projects_per_consultant,
f"max_parallel_projects_{consultant_id}_month_{month}",
)
@constraint
def min_days_per_assignment(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Ensure consultants work at least k days if planned on a project,
where k = MinWorkingDaysPerAssigment parameter"""
k = input_.min_working_days_per_assignment
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
yield (
variables.working_days[consultant_id][project_id][month]
>= k * variables.staffing_assignment[consultant_id][project_id][month],
f"min_days_per_assignment_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def all_or_nothing_demand(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""All-or-nothing project staffing: either staff consultants to fulfill
the full demand of the project across all months or none at all"""
for project_id in input_.project_ids:
project = input_.projects[project_id]
# Check if project starts before the planning horizon
starts_before_horizon = project["start_month"] < 1
if starts_before_horizon:
# Set project_fully_staffed=1 for projects that start before the planning horizon
yield (
variables.project_fully_staffed[project_id] == 1,
f"all_or_nothing_existing_project_{project_id}",
)
for month in input_.months_in_planning_horizon:
demand_for_project_in_month = input_.demand.get(project_id, {}).get(month, 0)
if demand_for_project_in_month > 0: # Only apply constraint if there is demand
# Sum of all consultant work equals demand times binary variable
yield (
pulp.lpSum(
variables.working_days[consultant_id][project_id][month]
for consultant_id in input_.consultants
)
== demand_for_project_in_month * variables.project_fully_staffed[project_id],
f"all_or_nothing_demand_{project_id}_month_{month}",
)
@constraint
def no_removal_after_lock_in(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Enforces consultant lock-in: if assigned for consecutive months, must continue in the next month."""
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
# Check if we can look ahead lock_in_window_months from current month
lock_in_end_month = month + input_.lock_in_window_months
# Ensure lock_in_end_month is within project duration (not after end_month)
if lock_in_end_month <= input_.projects[project_id]["end_month"]:
# Sum assignments in the lock-in window [month, month+1, ..., month+window-1]
assignments_in_window = pulp.lpSum(
variables.staffing_assignment[consultant_id][project_id][month + i]
for i in range(input_.lock_in_window_months)
)
# If assigned for all months in window, must continue in the next month
yield (
variables.staffing_assignment[consultant_id][project_id][lock_in_end_month]
>= assignments_in_window - input_.lock_in_window_months + 1,
f"lock_in_{consultant_id}_{project_id}_from_month_{month}",
)
@constraint
def initial_no_removal_after_lock_in(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Enforces consultant lock-in at the start of planning horizon:
if assigned for consecutive months before planning starts,
must continue in the first months of the planning horizon."""
months_in_lock_in_check = list(range(1, min(input_.lock_in_window_months - 1, input_.horizon) + 1))
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
# Check if consultant was assigned to this project before planning horizon
months_already_worked = input_.initial_staffing.get(consultant_id, {}).get(project_id, 0)
# If consultant was assigned to this project before planning horizon
if months_already_worked > 0:
# check if consultant must be assigned for the entire planning horizon
if months_already_worked >= input_.lock_in_window_months:
# Force assignment for all active months of the project
active_months = input_.active_months_of_project(project_id)
for planning_month in input_.months_in_planning_horizon:
if planning_month in active_months:
yield (
variables.staffing_assignment[consultant_id][project_id][planning_month] == 1,
f"initial_lock_in_force_{consultant_id}_{project_id}_month_{planning_month}",
)
else:
# use the months the consultant was already assigned to enforce lock-in
# for the first months of the planning horizon
for month in months_in_lock_in_check:
sum_previous_months = 0
pre_horizon_months = months_already_worked - month + 1
# the months the consultant was already assigned to are relevant
# for lock-in enforcement in this month
if pre_horizon_months > 0:
sum_previous_months += pre_horizon_months
# Add decision variables for months 1 to month within planning horizon
for i in range(1, month + 1):
sum_previous_months += variables.staffing_assignment[consultant_id][project_id][i]
yield (
variables.staffing_assignment[consultant_id][project_id][month + 1]
>= sum_previous_months - input_.lock_in_window_months + 1,
f"initial_lock_in_{consultant_id}_{project_id}_month_{month + 1}",
)
def set_up_objective(
self, input_: ConsultantStaffingInput, prob: pulp.LpProblem, variables: ConsultantStaffingVariables
) -> pulp.LpProblem:
"""Set up the objective function to minimize the total demand that cannot be fulfilled"""
# Calculate total demand for each project
total_demand_per_project = {}
for project_id in input_.project_ids:
total_demand_per_project[project_id] = sum(
input_.demand.get(project_id, {}).get(month, 0) for month in input_.months_in_planning_horizon
)
# Minimize the total demand that cannot be fulfilled
prob += pulp.lpSum(
(1 - variables.project_fully_staffed[project_id]) * total_demand_per_project[project_id]
for project_id in input_.project_ids
)
return prob
def solution_to_str(self, input_: ConsultantStaffingInput, solution: Solution) -> str:
"""Convert solution to human-readable string"""
solution_display = f"Objective value: {solution.objective:.2f}\n\n"
solution_display += f"Status: {solution.status}\n\n"
if solution.status == "Infeasible":
return solution_display
solution_display += "Assignments:\n\n"
# Show basic assignments by consultant
for consultant_id in input_.consultants:
solution_display += f" * {consultant_id}: "
assignments = []
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
working_days = solution.variables.working_days[consultant_id][project_id][month]
if working_days > ZERO_TOLERANCE:
start_date = pd.to_datetime(input_.planning_start_date)
month_date = start_date + pd.DateOffset(months=month - 1)
assignments.append(f"{project_id} ({month_date.strftime('%b %Y')}: {working_days:.1f}d)")
if assignments:
solution_display += ", ".join(assignments) + "\n\n"
else:
solution_display += "No assignments\n\n"
return solution_display
Note
The model includes several helper methods for writing solutions to Excel files and creating staffing matrices.
These methods use the decision_ai.hide_from_ai() decorator to exclude them from the context passed to the LLMs,
as they don’t provide any useful information for them.
Complete Example#
import pandas as pd
import pulp
from pydantic import Field
from rich.console import Console
from decision_ai import InputData, PulpDecisionAIModel, PulpVariables, Solution, constraint
from decision_ai.typing import ConstraintGenerator
ZERO_TOLERANCE = 1e-6
class ConsultantStaffingInput(InputData):
"""Input data for the consultant staffing optimization problem"""
# Consultant data
consultants: list[str] = Field(..., description="List of consultant IDs")
consultant_capacities: dict[str, int] = Field(
..., description="Monthly capacity for each consultant keyed by consultant ID"
)
# Project data
projects: dict[str, dict[str, int]] = Field(
..., description="Projects keyed by project ID with start_month and end_month"
)
# Demand data
demand: dict[str, dict[int, float]] = Field(..., description="Demand keyed by project ID and month")
# Planning parameters
max_utilization_rate: float = Field(
default=0.85,
description="Maximal fraction of consultant capacity that can be used for work on any project",
)
max_parallel_projects_per_consultant: int = Field(default=4, description="Max parallel projects per consultant")
min_working_days_per_assignment: int = Field(
default=1,
description="Minimum days a consultant must work in assigned project",
)
lock_in_window_months: int = Field(default=3, description="Lock-in window in months")
# Initial staffing data
initial_staffing: dict[str, dict[str, int]] = Field(
default_factory=dict,
description="Initial staffing keyed by consultant ID, then project ID, with months already worked as value",
)
# Planning configuration
planning_start_date: str = Field(description="The YYYY-MM start date of the planning horizon")
@property
def project_ids(self) -> list[str]:
"""Get list of project IDs for convenience"""
return list(self.projects.keys())
@property
def horizon(self) -> int:
"""Planning horizon in number of months from planning start date"""
if not self.projects:
return 1 # Default minimum horizon if no projects
# The planning start date is used as the base for month_string_to_int conversion,
# so it always becomes month 1, and project months are relative to this base.
# Therefore: horizon = latest_end_month - 1 + 1 = latest_end_month
return max(project["end_month"] for project in self.projects.values())
@property
def months_in_planning_horizon(self) -> list[int]:
"""Get list of months in the planning horizon (1 to horizon), where 1 corresponds to the planning start date"""
return list(range(1, self.horizon + 1))
def active_months_of_project(self, project_id: str) -> range:
"""Get the active months range for a project"""
project = self.projects[project_id]
return range(project["start_month"], project["end_month"] + 1)
@staticmethod
def _month_string_to_int(month_str: str, base_year: int, base_month: int) -> int:
"""Convert YYYY-MM string to month number relative to base period"""
year, month = map(int, month_str.split("-"))
return (year - base_year) * 12 + (month - base_month) + 1
@classmethod
def from_string_dates(
cls,
consultants: list[str],
consultant_capacities: dict[str, int],
demand: dict[str, dict[str, float]], # month keys as strings
planning_start_date: str,
initial_staffing: dict[str, dict[str, str]] = None, # since_month as strings
max_utilization_rate: float = 0.85,
max_parallel_projects_per_consultant: int = 4,
min_working_days_per_assignment: int = 1,
lock_in_window_months: int = 3,
) -> "ConsultantStaffingInput":
"""Create ConsultantStaffingInput from string date representations.
This method accepts demand and initial_staffing with string dates (YYYY-MM format)
and converts them to the internal integer representation relative to the planning_start_date.
Project start and end months are automatically derived from the demand data.
Args:
consultants: List of consultant IDs
consultant_capacities: Monthly capacity for each consultant
demand: Demand with string month keys, e.g., {"p1": {"2024-01": 15.0, "2024-02": 20.0}}
planning_start_date: Base date in YYYY-MM format (becomes month 1)
initial_staffing: Past staffing with since_month strings, e.g., {"c1": {"p1": "2023-12"}}
max_utilization_rate: Maximum utilization rate
max_parallel_projects_per_consultant: Maximum parallel projects per consultant
min_working_days_per_assignment: Minimum working days per assignment
lock_in_window_months: Lock-in window in months
Returns:
ConsultantStaffingInput instance with converted integer dates
"""
# Parse base year and month from planning start date
base_year, base_month = map(int, planning_start_date.split("-"))
# Convert demand dates from strings to integers and derive project dates
converted_demand = {}
converted_projects = {}
for project_id, project_demand in demand.items():
converted_demand[project_id] = {}
month_ints = []
for month_str, demand_value in project_demand.items():
month_int = cls._month_string_to_int(month_str, base_year, base_month)
converted_demand[project_id][month_int] = demand_value
month_ints.append(month_int)
# Derive project start and end months from demand data
if month_ints:
converted_projects[project_id] = {
"start_month": min(month_ints),
"end_month": max(month_ints),
}
# Convert initial staffing from since_month strings to months_already_worked integers
converted_initial_staffing = {}
if initial_staffing:
for consultant_id, consultant_projects in initial_staffing.items():
converted_initial_staffing[consultant_id] = {}
for project_id, since_month_str in consultant_projects.items():
# Calculate months already worked as difference from planning start
since_month_int = cls._month_string_to_int(since_month_str, base_year, base_month)
months_already_worked = abs(since_month_int - 1) # 1 is the planning start month
converted_initial_staffing[consultant_id][project_id] = months_already_worked
# Create instance with converted data
return cls(
consultants=consultants,
consultant_capacities=consultant_capacities,
projects=converted_projects,
demand=converted_demand,
planning_start_date=planning_start_date,
initial_staffing=converted_initial_staffing,
max_utilization_rate=max_utilization_rate,
max_parallel_projects_per_consultant=max_parallel_projects_per_consultant,
min_working_days_per_assignment=min_working_days_per_assignment,
lock_in_window_months=lock_in_window_months,
)
@staticmethod
def create_sample_input() -> "ConsultantStaffingInput":
"""Create a simple sample input for testing the model"""
consultants = ["c1", "c2", "c3"]
# Sample consultant capacities (different capacities to test individual handling)
consultant_capacities = {
"c1": 20, # full-time
"c2": 15, # part-time
"c3": 18, # near full-time
}
# Demand data - using string dates
# Project timelines are automatically derived from demand data
demand = {
"p1": {"2024-01": 15.0, "2024-02": 20.0, "2024-03": 10.0},
"p2": {"2024-02": 25.0, "2024-03": 15.0, "2024-04": 20.0},
"p3": {"2024-01": 10.0, "2024-02": 15.0},
}
# Initial staffing data - using string dates
initial_staffing = {
"c1": {"p1": "2023-12"} # consultant c1 has been on project p1 since 2023-12
}
return ConsultantStaffingInput.from_string_dates(
consultants=consultants,
consultant_capacities=consultant_capacities,
demand=demand,
initial_staffing=initial_staffing,
max_utilization_rate=0.85,
max_parallel_projects_per_consultant=3,
min_working_days_per_assignment=1,
lock_in_window_months=2,
planning_start_date="2024-01",
)
@staticmethod
def parse_planning_start_date(params_dict: dict) -> str:
"""Parse planning start date from parameters."""
console = Console()
try:
start_date_str = str(params_dict["PlanningStartDate"])
# Validate format
map(int, start_date_str.split("-")) # This will raise ValueError if invalid
except (KeyError, ValueError) as e:
console.print(f"[bold red]❌ Error parsing planning start date: {e}[/bold red]")
console.print("[yellow]💡 Using sample data instead.[/yellow]")
return "2024-01"
else:
return start_date_str
@staticmethod
def create_input_from_excel(filename: str) -> "ConsultantStaffingInput":
"""Create ConsultantStaffingInput object from Excel file data"""
try:
# 1. Read Consultants
consultants_df = pd.read_excel(filename, sheet_name="Consultants")
consultants = consultants_df["ConsultantID"].tolist()
consultant_capacities = dict(
zip(
consultants_df["ConsultantID"],
consultants_df["MonthlyCapacity_Q"],
strict=False,
)
)
# 2. Read Parameters to get planning start date
parameters_df = pd.read_excel(filename, sheet_name="Parameters")
params_dict = dict(zip(parameters_df["Parameter"], parameters_df["Value"], strict=False))
# Get planning start date
start_date_str = ConsultantStaffingInput.parse_planning_start_date(params_dict)
# 3. Read Demand - keep string dates, model will convert them
# Project timelines are automatically derived from demand data
demand_df = pd.read_excel(filename, sheet_name="Demand")
demand = {}
for _, row in demand_df.iterrows():
month_str = str(row["Month"])
project_id = row["ProjectID"]
demand_days = float(row["DemandDays"])
if project_id not in demand:
demand[project_id] = {}
demand[project_id][month_str] = demand_days
# 4. Read Current Staffing - use string dates, model will convert them
try:
current_staffing_df = pd.read_excel(filename, sheet_name="PastStaffing")
initial_staffing = {}
for _, row in current_staffing_df.iterrows():
consultant_id = row["ConsultantID"]
project_id = row["ProjectID"]
since_month = str(row["SinceMonth"])
if consultant_id not in initial_staffing:
initial_staffing[consultant_id] = {}
# Store the since_month string, model will convert to months_worked
initial_staffing[consultant_id][project_id] = since_month
except (KeyError, ValueError):
initial_staffing = {} # Empty if sheet doesn't exist or is empty
return ConsultantStaffingInput.from_string_dates(
consultants=consultants,
consultant_capacities=consultant_capacities,
demand=demand,
initial_staffing=initial_staffing,
max_utilization_rate=float(params_dict["UtilizationTarget"]),
max_parallel_projects_per_consultant=int(params_dict["MaxProjectsPerConsultant"]),
min_working_days_per_assignment=int(params_dict["MinWorkingDaysPerAssignment"]),
lock_in_window_months=int(params_dict["LockInWindow_Months"]),
planning_start_date=start_date_str,
)
except Exception as e:
error_msg = f"Error reading Excel file {filename}: {str(e)}"
raise ValueError(error_msg) from e
class ConsultantStaffingVariables(PulpVariables):
"""Decision variables for the consultant staffing optimization problem"""
working_days: dict = Field(
...,
description="Integer variable representing the number of days a consultant is assigned "
"to a project in a given month",
)
staffing_assignment: dict[str, dict[str, dict[int, pulp.LpVariable]]] = Field(
..., description="1 if consultant is staffed on project in month"
)
project_fully_staffed: dict[str, pulp.LpVariable] = Field(
..., description="1 if project is fully staffed (all-or-nothing)"
)
@staticmethod
def init_working_days(input_: ConsultantStaffingInput) -> dict:
"""Initialize working days variables: how many days each consultant works on each project per month"""
working_days = {}
for consultant_id in input_.consultants:
capacity = input_.consultant_capacities[consultant_id]
working_days[consultant_id] = {}
for project_id in input_.project_ids:
working_days[consultant_id][project_id] = {}
for month in input_.months_in_planning_horizon:
working_days[consultant_id][project_id][month] = pulp.LpVariable(
f"working_days_{consultant_id}_{project_id}_month_{month}",
lowBound=0,
upBound=capacity,
cat="Integer",
)
return working_days
@staticmethod
def init_staffing_assignment(input_: ConsultantStaffingInput) -> dict[str, dict[str, dict[int, pulp.LpVariable]]]:
"""Initialize staffing assignment variables: whether consultant is assigned to project in month"""
staffing_assignment = {}
for consultant_id in input_.consultants:
staffing_assignment[consultant_id] = {}
for project_id in input_.project_ids:
staffing_assignment[consultant_id][project_id] = {}
for month in input_.months_in_planning_horizon:
staffing_assignment[consultant_id][project_id][month] = pulp.LpVariable(
f"staffing_assignment_{consultant_id}_{project_id}_month_{month}",
cat="Binary",
)
return staffing_assignment
@staticmethod
def init_project_fully_staffed(input_: ConsultantStaffingInput) -> dict[str, pulp.LpVariable]:
"""Initialize project fully staffed variables: whether each project is fully staffed"""
project_fully_staffed = {}
for project_id in input_.project_ids:
project_fully_staffed[project_id] = pulp.LpVariable(
f"project_fully_staffed_{project_id}",
cat="Binary",
)
return project_fully_staffed
class ConsultantStaffingOptimizationModel(PulpDecisionAIModel[ConsultantStaffingInput, ConsultantStaffingVariables]):
"""Optimization model for consultant staffing and project assignment"""
variables_class = ConsultantStaffingVariables
@constraint
def consultant_monthly_utilization_limits(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Consultant monthly utilization cannot exceed capacity limits"""
for consultant_id in input_.consultants:
consultant_capacity = input_.consultant_capacities[consultant_id]
for month in input_.months_in_planning_horizon:
# Max utilization constraint: total work across all projects
yield (
pulp.lpSum(
variables.working_days[consultant_id][project_id][month] for project_id in input_.project_ids
)
<= input_.max_utilization_rate * consultant_capacity,
f"consultant_monthly_utilization_limits_{consultant_id}_month_{month}",
)
@constraint
def only_assigned_consultants_work(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Consultants can only work on projects they are assigned to"""
for consultant_id in input_.consultants:
consultant_capacity = input_.consultant_capacities[consultant_id]
for month in input_.months_in_planning_horizon:
# Assignment-work linkage: can only work if assigned to project
for project_id in input_.project_ids:
yield (
variables.working_days[consultant_id][project_id][month]
<= input_.max_utilization_rate
* consultant_capacity
* variables.staffing_assignment[consultant_id][project_id][month],
f"only_assigned_consultants_work_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def project_lifecycle(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""No staffing outside of project lifecycle"""
for project_id in input_.project_ids:
active_months = input_.active_months_of_project(project_id)
for month in input_.months_in_planning_horizon:
if month not in active_months:
for consultant_id in input_.consultants:
yield (
variables.staffing_assignment[consultant_id][project_id][month] == 0,
f"project_lifecycle_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def max_parallel_projects(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Maximum parallel projects per consultant constraint"""
for consultant_id in input_.consultants:
for month in input_.months_in_planning_horizon:
yield (
pulp.lpSum(
variables.staffing_assignment[consultant_id][project_id][month]
for project_id in input_.project_ids
)
<= input_.max_parallel_projects_per_consultant,
f"max_parallel_projects_{consultant_id}_month_{month}",
)
@constraint
def min_days_per_assignment(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Ensure consultants work at least k days if planned on a project,
where k = MinWorkingDaysPerAssigment parameter"""
k = input_.min_working_days_per_assignment
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
yield (
variables.working_days[consultant_id][project_id][month]
>= k * variables.staffing_assignment[consultant_id][project_id][month],
f"min_days_per_assignment_{consultant_id}_{project_id}_month_{month}",
)
@constraint
def all_or_nothing_demand(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""All-or-nothing project staffing: either staff consultants to fulfill
the full demand of the project across all months or none at all"""
for project_id in input_.project_ids:
project = input_.projects[project_id]
# Check if project starts before the planning horizon
starts_before_horizon = project["start_month"] < 1
if starts_before_horizon:
# Set project_fully_staffed=1 for projects that start before the planning horizon
yield (
variables.project_fully_staffed[project_id] == 1,
f"all_or_nothing_existing_project_{project_id}",
)
for month in input_.months_in_planning_horizon:
demand_for_project_in_month = input_.demand.get(project_id, {}).get(month, 0)
if demand_for_project_in_month > 0: # Only apply constraint if there is demand
# Sum of all consultant work equals demand times binary variable
yield (
pulp.lpSum(
variables.working_days[consultant_id][project_id][month]
for consultant_id in input_.consultants
)
== demand_for_project_in_month * variables.project_fully_staffed[project_id],
f"all_or_nothing_demand_{project_id}_month_{month}",
)
@constraint
def no_removal_after_lock_in(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Enforces consultant lock-in: if assigned for consecutive months, must continue in the next month."""
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
# Check if we can look ahead lock_in_window_months from current month
lock_in_end_month = month + input_.lock_in_window_months
# Ensure lock_in_end_month is within project duration (not after end_month)
if lock_in_end_month <= input_.projects[project_id]["end_month"]:
# Sum assignments in the lock-in window [month, month+1, ..., month+window-1]
assignments_in_window = pulp.lpSum(
variables.staffing_assignment[consultant_id][project_id][month + i]
for i in range(input_.lock_in_window_months)
)
# If assigned for all months in window, must continue in the next month
yield (
variables.staffing_assignment[consultant_id][project_id][lock_in_end_month]
>= assignments_in_window - input_.lock_in_window_months + 1,
f"lock_in_{consultant_id}_{project_id}_from_month_{month}",
)
@constraint
def initial_no_removal_after_lock_in(
input_: ConsultantStaffingInput, # noqa: N805
variables: ConsultantStaffingVariables,
) -> ConstraintGenerator:
"""Enforces consultant lock-in at the start of planning horizon:
if assigned for consecutive months before planning starts,
must continue in the first months of the planning horizon."""
months_in_lock_in_check = list(range(1, min(input_.lock_in_window_months - 1, input_.horizon) + 1))
for consultant_id in input_.consultants:
for project_id in input_.project_ids:
# Check if consultant was assigned to this project before planning horizon
months_already_worked = input_.initial_staffing.get(consultant_id, {}).get(project_id, 0)
# If consultant was assigned to this project before planning horizon
if months_already_worked > 0:
# check if consultant must be assigned for the entire planning horizon
if months_already_worked >= input_.lock_in_window_months:
# Force assignment for all active months of the project
active_months = input_.active_months_of_project(project_id)
for planning_month in input_.months_in_planning_horizon:
if planning_month in active_months:
yield (
variables.staffing_assignment[consultant_id][project_id][planning_month] == 1,
f"initial_lock_in_force_{consultant_id}_{project_id}_month_{planning_month}",
)
else:
# use the months the consultant was already assigned to enforce lock-in
# for the first months of the planning horizon
for month in months_in_lock_in_check:
sum_previous_months = 0
pre_horizon_months = months_already_worked - month + 1
# the months the consultant was already assigned to are relevant
# for lock-in enforcement in this month
if pre_horizon_months > 0:
sum_previous_months += pre_horizon_months
# Add decision variables for months 1 to month within planning horizon
for i in range(1, month + 1):
sum_previous_months += variables.staffing_assignment[consultant_id][project_id][i]
yield (
variables.staffing_assignment[consultant_id][project_id][month + 1]
>= sum_previous_months - input_.lock_in_window_months + 1,
f"initial_lock_in_{consultant_id}_{project_id}_month_{month + 1}",
)
def set_up_objective(
self, input_: ConsultantStaffingInput, prob: pulp.LpProblem, variables: ConsultantStaffingVariables
) -> pulp.LpProblem:
"""Set up the objective function to minimize the total demand that cannot be fulfilled"""
# Calculate total demand for each project
total_demand_per_project = {}
for project_id in input_.project_ids:
total_demand_per_project[project_id] = sum(
input_.demand.get(project_id, {}).get(month, 0) for month in input_.months_in_planning_horizon
)
# Minimize the total demand that cannot be fulfilled
prob += pulp.lpSum(
(1 - variables.project_fully_staffed[project_id]) * total_demand_per_project[project_id]
for project_id in input_.project_ids
)
return prob
def solution_to_str(self, input_: ConsultantStaffingInput, solution: Solution) -> str:
"""Convert solution to human-readable string"""
solution_display = f"Objective value: {solution.objective:.2f}\n\n"
solution_display += f"Status: {solution.status}\n\n"
if solution.status == "Infeasible":
return solution_display
solution_display += "Assignments:\n\n"
# Show basic assignments by consultant
for consultant_id in input_.consultants:
solution_display += f" * {consultant_id}: "
assignments = []
for project_id in input_.project_ids:
for month in input_.months_in_planning_horizon:
working_days = solution.variables.working_days[consultant_id][project_id][month]
if working_days > ZERO_TOLERANCE:
start_date = pd.to_datetime(input_.planning_start_date)
month_date = start_date + pd.DateOffset(months=month - 1)
assignments.append(f"{project_id} ({month_date.strftime('%b %Y')}: {working_days:.1f}d)")
if assignments:
solution_display += ", ".join(assignments) + "\n\n"
else:
solution_display += "No assignments\n\n"
return solution_display