这一阶段的目标是:能写日常自动化脚本、批处理脚本、实验脚本。
学完后,你应该能完成这些事情:
- 从命令行接收参数。
- 读取环境变量和配置文件。
- 批量处理 JSON、JSONL、CSV、文本文件。
- 用日志记录脚本执行过程。
- 处理路径、目录、输出文件。
- 写一个可测试、可复用、可扩展的小脚本。
- 为 Agent 原型、评估、数据清洗写基础工具。
第 1 阶段学的是 Python 语法;第 2 阶段学的是如何把语法组织成“能每天用的小工具”。
Java 对照速查(本阶段工具的 Java/Spring 近亲):
| Java / 生态 | Python | 说明 |
|---|---|---|
args[] / Picocli / commons-cli |
argparse |
解析命令行参数、--flag、子命令 |
System.getenv("X") |
os.environ.get("X") |
读环境变量 |
| SLF4J + Logback | 标准库 logging |
分级日志、格式、输出位置 |
java.nio.file.Path / Files |
pathlib.Path |
路径拼接、读写、遍历 |
| Jackson | json |
读写 JSON / JSONL |
| OpenCSV | csv |
读写 CSV |
java.time |
datetime |
时间戳、格式化 |
application.yml |
.env / JSON / YAML |
外部配置 |
| JUnit | pytest |
给核心函数写测试 |
一个推荐的 Python 脚本通常长这样:
import argparse
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
def run(input_path: Path, output_path: Path) -> None:
logger.info("Reading %s", input_path)
text = input_path.read_text(encoding="utf-8")
output_path.write_text(text.strip(), encoding="utf-8")
logger.info("Wrote %s", output_path)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Clean a text file")
parser.add_argument("--input", required=True, help="Input file path")
parser.add_argument("--output", required=True, help="Output file path")
return parser.parse_args()
def main() -> None:
logging.basicConfig(level=logging.INFO)
args = parse_args()
run(Path(args.input), Path(args.output))
if __name__ == "__main__":
main()关键点:
- 核心逻辑放在
run()里,方便测试。 - 参数解析放在
parse_args()里,方便维护。 main()负责把命令行和核心逻辑连接起来。if __name__ == "__main__"表示只有直接运行这个文件时才执行main()。
argparse 是 Python 标准库,用来处理命令行参数。
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True)
parser.add_argument("--output", required=True)
args = parser.parse_args()
print(args.input)
print(args.output)运行:
python script.py --input data/raw.json --output data/clean.jsonparser.add_argument("--limit", type=int, default=100)
parser.add_argument("--dry-run", action="store_true")运行:
python script.py --limit 10 --dry-run说明:
type=int会把字符串参数转换成整数。default=100表示不传时使用默认值。action="store_true"表示只要出现这个参数,值就是True。
parser.add_argument("--format", choices=["json", "csv"], default="json")这可以限制用户只能传固定选项。
环境变量常用于保存 API key、环境名、默认配置。
import os
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY is required")带默认值:
env = os.environ.get("APP_ENV", "dev")注意:
- 不要把 API key 写死在代码里。
- 读取不到关键环境变量时,要给出清晰错误。
- 日志里不要打印完整密钥。
pathlib.Path 是现代 Python 推荐的路径处理方式。
from pathlib import Path
root = Path("data")
input_path = root / "raw.json"
output_path = root / "clean.json"
print(input_path.exists())
print(input_path.suffix)
print(input_path.stem)创建目录:
output_dir = Path("outputs")
output_dir.mkdir(parents=True, exist_ok=True)遍历文件:
for path in Path("data").glob("*.json"):
print(path)递归遍历:
for path in Path("data").rglob("*.md"):
print(path)建议:
- 函数入参可以接收
str | Path。 - 函数内部尽快转换成
Path。 - 输出文件前先确保父目录存在。
def write_output(path: str | Path, text: str) -> None:
output_path = Path(path)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(text, encoding="utf-8")JSON 适合保存结构化对象或数组。
import json
from pathlib import Path
data = {"name": "Ada", "score": 0.9}
Path("result.json").write_text(
json.dumps(data, ensure_ascii=False, indent=2),
encoding="utf-8",
)
loaded = json.loads(Path("result.json").read_text(encoding="utf-8"))JSONL 是一行一个 JSON 对象,非常适合大批量评估结果。
import json
from pathlib import Path
rows = [
{"id": "1", "input": "hello", "score": 0.8},
{"id": "2", "input": "world", "score": 0.9},
]
with Path("results.jsonl").open("w", encoding="utf-8") as file:
for row in rows:
file.write(json.dumps(row, ensure_ascii=False) + "\n")读取:
items = []
with Path("results.jsonl").open("r", encoding="utf-8") as file:
for line in file:
if line.strip():
items.append(json.loads(line))为什么 Agent 评估常用 JSONL:
- 可以一边运行一边写入。
- 单条失败不影响整个文件。
- 大文件更容易流式处理。
- 适合记录每个样本的输入、输出、耗时、错误。
CSV 适合表格数据。
import csv
from pathlib import Path
rows = [
{"id": "1", "score": "0.8"},
{"id": "2", "score": "0.9"},
]
with Path("summary.csv").open("w", encoding="utf-8", newline="") as file:
writer = csv.DictWriter(file, fieldnames=["id", "score"])
writer.writeheader()
writer.writerows(rows)读取:
with Path("summary.csv").open("r", encoding="utf-8", newline="") as file:
reader = csv.DictReader(file)
rows = list(reader)不要只靠 print() 调试脚本。正式脚本建议用 logging。
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)
logger = logging.getLogger(__name__)
logger.info("Start processing")
logger.warning("This row is missing score")
logger.error("Failed to process item")常用级别:
DEBUG:调试细节INFO:正常进度WARNING:可恢复问题ERROR:处理失败CRITICAL:严重错误
建议:
- 脚本开始、结束要有日志。
- 输入路径、输出路径要有日志。
- 跳过数据、失败数据要有日志。
- 不要在日志里打印完整 API key。
脚本经常需要记录运行时间、生成时间戳文件名、统计耗时。
from datetime import datetime, timezone
started_at = datetime.now(timezone.utc)
# do something
finished_at = datetime.now(timezone.utc)
duration_seconds = (finished_at - started_at).total_seconds()
print(duration_seconds)生成文件名:
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
output_path = f"outputs/result-{timestamp}.jsonl"建议:
- 记录日志时间时尽量带时区。
- 文件名里不要用冒号,Windows 路径不友好。
- 评估结果里建议记录
started_at、duration_seconds。
小脚本可以先用 JSON 配置。
config.json:
{
"model": "gpt-example",
"temperature": 0.2,
"max_items": 100
}读取:
import json
from pathlib import Path
config = json.loads(Path("config.json").read_text(encoding="utf-8"))
model = config.get("model", "default-model")更复杂时可以使用:
.env:适合环境变量和密钥- YAML:适合层级配置
- Pydantic Settings:适合强校验配置
当前阶段先掌握 JSON 配置即可。
批处理脚本经常需要进度条。tqdm 是常用选择。
安装:
python -m pip install tqdm使用:
from tqdm import tqdm
items = range(100)
for item in tqdm(items, desc="Processing"):
pass如果项目暂时没有安装 tqdm,可以先用日志记录每处理多少条:
for index, item in enumerate(items, start=1):
if index % 100 == 0:
logger.info("Processed %s items", index)脚本也应该写测试,尤其是数据清洗、格式转换、评估统计这类逻辑。
被测试函数:
def clean_text(text: str) -> str:
return " ".join(text.strip().split())测试文件:
def test_clean_text() -> None:
assert clean_text(" hello python ") == "hello python"运行:
python -m pytest tests建议:
- 核心逻辑写成函数。
- 命令行解析和文件 IO 尽量薄。
- 测试核心函数,不要只测试命令行。
- 对边界情况写测试:空输入、缺字段、非法值、重复数据。
输入:
[
{"id": "1", "text": " hello world "},
{"id": "2", "text": ""}
]输出:
[
{"id": "1", "text": "hello world"}
]关键能力:
- JSON 读写
- 字段检查
- 字符串清洗
- 跳过无效数据
输入 JSONL:
{"id":"1","question":"What is Python?","expected":"A programming language."}
{"id":"2","question":"What is FastAPI?","expected":"A web framework."}输出 JSONL:
{"id":"1","answer":"...","score":0.8,"success":true,"duration_seconds":0.12}
{"id":"2","answer":"...","score":0.9,"success":true,"duration_seconds":0.10}关键能力:
- 一行一条处理
- 错误不影响整体
- 记录耗时
- 记录 success/error
- 输出可被 pandas 分析的数据
输入 JSONL,输出 CSV:
total,success_count,failed_count,average_score
100,95,5,0.82关键能力:
- 读取 JSONL
- 分组和统计
- 输出 CSV
- 为后续 pandas 分析打基础
scripts/
clean_texts.py
run_eval.py
summarize_eval.py
data/
raw_texts.json
eval_dataset.jsonl
outputs/
cleaned_texts.json
eval_results.jsonl
summary.csv
tests/
test_clean_texts.py
test_summarize_eval.py
当前阶段不追求复杂工程化,先追求:
- 能运行
- 能复用
- 能测试
- 日志清楚
- 输入输出明确
本阶段配套示例脚本:
.\.venv\Scripts\python.exe scripts\stage2_basic_scripting.py --input data\stage2_raw_texts.json --output data\stage2_cleaned_texts.json --summary data\stage2_summary.csv运行测试:
.\.venv\Scripts\python.exe -m pytest tests\test_stage2_basic_scripting.py你可以重点观察:
- 参数是如何通过
argparse传入的。 - 路径是如何用
Path处理的。 - JSON 和 CSV 是如何读写的。
- 日志在哪里记录。
- 核心逻辑如何被 pytest 测试。
学完这一阶段,你应该能够:
- 写一个带
--input、--output参数的脚本。 - 用
Path安全处理输入输出路径。 - 读取和写入 JSON、JSONL、CSV。
- 用
logging记录脚本过程。 - 用
datetime记录运行时间和耗时。 - 从环境变量读取配置。
- 用 JSON 文件保存简单配置。
- 写 pytest 测试核心处理函数。
- 为 Agent 原型和评估任务写小型自动化脚本。