Agent/examples/process_pipeline/scratch/SCRIPT_TEMPLATE.md

Scratch 工具脚本模板

本文档基于 run_prompt/run_prompt.py 的设计抽出来的可迁移模式， 供后续在 examples/process_pipeline/scratch/ 下新增工具脚本时作为参考。

不是教条 —— 每条规则都附了"为什么"，遇到不适用的场景请基于场景判断。

---

1. 目录结构

每个独立工具一个子目录，脚本、数据、输出三者物理隔离：

examples/process_pipeline/scratch/
├── SCRIPT_TEMPLATE.md              ← 本文档
└── <tool_name>/                    ← 一个工具一个目录
    ├── <tool_name>.py              ← 脚本本体（与目录同名）
    ├── inputs/                     ← （前端预留）通过前端 UI 上传的数据文件默认存放在此
    │   ├── <input_data>.json
    │   └── prompt_input.txt
    └── outputs/                    ← 所有输出都在这里
        ├── result_<mode>_<ts>.txt
        └── ...

为什么：

• 子目录隔离 → 不同工具的脚本/数据互不干扰
• inputs/ 和 outputs/ 沙盒 → 整体 .gitignore 一行搞定，git status 不被结果污染，且前端面板会自动探测这两个文件夹用于“文件复用”。
• 脚本与目录同名 → python <dir>/<dir>.py 路径直观

---

2. 文件头骨架

"""
<一句话说明工具用途>

用法：
    python <path>/<tool>.py --arg1 ... --arg2 ...

脚本通过探测 .git/pyproject.toml 自动定位项目根，可以放在仓库内任意位置。
"""

import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

# Windows 控制台 UTF-8（中文输出/输入必备）
for _s in (sys.stdout, sys.stderr):
    try:
        _s.reconfigure(encoding="utf-8")
    except (AttributeError, OSError):
        pass


def _find_project_root(start: Path) -> Path:
    """沿父目录上爬，找到含 .git / pyproject.toml 的目录。"""
    p = start.resolve()
    for ancestor in [p, *p.parents]:
        if (ancestor / ".git").exists() or (ancestor / "pyproject.toml").exists():
            return ancestor
    return start.resolve().parent


PROJECT_ROOT = _find_project_root(Path(__file__))
SCRIPT_DIR = Path(__file__).resolve().parent
OUTPUTS_DIR = SCRIPT_DIR / "outputs"

sys.path.insert(0, str(PROJECT_ROOT))

from dotenv import load_dotenv
load_dotenv(PROJECT_ROOT / ".env")

为什么这些选择：

• 智能项目根探测 → 脚本无论挪到哪都能正确 import agent.* 和加载 .env
• UTF-8 reconfigure → Windows 默认 cp1252/gbk，中文 print 会 UnicodeEncodeError
• SCRIPT_DIR / OUTPUTS_DIR 模块级常量 → 后续函数直接引用，避免每处重新计算

---

3. 输出沙盒：默认写文件，路径限制在 outputs/ 内

def default_output_path(tag: str) -> Path:
    """带时间戳的默认输出文件，永不覆盖历史。"""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    return OUTPUTS_DIR / f"result_{tag}_{ts}.txt"


def resolve_user_output(rel_path: str) -> Path:
    """
    把用户的 --output 解析到 OUTPUTS_DIR 下。
    - 必须是相对路径
    - 解析后必须落在 OUTPUTS_DIR 之内（防 `..` 越界）
    """
    p = Path(rel_path)
    if p.is_absolute():
        raise SystemExit(f"ERROR: --output 必须是相对路径: {rel_path!r}")
    target = (OUTPUTS_DIR / p).resolve()
    try:
        target.relative_to(OUTPUTS_DIR.resolve())
    except ValueError:
        raise SystemExit(f"ERROR: --output 越界到 {target}（不允许 '..'）")
    return target


def write_output(content: str, args, tag: str) -> None:
    """统一输出处理：默认写文件，--stdout 才打 stdout。"""
    if args.stdout:
        print(content)
        return
    out_path = resolve_user_output(args.output) if args.output else default_output_path(tag)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(content, encoding="utf-8")
    print(f"[info] written to {out_path}  ({len(content)} chars)", file=sys.stderr)

为什么：

• 默认写文件 → 探索性任务的结果天然该保留，stdout 一关就丢了
• 时间戳防覆盖 → LLM/agent 调用都贵，丢结果重跑代价高
• 沙盒约束 → 防止 --output C:\Windows\system32\evil.txt；也防 --output ../../../etc/passwd
• fail-fast 校验 → 在 run() 开头就 resolve_user_output() 一次，避免长任务跑完才报路径错

---

4. CLI 输入：prompt 的三种来源

