让模型给文章写摘要：一个会重试的 LLM 客户端，和一条 SSE 全链路
我把手写摘要交给了模型，但我不信任它。这篇讲两块最难的东西：客户端怎么处理超时、限流和退避抖动，以及 SSE 流式怎么从模型接口一路走到浏览器，中间三个环节各自的坑。
站里的文章摘要一直是我手写的，写到第三十篇的时候烦了：每次发完文章还要对着正文复述一遍。我决定交给模型写，但我不信任它。
下面这个功能是后台编辑页的一个按钮：点一下，模型开始写摘要，文字一个字一个字出现在输入框里。这篇文章只讲这个过程里最难的两块，客户端怎么容错，以及流式怎么从模型一路走到浏览器。代码是站里跑着的版本，删掉业务细节，照着抄能跑。
先定一条原则
动手前先定死一条：模型服务挂了，文章页和后台都不能报 500，也不能弹错误提示，最差是静默降级成"没有这个功能"。
这条原则决定了很多设计。摘要我一开始想做成实时生成，打开文章页时没缓存就现调模型，后来否了，因为它把模型接口放进了用户请求路径，接口偶发的三秒延迟会直接变成页面白屏三秒。改成离线预计算之后，用户路径上只剩一次文件读取。
但后台那个"帮我写摘要"的按钮必须是实时的，因为我在等结果，界面八秒纹丝不动我会以为按钮没生效。这就是流式的意义，不是性能，是反馈。
客户端：超时、重试和抖动
我见过不少项目直接在业务函数里 requests.post 一下，取 response.json()["choices"][0]["message"]["content"] 就用。这种写法在 demo 里没问题，上线后会出三类故障：网络超时把请求挂住、限流返回 429、模型偶尔返回不合法 JSON 导致解析异常。
所以我先写一个薄客户端把前两类处理掉。
# backend/app/ai/client.py
import asyncio
import random
import httpx
class LLMError(RuntimeError):
pass
class LLMClient:
def __init__(self, api_key: str, base_url: str, model: str, max_retries: int = 3) -> None:
self._api_key = api_key
self._base_url = base_url.rstrip("/")
self._model = model
self._max_retries = max_retries
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=5.0),
limits=httpx.Limits(max_connections=8, max_keepalive_connections=4),
)
async def chat(self, messages: list[dict], temperature: float = 0.3) -> str:
payload = {"model": self._model, "messages": messages, "temperature": temperature}
last_error = None
for attempt in range(self._max_retries):
try:
response = await self._client.post(
f"{self._base_url}/chat/completions",
headers={"Authorization": f"Bearer {self._api_key}"},
json=payload,
)
if response.status_code == 429 or response.status_code >= 500:
raise LLMError(f"upstream {response.status_code}")
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
except (httpx.TimeoutException, httpx.TransportError, LLMError) as exc:
last_error = exc
delay = min(8.0, 0.6 * (2**attempt)) * (0.7 + random.random() * 0.6)
await asyncio.sleep(delay)
except (KeyError, IndexError) as exc:
raise LLMError(f"malformed response: {exc}") from exc
raise LLMError(f"failed after {self._max_retries} attempts: {last_error}")三处决定要解释。
超时、连接错误、429、5xx 这四类归为"可以重试"，因为都是暂时性的。响应结构不对（拿不到 choices[0].message.content）归为"不该重试"，重试一次大概率还是同样结构，混在一起只会让日志里堆满没意义的重复请求。
退避时长里加了抖动，0.7 + random.random() * 0.6 得到 0.7 到 1.3 之间的随机乘数。这解决的是惊群：批量处理 48 篇文章时如果同时撞上限流，没有抖动的话 48 个请求会在同一毫秒一起重试，把限流撞得更死。
httpx.Limits(max_connections=8) 是限流自保，默认值是 100，意味着批量脚本可能瞬间打出上百个并发，那一定会被上游按在地上。8 个是我测出来的甜蜜点，再高就开始出 429。
流式：从上游一路到浏览器
流式这一条链路要过三个环节，每一环都有它自己的坑。
先让客户端能读上游的分片。
# backend/app/ai/client.py（在 LLMClient 内追加）
async def chat_stream(self, messages: list[dict], temperature: float = 0.3):
payload = {"model": self._model, "messages": messages,
"temperature": temperature, "stream": True}
async with self._client.stream(
"POST",
f"{self._base_url}/chat/completions",
headers={"Authorization": f"Bearer {self._api_key}"},
json=payload,
timeout=httpx.Timeout(60.0, connect=5.0),
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
chunk = line[6:].strip()
if chunk == "[DONE]":
return
try:
content = json.loads(chunk)["choices"][0]["delta"].get("content")
except (json.JSONDecodeError, KeyError, IndexError):
continue
if content:
yield content用 aiter_lines 按行迭代，别用 iter_bytes 自己切分。一个网络分片很可能切在一行中间，你可能拿到 da 和 ta: {...} 两段，自己拼的解析器要处理这种半行，很容易出 bug，而 aiter_lines 已经做了这件事。
delta 里的 content 可能是 None。流式响应第一个 chunk 通常只带 role，最后几个可能只带结束标记，直接 yield 会往 SSE 里塞一堆 None 字符串。
第二环是后端把自己包成 SSE。格式很简单，event: 一行，data: 一行，空行结束一条消息。
# backend/app/api/routes/ai.py
def sse_event(event: str, data: dict) -> str:
return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
@router.post("/summarize/stream")
async def summarize_stream(payload: SummarizePayload, client=Depends(get_llm_client),
_admin=Depends(require_admin)):
async def generator():
buffer: list[str] = []
yield sse_event("start", {"ok": True})
try:
async for chunk in client.chat_stream(
[{"role": "user", "content": payload.content[:12000]}]
):
buffer.append(chunk)
yield sse_event("delta", {"text": chunk})
except Exception as exc:
yield sse_event("error", {"message": str(exc)})
return
yield sse_event("done", {"summary": "".join(buffer).strip()})
return StreamingResponse(
generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)我定义了三类事件：start 让前端知道连接建立了，delta 是增量文本，done 带完整结果。X-Accel-Buffering: no 这个响应头是专门给 Nginx 看的，它默认会把小块数据攒成大块再发，这对普通接口是优化，对流式是灾难，会让"一个字一个字出来"变成"卡八秒然后全部出来"。这个头只解决上游到 Nginx 那一段，Nginx 自己到浏览器那段还得配 proxy_buffering off，两处都做才稳。
前端读流：缓冲区加边界查找
浏览器的 EventSource 只能发 GET，而我的接口按项目约定是 POST，还要带请求体和 Cookie。所以用 fetch 加 ReadableStream 自己解析。
// frontend/src/services/aiStream.js
export async function streamSummary(content, { onDelta, onDone, onError }) {
const controller = new AbortController();
const response = await fetch("/tiana/api/admin/ai/summarize/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
credentials: "include",
body: JSON.stringify({ content }),
signal: controller.signal,
});
if (!response.ok || !response.body) throw new Error(`HTTP ${response.status}`);
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let boundary = buffer.indexOf("\n\n");
while (boundary !== -1) {
const block = buffer.slice(0, boundary);
buffer = buffer.slice(boundary + 2);
boundary = buffer.indexOf("\n\n");
let event = "message";
let payload = "";
for (const line of block.split("\n")) {
if (line.startsWith("event: ")) event = line.slice(7).trim();
else if (line.startsWith("data: ")) payload += line.slice(6);
}
if (!payload) continue;
const data = JSON.parse(payload);
if (event === "delta") onDelta(data.text);
else if (event === "done") onDone(data.summary);
else if (event === "error") onError(data.message);
}
}
return () => controller.abort();
}decoder.decode(value, { stream: true }) 里的 stream: true 是关键。一个汉字在 UTF-8 里占三个字节，网络分片完全可能把一个汉字切成两半，一半在这个 chunk 里一半在下一个。不加这个参数，TextDecoder 会把不完整的字节序列解成替换字符，界面上出现一堆问号。加了之后它会把不完整的字节暂存，等下一块到了再拼起来。
外层反复往 buffer 追加、内层用 indexOf("\n\n") 找完整消息、找完就把用掉的部分切掉，这个"缓冲区加边界查找"的模式在处理任何流式文本协议时都适用。原理和 Python 那边的按行迭代一样：一条 SSE 消息可能跨两个网络分片，没有缓冲就会解析出半条。
AbortController 这次真的用上了。我在后台点了生成，写两秒觉得不对，转身去点别的文章，上一次的流应该被掐掉，否则它还会继续往已经不显示的界面上写文本，返回的 abort 函数给组件在 onBeforeUnmount 里调用。
效果和边界
摘要改成模型生成后，我拿三十篇自动摘要和三十篇手写摘要混着看，自己都分不太出来。这不是夸奖。摘要只是把正文里的信息压缩一遍，正文空洞，摘要就只能靠句式撑着，所以我现在把模型输出当初稿，读一遍把"全面介绍""深入实践"这类词换成具体说法，比如把"介绍了 Nginx 缓存的完整方案"改成"把 CSS 和 JS 钉死缓存一年，HTML 每次协商"。这一步两分钟，能让摘要从"看起来对"变成"确实说了点什么"。
成本上，48 篇首次全量生成加后续零星更新一共 61 次调用，按当量算是几毛钱。之所以这么省，是因为摘要按正文哈希做了缓存，改一次错别字只重新生成一次。
边界要说清：这套实现只解决了"调用可靠"和"流式链路"两件事，没解决结构化输出。我需要模型返回带标签的 JSON 时，它还经常裹一层 json 代码围栏，或者在前面加一句"好的，以下是结果"，这部分是另一个话题，展开写又是一篇。