# Scratch 工具脚本模板 本文档基于 `run_prompt/run_prompt.py` 的设计抽出来的**可迁移模式**， 供后续在 `examples/process_pipeline/scratch/` 下新增工具脚本时作为参考。 不是教条 —— 每条规则都附了"为什么"，遇到不适用的场景请基于场景判断。 --- ## 1. 目录结构 每个独立工具一个子目录，**脚本、数据、输出三者物理隔离**： ``` examples/process_pipeline/scratch/ ├── SCRIPT_TEMPLATE.md ← 本文档 └── / ← 一个工具一个目录 ├── .py ← 脚本本体（与目录同名） ├── inputs/ ← （前端预留）通过前端 UI 上传的数据文件默认存放在此 │ ├── .json │ └── prompt_input.txt └── outputs/ ← 所有输出都在这里 ├── result__.txt └── ... ``` **为什么**： - 子目录隔离 → 不同工具的脚本/数据互不干扰 - `inputs/` 和 `outputs/` 沙盒 → 整体 `.gitignore` 一行搞定，`git status` 不被结果污染，且前端面板会自动探测这两个文件夹用于“文件复用”。 - 脚本与目录同名 → `python /.py` 路径直观 --- ## 2. 文件头骨架 ```python """ <一句话说明工具用途> 用法： python /.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/` 内 ```python 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 的三种来源 ```python 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 容错 ```python 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` | ```python print(f"[info] starting...", file=sys.stderr) # 进度 print(content) # 仅当 --stdout 时 out_path.write_text(content, encoding="utf-8") # 默认 ``` **为什么**：管道兼容性 —— `python tool.py --stdout | jq ...` 不会被进度日志污染。 --- ## 7. 退码语义（统一约定） ```python """ 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" ```python 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）： ```python 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 分钟没动静像挂了。 ```python 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 习惯 ```python 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__.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//` 下建子目录 --- ## 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` —— 本文档的所有模式都能在这个脚本里找到具体实现。 读它比读这个文档更直接。