---

前言

主线：Agent Runtime → Harness。前几篇我们把 Agent「跑通了」——状态、工具循环、持久化都立住了。但直到现在，回答都是「憋到最后一口气才吐出来」的：一次 invoke 阻塞到底，等模型完整答完才把整段话一次性还给你。本篇拆掉这个「黑盒」，让回答边想边说——token 一级地往外流，再一路流到浏览器前端。流式输出不是锦上添花，它是 Agent 产品「能用」和「好用」的分水岭。

一、为什么需要流式输出：invoke 与 stream 的差别

前面几篇我们一直用 agent.invoke(...) 拿结果。invoke 是「同步一把梭」：把整轮对话跑完，返回最终状态。模型答出第一个字的那一刻，前端什么都看不到，用户只能盯着转圈。对一个 20 秒的长回答，这 20 秒是纯黑的。

stream 反着来：不返回最终结果，而是把过程中的每一小步「推」给你。模型每吐出一个 token（甚至一个词），你就能立刻拿到，前端马上渲染一个字。用户感知上的「快」不是真快了多少，而是首字延迟（TTFT，Time To First Token）从整段时长缩到几百毫秒——这正是 GPT-4、Claude 这类产品「边说边出字」的体感来源。

这不是把 invoke 换一个函数名那么简单。invoke 和 stream 是两套心智：

	invoke	stream
返回	最终状态 dict	一个迭代器/生成器，逐个吐中间产物
首字延迟	整轮跑完	一个 token 的时长
用途	批处理、脚本、拿最终结果	聊天界面、进度展示、长任务
粒度	无（只有终态）	可粗可细，由 stream_mode 决定

stream_mode 就是「流式输出的颗粒度开关」。同一个 stream 调用，传不同的 stream_mode，吐出来的东西完全不一样——这是本篇的核心概念。

二、stream_mode 五种取值：一张表看懂

LangGraph 的 stream_mode 有五种核心取值，各自对应一个「颗粒度」：

stream_mode	吐什么	粒度	典型用途
values	每个 super-step 结束后的完整状态	最粗	要完整 state、做状态回放
updates	每个节点产出的增量（dict，key 是节点名）	中	观察「哪一步做了什么」
messages	(消息块, 元数据) 元组	最细（token 级）	打字机式输出
custom	你自定义的事件（需 get_stream_writer()）	自定义	进度条、业务事件
debug	执行轨迹（step/type/payload）	调试	排查「图跑到哪卡住了」

两个后补模式：官方还提供 checkpoints（吐 checkpoint 事件，需挂 checkpointer）和 tasks（吐任务级起止事件，含结果/错误）两个模式。debug 本质上就是 checkpoints + tasks 的合体再加点元数据。日常开发先吃透上表五种，那两种用到再查。

运行前先配 key：本文所有代码都要读环境变量 DEEPSEEK_API_KEY（ChatDeepSeek 自动读取）——Windows 用 set DEEPSEEK_API_KEY=...、macOS/Linux 用 export DEEPSEEK_API_KEY=...，同第 2 篇约定。另外注意下面工具把「演示用假数据」写成了注释而非 docstring：docstring 会被当成工具描述喂给模型，模型可能把它当真话复述进回答，所以这类标注一律放注释。

用一个「查汇率」的 Agent（换掉前几篇的查天气，避免案例同质化）逐个看。完整真源见 code/06-流式输出/stream_modes.py：

import sys

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

# Windows 控制台默认 GBK，模型回复若带 emoji，print 会抛 UnicodeEncodeError；强制 UTF-8 输出
sys.stdout.reconfigure(encoding="utf-8")


@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """查询两种货币的实时汇率。"""  # 演示用假数据，不接真实汇率 API
    rates = {("USD", "CNY"): "7.25", ("EUR", "CNY"): "7.85", ("CNY", "USD"): "0.138"}
    return rates.get((from_currency, to_currency), f"{from_currency}→{to_currency}：1.0（默认）")


