流式回复中断后的恢复

用户盯着进度条，看见回答已经写到 99%，突然停住了。前端显示连接断开，后端日志却说请求还在运行。此时最容易写出的处理是再请求一次，然后把新回答直接接到旧回答后面。这样做经常带来三个问题：内容重复、工具被执行两次、用户以为已经完成的答案其实没有结束。

这篇文章沿用服务台助手案例，但不再增加新的业务工具。我们只解决一个工程问题：流式输出进行到一半时，如何保存现场、判断断点类型，并在不重复副作用的前提下恢复。文章仍然采用小步练习，你可以把每一步复制到自己的目录中运行，不需要先搭一个完整 Web 项目。

先分清三种“断了”

先不要急着写 retry。至少有三层状态需要分别判断：

层次	发生了什么	能不能原地继续
传输层	浏览器、网关或 HTTP 连接断开	只有在服务端支持事件游标时才可能恢复未收到的事件
模型生成层	模型还没结束，服务端任务被取消或超时	通常不能从某个 token 原地恢复，只能重新请求续写
业务执行层	工具已执行、参数只生成一半或结构化结果不完整	必须根据执行记录决定重试、补回答还是丢弃调用

把这三层混在一个 except 里，后面很难保证恢复正确。本文会先保存现场，再根据状态选择动作。

第 1 步：为一次流式请求建档

这一小步要完成什么

先建立一个运行记录，保存已经收到的内容和当前状态。

from dataclasses import dataclass, field
from typing import Literal


RunStatus = Literal[
    "running",
    "completed",
    "interrupted",
    "retrying",
    "failed",
]


@dataclass
class StreamRun:
    run_id: str
    status: RunStatus = "running"
    next_sequence: int = 0
    chunks: list[str] = field(default_factory=list)
    finish_reason: str | None = None
    last_event_at: float | None = None
    tool_call_buffer: dict[str, str] = field(default_factory=dict)
    tool_executed: dict[str, str] = field(default_factory=dict)

    @property
    def text(self) -> str:
        return "".join(self.chunks)

这里先关注几个字段：

• next_sequence 用来判断事件是否重复或乱序；
• chunks 保存已经收到的文本片段，不能等最后一个事件才写入；
• finish_reason 用来区分正常结束、长度限制和中断；
• tool_call_buffer 保存尚未拼完整的工具参数；
• tool_executed 记录已经产生副作用的调用，恢复时不能只看模型消息重新猜。

内存对象会随进程重启丢失。需要跨进程恢复时，应把这些内容写入 Redis、数据库或任务存储。

运行观察

run = StreamRun(run_id="run-1001")
run.chunks.append("VPN 的申请流程是")
print(run.text)
print(run.status, run.next_sequence)

现在输出的是未完成草稿。它还不能被标记成最终回答，也不能据此判断模型已经完成。

动手改一改

给 StreamRun 增加 user_id 和 tenant_id，想一想为什么恢复请求不能只凭一个容易猜到的 run_id 找回内容。

第 2 步：每收到一个事件就保存一次

这一小步要完成什么

把流式客户端想成一个不断收到事件的循环。不同 SDK 的事件类型和字段名称不完全一样，先用统一的内部事件表示，避免业务代码直接依赖某一家 SDK 的细节。

from dataclasses import dataclass


@dataclass
class StreamEvent:
    sequence: int
    text_delta: str = ""
    event_id: str | None = None
    finish_reason: str | None = None
    tool_name: str | None = None
    tool_arguments_delta: str = ""

写一个只负责更新状态的函数：

def apply_event(run: StreamRun, event: StreamEvent, now: float) -> None:
    if event.sequence < run.next_sequence:
        return  # 重连后收到的旧事件，丢弃
    if event.sequence > run.next_sequence:
        raise ValueError("event sequence 有缺口，不能静默拼接")

    if event.text_delta:
        run.chunks.append(event.text_delta)

    if event.tool_name and event.tool_arguments_delta:
        previous = run.tool_call_buffer.get(event.tool_name, "")
        run.tool_call_buffer[event.tool_name] = (
            previous + event.tool_arguments_delta
        )

    run.finish_reason = event.finish_reason or run.finish_reason
    run.last_event_at = now
    run.next_sequence += 1

