这是 LangChain 聊天模型流式传输的完整上手。核心就一句话：invoke() 是「等全量返回」，stream() 是「逐 token 吐出来」——LLM 天生按迭代方式生成 token，流式只是把这个过程实时暴露给用户[citation:web:12.8]。

一、为什么需要流式

用 invoke() 时，模型要全部生成完才一次性返回，模型思考越久，用户干等越久（20 秒、半分钟很正常），体验很差[citation:web:12.1]。流式则像 DeepSeek / ChatGPT 客户端那样打字机式逐字输出，首 token 延迟大幅降低。

| 方式 | 体验 | 适用场景 |

|------|------|----------|

| invoke() | 等待 → 一次性看到完整结果 | 脚本、批处理、API |

| stream() | 实时看到每个 token | 聊天界面、实时交互 |

二、同步流式 stream()

.stream() 返回一个迭代器，边生成边 yield 出 AIMessageChunk 消息块，取 .content 就是文本：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

for chunk in llm.stream("写一首关于秋天的诗"):

 print(chunk.content, end="", flush=True)

flush=True 很关键，否则 Python 的输出缓冲会「攒一批」再刷屏，破坏打字机效果。

三、异步流式 astream()

Web 服务、FastAPI 这类场景要用异步版本：

import asyncio

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

async def main():

 async for chunk in llm.astream("写一首关于秋天的诗"):

 print(chunk.content, end="", flush=True)

asyncio.run(main())

四、最常用的：LCEL 链式流式 + StrOutputParser

实际开发里很少裸调模型，而是用 LCEL（| 管道）拼 prompt + model + 解析器。链式结构同样支持流式，且 StrOutputParser 会帮你把每个 chunk 自动转成纯字符串：

from langchain_openai import ChatOpenAI

from langchain_core.prompts import ChatPromptTemplate

from langchaincore.outputparsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("用{style}风格写一首关于{topic}的诗")

chain = prompt | llm | StrOutputParser()

for chunk in chain.stream({"style": "幽默", "topic": "程序员"}):

 print(chunk, end="", flush=True)

这是最推荐的形式：非流式用 chain.invoke()，流式只需换成 chain.stream()，代码结构几乎不变。

五、深度探索：astream_events()

stream() 只能拿到「最终输出块」。如果你想在流式过程中观察中间事件（比如模型开始生成了、某个工具被调用了、每个 token 是什么），用 astream_events()：

async def main():

 async for event in llm.astream_events("你好", version="v2"):

 kind = event["event"]

 if kind == "onchatmodel_start":

 print("[模型开始生成]")

 elif kind == "onchatmodel_stream":

 # 每个 token 的增量内容

 print(event["data"]["chunk"].content, end="", flush=True)

 elif kind == "onchatmodel_end":

 print("\n[生成结束]")

asyncio.run(main())

事件类型（version="v2"）常见的有：

| 事件 | 含义 |

|------|------|

| onchatmodel_start | 模型开始生成 |

| onchatmodel_stream | 每个 token 增量 |

| onchatmodel_end | 生成完成 |

| ontoolstart / ontoolend | 工具调用开始/结束（Agent 场景） |

六、几个实用注意点

流式拿 token 用量：默认流式 chunk 里没有 usage 统计。需要时给模型加 streamoptions={"includeusage": True}，最后一个 chunk 的 usage_metadata 里才有完整 token 数[citation:web:12.10]。

前端集成走 SSE：后端 astream() 生成，用 FastAPI 的 StreamingResponse 包成 text/event-stream，前端用 EventSource / fetch 流式读取即可。

astream_events 别滥用：它的开销比 astream() 大，只在确实需要观察中间事件时才用；单纯要逐字输出，astream()/stream() 就够。

旧版 streaming=True 参数已过时：早期靠 ChatOpenAI(streaming=True) 开启，现在直接调 .stream() 即可，不需要那个参数。

---

需要的话，我可以给你一个「FastAPI 后端 astream() + 前端 SSE」的完整可跑示例，或者展开讲 astream_events 在 Agent 场景下怎么用事件流做进度条。