综合小项目：用 Python 构建可中断保存的 AI 流式回答器

流式输出能让用户尽快看到模型开始回答，但也带来一个容易被忽略的问题：回答还没结束时，终端断开、进程被中止或网络异常，已经收到的内容可能只停留在内存里。本篇围绕一个核心问题构建小项目：如何把流式增量安全地保存到文件，并区分“部分结果”和“完整结果”。程序不会假装中断后可以从模型端续传，而是保留已经收到的前缀，方便查看、重试和排查。

先定义保存边界

项目输入是一段提示词，输出是一个 Markdown 文件。生成过程中先写入同名的 .part 文件，例如目标是 answer.md，临时文件就是 answer.md.part。每收到一个文本增量，就追加到临时文件并刷新；只有流正常结束，程序才把临时文件改名为最终文件。

这个顺序很重要。若一开始就写 answer.md，其他程序很难判断它是完整回答还是半截回答；若只在内存中累加，进程异常时又会丢掉已经收到的内容。因此可以把文件状态约定为：存在 .part 表示“生成中或曾经中断”，存在最终文件表示“本次流已正常结束”。改名操作只发生在完整结果确认之后。

需要明确的是，保存前缀不等于断点续传。重新请求模型时，模型通常会重新生成一遍，程序不能把旧的 .part 文件直接拼接成一次新的模型响应。旧文件的价值是保留现场、减少信息损失，并帮助人工决定是否重试。

准备环境和配置

示例使用 OpenAI Python SDK 的 Responses API 流式接口。密钥和模型名都从环境变量读取，不写进源代码：

1
2
3

python -m pip install -U openai
export OPENAI_API_KEY="替换为你的真实密钥"
export MODEL_NAME="替换为你可用的模型名称"

如果暂时没有密钥，代码还提供 --self-test，它只用本地固定分片验证落盘和最终改名逻辑；那不是模型运行结果，只是离线控制流测试。

编写最小流式保存器

把下面代码保存为 stream_save.py。Responses API 的流中会包含多种事件，程序只处理类型为 response.output_text.delta 的文本增量，其余事件忽略。os.fsync() 用来把已经写入的内容尽量落到磁盘；它会增加一点开销，所以真实批处理可以按固定间隔检查点，而不是每个分片都同步。

import argparse
import os
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import Iterable

from openai import OpenAI


TEXT_DELTA = "response.output_text.delta"


def sync_file(handle) -> None:
    handle.flush()
    os.fsync(handle.fileno())


def save_stream(events: Iterable, target: Path) -> None:
    partial = target.with_name(target.name + ".part")
    target.parent.mkdir(parents=True, exist_ok=True)

    with partial.open("w", encoding="utf-8") as output:
        try:
            for event in events:
                if getattr(event, "type", None) != TEXT_DELTA:
                    continue
                piece = getattr(event, "delta", "")
                if not piece:
                    continue
                output.write(piece)
                output.flush()
                print(piece, end="", flush=True)
        except BaseException:
            sync_file(output)
            raise
        sync_file(output)

    # 只有循环正常结束，才把部分文件提升为完整文件。
    partial.replace(target)


def model_events(prompt: str):
    client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
    return client.responses.create(
        model=os.environ["MODEL_NAME"],
        input=prompt,
        stream=True,
    )


def demo_events():
    for piece in ("这是", "一个离线", "流式保存测试。"):
        yield SimpleNamespace(type=TEXT_DELTA, delta=piece)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("prompt", nargs="?", default="用三句话解释为什么要保存流式输出。")
    parser.add_argument("--output", default="answer.md")
    parser.add_argument("--self-test", action="store_true")
    args = parser.parse_args()
    target = Path(args.output)

    try:
        events = demo_events() if args.self_test else model_events(args.prompt)
        save_stream(events, target)
        print(f"\n已保存完整回答：{target}")
        return 0
    except KeyboardInterrupt:
        print(f"\n已中断，部分回答保存在：{target}.part", file=sys.stderr)
        return 130
    except Exception as exc:
        print(f"\n请求失败：{type(exc).__name__}: {exc}", file=sys.stderr)
        print(f"部分回答保存在：{target}.part", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

先做不联网的检查：

1 2	python stream_save.py --self-test --output build/answer.md cat build/answer.md

这次测试应该生成 build/answer.md，内容来自代码中的三个固定分片。真实调用时再设置环境变量并运行：

1	python stream_save.py "把流式输出解释给刚学 Python 的读者" --output answer.md

终端会边接收边显示文字，同时把相同的增量写入 answer.md.part。流正常结束后，.part 被改名为 answer.md；如果按下 Ctrl-C 或请求抛出异常，临时文件会保留已经收到的前缀。

为什么要用临时文件

这里的临时文件不是为了隐藏错误，而是为了让状态可观察。下游脚本只处理没有 .part 且存在最终文件的结果，就不会误把半截回答当成完整文档。Path.replace() 在同一文件系统内是一个简单的原子切换：读者要么看到旧的最终文件，要么看到新的完整文件，不会在切换过程中看到“写到一半的新文件”。

如果目标文件已经存在，本示例会在成功结束时替换它。更谨慎的业务可以先禁止覆盖，或把时间戳、请求编号和模型名放进文件名。无论采用哪种命名方式，都应让“请求开始”“流正常结束”“请求异常”这些状态能够被日志或文件名区分。

flush() 和 fsync() 也不是绝对的数据安全保证。操作系统、文件系统和远程磁盘仍可能有自己的缓存；它们的作用是降低进程异常时最近一小段内容尚未落盘的概率。若每个分片都同步造成明显变慢，可以累计若干字符或按秒执行一次检查点，但要接受最近检查点之后的内容可能丢失。

常见问题

中断后能不能从 .part 继续生成？ 通常不能直接续上同一条服务端流。可以把 .part 作为人工参考，或者在新的请求中明确要求模型继续，但新请求仍可能重复、改写或改变上下文，不能把它当作严格的断点续传。

为什么最终文件没有生成？ 最终改名只在事件循环正常结束后执行。网络错误、鉴权失败、用户中断或解析异常都会保留 .part，先查看临时文件和异常类型，再决定是否重试。

为什么不能把所有事件都写入文件？ Responses API 流里除了文本增量，还可能有创建、完成、工具或错误等事件。最终 Markdown 只需要文本内容；状态和调试信息应单独写日志，否则会把协议元数据混进用户回答。

每个分片都 fsync() 会不会太慢？ 会有额外磁盘开销。示例为了突出可靠性采用每片同步；生产程序可以改成定时检查点，并在正常结束前再强制同步一次。检查点间隔应根据回答长度、磁盘性能和可接受的数据丢失窗口来选择。

离线 --self-test 是否证明模型调用成功？ 不是。它只证明本地保存、异常边界和文件改名逻辑能运行。真实模型调用仍需使用实际环境变量，并单独观察网络、鉴权和服务端返回。

小结

这个小项目只解决一个边界清晰的问题：流式回答到达时立即落盘，但在正常结束前始终标记为 .part。文本增量由 response.output_text.delta 事件提供，程序只保存文本事件；flush() 与 fsync() 尽量保留已收到的前缀；只有流正常结束，才用改名把临时文件提升为最终结果。这样既保留了流式交互的即时性，也让中断状态可以被发现、检查和重试，而不会把半截内容伪装成完整答案。

搜索时光笔记