路线 A：线程池（ThreadPoolExecutor）
  → 同步代码（invoke/batch/stream）的并发手段
  → 适合"阻塞型"任务（如同步 HTTP 请求）

路线 B：事件循环 + 协程（asyncio）
  → 异步代码（ainvoke/abatch/astream）的并发手段
  → 适合"IO 型"任务（如异步 HTTP 请求，await 不阻塞）

# langchain_core/runnables/config.py:607（实测源码）
class ContextThreadPoolExecutor(ThreadPoolExecutor):
    # 比普通 ThreadPoolExecutor 多干一件事：
    # 提交任务时 copy_context().run(func, ...)
    # → 把当前线程的 contextvars（配置、回调）复制进工作线程

# langchain_core/runnables/config.py:678（实测源码）
async def run_in_executor(executor_or_config, func, *args, **kwargs):
    # 传 None 或 dict（RunnableConfig）→ 用 asyncio 事件循环自带的默认线程池
    if executor_or_config is None or isinstance(executor_or_config, dict):
        return await asyncio.get_running_loop().run_in_executor(
            None, partial(copy_context().run, wrapper))
    # 传显式 Executor → 用指定的
    return await asyncio.get_running_loop().run_in_executor(executor_or_config, wrapper)

# langchain_core/runnables/base.py:966（实测源码）
if len(inputs) > 1:
    with get_executor_for_config(configs[0]) as executor:
        return list(executor.map(invoke, inputs, configs))
# 每次 batch 调用都会【新建一个线程池】

# langchain_core/runnables/base.py:1100（实测源码）
return await gather_with_concurrency(configs[0].get("max_concurrency"), *coros)
# gather_with_concurrency（utils.py:65）= asyncio.Semaphore 限流 + asyncio.gather

# langchain_core/runnables/base.py:4169（实测源码）
with get_executor_for_config(config) as executor:
    futures = [executor.submit(step.invoke, ..., c) for ...]

# langchain_core/language_models/chat_models.py:1653（实测源码）
for input in input_messages:
    self._generate_with_cache(...)   # 逐个同步发请求

# langchain_core/language_models/chat_models.py:1780（实测源码）
async def agenerate(...):
    ...
    results = await asyncio.gather(*(coros))

model.batch([3 个问题])
  └─ Runnable.batch → ThreadPoolExecutor（3 个线程）
       └─ 每个线程跑 model.invoke(一个问题)
            └─ invoke → generate → for 循环（单线程内串行）

# 1. _security/_transport.py —— httpx（安全用）
#    SSRF 防护的 httpx transport，create_httpx_client 工厂
#    （用于安全地检查外部 chain，不是模型调用）

# 2. utils/utils.py —— requests（工具函数）
#    仅 raise_for_status_with_text 辅助，不在核心链路

# 3. runnables/graph_mermaid.py —— requests（可选依赖）
#    调 mermaid.ink 渲染图

# langchain_openai/chat_models/base.py:1318（实测源码）
# 同步客户端：
self.client = self.root_client.chat.completions   # root_client = openai.OpenAI
# 异步客户端：
self.root_async_client = openai.AsyncOpenAI(...)   # base.py:1335

# _generate 里（同步，base.py:1704）：
resp = self.client.with_raw_response.create(**payload)   # 阻塞 HTTP

# _agenerate 里（异步，base.py:1970）：
resp = await self.async_client.with_raw_response.create(...)   # 非阻塞 HTTP

# anthropic/_base_client.py（实测源码）
class SyncAPIClient(BaseClient[httpx.Client, ...]):
    _client: httpx.Client          # 同步：持有 httpx.Client
class AsyncAPIClient(BaseClient[httpx.AsyncClient, ...]):
    _client: httpx.AsyncClient     # 异步：持有 httpx.AsyncClient
# 版本依赖：anthropic 依赖 httpx<1,>=0.25.0

同步栈（model.invoke / model.batch / for chunk in model.stream）：
  当前线程 → 调度层（Runnable.invoke，当前线程直跑）
    → 模型层（generate，for 循环串行）
      → 传输层（httpx.Client，阻塞 HTTP）
    [batch 时] Runnable.batch 新建线程池，每个线程跑一个 invoke

异步栈（await model.ainvoke / abatch / astream）：
  事件循环 → 调度层（Runnable.ainvoke，基类默认丢线程池 / 具体类走协程）
    → 模型层（agenerate，asyncio.gather 并发）
      → 传输层（httpx.AsyncClient，await 非阻塞 HTTP）
    [兼容桥] 只有同步实现的模型 → run_in_executor 把同步 _generate 搬进默认线程池