用 Claude Code 改 Python 批处理：先量内存峰值，再做 JSONL 流式聚合

一段批处理跑完后只返回三个分组，看上去没有多少数据留在内存里。可在返回之前，它可能已经把整份文件读成字符串，再拆成行列表，最后把每一行解析成字典。最终结果小，并不能说明处理中途没有出现过大峰值。

把这项优化交给 Claude Code 时，先把问题收窄：在业务结果不变的前提下，是否可以减少同时存活的输入对象？本文提供一份原创的本地 JSONL 实验，比较全量读入、逐条聚合，以及一个故意把生成器重新转成列表的负对照。三份实现使用同一解析规则，预期计数独立写出。

以下数据于 2026 年 10 月 13 日用已安装的 CPython 3.12.14、Linux x86_64 实际运行得到，只用了 Python 标准库。没有安装依赖，没有调用 Claude Code 或其他模型，也没有用模型回答代替实验结果。文末给出的交接请求可用于你自己的已批准项目；这里验证的是 Python 程序和测量方法。

tracemalloc 官方文档把 get_traced_memory() 的两个值定义为当前和峰值追踪内存。本例在各个工作子进程完成导入后调用 start(1)，从读取输入之前开始追踪，在聚合返回之后立即取数，再停止追踪。报告序列化、父进程生成夹具和保存报告都不在这个窗口里。

因此，current_bytes 是取数那一刻仍被追踪的分配，peak_bytes 是本次追踪以来的最高值。两者都不是操作系统的 RSS，也不包含启动追踪之前的所有分配。不能把“峰值下降”改写为“整进程只用了这些内存”，也不能据此推断原生扩展、子进程、文件缓存或容器整体的占用。本例没有采集耗时，不能声称提速。

每一个模式、规模和重复次数都启动一个新的 python -I -B 进程，追踪深度固定为一帧。这里不需要在同一进程里 reset_peak()；复用进程再只重置峰值，并不会自动清空已有对象和缓存。隔离启动也不是安全沙箱，它只是减少这次比较中的状态混用。

输入是 UTF-8 文本，每个物理行恰好一个 JSON 对象。对象只能包含 group、units、payload 三个字段：group 是非空字符串，units 是非负整数且不接受布尔值，payload 是字符串。本例汇总总记录数、总 units，以及每组的这两个数，不保留 payload，也不输出原始记录。

先手算六条夹具：alpha 的 units 为 1 + 3 = 4，beta 为 2 + 5 = 7，gamma 为 4 + 6 = 10。每组两条，总计六条、21 units。这份预期写在 expected.json 里，不能通过先运行 full 模式再把其输出当作正确答案。

{"group":"alpha","units":1,"payload":"a"}
{"group":"beta","units":2,"payload":"b"}
{"group":"alpha","units":3,"payload":"c"}
{"group":"gamma","units":4,"payload":"d"}
{"group":"beta","units":5,"payload":"e"}
{"group":"gamma","units":6,"payload":"f"}

大夹具重复相同的六条分组和 units 模式，把每条 payload 换成 512 个 ASCII 字符 x。600、6,000、60,000 条分别重复 100、1,000、10,000 轮，所以最终总 units 应为 2,100、21,000、210,000。60,000 条时，每组 20,000 条，alpha、beta、gamma 的 units 分别是 40,000、70,000、100,000。数据生成器负责写输入，验证器只缩放手写的六条预期，不调用被测聚合器生成答案。

这让 payload 成为可控的内存负担，而不是改变业务输出的变量。它也限制了结论：固定三个分组、固定短字段，不能代表用户标识近乎唯一或单条内容特别长的真实数据。

full 模式先 read()，再拆成所有行，随后把解析对象装进 rows 列表，最后聚合。聚合开始时，完整文本、行列表和对象列表仍然可达。它不是内存泄漏；函数结束后这些引用会消失，但中途的大峰值已经发生。

stream 模式把 records(path) 直接交给 aggregate()。文件迭代器给出下一行，生成器解析一条，聚合器立即更新累计值，然后继续消费。Python 文件读写文档说明了逐行迭代文件的用法。重要的改动是逐条消费，而不只是把函数写成带 yield 的样子。

materialized 模式故意执行 aggregate(list(records(path)))。它不再保留完整原文和所有行字符串，却仍然把全部解析对象留下。这个负对照能揭穿一种常见的假优化：读取函数变成生成器，但下游为了方便又调用 list()、sorted()，或把所有对象收集到另一个容器里。是否可以避免排序或保留全部对象，取决于真实输出要求，不能靠改语法解决。