def read_prompt(prompt_arg: Optional[str], prompt_file: Optional[str]) -> str:
    """从 --prompt / --prompt-file / stdin 三选一。"""
    if prompt_file:
        return Path(prompt_file).read_text(encoding="utf-8").strip()
    if prompt_arg == "-":
        return sys.stdin.read().strip()
    if prompt_arg:
        return prompt_arg
    raise SystemExit("ERROR: 请提供 --prompt TEXT / --prompt - / --prompt-file PATH")

为什么：

• 命令行字符串适合短 prompt
• 文件适合长/含特殊字符（中文、括号、引号）的 prompt —— 避免 shell 转义地狱
• stdin 适合管道场景（Get-Content prompt.txt | python ...）

---

5. 文件输入：多文件 + UTF-8 容错

def read_file_for_prompt(path: str) -> str:
    """读单个文件。不做客户端预检 —— 信任下游报错。"""
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"File not found: {path}")
    return p.read_text(encoding="utf-8", errors="replace")

为什么：

• errors="replace" → 二进制污染的文本不会让脚本崩，只是会有少量乱码
• 不做客户端大小预检 → 服务端/LLM 端的报错最精确，客户端粗估只会过度防御（见 run_prompt.py 早期教训）

---

6. stderr / stdout / 文件分流

约定：

流	用途	例
stderr	所有进度日志、警告、错误	[info] mode=... [error] ... !!! WARNING ...
stdout	只用于最终结果（仅当 --stdout）	LLM 答案
--output 文件	最终结果常规去向	outputs/result_*.txt

print(f"[info] starting...", file=sys.stderr)   # 进度
print(content)                                   # 仅当 --stdout 时
out_path.write_text(content, encoding="utf-8")   # 默认

为什么：管道兼容性 —— python tool.py --stdout | jq ... 不会被进度日志污染。

---

7. 退码语义（统一约定）

"""
Return codes:
    0   — success
    1   — runtime error (LLM 调用失败、文件读不到、参数错…)
    2   — partial success (输出被截断、超出 max_tokens 等)
    3   — input validation error (沙盒越界、文件过大等)
    130 — KeyboardInterrupt (Ctrl+C)
"""

为什么：让自动化脚本能根据退码做不同处理（if exit==2: 重试加大 max_tokens）。 "成功 0 / 失败 1" 是底线，但不要止步于此 —— 多 1-2 个退码能让 shell 脚本/CI 流水线更聪明。

---

8. 全局异常兜底：消灭"孤儿 exit code"

def main():
    """全局兜底：任何未捕获异常都打 traceback + 明确退码。"""
    try:
        args = build_parser().parse_args()
        code = asyncio.run(run(args))
        sys.exit(code)
    except KeyboardInterrupt:
        print("\n[info] interrupted by user (Ctrl+C)", file=sys.stderr)
        sys.exit(130)
    except SystemExit:
        raise
    except BaseException as e:
        import traceback
        print(f"\n!!! UNEXPECTED ERROR: {type(e).__name__}: {e}", file=sys.stderr)
        traceback.print_exc(file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

为什么：

• 没有兜底时，asyncio.run 内部的异常会以"非 0 但无信息"退出 —— PowerShell 上会显示成 EXIT=-1
• 兜底后所有失败都有 traceback —— 调试不再是猜谜
• KeyboardInterrupt 退 130 是 Unix 约定，便于 shell 脚本检测

---

9. 服务端错误体完整打印（LLM 工具特化）

调用外部 LLM API 失败时，完整打印服务端 body（不只 exception message）：

try:
    response = await llm_call(...)
except Exception as e:
    print(f"!!! LLM CALL FAILED: {type(e).__name__}: {e}", file=sys.stderr)
    # httpx HTTPStatusError 等会带 response.text
    for attr in ("response", "body"):
        obj = getattr(e, attr, None)
        if obj is not None:
            try:
                text = obj.text if hasattr(obj, "text") else str(obj)
                print(f"!!! server body:", file=sys.stderr)
                print(text[:4000], file=sys.stderr)
            except Exception:
                pass
    import traceback; traceback.print_exc(file=sys.stderr)
    return 1

为什么：服务端的 400 body 通常包含具体的拒绝原因（"context exceeds 1M tokens" / "API key invalid" / "model not available"），比 Python 的 HTTPStatusError: 400 Bad Request 信息量大 10 倍。

---

10. 长任务的流式进度

任何会运行超过 5 秒的任务都应该在 stderr 流式打进度。 agent / 多步骤工具尤其重要 —— 否则 5 分钟没动静像挂了。

for tool_use in agent.tool_calls:
    print(f"[agent tool_use] {tool_use.name}({json.dumps(tool_use.input)[:200]})",
          file=sys.stderr)
# 完成时打统计
print(f"[info] done: turns={n} duration={ms}ms in={i_tok} out={o_tok}",
      file=sys.stderr)

为什么：可观察性 —— 用户能看到 agent 在做什么、卡在哪、走偏没。

---

11. CLI argparse 习惯

def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="...",
        formatter_class=argparse.RawDescriptionHelpFormatter,  # 让 \n 在 help 里生效
    )
    p.add_argument("--mode", choices=["fast", "slow", "smart"], default="smart", 
                   help="运行模式")  # 使用 choices 会在前端自动渲染为下拉单选框
    p.add_argument("--prompt", help="...")
    p.add_argument("--prompt-file", help="输入提示词文件")
    p.add_argument("--file", action="append",  # 多次传同一参数 → 列表
                   help="要处理的文档文件（可多次传）")
    p.add_argument("--output", help="相对 outputs/ 的路径")
    p.add_argument("--stdout", action="store_true", help="强制走 stdout")
    p.add_argument("--show-prompt", action="store_true", help="debug 用")
    return p

