一、引言Demo 与生产之间隔着一条工程鸿沟几乎每个团队都有过这样的经历周五下午用 200 行 Python 跑通了一个 RAG 问答 Demo老板在群里点了赞周一早上把它塞进测试环境问题就全来了——首字延迟 8 秒前端界面一片空白用户以为卡死了上游 API 偶发 429整个请求直接抛异常到前端到了月底看账单没人说得清这 3000 美元花在了哪个功能、哪个用户、哪个 prompt 上。Demo 关心的是能不能跑通生产关心的是在 99.9% 的情况下能不能稳定跑通且成本可控、故障可定位。这两者之间的差距本质上是三个工程问题响应时延的感知问题——LLM 推理是秒级甚至十秒级的必须靠流式输出把等待变成可见的进展依赖的不可靠性问题——你调用的是一个跨公网、有速率限制、有长尾延迟的第三方服务必须有重试与降级系统的可观测性问题——没有 TTFT、TPOT、token 用量、错误分布这些指标你就是在盲飞。本文不谈 prompt 工程只谈这三件事怎么落地。二、流式输出不只是打字机效果2.1 SSE 的真实形态主流大模型 APIOpenAI、Anthropic、通义、DeepSeek 等的流式接口基本都基于 SSEServer-Sent Events。HTTP 响应头是Content-Type: text/event-streambody 是一串以\n\n分隔的事件块data: {choices:[{delta:{content:你},index:0}]} data: {choices:[{delta:{content:好},index:0}]} data: [DONE]注意几个容易被忽略的细节一个事件可能被 TCP 分片切成多段你收到的 chunk 边界和data:边界没有任何关系最后一行[DONE]是判断流正常结束的唯一可靠信号不要靠连接关闭来判断真正的usage字段只有在stream_options: {include_usage: true}时才会在倒数第二个 chunk 里出现否则你只能自己估 token。2.2 流式链路的四个工程问题TTFTTime To First Token是用户感知的核心指标而不是总耗时。一个总耗时 20 秒但 1 秒就出首字的回答体验远好于总耗时 5 秒但 5 秒后才一次性吐出的回答。背压Backpressure是流式架构里最容易被忽视的一环。如果网关把上游 SSE 全量读进内存再转发流式就退化成了慢速非流式长回答会直接打爆内存。正确做法是逐 chunk 转发让 TCP 窗口天然限速。断流处理上游可能在输出到一半时超时或被中断。此时不能静默截断应该向前端发送一个明确的结束标记比如event: error否则用户看到的是回答说到一半没了而监控里一切正常。取消传播用户点了停止生成或刷新了页面网关必须立刻取消上游请求。否则你会为一段没人看的输出继续付费——这是新手最常见的成本泄漏点。三、异常重试把 LLM 当作不可靠依赖3.1 先分清哪些错误该重试无脑重试是灾难。判断标准是错误是否具有瞬时性类型例子是否重试网络层连接超时、读超时、ConnectionResetError✅ 重试限流429、Retry-After头✅ 重试按头等待服务端500 / 502 / 503 / 504✅ 重试客户端400 参数错、401 鉴权、404 模型不存在❌ 不重试内容策略400 content_filter❌ 不重试走降级话术400重试 3 次只会让你多等 3 倍时间并且可能触发风控。3.2 退避策略指数退避 Full Jitter固定间隔重试会让所有失败的客户端在同一时刻再次冲击上游形成惊群。业界推荐的是Full Jittersleep random(0, min(cap, base * 2 ** attempt))random(0, ...)而不是random(base, ...)能最大程度打散重试流量。3.3 流式场景的重试边界流式重试有个残酷的现实一旦你已经向客户端吐出了第一个 token就不能再重试了——重放会导致用户看到两份重复的开头。所以工程上的策略是在首个 token 到达之前允许完整重试此时客户端还没收到任何数据重试是透明的在首个 token 之后断开只能向上报错并把finish_reason ! stop标记为不完整响应对非流式的重试务必在请求头带上X-Request-Id或Idempotency-Key避免上游重复计费。四、监控没有可观测性就没有 SLOLLM 应用的监控和传统 Web 服务最大的不同在于延迟不是单一数字成本不是单一数字。建议至少采集这几类指标llm_ttft_secondsHistogram分模型、分场景统计首字延迟分布P95 才是你要对用户承诺的llm_tpot_secondsHistogramTime Per Output Token直接决定总耗时 ≈ TTFT TPOT × 输出长度llm_tokens_total{direction, model}输入/输出 token 分开统计输入 token 决定了 RAG 的上下文成本llm_request_errors_total{code, model}按 HTTP 状态码分类区分上游故障和我方参数错llm_cost_usd_total{model, tenant}把 token 折算成钱按租户维度聚合——这是唯一能说服老板的指标llm_stream_aborted_total流被中断的次数用来发现用户看到的回答其实是不完整的。标签设计上有一条铁律不要用 user_id、session_id 这类高基数标签它们会让 Prometheus 直接 OOM。用户维度的排查交给 Trace/日志交给指标就是灾难。五、实战一个能上生产的最小客户端下面这段代码把流式、重试、指标三件事合到一起。它是一份可直接改造的骨架。importasyncio,random,time,json,loggingfromdataclassesimportdataclass,fieldimporthttpx loglogging.getLogger(llm)RETRYABLE_STATUS{429,500,502,503,504}dataclassclassMetrics:ttft:float|NoneNoneoutput_tokens:int0attempts:int0aborted:boolFalsedefobserve(self,model:str,status:str):# 真实项目里替换为 Prometheus / OTel 上报log.info(json.dumps({event:llm_call,model:model,status:status,ttft_s:self.ttft,out_tokens:self.output_tokens,attempts:self.attempts,aborted:self.aborted,},ensure_asciiFalse))defdelay(self,attempt:int,retry_after:float|NoneNone)-float:ifretry_after:returnmin(retry_after,30.0)# Full Jitterbase 0.5scap 8sreturnrandom.uniform(0,min(8.0,0.5*(2**attempt)))asyncdefstream_chat(messages,modelgpt-4o-mini,max_attempts4,on_deltaNone):流式调用带重试。on_delta 每收到一段文本回调一次。mMetrics()payload{model:model,messages:messages,stream:True,stream_options:{include_usage:True},}# read timeout 是两次 chunk 之间的间隔不是总时长timeouthttpx.Timeout(connect5.0,read60.0,write10.0,pool5.0)forattemptinrange(max_attempts):m.attemptsattempt1emittedFalse# 是否已经向调用方吐出过 tokentry:asyncwithhttpx.AsyncClient(timeouttimeout)asclient:asyncwithclient.stream(POST,https://api.openai.com/v1/chat/completions,jsonpayload,headers{Authorization:fBearer{API_KEY},X-Request-Id:request_id()},)asresp:ifresp.status_code!200:body(awaitresp.aread()).decode()[:500]ifresp.status_codeinRETRYABLE_STATUSandattemptmax_attempts-1:raresp.headers.get(Retry-After)awaitasyncio.sleep(m.delay(attempt,float(ra)ifraelseNone))continueraiseLLMError(resp.status_code,body)t0time.perf_counter()asyncforlineinresp.aiter_lines():ifnotline.startswith(data:):continuedataline[5:].strip()ifdata[DONE]:m.observe(model,ok)returnchunkjson.loads(data)ifm.ttftisNoneandchunk[choices][0][delta].get(content):m.ttfttime.perf_counter()-t0fordeltain(chunk[choices][0][delta].get(content)or,):ifdelta:emittedTruem.output_tokens1awaiton_delta(delta)ifon_deltaelseNoneifusage:chunk.get(usage):m.output_tokensusage[completion_tokens]except(httpx.TransportError,asyncio.TimeoutError)ase:ifemitted:# 已经开始输出不能重试只能报错m.abortedTruem.observe(model,aborted)raiseStreamAborted()fromeifattemptmax_attempts-1:awaitasyncio.sleep(m.delay(attempt))continuem.observe(model,network_error)raise几个关键设计点emitted标志是整个重试逻辑的闸门它保证了要么完整重试要么干净失败aiter_lines()而非手动iter_chunks()httpx 已经帮你处理了跨 TCP 分片的问题但要注意它按\n切分所以必须保留line.startswith(data:)的判断跳过空行read60.0只约束两次数据之间的间隔。对于长思维链模型o1、R1 类首字前的静默期可能超过 60 秒这个值要按模型单独配置不要全局写死stream_options.include_usage让最后一个 chunk 带回真实 token 数比你自己数 chunk 准确得多。六更多硬核网安与AI工具包请扫码获取完整源码