# 全量：完整文本、所有行、所有对象都存活到聚合阶段
text = handle.read()
lines = text.split("\n")
rows = [parse_line(line, n) for n, line in enumerate(lines, 1)]
result = aggregate(rows)

# 流式：消费一条，更新一次累计值
result = aggregate(records(path))

# 负对照：生成器被重新全部物化
result = aggregate(list(records(path)))

上面只是结构示意，不能替代文末的可运行实现。完整源码处理了空文件、末尾换行和空白行。特别是 full 模式没有使用 str.splitlines()：该方法还会把 U+2028 等字符视为分隔，而这样的字符可以出现在合法 JSON 字符串中。本例先让文本模式统一 CRLF/CR 换行，再只按 LF 拆分，使两种读取路径遵守相同的物理行规则。

从正文保存 lab.py、make_fixtures.py、expected.json 三份文件到同一空目录，保持 UTF-8 编码。make_fixtures.py 会在旁边生成五份固定夹具，其中两个文件故意没有末尾换行。三份源码都在正文中，不需要额外下载实验文件。

python3 -I -B make_fixtures.py
python3 -I -B lab.py run evidence-01

运行器只生成自己的合成数据，在系统临时目录复制一份已知脚本并启动独立进程，结束后清理这些临时数据。它不接收已有项目作为实验输入。evidence-01 必须不存在，脚本会拒绝覆盖；复跑时改用 evidence-02。fixtures 已存在时，夹具生成器同样拒绝重建。失败时输出目录可能已经创建，请检查错误后使用新的输出目录。

正常结束应看到以下摘要。这里的 PASS 只断言业务结果和约定的错误行为，没有把某个内存字节数写成通过条件。

PASS: 15 fixture + 27 bulk + 4 limit business-result checks
PASS: 33 malformed-input checks; no memory byte threshold used
MEASURED: current/peak traced Python allocations; not RSS or runtime
REPORT: .../evidence-01/report.json

report.json 保存 Python 和平台版本、每次观察的模式和次数、输入大小与 SHA-256、结果、当前与峰值字节数，以及源文件和夹具哈希。输入哈希由父进程计算，不计入工作进程的测量窗口。哈希用于发现复跑文件是否变化，不是不可伪造的证明，也不能替代代码审阅。

测量环境为 CPython 3.12.14，编译器标识 Clang 22.1.3，Linux 6.18.44、x86_64、glibc 2.41。每个点重复三次，顺序是 full、stream、materialized；没有固定哈希种子，没有做运行时统计推断。三个输入文件分别为 331,600、3,316,000、33,160,000 字节。

下表的数值单位全部为字节。峰值中位数用于压低单次波动的影响，同时保留实际范围；它不是跨机器应复现的常量。

在这一组固定形状的输入里，full 和 materialized 的峰值随着条数大致成比例增长；stream 的三组中位数都在约 0.03 MiB 附近。60,000 条时，full 的峰值约为 119.33 MiB，materialized 约为 53.33 MiB，而 stream 约为 0.034 MiB。这是此次实际观察，不是对所有 JSONL 程序的性能保证。

再看 current：full 跑完只剩几十 KB 被追踪，远小于中途峰值。如果只在函数返回后看一次“当前占用”，就容易漏掉问题。也不能用两个 current 的差额推算整进程已向操作系统归还多少内存。

换一台机器或解释器，小额缓存和分配策略可能改变数值，流式的几次结果也未必单调。复跑时先看业务断言，再检查输入哈希、窗口与实现，最后解释规模变化。不要为了把结果变绿而规定“必须小于 40,000 字节”，也不要用单次有利结果替换完整的三次观察。

比较峰值只能指出差异，还需要解释原因。运行器另开一个诊断进程，使用 6,000 条输入，在 full 模式完成全部解析、尚未开始聚合时取快照。此时三个大容器仍存活，适合观察分配来源。诊断进程和上表测量进程分开，避免把创建和整理快照的成本混进性能比较。

这一份独立诊断的取快照前 current 为 12,553,835 字节。按行分组的前六项如下，显示的行号对应文末完整 lab.py，而不是上面的结构示意。

