流式取消与人审

生产 Agent 的交互协议是事件流，不是一次 JSON 响应。本篇写事件、取消和人审怎么恢复；图上的 interrupt_before / thread_id 见 LangGraph 编排实践；什么必须人审见 Agent 安全与权限；HTTP 直觉见 API 设计与 HTTP 实践；卡住时的保护见限流熔断与降级。

为什么必须流式

诉求	一次返回 JSON	事件流
首字时间（TTFT）	等整段生成完	先推 token
用户点「停止」	后端仍跑完	可取消
工具调用可见	结束后才知道调了什么	`tool_call` 当时就能画卡片
人审	同步阻塞到超时	`interrupt` 后连接可结束，批准再 resume

最小事件类型

token        增量文本
tool_call    模型要调工具（name + args，尚未执行）
tool_result  工具已执行（或校验失败）
interrupt    等人审，携带 tool + args + 过期时间
done         正常结束（含拒答）
error        编排/模型/工具不可恢复失败
cancelled    用户或系统取消，半截不算成功

一次请求里可以交错：若干 token → tool_call → tool_result → 再 token → done。人审则在 tool_call 之后发 interrupt，不要先执行再通知。

传输选型

浏览器到自家后端：优先 SSE（实现简单、反向代理好配）。需要客户端频繁上行（协同编辑、语音帧）再用 WebSocket。不要为 Agent 聊天先上 Socket 框架。

限流打在这条流上：同一 thread_id 同时只允许一条 in-flight 生成，避免双开 resume。

后端：事件与人审状态

# events.py
from enum import Enum


class Event(str, Enum):
    TOKEN = "token"
    TOOL_CALL = "tool_call"
    TOOL_RESULT = "tool_result"
    INTERRUPT = "interrupt"
    DONE = "done"
    ERROR = "error"
    CANCELLED = "cancelled"


# running → waiting_human → resumed | rejected | expired
# cancelled 可从 running / waiting_human 进入

def sse_line(event: Event, data: dict) -> str:
    import json
    return f"event: {event.value}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"

人审状态机：

running
  ├─ 只读工具 → 执行 → tool_result → 继续
  ├─ 不可逆工具 → interrupt（waiting_human）
  │     ├─ 用户批准 → resumed → 执行 → tool_result
  │     ├─ 用户拒绝 → rejected → 用只读结果收尾或 done
  │     └─ 超时     → expired → 默认拒绝不可逆；已产生的只读结果仍可返回
  └─ Abort / 超时生成 → cancelled（半截不得标 success）

LangGraph 对照（不重复 checkpointer 细节）：thread_id = 本会话恢复键；interrupt_before=["call_tool"] = 发出 interrupt 事件。Resume 时带上「批准 / 拒绝」，再 invoke 同一 thread_id。PoC 内存 checkpointer 即可；生产换持久化，否则进程重启人审单全丢。

前端：消费与取消

// 最小 SSE 消费；不要当完整 Chat UI
async function consumeAgentStream(url: string, onEvent: (e: string, data: unknown) => void) {
  const ac = new AbortController();
  const res = await fetch(url, { signal: ac.signal, headers: { Accept: "text/event-stream" } });
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  let buf = "";
  try {
    while (true) {
      const { value, done } = await reader.read();
      if (done) break;
      buf += decoder.decode(value, { stream: true });
      const chunks = buf.split("\n\n");
      buf = chunks.pop() ?? "";
      for (const chunk of chunks) {
        const ev = /event: (\w+)/.exec(chunk)?.[1];
        const raw = /data: (.*)/.exec(chunk)?.[1];
        if (ev && raw) onEvent(ev, JSON.parse(raw));
      }
    }
  } catch (err) {
    if ((err as Error).name === "AbortError") onEvent("cancelled", {});
    else throw err;
  }
  return () => ac.abort(); // 调用方在「停止」按钮上触发
}

取消时：前端 abort；后端停生成、发 cancelled（若连接还在）、不要把已生成半截当成任务成功。可观测里 outcome=cancelled，见可观测与线上评测。

interrupt 到达后：渲染确认卡片（工具名 + 参数摘要，敏感字段打码）。批准走 另一条短请求 POST /threads/:id/resume，不要指望同一条 SSE 永远挂着——用户可能关页再回来。

超时未审

策略	适用
默认拒绝不可逆	退款、发送、删除
保留只读部分结果	已经查到的订单状态仍可展示
过期时间写进 `interrupt` 事件	前端倒计时；后端到期把状态打成 `expired`

不要默默执行。过期后的 resume 必须再走一遍闸门，防止「过期邮件里的批准链接」补一刀。

常见坑

工具先执行再流式通知： 人审没有意义。
取消后仍写库： 以 cancelled 为准回滚或根本别提交。
人审只靠 Prompt： 安全篇已否；本篇必须是状态机。
SSE 断了当失败： 人审等待期间断线是正常的，靠 thread_id 恢复。
半截 token 当 done： 评测会虚高成功率。

实践要点

先定 7 种事件，前后端共用同一枚举。
不可逆：tool_call → interrupt → 批准后才执行。
取消用 AbortController + 后端停生成；半截不是 success。
超时未审默认拒绝；只读结果可保留。
面试挂钩：能画出 running / waiting_human / resume，并说到 LangGraph interrupt 只是实现，不是协议本身。

为什么必须流式​

最小事件类型​

传输选型​

后端：事件与人审状态​

前端：消费与取消​

超时未审​

常见坑​

实践要点​