model = ChatDeepSeek(model="deepseek-chat")
agent = create_agent(
    model,
    tools=[get_exchange_rate],
    system_prompt="你是简洁的助手，回答尽量短。",
)

q = {"messages": [{"role": "user", "content": "100 美元能换多少人民币？"}]}

# 1) values：每跑完一个 super-step，吐一次完整状态（messages 列表越滚越长）
print("=" * 30, "values", "=" * 30)
for chunk in agent.stream(q, stream_mode="values"):
    print(f"  messages 条数 = {len(chunk['messages'])}")

# 2) updates：每个节点产出的增量，dict 以节点名为 key
print("=" * 30, "updates", "=" * 30)
for chunk in agent.stream(q, stream_mode="updates"):
    for node, val in chunk.items():
        print(f"  节点 {node!r} 产出 {type(val).__name__}")

# 3) messages：最细粒度，吐 (消息块, 元数据) 元组
print("=" * 30, "messages", "=" * 30)
for chunk, meta in agent.stream(q, stream_mode="messages"):
    node = meta["langgraph_node"]
    txt = chunk.content if hasattr(chunk, "content") else ""
    print(f"  节点 {node!r} 类型 {type(chunk).__name__} 内容 {str(txt)[:20]!r}")

# 4) custom：create_agent 没有自定义节点，吐不出任何东西（自定义事件见 custom_events.py）
print("=" * 30, "custom", "=" * 30)
items = list(agent.stream(q, stream_mode="custom"))
print(f"  共收到 {len(items)} 个自定义事件（create_agent 下为空）")

# 5) debug：调试用的执行轨迹，含 step / type / payload
print("=" * 30, "debug", "=" * 30)
for ev in agent.stream(q, stream_mode="debug"):
    print(f"  step={ev.get('step')} type={ev.get('type')}")

跑一遍（模型输出每次略有不同，示意即可）：

============================== values ==============================
  messages 条数 = 1
  messages 条数 = 2
  messages 条数 = 3
  messages 条数 = 4
============================== updates ==============================
  节点 'model' 产出 dict
  节点 'tools' 产出 dict
  节点 'model' 产出 dict
============================== messages ==============================
  节点 'model' 类型 AIMessageChunk 内容 ''
  节点 'model' 类型 AIMessageChunk 内容 ''
  节点 'tools' 类型 ToolMessage 内容 '7.25'
  节点 'model' 类型 AIMessageChunk 内容 '按'
  节点 'model' 类型 AIMessageChunk 内容 '汇率'
  ...
============================== custom ==============================
  共收到 0 个自定义事件（create_agent 下为空）
============================== debug ==============================
  step=1 type=task
  step=1 type=task_result
  step=2 type=task
  step=2 type=task_result
  step=3 type=task
  step=3 type=task_result

几个关键观察：

• values 的 messages 条数从 1 涨到 4——每个 super-step 都吐一份完整状态，你看到的是「这张图的记忆在一步步长大」。
• updates 的顺序是 model → tools → model，正好是第 4 篇讲的工具调用循环：模型喊工具、工具干活、模型收尾。每个 chunk 只含这一步新增的那部分。
• messages 里出现了「内容为空」的 AIMessageChunk（后面第三节展开），还有一条完整的 ToolMessage。
• custom 是空的——因为 create_agent 的节点没写过自定义事件，这个模式要配合 get_stream_writer() 才有东西（第五节）。
• debug 把整轮拆成 3 个 step，每个 step 一对 task / task_result，对应「一步开始、一步结束」。（本示例没挂 checkpointer，所以只看到 task 半侧；挂了 checkpointer 才会多出 checkpoint 事件——这正是前面说的「debug = checkpoints + tasks 合体」。）

三、messages 模式：打字机式输出的首选

做「边说边出字」，stream_mode="messages" 是首选。它吐的是 (chunk, meta) 二元组：

