""" 多模型编程任务派发问题 —— 状态空间搜索求解器 场景：用 Opus / Sonnet / DeepSeek 三个 AI 编程助手完成 GridTrax v1.1 的 8 个开发任务， 在依赖、能力、额度、交叉审查等约束下，寻找总额度消耗最小的派发序列。 实现的搜索策略（与课程第二章一一对应）： BFS 宽度优先搜索 OPEN 表 FIFO DFS 有界深度优先搜索 OPEN 表 LIFO，深度界限 = 任务数 UCS 等代价搜索 OPEN 表按 g(n) 排序 HILL 局部择优搜索 子节点按 f(n) 排序后插入 OPEN 表首部 ASTAR_H1 A* 算法，h1(n) 类比八数码“不在位数 W(n)” ASTAR_H2 A* 算法，h2(n) 类比八数码“曼哈顿距离 P(n)” ASTAR_H3 A* 算法，h3(n) 分组感知的改进启发函数 只用 Python 标准库，运行：python3 solver.py 结果写入 viz/data.js（供可视化页面读取）和 results.json。 """ import heapq import itertools import json import time from collections import deque from dataclasses import dataclass, field from pathlib import Path # --------------------------------------------------------------------------- # 1. 问题定义 # --------------------------------------------------------------------------- MODELS = ["O", "S", "D"] MODEL_NAMES = {"O": "Opus", "S": "Sonnet", "D": "DeepSeek"} # 任务：编号、名称、所属模块组、前置依赖、说明 TASKS = [ {"id": "T1", "name": "TMDB↔Bangumi 映射设计", "group": "数据层", "deps": [], "note": "架构级设计，高风险，只有 Opus 能胜任"}, {"id": "T2", "name": "Bangumi 同步实现", "group": "数据层", "deps": ["T1"], "note": "核心业务代码，是代码审查的对象"}, {"id": "T3", "name": "豆瓣数据源接入", "group": "数据层", "deps": ["T1"], "note": "复用映射层，接入第二个元数据源"}, {"id": "T4", "name": "PWA 离线支持", "group": "前端", "deps": [], "note": "Service Worker 与缓存策略"}, {"id": "T5", "name": "年度追剧图表", "group": "前端", "deps": ["T1"], "note": "依赖映射后的数据模型"}, {"id": "T6", "name": "测试编写", "group": "质量", "deps": ["T2", "T3", "T4", "T5"], "note": "覆盖同步、数据源、离线、图表"}, {"id": "T7", "name": "代码审查", "group": "质量", "deps": ["T6"], "note": "审查者不能是 T2 的作者；DeepSeek 不承担审查"}, {"id": "T8", "name": "发布日志", "group": "文档", "deps": ["T7"], "note": "写博客版更新说明"}, ] # 单位：额度点（各家订阅按单价折算到统一单位）；None 表示该模型不能胜任 COST = { "T1": {"O": 9, "S": None, "D": None}, "T2": {"O": 8, "S": 6, "D": 4}, "T3": {"O": 7, "S": 5, "D": 4}, "T4": {"O": 6, "S": 4, "D": 3}, "T5": {"O": 6, "S": 5, "D": 3}, "T6": {"O": 5, "S": 3, "D": 2}, "T7": {"O": 4, "S": 3, "D": None}, "T8": {"O": 4, "S": 2, "D": 2}, } # 上下文复用：同一模型连续执行同组任务，代价减 DISCOUNT（最低为 1） DISCOUNT = 2 # 交叉审查约束：REVIEW_TASK 的执行者 != REVIEWED_TASK 的作者 REVIEW_TASK, REVIEWED_TASK = "T7", "T2" N = len(TASKS) TASK_IDX = {t["id"]: k for k, t in enumerate(TASKS)} GROUP = [t["group"] for t in TASKS] DEPS = [[TASK_IDX[d] for d in t["deps"]] for t in TASKS] REV_I, REVD_I = TASK_IDX[REVIEW_TASK], TASK_IDX[REVIEWED_TASK] @dataclass(frozen=True) class Problem: name: str quota: tuple # 各模型初始额度，顺序同 MODELS PROBLEMS = { "main": Problem("主实例", (16, 8, 11)), "infeasible": Problem("无解实例", (12, 2, 10)), } # 状态 S = (a1..a8, qO, qS, qD, k) # a_i ∈ {-1, 0, 1, 2}：任务 i 未派发 / 已由 MODELS[a_i] 完成 # q_m ：模型 m 剩余额度 # k ：上一步完成的任务下标（-1 表示尚未开始）；上一步的模型即 a_k # 状态用元组表示：(authors, quotas, k) def initial_state(p: Problem): return ((-1,) * N, p.quota, -1) def is_goal(state) -> bool: authors, _, _ = state if any(a < 0 for a in authors): return False return authors[REV_I] != authors[REVD_I] def step_cost(i, m, authors, last_task): """派发任务 i 给模型 m 的实际代价（含上下文复用折扣）""" c = COST[TASKS[i]["id"]][MODELS[m]] if last_task >= 0 and authors[last_task] == m and GROUP[last_task] == GROUP[i]: c = max(1, c - DISCOUNT) return c # 算符 assign(i, m)：把就绪任务 i 派给模型 m # 与八数码“只对空格定义算符”同理，只对当前就绪的任务生成算符 def ready_tasks(authors): return [i for i in range(N) if authors[i] < 0 and all(authors[d] >= 0 for d in DEPS[i])] def legal_check(state, i, m): """返回 (是否合法, 不合法原因, 代价)。用于合法性表和后继生成。""" authors, quotas, last_task = state c0 = COST[TASKS[i]["id"]][MODELS[m]] if c0 is None: return False, "能力不匹配", None if i == REV_I and authors[REVD_I] == m: return False, "审查者不能是作者", None c = step_cost(i, m, authors, last_task) if quotas[m] < c: return False, f"额度不足（剩 {quotas[m]}，需 {c}）", c return True, "", c def successors(state): """生成全部合法后继：[(action(i,m), cost, new_state)]""" authors, quotas, _ = state out = [] for i in ready_tasks(authors): for m in range(len(MODELS)): ok, _, c = legal_check(state, i, m) if not ok: continue na = list(authors); na[i] = m nq = list(quotas); nq[m] -= c out.append(((i, m), c, (tuple(na), tuple(nq), i))) return out # --------------------------------------------------------------------------- # 2. 启发函数 # --------------------------------------------------------------------------- def cheapest(i): """任务 i 在所有能胜任模型中的最低可能代价（按可能享受折扣估计，保证不高估）""" return min(max(1, c - DISCOUNT) for c in COST[TASKS[i]["id"]].values() if c is not None) CHEAPEST = [cheapest(i) for i in range(N)] GLOBAL_MIN = min(CHEAPEST) def h0(state): return 0 def h1(state): """剩余任务数 × 全局最低单价（类比八数码不在位数 W(n)）""" return sum(1 for a in state[0] if a < 0) * GLOBAL_MIN def h2(state): """各剩余任务最低可能代价之和（类比八数码曼哈顿距离 P(n)）""" return sum(CHEAPEST[i] for i, a in enumerate(state[0]) if a < 0) FULL_MIN = [min(c for c in COST[t["id"]].values() if c is not None) for t in TASKS] GROUPS = sorted(set(GROUP), key=GROUP.index) def h3(state): """改进：分组感知。一个模块组里最先执行的任务前面没有同组任务，享受不到折扣， 所以每个尚未开工的组至少有一个任务要付全价。在 h2 基础上补上这部分差额。""" authors = state[0] est = h2(state) for grp in GROUPS: members = [i for i in range(N) if GROUP[i] == grp] if all(authors[i] < 0 for i in members): est += min(FULL_MIN[i] - CHEAPEST[i] for i in members) return est # --------------------------------------------------------------------------- # 3. 通用图搜索（OPEN 表 + CLOSED 表） # --------------------------------------------------------------------------- @dataclass class Node: id: int state: tuple parent: int action: tuple # (i, m) 或 None g: int h: int depth: int @property def f(self): return self.g + self.h @dataclass class SearchResult: strategy: str found: bool path: list = field(default_factory=list) # 节点 id 序列 cost: int = None expanded: int = 0 generated: int = 0 max_open: int = 0 elapsed_ms: float = 0.0 nodes: list = field(default_factory=list) rounds: list = field(default_factory=list) # 每轮 OPEN/CLOSED 快照 STRATEGIES = { "BFS": ("宽度优先搜索", h0), "DFS": ("有界深度优先搜索", h0), "UCS": ("等代价搜索", h0), "HILL": ("局部择优搜索（爬山法）", h3), "ASTAR_H1": ("A* 算法 · h1", h1), "ASTAR_H2": ("A* 算法 · h2", h2), "ASTAR_H3": ("A* 算法 · h3（改进）", h3), } OPEN_SNAPSHOT = 14 # 每轮记录 OPEN 表前若干项 def search(problem: Problem, strategy: str, record=True) -> SearchResult: _, h = STRATEGIES[strategy] t0 = time.perf_counter() res = SearchResult(strategy, False) counter = itertools.count() nodes: list[Node] = [] def new_node(state, parent, action, g, depth): n = Node(len(nodes), state, parent, action, g, h(state), depth) nodes.append(n) return n root = new_node(initial_state(problem), -1, None, 0, 0) # OPEN 表的三种组织方式 priority = strategy in ("UCS", "ASTAR_H1", "ASTAR_H2", "ASTAR_H3") if priority: key = (lambda n: n.g) if strategy == "UCS" else (lambda n: n.f) open_heap = [(key(root), root.h, next(counter), root.id)] else: open_list = deque([root.id]) best_g = {root.state: 0} # 状态 -> OPEN 中或已扩展的最优 g closed = set() # 已扩展状态 closed_order = [] def open_ids(): if priority: return [x[3] for x in sorted(open_heap) if nodes[x[3]].state not in closed and best_g.get(nodes[x[3]].state) == nodes[x[3]].g] return list(open_list) def open_size(): return len(open_heap) if priority else len(open_list) while open_size(): res.max_open = max(res.max_open, open_size()) # —— 取出 OPEN 表首节点 —— if priority: _, _, _, nid = heapq.heappop(open_heap) n = nodes[nid] if n.state in closed or best_g.get(n.state) != n.g: continue # 过期条目（已被更优路径取代） else: nid = open_list.popleft() n = nodes[nid] if n.state in closed: continue closed.add(n.state) closed_order.append(n.id) res.expanded += 1 rnd = {"pop": n.id, "goal": False, "children": []} if is_goal(n.state): res.found = True rnd["goal"] = True if record: rnd["open"] = open_ids()[:OPEN_SNAPSHOT] rnd["open_size"] = open_size() rnd["closed_size"] = len(closed_order) res.rounds.append(rnd) path = [] k = n.id while k != -1: path.append(k) k = nodes[k].parent res.path = path[::-1] res.cost = n.g break # —— 扩展节点 —— children = [] if not (strategy == "DFS" and n.depth >= N): # 有界深度 for action, c, s in successors(n.state): if s in closed: continue g = n.g + c if priority: if s in best_g and best_g[s] <= g: continue else: if s in best_g: # 盲目/局部择优：已在 OPEN 中则不重复加入 continue best_g[s] = g child = new_node(s, n.id, action, g, n.depth + 1) children.append(child) res.generated += 1 # —— 按策略插入 OPEN 表 —— if priority: for ch in children: heapq.heappush(open_heap, (key(ch), ch.h, next(counter), ch.id)) elif strategy == "BFS": open_list.extend(ch.id for ch in children) elif strategy == "DFS": open_list.extendleft(ch.id for ch in reversed(children)) elif strategy == "HILL": children.sort(key=lambda ch: (ch.f, ch.h)) open_list.extendleft(ch.id for ch in reversed(children)) if record: rnd["children"] = [ch.id for ch in children] rnd["open"] = open_ids()[:OPEN_SNAPSHOT] rnd["open_size"] = open_size() rnd["closed_size"] = len(closed_order) res.rounds.append(rnd) res.elapsed_ms = (time.perf_counter() - t0) * 1000 res.nodes = nodes return res def greedy_no_backtrack(problem: Problem): """纯爬山（不回溯）：每步只取当前代价最小的派法，走进死胡同就失败""" state, g, steps = initial_state(problem), 0, [] while not is_goal(state): succ = successors(state) if not succ: return {"found": False, "cost": g, "steps": steps, "stuck_state": state} (i, m), c, s = min(succ, key=lambda x: (x[1], x[0])) steps.append({"task": TASKS[i]["id"], "model": MODELS[m], "cost": c}) state, g = s, g + c return {"found": True, "cost": g, "steps": steps} # --------------------------------------------------------------------------- # 4. 全状态空间（用于状态空间图与可采纳性验证） # --------------------------------------------------------------------------- def full_state_space(problem: Problem): """枚举全部可达状态，并逆向计算 h*""" start = initial_state(problem) seen = {start: 0} order = [start] edges = [] q = deque([start]) while q: cur = q.popleft() if is_goal(cur): continue for (i, m), c, nxt in successors(cur): if nxt not in seen: seen[nxt] = len(order) order.append(nxt) q.append(nxt) edges.append((seen[cur], seen[nxt], c, i, m)) # 逆向 Dijkstra 求 h* INF = float("inf") hstar = [INF] * len(order) rev = [[] for _ in order] for a, b, c, _, _ in edges: rev[b].append((a, c)) pq = [] for k, s in enumerate(order): if is_goal(s): hstar[k] = 0 heapq.heappush(pq, (0, k)) while pq: d, k = heapq.heappop(pq) if d > hstar[k]: continue for a, c in rev[k]: if d + c < hstar[a]: hstar[a] = d + c heapq.heappush(pq, (d + c, a)) return order, edges, hstar # --------------------------------------------------------------------------- # 5. 输出 # --------------------------------------------------------------------------- def fmt_state(state): authors, quotas, k = state a = "".join(MODELS[x] if x >= 0 else "·" for x in authors) return f"({a} | {quotas[0]},{quotas[1]},{quotas[2]} | {TASKS[k]['id'] if k >= 0 else '-'})" def node_json(n: Node): return { "id": n.id, "parent": n.parent, "depth": n.depth, "g": n.g, "h": n.h, "f": n.f, "authors": list(n.state[0]), "quota": list(n.state[1]), "last": n.state[2], "action": None if n.action is None else {"task": TASKS[n.action[0]]["id"], "model": MODELS[n.action[1]]}, } def result_json(r: SearchResult): nodes = [] for n in r.nodes: d = node_json(n) if n.action is not None: d["action"]["cost"] = n.g - r.nodes[n.parent].g nodes.append(d) return { "strategy": r.strategy, "label": STRATEGIES[r.strategy][0], "found": r.found, "cost": r.cost, "path": r.path, "expanded": r.expanded, "generated": r.generated, "max_open": r.max_open, "elapsed_ms": round(r.elapsed_ms, 3), "nodes": nodes, "rounds": r.rounds, } def legality_table(state): rows = [] authors = state[0] ready = set(ready_tasks(authors)) for i in range(N): if authors[i] >= 0: continue for m in range(len(MODELS)): if i not in ready: deps = [TASKS[d]["id"] for d in DEPS[i] if authors[d] < 0] rows.append({"task": TASKS[i]["id"], "model": MODELS[m], "ok": False, "reason": "前置未完成：" + "、".join(deps), "cost": None}) continue ok, why, c = legal_check(state, i, m) rows.append({"task": TASKS[i]["id"], "model": MODELS[m], "ok": ok, "reason": why, "cost": c}) return rows def timed(problem, strategy, repeat=20): """多次运行取中位数耗时，避免单次计时抖动""" ts = [] for _ in range(repeat): r = search(problem, strategy, record=False) ts.append(r.elapsed_ms) ts.sort() return ts[len(ts) // 2] def main(): out_dir = Path(__file__).parent data = { "models": MODELS, "model_names": MODEL_NAMES, "tasks": TASKS, "cost": COST, "discount": DISCOUNT, "review": [REVIEW_TASK, REVIEWED_TASK], "cheapest": dict(zip([t["id"] for t in TASKS], CHEAPEST)), "global_min": GLOBAL_MIN, "problems": {}, } for key, p in PROBLEMS.items(): print(f"\n===== {p.name} 初始额度 O/S/D = {p.quota} =====") order, edges, hstar = full_state_space(p) n_goal = sum(1 for s in order if is_goal(s)) print(f"可达搜索节点 {len(order)} 个，边 {len(edges)} 条，目标节点 {n_goal} 个") # 可采纳性验证：对全部可解节点检查 h1 <= h2 <= h3 <= h* # 一致性验证：对全部边检查 h(n) <= c(n, n') + h(n')（不重开 CLOSED 节点时 A* 仍最优的条件） adm = {"checked": 0, "h1_le_h2": True, "h2_le_h3": True, "h3_le_hstar": True, "consistent": {"h1": True, "h2": True, "h3": True}} for k, s in enumerate(order): if hstar[k] == float("inf"): continue adm["checked"] += 1 adm["h1_le_h2"] &= h1(s) <= h2(s) adm["h2_le_h3"] &= h2(s) <= h3(s) adm["h3_le_hstar"] &= h3(s) <= hstar[k] for a, b, c, _, _ in edges: for name, fn in (("h1", h1), ("h2", h2), ("h3", h3)): adm["consistent"][name] &= fn(order[a]) <= c + fn(order[b]) print(f"可采纳性验证（{adm['checked']} 个可解节点）：h1<=h2 {adm['h1_le_h2']}，" f"h2<=h3 {adm['h2_le_h3']}，h3<=h* {adm['h3_le_hstar']}；一致性 {adm['consistent']}") pdata = {"name": p.name, "quota": list(p.quota), "space": {"nodes": len(order), "edges": len(edges), "goals": n_goal, "hstar0": None if hstar[0] == float("inf") else hstar[0]}, "admissible": adm, "legality0": legality_table(initial_state(p)), "greedy": greedy_no_backtrack(p), "results": {}} # 状态空间图（规模适中时全部导出） pdata["graph"] = { "nodes": [{"authors": list(s[0]), "quota": list(s[1]), "last": s[2], "goal": is_goal(s), "hstar": None if hstar[k] == float("inf") else hstar[k]} for k, s in enumerate(order)], "edges": [[a, b, c] for a, b, c, _, _ in edges], } print(f"{'策略':<22}{'找到解':<6}{'代价':>5}{'扩展':>7}{'生成':>7}{'OPEN峰值':>9}{'耗时ms':>9}") for sname in STRATEGIES: r = search(p, sname) r.elapsed_ms = timed(p, sname) pdata["results"][sname] = result_json(r) print(f"{STRATEGIES[sname][0]:<20}{str(r.found):<8}{str(r.cost):>5}{r.expanded:>7}" f"{r.generated:>7}{r.max_open:>9}{r.elapsed_ms:>9.3f}") if sname == "ASTAR_H2" and r.found: print("最优派发序列：") for nid in r.path[1:]: nd = r.nodes[nid] i, m = nd.action print(f" assign({TASKS[i]['id']}, {MODELS[m]}) g={nd.g:<3} 状态 {fmt_state(nd.state)}") g = pdata["greedy"] print(f"纯爬山（不回溯）：{'找到解，代价 ' + str(g['cost']) if g['found'] else '陷入死局'} " + " → ".join(f"{s['task']}:{s['model']}" for s in g["steps"])) data["problems"][key] = pdata (out_dir / "results.json").write_text(json.dumps(data, ensure_ascii=False)) (out_dir / "viz").mkdir(exist_ok=True) (out_dir / "viz" / "data.js").write_text( "window.SEARCH_DATA = " + json.dumps(data, ensure_ascii=False) + ";\n") print("\n已写入 results.json 与 viz/data.js") if __name__ == "__main__": main()