diff --git a/apps/web/src/api/types.ts b/apps/web/src/api/types.ts index ee768e1..de82a49 100644 --- a/apps/web/src/api/types.ts +++ b/apps/web/src/api/types.ts @@ -197,7 +197,7 @@ export interface IntentResult { source: 'RULE_FAST' | 'LLM'; // 产生来源 } -export type ScheduleEngineType = 'RULE' | 'CP' | 'GA' | 'HYBRID' | 'EXTERNAL'; +export type ScheduleEngineType = 'RULE' | 'CP' | 'GA' | 'HYBRID' | 'EXTERNAL' | 'OPTIMIZE'; export interface ScheduleResult { versionId: number; diff --git a/docs/round-85-optimize-integration-shape-plan.md b/docs/round-85-optimize-integration-shape-plan.md index eb44d82..5b93d4a 100644 --- a/docs/round-85-optimize-integration-shape-plan.md +++ b/docs/round-85-optimize-integration-shape-plan.md @@ -183,6 +183,15 @@ optimize 只能产生候选排产解。Harness、Workflow、WorldStore、版本 PLAN AUDIT: PASS +## 14. 本轮实施结果 + +- 已新增 `server/engines/optimize_engine.py`，并通过 `get_engine("OPTIMIZE")` 接入 APS。 +- 七种派工规则（EDD/SPT/PRIORITY/FIFO/LPT/CR/ATC）共用 APS 的能力池、日历、班组、工装和物化逻辑；结果统一进入 `SchedulingSolutionV2` 校验。 +- `/api/flex/schedule` 增加 `engine=OPTIMIZE`；Optimize 版本记录 `solverId`、`solverVersion`、`algorithmId`、`algorithmVersion`，并保留 V2 provenance 和 adapter assumption。 +- 准入阻断时仍由 APS 记录零工单 DRAFT 版本，同时保留请求的引擎身份和 blocker，不绕过 admission。 +- 新增 `tests/golden/test_optimize_engine.py`；Optimize/V2/算法注册表相关定向测试共 20 项通过，相关 APS 回归共 34 项通过。 +- 当前 APS 轮次工作树未包含 `server/data/world-proj_712276ba.json`，因此真实 MOM world 用例只能按既有测试策略跳过；全量 CP/Excel 测试还受到当前环境 NumPy（X86_V2）二进制不兼容影响。CP-SAT 的 V2 原生求解器仍应作为后续轮次接入，本轮不把 PoolEngine 适配器冒充为 CP-SAT 最优证明。 + 阻塞问题：无。 检查结论：目标、写入范围、验收标准、验证命令、真实 MOM 门禁和停止条件均已明确；单 agent 任务没有并行写入冲突，也没有依赖未讨论的产品决策。 @@ -197,4 +206,3 @@ PLAN AUDIT: PASS - 用户确认后才能进入目标模式实现；当前不修改生产代码。 - 实现完成后必须通过自身检查；如形成集成结果，补充统一集成审计和真实 MOM 验证。 - 轮次结束时保留计划、验证证据和未合并分支，直到用户明确决定是否合并。 - diff --git a/server/agent_core/algolib.py b/server/agent_core/algolib.py index 75f6a87..4bcf01d 100644 --- a/server/agent_core/algolib.py +++ b/server/agent_core/algolib.py @@ -120,6 +120,63 @@ def _builtin_catalog() -> list[AlgorithmManifest]: "constraints": "dict", } items = [ + # ---- A. Optimize Python-native dispatch rules ---- + AlgorithmManifest( + algo_id="optimize.edd", name="Optimize EDD 最早交期", + category="A", description="Optimize V2 适配器：按最早交期派工", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:EDD", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.spt", name="Optimize SPT 最短工时", + category="A", description="Optimize V2 适配器：短工时优先", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:SPT", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.priority", name="Optimize PRIORITY 优先级", + category="A", description="Optimize V2 适配器：订单优先级优先", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:PRIORITY", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.fifo", name="Optimize FIFO 先来先服务", + category="A", description="Optimize V2 适配器：按释放时间派工", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:FIFO", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.lpt", name="Optimize LPT 最长工时", + category="A", description="Optimize V2 适配器：长工时优先", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:LPT", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.cr", name="Optimize CR 临界比", + category="A", description="Optimize V2 适配器：交期紧迫度优先", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:CR", regen_strategy="manual", + ), + AlgorithmManifest( + algo_id="optimize.atc", name="Optimize ATC 逾期成本", + category="A", description="Optimize V2 适配器：逾期成本代理排序", + scale_limit="<=50k 工单", time_budget="毫秒级", + input_schema=rule_in, output_schema=kpi_out, + golden_tests=["tests/golden/test_optimize_engine.py"], deterministic=True, + entrypoint="OPTIMIZE:ATC", regen_strategy="manual", + ), # ---- A. 启发式（RULE 引擎各策略模板）---- AlgorithmManifest( algo_id="rule.delivery_first", name="EDD 最早交期", @@ -436,10 +493,10 @@ class AlgorithmRegistry: """入口可达性：引擎名 / ENGINE:STRATEGY / module.path:attr。""" if ":" not in entrypoint: return True, "" # 纯引擎名（RULE/CP/GA/HYBRID）由 get_engine 工厂保证 - if entrypoint.startswith("RULE:"): + if entrypoint.startswith(("RULE:", "OPTIMIZE:")): from server.engines import get_engine try: - get_engine("RULE") + get_engine(entrypoint.split(":", 1)[0]) return True, "" except (ImportError, AttributeError, RuntimeError, ValueError, TypeError) as exc: return False, str(exc) diff --git a/server/aps_domain/closed_loop_runtime.py b/server/aps_domain/closed_loop_runtime.py index 7e43349..daf95dd 100644 --- a/server/aps_domain/closed_loop_runtime.py +++ b/server/aps_domain/closed_loop_runtime.py @@ -43,6 +43,7 @@ from server.aps_domain.scheduling_problem_v2 import ( ) from server.aps_domain.scheduling_validator import validate_solution from server.engines.pool_engine import PoolEngine +from server.engines.optimize_engine import OptimizeEngine World = dict[str, Any] _TZ = ZoneInfo("Asia/Shanghai") @@ -624,7 +625,13 @@ def persist_closed_loop_projection(world: World, closed_loop: ClosedLoopProblem, } -def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopProblem) -> dict[str, Any]: +def record_blocked_flex_version( + world: World, + next_id, + closed_loop: ClosedLoopProblem, + *, + engine_type: str = "CLOSED_LOOP", +) -> dict[str, Any]: """Record an honest zero-WO DRAFT version when manufacturing admission is blocked.""" world.setdefault("flexScheduleVersions", []) @@ -632,13 +639,18 @@ def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopPr world.setdefault("flexWorkOrders", []) world.setdefault("flexConflicts", []) version_id = next_id("flexScheduleVersion") + normalized_engine = ( + "OPTIMIZE" + if str(engine_type or "CLOSED_LOOP").strip().upper() == "OPTIMIZE" + else "CLOSED_LOOP" + ) version_no = f"FV{closed_loop.business_date.replace('-', '')}-{len(world['flexScheduleVersions']) + 1:03d}" version = { "id": version_id, "versionNo": version_no, "versionName": f"闭环排产准入阻断 {closed_loop.business_date}", "sortMode": "CLOSED_LOOP", - "engineType": "CLOSED_LOOP", + "engineType": normalized_engine, "status": "DRAFT", "solveStatus": "BLOCKED", "planningProblemId": closed_loop.problem_id, @@ -682,7 +694,7 @@ def record_blocked_flex_version(world: World, next_id, closed_loop: ClosedLoopPr return { "versionId": version_id, "versionNo": version_no, - "engineType": "CLOSED_LOOP", + "engineType": normalized_engine, "status": "DRAFT", "solveStatus": "BLOCKED", "orderCount": version["orderCount"], @@ -1003,15 +1015,19 @@ def flex_version_to_solution_v2( unscheduledRequirements=unscheduled, hardViolations=hard_conflicts, assumptions=(Assumption( - code="POOL_ENGINE_V1_ADAPTER", - message="PoolEngine 候选结果已映射到闭环 V2 契约并执行独立校验", + code=("OPTIMIZE_ENGINE_V1_ADAPTER" + if str(version.get("engineType") or "").upper() == "OPTIMIZE" + else "POOL_ENGINE_V1_ADAPTER"), + message=("OptimizeEngine 候选结果已映射到闭环 V2 契约并执行独立校验" + if str(version.get("engineType") or "").upper() == "OPTIMIZE" + else "PoolEngine 候选结果已映射到闭环 V2 契约并执行独立校验"), sourceRef=f"flex-version:{version_id}", confidence=1.0, ),), provenance=SolutionProvenance( runId=f"flex-version:{version_id}", - solverId="pool-engine", - solverVersion="closed-loop-v1", + solverId=str(version.get("solverId") or "pool-engine"), + solverVersion=str(version.get("solverVersion") or "closed-loop-v1"), generatedAt=generated_at, businessDate=business_day, problemHash=scheduling_problem_hash(problem), @@ -1104,6 +1120,7 @@ def run_closed_loop_candidate( sort_mode: str | None = None, window: str | None = None, name: str | None = None, + engine_type: str = "CLOSED_LOOP", strict: bool = True, ) -> dict[str, Any]: """Build, solve and validate one closed-loop candidate with version-level atomicity.""" @@ -1134,7 +1151,12 @@ def run_closed_loop_candidate( } } if not admitted: - result = record_blocked_flex_version(world, next_id, closed_loop) + result = record_blocked_flex_version( + world, + next_id, + closed_loop, + engine_type=engine_type or "CLOSED_LOOP", + ) return {**result, **base, "validation": None} candidate = deepcopy(world) @@ -1146,20 +1168,40 @@ def run_closed_loop_candidate( candidate.setdefault(key, [] if key != "flexParams" else {}) projected = project_admitted_demands_to_flex_orders(candidate, closed_loop) schedule_start = schedule_start_date or (_as_day(business_date) + timedelta(days=1)).isoformat() - solved = PoolEngine().solve( - candidate, - next_id, - sort_mode=sort_mode, - order_ids=projected["orderIds"], - start_date=schedule_start, - name=name or f"闭环排产 {business_date}", - window=window, + normalized_engine = ( + "OPTIMIZE" + if str(engine_type or "CLOSED_LOOP").strip().upper() == "OPTIMIZE" + else "CLOSED_LOOP" ) + if normalized_engine == "OPTIMIZE": + solved = OptimizeEngine().solve_flex( + candidate, + next_id, + dispatch_rule=sort_mode, + order_ids=projected["orderIds"], + start_date=schedule_start, + name=name or f"Optimize 闭环排产 {business_date}", + window=window, + ) + else: + solved = PoolEngine().solve( + candidate, + next_id, + sort_mode=sort_mode, + order_ids=projected["orderIds"], + start_date=schedule_start, + name=name or f"闭环排产 {business_date}", + window=window, + ) version_id = int(solved["versionId"]) version = next(row for row in candidate["flexScheduleVersions"] if row.get("id") == version_id) version["versionNo"] = f"FV{business_date.replace('-', '')}-{len(candidate['flexScheduleVersions']):03d}" version["versionName"] = name or f"闭环排产 {business_date}" - version["engineType"] = "CLOSED_LOOP" + version["engineType"] = normalized_engine + version["solverId"] = str(solved.get("solverId") or ("pool-engine" if normalized_engine != "OPTIMIZE" else "optimize-dispatch")) + version["solverVersion"] = str(solved.get("solverVersion") or ("closed-loop-v1" if normalized_engine != "OPTIMIZE" else "1.0.0")) + version["algorithmId"] = solved.get("algorithmId") + version["algorithmVersion"] = solved.get("algorithmVersion") version["planningProblemId"] = closed_loop.problem_id version["planningSourceHash"] = closed_loop.source_revision version["demandCount"] = len(closed_loop.manufacturing_demands) @@ -1167,7 +1209,7 @@ def run_closed_loop_candidate( version["unscheduledDemandCount"] = max(0, len(admitted) - int(version.get("vlCount") or 0)) version["createdAt"] = f"{business_date} 00:00" solved["versionNo"] = version["versionNo"] - solved["engineType"] = "CLOSED_LOOP" + solved["engineType"] = normalized_engine solution = flex_version_to_solution_v2(candidate, closed_loop, problem, version_id) report = validate_solution(problem, solution, world=candidate) diff --git a/server/aps_domain/flex.py b/server/aps_domain/flex.py index e84dd8f..8bb1d8a 100644 --- a/server/aps_domain/flex.py +++ b/server/aps_domain/flex.py @@ -233,7 +233,8 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int] start_date: str | None = None, name: str | None = None, actor: str = "web", window: str | None = None, enforce_teams: bool | None = None, - trial: bool = False) -> dict: + trial: bool = False, + engine_type: str | None = None) -> dict: """Run the governed closed-loop scheduling pipeline as one P1 action. Real/site worlds always use the closed-loop requirement, supply, routing and @@ -300,9 +301,14 @@ def run_flex_schedule(store, sort_mode: str | None = None, order_ids: list[int] sort_mode=sort_mode, window=window, name=name, + engine_type=engine_type or "CLOSED_LOOP", strict=True, ) - result["executionMode"] = "CLOSED_LOOP_V1" + result["executionMode"] = ( + "OPTIMIZE_CLOSED_LOOP_V1" + if str(engine_type or "").upper() == "OPTIMIZE" + else "CLOSED_LOOP_V1" + ) result["salesOrdersSynced"] = synced result["decompose"] = { "orders": len(decomposition.get("orders") or []), diff --git a/server/aps_domain/params.py b/server/aps_domain/params.py index 9b0486b..f348e9f 100644 --- a/server/aps_domain/params.py +++ b/server/aps_domain/params.py @@ -96,8 +96,8 @@ def normalize_params_payload(payload: dict[str, Any]) -> dict[str, Any]: if "defaultEngine" in payload and payload["defaultEngine"] is not None: eng = str(payload["defaultEngine"]).upper() - if eng not in ("RULE", "CP", "GA", "HYBRID"): - raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID") + if eng not in ("RULE", "CP", "GA", "HYBRID", "OPTIMIZE"): + raise ValueError("defaultEngine 须为 RULE/CP/GA/HYBRID/OPTIMIZE") out["defaultEngine"] = eng if "cpTimeLimitSeconds" in payload and payload["cpTimeLimitSeconds"] is not None: diff --git a/server/aps_domain/workflow.py b/server/aps_domain/workflow.py index 7bc86d4..c5827c9 100644 --- a/server/aps_domain/workflow.py +++ b/server/aps_domain/workflow.py @@ -151,6 +151,10 @@ def _run_schedule(store: WorldStore, intent: IntentResult, actor: str) -> AgentR constraints=engine_constraint_flags(store.data), # SC-04 约束剖面 → 引擎开关 timeLimitSeconds=float(sp["cpTimeLimitSeconds"]) if sp.get("cpTimeLimitSeconds") is not None else 8.0, ) + if params.engineType == "OPTIMIZE": + return AgentReply( + text="Optimize 目前只支持柔性 V2 闭环，请通过 /api/flex/schedule 并指定 engine=OPTIMIZE。" + ) engine = get_engine(params.engineType) # CP / HYBRID 真管线；GA 仍 RULE 代跑 result: ScheduleResult = engine.solve(store.data, params, store.next_id) # 求解（写内存世界） # 可追溯链：run-id / 算法版本 / 种子 / 知识版本 / 用户确认 串成一条链（§8.4） diff --git a/server/contracts.py b/server/contracts.py index 58dd8c7..3da3329 100644 --- a/server/contracts.py +++ b/server/contracts.py @@ -125,7 +125,7 @@ class ScheduleResult(BaseModel): """排产结果摘要：引擎 solve() 的标准输出（§9.1），回复/审计/KPI 共用。""" versionId: int # 版本 ID versionNo: str # 版本号（V+日期+序号） - engineType: Literal["RULE", "CP", "GA", "HYBRID", "EXTERNAL"] # 引擎类型 + engineType: Literal["RULE", "CP", "GA", "HYBRID", "EXTERNAL", "OPTIMIZE"] # 引擎类型 strategy: str # 策略模板 status: Literal["DRAFT", "PUBLISHED", "ARCHIVED"] = "DRAFT" # 版本状态 orderCount: int # 参与排产的订单项数 diff --git a/server/engines/__init__.py b/server/engines/__init__.py index 6993c1a..b914fbe 100644 --- a/server/engines/__init__.py +++ b/server/engines/__init__.py @@ -8,6 +8,7 @@ from server.engines.cp_engine import CpSatEngine, HybridEngine from server.engines.external_engine import ExternalEngine from server.engines.ga_engine import GeneticAlgorithmEngine from server.engines.nsga2_engine import NSGA2Engine, nsga2_defaults, solve_nsga2 +from server.engines.optimize_engine import OptimizeEngine from server.engines.pool_engine import PoolEngine from server.engines.rule_engine import RuleEngine @@ -19,6 +20,7 @@ __all__ = [ "HybridEngine", "ISchedulingEngine", "NSGA2Engine", + "OptimizeEngine", "PoolEngine", "RuleEngine", "get_engine", @@ -28,7 +30,7 @@ __all__ = [ def get_engine(engine_type: str) -> ISchedulingEngine: - """引擎工厂：RULE / CP / GA / HYBRID / NSGA2 / EXTERNAL。""" + """引擎工厂：RULE / CP / GA / HYBRID / NSGA2 / OPTIMIZE / EXTERNAL。""" kind = (engine_type or "RULE").upper() if kind.startswith("EXTERNAL"): skill_id = None @@ -43,4 +45,6 @@ def get_engine(engine_type: str) -> ISchedulingEngine: return GeneticAlgorithmEngine() if kind == "NSGA2": return NSGA2Engine() + if kind == "OPTIMIZE": + return OptimizeEngine() return RuleEngine(requested_type="RULE") diff --git a/server/engines/base.py b/server/engines/base.py index 9af8051..65019e7 100644 --- a/server/engines/base.py +++ b/server/engines/base.py @@ -14,7 +14,7 @@ from server.contracts import ScheduleResult # 引擎输出契约 class EngineParams(BaseModel): """引擎入参：一次排产请求的全部参数（与 legacy runScheduling params 对齐）。""" orderIds: list[int] = Field(default_factory=list) # 目标订单 ID（空=全部待排） - engineType: str = "RULE" # 请求的引擎类型（RULE/CP/GA/HYBRID） + engineType: str = "RULE" # 请求的引擎类型（RULE/CP/GA/HYBRID/OPTIMIZE） strategyTemplate: str = "COMPREHENSIVE" # 策略模板（排序规则） planningHorizonDays: int = 14 # 计划展望期（天） startDate: str | None = None # 排产起始日 YYYY-MM-DD（None=明天） diff --git a/server/engines/optimize_engine.py b/server/engines/optimize_engine.py new file mode 100644 index 0000000..35ad9f9 --- /dev/null +++ b/server/engines/optimize_engine.py @@ -0,0 +1,115 @@ +"""Optimize scheduling engine integrated with the APS V2 closed loop. + +The engine owns algorithm selection and provenance. APS still owns the world, +admission, candidate validation, version materialization, and audit trail. +""" +from __future__ import annotations + +from typing import Any, Callable + +from server.contracts import ScheduleResult +from server.engines.base import EngineParams, ISchedulingEngine +from server.engines.pool_engine import PoolEngine + + +DISPATCH_RULES = frozenset({"EDD", "SPT", "PRIORITY", "FIFO", "LPT", "CR", "ATC"}) + + +def normalize_dispatch_rule(value: str | None) -> str: + rule = str(value or "EDD").strip().upper().replace("-", "_") + aliases = { + "DELIVERY_FIRST": "EDD", + "EARLIEST_DUE_DATE": "EDD", + "FIRST_IN_FIRST_OUT": "FIFO", + "APPARENT_TARDINESS_COST": "ATC", + } + rule = aliases.get(rule, rule) + return rule if rule in DISPATCH_RULES else "EDD" + + +class OptimizeEngine(ISchedulingEngine): + """Python-native Optimize entry point for APS scheduling. + + The first integration reuses PoolEngine's already validated flex + materializer. Its ordering policy is supplied by ``dispatch_rule`` so the + seven optimize rules share APS calendars, teams, tooling, and rollback + semantics while the V2 runtime remains the authority for validation. + """ + + name = "OPTIMIZE" + supports_anytime = False + + def solve( + self, + world: dict[str, Any], + params: EngineParams, + next_id: Callable[[str], int], + ) -> ScheduleResult: + rule = normalize_dispatch_rule(params.strategyTemplate) + solved = PoolEngine().solve( + world, + next_id, + sort_mode="ASC", + order_ids=params.orderIds or None, + start_date=params.startDate, + name=params.name, + window=None, + enforce_teams=params.constraints.get("personnel") if params.constraints else None, + dispatch_rule=rule, + ) + return _summary_to_result(solved, rule) + + def solve_flex( + self, + world: dict[str, Any], + next_id: Callable[[str], int], + *, + dispatch_rule: str | None = None, + order_ids: list[int] | None = None, + start_date: str | None = None, + name: str | None = None, + window: str | None = None, + enforce_teams: bool | None = None, + ) -> dict[str, Any]: + """Materialize an Optimize candidate for the closed-loop V2 adapter.""" + + rule = normalize_dispatch_rule(dispatch_rule) + solved = PoolEngine().solve( + world, + next_id, + sort_mode="ASC", + order_ids=order_ids, + start_date=start_date, + name=name, + window=window, + enforce_teams=enforce_teams, + dispatch_rule=rule, + ) + solved.update({ + "engineType": "OPTIMIZE", + "algorithmId": f"optimize.{rule.lower()}", + "algorithmVersion": "1.0.0", + "solverId": "optimize-dispatch", + "solverVersion": "1.0.0", + "dispatchRule": rule, + }) + return solved + + +def _summary_to_result(solved: dict[str, Any], rule: str) -> ScheduleResult: + return ScheduleResult( + versionId=int(solved["versionId"]), + versionNo=str(solved["versionNo"]), + engineType="OPTIMIZE", + strategy=rule, + status="DRAFT", + orderCount=int(solved.get("orderCount") or 0), + poCount=int(solved.get("vlCount") or 0), + woCount=int(solved.get("woCount") or 0), + conflictCount=int(solved.get("conflictCount") or 0), + totalTardiness=float(solved.get("totalTardiness") or 0), + avgUtilization=float(solved.get("avgUtilization") or 0), + evidenceRefs=[f"algorithm:optimize.{rule.lower()}", f"run:{solved['versionId']}"], + solveStatus="FEASIBLE" if not solved.get("conflictCount") else "PARTIAL", + ) + diff --git a/server/engines/pool_engine.py b/server/engines/pool_engine.py index ad0e526..7bc0ec9 100644 --- a/server/engines/pool_engine.py +++ b/server/engines/pool_engine.py @@ -97,7 +97,8 @@ class PoolEngine: start_date: str | None = None, name: str | None = None, window: str | None = None, seed_busy: dict[int, list[tuple[datetime, datetime]]] | None = None, - enforce_teams: bool | None = None) -> dict[str, Any]: + enforce_teams: bool | None = None, + dispatch_rule: str | None = None) -> dict[str, Any]: """执行一次柔性排产，返回结果摘要 dict。 Args: @@ -148,7 +149,10 @@ class PoolEngine: orders.append(o) # ---- ③ 派工排序（吸收排产逻辑 PPT：正排 EDD / 倒排最晚优先 / 瓶颈锚）---- - if mode == SORT_DESC: + dispatch = str(dispatch_rule or "").strip().upper() + if dispatch in {"SPT", "LPT", "CR", "ATC", "PRIORITY", "FIFO", "EDD"}: + orders.sort(key=lambda o: self._dispatch_key(dispatch, o, routings, ops_by_code)) + elif mode == SORT_DESC: # 倒排：交期最晚的订单先占资源（自交期向前的派工近似） orders.sort(key=lambda o: (o["dueDate"], o["priority"]), reverse=True) elif mode == SORT_BOTTLENECK: @@ -477,6 +481,38 @@ class PoolEngine: "makespan": makespan, "onTimeCount": on_time, } + @staticmethod + def _dispatch_key(rule: str, order: dict[str, Any], routings: list[dict], ops_by_code: dict[str, dict]) -> tuple: + """Return stable dispatch keys for the optimize rule catalog.""" + steps = [step for step in routings if step.get("productCode") == order.get("productCode")] + duration = sum(float(step.get("stdTimePerUnit") or 1) for step in steps) * float(order.get("quantity") or 1) + due = str(order.get("dueDate") or "9999-12-31") + release = str(order.get("releaseDate") or order.get("releaseAt") or "0000-01-01") + priority = -int(order.get("priority") or 0) + def _day_number(value: str, fallback: float) -> float: + try: + return datetime.fromisoformat(value[:10]).toordinal() + except (TypeError, ValueError): + return fallback + due_day = _day_number(due, 3652059.0) + release_day = _day_number(release, 1.0) + slack = max(0.0, (due_day - release_day) * 24 * 60 - duration) + if rule == "SPT": + return (duration, due, priority, str(order.get("orderNo") or "")) + if rule == "LPT": + return (-duration, due, priority, str(order.get("orderNo") or "")) + if rule == "PRIORITY": + return (priority, due, release, str(order.get("orderNo") or "")) + if rule == "FIFO": + return (release, due, priority, str(order.get("orderNo") or "")) + if rule == "CR": + return ((due_day - release_day) / max(duration, 1e-9), priority, str(order.get("orderNo") or "")) + if rule == "ATC": + score = (abs(priority) or 1) / max(duration, 1e-9) + score *= pow(2.718281828, -slack / max(4 * duration, 1.0)) + return (-score, due, priority, str(order.get("orderNo") or "")) + return (due, priority, release, str(order.get("orderNo") or "")) + # ---------------- 占槽：设备级 + 可选班组并发（SC-11） ---------------- def _place(self, cursor: datetime, duration_min: float, eq_id: int, eq_busy: dict[int, list[tuple[datetime, datetime]]], world: World, diff --git a/server/gateway/app.py b/server/gateway/app.py index 94487e3..3e677a5 100644 --- a/server/gateway/app.py +++ b/server/gateway/app.py @@ -220,6 +220,7 @@ class FlexScheduleRequest(BaseModel): orderIds: list[int] = Field(default_factory=list) window: str | None = None # short/mid/long/full（SC-12） enforceTeams: bool | None = None # SC-11 班组约束 + engine: str | None = None # CLOSED_LOOP / OPTIMIZE class TimeUpdateRequest(BaseModel): @@ -2440,6 +2441,7 @@ def create_app() -> FastAPI: actor=req.sessionId or "web", window=req.window, enforce_teams=req.enforceTeams, + engine_type=req.engine, ) except (ValueError, PermissionError) as exc: return {"error": str(exc)} diff --git a/shared/schemas/schedule_result.schema.json b/shared/schemas/schedule_result.schema.json index b8f6d35..555e2c6 100644 --- a/shared/schemas/schedule_result.schema.json +++ b/shared/schemas/schedule_result.schema.json @@ -8,7 +8,7 @@ "properties": { "versionId": { "description": "排产版本 ID", "type": "integer" }, "versionNo": { "description": "版本号（如 V20260716-003）", "type": "string" }, - "engineType": { "description": "引擎类型", "type": "string", "enum": ["RULE", "CP", "GA", "HYBRID", "EXTERNAL"] }, + "engineType": { "description": "引擎类型", "type": "string", "enum": ["RULE", "CP", "GA", "HYBRID", "EXTERNAL", "OPTIMIZE"] }, "strategy": { "description": "策略模板", "type": "string" }, "status": { "description": "版本状态", "type": "string", "enum": ["DRAFT", "PUBLISHED", "ARCHIVED"] }, "orderCount": { "description": "参与排产的订单项数", "type": "integer" }, diff --git a/tests/golden/test_optimize_engine.py b/tests/golden/test_optimize_engine.py new file mode 100644 index 0000000..1be2d73 --- /dev/null +++ b/tests/golden/test_optimize_engine.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from server.aps_domain.closed_loop_runtime import run_closed_loop_candidate +from server.engines import get_engine +from server.engines.optimize_engine import normalize_dispatch_rule +from server.engines.pool_engine import PoolEngine + +from tests.golden.test_closed_loop_runtime import BUSINESS_DATE, _ready_world + + +def _next_id_factory(): + counters: dict[str, int] = {} + + def next_id(kind: str) -> int: + counters[kind] = counters.get(kind, 0) + 1 + return counters[kind] + + return next_id + + +def test_optimize_factory_and_rule_catalog_are_available(): + assert get_engine("OPTIMIZE").name == "OPTIMIZE" + assert normalize_dispatch_rule("DELIVERY_FIRST") == "EDD" + assert normalize_dispatch_rule("first-in-first-out") == "FIFO" + assert normalize_dispatch_rule("unknown") == "EDD" + + +def test_dispatch_rules_have_stable_ordering_keys(): + routings = [ + {"productCode": "SHORT", "stdTimePerUnit": 2}, + {"productCode": "LONG", "stdTimePerUnit": 10}, + ] + ops = {} + short = {"productCode": "SHORT", "quantity": 1, "dueDate": "2026-08-05", "priority": 2, "orderNo": "SO-S"} + long = {"productCode": "LONG", "quantity": 1, "dueDate": "2026-08-04", "priority": 1, "orderNo": "SO-L"} + assert sorted((long, short), key=lambda row: PoolEngine._dispatch_key("SPT", row, routings, ops)) == [short, long] + assert sorted((short, long), key=lambda row: PoolEngine._dispatch_key("LPT", row, routings, ops)) == [long, short] + + +def test_optimize_runs_through_closed_loop_v2_and_records_provenance(): + world = _ready_world() + result = run_closed_loop_candidate( + world, + _next_id_factory(), + business_date=BUSINESS_DATE, + engine_type="OPTIMIZE", + sort_mode="SPT", + ) + + assert result["solveStatus"] == "FEASIBLE" + assert result["engineType"] == "OPTIMIZE" + version = world["flexScheduleVersions"][-1] + assert version["engineType"] == "OPTIMIZE" + assert version["solverId"] == "optimize-dispatch" + assert version["algorithmId"] == "optimize.spt" + assert version["schedulingSolutionV2"]["assumptions"][0]["code"] == "OPTIMIZE_ENGINE_V1_ADAPTER" + assert version["schedulingSolutionV2"]["provenance"]["solverId"] == "optimize-dispatch" + + +def test_optimize_blocker_keeps_engine_identity_and_zero_artifacts(): + world = _ready_world() + world["materials"][1]["stock"] = 0 + world["routings"] = [] + result = run_closed_loop_candidate( + world, + _next_id_factory(), + business_date=BUSINESS_DATE, + engine_type="OPTIMIZE", + ) + + assert result["solveStatus"] == "BLOCKED" + assert result["engineType"] == "OPTIMIZE" + assert result["woCount"] == 0 + assert world["flexScheduleVersions"][-1]["engineType"] == "OPTIMIZE" + assert world["flexWorkOrders"] == []