• chunk：一个消息块。模型节点吐的是 AIMessageChunk（content 是一小段文字，多字一块，比如 DeepSeek 常吐「按」「汇率」这种 1~2 字块）；工具节点吐的是完整的 ToolMessage。
• meta：这个块的出处元数据。最常用的是 meta["langgraph_node"]——它告诉你这块是 "model" 产的文字，还是 "tools" 产的工具结果。

打字机输出的核心过滤就一行——只要 model 节点的、且 content 非空的块：

for chunk, meta in agent.stream(q, stream_mode="messages"):
    if meta["langgraph_node"] == "model" and chunk.content:
        print(chunk.content, end="", flush=True)

为什么必须判 chunk.content 而不是「看到 model 节点就 print」？看上一节的输出：模型节点吐了一串 content 为空的 AIMessageChunk。这些不是 bug——它们是模型在**「决定调工具」阶段**吐的增量，真正的信息在 tool_call_chunks 字段里（工具名、参数一段段拼出来），content 是空的。如果不判 chunk.content，前端就会刷出一堆空白。完整真源见 code/06-流式输出/typewriter.py：

import sys

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

sys.stdout.reconfigure(encoding="utf-8")


@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """查询两种货币的实时汇率。"""  # 演示用假数据，不接真实汇率 API
    rates = {("USD", "CNY"): "7.25", ("EUR", "CNY"): "7.85", ("CNY", "USD"): "0.138"}
    return rates.get((from_currency, to_currency), f"{from_currency}→{to_currency}：1.0（默认）")


model = ChatDeepSeek(model="deepseek-chat")
agent = create_agent(
    model,
    tools=[get_exchange_rate],
    system_prompt="你是简洁的助手，回答尽量短。",
)

q = {"messages": [{"role": "user", "content": "100 美元能换多少人民币？"}]}

# 逐 token 打印：model 节点吐的是 AIMessageChunk，其 content 就是一段文字
# 工具节点吐的是完整的 ToolMessage，这里直接跳过，避免把工具内部结果打到屏幕上
for chunk, meta in agent.stream(q, stream_mode="messages"):
    if meta["langgraph_node"] == "model" and chunk.content:
        print(chunk.content, end="", flush=True)
print()

跑一遍：

按汇率 1 美元 = 7.25 人民币，100 美元 ≈ **725 元人民币**。

顺带看清 meta 里还有哪些信息——这对写「协议层」很有用（langgraph_step 是第几个 super-step，langgraph_triggers / langgraph_path 是它怎么被路由到这里的，langgraph_checkpoint_ns 是第 5 篇埋过线的 checkpoint 命名空间，ls_* 开头的是 LangSmith 追踪用的）：

'model' 节点的 meta 里有：
  langgraph_step=3, langgraph_node='model',
  langgraph_triggers=('branch:to:model',),
  langgraph_path=('__pregel_pull', 'model'),
  langgraph_checkpoint_ns='model:...',
  ls_integration='langchain_chat_model', ls_provider='deepseek'

（langgraph_step 的编号随回答长短浮动：上面这个 3 是因为「决定调工具」占掉了 step 1/2，收尾回答在第 3 步。）

四、astream_events：更细粒度的事件流

stream_mode 是按「图的结构」分颗粒度的；langchain-core 的 Runnable 层还有一条按「运行事件」分的流：astream_events(version="v2")——LangGraph 图和 create_agent 都实现了它。它吐的是一串 on_xxx 命名的事件，比 stream_mode 更「内行」——你能看到每一次模型调用、每一次工具调用的起止。

注意两点：它是异步的（要 async for），事件名用 on_ 前缀。完整真源见 code/06-流式输出/astream_events.py：

import asyncio
import sys

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

sys.stdout.reconfigure(encoding="utf-8")


@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """查询两种货币的实时汇率。"""  # 演示用假数据，不接真实汇率 API
    rates = {("USD", "CNY"): "7.25", ("EUR", "CNY"): "7.85", ("CNY", "USD"): "0.138"}
    return rates.get((from_currency, to_currency), f"{from_currency}→{to_currency}：1.0（默认）")


