12 · 并发与网络模型：LangChain 的调度机制拆解

本片目标：回答生产级三个必问题——100 个用户同时问扛得住吗？同步/异步/批量怎么选？底层发了什么 HTTP 请求？本片把 LangChain 的线程模型和网络栈彻底拆开。 新增规定性：12（执行协议：invoke/batch/stream/ainvoke 四件套 + 并发配置） 数据字典：ContextThreadPoolExecutor / run_in_executor / RunnableConfig。 进程线程模型：本片主角——线程池、事件循环、async 并发。 网络模型：httpx Client / AsyncClient、SSE 流、超时重试。

1. 上集回顾

第 11 篇模型能调用工具了——但一次工具问答 = 多次模型调用（1 工具 = 2 次）。部署到生产前必须回答：

100 个用户同时提问，串行 invoke 一个占 1~3 秒，第 100 个用户要等 5 分钟——怎么扛？
同步、异步、批量，什么时候用哪个？为什么有 ainvoke 这种东西？
底层到底走什么 HTTP 客户端？超时、重试、连接池在哪配？

答案都在 LangChain 的执行层。本片用源码 + 实测说话。

2. 数据字典：Runnable 四件套（执行协议）

方法	底层原语	特点
`invoke`	当前线程直接执行	同步、阻塞、单次
`batch`	ThreadPoolExecutor（每次新建）	同步、多线程并发、输入是列表
`stream`	生成器	逐块 yield（第 04 篇）
`ainvoke`	asyncio	异步、事件循环、不阻塞线程
`abatch`	asyncio.gather + Semaphore	异步并发、`max_concurrency` 限流
`astream`	异步生成器	异步流式

对照规则：a 前缀 = 异步版；输入列表 = 批量版。

2.1 RunnableConfig（并发控制的关键参数）

config = {
    "max_concurrency": 5,   # batch/abatch 最多同时跑几个（默认不限）
    "timeout": 60,          # 超时秒数
    "run_id": ...,          # 追踪 ID
    "metadata": {...},      # 附加元数据（LangSmith 追踪用）
}
chain.batch(inputs, config=config)

3. 数据字典：LangChain 的线程池

3.1 `ContextThreadPoolExecutor`（langchain_core 的专用线程池）

from langchain_core.runnables.config import ContextThreadPoolExecutor, run_in_executor

它和标准 ThreadPoolExecutor 的区别：submit 时用 copy_context() 复制当前线程的 contextvars 上下文到工作线程。为什么？——因为 LangChain 用 contextvars 传递 run_id、追踪信息，如果线程不复制上下文，异步调用里的追踪就断了。

3.2 `run_in_executor`（同步↔异步的桥）

await run_in_executor(None, chain.invoke, input)   # 把同步调用丢进线程池，等结果

用途：在 async 代码里调用同步函数，避免阻塞事件循环。

4. 实测：并发行为的真相（2026-08 本机，智谱 glm-4.7）

我用 3 个问题做了一组对照实验：

串行 invoke ×3 :  4.1s   ← 一个接一个，每次等模型返回
batch ×3       : 62.2s   ← 线程池并发 3 个请求
abatch ×3      : 12.0s   ← 事件循环并发 3 个请求

⚠️ 最重要的发现：batch 比串行还慢！ 为什么？ 因为智谱 API 有速率限制（rate limit）——3 个请求同时打过去，部分触发 429 Too Many Requests，SDK 自动重试（默认最多 2 次），重试又要等。所以：

并发 ≠ 一定更快。并发是”潜力”，实际收益取决于 API 的限流策略。 生产环境要对 API 的 QPS 上限做压测，再定 max_concurrency。

另一个实测确认：RunnableParallel（第 09 篇的 {"context":..., "question":...}）的同步 invoke 内部确实用 ThreadPoolExecutor——两条分支是真并行：

RunnableParallel.invoke → ThreadPoolExecutor-2_0 → 两条子任务并行执行

5. 进程线程模型（全图）

┌─ 同步世界（你的主线程）─────────────────────────────┐
│  chain.invoke()  → 当前线程一步步执行               │
│  chain.batch()   → 新建 ThreadPoolExecutor          │
│                     └─ ContextThreadPoolExecutor     │
│                        （复制 contextvars → 工作线程）│
└───────────────────────────────────────────────────┘

┌─ 异步世界（asyncio 事件循环）───────────────────────┐
│  chain.ainvoke() → 协程，事件循环调度                │
│  chain.abatch()  → asyncio.gather + Semaphore        │
│  await model.ainvoke() → httpx.AsyncClient 非阻塞    │
│  run_in_executor(None, sync_fn) → 同步函数丢线程池    │
└───────────────────────────────────────────────────┘

两条世界线的桥：run_in_executor。同步调用会被包装成 Future 丢进线程池，事件循环等 Future 完成——这是 FastAPI（第 14 篇）里混用同步/异步的关键。

6. 网络模型：httpx 全栈

LangChain 生态的 HTTP 客户端是 httpx（不是 requests、不是 urllib）。本机实测依赖：

包	对 httpx 的版本要求	用同步还是异步
anthropic SDK	`httpx>=0.25,<1`	同步 `Client` / 异步 `AsyncClient`
openai SDK	`httpx>=0.23,<1`	同上
chromadb	`httpx>=0.27`	同步为主
httpx-sse	—	解析 SSE 流（第 04 篇）

6.1 同步栈（invoke 路径）

