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 = "", spec_name: str = "spec") -> 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 {},
        "spec_name": spec_name or "spec",
    })


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

# 与 lint-case.py 同口径: 行首两位标号 NN| (01..99), 行首要求天然排除 "图 0N|" 配图说明
_SEC_RE = re.compile(r'(?m)^\s*(\d{2})\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 _io_labels(st: dict, key: str) -> List[str]:
    """取步骤的 input/output 标签, 字符串或数组都收 (生成步典型是 [提示词, 参考图] 两个输入)。"""
    v = st.get(key)
    if v is None:
        v = st.get(key + "s")
    if isinstance(v, str):
        v = [v]
    if not isinstance(v, list):
        return []
    return [str(x).strip()[:40] for x in v if str(x).strip()]


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}"
            ins = _io_labels(st, "input")
            outs = _io_labels(st, "output")
            tool = (st.get("tool") or "").strip()
            loop = (st.get("loop") or "").strip()[:40]
            if loop:
                # 循环步 → block 父 + nested 子 (README「有循环/并行怎么切」的标准形)。
                # block: 输入=被遍历的集合, 输出=结果列表; 子步带真实工具和逐项 IO。
                steps_out.append({
                    "id": sid, "kind": "block", "via": "-", "directive": "",
                    "inputs": [{"type": loop, "value": "", "anchor": ""}],
                    "outputs": [{"id": f"{sid}o1",
                                 "type": (outs[0] + "列表")[:40] if outs else "结果列表",
                                 "value": "", "anchor": ""}],
                })
                steps_out.append({
                    "id": f"{sid}.1", "kind": "nested", "group": sid, "via": tool,
                    "directive": "",
                    "inputs": [{"type": t, "value": "", "anchor": ""} for t in ins],
                    "outputs": [{"id": f"{sid}.1o{oi}", "type": t, "value": "", "anchor": ""}
                                for oi, t in enumerate(outs, 1)],
                })
                continue
            steps_out.append({
                "id": sid,
                "kind": "step",
                "via": tool,
                "directive": "",   # 待 LLM 用 wf-patch 填真实 prompt
                "inputs": [{"type": t, "value": "", "anchor": ""} for t in ins],
                "outputs": [{"id": f"{sid}o{oi}", "type": t, "value": "", "anchor": ""}
                            for oi, t in enumerate(outs, 1)],
            })
        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); "
    '对一批数据逐个重复的步骤加 "loop":"被遍历数据的短标签(如 分镜脚本)", 骨架会自动展开成循环块+子步; '
    "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] = []
    io_missing: List[str] = []   # 缺 input/output 标签是硬门禁: 骨架会带空 IO 出生, 下游全瞎
    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", "does") if not str(st.get(k) or "").strip()]
            if miss:
                warnings.append(f"工序『{name}』step{sj} 缺 {'/'.join(miss)} (四要素: 工具 tool·输入 input·动作 does·输出 output)")
            io_miss = [k for k in ("input", "output") if not _io_labels(st, k)]
            if io_miss:
                io_missing.append(f"工序『{name}』step{sj} 缺 {'/'.join(io_miss)}")

    # ---- 硬门禁: 每步必有 input + output 标签 (步骤=对输入操作产生新产物, 缺标签=骨架洞) ----
    if io_missing:
        lines = "\n".join(f"  · {m}" for m in io_missing)
        return ToolResult(
            title=f"{len(io_missing)} 个步骤缺 input/output 标签",
            output=(
                f"以下步骤缺输入或输出的短标签, 骨架会带着空 IO 生成, 后续校验/渲染全过不去:\n{lines}\n\n"
                f"每步都要写 input(用到什么数据)和 output(产出什么) 的短标签 (字符串或数组)。"
                f"原文没明写的, 按工艺推断该有什么(如生图步 input 必有 提示词)。"
                f"补全后**重新调用 plan_procedures**。"
            ),
            error=f"steps missing input/output: {len(io_missing)}",
        )

    # ---- 硬门禁: 章节覆盖 (只在原文确有 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())

    # ---- 回灌进对话: 计划摘要 + 骨架结构 + 指针 (规则细节在 README, 这里不复述) ----
    spec_name = _PLAN_CTX.get("spec_name", "spec")
    recap = [f"✅ 计划已通过校验, 已据此生成 workflow.json 骨架: {nproc} 工序 / {nstep} 步。"]
    if present:
        recap.append(f"章节覆盖: 原文 {sorted(present)} 全部被认领 ✓")
    else:
        recap.append("ℹ 原文未检测到 `NN|` 分章标号, 章节覆盖门禁未启用 — 自行核对没有整段漏抽。")
    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 = "+".join(io["type"] for io in st["inputs"]) or "—"
            outs = "+".join(io["type"] for io in st["outputs"]) or "—"
            tag = f" [{st['kind']}]" if st.get("kind") != "step" else ""
            recap.append(f"    {st['id']}{tag} via={st['via'] or '—'}: [{ins}] → [{outs}]")
    if warnings:
        recap.append("")
        recap.append("⚠ 警告 (不阻塞, 但请核对):")
        for w in warnings:
            recap.append(f"  - {w}")
    recap.append("")
    recap.append(
        "锁定范围 = 工序划分与步骤序列; **在已有步骤里补输入/输出是允许且常见的**"
        "(如生成步补第二个输入, wf-patch 路径不存在默认 upsert)。"
        f"接下来按 {spec_name}/README.md「第二阶段 · 2.0 填内容」执行: "
        "wf-patch 填 value(@quote)/anchor → verify-io.py 校验通过 → 进 2.1 归类。"
        "若工序/步骤划分本身不对, 修改计划后重新调用 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)},
    )