async def main():
    model = ChatDeepSeek(model="deepseek-chat")
    agent = create_agent(
        model,
        tools=[get_exchange_rate],
        system_prompt="你是简洁的助手，回答尽量短。",
    )
    q = {"messages": [{"role": "user", "content": "100 美元能换多少人民币？"}]}

    # astream_events 是异步生成器，必须 async for；事件名是 on_xxx 形式
    async for ev in agent.astream_events(q, version="v2"):
        name = ev.get("event")
        if name == "on_chat_model_stream":
            chunk = ev["data"]["chunk"]
            txt = chunk.content if hasattr(chunk, "content") else ""
            if txt:
                print(txt, end="", flush=True)
        elif name == "on_tool_start":
            print(f"\n[调用工具] {ev.get('name')} 输入={ev['data'].get('input')}")
        elif name == "on_tool_end":
            print(f"[工具返回] {ev.get('name')} 输出={ev['data'].get('output')!r}")
    print()


asyncio.run(main())

跑一遍：

[调用工具] get_exchange_rate 输入={'from_currency': 'USD', 'to_currency': 'CNY'}
[工具返回] get_exchange_rate 输出=ToolMessage(content='7.25', name='get_exchange_rate', ...)
按汇率 1 USD = 7.25 CNY，100 美元可换 **725 元人民币**。

一轮跑下来，事件类型大致是这些（计数随回答长短浮动）：

事件	含义
on_chain_start / on_chain_end / on_chain_stream	agent 这条「链」的起止与流
on_chat_model_start / on_chat_model_end / on_chat_model_stream	每次模型调用的起止，on_chat_model_stream 是 token 流
on_tool_start / on_tool_end	每次工具调用的起止

一个值得记的坑：v2 里 on_tool_stream 和 on_retriever_stream 被移除了，看工具结果用 on_tool_end（它的 output 是个 ToolMessage，真正的文字在 .content）。老教程里按 on_tool_stream 抓工具输出的代码，在 v2 下会静默收不到东西。

什么时候用哪个？ 简单打字机用 stream_mode="messages" 就够，干净省事；需要「模型调了几次、每次耗时多少、工具传了什么参数」这类审计级信息时，上 astream_events。两者不是二选一，你可以在 astream_events 里同时拿到 token 流和工具调用全貌。

五、StreamWriter 自定义事件

stream_mode 和 astream_events 管的是「模型/工具」这些内置事件；可你的 Agent 里还有些业务事件——「正在查第 3 个数据源」「进度 60%」「这一步要做 5 秒」——这些 LangGraph 不知道，得你自己发。这就要 get_stream_writer()（导入自 langgraph.config）：它只能在节点内部调用，返回一个 writer，把任意 dict 打出去。

注意：stream_mode="custom" 本身不产生任何东西，它只是一个通道——你 writer(...) 打进通道的东西，才会从这里流出来。上一节 create_agent 的自定义事件为空，正是因为它的节点从没调用过 writer。手写一个图看效果，完整真源见 code/06-流式输出/custom_events.py：

import sys
from typing import Annotated, TypedDict

from langchain_core.messages import HumanMessage
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

sys.stdout.reconfigure(encoding="utf-8")


class State(TypedDict):
    messages: Annotated[list, add_messages]


def answer_node(state: State) -> dict:
    # get_stream_writer() 只能在节点内部调用；它返回一个 writer，把任意 dict 打出去
    writer = get_stream_writer()
    writer({"type": "progress", "msg": "第一步：读取输入"})
    writer({"type": "progress", "msg": "第二步：生成回答"})
    writer({"type": "done", "answer": "你好，我是助手。"})
    return {"messages": [{"role": "assistant", "content": "你好，我是助手。"}]}


graph = StateGraph(State)
graph.add_node("answer_node", answer_node)
graph.add_edge(START, "answer_node")
graph.add_edge("answer_node", END)
app = graph.compile()