decoder.py:354  4,432,000 bytes  30,000 blocks
lab.py:72  3,609,152 bytes  6,001 blocks
<frozen codecs>:322  3,316,041 bytes  1 blocks
__init__.py:359  720,000 bytes  6,000 blocks
lab.py:19  383,616 bytes  5,994 blocks
lab.py:75  53,024 bytes  1 blocks

这里的 lab.py:72 对应所有行字符串的拆分，lab.py:75 对应 rows 容器的创建。json/decoder.py 和 json/__init__.py 的条目来自 JSON 解码路径，frozen codecs 的条目对应读取文本时的解码分配。它们是这次快照里的分配位置，不是一个“应该替换标准库”的结论。尤其不能把 decoder.py 占得多直接解释为 JSON 库泄漏。

快照只看到取样时仍存在的追踪块，不能保证捕获每个已经释放的短暂峰值。一帧也不能还原完整调用路径。若要进一步定位真实项目，可以在独立诊断轮次增加追踪深度或比较两个快照，同时保留测量窗口定义；不要把诊断配置改变后的数字混回原表。

五份有效夹具分别验证六条手工样例、真正的零字节空文件、中文与 emoji、末行无换行，以及 CRLF 换行。Unicode 夹具还包含实际的 U+2028 字符，并把两个相同中文分组合并。所有夹具都分别经过三种模式，因此不是只证明一个 happy path。

空文件应返回零记录和空分组；空白行则按这份输入合同报错，不能偷偷跳过。末行没有换行仍是一条有效记录。错误样例还覆盖非法 JSON、缺字段、额外字段、布尔 units、负 units、重复对象键、NaN、数组顶层、第二行损坏和非法 UTF-8。三种模式都应失败退出，不产出成功的部分汇总。

Python JSON 文档的兼容性说明值得单独核对：默认解析器对重复键和非标准数值有自己的行为。本例通过 object_pairs_hook 拒绝重复键，用 parse_constant 拒绝 NaN 等常量，再做字段校验。不要假设调用一次 json.loads() 就已经执行了业务输入合同。

这里没有做不可信输入的完整防护：没有设置单行长度、嵌套深度、整数位数或文件大小的应用级上限，也没有测试并发改写、截断恢复、压缩输入、断点续跑和错误隔离。非法 UTF-8 的错误来自缓冲解码，不承诺精确业务行号。深层嵌套等资源型错误也不在这 33 个受控错误检查的覆盖范围内。

流式算法在遇到后面的坏行前可能已经计算了前面几条的内存累计值；本例没有对外写入这些中间值，所以失败不留下业务输出。如果真实程序边读边发请求或提交数据库，错误处理就有了副作用，需要重新设计事务或幂等策略，不能从本例的“同样失败”推断外部状态也相同。

2026 年 10 月 20 日补充复验发现，测量窗口结束并不等于输出合同已经兑现。原版 worker 在停止 tracemalloc 后直接调用 json.dumps()；编码错误不在原先的 try 范围内。两条记录的 units 都是 4,300 位的正整数时，每条都能解析，求和也能完成，但总数变成 4,301 位。在本次 CPython 3.12.14 环境中，三种模式都因此以退出码 1 结束，stdout 为空。这份反例只有 8,682 字节，不需要大文件或内存耗尽。

CPython 的整数与字符串转换限制约束十进制输入和输出；默认上限是 4,300 位。整数加法本身不会因为结果超过该十进制位数就失败。输入可以解析，不能推出累计结果一定可以按相同规则编码。

文末 lab.py 已把成功响应先编码到 encoded，捕获这里的 ValueError 并返回退出码 2 与 ok: false，错误前缀为 result serialization failed:。只有编码完成后才打印成功响应，因此这类错误不会先输出部分成功 JSON。tracemalloc.stop() 仍在编码之前，原来的内存测量窗口、业务算法和 10 月 13 日实测表格均未改变。这不是扩大允许数值范围，也没有把整数悄悄改成浮点数或字符串。

将下面的补充脚本保存为同目录的 check_output_boundary.py。它用固定的 4,300 位进程启动配置复验四种情况：单条恰好在上限内、同组求和超限、不同组总和超限、单条输入已经超限。每种都经过 full、stream、materialized，共 12 项；成功用独立整数公式检查完整结果，失败检查退出码、无 stderr 和完整的错误 JSON。脚本只创建临时合成输入，并对每个子进程设 30 秒超时，不修改全局设置或关闭位数限制。

