docs(mode_workflow): Part A1 query-score 后端 实现计划

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docs/superpowers/plans/2026-06-18-partA1-query-score-backend.md

@@ -0,0 +1,505 @@
+# Part A1：query-builder 后端（矩阵/分类树/评分脚本/接口）Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 提供 query-builder 弹层所需的全部后端：返回内容树矩阵、拉取并缓存实质/形式分类树、用 Sonnet 对 tier≥1 格批量评分的脚本，以及触发/取结果的接口（带按选择哈希的缓存，省钱）。
+
+**Architecture:** 新增独立脚本 `stages/query_score.py`（异步并发分批评分，结果原子落盘 `.cache/query_score/<sel>.json`）+ prompt 文件。`server.py` 加 4 个路由：`GET /api/query_matrix`、`GET /api/category_tree`、`POST /api/query_score`（命中缓存短路，否则 `_spawn_task` 起脚本）、`GET /api/query_score?sel=`（读结果）。server 负责选择哈希与缓存判定，脚本只按 `--sel` 写对应缓存文件，二者不互相 import（避免把 agent 重栈拖进 server）。
+
+**Tech Stack:** Python 3 stdlib（`http.server`/`urllib`）+ 现有 `agent.llm.openrouter`。无 pytest——验证用 `python3 -m py_compile`、`python3 -c "import server; ..."`（手册已这么用，`import server` 无副作用）、脚本 `--dry-run`/`--limit`、`curl`（需手动重启 server 后）。
+
+参考 spec：`docs/superpowers/specs/2026-06-18-query-builder-design.md`（Part A / 评分脚本 / 评分 Prompt）。
+
+工作目录：所有命令在 `cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow` 下；提交在仓库根 `/Users/max_liu/max_liu/company/Agent`。分支 `dev-lxn`。
+
+---
+
+## File Structure
+
+- `examples/mode_workflow/prompts/query_score_system.md` — 评分 system prompt（可单独迭代）。
+- `examples/mode_workflow/stages/query_score.py` — 评分脚本（读矩阵→筛 tier≥1→拼词→分批 Sonnet→原子落盘）。
+- `examples/mode_workflow/server.py` — 加 `_matrix()`/`_category_tree()`/`_query_sel_hash()` 三个 helper + 4 个路由。
+
+---
+
+### Task 1: 评分 Prompt 文件
+
+**Files:**
+- Create: `examples/mode_workflow/prompts/query_score_system.md`
+
+- [ ] **Step 1: 写文件**
+
+```markdown
+你是「内容制作知识库」的检索词评审。我们要从小红书/公众号等公共平台搜到
+"怎么做某类内容"的工序/工具教程。用户会给你一批由内容树维度正交生成的候选 query
+(每批共享一段固定上下文)。
+
+【逐条打分,只看这条 query 本身】
+- natural(0-10): 像不像真人会在搜索框打的人话(越口语、越具体越高;生造词/学术腔越低)
+- findable(0-10): 公共域是否真有对口的"怎么做"内容能被搜到(可参考 tier,但以 query 实际可搜性为准)
+- useful(0-10): 命中的内容对"内容制作知识库"是否有价值(教程/方法/工具 高;成品展示/抽象概念 低)
+- keep(bool): 三者均衡后是否值得真的拿去搜(有意义=true)
+- rewrite: 若 keep 但措辞生硬,给一个更像人话的等价搜索词(否则原样返回该 query)
+- reason: 一句话理由(≤30字)
+
+严格只输出 JSON 数组,每个候选一项,idx 对应输入序号,不要任何额外文字/解释/markdown 围栏:
+[{"idx":0,"natural":8,"findable":7,"useful":9,"keep":true,"rewrite":"...","reason":"..."}]
+```
+
+- [ ] **Step 2: 提交**
+
+```bash
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/prompts/query_score_system.md
+git commit -m "feat(mode_workflow): 新增 query 评分 system prompt"
+```
+
+---
+
+### Task 2: 评分脚本 `stages/query_score.py`
+
+**Files:**
+- Create: `examples/mode_workflow/stages/query_score.py`
+
+- [ ] **Step 1: 写脚本**
+
+```python
+# -*- coding: utf-8 -*-
+"""Query 正交格评分 · 对 judged_matrix 的 tier≥1 格(动作×类型)在当前维度上下文下用 Sonnet 打分,
+挑出有意义、人话、有助于内容制作知识库目的的组合。结果原子写 .cache/query_score/<sel>.json。
+
+由 server.py /api/query_score 起子进程调;也可独立跑:
+  python stages/query_score.py --tool-type AI --modality 图片 --suffix 怎么做 \
+      --substance-path 表象,实体 --form-path 呈现,视觉 --sel adhoc --dry-run
+"""
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[3]   # …/Agent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from dotenv import load_dotenv
+load_dotenv()
+
+from examples.process_pipeline.script.llm_helper import call_llm_with_retry
+
+HERE = Path(__file__).resolve().parent
+MW = HERE.parent
+MATRIX_FILE = MW / "reference" / "judged_matrix.json"
+PROMPT_FILE = MW / "prompts" / "query_score_system.md"
+CACHE_DIR = MW / ".cache" / "query_score"
+DEFAULT_MODEL = "anthropic/claude-sonnet-4-6"
+BATCH = 40
+CONCURRENCY = 5
+
+
+def _build_cells(matrix, tool_type, modality, suffix):
+    """筛 tier≥1 格,产出 [{a_idx,t_idx,action,type,tier,query}]。
+    query = [工具类型] 动作叶 类型叶 [模态] [后缀],"无"/空跳过。"""
+    actions, types, grid = matrix["actions"], matrix["types"], matrix["matrix"]
+    pre, mod, suf = (tool_type or "").strip(), (modality or "").strip(), (suffix or "").strip()
+    cells = []
+    for ai, arow in enumerate(grid):
+        action = actions[ai]["name"]
+        for ti, cell in enumerate(arow):
+            if cell.get("tier", 0) < 1:
+                continue
+            typ = types[ti]["name"]
+            parts = [p for p in (pre, action, typ, mod, suf) if p and p != "无"]
+            cells.append({"a_idx": ai, "t_idx": ti, "action": action,
+                          "type": typ, "tier": cell["tier"], "query": " ".join(parts)})
+    return cells
+
+
+def _build_user(batch, ctx):
+    lines = [f'{i}. "{c["query"]}"   (动作={c["action"]} 类型={c["type"]} 内容树tier={c["tier"]})'
+             for i, c in enumerate(batch)]
+    sub = (ctx["substance"] or "无").replace(",", "›")
+    form = (ctx["form"] or "无").replace(",", "›")
+    return (f"【固定上下文(本批共享)】\n"
+            f"工具类型: {ctx['tool_type'] or '无'}   模态: {ctx['modality'] or '无'}   后缀: {ctx['suffix'] or '无'}\n"
+            f"(实质/形式不参与拼词，仅供理解领域定位: 实质路径={sub}  形式路径={form})\n\n"
+            f"【候选列表】每条 = 动作 + 类型 + 上下文词拼成的 query:\n" + "\n".join(lines))
+
+
+def _validate(d):
+    return None if isinstance(d, list) else "需 JSON 数组"
+
+
+async def _score_batch(batch, ctx, system, llm_call, model, sem):
+    messages = [{"role": "system", "content": system},
+                {"role": "user", "content": _build_user(batch, ctx)}]
+    async with sem:
+        data, cost = await call_llm_with_retry(
+            llm_call=llm_call, messages=messages, model=model,
+            temperature=0.1, max_tokens=4000, validate_fn=_validate,
+            task_name=f"QueryScore[{batch[0]['query'][:12]}]")
+    out = {}
+    for v in (data or []):
+        if not isinstance(v, dict):
+            continue
+        i = v.get("idx")
+        if not isinstance(i, int) or not (0 <= i < len(batch)):
+            continue
+        c = batch[i]
+        try:
+            score = round(float(v.get("natural")) * 0.4 + float(v.get("findable")) * 0.3
+                          + float(v.get("useful")) * 0.3, 1)
+        except (TypeError, ValueError):
+            score = None
+        out[f"{c['a_idx']}_{c['t_idx']}"] = {
+            "query": c["query"], "natural": v.get("natural"), "findable": v.get("findable"),
+            "useful": v.get("useful"), "keep": bool(v.get("keep")),
+            "rewrite": (v.get("rewrite") or c["query"]), "reason": v.get("reason", ""),
+            "score": score}
+    return out, cost
+
+
+async def run(args):
+    matrix = json.loads(MATRIX_FILE.read_text(encoding="utf-8"))
+    ctx = {"tool_type": args.tool_type, "modality": args.modality, "suffix": args.suffix,
+           "substance": args.substance_path, "form": args.form_path}
+    cells = _build_cells(matrix, args.tool_type, args.modality, args.suffix)
+    if args.limit:
+        cells = cells[:args.limit]
+    print(f"📋 tier≥1 候选 {len(cells)} 格" + (f" (--limit {args.limit})" if args.limit else ""))
+    if args.dry_run:
+        for c in cells[:10]:
+            print(f"  [{c['tier']}] {c['query']}")
+        print(f"…共 {len(cells)} 格(dry-run,未调 LLM)")
+        return 0
+
+    system = PROMPT_FILE.read_text(encoding="utf-8")
+    from agent.llm.openrouter import create_openrouter_llm_call
+    llm_call = create_openrouter_llm_call(model=args.model)
+    sem = asyncio.Semaphore(CONCURRENCY)
+    batches = [cells[i:i + BATCH] for i in range(0, len(cells), BATCH)]
+    print(f"🤖 {len(batches)} 批 × ≤{BATCH} 格 · 并发 {CONCURRENCY} · 模型 {args.model}")
+    results = await asyncio.gather(*[
+        _score_batch(b, ctx, system, llm_call, args.model, sem) for b in batches])
+    merged, cost = {}, 0.0
+    for cmap, c in results:
+        merged.update(cmap)
+        cost += c
+    kept = sum(1 for v in merged.values() if v.get("keep"))
+    out = {"sel": ctx, "model": args.model, "kept": kept, "total": len(merged),
+           "cost_usd": round(cost, 4), "cells": merged}
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    dest = CACHE_DIR / f"{args.sel}.json"
+    tmp = dest.with_suffix(".tmp")
+    tmp.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp.replace(dest)   # 原子落盘,避免前端读到半截
+    print(f"✅ 评分完成 {len(merged)} 格 · keep {kept} · ${cost:.4f} → {dest.name}")
+    return 0
+
+
+def main():
+    p = argparse.ArgumentParser(description="Query 正交格评分(tier≥1 × 当前维度 → Sonnet 打分)")
+    p.add_argument("--tool-type", default="")
+    p.add_argument("--modality", default="")
+    p.add_argument("--suffix", default="")
+    p.add_argument("--substance-path", default="", help="实质祖先路径,逗号分隔(仅作上下文)")
+    p.add_argument("--form-path", default="", help="形式祖先路径,逗号分隔(仅作上下文)")
+    p.add_argument("--model", default=DEFAULT_MODEL)
+    p.add_argument("--sel", default="adhoc", help="缓存文件名(server 传 sel_hash)")
+    p.add_argument("--limit", type=int, default=None, help="只评前 N 格(调试)")
+    p.add_argument("--dry-run", action="store_true", help="只拼词打印,不调 LLM、不落盘")
+    p.add_argument("--force", action="store_true", help="(占位)缓存短路在 server 侧,本脚本恒重算")
+    args = p.parse_args()
+    raise SystemExit(asyncio.run(run(args)))
+
+
+if __name__ == "__main__":
+    main()
+```
+
+- [ ] **Step 2: dry-run 验证拼词与筛选（不花钱）**
+
+Run:
+```bash
+cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow
+python3 stages/query_score.py --tool-type AI --modality 图片 --suffix 怎么做 --dry-run
+```
+Expected: 打印 `📋 tier≥1 候选 643 格`，随后 10 行形如 `[3] AI 检索 提示词 图片 怎么做`，末行 `…共 643 格(dry-run,未调 LLM)`。
+
+- [ ] **Step 3: 小样真评（~$0.01，验证 JSON 结构与落盘）**
+
+Run:
+```bash
+python3 stages/query_score.py --tool-type AI --suffix 怎么做 --sel _smoketest --limit 8
+python3 - <<'PY'
+import json, pathlib
+p = pathlib.Path(".cache/query_score/_smoketest.json")
+d = json.loads(p.read_text(encoding="utf-8"))
+assert d["total"] >= 1 and "cells" in d
+k, v = next(iter(d["cells"].items()))
+assert set(["query","natural","findable","useful","keep","rewrite","reason","score"]) <= set(v)
+assert "_" in k  # 键形如 a_idx_t_idx
+print("✔ 评分结果结构 OK:", d["kept"], "/", d["total"], "keep, $", d["cost_usd"])
+PY
+rm -f .cache/query_score/_smoketest.json
+```
+Expected: `✔ 评分结果结构 OK: ...`
+
+- [ ] **Step 4: 编译 + 提交**
+
+```bash
+python3 -m py_compile stages/query_score.py && echo OK
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/stages/query_score.py
+git commit -m "feat(mode_workflow): 新增 query_score 评分脚本(tier≥1×维度→Sonnet,原子落盘缓存)"
+```
+
+---
+
+### Task 3: server `_matrix()` + `GET /api/query_matrix`
+
+**Files:**
+- Modify: `examples/mode_workflow/server.py`（在 `MATRIX_FILE = ...` 行之后加 helper；在 `do_GET` 的 `elif u.path == "/api/dashboard"` 之前加路由）
+
+- [ ] **Step 1: 加矩阵缓存 helper**
+
+在 `server.py` 的 `MATRIX_FILE = HERE / "reference" / "judged_matrix.json"` 行之后插入：
+
+```python
+_MATRIX_CACHE = None
+
+
+def _matrix():
+    """judged_matrix.json 解析后模块级缓存(只读,进程内不变)。"""
+    global _MATRIX_CACHE
+    if _MATRIX_CACHE is None:
+        _MATRIX_CACHE = json.loads(MATRIX_FILE.read_text(encoding="utf-8"))
+    return _MATRIX_CACHE
+```
+
+- [ ] **Step 2: 加路由**
+
+在 `do_GET` 里 `elif u.path == "/api/dashboard":` 这一行之前插入：
+
+```python
+            elif u.path == "/api/query_matrix":
+                self._json_etag(_matrix())
+```
+
+- [ ] **Step 3: 验证（import 无副作用，直接调 helper）**
+
+Run:
+```bash
+cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow
+python3 -c "import server; m=server._matrix(); print('✔', len(m['actions']),'动作', len(m['types']),'类型', len(m['matrix']),'行')"
+python3 -m py_compile server.py && echo OK
+```
+Expected: `✔ 27 动作 50 类型 27 行` 和 `OK`
+
+- [ ] **Step 4: 提交**
+
+```bash
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/server.py
+git commit -m "feat(mode_workflow): GET /api/query_matrix 返回内容树矩阵(ETag,模块级缓存)"
+```
+
+---
+
+### Task 4: server `_category_tree()` + `GET /api/category_tree`
+
+**Files:**
+- Modify: `examples/mode_workflow/server.py`（顶部加 `import urllib.request, urllib.parse`；`_matrix` 之后加 helper；`do_GET` 加路由）
+
+- [ ] **Step 1: 顶部加 import**
+
+在 `server.py` 现有 import 区（与其它 `import` 同处）加：
+
+```python
+import urllib.request
+import urllib.parse
+```
+
+- [ ] **Step 2: 加 helper（紧跟 `_matrix` 之后）**
+
+```python
+CATEGORY_API = "https://library.aiddit.com/api/pattern/executions/401/category-tree"
+CAT_CACHE_DIR = HERE / ".cache" / "category_tree"
+
+
+def _category_tree(source_type):
+    """拉取并缓存 实质/形式 分类树(library.aiddit.com)。缓存命中直接读盘;
+    上游不可达则抛异常(do_GET 转 500,前端降级)。"""
+    if source_type not in ("实质", "形式"):
+        raise ValueError("source_type 只能是 实质/形式")
+    CAT_CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    cache = CAT_CACHE_DIR / f"{source_type}.json"
+    if cache.is_file():
+        return json.loads(cache.read_text(encoding="utf-8"))
+    url = CATEGORY_API + "?" + urllib.parse.urlencode({"source_type": source_type})
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    with urllib.request.urlopen(req, timeout=20) as resp:
+        data = json.loads(resp.read().decode("utf-8"))
+    tmp = cache.with_suffix(".tmp")
+    tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+    tmp.replace(cache)
+    return data
+```
+
+- [ ] **Step 3: 加路由（`do_GET` 的 `elif u.path == "/api/query_matrix":` 之后）**
+
+```python
+            elif u.path == "/api/category_tree":
+                self._json_etag(_category_tree(qs.get("source_type", "实质")))
+```
+
+- [ ] **Step 4: 验证（真拉一次 + 缓存命中）**
+
+Run:
+```bash
+cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow
+python3 -c "import server; d=server._category_tree('实质'); print('✔ tree 顶层', len(d['tree']), '节点; 首个=', d['tree'][0]['name'])"
+test -f .cache/category_tree/实质.json && echo '✔ 已缓存落盘'
+python3 -m py_compile server.py && echo OK
+```
+Expected: `✔ tree 顶层 N 节点; 首个= 表象`、`✔ 已缓存落盘`、`OK`
+
+- [ ] **Step 5: 提交**
+
+```bash
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/server.py
+git commit -m "feat(mode_workflow): GET /api/category_tree 拉取+缓存 实质/形式 分类树"
+```
+
+---
+
+### Task 5: server `_query_sel_hash()` + `POST/GET /api/query_score`
+
+**Files:**
+- Modify: `examples/mode_workflow/server.py`（`_category_tree` 之后加 helper；`do_GET` 加 GET 路由；`do_POST` 加 POST 路由）
+
+- [ ] **Step 1: 加选择哈希 helper**
+
+紧跟 `_category_tree` 之后插入（`hashlib` 已在 server.py 顶部 import）：
+
+```python
+SCORE_CACHE_DIR = HERE / ".cache" / "query_score"
+
+
+def _query_sel_hash(sel):
+    """对评分选择 + 矩阵指纹算稳定短哈希,作为缓存文件名(同选择不重复付费)。"""
+    matrix_sig = hashlib.md5(MATRIX_FILE.read_bytes()).hexdigest()[:8]
+    canon = json.dumps([sel, matrix_sig], ensure_ascii=False, sort_keys=True)
+    return hashlib.md5(canon.encode("utf-8")).hexdigest()[:16]
+```
+
+- [ ] **Step 2: 加 GET 路由（`do_GET` 的 `elif u.path == "/api/category_tree":` 之后）**
+
+```python
+            elif u.path == "/api/query_score":
+                cache = SCORE_CACHE_DIR / f"{qs.get('sel', '')}.json"
+                if cache.is_file():
+                    self._json_etag(json.loads(cache.read_text(encoding="utf-8")))
+                else:
+                    self._json({"pending": True}, 202)
+```
+
+- [ ] **Step 3: 加 POST 路由（`do_POST` 的 `elif u.path == "/api/run_search":` 之前）**
+
+```python
+            elif u.path == "/api/query_score":
+                sel = {
+                    "tool_type": payload.get("tool_type", ""),
+                    "modality": payload.get("modality", ""),
+                    "suffix": payload.get("suffix", ""),
+                    "substance_path": payload.get("substance_path") or [],
+                    "form_path": payload.get("form_path") or [],
+                    "model": payload.get("model", "anthropic/claude-sonnet-4-6"),
+                }
+                sel_hash = _query_sel_hash(sel)
+                cache = SCORE_CACHE_DIR / f"{sel_hash}.json"
+                if cache.is_file() and not payload.get("force"):
+                    return self._json({"sel": sel_hash, "cached": True})
+                cmd = [sys.executable, "stages/query_score.py", "--sel", sel_hash,
+                       "--tool-type", sel["tool_type"], "--modality", sel["modality"],
+                       "--suffix", sel["suffix"],
+                       "--substance-path", ",".join(sel["substance_path"]),
+                       "--form-path", ",".join(sel["form_path"]),
+                       "--model", sel["model"]]
+                if payload.get("force"):
+                    cmd += ["--force"]
+                self._json({"sel": sel_hash, "task_id": _spawn_task("score", cmd), "cached": False})
+```
+
+- [ ] **Step 4: 验证哈希稳定性 + 编译**
+
+Run:
+```bash
+cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow
+python3 -c "
+import server
+s = {'tool_type':'AI','modality':'图片','suffix':'怎么做','substance_path':['表象'],'form_path':[],'model':'anthropic/claude-sonnet-4-6'}
+h1 = server._query_sel_hash(s); h2 = server._query_sel_hash(dict(s))
+assert h1 == h2 and len(h1) == 16, '哈希应稳定且 16 位'
+s2 = dict(s); s2['modality'] = '视频'
+assert server._query_sel_hash(s2) != h1, '选择变了哈希应变'
+print('✔ sel_hash 稳定且区分:', h1)
+"
+python3 -m py_compile server.py && echo OK
+```
+Expected: `✔ sel_hash 稳定且区分: ...`、`OK`
+
+- [ ] **Step 5: 端到端冒烟（手动重启 server 后 curl）**
+
+> server.py 不自动重载，需先重启：若已在跑，`kill $(cat .cloudflared.pid 2>/dev/null; pgrep -f "python.*server.py")` 后 `python3 server.py &`（或用你平时启动 server 的方式）。仅本步需要 server 在跑。
+
+Run:
+```bash
+cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow
+curl -s "http://localhost:8772/api/query_matrix" | python3 -c "import sys,json; print('matrix actions=', len(json.load(sys.stdin)['actions']))"
+curl -s "http://localhost:8772/api/category_tree?source_type=形式" | python3 -c "import sys,json; print('form tree top=', len(json.load(sys.stdin)['tree']))"
+curl -s -X POST "http://localhost:8772/api/query_score" -H 'Content-Type: application/json' \
+  -d '{"tool_type":"AI","suffix":"怎么做"}' | python3 -c "import sys,json; d=json.load(sys.stdin); print('score POST ->', d)"
+```
+Expected: `matrix actions= 27`、`form tree top= N`、`score POST -> {'sel': ..., 'task_id'/'cached': ...}`。
+若返回 `task_id`，稍候 `curl "http://localhost:8772/api/query_score?sel=<上面的sel>"` 应从 202 `{"pending":true}` 变为完整结果。
+
+- [ ] **Step 6: 提交**
+
+```bash
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/server.py
+git commit -m "feat(mode_workflow): /api/query_score 触发+取结果(选择哈希缓存,命中短路)"
+```
+
+---
+
+### Task 6: 文档（README 结构表 + 接口说明）
+
+**Files:**
+- Modify: `examples/mode_workflow/README.md`
+
+- [ ] **Step 1: 结构表加两行**
+
+在 README「结构」表里 `reference/judged_matrix.json` 行之后加：
+
+```
+| `stages/query_score.py` | Query 正交格评分:tier≥1 格 × 维度上下文 → Sonnet 打分,结果缓存 `.cache/query_score/` |
+| `prompts/query_score_system.md` | 评分 system prompt(可单独迭代) |
+```
+
+- [ ] **Step 2: 提交**
+
+```bash
+cd /Users/max_liu/max_liu/company/Agent
+git add examples/mode_workflow/README.md
+git commit -m "docs(mode_workflow): README 补 query_score 脚本/prompt"
+```
+
+---
+
+## Self-Review
+
+- **Spec coverage（Part A 后端）**：`/api/query_matrix` → Task 3 ✓；`/api/category_tree`（拉+缓存+降级）→ Task 4 ✓；`stages/query_score.py`（tier≥1 筛选、拼词、分批并发、原子落盘、`--dry-run`/`--limit`）→ Task 2 ✓；评分 Prompt（natural/findable/useful/keep/rewrite/reason + JSON 数组）→ Task 1 ✓；综合分 `natural*0.4+findable*0.3+useful*0.3` → Task 2 `_score_batch` ✓；`/api/query_score` POST 触发+缓存短路、GET 取结果、`sel_hash`(选择+matrix_sig) → Task 5 ✓；cells 键 `<a_idx>_<t_idx>` → Task 2 ✓。
+- **Placeholder scan**：无 TBD/TODO；每步含完整代码与可运行验证。`--force` 在脚本里是占位参数（缓存短路由 server 负责），已在 help 注明，非占位逻辑缺口。
+- **Type consistency**：脚本输出 `cells[key]` 字段集与 spec/前端约定一致（query/natural/findable/useful/keep/rewrite/reason/score）；server `_query_sel_hash(sel)` 入参为 dict、`sel` 字段集与脚本 CLI 参数一一对应（tool_type/modality/suffix/substance_path/form_path/model）；GET/POST 同路径 `/api/query_score` 分属 do_GET/do_POST，不冲突。