Domů Procházet Nápověda
Přihlásit se
howard
/
Agent
2
0
Rozštěpit 0
Soubory Úkoly 0 Pull Requesty 0 Wiki
Větev: dev-lxn
Větve Značky
Agent
/
docs
/
superpowers
/
plans
/
2026-06-18-partA1-query-score-backend.md

2026-06-18-partA1-query-score-backend.md 22 KB
Trvalý odkaz Historie Surový

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: 写文件

你是「内容制作知识库」的检索词评审。我们要从小红书/公众号等公共平台搜到
"怎么做某类内容"的工序/工具教程。用户会给你一批由内容树维度正交生成的候选 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: 提交
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: 写脚本

# -*- 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:

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:

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: 编译 + 提交
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" 行之后插入：

_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": 这一行之前插入：

            elif u.path == "/api/query_matrix":
                self._json_etag(_matrix())
  • Step 3: 验证（import 无副作用，直接调 helper）

Run:

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: 提交
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 同处）加：

import urllib.request
import urllib.parse
  • Step 2: 加 helper（紧跟 _matrix 之后）
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": 之后）
            elif u.path == "/api/category_tree":
                self._json_etag(_category_tree(qs.get("source_type", "实质")))
  • Step 4: 验证（真拉一次 + 缓存命中）

Run:

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: 提交
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）：

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": 之后）
            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": 之前）
            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:

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:

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: 提交
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: 提交
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，不冲突。