python3 -I -B -X int_max_str_digits=4300 check_output_boundary.py
PASS: 12 output-boundary checks; integer digit limit=4300

#!/usr/bin/env python3
"""Synthetic output-contract checks; no production input or memory threshold."""
import json
from pathlib import Path
import subprocess
import sys
import tempfile

HERE = Path(__file__).resolve().parent
LIMIT = 4300


def record(group, digits):
    # Build text directly: fixture creation must not invoke the serializer under test.
    return '{"group":"' + group + '","units":' + digits + ',"payload":"x"}\n'


def main():
    if sys.get_int_max_str_digits() != LIMIT:
        raise SystemExit("Run with: python3 -I -B -X int_max_str_digits=4300 check_output_boundary.py")
    value = 10 ** LIMIT - 1
    one = record("alpha", "9" * LIMIT)
    expected = {"records": 1, "units": value,
                "groups": {"alpha": {"records": 1, "units": value}}}
    cases = (
        ("at_limit", one, None),
        ("same_group_sum", one + one, "result serialization failed:"),
        ("cross_group_total", one + record("beta", "9" * LIMIT),
         "result serialization failed:"),
        ("input_over_limit", record("alpha", "9" * (LIMIT + 1)),
         "line 1: invalid JSON"),
    )
    checks = 0
    with tempfile.TemporaryDirectory(prefix="jsonl-output-boundary-") as temp:
        path = Path(temp) / "synthetic.jsonl"
        for name, text, error in cases:
            path.write_text(text, encoding="utf-8")
            for mode in ("full", "stream", "materialized"):
                cp = subprocess.run(
                    [sys.executable, "-I", "-B", "-X", "int_max_str_digits=4300",
                     str(HERE / "lab.py"), "worker", mode, str(path)],
                    capture_output=True, text=True, encoding="utf-8", timeout=30)
                assert not cp.stderr, (name, mode, cp.returncode)
                outcome = json.loads(cp.stdout)
                if error is None:
                    assert cp.returncode == 0 and outcome["ok"] is True
                    assert outcome["result"] == expected
                else:
                    assert cp.returncode == 2 and outcome["ok"] is False
                    assert set(outcome) == {"ok", "error"}
                    assert error in outcome["error"], (name, mode)
                checks += 1
    print(f"PASS: {checks} output-boundary checks; integer digit limit={LIMIT}")


if __name__ == "__main__":
    main()

此次补充实跑通过原来的 79 项检查与新增 12 项输出边界检查。12 项不是内存或性能验收，不与原来的 33 项坏输入混为一类。遇到这种边界，应一起定义单条值、累计值和输出格式的业务上限，不应为了“跑通”而直接关闭解释器保护。

这一修正只覆盖这里复现的 JSON 编码 ValueError。MemoryError、深递归，以及 stdout 写入、flush、管道断开或磁盘失败仍未验证，也没有提供原子文件发布或外部事务保证。更完整的端到端内存预算还须另测序列化、输出缓冲和父进程；不能拿本文的聚合峰值代替它。

第一个边界实验只放一条记录：把 payload 从 512 个字符扩大到 2,097,152 个字符，流式峰值从 15,978 增加到 4,212,762 字节。文件按行迭代，仍需要拿到这一整行，然后产生解析对象。所以“按行读”不能为任意大的单行提供固定内存上限。

第二个边界实验让每条记录都有不同的 group。600 个分组与 6,000 个分组的流式峰值分别为 204,801 和 1,652,581 字节。输入并没有被整份留下，但聚合字典必须保存所有不同键，输出自身就在长大。四次边界观察都在独立进程里运行，每次只测一轮，不能与主表的三轮中位数混淆。

更实用的描述是：对于记录长度和分组数有界的输入，逐条聚合可避免峰值随总条数线性增加；当最大记录、分组键总长度或不同分组数增长时，内存仍会增长。累计整数的位数和解释器开销也不是严格常量。要处理高基数数据，可以评估分区后分批聚合、外部排序或数据库聚合，但应把这些作为新的设计与测量任务。

如果业务必须返回所有原始记录，或者全局排序要求把所有键留下，流式读取本身不会消除最终结果的体积。先核实输出合同，再决定是否值得改存储格式或算法，比为“流式”这个名字重写代码更有用。

Claude Code 官方最佳实践建议给出清晰的验证方式。针对这个任务，验证方式不应只是“内存小一些”，而应包含独立业务预期、输入规模、测量口径和负对照。可以按下面的模板填写你的仓库与文件范围。