这段代码做了两件重要的事：事件序号有缺口时暂停拼接，旧事件重复到达时不重复追加。实际 API 如果没有序号，可以在自己的服务端接收层生成序号；如果供应商提供稳定的事件 ID，则同时保存事件 ID。

运行观察

events = [
    StreamEvent(sequence=0, text_delta="VPN 的申请流程是"),
    StreamEvent(sequence=1, text_delta="先提交权限申请。"),
]
run = StreamRun(run_id="run-1002")
for event in events:
    apply_event(run, event, now=1.0 + event.sequence)

print(run.text)
print(run.next_sequence)

再把第一条事件重复放进列表。输出不应多出一遍“VPN 的申请流程是”。把 sequence=1 改成 sequence=3，程序应该报缺口，而不是把中间丢失的内容假装不存在。

常见错误

• 只把文本追加到前端，不在服务端保存，浏览器一断就没有恢复依据；
• 把事件到达顺序当成模型生成顺序，重连或并发转发后可能出现乱序；
• 收到 finish_reason 就立即认为答案可靠，忽略结构化输出和工具调用是否完整。

第 3 步：用结束信号判断是否真的完成

这一小步要完成什么

流式事件结束不等于模型任务正常结束。把结束判断集中在一个函数里：

def classify_end(run: StreamRun) -> str:
    if run.finish_reason == "stop":
        if run.tool_call_buffer:
            return "interrupted_tool_call"
        return "completed"
    if run.finish_reason in {"length", "max_tokens"}:
        return "truncated_by_limit"
    if run.status == "interrupted":
        return "transport_interrupted"
    return "still_running"

不同供应商的 finish_reason 名称可能不同，先在适配层归一化，再让业务代码使用自己的状态。这里故意把“正常结束但工具参数还没拼完”视为异常，因为不能把不完整参数交给工具。

运行观察

分别构造三种状态：

run.finish_reason = "stop"
print(classify_end(run))

run.finish_reason = "length"
print(classify_end(run))

run.finish_reason = None
run.status = "interrupted"
print(classify_end(run))

你应该得到 completed、truncated_by_limit 和 transport_interrupted。如果所有情况最后都只显示“失败”，后续就无法决定该重连还是续写。

动手改一改

把“网关主动超时”和“用户点击停止”分别加入状态。两者都可能没有 finish_reason，但前者通常允许自动恢复，后者应该尊重用户停止意图。

第 4 步：先判断是重连，还是重新请求续写

连接可以恢复，不代表模型可以恢复

如果你的事件服务支持游标，可以尝试从最后一个已确认事件之后继续接收：

last_sequence = run.next_sequence - 1
headers = {"Last-Event-ID": str(last_sequence)}
# 重新连接事件服务，并让服务端从 last_sequence + 1 开始发送

这只适用于“模型任务仍然存在，服务端还保留事件”的架构。很多直接调用模型 API 的场景没有这个能力，重新发起 HTTP 请求只会创建一次新的生成任务。

因此恢复代码应先询问任务状态：

def choose_recovery(run: StreamRun, provider_supports_resume: bool) -> str:
    kind = classify_end(run)
    if kind == "transport_interrupted" and provider_supports_resume:
        return "resume_events"
    if kind in {"transport_interrupted", "truncated_by_limit"}:
        return "request_continuation"
    if kind == "interrupted_tool_call":
        return "restart_tool_call"
    if kind == "completed":
        return "show_result"
    return "inspect_task"

不要把 provider_supports_resume=True 写死在业务代码里。它属于模型供应商和网关的能力配置，而且要经过实际断线测试。

运行观察

为同一个 run 分别传入 True 和 False，观察动作从 resume_events 变成 request_continuation。这两个动作的成本和重复风险不同，日志里要记录最终选择了哪一种。

第 5 步：普通文本用“续写请求”接回来

先保存恢复所需的事实

续写请求至少需要这些信息：

continuation_context = {
    "original_user_input": "请说明 VPN 申请流程",
    "system_prompt_version": "service-desk-v3",
    "confirmed_tool_results": [],
    "partial_text": run.text,
    "last_sequence": run.next_sequence - 1,
}

partial_text 是未完成草稿，不是可信事实。真正的事实应该来自用户输入、已经完成的工具结果和业务状态。把草稿单独存出来，也能避免恢复时把它错误地当成一条完整 assistant 消息。

