《AI Agent 工程实战》9.3 断线重连：Last-Event-ID 续传不丢步

import asyncio
import json

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

EVENTS = [
    (1, "run.started", {"run_id": "r_d98c8f3c"}),
    (2, "step", {"index": 0, "kind": "tool_call"}),
    (3, "tool_call", {"id": "c1", "name": "search_corpus", "arguments": {"q": "pgvector 召回率"}}),
    (4, "tool_result", {"id": "c1", "name": "search_corpus", "result": "命中 3 条"}),
    (5, "token", {"text": "pgvector"}),
    (6, "token", {"text": " 在"}),
    (7, "token", {"text": "低维"}),
    (8, "token", {"text": "场景更省"}),
    (9, "step", {"index": 1, "kind": "final"}),
    (10, "final", {"content": "pgvector 在低维场景更省。"}),
    (11, "run.succeeded", {"run_id": "r_d98c8f3c", "steps": 2}),
]
BUFFER_START = 1  # 事件缓冲保留的最早 id

app = FastAPI()


def frame(seq: int, event: str, data: dict) -> str:
    return f"id: {seq}\nevent: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


def stream_after(last_id: int):
    if last_id < BUFFER_START - 1:                      # 缓冲窗口丢了，降级
        yield frame(last_id, "resync", {"from": BUFFER_START})
        return
    for seq, event, data in EVENTS:
        if seq > last_id:                               # 只推断点之后的
            yield frame(seq, event, data)


@app.post("/runs/{rid}/stream")
def stream(rid: str, request: Request) -> StreamingResponse:
    last_id = int(request.headers.get("last-event-id", "0"))
    return StreamingResponse(stream_after(last_id), media_type="text/event-stream")


async def read_frames(client, url, last_id=0, limit=None):
    headers = {"Last-Event-ID": str(last_id)} if last_id else {}
    got = []
    async with client.stream("POST", url, json={"input": "x"}, headers=headers) as resp:
        cur = {}
        async for line in resp.aiter_lines():
            if line == "":
                if "id" in cur:
                    got.append((int(cur["id"]), cur.get("event", ""), cur.get("data", "")))
                cur = {}
                if limit and len(got) >= limit:
                    break                                   # 主动断开，模拟网络中断
            elif line.startswith("id:"):
                cur["id"] = line[3:].strip()
            elif line.startswith("event:"):
                cur["event"] = line[6:].strip()
            elif line.startswith("data:"):
                cur["data"] = line[5:].strip()
    return got


async def main() -> None:
    transport = httpx.ASGITransport(app=app)
    url = "/runs/r_d98c8f3c/stream"
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        first = await read_frames(client, url, 0, limit=5)
        print("连接 1（消费 5 条后主动断开）:")
        for seq, event, data in first:
            print(f"  #{seq} {event} {data}")
        last = first[-1][0]
        print(f"  记录 Last-Event-ID = {last}")

        second = await read_frames(client, url, last)
        print("连接 2（带 Last-Event-ID 重连，续推剩余）:")
        for seq, event, data in second:
            print(f"  #{seq} {event} {data}")

        merged = [seq for seq, _, _ in first + second]
        print("--- 断言 ---")
        print(f"合并 id 序列: {merged}")
        print(f"无重复: {len(merged) == len(set(merged))}")
        print(f"无缺失: {merged == list(range(1, len(EVENTS) + 1))}")


asyncio.run(main())

连接 1（消费 5 条后主动断开）:
  #1 run.started {"run_id": "r_d98c8f3c"}
  #2 step {"index": 0, "kind": "tool_call"}
  #3 tool_call {"id": "c1", "name": "search_corpus", "arguments": {"q": "pgvector 召回率"}}
  #4 tool_result {"id": "c1", "name": "search_corpus", "result": "命中 3 条"}
  #5 token {"text": "pgvector"}
  记录 Last-Event-ID = 5
连接 2（带 Last-Event-ID 重连，续推剩余）:
  #6 token {"text": " 在"}
  #7 token {"text": "低维"}
  #8 token {"text": "场景更省"}
  #9 step {"index": 1, "kind": "final"}
  #10 final {"content": "pgvector 在低维场景更省。"}
  #11 run.succeeded {"run_id": "r_d98c8f3c", "steps": 2}
--- 断言 ---
合并 id 序列: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
无重复: True
无缺失: True

BUFFER_START = 8  # 只保留最后 4 条（8..11），更早的已被淘汰


def stream_after(last_id: int):
    if last_id < BUFFER_START - 1:                       # 断点早于缓冲窗口
        yield frame(last_id, "resync",
                    {"reason": "buffer_evicted", "from": BUFFER_START})
        return
    for seq, event, data in EVENTS:
        if seq > last_id:
            yield frame(seq, event, data)

id: 3
event: resync
data: {"reason": "buffer_evicted", "from": 8}