请在已批准的项目目录中检查指定 JSONL 批处理文件。
先记录输入规则、输出字段和现有错误处理，不改变业务语义。
用手写的小夹具确定独立预期，不用当前实现的输出生成答案。
分别测原实现、流式实现，以及 list(生成器) 的负对照。
每次在独立 Python 进程运行，记录版本、输入哈希、命令和退出码。
明确 tracemalloc 从哪里启动、在哪里取数，分别报告 current 与 peak。
用多个输入规模，保留重复观察；不要把固定字节数当作通用通过门槛。
补空输入、Unicode、末行无换行、约定的坏输入及高基数/长行边界。
只修改获准文件，不安装依赖，不读取真实敏感数据，不提交或推送。
最后给出实际运行结果、失败项、未覆盖项，以及每个结论对应的证据。

验收时看两份东西：实现差异，以及能复跑的证据。若报告只有最终 current，要求补峰值；若只换成 yield，却在调用方保留 list()，要求解释消费端；若业务断言只比较新旧输出相同，要求回到手写预期。新旧实现完全可能共享同一个错误。

这份请求模板没有实际运行模型，也不承诺模型会自动正确选择测量点。在真实项目里，导入测试和执行脚本都可能运行项目代码，仍应使用你已经批准的环境、路径和命令范围。

只依赖标准库。run 子命令构造合成实验；worker 和 diagnose 供运行器在临时目录中调用。验证逻辑使用显式异常，不依赖 Python assert 是否被优化掉。

#!/usr/bin/env python3
"""Synthetic JSONL lab: stdlib only; independent worker per observation."""
import argparse
import hashlib
import json
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile
import tracemalloc

MODES = ("full", "stream", "materialized")
HERE = Path(__file__).resolve().parent


def unique_object(pairs):
    obj = {}
    for key, value in pairs:
        if key in obj:
            raise ValueError("duplicate object key")
        obj[key] = value
    return obj


def reject_constant(value):
    raise ValueError("non-standard JSON constant")


def parse_line(line, number):
    if not line.strip():
        raise ValueError(f"line {number}: blank record")
    try:
        row = json.loads(line, object_pairs_hook=unique_object,
                         parse_constant=reject_constant)
    except ValueError as exc:
        raise ValueError(f"line {number}: invalid JSON") from exc
    if not isinstance(row, dict) or set(row) != {"group", "units", "payload"}:
        raise ValueError(f"line {number}: expected group/units/payload object")
    if not isinstance(row["group"], str) or not row["group"]:
        raise ValueError(f"line {number}: group must be a nonempty string")
    if type(row["units"]) is not int or row["units"] < 0:
        raise ValueError(f"line {number}: units must be a nonnegative integer")
    if not isinstance(row["payload"], str):
        raise ValueError(f"line {number}: payload must be a string")
    return row


def records(path):
    with path.open("r", encoding="utf-8", newline=None) as handle:
        for number, line in enumerate(handle, 1):
            yield parse_line(line, number)


def aggregate(rows):
    result = {"records": 0, "units": 0, "groups": {}}
    for row in rows:
        result["records"] += 1
        result["units"] += row["units"]
        group = result["groups"].setdefault(row["group"], {"records": 0, "units": 0})
        group["records"] += 1
        group["units"] += row["units"]
    return result


def full_read(path, diagnostic=None):
    with path.open("r", encoding="utf-8", newline=None) as handle:
        text = handle.read()
    # Use physical LF boundaries after universal-newline translation.
    # str.splitlines() also splits U+2028 inside a valid JSON string.
    lines = text.split("\n")
    if lines[-1] == "":
        lines.pop()
    rows = [parse_line(line, n) for n, line in enumerate(lines, 1)]
    if diagnostic is not None:
        current, peak = tracemalloc.get_traced_memory()
        snapshot = tracemalloc.take_snapshot()
        diagnostic.update(stage="full: parsed rows retained, before aggregation",
                          current_bytes=current, peak_bytes=peak,
                          top=[{"file": s.traceback[0].filename,
                                "line": s.traceback[0].lineno,
                                "bytes": s.size, "blocks": s.count}
                               for s in snapshot.statistics("lineno")[:6]])
    return aggregate(rows)