组织一个明确的 continuation 请求

不同模型 API 的消息格式不同，先用伪代码表达控制边界：

def build_continuation_prompt(ctx: dict) -> str:
    return f"""
你正在恢复一段被中断的回答。
原始问题：{ctx['original_user_input']}
已经生成但未完成的草稿：{ctx['partial_text']}

请只补上草稿后面缺失的内容，不要重复已有句子。
如果信息不足，请明确说明，不要重新执行已经完成的工具。
""".strip()

真正发送请求时，应继续使用原来的 system prompt、模型和工具权限，并把已确认的工具结果作为事实输入。不要只把 partial text 贴进一个全新的对话，那里可能丢失原始问题、权限和工具状态。

追加前做重叠检测

续写模型可能再次输出“先提交权限申请”这几个字。先在服务端合并，再把结果推给前端：

def append_without_simple_overlap(prefix: str, suffix: str) -> str:
    max_overlap = min(len(prefix), len(suffix), 80)
    for size in range(max_overlap, 0, -1):
        if prefix.endswith(suffix[:size]):
            return prefix + suffix[size:]
    return prefix + suffix

这个函数只匹配首尾完全相同的字符，不处理标点变化、Markdown 代码块、分词边界或模型重复改写。宁可保留一个短重复，也不要用过于激进的规则误删原文。

运行观察

先手动设置：

print(append_without_simple_overlap("先提交权限申请。", "先提交权限申请。然后等待审批。"))

再把续写结果故意从句子中间开始，观察简单算法为什么不一定能处理。这一步的目的不是造一个完美去重器，而是让你把“续写结果不能直接原样拼接”纳入设计。

第 6 步：工具调用或 JSON 中断时，不要续写字符

普通文本可以尝试续写，工具调用要先判断结构是否完整：

import json


def parse_complete_tool_call(run: StreamRun, tool_name: str) -> dict | None:
    raw = run.tool_call_buffer.get(tool_name, "")
    try:
        value = json.loads(raw)
    except json.JSONDecodeError:
        return None
    if not isinstance(value, dict):
        return None
    return value

如果返回 None，不要把半截 JSON 传给 create_ticket。正确的恢复顺序是：

1. 丢弃这次不完整的工具调用缓存；
2. 根据原始用户请求重新请求一个完整的工具调用；
3. 对工具名、参数 schema、用户身份和权限重新校验；
4. 只有校验通过才执行工具。

如果 tool_executed 已经记录了调用 ID 或幂等键，则恢复时先查执行结果。工具已经成功时，只补最终回答，不要再次调用。

运行观察

准备两个参数片段：

run.tool_call_buffer["create_ticket"] = '{"title":"VPN 权限'
print(parse_complete_tool_call(run, "create_ticket"))

run.tool_call_buffer["create_ticket"] = (
    '{"title":"VPN 权限申请","priority":"high"}'
)
print(parse_complete_tool_call(run, "create_ticket"))

第一个结果应该是 None，第二个才允许进入后续 schema 和权限校验。JSON 能解析也不代表业务允许执行，两个检查不能省略。

第 7 步：给恢复过程加幂等和硬限制

副作用工具要有幂等键

假设用户请求“创建 VPN 工单”，生成到 99% 时连接断开。你不能因为没有收到最终文本，就默认工具没有执行。为一次业务动作生成稳定的幂等键：

import hashlib


def idempotency_key(run_id: str, tool_name: str, arguments: dict) -> str:
    raw = f"{run_id}:{tool_name}:{sorted(arguments.items())}"
    return hashlib.sha256(raw.encode()).hexdigest()

真实工具执行前，先用这个键查询执行记录；已成功就返回原结果，处理中就等待或查询状态，不存在才创建新任务。幂等键不能替代权限检查，也不能把不稳定的模型文本直接作为唯一输入。

自动恢复要有上限

MAX_RECOVERIES = 2
MAX_TOTAL_SECONDS = 45
MAX_CONTINUATION_CHARS = 4000


def can_recover(recovery_count: int, elapsed: float) -> bool:
    return (
        recovery_count < MAX_RECOVERIES
        and elapsed < MAX_TOTAL_SECONDS
    )

