一、本章导学——为什么你必须使用流式输出？
当你应用时最让人抓狂的不一定是模型本身，而是使用者在等待第一个字时产生的焦虑感和无反馈体验。
如果一次普通 Agent 调用需要
这正是SSE+Streaming 的意义所在——把“等不到第一个 token”变成“看得到正在生成”的实时交互。
本章从 LangChain 的基础 API 开始，到 LangGraph Agent 的四种流式模式。再到 FastAPI SSE 的全链路实现与异步处理，一路帮你把握痛点并解决掉它们。
二、流式输出基础——如何让模型边生成边返回？
1️⃣ 为什么需要流式输出？
-
传统 .invoke: 模型先生成完整响应 → 工具调用 → 最终拼接；整个过程阻塞，只要耗时超过 10‑15 s 使用者就会觉得卡死。后面再说超时处理技巧！💡
)
-
使用 Streaming 可以把 “一次性返回” 改成 “逐步推送”。
-
TFTT 从秒降到毫秒级；
-
User 在看到文字出现时立即知道程序在工作，焦虑大幅下降。
-
E.g.,GPT‑4 在 streaming 下首个 token 通常只需 ~200 ms。
从**结论**来看，当业务上有 *任何* 等待时间导致 UX 降级时就应该考虑 Streaming！
css
/* 示例 CSS 用于渲染表格 */
table {
width这方面，100%;border-collapse:collapse;}
th { background:#dfe7ff;padding:.5rem;按理说，}
tbody tr:nth-child{ background:#fafafa;}
tbody tr:hover { background:#ffe6cc;}
`
` `
` `
` `
` `
` `
` `
` `
` `
`
` `
` `
`
### 完整代码示例 – 同步 Streaming `)
python
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
load_dotenv
model = init_chat_model(
model=os.getenv,api_key=os.getenv,base_url=os.getenv,model_provider="openai"。)
# 同步 streaming：逐 token 输出到终端
for chunk in model.stream:
print
print
### 完整代码示例 – 异步 Streaming `)
python
import asyncio
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
load_dotenv
model = init_chat_model(
model=os.getenv,api_key=os.getenv,base_url=os.getenv,model_provider ="openai",)
async def stream_response:
async for chunk in model.astream:
print
print
asyncio.run)
2️⃣ 四种 Stream Mode 详解 – 为不同需求挑选合适粒度！🔧️️️️️️️️️️🛠️⚙️🧩📈🛠️🎯🚀🪐🌍🏆⏳💬❗️✅✖️⚠️⚡️🤖🤔💡🔎📊📈📉📚🔢📑📦🔧🔬⏱♀♂🚀👨💻👩💻🚨🤝🙌🌟🚧🥇🎉⚔💥⌛︎⌚︎😃😞😐😂😭🙁😊😉🙇🙅😶😶🤷♀ 🤷♂ 🕵 👨🏽🏫👩🏽🏫📝💭🤓👀🎯🍃✨🔥☀🌈🌊🌪🐾🌌🛰🛸🔭🔬⚙🔒✈✉☎📞☎☎🗺☑✔〰↕⇄↔↕❄⛄☃🎃❄♿🍂🍃🍂❣❤️❤️❤❤😍😘🥰😍😘😘🥰🥺🙏👏✌🎁👍🏼👋🏻🙂😉😜🙋🏻👍🏼🙏🏼👍🏼😊☺︎♥︎💗🐱🐶🐵🐴🐮🐹🐭🐹😀😂🤣 😆😅🤪 🙈🙉🙊 🐶 🐱 💓✨🔥☀︎ 🌝 ☎ ☑ ✅ ⏰ ⌚ 📅 📆 📰 🎤 🎬 📹 🎼 🎧 🔊 🔈 🔔 🔕 ❗‼‽ ⁉⁂ •• ➤ ➦ ➤ ➜ ⬅➡⬇⬆ ↩↪➜ ➡➖ ⏎ ⚙ ⚙ ⚙' #just demo...
### 怎么选？| 模式 | 输出内容 | 数据粒度 | 常见用途 |
|------|--------|----------|----------|
| **values** | 完整状态快照 | 粗 | 调试、回放 |
| **messages** | 每个 Token | 精细 | 前端实时展示 |
| **updates** | 节点增量变化 | 中等 | 节点监控、日志 |
| **custom** | 开发者自定义事件 | 任意 | 长任务进度、额外数据 |
---
### 示例：创建带工具的 Agent 并开启不同 Stream Mode
python
# 创建带工具 Agent 示例
agent = create_agent
#### values 模式 – 调试视角
python
for state in agent.stream]}。stream_mode='values'):
messages = state
latest = messages
print
#### messages 模式 – 打字机效果
python
for msg,metadata in agent.stream]},stream_mode='messages'):
if msg.content:
print
#### updates 模式 – 节点级实时反馈
python
for update in agent.stream]},stream_mode='updates'):
for step,data in update.items:
if 'content_blocks' in data:
# 简化展示，仅打印 token 信息…pass
#### custom + messages 双重模式 – 同时获取进度与 LLM 文本
python
@tool
def search_with_progress->str:
writer=get_stream_writer
writer
time.sleep
writer
time.sleep
writer
return f"关于「{query}」..."
agent_v2=create_agent
for mode,chunk in agent_v2.stream]],stream_mode=):
if mode=='custom':
print}")
说到else，msg,_=chunk if isinstance else chunk
if msg.content:print
---
### FastAPI SSE 实战 – 全链路实现 🚀
#### 服务端代码
python
# demo_sse.py - FastAPI SSE 服务示例
import json。asyncio,json from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain.chat_models import init_chat_model from langchain.tools import tool from langchain.agents import create_agent from dotenv import load_dotenv import os load_dotenv
app=FastAPI
model=init_chat_model(
model=os.getenv,api_key=os.getenv,base_url=os.getenv,model_provider='openai',)
@tool def search_knowledge->str:return f"关于「{query}」…"
@tool def calculate->str:
try的观点是，return str)
except Exception as e:return f"错误:{e}"
agent=create_agent
async def generate_stream:
yield json.dumps+'
'
async for event in agent.astream_events]},version='v2'):
k,event_name,d=data:=event,event,event
if k=='on_chat_model_stream':
token=d.content or ''
if token:yield json.dumps+'
'
elif k=='on_tool_start':
yield json.dumps}。ensure_ascii=False)+'
'
elif k=='on_tool_end':
yield json.dumps+'
'
yield "data:
"
@app.get
async def chat:
return StreamingResponse,media_type='text/event-stream',headers={"Cache-Control":"no-cache","Connection":"keep-alive","X-Accel-Buffering":"no"})
#### 前端消费示例
SSE Demo ChatGPT-like UI 🚀🚀🚀!说起来，*{box-sizing:border-box;margin:0，padding:0;}body{font-family:sans-serif;background:#fafafa;padding-top:20px;}#container{max-width:800px;margin:auto，background:white;border-radius:.25rem;padding:.75rem;}#output{height:400px;border:.0625rem solid #ddd;其实，border-radius:.125rem;padding:.75rem;background:white;font-family:'Consolas','Monaco';
overflow-y:auto;font-size:.875rem;color:black，}#inputArea{display:flex;margin-top:.75rem;}#inputArea input{flex-grow:1;padding:.625rem;border:.0625rem solid #ddd;border-radius:.125rem;font-size:.875rem;按理说，}#sendBtn{padding:.625rem .9375rem;margin-left:.3125rem;background:#1976d2;color:white，border:none;border-radius:.125rem;font-size:.875rem;按理说，cursor:pointer;}
#sendBtn{background:#bbb;}
.cursor{display:inline-block;width:8px，height:auto;老实说，background:black;margin-left:-8px;}
@keyframes blink{50%{opacity:.25;}}
.msg.user{text-align:right;color:dimgrey;}
.msg.ai{text-align:left;color:black，其实，}
.msg.tool{text-align:left;color:dimgrey;font-style:italic;}
textarea,#output {font-family:"Consolas"。monospace}
textarea::-webkit-scrollbar {width:6px}
textarea::-webkit-scrollbar-track {background:white}
textarea::-webkit-scrollbar-thumb {background:dimgrey}
textarea::-webkit-scrollbar-thumb:hover {background:black}
}
Demo ©2026 AI Lab.
。