def execute(mode, path):
    if mode == "full":
        return full_read(path)
    if mode == "stream":
        return aggregate(records(path))
    # Deliberate negative control: a generator is not enough.
    return aggregate(list(records(path)))


def worker(mode, path):
    # Imports and fixture generation are outside this measurement window.
    tracemalloc.start(1)
    try:
        result = execute(mode, path)
        current, peak = tracemalloc.get_traced_memory()
    except (ValueError, UnicodeError, OSError) as exc:
        tracemalloc.stop()
        print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=True))
        return 2
    tracemalloc.stop()
    # Encoding stays outside the allocation window, but inside the failure contract.
    try:
        encoded = json.dumps({"ok": True, "result": result,
                              "current_bytes": current, "peak_bytes": peak},
                             ensure_ascii=True, sort_keys=True)
    except ValueError as exc:
        print(json.dumps({"ok": False, "error": f"result serialization failed: {exc}"},
                         ensure_ascii=True))
        return 2
    print(encoded)
    return 0


def child(root, mode, path, expected):
    cp = subprocess.run([sys.executable, "-I", "-B", str(root / "lab.py"),
                         "worker", mode, str(path)], cwd=root,
                        capture_output=True, text=True, encoding="utf-8", check=False)
    if cp.returncode != 0:
        raise RuntimeError(f"worker {mode} failed: {cp.stdout} {cp.stderr}")
    observation = json.loads(cp.stdout)
    if not observation["ok"] or observation["result"] != expected:
        raise AssertionError(f"business result differs from expected: {path.name}/{mode}")
    return {"mode": mode, "rc": cp.returncode, **observation}


def scaled_expected(base, cycles):
    # Scale handwritten six-record totals; never ask a tested mode for the oracle.
    return {"records": base["records"] * cycles, "units": base["units"] * cycles,
            "groups": {k: {field: value * cycles for field, value in v.items()}
                       for k, v in base["groups"].items()}}


