面经策

在 LangChain 中如何实现流式输出？ ​

  • 标签：LangChain、流式输出

🏛 业务架构师 ​

LangChain 里实现流式输出（以 0.2/0.3 版 LCEL 为准）：

基础用法：

python
chain = prompt | model | StrOutputParser()

for chunk in chain.stream({"question": "退货政策"}):
    print(chunk, end="", flush=True)  # 逐 token 打印

LCEL 链天然支持 stream——组件实现了 Runnable 协议，检索、提示、模型各环节自动流式衔接。

服务端推送（Web 应用）：用 astream 异步流 + SSE（Server-Sent Events）推给前端：

python
async for chunk in chain.astream(inputs):
    yield f"data: {chunk}\n\n"  # SSE 格式

FastAPI/Spring 侧把 SSE 响应透传，前端 EventSource/fetch 逐块渲染。

工程要点（生产必看）：

  1. 链路各环节的流式兼容：模型输出流式、但若后面接了非流式的 OutputParser（如 JSON 解析），流会被"憋住"——结构化输出场景要么放弃逐字流式（解析完再一次性返回），要么用支持流式拼接的解析器；
  2. 检索在流之前：RAG 链路里检索是阻塞环节（几百 ms）——优化 TTFT（首字延迟）主要靠检索加速，流式只是让"生成时间"变得可感知；
  3. 异常与超时：流中途失败要有错误帧协议（前端显示"生成中断"并给重试）；全链路超时兜底；
  4. 可观测：流式的 token 计量与 trace 照常埋点（Callback 的 on_llm_new_token）。

业务价值：流式输出是 LLM 产品体验的第一杠杆——把"等 3 秒看到全部"变成"0.3 秒开始逐字看"，用户感知延迟降 80%+，工作量一天以内——这是我给每个 C 端 AI 产品的必做项。

🏗 系统架构师 ​

LangChain 流式输出（LCEL 原生设计，stream() 一等支持）：

python
# 简单链：token 级流式
chain = prompt | model | StrOutputParser()
for chunk in chain.stream("什么是 RAG？"):
    print(chunk, end="", flush=True)

# 异步流式（服务端）
async for chunk in chain.astream("..."):
    ...

# FastAPI SSE 对接
from fastapi.responses import StreamingResponse

@app.get("/chat")
def chat(q: str):
    def gen():
        for c in chain.stream(q):
            yield f"data: {json.dumps({'delta': c})}\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

架构师视角的要点（比 API 重要）：

  • 流式的体验价值：TTFT（首 token 时间）从秒级降到几百 ms，用户"看着它打字"的耐心远高于白屏等待——总时长不变，感知延迟数量级改善，是 LLM 应用的第一优先体验优化；
  • 流式透传链路：LCEL 的 stream 从模型层 token 粒度透传到管道出口（中间组件必须是流式友好的——parser 要支持增量解析，如 JSON 的 partial parser）；Agent 场景流式更复杂（工具调用步骤事件 + 文本 delta 混合，LangGraph 用事件流 astream_events 区分类型）；
  • 工程配套：SSE/WebSocket 到前端；超时与心跳（长间隔 keepalive）；背压控制（客户端慢时缓冲策略）；流式中断处理（用户停止 → 服务端取消上游，别烧完剩余 token）；
  • 注意：流式不省 token 成本（总生成量不变），只优化体验——省成本要靠 max_tokens 与缓存。

💻 全栈开发专家 ​

LangChain 流式输出三行核心 + 前端对接全链路：

1. 后端流式（LCEL 自动透传）：

python
chain = prompt | llm | StrOutputParser()

for chunk in chain.stream("介绍一下RAG"):    # 同步流式
    print(chunk, end="", flush=True)

async for chunk in chain.astream("RAG"):    # 异步（FastAPI 场景）
    yield chunk

任何 Runnable 链只要底层模型支持流式，.stream() 全链路逐 token 产出——LCEL 相比旧 Chain 类的核心优势。

2. Web 层 SSE 转发（FastAPI 示例）：

python
from fastapi.responses import StreamingResponse

@app.get("/chat")
async def chat(q: str):
    async def gen():
        async for chunk in chain.astream(q):
            yield f"data: {json.dumps({'delta': chunk})}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

3. 前端消费：

javascript
const es = new EventSource('/chat?q=' + encodeURIComponent(q));
es.onmessage = (e) => {
    if (e.data === '[DONE]') return es.close();
    appendToken(JSON.parse(e.data).delta);   // 打字机渲染
};

工程坑清单（实战）：

  • Nginx 反代要关 proxy_buffering，否则「流式」变「一次性」；
  • Agent 循环场景：工具执行阶段没有 token 可吐——发阶段事件（{"stage":"retrieving"}）保持进度感；
  • 客户端断连要传播取消（AbortController → 服务端 cancel chain），不然白白烧完剩余 token；
  • 首 token（TTFT）才是体感关键：检索并行化、prompt 瘦身、前缀缓存都为它服务。