Agent/examples/process_pipeline/script/search_eval/procedure-dsl/plan_tool.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""plan_tool.py — procedure-dsl 的「计划」内置工具 (技能本地, 仅 Cyber/run_cyber 引擎)。

为什么是工具而不是"写 understanding.md":
  弱模型读了"先想清楚工序"的 prompt 也照样跳过、直接堆骨架。把这一步做成**工具**,
  LLM 必须**调用一次**把【工序分解 + 每工序步骤逐条展开 + 每工序覆盖原文哪些章节】
  作为结构化参数交上来 —— 工具当场校验完整性 (混合门禁), 通过就**自动据计划生成
  workflow.json 骨架**, 结果回灌进对话锚定后续填充。于是:
    · understanding 重新独立成一步 (但由工具承载, 不靠自觉)
    · 工序由计划显式决定 (章节认领是声明式, 不再靠词汇模糊匹配)
    · workflow.json 结构严格按计划生成 (工序数/步骤数/顺序锁定, LLM 只填值)

设计边界 (遵守"runner 零业务知识"): 本模块是**技能本地**, 不进 agent/tools/builtin/,
run_cyber 仅 `import plan_tool` 触发 @tool 注册 + 调 set_plan_context 注入原文/路径。
Claude SDK 路 (run_procedure_dsl) 没有 repo 工具系统, 那条路 understanding 仍走 prompt。
"""
# ⚠ 不能用 `from __future__ import annotations`: 它把类型注解变成字符串, 而框架
#   SchemaGenerator 靠运行时 get_origin() 内省真实类型 — 拿字符串会把 procedures 这种
#   List[Dict] 退化成 "string", LLM 就不知道该传数组了。
import json
import re
from pathlib import Path
from typing import Any, Dict, List, Optional

from agent.tools import tool, ToolResult, ToolContext


# ===========================================================================
# 运行时上下文 (run_cyber.main() 在跑 runner 前注入; 工具闭包读它, 避免让 LLM 传路径)
# ===========================================================================

_PLAN_CTX: Dict[str, Any] = {}


def set_plan_context(*, body_text: str = "", out_dir: Optional[Path] = None,
                     case_id: Any = None, source: Optional[dict] = None,
                     ocr: str = "") -> None:
    """run_cyber 在执行前调用, 把原文正文 / 配图 OCR / 输出目录 / source 元信息交给工具。"""
    _PLAN_CTX.clear()
    _PLAN_CTX.update({
        "body_text": body_text or "",
        "ocr": ocr or "",
        "out_dir": Path(out_dir) if out_dir else None,
        "case_id": case_id,
        "source": source or {},
    })


# ===========================================================================
# 原文章节解析 (与 lint-case.py 同口径: 行首 0N| 标号, 排除 "图 0N|" 配图说明)
# ===========================================================================

_SEC_RE = re.compile(r'(?m)^\s*(0\d)\s*[|｜]')


def _source_sections(body: str) -> List[tuple]:
    """返回 [(章节号, 标题)]，如 [('01','从一个案例开始'), ('02','结构化 Prompt 框架'), ...]。"""
    marks = [(m.start(), m.group(1)) for m in _SEC_RE.finditer(body or "")]
    out: List[tuple] = []
    for i, (pos, num) in enumerate(marks):
        end = marks[i + 1][0] if i + 1 < len(marks) else len(body)
        seg = body[pos:end]
        after = re.split(r'[|｜]', seg, 1)
        tail = after[-1] if len(after) > 1 else seg
        title = ""
        for ln in tail.splitlines():
            ln = ln.strip()
            if ln:
                title = ln[:24]
                break
        out.append((num, title))
    return out


def _norm_sec(s: Any) -> str:
    """把 '1' / '01' / '第1章' 之类归一成两位章节号 '01'。无数字返回原串。"""
    m = re.search(r'\d+', str(s))
    return f"0{int(m.group())}"[-2:] if m and int(m.group()) < 10 else (m.group() if m else str(s).strip())


# ===========================================================================
# 入参容错 (弱模型常把 list 传成 JSON 字符串 / 包一层 {"procedures": [...]})
# ===========================================================================

def _coerce_procedures(procedures: Any) -> List[dict]:
    if isinstance(procedures, str):
        try:
            procedures = json.loads(procedures)
        except Exception:
            return []
    if isinstance(procedures, dict):
        procedures = procedures.get("procedures", procedures.get("list", []))
    return procedures if isinstance(procedures, list) else []


# ===========================================================================
# 据计划生成 workflow.json 骨架
# ===========================================================================

def _build_skeleton(summary: str, procedures: List[dict]) -> dict:
    src = _PLAN_CTX.get("source") or {}
    skeleton: Dict[str, Any] = {
        "source": {
            "platform": src.get("platform", ""),
            "author": src.get("author", ""),
            "date": src.get("date", ""),
            "url": src.get("url", ""),
            "title": src.get("title", ""),
            "excerpt": src.get("excerpt", "") or (summary or "")[:120],
        },
        "procedures": [],
    }
    for pi, proc in enumerate(procedures, 1):
        pid = f"p{pi}"
        steps_out: List[dict] = []
        for si, st in enumerate(proc.get("steps") or [], 1):
            if not isinstance(st, dict):
                continue
            sid = f"s{si}"
            in_label = (st.get("input") or "").strip()
            out_label = (st.get("output") or "").strip()
            step: Dict[str, Any] = {
                "id": sid,
                "kind": "step",
                "via": (st.get("tool") or "").strip(),
                "directive": "",   # 待 LLM 用 wf-patch 填真实 prompt
                "inputs": ([{"type": in_label[:40], "value": "", "anchor": ""}]
                           if in_label else []),
                "outputs": ([{"id": f"{sid}o1", "type": out_label[:40], "value": "", "anchor": ""}]
                            if out_label else []),
            }
            steps_out.append(step)
        skeleton["procedures"].append({
            "id": pid,
            "name": (proc.get("name") or "").strip(),
            "purpose": (proc.get("final_product") or proc.get("purpose") or "").strip(),
            "category": (proc.get("category") or "").strip(),
            "platform": src.get("platform", ""),
            "author": src.get("author", ""),
            "declarations": {"inputs": [], "resources": [], "returns": {}},
            "steps": steps_out,
        })
    return skeleton


# ===========================================================================
# 工具本体
# ===========================================================================

_PLAN_DESC = (
    "【第一步必做·只调用一次】提交你对这篇教程的工序计划 (understanding)。读完原文+配图后, "
    "把它拆成若干工序、每个工序的步骤逐条展开、并声明每个工序覆盖原文哪些 0N 章节。"
    "工具会校验完整性 (有章节没被任何工序认领 → 报错让你补全; 工序只有单步 → 警告), "
    "通过后**自动据此生成 workflow.json 骨架** (工序/步骤/顺序锁定), 你之后只在骨架上填 "
    "value/directive/anchor, 不再增删工序或步骤。\n"
    "procedures 每项形如:\n"
    '  {"name":"工序名", "category":"产物创造|资产建设|自动化|分析|学习", '
    '"final_product":"终态产物", "source_sections":["01","03"], '
    '"steps":[{"tool":"工具名/human", "input":"输入数据的短标签(如 提示词)", '
    '"does":"这步做什么的一句话自由描述", "output":"产出的短标签(如 场景图)"}, ...]}\n'
    "要点: source_sections 必填 (原文每个 0N 章节都要被某个工序认领, 别漏整段); "
    "steps 逐步展开别压成单步; input/output 写**短标签**(它们会变成步骤输入/输出的 type); "
    "does 只是计划期的自由描述, **不是** workflow 的 taxonomy 动作词 (那留到 Phase 2 查 action.json 归类)。"
)


@tool(description=_PLAN_DESC, hidden_params=["context"], groups=["core"])
async def plan_procedures(
    summary: str,
    procedures: List[Dict[str, Any]],
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """提交工序计划 → 校验 → 自动生成 workflow.json 骨架。

    Args:
        summary: 2-4 句说清这篇在教什么、分几大板块。
        procedures: 工序列表 (结构见工具描述)。
    """
    procs = _coerce_procedures(procedures)
    if not procs:
        return ToolResult(
            title="计划为空",
            output="procedures 解析为空。请传一个工序数组, 每项含 name/category/"
                   "final_product/source_sections/steps (steps 逐步展开)。",
            error="empty plan",
        )

    body = _PLAN_CTX.get("body_text", "")
    secs = _source_sections(body)
    present = {num for num, _ in secs}
    sec_title = dict(secs)

    # ---- 收集声明 + 基础校验 ----
    claimed: set = set()
    warnings: List[str] = []
    for pi, proc in enumerate(procs, 1):
        if not isinstance(proc, dict):
            warnings.append(f"procedures[{pi-1}] 不是对象, 已跳过")
            continue
        name = (proc.get("name") or "").strip() or f"(工序{pi})"
        steps = proc.get("steps") or []
        for s in (proc.get("source_sections") or []):
            claimed.add(_norm_sec(s))
        if not steps:
            warnings.append(f"工序『{name}』没有 steps — 至少要有步骤序列")
        elif len(steps) == 1:
            warnings.append(f"工序『{name}』只有 1 个步骤 — 确认它真的无法再展开? (多数工序是多步)")
        for sj, st in enumerate(steps, 1):
            if not isinstance(st, dict):
                continue
            miss = [k for k in ("tool", "input", "does", "output") if not (st.get(k) or "").strip()]
            if miss:
                warnings.append(f"工序『{name}』step{sj} 缺 {'/'.join(miss)} (四要素: 工具 tool·输入 input·动作 does·输出 output)")

    # ---- 硬门禁: 章节覆盖 (只在原文确有 0N 分章时启用) ----
    if present:
        unclaimed = sorted(present - claimed)
        if unclaimed:
            lines = "\n".join(f"  · 章节 {n} 『{sec_title.get(n,'')}』" for n in unclaimed)
            return ToolResult(
                title=f"计划漏了 {len(unclaimed)} 个章节",
                output=(
                    f"原文有 {len(present)} 个章节, 你的计划只认领了 {sorted(claimed) or '无'}。"
                    f"下面这些章节**没有任何工序认领**, 极可能被整段漏抽:\n{lines}\n\n"
                    f"请逐个判断: 它是一条独立工序 (有自己的做法/产物), 还是某工序的若干步骤? "
                    f"想清楚后**重新调用 plan_procedures**, 让每个工序的 source_sections 把这些章节都覆盖上。"
                    f"(无独立做法、纯展示的章节可以并进相邻工序的 steps, 但要在某工序的 source_sections 里出现。)"
                ),
                error=f"unclaimed sections: {','.join(unclaimed)}",
            )
        bogus = sorted(claimed - present - {""})
        if bogus:
            warnings.append(f"source_sections 里 {bogus} 在原文里找不到对应 0N 章节 (写错章节号?)")

    # ---- 通过: 生成骨架 + 落盘 ----
    skeleton = _build_skeleton(summary, procs)
    out_dir: Optional[Path] = _PLAN_CTX.get("out_dir")
    nproc = len(skeleton["procedures"])
    nstep = sum(len(p["steps"]) for p in skeleton["procedures"])

    written = []
    if out_dir:
        scratch = out_dir / "_scratch"
        scratch.mkdir(parents=True, exist_ok=True)
        wf_path = out_dir / "workflow.json"
        wf_path.write_text(json.dumps(skeleton, ensure_ascii=False, indent=2), encoding="utf-8")
        written.append(wf_path.as_posix())
        # 计划原文留档 (供审计 workflow 是否仍按计划; 也方便你回看)
        plan_path = scratch / "understanding.json"
        plan_path.write_text(json.dumps(
            {"summary": summary, "procedures": procs}, ensure_ascii=False, indent=2), encoding="utf-8")
        written.append(plan_path.as_posix())

    # ---- 回灌进对话: 计划摘要 + 骨架结构 + 下一步指令 ----
    recap = [f"✅ 计划已通过校验, 已据此生成 workflow.json 骨架: {nproc} 工序 / {nstep} 步。"]
    if present:
        recap.append(f"章节覆盖: 原文 {sorted(present)} 全部被认领 ✓")
    recap.append("")
    for p in skeleton["procedures"]:
        recap.append(f"【{p['id']}】{p['name']} ({p['category']}) — {len(p['steps'])} 步")
        for st in p["steps"]:
            ins = st["inputs"][0]["type"] if st["inputs"] else "—"
            outs = st["outputs"][0]["type"] if st["outputs"] else "—"
            recap.append(f"    {st['id']} via={st['via'] or '—'}: [{ins}] → [{outs}]")
    if warnings:
        recap.append("")
        recap.append("⚠ 警告 (不阻塞, 但请核对):")
        for w in warnings:
            recap.append(f"  - {w}")
    recap.append("")
    recap.append(
        "下一步 (在已生成的骨架上, 别增删工序/步骤):\n"
        "  1. 逐步用 wf-patch.py 填 inputs/outputs 的 value(文本类用 @quote 拽**完整逐字**原文)、"
        "anchor(数据流 ← / →); **提示词建成 type=提示词 的输入/输出 value(整段逐字), 别塞 directive**"
        "(directive 只放严格反推/比例 2:3 这类元指令, 多数生图步留空);\n"
        "  2. 填完后**必须跑 `python spec/tools/verify-io.py --workflow <wf> --source <原文> --ocr <ocr>`**: "
        "它校验文本 IO 的 value 逐字、生成步有没有 type=提示词 输入、提示词 value 完不完整、declarations; "
        "报问题就修(此时可重读原文, 提示词用 @quote 提全), 修完重跑直到通过, 才进 Phase 2;\n"
        "  3. type 现在是描述性短标签, Phase 2 再归一化到词表 + 补 effect/action/substance/form/intent;\n"
        "  4. 结构已按计划锁定 —— 若发现确实要加工序/步骤, 说明计划不完整, 重新调 plan_procedures。"
    )
    return ToolResult(
        title=f"计划通过: {nproc} 工序 / {nstep} 步",
        output="\n".join(recap),
        metadata={"procedures": nproc, "steps": nstep,
                  "sections_present": sorted(present), "written": written,
                  "warnings": len(warnings)},
    )