def make_bulk(path, count, payload_size=512, unique_groups=False):
    pattern = ("alpha", "beta", "alpha", "gamma", "beta", "gamma")
    with path.open("w", encoding="utf-8", newline="\n") as handle:
        for i in range(count):
            row = {"group": f"g{i:06d}" if unique_groups else pattern[i % 6],
                   "units": i % 6 + 1, "payload": "x" * payload_size}
            handle.write(json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n")


def run(out):
    # Fail before any experiments if the output already exists.
    out.mkdir(parents=True, exist_ok=False)
    expected = json.loads((HERE / "expected.json").read_text(encoding="utf-8"))
    report = {"environment": {"python": sys.version, "implementation": platform.python_implementation(),
              "platform": platform.platform(), "machine": platform.machine(),
              "traced_frames": 1, "worker_flags": ["-I", "-B"],
              "hash_seed": "interpreter default; fresh process for every observation"},
              "measurements": [], "fixtures": {}, "invalid": [], "limits": [],
              "source_sha256": {str(p.relative_to(HERE)): hashlib.sha256(p.read_bytes()).hexdigest()
                                for p in ([HERE / name for name in ("lab.py", "make_fixtures.py", "expected.json")]
                                          + [HERE / "fixtures" / (name + ".jsonl")
                                             for name in ("manual", "empty", "unicode", "lastline", "crlf")])}}
    with tempfile.TemporaryDirectory(prefix="python-jsonl-memory-") as temp:
        root = Path(temp)
        shutil.copyfile(HERE / "lab.py", root / "lab.py")
        for name in ("manual", "empty", "unicode", "lastline", "crlf"):
            path = root / (name + ".jsonl")
            shutil.copyfile(HERE / "fixtures" / path.name, path)
            report["fixtures"][name] = [child(root, mode, path, expected[name]) for mode in MODES]
        for n in (600, 6000, 60000):
            path = root / f"bulk-{n}.jsonl"
            make_bulk(path, n)
            oracle = scaled_expected(expected["manual"], n // 6)
            digest = hashlib.sha256(path.read_bytes()).hexdigest()
            for trial in (1, 2, 3):
                for mode in MODES:
                    report["measurements"].append({"n": n, "trial": trial,
                        "input_bytes": path.stat().st_size, "input_sha256": digest,
                        **child(root, mode, path, oracle)})
        cp = subprocess.run([sys.executable, "-I", "-B", str(root / "lab.py"),
             "diagnose", str(root / "bulk-6000.jsonl")], cwd=root, check=True,
             capture_output=True, text=True, encoding="utf-8")
        report["diagnostic"] = json.loads(cp.stdout.replace(str(root), "<synthetic-root>"))
        for case in expected["invalid"]:
            path = root / "bad.jsonl"
            path.write_bytes(bytes.fromhex(case["hex"]) if "hex" in case else case["text"].encode("utf-8"))
            for mode in MODES:
                cp = subprocess.run([sys.executable, "-I", "-B", str(root / "lab.py"),
                    "worker", mode, str(path)], cwd=root, capture_output=True,
                    text=True, encoding="utf-8", check=False)
                result = json.loads(cp.stdout)
                if cp.returncode != 2 or result["ok"] or case["contains"] not in result["error"]:
                    raise AssertionError(f"wrong invalid-input behavior: {case['name']}/{mode}")
                report["invalid"].append({"case": case["name"], "mode": mode,
                                         "rc": cp.returncode, **result})
        for width in (512, 2 * 1024 * 1024):
            path = root / f"one-{width}.jsonl"
            make_bulk(path, 1, payload_size=width)
            oracle = {"records": 1, "units": 1, "groups": {"alpha": {"records": 1, "units": 1}}}
            report["limits"].append({"case": "one long record", "payload_characters": width,
                                    **child(root, "stream", path, oracle)})
        for n in (600, 6000):
            path = root / f"groups-{n}.jsonl"
            make_bulk(path, n, unique_groups=True)
            oracle = {"records": n, "units": (n // 6) * 21,
                      "groups": {f"g{i:06d}": {"records": 1, "units": i % 6 + 1} for i in range(n)}}
            observation = child(root, "stream", path, oracle)
            # Keep the full worker result checked above; compact only the saved limit report.
            observation["result"] = {"records": n, "units": oracle["units"], "group_count": n}
            report["limits"].append({"case": "many groups", "group_count": n, **observation})
    report["checks"] = {"fixture_mode_checks": 15, "bulk_mode_checks": 27,
                        "invalid_mode_checks": len(report["invalid"]), "limit_checks": 4,
                        "memory_threshold_assertions": 0, "business_results_passed": True}
    (out / "report.json").write_text(json.dumps(report, ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
    print("PASS: 15 fixture + 27 bulk + 4 limit business-result checks")
    print(f"PASS: {len(report['invalid'])} malformed-input checks; no memory byte threshold used")
    print("MEASURED: current/peak traced Python allocations; not RSS or runtime")
    print(f"REPORT: {out / 'report.json'}")


def main():
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="action", required=True)
    p = sub.add_parser("run")
    p.add_argument("output", type=Path)
    p = sub.add_parser("worker")
    p.add_argument("mode", choices=MODES)
    p.add_argument("input", type=Path)
    p = sub.add_parser("diagnose")
    p.add_argument("input", type=Path)
    args = parser.parse_args()
    if args.action == "worker":
        return worker(args.mode, args.input)
    if args.action == "diagnose":
        diagnostic = {}
        tracemalloc.start(1)
        full_read(args.input, diagnostic)
        tracemalloc.stop()
        print(json.dumps(diagnostic, ensure_ascii=True, sort_keys=True))
        return 0
    run(args.output.resolve())
    return 0


if __name__ == "__main__":
    sys.exit(main())

从正文复刻时运行一次。Unicode 字符串里的 Python 转义 \u2028 会写入一个真实的 U+2028 字符，专门检查“物理行”与 str.splitlines() 的差异。程序生成的文件字节已经与本次实验的原始夹具逐份核对一致。

#!/usr/bin/env python3
"""Create only synthetic fixtures beside this file; refuses existing directory."""
from pathlib import Path

root = Path(__file__).resolve().parent / "fixtures"
root.mkdir(exist_ok=False)
manual = (
    '{"group":"alpha","units":1,"payload":"a"}\n'
    '{"group":"beta","units":2,"payload":"b"}\n'
    '{"group":"alpha","units":3,"payload":"c"}\n'
    '{"group":"gamma","units":4,"payload":"d"}\n'
    '{"group":"beta","units":5,"payload":"e"}\n'
    '{"group":"gamma","units":6,"payload":"f"}\n'
)
unicode_records = (
    '{"group":"中文","units":2,"payload":"é🙂\u2028"}\n'
    '{"group":"café","units":3,"payload":"简体中文"}\n'
    '{"group":"中文","units":0,"payload":"末行无换行"}'
)
(root / "manual.jsonl").write_bytes(manual.encode("utf-8"))
(root / "empty.jsonl").write_bytes(b"")
(root / "unicode.jsonl").write_bytes(unicode_records.encode("utf-8"))
(root / "lastline.jsonl").write_bytes(b'{"group":"alpha","units":1,"payload":"last line"}')
(root / "crlf.jsonl").write_bytes(manual.replace("\n", "\r\n").encode("utf-8"))
print("Created five UTF-8 fixtures; unicode and lastline have no final newline")

这里保存手算结果和错误样例。大规模预期只根据 manual 的六条总数缩放；不要用报告反向覆盖这个文件。

{
  "manual": {
    "records": 6,
    "units": 21,
    "groups": {
      "alpha": {
        "records": 2,
        "units": 4
      },
      "beta": {
        "records": 2,
        "units": 7
      },
      "gamma": {
        "records": 2,
        "units": 10
      }
    }
  },
  "empty": {
    "records": 0,
    "units": 0,
    "groups": {}
  },
  "unicode": {
    "records": 3,
    "units": 5,
    "groups": {
      "中文": {
        "records": 2,
        "units": 2
      },
      "café": {
        "records": 1,
        "units": 3
      }
    }
  },
  "lastline": {
    "records": 1,
    "units": 1,
    "groups": {
      "alpha": {
        "records": 1,
        "units": 1
      }
    }
  },
  "crlf": {
    "records": 6,
    "units": 21,
    "groups": {
      "alpha": {
        "records": 2,
        "units": 4
      },
      "beta": {
        "records": 2,
        "units": 7
      },
      "gamma": {
        "records": 2,
        "units": 10
      }
    }
  },
  "invalid": [
    {
      "name": "blank",
      "text": "\n",
      "contains": "line 1: blank record"
    },
    {
      "name": "syntax",
      "text": "{oops}\n",
      "contains": "line 1: invalid JSON"
    },
    {
      "name": "missing",
      "text": "{\"group\":\"alpha\",\"units\":1}\n",
      "contains": "expected group/units/payload object"
    },
    {
      "name": "boolean",
      "text": "{\"group\":\"alpha\",\"units\":true,\"payload\":\"x\"}\n",
      "contains": "units must be a nonnegative integer"
    },
    {
      "name": "negative",
      "text": "{\"group\":\"alpha\",\"units\":-1,\"payload\":\"x\"}\n",
      "contains": "units must be a nonnegative integer"
    },
    {
      "name": "duplicate",
      "text": "{\"group\":\"alpha\",\"group\":\"beta\",\"units\":1,\"payload\":\"x\"}\n",
      "contains": "invalid JSON"
    },
    {
      "name": "nan",
      "text": "{\"group\":\"alpha\",\"units\":NaN,\"payload\":\"x\"}\n",
      "contains": "invalid JSON"
    },
    {
      "name": "second_line",
      "text": "{\"group\":\"alpha\",\"units\":1,\"payload\":\"last line\"}\n{oops}\n",
      "contains": "line 2: invalid JSON"
    },
    {
      "name": "array",
      "text": "[]\n",
      "contains": "expected group/units/payload object"
    },
    {
      "name": "extra",
      "text": "{\"group\":\"alpha\",\"units\":1,\"payload\":\"x\",\"extra\":1}\n",
      "contains": "expected group/units/payload object"
    },
    {
      "name": "utf8",
      "hex": "ff0a",
      "contains": "utf-8"
    }
  ]
}

一次可接受的改动，应先保证独立预期全部通过，再用同一输入形状比较多个规模，最后让读者知道哪些边界还会增长。本文展示的是一条具体的改造路径：移除无用的全量输入保留，同时保留真实需要的聚合状态。它不构成整进程的内存预算，也没有证明 CPU、I/O 或端到端时延改善。

本文核对的是 Python 3.12 文档分支，官方页面当前显示 3.12.15，实验解释器为 3.12.14。上述 API 定义的来源是官方文档；表格、图、夹具、程序和数值来自本文的合成实验。生产版本不同，应保留自己的环境记录。

如果你还没有为旧程序固定行为，可以先看给旧 CSV 导出器补特征测试；如果测试结果还没有绑定到准备提交的版本，再看工作区与暂存区的测试证据。这两个问题处理好，内存优化的“结果不变”才有明确对象。