# stream_mode="custom" 拿到的正是 writer 打出去的 dict
print("=== stream_mode='custom' 拿到的自定义事件 ===")
for chunk in app.stream({"messages": [HumanMessage(content="你好")]}, stream_mode="custom"):
    print("  ", chunk)

跑一遍：

=== stream_mode='custom' 拿到的自定义事件 ===
   {'type': 'progress', 'msg': '第一步：读取输入'}
   {'type': 'progress', 'msg': '第二步：生成回答'}
   {'type': 'done', 'answer': '你好，我是助手。'}

自定义事件的价值在协议上：前端拿到 {"type": "progress", ...} 就渲染进度条，拿到 {"type": "done", ...} 就结束 loading。下一节把它和 messages 一起接进 FastAPI，就是一个完整的前后端流式协议。

六、实战案例：FastAPI 流式接口（SSE）

把上面几条线索收拢，做一个真实可跑的流式后端。选 SSE（Server-Sent Events） 而不是 WebSocket——Agent 的流式输出是单向的（服务端→前端），SSE 足够、更简单、还能复用 HTTP 的一切（负载均衡、鉴权、代理）。

SSE 的线格式极简，每个事件两行加一个空行：

event: token
data: 按

event: token
data: 汇率

event: done
data:

event: error
data: 模型调用超时

后端把 stream_mode="messages" 吐出的每个 token 包装成 {"event": "token", "data": ...} 让出即可——注意 async 上下文里必须用 agent.astream（异步流），同步 stream 会阻塞事件循环（第七节③展开）。sse-starlette 的 EventSourceResponse 就干这件事——给它一个异步生成器，它负责把 dict 编码成上面的线格式。完整真源见 code/06-流式输出/server.py：

import sys

from fastapi import FastAPI, Request
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from sse_starlette.sse import EventSourceResponse

sys.stdout.reconfigure(encoding="utf-8")


@tool
def get_exchange_rate(from_currency: str, to_currency: str) -> str:
    """查询两种货币的实时汇率。"""  # 演示用假数据，不接真实汇率 API
    rates = {("USD", "CNY"): "7.25", ("EUR", "CNY"): "7.85", ("CNY", "USD"): "0.138"}
    return rates.get((from_currency, to_currency), f"{from_currency}→{to_currency}：1.0（默认）")


model = ChatDeepSeek(model="deepseek-chat")
agent = create_agent(
    model,
    tools=[get_exchange_rate],
    system_prompt="你是简洁的助手，回答尽量短。",
)

app = FastAPI()


@app.get("/chat")
async def chat(request: Request, q: str):
    """把一轮对话以 SSE 流式推给前端：每个 token 一个 `token` 事件，结束 `done`、出错 `error`。"""

    async def event_stream():
        try:
            input_ = {"messages": [HumanMessage(content=q)]}
            # async 上下文里必须用 astream（同步 stream 会阻塞事件循环，见第七节③）
            async for chunk, meta in agent.astream(input_, stream_mode="messages"):
                # 前端断开就立刻停，别为没人看的 token 继续计费
                if await request.is_disconnected():
                    break
                # 只推模型吐出的文字 token；工具内部的 ToolMessage 不推给前端
                if meta["langgraph_node"] == "model" and chunk.content:
                    yield {"event": "token", "data": chunk.content}
            yield {"event": "done", "data": ""}
        except Exception as e:  # 流式中途出错，兜底发一个 error 事件，别让连接裸挂
            yield {"event": "error", "data": str(e)}

    # ping=15：每 15 秒发一条心跳注释帧，防止长连接被网关/代理掐断
    return EventSourceResponse(event_stream(), ping=15)


# 启动方式（在终端执行，别用 python 直接 import 本文件）：
#   uvicorn server:app --reload
# 然后在另一个终端跑 client.py

客户端用 httpx 流式读回、边到边打印，完整真源见 code/06-流式输出/client.py：

import sys

import httpx

