# ============================================================ # 内置算法库注册表（moduleId: core-algolib, 可重生 ✅） # plan.md §9.5：算法收敛为带注册表的可重生资产（契约/黄金测试/适用边界）； # 与 core-skills（外部算法 SkillRegistry）区分——本表只登记**内置**算法， # SkillRegistry 管外部可插拔算法（HTTP/local 端点）。 # 资产分类（矩阵 86 行口径，A-E 五类）： # A=启发式（调度规则/构造启发式） B=精确（CP-SAT/MILP） # C=元启发（GA/LNS/SA/NSGA-II） D=ML（预测器/代理模型） # E=集成（混合引擎/评估链：HYBRID、敏感性、鲁棒性、参数优化） # 注：plan.md §9.5.1 的五类与任务矩阵 86 行分类略有差异，本表以 # 任务矩阵口径为准（元启发独立成 C，ML 归 D，E 为集成/评估链）。 # ============================================================ from __future__ import annotations import importlib import threading import time from collections.abc import Callable from typing import Any from pydantic import BaseModel, Field # 类别定义：A-E 五类资产的用途与适用边界（查询/健康检查展示用） CATEGORY_LABELS: dict[str, str] = { "A": "启发式：调度规则/构造启发式（EDD/SPT/ATC/贪心插入），毫秒级，作初始解与兜底", "B": "精确：CP-SAT/MILP 等精确或近似优化，秒~分钟级，规模与时限需匹配", "C": "元启发：GA/LNS/SA/NSGA-II，非确定性算法需固定随机种子才可复现", "D": "ML：预测器/代理模型（工时回归/交期概率/需求时序），需历史数据、输出带置信区间", "E": "集成：混合引擎与评估链（RULE+CP HYBRID、敏感性分析、蒙特卡洛鲁棒性、参数优化）", } # 默认随机种子（注册表级）：非确定性内置算法复现的统一种子源 DEFAULT_SEED = 20260802 # 健康检查历史保留条数 _HEALTH_HISTORY_KEEP = 20 class AlgorithmManifest(BaseModel): """算法注册表条目（plan.md §9.5.2 AlgorithmManifest 的内置实现形态）。 每个算法自描述：编排器/LLM 据此选型与调用；golden_tests 是重生验收。 """ algo_id: str # 算法唯一 ID，如 "rule.delivery_first" name: str # 展示名 category: str # A/B/C/D/E（对应 CATEGORY_LABELS） description: str = "" # 一句话用途 scale_limit: str = "" # 适用规模声明，如 "<=50k 工单" time_budget: str = "" # 典型时限，如 "5s~5min" input_schema: dict[str, Any] = Field(default_factory=dict) # 输入契约（JSON Schema 形态） output_schema: dict[str, Any] = Field(default_factory=dict) # 输出契约（含证据字段） golden_tests: list[str] = Field(default_factory=list) # 黄金测试用例路径（重生验收） deterministic: bool = True # 是否确定性（GA/SA 需固定种子） random_seed: int | None = None # 随机种子（非确定性算法的复现锚点） version: str = "1.0" # 语义化版本 regen_strategy: str = "manual" # 重生策略：llm / manual / hybrid available: bool = True # 内置实现是否就绪（False=目录资产未落地） entrypoint: str = "" # 取用入口：引擎名（RULE/CP/GA/HYBRID）、 # "ENGINE:STRATEGY" 或 "module.path:attr" health_fn: str | Callable[[], dict[str, Any]] = "" # 可选健康探针（点路径或可调用对象） def validate_manifest(self) -> list[str]: """元数据完整性校验，返回问题列表（空=通过）。""" problems: list[str] = [] if not self.algo_id or not self.algo_id.strip(): problems.append("algo_id 缺失") if not self.name or not self.name.strip(): problems.append("name 缺失") if self.category not in CATEGORY_LABELS: problems.append(f"category 非法：{self.category!r}（须为 A/B/C/D/E）") if not isinstance(self.input_schema, dict) or not isinstance(self.output_schema, dict): problems.append("input_schema/output_schema 必须是 dict") if not isinstance(self.golden_tests, list): problems.append("golden_tests 必须是 list") if not isinstance(self.deterministic, bool): problems.append("deterministic 必须是 bool") if not self.version or not str(self.version).strip(): problems.append("version 缺失") if self.regen_strategy not in ("llm", "manual", "hybrid"): problems.append(f"regen_strategy 非法：{self.regen_strategy!r}") if not self.deterministic and self.random_seed is None: problems.append("非确定性算法必须声明 random_seed（复现锚点）") return problems def _resolve_health_fn( spec: str | Callable[[], dict[str, Any]], ) -> Callable[[], dict[str, Any]] | None: """把 health_fn 规范解析为可调用对象（点路径或原对象）。""" if callable(spec): return spec if isinstance(spec, str) and ":" in spec: mod_path, attr = spec.rsplit(":", 1) try: fn = getattr(importlib.import_module(mod_path), attr) return fn if callable(fn) else None except (ImportError, AttributeError): return None return None def _builtin_catalog() -> list[AlgorithmManifest]: """内置算法目录：只登记仓库内真实存在的实现（available=True）， 未落地的规划资产（MILP/ML 预测器）以 available=False 登记， 健康检查如实报告未就绪，避免把目录当成果（LNS 已随 round-36 交付转为可用）。""" rule_tests = ["tests/golden/test_planning.py", "tests/golden/test_schedule_wizard.py"] cp_tests = ["tests/golden/test_cp_engine.py"] kpi_out = { "scheduleVersion": {"id": "int", "versionNo": "str", "status": "str"}, "productionOrders": "array<生产订单>", "workOrders": "array<工单>", "conflicts": "array<冲突>", "kpi": {"tardiness": "float(小时)", "conflicts": "int", "utilization": "float", "changeoverMin": "float"}, } rule_in = { "salesOrders": "array<订单>", "lines": "array<产线>", "strategy": "RULE 策略模板", "planningHorizonDays": "int", "constraints": "dict", } items = [ # ---- A. 启发式（RULE 引擎各策略模板）---- AlgorithmManifest( algo_id="rule.delivery_first", name="EDD 最早交期", category="A", description="交期优先派工（单机 1||Lmax 最优，Jackson 定理）", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=list(rule_tests), deterministic=True, entrypoint="RULE:DELIVERY_FIRST", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.fifo", name="FIFO 先来先服务", category="A", description="按订单下达时间排序", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=list(rule_tests), deterministic=True, entrypoint="RULE:FIFO", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.comprehensive", name="COMPREHENSIVE 综合策略", category="A", description="等级权重+优先级+交期综合排序（默认派工策略）", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=list(rule_tests), deterministic=True, entrypoint="RULE:COMPREHENSIVE", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.capacity_balance", name="CAPACITY_BALANCE 产能均衡", category="A", description="按等级+优先级+累计占用均衡选线", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=list(rule_tests), deterministic=True, entrypoint="RULE:CAPACITY_BALANCE", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.changeover_min", name="CHANGEOVER_MIN 换型最小化", category="A", description="贪心最近邻最小化换型（SC-07）", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=["tests/golden/test_changeover.py"], deterministic=True, entrypoint="RULE:CHANGEOVER_MIN", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.cost_first", name="COST_FIRST 成本优先", category="A", description="以换型耗时为成本代理排序（SC-07 备选）", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=["tests/golden/test_changeover.py"], deterministic=True, entrypoint="RULE:COST_FIRST", regen_strategy="manual", ), AlgorithmManifest( algo_id="rule.campaign", name="CAMPAIGN 战役合并", category="A", description="同产品交期窗口战役合并，按换型最小化排战役序（SC-08）", scale_limit="<=50k 工单", time_budget="毫秒级", input_schema=rule_in, output_schema=kpi_out, golden_tests=["tests/golden/test_campaign.py"], deterministic=True, entrypoint="RULE:CAMPAIGN", regen_strategy="manual", ), # ---- B. 精确 ---- AlgorithmManifest( algo_id="cp_sat", name="CP-SAT 约束规划", category="B", description="OR-Tools CP-SAT 详排求优（IntervalVar+NoOverlap+Cumulative）", scale_limit="<=50k 工单", time_budget="5s~5min", input_schema={**rule_in, "timeLimitSeconds": "float(秒，anytime)"}, output_schema=kpi_out, golden_tests=list(cp_tests), deterministic=True, entrypoint="CP", regen_strategy="hybrid", ), AlgorithmManifest( algo_id="milp", name="MILP 混合整数规划", category="B", description="S1 集团产能平衡与小规模基准校核（析取建模，Manne 1960）", scale_limit="<=2k 工单", time_budget="分钟级", input_schema=rule_in, output_schema=kpi_out, golden_tests=[], deterministic=True, entrypoint="", available=False, regen_strategy="manual", ), # ---- C. 元启发 ---- AlgorithmManifest( algo_id="ga", name="遗传算法 GA", category="C", description="种群/代数演化搜索近优解", scale_limit="<=10k 工单", time_budget="秒~分钟级", input_schema={**rule_in, "population": "int", "generations": "int"}, output_schema=kpi_out, golden_tests=["tests/golden/test_ga_engine.py"], deterministic=False, random_seed=DEFAULT_SEED, entrypoint="GA", regen_strategy="hybrid", ), AlgorithmManifest( algo_id="lns", name="LNS/ALNS 大邻域搜索", category="C", description="插单局部修复：固定窗口最小扰动，超阈值升级全量重排（§9.11，round-36 交付）", scale_limit="<=50k 工单", time_budget="<30s", input_schema={**rule_in, "frozenWindowHours": "float", "maxAffectedOrders": "int", "disturbanceTolerance": "float"}, output_schema=kpi_out, golden_tests=["tests/golden/test_rush_lns.py"], deterministic=False, random_seed=DEFAULT_SEED, entrypoint="server.aps_domain.lns:lns_local_repair", available=True, regen_strategy="hybrid", ), AlgorithmManifest( algo_id="nsga2", name="NSGA-II 多目标遗传", category="C", description="多目标帕累托解集：快速非支配排序+拥挤度+精英保留，3 目标（总延迟/容量冲突/负载均衡）（§9.7，round-40 交付）", scale_limit="<=10k 工单", time_budget="分钟级", input_schema={**rule_in, "objectives": "array<目标键>", "population": "int", "generations": "int"}, output_schema={**kpi_out, "paretoSet": "array<方案>"}, golden_tests=["tests/golden/test_nsga2_engine.py"], deterministic=False, random_seed=DEFAULT_SEED, entrypoint="server.engines.nsga2_engine:solve_nsga2", available=True, regen_strategy="hybrid", ), # ---- D. ML（预测器/代理模型，当前目录资产，未落地实现）---- AlgorithmManifest( algo_id="ml.worktime_regression", name="实际工时回归", category="D", description="LightGBM/XGBoost 工时估计（§9.1a OR 输入参数估准）", scale_limit="需历史报工数据", time_budget="训练级", input_schema={"features": "dict", "history": "array<报工记录>"}, output_schema={"estimate": "float", "confidenceInterval": "[low, high]"}, golden_tests=[], deterministic=False, random_seed=DEFAULT_SEED, entrypoint="", available=False, regen_strategy="llm", ), AlgorithmManifest( algo_id="ml.due_date_survival", name="交期达成概率", category="D", description="生存分析：订单交期达成概率", scale_limit="需历史订单数据", time_budget="训练级", input_schema={"order": "dict", "history": "array<历史订单>"}, output_schema={"deliveryProbability": "float", "confidenceInterval": "[low, high]"}, golden_tests=[], deterministic=False, random_seed=DEFAULT_SEED, entrypoint="", available=False, regen_strategy="llm", ), # ---- E. 集成（混合引擎与评估链）---- AlgorithmManifest( algo_id="hybrid", name="HYBRID 规则+CP 混合", category="E", description="RULE 热启动构造初始解 + CP-SAT 局部改良", scale_limit="<=50k 工单", time_budget="5s~5min", input_schema={**rule_in, "timeLimitSeconds": "float(秒，anytime)"}, output_schema=kpi_out, golden_tests=list(cp_tests), deterministic=False, random_seed=DEFAULT_SEED, entrypoint="HYBRID", regen_strategy="hybrid", ), AlgorithmManifest( algo_id="eval.sensitivity_tornado", name="敏感性 Tornado 分析", category="E", description="OAT 单因子扰动排序（SC-06，跑 Explore 沙盒）", scale_limit="按需", time_budget="秒级/因子", input_schema={"world": "dict", "strategy": "str"}, output_schema={"rows": "array<因子摆幅>", "monteCarlo": "dict", "markdown": "str"}, golden_tests=["tests/golden/test_sensitivity.py"], deterministic=True, entrypoint="server.aps_domain.sensitivity:run_sensitivity", regen_strategy="manual", ), AlgorithmManifest( algo_id="eval.monte_carlo_robustness", name="蒙特卡洛鲁棒性", category="E", description="固定种子情景仿真，鲁棒性评分（§9.9）", scale_limit="按需", time_budget="后台批量", input_schema={"world": "dict", "trials": "int", "seed": "int"}, output_schema={"robustness": "float", "distribution": "dict", "seed": "int"}, golden_tests=["tests/golden/test_sensitivity.py"], deterministic=True, random_seed=DEFAULT_SEED, entrypoint="server.aps_domain.robustness:run_monte_carlo", regen_strategy="manual", ), AlgorithmManifest( algo_id="opt.parameter_optimizer", name="参数优化闭环", category="E", description="回放→灰度→验证→退化自动回滚（§9.8）", scale_limit="按需", time_budget="版本级", input_schema={"world": "dict", "candidates": "array<参数字典>"}, output_schema={"experiments": "array<实验记录>", "summary": "dict"}, golden_tests=["tests/golden/test_param_opt.py"], deterministic=True, random_seed=DEFAULT_SEED, entrypoint="server.agent_core.param_opt:ParameterOptimizer", regen_strategy="manual", ), ] return items class AlgorithmRegistry: """内置算法注册表（内存态，无文件持久化——内置资产随代码版本演进）。 与 core-skills 的 SkillRegistry 区分：SkillRegistry 管**外部**算法 （manifest.json + HTTP 端点），本注册表管**内置**算法（引擎/策略/评估链）。 """ def __init__(self, seed: int = DEFAULT_SEED) -> None: self._lock = threading.Lock() self._seed = int(seed) self._algos: dict[str, AlgorithmManifest] = {} self.health_history: dict[str, list[dict[str, Any]]] = {} for m in _builtin_catalog(): self._algos[m.algo_id] = m # ---------------- 注册 / 注销 ---------------- def register(self, manifest: AlgorithmManifest | dict[str, Any]) -> dict[str, Any]: """注册或更新（upsert）一个内置算法。元数据不完整直接拒绝。""" m = manifest if isinstance(manifest, AlgorithmManifest) else AlgorithmManifest(**manifest) problems = m.validate_manifest() if problems: raise ValueError("算法元数据不完整：" + "; ".join(problems)) with self._lock: self._algos[m.algo_id] = m return m.model_dump() def unregister(self, algo_id: str) -> bool: """注销算法（内置资产不建议注销；测试/热替换用）。""" with self._lock: return self._algos.pop(algo_id, None) is not None # ---------------- 查询 ---------------- def get(self, algo_id: str) -> dict[str, Any] | None: m = self._algos.get(algo_id) return m.model_dump() if m else None def list(self) -> list[dict[str, Any]]: return [m.model_dump() for m in self._algos.values()] def query( self, *, category: str | None = None, available: bool | None = None, keywords: str = "", max_time_budget: str | None = None, ) -> list[dict[str, Any]]: """按类别/可用性/关键词/时限声明查询（编排器选型入口）。""" kw = keywords.strip().lower() out: list[dict[str, Any]] = [] for m in self._algos.values(): if category and m.category != category: continue if available is not None and m.available != available: continue if kw and kw not in (m.name + m.description + m.algo_id).lower(): continue if max_time_budget and not self._budget_le(m.time_budget, max_time_budget): continue out.append(m.model_dump()) out.sort(key=lambda x: (x["category"], x["algo_id"])) return out def categories(self) -> dict[str, str]: return dict(CATEGORY_LABELS) def version(self, algo_id: str) -> dict[str, Any] | None: """版本查询：返回 {algo_id, version, regen_strategy, available}。""" m = self._algos.get(algo_id) if not m: return None return { "algo_id": m.algo_id, "version": m.version, "regen_strategy": m.regen_strategy, "available": m.available, } def next_seed(self, algo_id: str) -> int: """随机种子服务：确定性算法返回声明的种子（无则 None 语义用 0）； 非确定性算法返回按 algo_id 稳定的种子（同注册表种子源可复现）。""" m = self._algos.get(algo_id) if not m: return self._seed if m.deterministic: return m.random_seed if m.random_seed is not None else 0 return m.random_seed if m.random_seed is not None else (self._seed + sum(map(ord, algo_id))) # ---------------- 健康检查 ---------------- def health(self, algo_id: str | None = None) -> list[dict[str, Any]]: """健康检查：元数据完整性 + 就绪状态 + 可选探针；记录历史（最近 N 条）。""" targets = [self._algos[algo_id]] if algo_id and algo_id in self._algos else ([] if algo_id else list(self._algos.values())) out: list[dict[str, Any]] = [] for m in targets: status = self._probe(m) record = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), **status} hist = self.health_history.setdefault(m.algo_id, []) hist.append(record) del hist[:-_HEALTH_HISTORY_KEEP] out.append({ "algo_id": m.algo_id, "name": m.name, "category": m.category, "version": m.version, "available": m.available, **status, }) return out def history(self, algo_id: str) -> list[dict[str, Any]]: return list(self.health_history.get(algo_id) or []) def _probe(self, m: AlgorithmManifest) -> dict[str, Any]: """单算法探针：元数据→可用性→入口→自定义探针，逐级短路报告。""" started = time.time() problems = m.validate_manifest() if problems: return {"ok": False, "latencyMs": 0, "detail": "元数据不完整: " + "; ".join(problems)} if not m.available: return {"ok": False, "latencyMs": 0, "detail": "目录资产未落地（available=False）"} detail = "builtin" if m.entrypoint: ok, err = self._check_entrypoint(m.entrypoint) if not ok: return {"ok": False, "latencyMs": 0, "detail": f"入口不可达: {err}"} detail = f"entrypoint ok ({m.entrypoint})" if m.health_fn: fn = _resolve_health_fn(m.health_fn) if fn is None: return {"ok": False, "latencyMs": 0, "detail": "health_fn 不可解析"} try: r = fn() or {} if not isinstance(r, dict) or "ok" not in r: return {"ok": False, "latencyMs": 0, "detail": "health_fn 返回格式非法"} latency = int((time.time() - started) * 1000) return {"ok": bool(r["ok"]), "latencyMs": latency, "detail": str(r.get("detail") or detail)} except Exception as exc: # noqa: BLE001 - 健康探针必须兜底任意异常（异常即不健康） return {"ok": False, "latencyMs": 0, "detail": f"health_fn 异常: {exc}"} return {"ok": True, "latencyMs": int((time.time() - started) * 1000), "detail": detail} @staticmethod def _check_entrypoint(entrypoint: str) -> tuple[bool, str]: """入口可达性：引擎名 / ENGINE:STRATEGY / module.path:attr。""" if ":" not in entrypoint: return True, "" # 纯引擎名（RULE/CP/GA/HYBRID）由 get_engine 工厂保证 if entrypoint.startswith("RULE:"): from server.engines import get_engine try: get_engine("RULE") return True, "" except (ImportError, AttributeError, RuntimeError, ValueError, TypeError) as exc: return False, str(exc) mod_path, attr = entrypoint.rsplit(":", 1) try: getattr(importlib.import_module(mod_path), attr) return True, "" except (ImportError, AttributeError) as exc: return False, str(exc) @staticmethod def _budget_le(declared: str, limit: str) -> bool: """时限声明比较（粗粒度：解析首段数值，如 "5s~5min" → 5min）。""" def parse(spec: str) -> float: text = spec.strip().lower() if not text: return 0.0 num = "" for ch in text: if ch.isdigit() or ch == ".": num += ch elif num: break try: val = float(num) except ValueError: return 0.0 if "ms" in text: return val / 1000.0 if "s" in text: return val if "min" in text: return val * 60 if "h" in text: return val * 3600 return val return parse(declared) <= parse(limit) if parse(limit) > 0 else True _registry: AlgorithmRegistry | None = None def get_algolib() -> AlgorithmRegistry: """单例获取（与 get_skills 对齐）。""" global _registry if _registry is None: _registry = AlgorithmRegistry() return _registry def reset_algolib_registry() -> None: """测试用：丢弃单例，下次按默认目录重建。""" global _registry _registry = None