网络暂时中断可以指数退避；认证失败、权限错误、参数错误和用户主动停止不应盲目重试。达到上限后，保留已收到的内容，并把状态改成可供用户手动继续的 failed，不要无限自动请求。

动手改一改

把 MAX_RECOVERIES 改成 0、1、3，记录三种情况下的用户体验和成本。再想一想：实时客服和长篇报告是否应该使用同一个总超时？通常不会。

第 8 步：把恢复状态接回三种实现

前面三篇文章的框架边界可以这样对应：

实现	恢复状态放在哪里	仍然需要自己负责什么
Raw API	自己维护 StreamRun、事件序号和任务存储	事件去重、续写请求、幂等、重试和权限
LangGraph	把 status、pending_action、tool_executed 放进 State，并在节点间 checkpoint	State 字段设计、恢复入口、执行幂等和恢复预算
Agent SDK	使用 SDK 的 session、事件和 tracing 能力，另存业务恢复记录	供应商是否支持断点、工具结果一致性、敏感信息和副作用保护

框架可以帮你保存会话或暴露事件，但不会自动知道“这次工单到底创建成功没有”。凡是会改变外部系统的动作，都要以业务服务的执行记录为准，而不是以最后一段模型文本为准。

第 9 步：设计用户看到的恢复体验

自动恢复期间，前端不要把旧内容清空再重新刷一遍。可以采用这样的状态：

已经收到的回答保留
        ↓
显示“连接中断，正在恢复”
        ↓
恢复成功：只追加新片段
恢复失败：保留已有内容，提供“继续生成”

服务端给前端发送的事件至少要区分：

• text_delta：新的文本片段；
• status：正在生成、正在恢复、已完成、失败；
• tool_pending：等待工具或人工确认；
• error：可展示的错误码和下一步动作。

不要把内部异常堆栈、API Key、完整提示词或未经脱敏的工具参数发给浏览器。用户需要知道的是当前状态和可执行的下一步。

第 10 步：用故障注入验证恢复

不要只测试“网络正常时能否输出”。在自己的练习代码中加入一个故障开关，在不同位置主动抛出中断：

def maybe_interrupt(sequence: int, stop_at: int | None) -> None:
    if stop_at is not None and sequence == stop_at:
        raise ConnectionError(f"模拟在事件 {sequence} 断开")

至少测试这些情况：

1. 普通文本生成到一半断开，续写后不重复整段；
2. 断在 finish_reason 之前，状态能标记为未完成；
3. 事件重复到达，不会重复追加；
4. 事件序号出现缺口，不会静默拼接；
5. 工具参数只收到一半，不会执行工具；
6. 工具已经成功后回答断开，不会再次执行工具；
7. 连续恢复超过上限，能够保留部分内容并结束；
8. 用户主动停止，不会被后台自动续写。

记录每次测试的 run_id、断点位置、恢复动作、追加字符数、工具调用数和最终状态。只有“看起来恢复了”还不够，你要能回答恢复多花了多少时间和费用，以及有没有重复副作用。

常见误区

• 把重新连接当成模型断点续传：先确认供应商是否真的支持事件恢复。
• 收到 99% 就直接标记成功：没有完成信号、完整 JSON 或工具执行记录时，答案仍可能不完整。
• 把半段回答直接当 assistant 历史：它可能包含未完成句子、错误判断或未闭合的 Markdown。
• 重试时重新执行所有工具：先查幂等记录，已经成功的副作用只读取结果。
• 让模型决定是否无限续写：恢复次数、时间、费用和工具次数必须由程序限制。
• 只修前端按钮：前端可以提供“继续生成”，但服务端必须先保存事件和业务状态。

本篇结束时你应该能说明什么

拿着自己的练习代码，尝试回答：

• 这次断开属于传输、模型生成还是业务执行层？
• 已经收到的片段保存在哪里，如何判断是否有事件缺口？
• 普通文本、半截 JSON 和已执行工具分别怎样恢复？
• 自动恢复什么时候停止，用户如何拿回控制权？

如果这些问题还答不清，先回到第 2 步打印事件序号，再故意在不同位置注入中断。恢复问题通常不是某个 SDK 参数没填对，而是系统没有把“已经发生了什么”记录下来。