sys.stdout.reconfigure(encoding="utf-8")

# 先启动 server.py（uvicorn server:app --reload），再运行本文件
with httpx.stream(
    "GET",
    "http://127.0.0.1:8000/chat",
    params={"q": "100 美元能换多少人民币？"},
    timeout=httpx.Timeout(None, connect=5.0),  # 只限连接超时；读不限，SSE 事件间隔可能很长
) as r:
    event = ""
    finished = None  # "done" / "error" / None（流异常结束）
    for line in r.iter_lines():
        if line.startswith("event: "):
            event = line[len("event: "):]
        elif line.startswith("data: "):
            data = line[len("data: "):]
            if event == "done":
                finished = "done"
                print("\n[结束]")
                break
            elif event == "error":
                finished = "error"
                print(f"\n[出错] {data}")
                break
            # 普通 token：边到边打印，不打换行 —— 模拟打字机
            print(data, end="", flush=True)

    if finished is None:
        print("\n[连接中断] 未收到 done 事件，服务端可能中途异常")

跑起来（终端一：cd code/06-流式输出 && uvicorn server:app --reload；终端二：cd code/06-流式输出 && python client.py）：

按 1 美元 = 7.25 人民币计算，100 美元约可换 **725 元人民币**。
[结束]

一个后端接口 + 一个客户端，就闭环了「模型 token → SSE 事件 → 前端逐字渲染」的整条链路。

七、工程化要点：前后端流式协议与背压

① 协议设计：别裸推文本，给每个事件命名。 上面的 token / done / error 三个事件名就是最小协议；真实系统再加 progress（自定义事件）、meta（工具名/耗时等）。前端按 event 字段分派渲染逻辑，而不是去猜文本内容。协议一但定下，前后端各自独立演进。

② SSE 够用，WebSocket 别急着上。 Agent 流式输出是单向推送，SSE 是标准 HTTP、天然过代理/负载均衡，且浏览器 EventSource 原生支持（注意它无法设置 Authorization 头，鉴权只能走 Cookie/查询参数；要 header 鉴权就得用 fetch 手动读流）。只有需要双向（用户中途打断、再发指令）时，才考虑 WebSocket 或 SSE + 独立 POST 的组合。

③ 背压来自 ASGI 写入侧，重活一律 await。 真正的背压机制在 EventSourceResponse 的写入侧：它每收一个事件就 await send(...)，当前端（或 TCP 接收窗口）慢时，这个 send 会阻塞，阻塞向上传导到 async for，模型才不再往下拉 token——所以前端慢，token 就自然停在前方，不会堆爆内存。前提是别在生成器里放同步阻塞代码：同步调用会冻住整个事件循环，并发请求被串行化、心跳与断线检测全部失效。这就是六节为什么用 agent.astream（异步）而不是同步 stream——这是流式后端的第一条红线，重活一律 await 或丢线程池。

④ 断线重连 ≠ token 级续传，别混为一谈。 两个机制要分开：(a) 传输层重连——SSE 的 Last-Event-ID 只有在服务端给每个事件发 id: 字段时才有效，浏览器断线重连带上的「最后事件 id」只能让服务端从事件日志重放，这要求你自建事件缓冲，且对非确定性 LLM 的实时 token 流不适用；(b) 执行层恢复——第 5 篇的 Checkpointer + thread_id 做的是「从最后一个 checkpoint 恢复图执行」，它只保证已完成的工具副作用不重跑，并不能把已推送的 token T1…T50 接着续传 T51。模型生成中途断线，按同一 thread_id 续跑会重新调模型重新生成（输出可能变、token 重复计费）。所以正确姿势是：断线即取消本轮（见⑥），客户端重发整轮，thread_id 只负责不让工具副作用重复执行。

