LangChain 指南 · 第 04 章
流式与并发
本章你将学会 - asyncio 前置速览：写过 JS 的话，一张表完成 Promise → 协程的迁移；gather、waitfor、任务取消各配一个最小示例 - .stream() / astream()：让拾…
本章你将学会
- asyncio 前置速览：写过 JS 的话，一张表完成 Promise → 协程的迁移；
gather、wait_for、任务取消各配一个最小示例.stream()/astream()：让拾光书屋的「AI 荐书官」像打字机一样边想边说，而不是憋满一整篇才给- 流式在链上的传递：链上哪个环节会把「流」吞成一整块，怎么自查
.batch()/abatch()：一份书单五本书并发跑，max_concurrency控制「店员数」，return_exceptions让个别失败不掀桌- 超时与回退：
asyncio.wait_for当沙漏、with_fallbacks找替补、with_retry再敲两下门、外加一个「质量软回退」的小心机
上一章（03 · LCEL 与 Runnable）我们用 | 把提示词、模型、解析器串成了链，跑的是 invoke——一问一答，等模型把整段话生成完才返回。这一章要解决拾光书屋 AI 荐书官上线前的两个体验问题：
- 等得久：一句荐语模型要生成十几秒，页面干转圈。用户要的不是更快生成完，而是更早看到第一个字——这就是流式。
- 一批活儿一起干：运营丢来一份「夏日书单」，20 本书要 20 条荐语，一本一本排队等到明天。这就是并发批量。
另外，本地 Ollama 偶尔会卡死（内存被挤爆时尤其常见），所以我们顺手把超时、回退、重试这三张「保险单」也一起配齐。
技术基线：Python + LangChain 1.x（langchain + langchain-ollama），本地 Ollama 跑 qwen3:4b。没装的话回 01 章看环境准备。
1. asyncio 前置速览：一张表从 Promise 迁到协程
作用：LangChain 的异步能力（astream、abatch）都建立在 asyncio 上。如果你写过 JS，这一节就是给你做的「方言对照」——概念同源，写法换皮。没写过 JS 也不亏，这些就是 asyncio 的核心四件套：怎么定义、怎么并发、怎么限时、怎么取消。
1.1 迁移对照表
| JS 写法 | Python 对应 | 一句话备注 |
|---|---|---|
async function fn() {} |
async def fn(): |
定义异步函数 |
await promise |
await coroutine |
等结果，期间让出执行权 |
Promise.all([a, b]) |
asyncio.gather(a(), b()) |
一个失败整体抛错，行为一致 |
Promise.allSettled([...]) |
asyncio.gather(..., return_exceptions=True) |
成败都要，异常当结果返回 |
Promise.race 手搓超时 |
asyncio.wait_for(coro, timeout) |
超时这件事 Python 直接给了成品 |
AbortController.abort() |
task.cancel() |
一个是发信号，一个是直接拔电源 |
.then().catch() 链 |
顺序写 + try/except |
Python 没有 then 链，天然线性 |
两个生态的心智模型是同一套：单线程 + 事件循环。一个线程在多个任务之间快速切换，营造「同时进行」的效果——这叫并发（concurrency）；要真正多核同时算（并行，parallelism），Python 得请 multiprocessing 出场，就像杂技演员一个人能让五个盘子都转着（并发），马戏团五个人各转一个（并行）。本章用到的全是并发，I/O 等待型任务（等模型吐字）正是它的主场。
1.2 最小示例：async/await 与 gather
import asyncio
async def gen_summary(book: str, seconds: float) -> str:
"""模拟模型生成荐语：耗时 seconds 秒。"""
await asyncio.sleep(seconds) # 真实场景里是等待模型吐字
return f"《{book}》荐语已生成"
async def main() -> None:
# gather ≈ Promise.all：两个任务同时开跑，总耗时取最慢那个
r1, r2 = await asyncio.gather(
gen_summary("夜航船", 1.0),
gen_summary("山月记", 2.0),
)
print(r1, r2) # 整体约 2 秒，而不是 3 秒
asyncio.run(main()) # 协程自己不会跑，得有人点火
想要 allSettled 风格——某本书失败了，其他结果照样要——加一个参数：
async def boom() -> str:
raise ValueError("这本书暂时缺货")
async def main() -> None:
results = await asyncio.gather(
gen_summary("夜航船", 1.0),
boom(),
gen_summary("山月记", 2.0),
return_exceptions=True, # 失败不炸锅，异常实例原样放在结果里
)
print(results)
# ['《夜航船》荐语已生成', ValueError('这本书暂时缺货'), '《山月记》荐语已生成']
# 顺序永远和传入顺序一致，和完成先后无关
1.3 超时与取消：wait_for 和 task.cancel()
超时——JS 里要拿 Promise.race 手搓一个 withTimeout，Python 里是一行成品：
async def main() -> None:
try:
text = await asyncio.wait_for(gen_summary("山月记", 2.0), timeout=1.0)
print(text)
except TimeoutError: # 3.11+ 里 asyncio.TimeoutError 就是内置 TimeoutError
print("沙漏漏完了，不等了")
asyncio.run(main()) # 输出：沙漏漏完了，不等了
Python 3.11+ 还有更顺手写法 async with asyncio.timeout(1.0):，包一段代码而不是一个协程，后面 4.1 节细说。
取消——JS 里是 AbortController 发信号、AbortError 冒出来；Python 里是拿到任务句柄直接 cancel()：
async def main() -> None:
task = asyncio.create_task(gen_summary("山月记", 5.0))
await asyncio.sleep(1.0)
task.cancel() # 拔电源
try:
await task
except asyncio.CancelledError:
print("任务已被取消")
asyncio.run(main()) # 输出：任务已被取消
常见坑：
- 协程只创建不运行。
gen_summary("山月记", 2.0)单独写一行什么都不会发生，只会收到一条 RuntimeWarning——必须await它或交给gather/create_task。 async def里禁用time.sleep。它会把整个事件循环冻住，同一循环里所有任务集体陪跑。要用await asyncio.sleep。gather默认「一个炸全炸」。它对应的是Promise.all而不是allSettled；要每份结果都拿到，加return_exceptions=True。- 别想用
except Exception接CancelledError。它继承自BaseException，要么显式except asyncio.CancelledError，要么别接。
2. 流式输出：边想边说的 AI 荐书官
2.1 .stream()：同步逐块拿文字
作用：invoke 是「整封信写完才寄」，stream 是「电报机一边敲你一边收」。模型每生成一小块（大致是 token 级别）就交给你一块，用户 1 秒内就能看到第一个字，而不是盯着空白转圈 15 秒。
签名：
def stream(
self,
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> Iterator[Output]: ...
返回一个普通迭代器，for 循环逐块消费。
最小示例（拾光书屋荐书链，第 03 章的老朋友）：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
prompt = PromptTemplate.from_template(
"你是拾光书屋的荐书官，用不超过两句话向读者推荐《{book}》"
)
model = ChatOllama(model="qwen3:4b", temperature=0.7)
chain = prompt | model | StrOutputParser()
for chunk in chain.stream({"book": "夜航船"}):
print(chunk, end="", flush=True) # 打字机效果
print()
链尾没接 StrOutputParser 时，拿到的是 AIMessageChunk，文字在它的 .content 里：
for chunk in (prompt | model).stream({"book": "夜航船"}):
print(chunk.content, end="")
常见坑：
- chunk 不是
str。模型直出是AIMessageChunk；想直接拿文字，链尾接StrOutputParser，出来的每块就是str。 - 流式省的是「首字等待」和内存，不省总耗时。15 秒的活还是要 15 秒才吐完，区别只是「憋 15 秒一次性给」变成「边产边收」。
print加flush=True。尤其是输出被重定向到日志/管道时，不加的话缓冲区可能攒一大坨才刷出来，打字机效果直接失效。- 同步
.stream()会占住当前线程。普通脚本没问题；Web 服务里每个请求都在事件循环上，应该用下一节的astream。
2.2 astream：异步逐块 + 超时保护
作用：异步版流式。每 await 一块、拿一块，等待期间事件循环还能去伺候别的任务——这正是「一本书慢慢流式输出、同时另一本书并发生成」的底座。它也和上一节的 wait_for 天然契合：超时一触发，消费任务被取消，流随之关闭（JS 里要 AbortController + 手搓 withTimeout 配合才能做到的事，Python 一个 wait_for 全包了）。
签名：
def astream(
self,
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Output]: ...
最小示例：
import asyncio
async def recommend(reader: str, book: str) -> None:
"""向某类读者流式推荐一本书。"""
async for chunk in chain.astream({"book": book}):
print(chunk, end="", flush=True)
print()
async def main() -> None:
# 两本书并发流式：屏幕上两段文字交替往外冒
await asyncio.gather(
recommend("偏爱冷门历史", "山月记"),
recommend("想给孩子挑绘本", "小王子"),
)
asyncio.run(main())
给单次生成套上超时（对应本课程 JS 版里的 streamExplain + withTimeout 组合）：
STREAM_TIMEOUT_S = 60.0
async def consume() -> None:
async for chunk in chain.astream({"book": "小王子"}):
print(chunk, end="", flush=True)
async def main() -> None:
try:
await asyncio.wait_for(consume(), timeout=STREAM_TIMEOUT_S)
except TimeoutError:
# 超时的实现方式是「取消任务」：
# consume 在最近一个 await 点被取消，astream 的异步生成器随之关闭
print(f"\n[超时] 超过 {STREAM_TIMEOUT_S} 秒，不等了")
asyncio.run(main())
常见坑：
astream返回异步迭代器。只能async for，不能next()，也不能对同一个迭代器跑两轮——要跑第二遍就重新调用astream。- 超时靠的是取消。
CancelledError会在await点冒出来向上传播，别在中间用空except把它吞了。 astream必须在事件循环里跑。脚本里记得asyncio.run(main())，在 Jupyter 里直接await即可（Jupyter 自己有循环）。
2.3 流式在链上的传递：别让一个环节把水堵死
作用：链是 prompt | model | parser 一节一节接起来的水管，「流」能不能一路流到链尾，取决于每个环节是否支持流式。模型和 StrOutputParser 都支持；你自己写的环节就未必了。
最小示例（不需要模型，本地直接跑）：
from collections.abc import Iterator
from langchain_core.messages import AIMessageChunk
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
prompt = PromptTemplate.from_template("给{book}写一句推荐语")
def fake_model(pv) -> Iterator[AIMessageChunk]:
"""假模型：yield 一块给一块——流式友好。"""
text = f"《{pv.to_string()}》：夜航人的掌灯。"
for i in range(0, len(text), 4):
yield AIMessageChunk(content=text[i : i + 4])
chain = prompt | RunnableLambda(fake_model) | StrOutputParser()
print(list(chain.stream({"book": "夜航船"})))
# ['《给夜航', '船写一句', '推荐语》', '：夜航人', '的掌灯。'] —— 一块块给
而同样一个环节，写成普通函数（一次性 return 完整结果），流到这里就被「堵死」，下游只能拿到一整块：
whole = prompt | RunnableLambda(lambda pv: "一整段文字") | StrOutputParser()
print(list(whole.stream({"book": "夜航船"}))) # ['一整段文字'] —— 只有一块
StrOutputParser 本身是流式友好的：它逐块转换，不会把上游的流攒成一坨。
常见坑：
- 「必须拿到全文才能干活」的步骤天生断流。比如把整段荐语送去做评分、做全文翻译的中间环节——这是业务性质决定的，不是 bug；要么挪到链外，要么接受它退化成单块。
- 自定义环节想要流式，写成生成器函数（
def+yield）或异步生成器（async def+yield）。普通return函数在流式调用里只会产出一个大块。 - token 级体验的上限在模型端。本地 Ollama 按它自己的生成节奏吐块，块的大小由模型分词决定，链上没法把块切得更细。
3. 批量执行：.batch() 让书单一起跑
3.1 .batch() 与 max_concurrency
作用：运营丢来一份 5 本书的书单，要 5 条荐语。串行 for 循环是一本一本排队；batch 是五个店员同时开工，总耗时取决于最慢的那本，而不是五本之和。
签名：
def batch(
self,
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]: ...
最小示例（不需要模型，肉眼可见的耗时对比）：
import time
from langchain_core.runnables import RunnableLambda
def stamp(book: dict) -> str:
"""模拟盖章：耗时 0.3 秒的慢任务。"""
time.sleep(0.3)
return f"已盖章：{book['title']}"
stamper = RunnableLambda(stamp)
books = [{"title": t} for t in ("夜航船", "山月记", "小王子")]
t0 = time.perf_counter()
stamper.batch(books) # 3 个任务并发
print(f"batch 默认并发：{time.perf_counter() - t0:.2f}s") # ≈ 0.3s
t0 = time.perf_counter()
stamper.batch(books, config={"max_concurrency": 1}) # 一次只准跑 1 个
print(f"max_concurrency=1：{time.perf_counter() - t0:.2f}s") # ≈ 0.9s
max_concurrency 就是「同时在岗的店员数」。不给它时由内部线程池自定（一般跟 CPU 核数相关，日常批量够用）；要保护本地 Ollama 或者上游接口时，显式设小一点。
对真实的荐书链，用法一模一样：
summaries = chain.batch(
[{"book": t} for t in ("夜航船", "山月记", "小王子")],
config={"max_concurrency": 2},
)
常见坑：
- 同步
batch里只能放同步函数。给RunnableLambda传了async def函数再去.batch()，会直接报错Cannot invoke a coroutine function synchronously——异步链用下一节的abatch。 - 结果是按输入顺序排好的。不管哪本书先跑完，
results[i]永远对应inputs[i]，别按完成时间对号。 - 并发上限不只看你的代码。20 路并发打本地 Ollama，模型推理是串着排队的，内存小的机器还容易被挤爆——
max_concurrency帮你稳住。
3.2 abatch：异步版批量
作用：链的环节是异步的（或者你就是想在事件循环里统一调度），用 abatch。行为和 batch 一致，只是全程 await。
签名：
async def abatch(
self,
inputs: list[Input],
config: RunnableConfig | list[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any | None,
) -> list[Output]: ...
最小示例：
import asyncio
from langchain_core.runnables import RunnableLambda
async def astamp(book: dict) -> str:
await asyncio.sleep(0.3) # 异步慢任务
return f"已盖章：{book['title']}"
async def main() -> None:
astamper = RunnableLambda(astamp)
results = await astamper.abatch(
[{"title": "夜航船"}, {"title": "山月记"}, {"title": "小王子"}],
config={"max_concurrency": 3},
)
print(results) # 约 0.3 秒跑完，而不是 0.9 秒
asyncio.run(main())
常见坑：
abatch是协程，忘了await等于没跑——和 1.2 节那个「协程只创建不运行」是同一个坑的批量版。- 同步函数也能喂给
abatch（框架会把它丢进线程池），但混搭时想清楚哪个环节真正阻塞：time.sleep占线程、asyncio.sleep占循环，语义不一样。
3.3 return_exceptions：个别翻车别掀桌
作用：默认情况下 5 本书里有 1 本生成失败，整个 batch 直接抛异常——另外 4 本已经跑完的结果也跟着陪葬。加 return_exceptions=True，失败的那个位置放一个异常实例，其余结果照常返回。
最小示例：
def stamp_or_die(book: dict) -> str:
if book["title"] == "绝版书":
raise ValueError("绝版书盖不了章")
time.sleep(0.1)
return f"已盖章：{book['title']}"
results = RunnableLambda(stamp_or_die).batch(
[{"title": "夜航船"}, {"title": "绝版书"}, {"title": "小王子"}],
return_exceptions=True,
)
# ['已盖章：夜航船', ValueError('绝版书盖不了章'), '已盖章：小王子']
for book, r in zip(books, results):
status = "失败" if isinstance(r, Exception) else "成功"
print(f"{book['title']}: {status}")
常见坑：
- 拿到结果先做
isinstance(r, Exception)体检，别把异常对象当文案打给用户。 - 它只接住「任务执行中」抛的异常；输入本身不合法（比如根本不是 list）这类错误还是当场报。
4. 超时与回退：给链装上三张保险
4.1 超时：asyncio.wait_for / asyncio.timeout
作用：本地 Ollama 卡死时不陪它等天荒地老。Python 的超时就是 1.3 节那两招，作用在「协程」上，和 LangChain 的链天然兼容——因为 ainvoke/astream 本来就是协程/异步迭代器。
最小示例——两种写法：
# 写法一：wait_for 包一个协程（ainvoke 的超时）
try:
text = await asyncio.wait_for(chain.ainvoke({"book": "夜航船"}), timeout=60.0)
except TimeoutError:
print("生成超时，走兜底文案")
# 写法二：asyncio.timeout 包一段代码（Python 3.11+，astream 的超时首选）
try:
async with asyncio.timeout(60.0):
async for chunk in chain.astream({"book": "夜航船"}):
print(chunk, end="", flush=True)
except TimeoutError:
print("\n流式生成超时")
常见坑：
RunnableConfig里没有timeout这个键。当前 langchain-core 1.x 它只有tags/metadata/callbacks/run_name/max_concurrency/recursion_limit/configurable/run_id——往 config 里塞timeout不会报错但也不生效，别被旧资料带偏。ChatOllama构造参数里同样没有统一的timeout。请求级超时就自己用wait_for/asyncio.timeout包，这是 Python 侧的标准姿势。asyncio.TimeoutError与内置TimeoutError：3.11+ 两者是同一个类，except TimeoutError通吃；要兼容老版本就写except (TimeoutError, asyncio.TimeoutError)。- 超时是「取消」实现的，被超时的链不会继续在后台白跑——但已经发出的底层 HTTP 请求要等生成器关闭时才释放。
4.2 with_fallbacks：主演忘词，替补顶上
作用：主链挂了自动切备用链，调用方基本无感。像剧场里的替补演员——主演在台上忘词（主链抛错），替补直接接上台词（备用链接手），观众看出的顶多是换了个嗓音。
签名：
def with_fallbacks(
self,
fallbacks: Sequence[Runnable[Input, Output]],
*,
exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
exception_key: str | None = None,
) -> RunnableWithFallbacksT[Input, Output]: ...
最小示例——大模型配小模型当备胎：
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
prompt = PromptTemplate.from_template("用一句话推荐《{book}》")
big = ChatOllama(model="qwen3:4b", temperature=0.7) # 主演：主力模型
small = ChatOllama(model="qwen3:1.7b", temperature=0.7) # 替补：更小的模型
primary = prompt | big | StrOutputParser()
backup = prompt | small | StrOutputParser()
safe_chain = primary.with_fallbacks([backup])
print(safe_chain.invoke({"book": "夜航船"}))
没有第二个模型也行，RunnableLambda 返回静态兜底文案，一样是合法的替补：
from langchain_core.runnables import RunnableLambda
static_backup = RunnableLambda(
lambda book: f"《{book['book']}》：今晚，让拾光书屋替你留一盏灯。"
)
safe_chain = primary.with_fallbacks([static_backup])
常见坑：
- 回退只在「抛异常」时触发。模型慢、输出质量差这类「软失败」不抛错，不会触发——想要质量不达标也换人，看 4.4 节。
fallbacks是有序的，主链失败后按列表顺序依次试，全失败才把最后的异常抛出来。- 主链和备链的输入输出类型要对齐。同一个
Input喂进去、同一种Output拿出来（上面两个例子里都是「书名字典进、荐语字符串出」），不然链式替换会对不上号。 - 默认捕获
Exception。只想在特定错误时换备胎（比如只对超时换人），用exceptions_to_handle=(TimeoutError,)收紧。
4.3 with_retry：没人应门就再敲两下
作用：网络抖一下、模型偶发 500，这种「歇一下就好」的错，直接报给用户太亏。with_retry 让链自动重试——像敲门：没人应门，等几秒再敲，敲三次还没人就走人（去触发你后面的 fallback）。
签名：
def with_retry(
self,
*,
retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
wait_exponential_jitter: bool = True,
exponential_jitter_params: ExponentialJitterParams | None = None,
stop_after_attempt: int = 3,
) -> Runnable[Input, Output]: ...
最小示例：
# 默认：最多试 3 次，带指数退避（第 2 次等 ~1 秒，第 3 次等 ~2 秒，再加随机抖动）
robust = chain.with_retry(stop_after_attempt=3)
# 和 fallback 串起来：先重试 2 次，还不行就换备胎
safe_robust = chain.with_retry(stop_after_attempt=2).with_fallbacks([backup])
print(safe_robust.invoke({"book": "夜航船"}))
常见坑：
- 重试只对「偶发错误」有意义。提示词写错、模型名拼错这种「必现错误」，重试三次等于白等三倍时间。
- 等待会越来越长。默认
wait_exponential_jitter=True（指数退避 + 随机抖动），急性子场景可以关掉换固定间隔。 - 重试会重新执行整条链。如果链里混了写库、发消息这类带副作用的环节，小心同一动作被执行多次。
4.4 质量软回退：结果太差也算失败
作用：模型说「抱歉，我无法推荐这本书」时并不抛异常——但对荐书官来说这已经是演砸了。做法：在主链尾部挂一个「校验器」，不达标就主动抛错，把「软失败」翻译成「硬失败」，让 4.2 的回退机制接管。
最小示例：
import re
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
def reject_if_lazy(text: str) -> str:
"""输出里出现敷衍话术，就视为质量不达标。"""
if re.search(r"抱歉|无法|我不能", text):
raise ValueError("荐语质量不达标")
return text
primary_checked = prompt | big | StrOutputParser() | RunnableLambda(reject_if_lazy)
resilient = primary_checked.with_fallbacks([backup]) # 备链不需要挂校验器
print(resilient.invoke({"book": "夜航船"}))
常见坑：
- 校验器抛的异常类型要能被
exceptions_to_handle接住。默认(Exception,)就覆盖了ValueError，够用；如果你收紧了捕获范围，记得把校验器的异常也加进去。 - 校验器是主链的一部分，每次重试、每次流式分块之后的最终输出都会过它——校验逻辑要轻，别再塞一个模型调用进去。
5. 本章小结
一张表带走本章所有开关：
| 你的需求 | 用什么 |
|---|---|
| 打字机效果，早点看到第一个字 | .stream() / .astream() |
| 一批输入并发跑 | .batch() / abatch() |
| 控制同时在岗的「店员数」 | config={"max_concurrency": n} |
| 个别失败不掀桌 | return_exceptions=True |
| 限时等待 | asyncio.wait_for / async with asyncio.timeout(...) |
| 主链挂了换备胎 | .with_fallbacks([...]) |
| 偶发错误再试几次 | .with_retry(stop_after_attempt=3) |
| 质量不达标也算失败 | 校验器主动抛错 + with_fallbacks |
配套 demo（都在本章 demo/ 目录）：
demo_stream.py：荐书官的打字机效果、并发流式、超时保护——需要本机 Ollama 已启动且拉取了qwen3:4b，文件头部注释写了全部前提。demo_batch.py：用RunnableLambda模拟慢任务，串行 vs 并发、max_concurrency限流、abatch异步版、return_exceptions——不需要模型，直接跑。
下一章我们解决另一个问题：模型输出的本质是字符串，而我们的代码想要结构化数据。下一章：05 · 输出解析器