# mode_workflow 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:** 在 `examples/mode_workflow/` 建一套"搜索评估 + 工序/工具大模型解构"系统:MySQL 三张新表为唯一事实源,单 server(8772) + 单页 HTML(Dashboard / Dataset / 聚类库)。 **Architecture:** DB-first(区别于旧系统文件为主双写)。搜索/解构引擎函数从 `examples/process_pipeline/script/` 只读 import(沿用 `run_search.py` 先例),mode_workflow 只持有编排脚本、prompts、db 层、server、前端。解构由 server 起子进程跑 pipeline 脚本,结果直接写库,前端轮询任务状态。 **Tech Stack:** Python(http.server + pymysql + asyncio)、MySQL(.env `MYSQL_*`)、OpenRouter LLM(经 `call_llm_with_retry`)、单文件 HTML + 原生 JS + ECharts(CDN)。 **Spec:** `docs/superpowers/specs/2026-06-12-mode-workflow-design.md` **前置条件(执行前确认):** - `.env` 含 `MYSQL_HOST/MYSQL_PORT/MYSQL_USER/MYSQL_PASSWORD/MYSQL_DATABASE`(Task 2 起必需) - `.env` 含 `OPEN_ROUTER_API_KEY`(Task 4-6 的冒烟测试需要,会产生 ~$0.01-0.05 真实费用;没有 key 则跳过冒烟,只做代码审查) - `pip install -e .` 已装(`agent.llm.openrouter` 等可 import) **Spec 修正(已核实):** 评估 rubric 固化在 `llm_evaluate_sources` 的 `eval_prompt_template.md`(见 `search_and_evaluate.py:424` 注释),**不需要**复制 `eval_prompt.md` 到 prompts/。prompts/ 只放工序、工具两个解构 prompt。 --- ## 既有代码事实(执行者必读,均已核实) | 事实 | 位置 | |---|---| | MySQL 连接模式:pymysql + DictCursor + autocommit,`_enabled()` 检查 `MYSQL_HOST` | `examples/process_pipeline/script/search_eval/fixed_query_eval/db.py:32-47` | | `search_all(platforms, queries, max_count, max_concurrent, query_overrides=None)` 返回 source 列表(case_id/platform/source_url/post/found_by_queries) | `search_and_evaluate.py:212` | | `evaluate_posts(sources, requirement, llm_call, model, max_concurrent, include_images, max_images, image_mode, query)` 返回 `(sources, total_cost)`,评估挂在 `source["llm_evaluation"]` | `search_and_evaluate.py:403` | | `transcribe_video_posts(sources, concurrency)`、`build_query_overrides(platforms, phrasings, llm, model)`、`_attach_image_refs(sources, max_images, concurrency, mode)` | `search_and_evaluate.py` | | `build_eval_llm_call(choice)` 返回 `(llm_call, model_id)`;`EVAL_MODELS`/`DEFAULT_EVAL_MODEL`;`_format_post_for_eval(source)`;`_evaluate_one` | `examples/process_pipeline/script/llm_evaluate_sources.py` | | `call_llm_with_retry(llm_call, messages, model, temperature, max_tokens, validate_fn, task_name)` 返回 `(data_dict, cost)`,只认 JSON 对象 | `examples/process_pipeline/script/llm_helper.py` | | 任意 OpenRouter 模型:`from agent.llm.openrouter import create_openrouter_llm_call; llm_call = create_openrouter_llm_call(model=...)` | `tool_extract.py:121-122` | | 工具解构返回 `{"tools":[{工具名称,实质作用域,形式作用域,创作层级,来源链接,输入,输出,用法,案例,缺点,最新更新时间}]}` | `fixed_query_eval/prompts/tool_extract_system.md` | | 工序解构返回 `{"source":{...},"procedures":[{id,name,purpose,category,declarations,type_registry,steps:[{id,kind,via,effect,action,substance,form,directive,intent,inputs,outputs}]}]}` | `mode_procedure/mode-dsl/prompts/procedure_extract_system.md` | | 工序解构的图片走 base64(绕防盗链),`_detect_image_mime`/`_fetch_data_url`/`_collect_images`/`_sanitize_workflow` 可整段复制 | `mode_procedure/mode-dsl/procedure_model_extract.py:34-126` | | judged_matrix:`{actions:[{name,l1,l2}×27], types:[{name,l1,l2}×50], matrix:[27行×50格,每格{tier,r,q}], stats:{valid:643}}`,有效节点 = tier≥1 | `search_eval/evaluation/judged_matrix.json` | | 历史搜索数据:`fixed_query_eval/runs_full/q0000-q0003/form_A.json`,顶层 `{form,query,original_q,platforms,total,failed,results:[...]}` | 同目录 | | 仓库无 pytest 体系(CLAUDE.md),验证用独立命令/冒烟脚本 | `CLAUDE.md` | 路径深度:`examples/mode_workflow/x.py` 的 `PROJECT_ROOT = Path(__file__).resolve().parents[2]`;`examples/mode_workflow/pipeline/x.py` 是 `parents[3]`。 --- ### Task 1: 脚手架与资产同步 **Files:** - Create: `examples/mode_workflow/.gitignore` - Create: `examples/mode_workflow/prompts/`(拷贝 2 个 prompt) - Create: `examples/mode_workflow/reference/judged_matrix.json`(拷贝) - Create: `examples/mode_workflow/pipeline/`、`examples/mode_workflow/runs/` 目录 - [ ] **Step 1: 建目录并同步资产** ```bash cd /Users/max_liu/max_liu/company/Agent SE=examples/process_pipeline/script/search_eval mkdir -p examples/mode_workflow/{pipeline,prompts,reference,runs} cp "$SE/mode_procedure/mode-dsl/prompts/procedure_extract_system.md" examples/mode_workflow/prompts/ cp "$SE/fixed_query_eval/prompts/tool_extract_system.md" examples/mode_workflow/prompts/ cp "$SE/evaluation/judged_matrix.json" examples/mode_workflow/reference/ ``` - [ ] **Step 2: 写 .gitignore** `examples/mode_workflow/.gitignore`: ```gitignore runs/ __pycache__/ *.pyc ``` - [ ] **Step 3: 验证** ```bash ls examples/mode_workflow/prompts/ examples/mode_workflow/reference/ python3 -c "import json; d=json.load(open('examples/mode_workflow/reference/judged_matrix.json')); print(d['stats'])" ``` Expected: 两个 prompt 文件名 + judged_matrix.json;stats 行含 `'valid': 643`。 - [ ] **Step 4: Commit** ```bash git add examples/mode_workflow git commit -m "feat(mode_workflow): 脚手架与 prompt/内容树资产同步" ``` --- ### Task 2: db.py — 三表 DDL + 读写层 **Files:** - Create: `examples/mode_workflow/db.py` **设计要点:** - 风格沿用 `fixed_query_eval/db.py`(pymysql/DictCursor/autocommit),但本系统 **DB 是主存储**:写失败要 `raise`(由调用方决定怎么处理),不再静默吞掉。读侧保留防御。 - `mode_process`/`mode_tools` 写入语义:删掉 `(case_id, version)` 同版本旧行再插(同版本重跑幂等,跨版本保留历史)。 - `cost_usd`/`duration_s` 是**每次解构调用**的值,同一 `(case_id, version)` 的多行重复存同一个值;聚合时必须按 `(case_id, version)` 去重(见 Task 7)。 - [ ] **Step 1: 写 `examples/mode_workflow/db.py`(完整文件)** ```python # -*- coding: utf-8 -*- """mode_workflow · MySQL 持久化(DB 为唯一事实源) ================================================================================ 读 .env 的 MYSQL_* 连接 MySQL。三张表: search_data —— 每行一个 (query, 帖子):搜索 + llm 评估结果 mode_process —— 每行一个解构出的工序(steps 等嵌套结构存 JSON 列) mode_tools —— 每行一个解构出的工具 与旧 fixed_query_eval/db.py 的关键差异:本系统 DB 是主存储,写入失败直接 raise, 不做"失败不阻断"。读侧保留防御(返回空/None)。 用法: python db.py init # 建表(幂等) python db.py check # 打印三表行数 """ import json import os import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from dotenv import load_dotenv load_dotenv() import pymysql from pymysql.cursors import DictCursor def _conn(): if not os.getenv("MYSQL_HOST"): raise RuntimeError("缺 MYSQL_HOST:检查 .env 的 MYSQL_* 配置") return pymysql.connect( host=os.getenv("MYSQL_HOST"), port=int(os.getenv("MYSQL_PORT", 3306)), user=os.getenv("MYSQL_USER"), password=os.getenv("MYSQL_PASSWORD"), database=os.getenv("MYSQL_DATABASE"), charset="utf8mb4", cursorclass=DictCursor, autocommit=True, connect_timeout=10, ) # ── DDL ────────────────────────────────────────────────────────────────────── DDL_SEARCH = """ CREATE TABLE IF NOT EXISTS search_data ( id BIGINT AUTO_INCREMENT PRIMARY KEY, query_id VARCHAR(32) NOT NULL COMMENT 'q0000', query_text VARCHAR(512) NULL, case_id VARCHAR(128) NOT NULL COMMENT 'platform_channelContentId', platform VARCHAR(32) NULL, channel_content_id VARCHAR(128) NULL, title VARCHAR(512) NULL, url VARCHAR(1024) NULL, content_type VARCHAR(32) NULL, body LONGTEXT NULL, images JSON NULL, videos JSON NULL, like_count INT NULL, publish_time VARCHAR(64) NULL, quality_score FLOAT NULL COMMENT 'post._quality_score', quality_grade VARCHAR(8) NULL, found_by JSON NULL COMMENT '命中的措辞数组', knowledge_type JSON NULL COMMENT '["能力","工序","工具"] 子集', overall_score FLOAT NULL COMMENT '(相关均值+质量均值)/2', llm_evaluation JSON NULL COMMENT '评估全量 blob', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_qid_case (query_id, case_id), KEY idx_platform (platform) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='搜索+评估结果'; """ DDL_PROCESS = """ CREATE TABLE IF NOT EXISTS mode_process ( id BIGINT AUTO_INCREMENT PRIMARY KEY, query_id VARCHAR(32) NOT NULL, case_id VARCHAR(128) NOT NULL, platform VARCHAR(32) NULL, post_title VARCHAR(512) NULL, source JSON NULL COMMENT '解构返回的 source 块', procedure_id VARCHAR(16) NULL COMMENT 'p1,p2…', name VARCHAR(255) NULL, purpose TEXT NULL, category VARCHAR(32) NULL COMMENT '产物创造/资产建设/自动化/分析/学习', declarations JSON NULL, type_registry JSON NULL, steps JSON NULL COMMENT '步骤数组全量', step_count INT NULL, tools_used JSON NULL COMMENT '从 steps[].via 去重提取', model VARCHAR(64) NULL, version VARCHAR(16) NULL COMMENT 'v_MMDDHHMM,保留历史', cost_usd DECIMAL(10,6) NULL COMMENT '本次解构调用成本(同版本各行相同,聚合需按 case+version 去重)', duration_s FLOAT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, KEY idx_case_ver (case_id, version), KEY idx_qid (query_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构结果(每行一个工序)'; """ DDL_TOOLS = """ CREATE TABLE IF NOT EXISTS mode_tools ( id BIGINT AUTO_INCREMENT PRIMARY KEY, query_id VARCHAR(32) NOT NULL, case_id VARCHAR(128) NOT NULL, platform VARCHAR(32) NULL, post_title VARCHAR(512) NULL, tool_name VARCHAR(255) NULL, substance_scope JSON NULL COMMENT '实质作用域(数组)', form_scope JSON NULL COMMENT '形式作用域(数组或null)', creation_layer VARCHAR(32) NULL COMMENT '制作层/创作层', source_link VARCHAR(1024) NULL, input_desc TEXT NULL, output_desc TEXT NULL, usage_json JSON NULL, cases_json JSON NULL, defects_json JSON NULL, updated_time VARCHAR(64) NULL COMMENT '工具最新更新时间', model VARCHAR(64) NULL, version VARCHAR(16) NULL, cost_usd DECIMAL(10,6) NULL COMMENT '同 mode_process,聚合按 case+version 去重', duration_s FLOAT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, KEY idx_case_ver (case_id, version), KEY idx_qid (query_id), KEY idx_tool_name (tool_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工具解构结果(每行一个工具)'; """ def init_tables(): conn = _conn() try: with conn.cursor() as cur: cur.execute(DDL_SEARCH) cur.execute(DDL_PROCESS) cur.execute(DDL_TOOLS) print("✅ 建表完成:search_data, mode_process, mode_tools") finally: conn.close() # ── 工具函数 ────────────────────────────────────────────────────────────────── def _loads(v, default=None): """pymysql 的 JSON 列可能返回字符串,统一解析。""" if v is None: return default if isinstance(v, (list, dict)): return v try: return json.loads(v) except Exception: return default def _j(v): """写入 JSON 列:None 保持 NULL,其余 dumps。""" return None if v is None else json.dumps(v, ensure_ascii=False) def _collect_scores(node): """递归收集嵌套评估里所有数值「得分」。""" out = [] if isinstance(node, dict): for k, v in node.items(): if k == "得分" and isinstance(v, (int, float)): out.append(float(v)) else: out.extend(_collect_scores(v)) elif isinstance(node, list): for v in node: out.extend(_collect_scores(v)) return out def overall_score(e): """综合分 = (相关性各项均值 + 质量各项均值) / 可得部分数。算不出返回 None。""" parts = [] for key in ("相关性", "质量"): scores = _collect_scores((e or {}).get(key)) if scores: parts.append(sum(scores) / len(scores)) return round(sum(parts) / len(parts), 2) if parts else None # ── search_data ────────────────────────────────────────────────────────────── def upsert_search_posts(query_id, query_text, results): """一组搜索结果写入 search_data(按 (query_id, case_id) upsert)。返回写入条数。""" if not results: return 0 rows = [] for r in results: post = r.get("post") or {} e = r.get("llm_evaluation") or {} rows.append(( query_id, query_text, r.get("case_id"), r.get("platform"), r.get("channel_content_id"), (post.get("title") or post.get("desc") or "")[:500], r.get("source_url"), post.get("content_type"), post.get("body_text") or post.get("desc") or "", _j(post.get("images") or []), _j(post.get("videos") or []), post.get("like_count"), str(post.get("publish_time") or post.get("publish_timestamp") or "")[:64], post.get("_quality_score"), post.get("_quality_grade"), _j(r.get("found_by_queries") or []), _j(e.get("知识类型") or []), overall_score(e), _j(e), )) sql = """ INSERT INTO search_data (query_id, query_text, case_id, platform, channel_content_id, title, url, content_type, body, images, videos, like_count, publish_time, quality_score, quality_grade, found_by, knowledge_type, overall_score, llm_evaluation) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE query_text=VALUES(query_text), platform=VALUES(platform), channel_content_id=VALUES(channel_content_id), title=VALUES(title), url=VALUES(url), content_type=VALUES(content_type), body=VALUES(body), images=VALUES(images), videos=VALUES(videos), like_count=VALUES(like_count), publish_time=VALUES(publish_time), quality_score=VALUES(quality_score), quality_grade=VALUES(quality_grade), found_by=VALUES(found_by), knowledge_type=VALUES(knowledge_type), overall_score=VALUES(overall_score), llm_evaluation=VALUES(llm_evaluation); """ conn = _conn() try: with conn.cursor() as cur: cur.executemany(sql, rows) return len(rows) finally: conn.close() def fetch_queries(): """query 列表 + 帖子数 + 解构进度。""" conn = _conn() try: with conn.cursor() as cur: cur.execute("""SELECT query_id, MAX(query_text) AS query_text, COUNT(*) AS post_count FROM search_data GROUP BY query_id ORDER BY query_id""") queries = cur.fetchall() cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM mode_process GROUP BY query_id") np = {r["query_id"]: r["n"] for r in cur.fetchall()} cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM mode_tools GROUP BY query_id") nt = {r["query_id"]: r["n"] for r in cur.fetchall()} finally: conn.close() for q in queries: q["process_done"] = np.get(q["query_id"], 0) q["tools_done"] = nt.get(q["query_id"], 0) return queries def fetch_posts(query_id): """某 query 下全部帖子(JSON 列已解析),带 has_process/has_tools 标记。""" conn = _conn() try: with conn.cursor() as cur: cur.execute("""SELECT * FROM search_data WHERE query_id=%s ORDER BY overall_score DESC, id""", (query_id,)) rows = cur.fetchall() cur.execute("SELECT DISTINCT case_id FROM mode_process WHERE query_id=%s", (query_id,)) hp = {r["case_id"] for r in cur.fetchall()} cur.execute("SELECT DISTINCT case_id FROM mode_tools WHERE query_id=%s", (query_id,)) ht = {r["case_id"] for r in cur.fetchall()} finally: conn.close() for r in rows: for col in ("images", "videos", "found_by", "knowledge_type", "llm_evaluation"): r[col] = _loads(r[col]) r["has_process"] = r["case_id"] in hp r["has_tools"] = r["case_id"] in ht r.pop("created_at", None); r.pop("updated_at", None) return rows def fetch_post(query_id, case_id): """单帖完整行(给 pipeline 脚本重建 source 用)。无则 None。""" conn = _conn() try: with conn.cursor() as cur: cur.execute("SELECT * FROM search_data WHERE query_id=%s AND case_id=%s", (query_id, case_id)) row = cur.fetchone() finally: conn.close() if not row: return None for col in ("images", "videos", "found_by", "knowledge_type", "llm_evaluation"): row[col] = _loads(row[col]) return row # ── mode_process ───────────────────────────────────────────────────────────── def replace_process(query_id, case_id, platform, post_title, payload, model, version, cost_usd, duration_s): """写入一帖某版本的工序解构结果(payload = {source, procedures})。 删 (case_id, version) 旧行再插,同版本重跑幂等、跨版本保留历史。返回工序条数。""" source = payload.get("source") procedures = payload.get("procedures") or [] conn = _conn() try: with conn.cursor() as cur: cur.execute("DELETE FROM mode_process WHERE case_id=%s AND version=%s", (case_id, version)) if procedures: rows = [] for p in procedures: steps = p.get("steps") or [] vias = [] for s in steps: v = s.get("via") if v and v not in vias: vias.append(v) rows.append(( query_id, case_id, platform, (post_title or "")[:500], _j(source), p.get("id"), (p.get("name") or "")[:250], p.get("purpose"), p.get("category"), _j(p.get("declarations")), _j(p.get("type_registry")), _j(steps), len(steps), _j(vias), model, version, cost_usd, duration_s, )) cur.executemany(""" INSERT INTO mode_process (query_id, case_id, platform, post_title, source, procedure_id, name, purpose, category, declarations, type_registry, steps, step_count, tools_used, model, version, cost_usd, duration_s) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, rows) return len(procedures) finally: conn.close() def fetch_process_versions(case_id): conn = _conn() try: with conn.cursor() as cur: cur.execute("""SELECT version, COUNT(*) AS n, MAX(model) AS model FROM mode_process WHERE case_id=%s GROUP BY version ORDER BY version DESC""", (case_id,)) return cur.fetchall() finally: conn.close() def fetch_process(case_id, version=None): """重建 {case_id, version, model, source, procedures:[...]}。version=None 取最新。""" conn = _conn() try: with conn.cursor() as cur: if version is None: cur.execute("""SELECT version FROM mode_process WHERE case_id=%s ORDER BY version DESC, id DESC LIMIT 1""", (case_id,)) row = cur.fetchone() if not row: return None version = row["version"] cur.execute("""SELECT * FROM mode_process WHERE case_id=%s AND version=%s ORDER BY id""", (case_id, version)) rows = cur.fetchall() finally: conn.close() if not rows: return None procedures = [{ "id": r["procedure_id"], "name": r["name"], "purpose": r["purpose"], "category": r["category"], "declarations": _loads(r["declarations"]), "type_registry": _loads(r["type_registry"]), "steps": _loads(r["steps"], []), "tools_used": _loads(r["tools_used"], []), } for r in rows] return {"case_id": case_id, "version": version, "platform": rows[0]["platform"], "title": rows[0]["post_title"], "model": rows[0]["model"], "cost_usd": float(rows[0]["cost_usd"]) if rows[0]["cost_usd"] is not None else None, "duration_s": rows[0]["duration_s"], "source": _loads(rows[0]["source"]), "procedures": procedures} # ── mode_tools ─────────────────────────────────────────────────────────────── def replace_tools(query_id, case_id, platform, post_title, tools, model, version, cost_usd, duration_s): """写入一帖某版本的工具解构结果。语义同 replace_process。返回工具条数。""" conn = _conn() try: with conn.cursor() as cur: cur.execute("DELETE FROM mode_tools WHERE case_id=%s AND version=%s", (case_id, version)) if tools: rows = [( query_id, case_id, platform, (post_title or "")[:500], (t.get("工具名称") or "")[:250], _j(t.get("实质作用域")), _j(t.get("形式作用域")), t.get("创作层级"), t.get("来源链接"), t.get("输入"), t.get("输出"), _j(t.get("用法")), _j(t.get("案例")), _j(t.get("缺点")), t.get("最新更新时间"), model, version, cost_usd, duration_s, ) for t in tools] cur.executemany(""" INSERT INTO mode_tools (query_id, case_id, platform, post_title, tool_name, substance_scope, form_scope, creation_layer, source_link, input_desc, output_desc, usage_json, cases_json, defects_json, updated_time, model, version, cost_usd, duration_s) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, rows) return len(tools) finally: conn.close() def fetch_tools_versions(case_id): conn = _conn() try: with conn.cursor() as cur: cur.execute("""SELECT version, COUNT(*) AS n, MAX(model) AS model FROM mode_tools WHERE case_id=%s GROUP BY version ORDER BY version DESC""", (case_id,)) return cur.fetchall() finally: conn.close() def fetch_tools(case_id, version=None): """重建 {case_id, version, model, tool_count, tools:[...]}。version=None 取最新。""" conn = _conn() try: with conn.cursor() as cur: if version is None: cur.execute("""SELECT version FROM mode_tools WHERE case_id=%s ORDER BY version DESC, id DESC LIMIT 1""", (case_id,)) row = cur.fetchone() if not row: return None version = row["version"] cur.execute("""SELECT * FROM mode_tools WHERE case_id=%s AND version=%s ORDER BY id""", (case_id, version)) rows = cur.fetchall() finally: conn.close() if not rows: return None tools = [{ "工具名称": r["tool_name"], "实质作用域": _loads(r["substance_scope"]), "形式作用域": _loads(r["form_scope"]), "创作层级": r["creation_layer"], "来源链接": r["source_link"], "输入": r["input_desc"], "输出": r["output_desc"], "用法": _loads(r["usage_json"]), "案例": _loads(r["cases_json"]), "缺点": _loads(r["defects_json"]), "最新更新时间": r["updated_time"], } for r in rows] return {"case_id": case_id, "version": version, "platform": rows[0]["platform"], "title": rows[0]["post_title"], "model": rows[0]["model"], "cost_usd": float(rows[0]["cost_usd"]) if rows[0]["cost_usd"] is not None else None, "duration_s": rows[0]["duration_s"], "tool_count": len(tools), "tools": tools} # ── Dashboard 原始行(指标计算在 server.py)───────────────────────────────────── def fetch_dashboard_rows(): """拉 Dashboard 计算所需的轻量行。数据量级:百~千行,Python 聚合足够。""" conn = _conn() try: with conn.cursor() as cur: cur.execute("SELECT query_id, case_id, knowledge_type FROM search_data") posts = cur.fetchall() cur.execute("""SELECT case_id, version, steps, tools_used, cost_usd, duration_s, created_at FROM mode_process""") procs = cur.fetchall() cur.execute("""SELECT case_id, version, tool_name, substance_scope, form_scope, cost_usd, duration_s, created_at FROM mode_tools""") tools = cur.fetchall() finally: conn.close() for p in posts: p["knowledge_type"] = _loads(p["knowledge_type"], []) for r in procs: r["steps"] = _loads(r["steps"], []) r["tools_used"] = _loads(r["tools_used"], []) r["cost_usd"] = float(r["cost_usd"]) if r["cost_usd"] is not None else None r["created_at"] = str(r["created_at"]) if r["created_at"] else None for r in tools: r["substance_scope"] = _loads(r["substance_scope"], []) r["form_scope"] = _loads(r["form_scope"], []) r["cost_usd"] = float(r["cost_usd"]) if r["cost_usd"] is not None else None r["created_at"] = str(r["created_at"]) if r["created_at"] else None return posts, procs, tools def check(): conn = _conn() try: with conn.cursor() as cur: for t in ("search_data", "mode_process", "mode_tools"): cur.execute(f"SELECT COUNT(*) AS n FROM {t}") print(f"{t}: {cur.fetchone()['n']} 行") finally: conn.close() if __name__ == "__main__": cmd = sys.argv[1] if len(sys.argv) > 1 else "" if cmd == "init": init_tables() elif cmd == "check": check() else: print("用法:\n python db.py init # 建表\n python db.py check # 三表行数") ``` - [ ] **Step 2: 建表并验证** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python db.py init && python db.py check ``` Expected: `✅ 建表完成:search_data, mode_process, mode_tools`,三表各 0 行。失败提示缺 `MYSQL_HOST` 则确认 `.env`。 - [ ] **Step 3: 单元自检(overall_score)** ```bash python3 -c " import db e = {'相关性': {'和内容制作知识相关': {'得分': 8}, '和 query 相关': {'得分': 6}}, '质量': {'固定维度': {'热度性': {'得分': 4}}}} s = db.overall_score(e) assert s == round((7.0 + 4.0) / 2, 2), s assert db.overall_score({}) is None print('overall_score OK:', s) " ``` Expected: `overall_score OK: 5.5` - [ ] **Step 4: Commit** ```bash git add examples/mode_workflow/db.py git commit -m "feat(mode_workflow): 三表 DDL 与 MySQL 读写层" ``` --- ### Task 3: import_history.py — 历史搜索结果入库 **Files:** - Create: `examples/mode_workflow/import_history.py` - [ ] **Step 1: 写完整文件** ```python # -*- coding: utf-8 -*- """一次性导入:fixed_query_eval/runs_full/*/form_A.json → search_data 表。 幂等(upsert),可反复执行。 用法: python import_history.py python import_history.py --runs-dir /path/to/runs_full """ import argparse import json import sys from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import db DEFAULT_RUNS = (HERE.parent / "process_pipeline" / "script" / "search_eval" / "fixed_query_eval" / "runs_full") def main(): p = argparse.ArgumentParser(description="历史搜索结果导入 search_data") p.add_argument("--runs-dir", default=str(DEFAULT_RUNS)) args = p.parse_args() runs = Path(args.runs_dir) files = sorted(runs.glob("q*/form_A.json")) if not files: print(f"❌ {runs} 下没有 q*/form_A.json"); return 1 total = 0 for f in files: data = json.loads(f.read_text(encoding="utf-8")) qid = f.parent.name results = data.get("results", []) n = db.upsert_search_posts(qid, data.get("query") or data.get("original_q"), results) print(f" {qid}: 文件 {len(results)} 条 → 入库 {n} 条") total += n print(f"✅ 共导入 {total} 条 → search_data") return 0 if __name__ == "__main__": raise SystemExit(main()) ``` - [ ] **Step 2: 执行并核对行数** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python import_history.py python db.py check python3 -c " import json, glob n = sum(len(json.load(open(f))['results']) for f in glob.glob('../process_pipeline/script/search_eval/fixed_query_eval/runs_full/q*/form_A.json')) print('JSON 总条数:', n) " ``` Expected: 每个 q 打一行;`search_data` 行数 == JSON 总条数(q0000-q0003)。再跑一次 `python import_history.py`,行数不变(幂等)。 - [ ] **Step 3: Commit** ```bash git add examples/mode_workflow/import_history.py git commit -m "feat(mode_workflow): 历史搜索结果导入脚本" ``` --- ### Task 4: pipeline/tool_extract.py — 工具解构(读库→LLM→写库) **Files:** - Create: `examples/mode_workflow/pipeline/tool_extract.py` **与旧版差异:** 帖子源从 `search_data` 表读(不再读 form_A.json);结果直接写 `mode_tools`(不再写本地 json 为主),`runs/tools/` 只留调试副本;记录 cost/duration。 - [ ] **Step 1: 写完整文件** ```python # -*- coding: utf-8 -*- """工具解构 · search_data 帖子 → 结构化工具条目 → mode_tools 表 ================================================================================ - 帖子源:search_data 表(--query-id + --case-ids 定位) - 模型默认 google/gemini-3.1-flash-lite,可 --model 传任意 OpenRouter id - 多模态:复用 search_and_evaluate._attach_image_refs(URL 直传) - 写库:db.replace_tools(同版本幂等,跨版本保留);runs/tools/ 留调试副本 用法(一般由 server.py 起子进程调): python pipeline/tool_extract.py --query-id q0000 --case-ids xhs_abc,gzh_def python pipeline/tool_extract.py --query-id q0000 --case-ids xhs_abc --model anthropic/claude-sonnet-4-6 """ import argparse import asyncio import json import sys import time from datetime import datetime 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.search_eval.search_and_evaluate import _attach_image_refs from examples.process_pipeline.script.llm_evaluate_sources import _format_post_for_eval, build_eval_llm_call from examples.process_pipeline.script.llm_helper import call_llm_with_retry HERE = Path(__file__).resolve().parent # pipeline/ MW = HERE.parent # mode_workflow/ sys.path.insert(0, str(MW)) import db TOOL_SYSTEM = (MW / "prompts" / "tool_extract_system.md").read_text(encoding="utf-8") DEFAULT_MODEL_CHOICE = "gemini-flash-lite" MAX_IMAGES = 6 def _row_to_source(row): """search_data 行 → 引擎函数认的 source dict。""" return { "case_id": row["case_id"], "platform": row["platform"], "channel_content_id": row["channel_content_id"], "source_url": row["url"], "post": { "title": row["title"], "body_text": row["body"], "images": row["images"] or [], "like_count": row["like_count"], "publish_timestamp": row["publish_time"], "link": row["url"], }, } def _validate_tools(data): if not isinstance(data, dict) or "tools" not in data: return '缺少顶层 "tools" 字段' if not isinstance(data["tools"], list): return '"tools" 必须是数组' return None async def extract_one(source, llm_call, model): """对一条 source 抽工具,返回 (tools_list, cost)。失败返回 ([], cost)。""" user_text = "【内容】\n" + _format_post_for_eval(source) image_urls = source.get("_image_data_urls") or None if image_urls: user_content = [{"type": "text", "text": user_text}] for u in image_urls: user_content.append({"type": "image_url", "image_url": {"url": u}}) messages = [{"role": "system", "content": TOOL_SYSTEM}, {"role": "user", "content": user_content}] else: messages = [{"role": "system", "content": TOOL_SYSTEM}, {"role": "user", "content": user_text}] data, cost = await call_llm_with_retry( llm_call=llm_call, messages=messages, model=model, temperature=0.1, max_tokens=4000, validate_fn=_validate_tools, task_name=f"ToolExtract[{source.get('case_id', '?')}]", ) if not data: return [], cost return data.get("tools", []), cost async def run(args): qid = args.query_id case_ids = [c.strip() for c in args.case_ids.split(",") if c.strip()] sources = [] for cid in case_ids: row = db.fetch_post(qid, cid) if row is None: print(f"⚠️ {qid}/{cid} 不在 search_data,跳过") continue sources.append(_row_to_source(row)) if not sources: print("❌ 没有可解构的帖子"); return 1 if args.model and "/" in args.model: from agent.llm.openrouter import create_openrouter_llm_call llm_call, model_id = create_openrouter_llm_call(model=args.model), args.model else: llm_call, model_id = build_eval_llm_call(args.model or DEFAULT_MODEL_CHOICE) version = args.version or ("v_" + datetime.now().strftime("%m%d%H%M")) print(f"🔧 工具解构 {len(sources)} 帖 · 模型 {model_id} · 版本 {version}") await _attach_image_refs(sources, MAX_IMAGES, max(2, args.max_concurrent * 2), "url") sem = asyncio.Semaphore(args.max_concurrent) out_dir = MW / "runs" / "tools" out_dir.mkdir(parents=True, exist_ok=True) async def _work(s): t0 = time.monotonic() async with sem: tools, cost = await extract_one(s, llm_call, model_id) dur = round(time.monotonic() - t0, 1) n = db.replace_tools(qid, s["case_id"], s.get("platform"), (s.get("post") or {}).get("title", ""), tools, model_id, version, cost, dur) (out_dir / f"{s['case_id']}_{version}.json").write_text(json.dumps({ "case_id": s["case_id"], "version": version, "model": model_id, "cost_usd": cost, "duration_s": dur, "tools": tools, }, ensure_ascii=False, indent=2), encoding="utf-8") print(f" ✅ {s['case_id']} → {n} 个工具 · ${cost:.4f} · {dur}s") return cost costs = await asyncio.gather(*[_work(s) for s in sources]) print(f"\n📊 完成 {len(sources)} 帖 · 总成本 ${sum(costs):.4f}") return 0 def main(): p = argparse.ArgumentParser(description="工具解构:search_data 帖子 → mode_tools") p.add_argument("--query-id", required=True) p.add_argument("--case-ids", required=True, help="逗号分隔 case_id 列表") p.add_argument("--model", default=None, help="默认 gemini-flash-lite,可传 OpenRouter id") p.add_argument("--max-concurrent", type=int, default=3) p.add_argument("--version", default=None, help="默认自动 v_月日时分") args = p.parse_args() raise SystemExit(asyncio.run(run(args))) if __name__ == "__main__": main() ``` - [ ] **Step 2: 冒烟(需 OPEN_ROUTER_API_KEY;无 key 跳过本步)** 先从库里挑一个真实 case_id: ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow CID=$(python3 -c "import db; print(db.fetch_posts('q0000')[0]['case_id'])") python pipeline/tool_extract.py --query-id q0000 --case-ids "$CID" python3 -c "import db; r=db.fetch_tools('$CID'); print(r['version'], r['tool_count'], r['cost_usd'], r['duration_s'])" ``` Expected: 解构日志含 `✅ → N 个工具`;fetch_tools 打出版本号、工具数 ≥0、cost/duration 非 None。 - [ ] **Step 3: Commit** ```bash git add examples/mode_workflow/pipeline/tool_extract.py git commit -m "feat(mode_workflow): 工具解构 pipeline(读库→LLM→mode_tools)" ``` --- ### Task 5: pipeline/procedure_extract.py — 工序解构 **Files:** - Create: `examples/mode_workflow/pipeline/procedure_extract.py` **与旧版差异:** 帖子源从库读;**不渲染 HTML**(前端直接渲染 DB 数据);写 `mode_process`;支持一次多帖。`_detect_image_mime`/`_fetch_data_url`/`_collect_images`/`_sanitize_workflow` 从 `procedure_model_extract.py:34-126` 原样复制(已验证可用,不重新发明)。 - [ ] **Step 1: 写完整文件** ```python # -*- coding: utf-8 -*- """工序解构 · search_data 帖子 → workflow JSON → mode_process 表 ================================================================================ 单次大模型直出(无 agent / 无 validate 循环),prompt 见 prompts/procedure_extract_system.md。 配图下载转 base64(绕防盗链)随文本一起发。结果按工序拆行写 mode_process。 用法(一般由 server.py 起子进程调): python pipeline/procedure_extract.py --query-id q0000 --case-ids xhs_abc python pipeline/procedure_extract.py --query-id q0000 --case-ids xhs_abc --model google/gemini-3.1-flash-lite """ import argparse import asyncio import base64 import json import sys import time from datetime import datetime 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 sys.path.insert(0, str(MW)) import db PROMPT_FILE = MW / "prompts" / "procedure_extract_system.md" DEFAULT_MODEL = "anthropic/claude-sonnet-4-6" MAX_IMAGES = 8 # ── 以下 4 个助手原样取自 mode_procedure/mode-dsl/procedure_model_extract.py ── def _detect_image_mime(data: bytes): if not data or len(data) < 12: return None if data[:3] == b"\xff\xd8\xff": return "image/jpeg" if data[:8] == b"\x89PNG\r\n\x1a\n": return "image/png" if data[:4] == b"RIFF" and data[8:12] == b"WEBP": return "image/webp" if data[:6] in (b"GIF87a", b"GIF89a"): return "image/gif" return None async def _fetch_data_url(url, sem): from agent.tools.builtin.file.image_cdn import _download_image async with sem: try: data = await _download_image(url) except Exception: return None mime = _detect_image_mime(data) if mime is None: return None return f"data:{mime};base64,{base64.b64encode(data).decode()}" async def _collect_images(urls, max_images, concurrency): urls = [u for u in urls if isinstance(u, str) and u][:max_images] if not urls: return [] sem = asyncio.Semaphore(concurrency) results = await asyncio.gather(*[_fetch_data_url(u, sem) for u in urls]) return [d for d in results if d] def _validate_wf(data): if not isinstance(data, dict): return "顶层必须是 JSON 对象" if "procedures" not in data: return '缺少 "procedures" 字段' if not isinstance(data["procedures"], list): return '"procedures" 必须是数组' return None def _sanitize_workflow(data): dropped = {"procedures": 0, "steps": 0, "io": 0} procs = data.get("procedures") if not isinstance(procs, list): return data, dropped clean_procs = [] for p in procs: if not isinstance(p, dict): dropped["procedures"] += 1 continue steps = p.get("steps") if isinstance(steps, list): kept = [] for s in steps: if not isinstance(s, dict): dropped["steps"] += 1 continue for io in ("inputs", "outputs"): if isinstance(s.get(io), list): before = len(s[io]) s[io] = [x for x in s[io] if isinstance(x, dict)] dropped["io"] += before - len(s[io]) kept.append(s) p["steps"] = kept if not isinstance(p.get("declarations"), dict): p.pop("declarations", None) if not isinstance(p.get("type_registry"), dict): p.pop("type_registry", None) clean_procs.append(p) data["procedures"] = clean_procs return data, dropped # ── 助手复制结束 ────────────────────────────────────────────────────────────── async def extract_one(row, system, llm_call, model, args): """单帖工序解构 → 写 mode_process。返回 cost。""" cid = row["case_id"] t0 = time.monotonic() post_text = (f"【标题】{row['title'] or ''}\n【来源】{row['url'] or ''}\n" f"【正文】\n{row['body'] or ''}") data_urls = [] if args.no_images else await _collect_images( row["images"] or [], args.max_images, args.max_concurrent) print(f"🖼️ {cid} 配图 {len(data_urls)}/{len(row['images'] or [])} 张") if data_urls: user_content = [{"type": "text", "text": post_text}] for u in data_urls: user_content.append({"type": "image_url", "image_url": {"url": u}}) messages = [{"role": "system", "content": system}, {"role": "user", "content": user_content}] else: messages = [{"role": "system", "content": system}, {"role": "user", "content": post_text}] data, cost = await call_llm_with_retry( llm_call=llm_call, messages=messages, model=model, temperature=0.2, max_tokens=args.max_tokens, validate_fn=_validate_wf, task_name=f"ProcExtract[{cid}]", ) if not data: print(f"❌ {cid} 解构失败(重试耗尽)") return cost data, dropped = _sanitize_workflow(data) if any(dropped.values()): print(f"🧹 {cid} 清洗:丢弃 procedure {dropped['procedures']} / " f"step {dropped['steps']} / io {dropped['io']}") dur = round(time.monotonic() - t0, 1) n = db.replace_process(args.query_id, cid, row["platform"], row["title"], data, model, args.version, cost, dur) out_dir = MW / "runs" / "procedures" out_dir.mkdir(parents=True, exist_ok=True) (out_dir / f"{cid}_{args.version}.json").write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") print(f" ✅ {cid} → {n} 个工序 · ${cost:.4f} · {dur}s") return cost async def run(args): case_ids = [c.strip() for c in args.case_ids.split(",") if c.strip()] rows = [] for cid in case_ids: row = db.fetch_post(args.query_id, cid) if row is None: print(f"⚠️ {args.query_id}/{cid} 不在 search_data,跳过") continue rows.append(row) if not rows: print("❌ 没有可解构的帖子"); return 1 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) args.version = args.version or ("v_" + datetime.now().strftime("%m%d%H%M")) print(f"🤖 工序解构 {len(rows)} 帖 · 模型 {args.model} · 版本 {args.version}") costs = [] for row in rows: # 工序解构 token 重,串行跑,避免 OpenRouter 限流 costs.append(await extract_one(row, system, llm_call, args.model, args)) print(f"\n📊 完成 {len(rows)} 帖 · 总成本 ${sum(costs):.4f}") return 0 def main(): p = argparse.ArgumentParser(description="工序解构:search_data 帖子 → mode_process") p.add_argument("--query-id", required=True) p.add_argument("--case-ids", required=True, help="逗号分隔 case_id 列表") p.add_argument("--model", default=DEFAULT_MODEL) p.add_argument("--version", default=None, help="默认自动 v_月日时分") p.add_argument("--max-images", type=int, default=MAX_IMAGES) p.add_argument("--max-concurrent", type=int, default=4) p.add_argument("--max-tokens", type=int, default=8000) p.add_argument("--no-images", action="store_true") args = p.parse_args() raise SystemExit(asyncio.run(run(args))) if __name__ == "__main__": main() ``` - [ ] **Step 2: 冒烟(需 API key;工序解构默认模型是 claude-sonnet,单帖约 $0.03-0.1;省钱可加 `--model google/gemini-3.1-flash-lite`)** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow CID=$(python3 -c "import db; print(db.fetch_posts('q0000')[0]['case_id'])") python pipeline/procedure_extract.py --query-id q0000 --case-ids "$CID" --model google/gemini-3.1-flash-lite python3 -c " import db r = db.fetch_process('$CID') print(r['version'], len(r['procedures']), '个工序') p = r['procedures'][0] print('steps:', len(p['steps']), 'tools_used:', p['tools_used']) " ``` Expected: `✅ → N 个工序`;fetch_process 返回 version、工序数 ≥1、steps 非空、tools_used 列出 via。 - [ ] **Step 3: Commit** ```bash git add examples/mode_workflow/pipeline/procedure_extract.py git commit -m "feat(mode_workflow): 工序解构 pipeline(读库→LLM→mode_process)" ``` --- ### Task 6: pipeline/search_eval.py — 搜索 + 评估入库 **Files:** - Create: `examples/mode_workflow/pipeline/search_eval.py` **与 run_search.py 的差异:** 不写死 4 组 query,接受任意 `--query`(可带 `--synonyms` 同义措辞);产出直接入库 `search_data`,`runs/search/` 留调试副本。 - [ ] **Step 1: 写完整文件** ```python # -*- coding: utf-8 -*- """搜索 + 评估 · 任意 query → 多渠道搜索去重 → LLM 逐帖评估 → search_data 表 ================================================================================ 引擎函数全部只读复用 search_and_evaluate.py(搜索/去重/转写/评估/英平台翻译)。 用法(一般由 server.py 起子进程调): python pipeline/search_eval.py --query-id q0004 --query "AI 人像 图片 生成 怎么做" python pipeline/search_eval.py --query-id q0005 --query "GPT image2 评测" \ --synonyms "GPT image2 测评,GPT image2 实测" --platforms xhs,gzh --max-count 10 """ 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.search_eval.search_and_evaluate import ( search_all, evaluate_posts, transcribe_video_posts, build_query_overrides, ) from examples.process_pipeline.script.llm_evaluate_sources import ( build_eval_llm_call, EVAL_MODELS, DEFAULT_EVAL_MODEL, ) HERE = Path(__file__).resolve().parent MW = HERE.parent sys.path.insert(0, str(MW)) import db async def run(args): phrasings = [args.query] + [s.strip() for s in (args.synonyms or "").split(",") if s.strip()] # 去重保序 seen, uniq = set(), [] for q in phrasings: if q not in seen: seen.add(q); uniq.append(q) phrasings = uniq platforms = [p.strip() for p in args.platforms.split(",") if p.strip()] eval_llm, eval_model_id = build_eval_llm_call(args.eval_model) print(f"▶ {args.query_id} query={args.query!r} 措辞={phrasings} 渠道={platforms}") overrides = await build_query_overrides(platforms, phrasings, eval_llm, eval_model_id) sources = await search_all(platforms, phrasings, args.max_count, args.max_concurrent, query_overrides=overrides) print(f"🔎 去重后 {len(sources)} 帖") if not sources: print("❌ 搜索无结果"); return 1 try: from examples.process_pipeline.script.extract_sources import _convert_timestamps _convert_timestamps(sources) except Exception: pass if not args.no_transcribe: n = await transcribe_video_posts(sources, concurrency=args.max_concurrent) if n: print(f"🎙️ 视频转写 {n} 条") cost = 0.0 if not args.no_eval: sources, cost = await evaluate_posts( sources, "", eval_llm, eval_model_id, args.max_concurrent, include_images=not args.no_images, max_images=args.max_images, image_mode=args.image_mode, query=args.query, ) for s in sources: s.pop("_image_data_urls", None) n = db.upsert_search_posts(args.query_id, args.query, sources) print(f"🗄️ search_data 入库 {n} 行 · 评估成本 ${cost:.4f}") out_dir = MW / "runs" / "search" out_dir.mkdir(parents=True, exist_ok=True) (out_dir / f"{args.query_id}.json").write_text(json.dumps({ "query_id": args.query_id, "query": args.query, "phrasings": phrasings, "platforms": platforms, "total": len(sources), "results": sources, }, ensure_ascii=False, indent=2), encoding="utf-8") return 0 def main(): p = argparse.ArgumentParser(description="搜索+评估 → search_data") p.add_argument("--query-id", required=True, help="如 q0004(server 自动分配)") p.add_argument("--query", required=True, help="基准 query(评估锚点)") p.add_argument("--synonyms", default="", help="逗号分隔的同义措辞(可选)") p.add_argument("--platforms", default="xhs,gzh") p.add_argument("--max-count", type=int, default=10) p.add_argument("--eval-model", default=DEFAULT_EVAL_MODEL, choices=list(EVAL_MODELS)) p.add_argument("--max-concurrent", type=int, default=3) p.add_argument("--max-images", type=int, default=4) p.add_argument("--image-mode", choices=["url", "base64"], default="url") p.add_argument("--no-transcribe", action="store_true") p.add_argument("--no-eval", action="store_true") p.add_argument("--no-images", action="store_true") args = p.parse_args() raise SystemExit(asyncio.run(run(args))) if __name__ == "__main__": main() ``` - [ ] **Step 2: 冒烟(需 API key + 搜索通道可用;成本与时长都高,可只验 import 与参数解析)** 最小验证(不花钱): ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python pipeline/search_eval.py --help ``` Expected: 正常打出 usage(证明 import 链 OK——这条会触发 agent.tools 注册链,import 错误在这暴露)。 完整冒烟(可选,~1-3 分钟、~$0.05): ```bash python pipeline/search_eval.py --query-id q9999 --query "GPT image2 评测" --platforms xhs --max-count 3 python3 -c "import db; print(len(db.fetch_posts('q9999')), '帖')" ``` Expected: 入库 ≥1 行。验完可清理:`python3 -c "import db; c=db._conn(); c.cursor().execute(\"DELETE FROM search_data WHERE query_id='q9999'\"); c.close()"` - [ ] **Step 3: Commit** ```bash git add examples/mode_workflow/pipeline/search_eval.py git commit -m "feat(mode_workflow): 搜索评估 pipeline(任意 query → search_data)" ``` --- ### Task 7: server.py — API + 任务管理 + Dashboard 聚合 **Files:** - Create: `examples/mode_workflow/server.py` **API 契约(index.html 依赖,Task 8 不得偏离):** | Method+Path | 入参 | 返回 | |---|---|---| | GET `/` | - | index.html | | GET `/api/dashboard` | - | 见下方 `_dashboard()` 返回结构 | | GET `/api/queries` | - | `[{query_id, query_text, post_count, process_done, tools_done}]` | | GET `/api/posts` | `query_id` | `[{...search_data 行, has_process, has_tools}]` | | GET `/api/process_versions` | `case_id` | `[{version, n, model}]` | | GET `/api/process` | `case_id`, `version?` | `{case_id, version, model, source, procedures:[...]}` 或 404 | | GET `/api/tools_versions` | `case_id` | `[{version, n, model}]` | | GET `/api/tools` | `case_id`, `version?` | `{case_id, version, model, tool_count, tools:[...]}` 或 404 | | POST `/api/extract_process` | `{query_id, case_ids:[..], model?}` | `{task_id}` | | POST `/api/extract_tools` | `{query_id, case_ids:[..], model?}` | `{task_id}` | | POST `/api/run_search` | `{query, synonyms?, platforms?, max_count?}` | `{task_id, query_id}`(query_id 自动分配下一个 qNNNN) | | GET `/api/task_status` | `task_id` | `{status: running\|done\|failed, log_tail}` | - [ ] **Step 1: 写完整文件** ```python # -*- coding: utf-8 -*- """mode_workflow server · 页面 + API + 解构任务管理 ================================================================================ 单服务(默认 8772): - GET / index.html - GET /api/dashboard Dashboard 全部聚合指标(含内容树覆盖) - GET /api/queries|posts|process|tools(+_versions) Dataset 数据 - POST /api/run_search|extract_process|extract_tools 起子进程跑 pipeline - GET /api/task_status 轮询任务状态(读日志尾部) 用法:python server.py [port] """ import json import subprocess import sys import threading from collections import Counter from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse, parse_qs try: sys.stdout.reconfigure(encoding="utf-8") except Exception: pass HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import db PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8772 MATRIX_FILE = HERE / "reference" / "judged_matrix.json" LOG_DIR = HERE / "runs" / "logs" # ── 任务管理:task_id → {proc, log, status} ────────────────────────────────── TASKS = {} _TASK_LOCK = threading.Lock() def _spawn_task(kind, cmd): LOG_DIR.mkdir(parents=True, exist_ok=True) task_id = f"{kind}_{datetime.now().strftime('%m%d%H%M%S%f')}" log_path = LOG_DIR / f"{task_id}.log" f = open(log_path, "w", encoding="utf-8") proc = subprocess.Popen(cmd, stdout=f, stderr=subprocess.STDOUT, cwd=str(HERE), text=True) with _TASK_LOCK: TASKS[task_id] = {"proc": proc, "log": log_path, "status": "running"} def _wait(): rc = proc.wait() f.close() with _TASK_LOCK: TASKS[task_id]["status"] = "done" if rc == 0 else "failed" threading.Thread(target=_wait, daemon=True).start() return task_id def _task_status(task_id): with _TASK_LOCK: t = TASKS.get(task_id) if not t: return None tail = "" try: text = t["log"].read_text(encoding="utf-8", errors="replace") tail = text[-3000:] except Exception: pass return {"status": t["status"], "log_tail": tail} def _next_query_id(): qs = [q["query_id"] for q in db.fetch_queries()] nums = [int(q[1:]) for q in qs if q.startswith("q") and q[1:].isdigit()] return f"q{(max(nums) + 1 if nums else 0):04d}" # ── Dashboard 聚合 ──────────────────────────────────────────────────────────── def _split_values(v): """substance/form 字段:数组直接用;字符串按 、,/ 分割;None 丢弃。""" out = [] items = v if isinstance(v, list) else [v] for it in items: if not it or not isinstance(it, str): continue for piece in it.replace(",", "、").replace("/", "、").split("、"): piece = piece.strip() if piece: out.append(piece) return out def _dashboard(): posts, procs, tools = db.fetch_dashboard_rows() # 最新版本行集(覆盖度/Top10 用最新版,成本/耗时按全部版本累计) def latest(rows): best = {} for r in rows: cid = r["case_id"] if cid not in best or (r["version"] or "") > (best[cid][0] or ""): best[cid] = (r["version"], []) out = [] for r in rows: if r["version"] == best[r["case_id"]][0]: out.append(r) return out latest_procs = latest(procs) latest_tools = latest(tools) # 内容树覆盖:steps 的 (action 叶子 × 输入/输出 type) ∩ 有效节点(tier≥1) jm = json.loads(MATRIX_FILE.read_text(encoding="utf-8")) a_idx = {a["name"]: i for i, a in enumerate(jm["actions"])} t_idx = {t["name"]: i for i, t in enumerate(jm["types"])} valid = set() for ai, row in enumerate(jm["matrix"]): for ti, cell in enumerate(row): if isinstance(cell, dict) and cell.get("tier", 0) >= 1: valid.add((ai, ti)) covered = set() via_counter = Counter() substance_counter = Counter() form_counter = Counter() for r in latest_procs: for s in r["steps"]: leaf = (s.get("action") or "").split("/")[-1].strip() types = [] for io in ("inputs", "outputs"): for x in s.get(io) or []: if isinstance(x, dict) and x.get("type"): types.append(str(x["type"]).strip()) if leaf in a_idx: for tp in types: if tp in t_idx and (a_idx[leaf], t_idx[tp]) in valid: covered.add((a_idx[leaf], t_idx[tp])) via = (s.get("via") or "").strip() if via: via_counter[via] += 1 for v in _split_values(s.get("substance")): substance_counter[v] += 1 for v in _split_values(s.get("form")): form_counter[v] += 1 for r in latest_tools: for v in _split_values(r["substance_scope"]): substance_counter[v] += 1 for v in _split_values(r["form_scope"]): form_counter[v] += 1 # 成本/耗时:同一 (case_id, version) 只计一次(各行重复存同一次调用的值) def cost_groups(rows): g = {} for r in rows: key = (r["case_id"], r["version"]) if key not in g and r["cost_usd"] is not None: g[key] = (r["cost_usd"], r["duration_s"] or 0.0, r["created_at"]) return list(g.values()) runs = cost_groups(procs) + cost_groups(tools) total_cost = round(sum(c for c, _, _ in runs), 4) total_dur = round(sum(d for _, d, _ in runs), 1) # 按日成本趋势 daily = Counter() for c, _, ts in runs: if ts: daily[ts[:10]] += c cost_trend = [{"date": d, "cost": round(v, 4)} for d, v in sorted(daily.items())] # 进度:分母 = knowledge_type 含对应类型的帖子(distinct case) proc_targets = {p["case_id"] for p in posts if "工序" in (p["knowledge_type"] or [])} tool_targets = {p["case_id"] for p in posts if "工具" in (p["knowledge_type"] or [])} proc_done = {r["case_id"] for r in procs} tool_done = {r["case_id"] for r in tools} return { "result": { "matrix_covered": len(covered), "matrix_valid": len(valid), "matrix_cells": sorted([ai, ti] for ai, ti in covered), "matrix_actions": [a["name"] for a in jm["actions"]], "matrix_types": [t["name"] for t in jm["types"]], "substance_count": len(substance_counter), "substance_top": substance_counter.most_common(15), "form_count": len(form_counter), "form_top": form_counter.most_common(15), "post_count": len(posts), "extracted_post_count": len(proc_done | tool_done), "tool_count": len({r["tool_name"] for r in latest_tools if r["tool_name"]}), "via_top10": via_counter.most_common(10), }, "process_data": { "run_count": len(runs), "avg_cost": round(total_cost / len(runs), 4) if runs else 0, "total_cost": total_cost, "avg_duration": round(total_dur / len(runs), 1) if runs else 0, "total_duration": total_dur, "cost_trend": cost_trend, "process_progress": {"done": len(proc_done), "total": len(proc_targets)}, "tools_progress": {"done": len(tool_done), "total": len(tool_targets)}, }, } # ── HTTP handler ───────────────────────────────────────────────────────────── class Handler(BaseHTTPRequestHandler): def _json(self, data, code=200): body = json.dumps(data, ensure_ascii=False, default=str).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _err(self, msg, code=400): self._json({"error": msg}, code) def do_GET(self): u = urlparse(self.path) qs = {k: v[0] for k, v in parse_qs(u.query).items()} try: if u.path == "/" or u.path == "/index.html": body = (HERE / "index.html").read_bytes() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) elif u.path == "/api/dashboard": self._json(_dashboard()) elif u.path == "/api/queries": self._json(db.fetch_queries()) elif u.path == "/api/posts": self._json(db.fetch_posts(qs.get("query_id", ""))) elif u.path == "/api/process_versions": self._json(db.fetch_process_versions(qs.get("case_id", ""))) elif u.path == "/api/process": r = db.fetch_process(qs.get("case_id", ""), qs.get("version")) self._json(r) if r else self._err("无解构记录", 404) elif u.path == "/api/tools_versions": self._json(db.fetch_tools_versions(qs.get("case_id", ""))) elif u.path == "/api/tools": r = db.fetch_tools(qs.get("case_id", ""), qs.get("version")) self._json(r) if r else self._err("无解构记录", 404) elif u.path == "/api/task_status": r = _task_status(qs.get("task_id", "")) self._json(r) if r else self._err("未知 task_id", 404) else: self._err("not found", 404) except Exception as e: self._err(f"{type(e).__name__}: {e}", 500) def do_POST(self): u = urlparse(self.path) try: n = int(self.headers.get("Content-Length") or 0) payload = json.loads(self.rfile.read(n) or b"{}") except Exception: return self._err("body 必须是 JSON") try: if u.path in ("/api/extract_process", "/api/extract_tools"): qid = payload.get("query_id") cids = payload.get("case_ids") or [] if not qid or not cids: return self._err("缺 query_id / case_ids") script = ("pipeline/procedure_extract.py" if u.path.endswith("process") else "pipeline/tool_extract.py") cmd = [sys.executable, script, "--query-id", qid, "--case-ids", ",".join(cids)] if payload.get("model"): cmd += ["--model", payload["model"]] kind = "proc" if u.path.endswith("process") else "tool" self._json({"task_id": _spawn_task(kind, cmd)}) elif u.path == "/api/run_search": query = (payload.get("query") or "").strip() if not query: return self._err("缺 query") qid = payload.get("query_id") or _next_query_id() cmd = [sys.executable, "pipeline/search_eval.py", "--query-id", qid, "--query", query] if payload.get("synonyms"): cmd += ["--synonyms", payload["synonyms"]] if payload.get("platforms"): cmd += ["--platforms", payload["platforms"]] if payload.get("max_count"): cmd += ["--max-count", str(payload["max_count"])] self._json({"task_id": _spawn_task("search", cmd), "query_id": qid}) else: self._err("not found", 404) except Exception as e: self._err(f"{type(e).__name__}: {e}", 500) def log_message(self, fmt, *a): pass # 静默访问日志 if __name__ == "__main__": print(f"🚀 mode_workflow server → http://0.0.0.0:{PORT}") ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever() ``` - [ ] **Step 2: 启动 + 逐端点验证** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python server.py 8772 & sleep 1 curl -s localhost:8772/api/queries | python3 -m json.tool | head -20 curl -s localhost:8772/api/dashboard | python3 -c " import json,sys d = json.load(sys.stdin) print('采集帖子:', d['result']['post_count']) print('内容树:', d['result']['matrix_covered'], '/', d['result']['matrix_valid']) print('进度:', d['process_data']) " curl -s "localhost:8772/api/posts?query_id=q0000" | python3 -c "import json,sys; r=json.load(sys.stdin); print(len(r),'帖, 首帖:', r[0]['case_id'], r[0]['has_process'], r[0]['has_tools'])" curl -s "localhost:8772/api/process?case_id=不存在" ; echo kill %1 ``` Expected: queries 返回 q0000-q0003;dashboard 的 post_count>0、matrix_valid==643;posts 返回该 query 帖子;不存在的 case 返回 `{"error": "无解构记录"}` 404。若 Task 4/5 冒烟跑过,对应 case 的 process/tools 端点也应返回数据。 - [ ] **Step 3: 任务接口验证(不花钱,用 --help 不行,改用故意缺参验证错误路径 + 真实任务可选)** ```bash curl -s -X POST localhost:8772/api/extract_tools -d '{}' ; echo ``` Expected: `{"error": "缺 query_id / case_ids"}`。(真实解构任务在 Task 9 端到端验收时从 UI 触发。) - [ ] **Step 4: Commit** ```bash git add examples/mode_workflow/server.py git commit -m "feat(mode_workflow): server(API+任务管理+Dashboard聚合)" ``` --- ### Task 8: index.html — 三 tab 前端 **Files:** - Create: `examples/mode_workflow/index.html` **实现方式:** 本任务**必须先invoke `frontend-design:frontend-design` 技能**再写代码——产出单文件 HTML(原生 JS,ECharts 5 CDN:`https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js`),禁止引入构建链。API 契约见 Task 7 表格,字段名不得偏离。 **视觉方向(延续用户截图的既有风格):** 白底卡片 + 深色(藏青/石板)表头多色分区表格;标签 pill(类型彩色底);需求区深蓝表头、输入区橙黄表头、输出区绿表头的分区配色;推断字段用「推」角标 + 高亮底色,hover 显示 inferred_reason。整体配中文界面。 **页面结构与行为(验收依据):** 1. **顶部导航**:Dashboard / Dataset / 聚类库 三 tab(hash 路由 `#dashboard` `#dataset` `#cluster`,刷新保位)。 2. **Dashboard tab**(数据源 `GET /api/dashboard`): - 第一行数字卡:采集帖子数量、解构帖子数量、工具数量、内容树覆盖(`covered/valid` + 百分比)。 - 第二行数字卡(过程数据):累计解构成本(`$total_cost`,副标题单条均值 `$avg_cost`)、总耗时(副标题平均耗时)、工序进度环(`done/total`)、工具进度环。 - 图表区: - 内容树覆盖热力图:ECharts heatmap,x=50 types,y=27 actions,covered cell 高亮(数据来自 `matrix_cells`/`matrix_actions`/`matrix_types`)。 - 工序提及工具 Top10:横向条形图(`via_top10`)。 - 实质覆盖分布:条形图(`substance_top`,标题带总数 `substance_count`)。 - 形式覆盖分布:条形图(`form_top` + `form_count`)。 - 成本趋势:折线图(`cost_trend`)。 - 全部指标为 0/空时不报错,显示空态文案。 3. **Dataset tab**:顶部「工序 / 工具」子切换 + 「新建搜索」按钮;三栏: - 左栏 query 列表(`/api/queries`):query_text + 帖子数 + 解构进度(`process_done` 或 `tools_done` 随子模式切换);点击选中。 - 中栏帖子列表(`/api/posts?query_id=`):卡片含标题、平台徽标(xhs/gzh/zhihu/x)、overall_score、knowledge_type pill、已解构标记(has_process/has_tools);卡片上「解构」按钮 + 多选批量解构;点击卡片在右栏加载。 - 右栏解构结果: - 工序模式:头部「大模型工序:已提取」+ 版本下拉(`/api/process_versions`,格式 `v_xxx (最新) · N工序`)+「重新生成」按钮;每个工序渲染:工序名 + 目的 + 类别 pill + 平台/作者;输入/返回声明区;步骤明细表(分区表头:需求[#/目的/作用/实质/形式] 深蓝、输入[类型/值/来源] 橙、实现[外部工具/动作/指令] 绿、输出[类型/值/去处] 深绿),`inferred:true` 的单元格高亮 +「推」角标,hover title 显示 inferred_reason。 - 工具模式:版本下拉 + 重新生成;每个工具一张卡:工具名、创作层级 pill、实质/形式作用域 pill 组、输入/输出、用法列表、案例(输入/输出/效果)、缺点列表、来源链接。 - 无解构记录(404)时显示空态 +「开始解构」按钮。 - 解构/搜索触发后:弹出任务面板,轮询 `/api/task_status`(2s 间隔),显示 log_tail,done 后自动刷新当前栏;failed 显示日志。 - 「新建搜索」:弹窗输 query/synonyms/platforms/max_count → POST `/api/run_search` → 任务面板,完成后左栏刷新出现新 query。 4. **聚类库 tab**:居中空态占位(图标 + 「聚类库 · 敬请期待」)。 **最小骨架(代码结构基线,frontend-design 在此骨架上完成完整实现):** ```html mode_workflow
``` - [ ] **Step 1: invoke frontend-design 技能,按上述规格产出完整 index.html** - [ ] **Step 2: 浏览器验收** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python server.py 8772 & ``` 打开 `http://localhost:8772`,核对上面"页面结构与行为"逐条:三 tab 切换、Dashboard 全部卡片与 5 张图渲染、Dataset 三栏联动、版本下拉、空态、聚类库占位。若 Task 4/5 冒烟跑过,选对应帖子确认工序表格分区配色与「推」高亮、工具卡片完整。 - [ ] **Step 3: Commit** ```bash git add examples/mode_workflow/index.html git commit -m "feat(mode_workflow): 三tab单页前端(Dashboard/Dataset/聚类库)" ``` --- ### Task 9: README + 端到端验收 **Files:** - Create: `examples/mode_workflow/README.md` - [ ] **Step 1: 写 README** ```markdown # mode_workflow · 搜索评估 + 工序/工具解构工作台 MySQL 三表(search_data / mode_process / mode_tools)为唯一事实源的单页工作台: Dashboard(结果/过程指标可视化)、Dataset(query → 帖子 → 工序/工具解构)、聚类库(占位)。 设计文档:`docs/superpowers/specs/2026-06-12-mode-workflow-design.md` ## 启动 ```bash # 0. 前置:.env 配 MYSQL_* 与 OPEN_ROUTER_API_KEY;pip install -e . python db.py init # 建三张表(幂等) python import_history.py # (可选)导入 fixed_query_eval 历史搜索结果 python server.py # http://localhost:8772 ``` ## 结构 | 文件 | 职责 | |---|---| | `db.py` | 三表 DDL + 全部读写(读 .env MYSQL_*) | | `server.py` | 页面 + API + 解构任务子进程管理(端口 8772) | | `index.html` | 单文件前端:Dashboard / Dataset / 聚类库 | | `pipeline/search_eval.py` | 任意 query 搜索+评估 → search_data | | `pipeline/procedure_extract.py` | 工序解构(LLM 直出)→ mode_process | | `pipeline/tool_extract.py` | 工具解构 → mode_tools | | `prompts/` | 工序/工具解构 system prompt(可单独迭代) | | `reference/judged_matrix.json` | 内容树(27 动作×50 类型),Dashboard 覆盖度用 | | `runs/` | 运行日志与调试副本(gitignore) | 搜索/评估/转写引擎函数只读复用 `examples/process_pipeline/script/search_eval/search_and_evaluate.py`,本目录不复制引擎代码。 ## 与旧 search_eval 的关系 取代 `fixed_query_eval`(8770)+ `mode_procedure`(8771)两套服务的"搜索评估 + 大模型解构" 部分;procedure-dsl 执行引擎、mode-dsl 模式提取、A/B/C 三形式对比未迁移(按设计裁剪)。 ``` - [ ] **Step 2: 端到端验收(需 API key)** ```bash cd /Users/max_liu/max_liu/company/Agent/examples/mode_workflow python server.py 8772 & ``` 浏览器走完整链路: 1. Dataset → 选 q0000 → 选 1 帖 → 触发工具解构 → 任务面板出日志 → 完成后右栏出工具卡。 2. 同帖触发工序解构(模型选 `google/gemini-3.1-flash-lite` 控成本)→ 右栏出工序分区表格。 3. 回 Dashboard → 解构帖子数量/工具数量/Top10/成本/进度环全部联动更新。 4. 新建搜索(query 任意,platforms=xhs,max_count=3)→ 完成后左栏出现新 query。 - [ ] **Step 3: 最终 Commit** ```bash git add examples/mode_workflow/README.md git commit -m "docs(mode_workflow): README 与启动说明" ``` - [ ] **Step 4: 收尾检查** ```bash git status examples/mode_workflow # runs/ 不应出现在未跟踪列表(被 .gitignore 覆盖) ``` --- ## Self-Review 记录 - **Spec 覆盖**:三表(Task 2)、历史导入(Task 3)、三条 pipeline(Task 4-6)、API+Dashboard 口径(Task 7)、三 tab 前端+截图复刻(Task 8)、聚类库占位(Task 8)、README(Task 9)——spec 全部章节有对应任务。spec 中 `prompts/eval_prompt.md` 一项经核实不需要(rubric 固化在引擎内),已在计划头部标注修正。 - **占位符扫描**:无 TBD/TODO;Task 8 的 HTML 完整实现委托 frontend-design 技能,但 API 契约、页面行为、骨架代码、验收清单均已写死,不属于开放占位。 - **类型/命名一致性**:`replace_process(payload)` 与 `fetch_process()` 的 procedures 字段对称;server 调 `db.fetch_*` 名称与 Task 2 定义一致;pipeline 脚本参数名与 server `_spawn_task` 构造的 cmd 一致(`--query-id/--case-ids/--model`);`_dashboard()` 返回字段与 Task 8 前端读取字段一致(`matrix_cells/via_top10/substance_top/cost_trend/process_progress` 等)。