⑤ 心跳防掐线。 长连接穿过 Nginx/云网关时，一段时间没有数据会被当成死连接掐掉。模型思考（或工具调用）的那几秒正是「空窗期」，所以要发心跳——SSE 里一条 : 开头的注释帧就是标准心跳（sse-starlette 默认发 : ping - <时间戳>）。六节代码里 EventSourceResponse(..., ping=15) 显式写出 15 秒间隔（其实这正是库的默认值，写出来只为说明），不用自己发。

⑥ 断线取消 + 中途兜底，是流式后端的成本与体验底线。 两件事都必须做：(a) 断线取消——SSE 框架（Starlette/sse-starlette）在检测到客户端断开后会取消响应生成器，但检测/取消可能滞后（长 prefill、上游 SDK 取消传导不及时、同步阻塞时尤其如此）。在生成器里显式 await request.is_disconnected() 检测断开即 break（六节已做），是为了更早停掉上游生成，少付这段空转的 token 钱。(b) 中途兜底——流式没有「一次性返回」这个明确结束点，模型报错/工具异常/超时都可能在中途发生，务必 try/except 并 yield {"event": "error", ...}（六节已做），前端据此把已渲染的半截答案标记为「不完整/不可信」。注意 error 事件只回错误码/简短原因，完整 traceback 记到服务端日志，别把异常原文（可能含内部路径）裸发客户端。另外记一句成本定性：流式只改善体感（首字延迟），不省 token——所有生成 token 照计费。

⑦ 并发与连接管理。 SSE 长连接会长期占用一条连接 + 一次 LLM 调用。六节的 model/agent 是模块级单例，演示够用；上生产要给每轮一个 thread_id（顺带隔离会话）、加并发上限（信号量/队列）、连接超时，以及鉴权/限流/PII 脱敏——这些属于后续「服务化与可观测」「评估与护栏」篇的主题，这里先立住流式这条线。

（本文代码实测于 langgraph 1.2.14 / langchain-core 1.6.9 / langchain 1.4.4 / langchain-deepseek 1.1.1 / fastapi 0.162.0 / uvicorn 0.54.0 / sse-starlette 3.6.0 / httpx 0.28.1；完整真源见 code/06-流式输出/ 下 6 个 .py。）

附：「Deep Agents 等价实现」贴士

Deep Agents 等价实现
本文的流式输出在 Deep Agents 套件层写法通用——create_deep_agent 内部把 create_agent 的编译图加了文件、子 Agent 等中间件，但 .stream(stream_mode=...) / .astream_events(version="v2") 接口一致，同一套代码原样可用（本系列锁 deepagents 0.7.23；其事件流 API 已发布 version="v3"，本文仍用 v2，用前先对一下锁定的版本）：

from deepagents import create_deep_agent

agent = create_deep_agent(model=..., tools=[...])
for chunk, meta in agent.stream({"messages": [...]}, stream_mode="messages"):
    if meta["langgraph_node"] == "model" and chunk.content:
        print(chunk.content, end="", flush=True)

区别在于套件的内置中间件（文件、子 Agent 等）也在往流里吐东西：在 astream_events 下写文件表现为 on_tool_start / on_tool_end，在 stream_mode="messages" 下则表现为工具节点的 ToolMessage；子 Agent 派发会带出嵌套的 langgraph_checkpoint_ns。所以「只取模型文字」时，meta["langgraph_node"] == "model" 这个过滤器比在裸 LangGraph 下更关键——否则你会把套件的内部动作也打到前端。

参考文献

[1] LangGraph 官方文档 · Streaming（stream_mode 各取值）— https://docs.langchain.com/oss/python/langgraph/streaming
[2] LangChain 官方文档 · Streaming（astream_events v2）— https://docs.langchain.com/oss/python/langchain/streaming
[3] sse-starlette · 项目主页 — https://github.com/sysid/sse-starlette
[4] MDN · Server-Sent Events 协议 — https://developer.mozilla.org/zh-CN/docs/Web/API/Server-sent_events

---

🎯 更多专栏系列文章可以查看博客主页📑 👍 若文章对你有所触动，恳请点赞 ⭐ 关注 ⭐ 收藏