前端解析与 UI 生成规则（重要！）：

• 文件上传 UI：如果希望前端将某个参数渲染为高级的【📂 +上传文件】及【沙盒文件复用】组件，参数名必须包含 file / path，或者它的 help 提示语中必须包含 文件 两字（例如 --data 的 help 设为 "输入的数据文件" 也能生效）。
• 多文件支持（Tag 标签）：如需支持前端用户一次上传/选择多个文件，请使用 action="append"。前端会为每个文件生成一个 Tag，并发送多次相同的参数名。
• 开关按钮（Checkbox）：action="store_true" 会被前端自动渲染为复选框。
• 下拉菜单（Select）：定义了 choices=["a", "b"] 的参数，前端会自动渲染为下拉选择框。
• 必填项与默认值：如果参数没有 default 且 required=True，前端会打上红色红星。help 字段也会直接作为输入框的 placeholder 提示。

---

12. 检查清单（写新脚本时勾一遍）

基础设施

• 文件头加 UTF-8 reconfigure（Windows 中文必备）
• 用 _find_project_root() 探测项目根（不硬编码 parent.parent.parent）
• sys.path.insert(0, str(PROJECT_ROOT)) + load_dotenv(PROJECT_ROOT / ".env")
• 模块级常量：PROJECT_ROOT / SCRIPT_DIR / OUTPUTS_DIR

文件 IO

• 所有 read_text / write_text 显式 encoding="utf-8"
• 输出默认写 outputs/result_<tag>_<ts>.txt
• 用户指定的 --output 限定在 outputs/ 沙盒内
• OUTPUTS_DIR.mkdir(parents=True, exist_ok=True)

CLI

• prompt 支持 --prompt / --prompt-file / --prompt -（stdin）
• 多文件输入用 --file + action="append"
• --stdout flag 保留管道场景
• 每个 argument 都有清晰的 help=...

错误处理

• main() 全局 try/except 兜底（包括 KeyboardInterrupt 和 BaseException）
• 退码约定：0/1/2/3/130（按需用）
• LLM/HTTP 调用失败打印完整服务端 body
• Fail-fast：路径/参数校验在长任务之前

可观察性

• 启动时 stderr 打 [info] mode=... model=... ...
• 长任务过程中流式打进度（每步骤一行）
• 完成时打统计（duration、token 用量等）
• 进度日志全部走 stderr，stdout 留给结果

文档

• 文件头 docstring 包含：用途 + 用法示例 + 关键参数
• 写新工具时，在 examples/process_pipeline/scratch/<tool_name>/ 下建子目录

---

13. 反模式（不要这么写）

错误做法	为什么不好	正确做法
PROJECT_ROOT = Path(__file__).parent.parent.parent	脚本一挪位置就崩	_find_project_root()
os.chdir(...) 改全局 cwd	影响其它子进程/线程	用 cwd= 参数传给子进程，本进程不改
print(content) 默认输出	用户关 terminal 就丢了	默认写文件，--stdout 显式打开
中文文本 open(path, 'r')	Windows 用 cp1252 编码读 → 乱码	显式 encoding="utf-8"
except Exception: pass	静默失败 → 调试地狱	至少 traceback.print_exc()
--max-tokens 4000 写死小值	限制了模型能力	默认取 model 上限，用户想小再 --max-tokens 100
客户端做输入大小预检	经常过度防御 → 拒绝合法请求	信任服务端报错，把错误透传
if mode == "X": ...; elif "Y": ...; elif "Z": ... 长链	加新 mode 要改多处	MODES = {"X": fn_x, ...}; MODES[mode]()

---

参考实现

run_prompt/run_prompt.py —— 本文档的所有模式都能在这个脚本里找到具体实现。 读它比读这个文档更直接。