model.invoke() → anthropic 同步 Client → httpx.Client
  └─ POST https://open.bigmodel.cn/api/anthropic/v1/messages
       │ 连接池：httpx.Client 内部管理 keep-alive 连接
       │ 超时：默认 600s（可配 timeout 参数）
       │ 重试：429/5xx 自动重试（max_retries，默认 2）
       └─ 响应 → AIMessage

6.2 异步栈（ainvoke 路径）

await model.ainvoke() → anthropic AsyncClient → httpx.AsyncClient
  └─ POST（非阻塞！发出请求后事件循环可以去干别的）
       └─ 响应到达 → 事件循环唤醒协程 → AIMessage

异步为什么能扛高并发：invoke 阻塞线程 1~3 秒，100 并发需要 100 个线程；ainvoke 在事件循环里等待时不占线程，100 并发只需要 1 个事件循环线程 + N 个协程。这是 FastAPI async 路由的核心优势（第 14 篇）。

6.3 流式网络模型（SSE）

stream() → POST（stream: true）→ HTTP 长连接
  └─ 服务端逐事件推送（content_block_delta...）
       └─ httpx-sse 解析 → AIMessageChunk 逐块 yield

7. 关键方法速查

方法	场景	建议
`invoke`	单次问答（CLI/测试）	默认
`batch`	离线批量处理（建库时批量 embedding）	注意 API 限流，设 max_concurrency
`stream`	打字机效果	用户体验最好
`ainvoke/abatch`	服务端高并发（第 14 篇 FastAPI）	生产首选
`run_in_executor`	异步代码里调同步函数	桥

8. 验证：跑起来

配套代码 code/12_async_bench.py：

对照实验：串行 invoke vs batch vs abatch（3 问 × 3 种方式计时）；
观察 RunnableParallel 内部线程名（证明线程池存在）；
abatch 限流演示：max_concurrency=1 时退化为串行；
打印 httpx 客户端信息。

cd enterprise-rag-course\code
python 12_async_bench.py

预期输出（节选，数字随网络波动）：

串行 invoke ×3 :  4.1s
batch ×3       : 62.2s   ← 可能比串行慢！API 限流触发 429 重试
abatch ×3      : 12.0s   ← 事件循环并发

RunnableParallel 子任务线程: ThreadPoolExecutor-2_0   ← 证明内部用线程池

abatch max_concurrency=1 : 串行执行（限流退化为顺序）

9. 边界

batch 不一定快——API 限流时并发触发 429 重试反而更慢。先压测再定并发数。
线程池每次 batch 新建——高频调用建议自己管理连接复用（或直接走异步）。
同步阻塞代码里别混 async——asyncio.run() 在同一线程二次调用会报错。
本地 Chroma 的 HNSW 索引是线程安全的吗——多线程并发检索建议用同一实例（第 08 篇已提）。写入要加锁。

10. 未完待续

并发模型懂了。但这一篇我们反复钻进 langchain_core.runnables.config、httpx 这些底层——你已经在用 langchain_core，但只用了它 36 个模块里的几个。 它到底还有多少模块？哪些是你以后的武器、哪些是必须避开的坑？——这就是第 13 篇：langchain_core 全地图。 → 13 · langchain_core 全地图

RAG 教程

小林学 LangChain

12 · 并发与网络模型：LangChain 的调度机制拆解

12 · 并发与网络模型：LangChain 的调度机制拆解

1. 上集回顾

2. 数据字典：Runnable 四件套（执行协议）

2.1 RunnableConfig（并发控制的关键参数）

3. 数据字典：LangChain 的线程池

3.1 `ContextThreadPoolExecutor`（langchain_core 的专用线程池）

3.2 `run_in_executor`（同步↔异步的桥）

4. 实测：并发行为的真相（2026-08 本机，智谱 glm-4.7）

5. 进程线程模型（全图）

6. 网络模型：httpx 全栈

6.1 同步栈（invoke 路径）

6.2 异步栈（ainvoke 路径）

6.3 流式网络模型（SSE）

7. 关键方法速查

8. 验证：跑起来

9. 边界

推荐资料（延伸阅读）

10. 未完待续

​12 · 并发与网络模型：LangChain 的调度机制拆解

​1. 上集回顾

​2. 数据字典：Runnable 四件套（执行协议）

​2.1 RunnableConfig（并发控制的关键参数）

​3. 数据字典：LangChain 的线程池

​3.1 ContextThreadPoolExecutor（langchain_core 的专用线程池）

​3.2 run_in_executor（同步↔异步的桥）

​4. 实测：并发行为的真相（2026-08 本机，智谱 glm-4.7）

​5. 进程线程模型（全图）

​6. 网络模型：httpx 全栈

​6.1 同步栈（invoke 路径）

​6.2 异步栈（ainvoke 路径）

​6.3 流式网络模型（SSE）

​7. 关键方法速查

​8. 验证：跑起来

​9. 边界

​推荐资料（延伸阅读）

​10. 未完待续

12 · 并发与网络模型：LangChain 的调度机制拆解

1. 上集回顾

2. 数据字典：Runnable 四件套（执行协议）

2.1 RunnableConfig（并发控制的关键参数）

3. 数据字典：LangChain 的线程池

3.1 `ContextThreadPoolExecutor`（langchain_core 的专用线程池）

3.2 `run_in_executor`（同步↔异步的桥）

4. 实测：并发行为的真相（2026-08 本机，智谱 glm-4.7）

5. 进程线程模型（全图）

6. 网络模型：httpx 全栈

6.1 同步栈（invoke 路径）

6.2 异步栈（ainvoke 路径）

6.3 流式网络模型（SSE）

7. 关键方法速查

8. 验证：跑起来

9. 边界

推荐资料（延伸阅读）

10. 未完待续