LangGraph ToolNode 设置 max_concurrency=1 仍然并发执行：异步路径原因与临时修复

LangGraph ToolNode 设置 max_concurrency=1 仍然并发执行：异步路径原因与临时修复

先说结论：在 langgraph==1.2.10 的直接多调用 ToolNode 路径中，graph.invoke() 会遵守 RunnableConfig.max_concurrency，graph.ainvoke() 却可能忽略这个上限。 我在一个不调用模型、不需要 API Key 的最小图里一次传入四个 Tool Call。配置 max_concurrency=1 时，同步路径测得最大同时运行工具数为 1，异步路径为 4；配置为 2 时，同步路径为 2，异步路径仍为 4。

最快的临时处理不是把异步 Tool 全部改成同步，而是在真正共享限额的资源边界加一个共享 asyncio.Semaphore。这可以保护限流 API、数据库连接池、浏览器实例和其他非并发安全资源，但它只是应用层绕过方案，不会修复 ToolNode 内部的调度差异。正式方案仍应由上游让异步路径读取 max_concurrency，并补齐同步/异步一致性测试。

这个问题由 LangGraph 官方仓库 Issue #8517 在 2026 年 8 月 3 日报告。截至 2026 年 8 月 4 日，Issue 仍为 Open，已有外部开发者提出补回归测试和受限 gather 的修复方向，但还没有维护者确认或已发布修复。

本文只解决一个搜索任务：

我已经给 graph.ainvoke(..., config={"max_concurrency": 1}) 设置了并发上限，为什么同一个 ToolNode 里的多个异步 Tool 仍然一起开始？

问题边界：不是所有 Agent 并发路径都能这样推导

本次实验结构很明确：一个自定义 StateGraph，中间只有一个 ToolNode，输入消息一次携带四个 Tool Call。结论只适用于“多个调用直接进入同一个 ToolNode 的异步批处理”。

不要把它扩大成“LangGraph 所有异步 Agent 都不支持 max_concurrency”。Issue 作者在另一项对照中提到，create_agent 的 Send 调度路径可能遵守限制；XBSTACK 本次没有验证那条路径，因此文章不把它写成既定结论。生产排查时，应先确认并发究竟发生在一个 ToolNode 内部，还是发生在图级任务、Send、子图或多个节点之间。

如果你遇到的是工具超时、重试后重复执行或失败恢复，请转到 LangGraph 多智能体失败恢复：Tool Error、Timeout 与重试策略；如果需要看每个调用何时开始、何时结束，则配合 LangGraph Observability：追踪 Agent 决策路径 使用。

离线复现环境

实验在 2026 年 8 月 4 日运行，环境如下：

组件	版本
操作系统	macOS 26.5.2 arm64
Python	3.11.15
langgraph	1.2.10
langgraph-prebuilt	1.1.0
langchain-core	1.5.3
pytest	9.1.1
模型/API Key	不需要

实验目录包含复现脚本、Semaphore 方案、原始结果、版本矩阵和测试：

experiments/langgraph-toolnode-max-concurrency/
├── README.md
├── requirements.txt
├── repro.py
├── workaround.py
├── tests/
│   └── test_max_concurrency.py
└── results/
    ├── repro-results.json
    ├── repro-results.txt
    └── version-matrix.md

独立复现仓库：

langgraph-toolnode-max-concurrency-repro

最小复现：同一批四个 Tool Call

核心做法是给每个 Tool 一个相同延迟，并在开始和结束时更新共享计数器。下面省略了输出格式和文件写入，只保留关键部分：

import asyncio

from langchain_core.messages import AIMessage
from langchain_core.tools import StructuredTool
from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode


class AsyncProbe:
    def __init__(self):
        self.active = 0
        self.max_active = 0

    async def run(self, delay: float = 0.08) -> str:
        self.active += 1
        self.max_active = max(self.max_active, self.active)
        try:
            await asyncio.sleep(delay)
            return "ok"
        finally:
            self.active -= 1


async def main():
    probe = AsyncProbe()
    names = [f"tool_{index}" for index in range(4)]
    tools = []

    for name in names:
        async def tool_coroutine() -> str:
            return await probe.run()

        tools.append(
            StructuredTool.from_function(
                coroutine=tool_coroutine,
                name=name,
                description=f"Probe {name}",
            )
        )

    builder = StateGraph(MessagesState)
    builder.add_node("tools", ToolNode(tools))
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    graph = builder.compile()

    message = AIMessage(
        content="",
        tool_calls=[
            {"id": f"call-{i}", "name": name, "args": {}}
            for i, name in enumerate(names)
        ],
    )

    await graph.ainvoke(
        {"messages": [message]},
        config={"max_concurrency": 1},
    )
    print(probe.max_active)


asyncio.run(main())

在受影响版本中，输出不是期望的 1，而是 4。

完整仓库里的 repro.py 同时运行同步与异步矩阵，避免只凭一次时序判断问题。

实验结果：同步受限，异步对 1 和 2 无效

四个工具每个等待约 80 毫秒，结果如下：

执行路径	配置上限	实际最大同时运行	是否遵守
invoke()	1	1	是
ainvoke()	1	4	否
invoke()	2	2	是
ainvoke()	2	4	否
invoke()	4	4	是
ainvoke()	4	4	表面一致，因为上限等于调用数

对应耗时为：

sync  max_concurrency=1 max_active=1 elapsed=0.3461s
async max_concurrency=1 max_active=4 elapsed=0.0877s
sync  max_concurrency=2 max_active=2 elapsed=0.1727s
async max_concurrency=2 max_active=4 elapsed=0.0845s
sync  max_concurrency=4 max_active=4 elapsed=0.0913s
async max_concurrency=4 max_active=4 elapsed=0.0953s

耗时会受到机器调度影响，不能单独作为证据。更可靠的是 max_active 和轨迹：限制为 1 时，同步轨迹是一个结束后下一个才开始；异步轨迹则是四个 start 连续出现，然后才陆续 end。

根因：同步走配置化 Executor，异步直接 gather

本地安装包和 LangGraph 当前源码的关键差异在两个方法。

同步 _func() 会读取配置并创建受限执行器：

with get_executor_for_config(config) as executor:
    outputs = list(
        executor.map(self._run_one, tool_calls, input_types, tool_runtimes)
    )

get_executor_for_config(config) 能读取 max_concurrency，所以四个同步 Tool 在限制为 1 时串行执行，在限制为 2 时分两批执行。

异步 _afunc() 则先收集全部协程，再一次性交给 asyncio.gather()：

coros = []
for call, tool_runtime in zip(tool_calls, tool_runtimes, strict=False):
    coros.append(self._arun_one(call, input_type, tool_runtime))

outputs = await asyncio.gather(*coros)

这段路径没有读取 config.get("max_concurrency")。asyncio.gather() 负责等待所有协程完成，但不会替你限制同时进入工具函数的数量。因此，RunnableConfig 虽然被传到了每个 Tool 的运行上下文中，却没有约束这一批协程的启动。

LangChain Core 已提供 gather_with_concurrency(n, *coros)。上游可以考虑用它或等价实现替换无界 gather，但最终修复需要由维护者确认，因为还要处理异常传播、取消、Command 输出和现有行为兼容性。

为什么这不是一个“只是快一点”的小差异

当 Tool 只做本地无副作用计算时，额外并发可能暂时看不出问题。但生产 Tool 往往连接有限资源：

• 第三方 API 规定每秒或每租户最大并发；
• PostgreSQL、Redis、浏览器或 HTTP Client 使用有限连接池；
• 一个 Tool 持有同一文件、设备、GPU 或会话锁；
• 多个工具共同消耗昂贵配额；
• 外部系统对重复或重叠写操作不具备幂等性。

此时你在图调用处写下 max_concurrency=1，会自然认为资源已经受到保护。如果同步测试通过、生产却切换到 ainvoke()，同一批调用可能瞬间压向外部服务，出现 429、连接池耗尽、顺序错乱或重复副作用。并发失控之后的重试还可能放大流量，因此应把本文与 LangGraph 失败恢复与重试边界 一起检查，而不是简单增加重试次数。

临时方案：在受限资源边界共享 Semaphore

应用层最直接的临时方案，是让所有共享同一资源配额的 Tool 使用同一个 asyncio.Semaphore：

import asyncio

api_semaphore = asyncio.Semaphore(1)


def limit_tool(coroutine):
    async def guarded(*args, **kwargs):
        async with api_semaphore:
            return await coroutine(*args, **kwargs)
    return guarded


@limit_tool
async def search_customers(query: str) -> str:
    return await customer_api.search(query)


@limit_tool
async def search_orders(query: str) -> str:
    return await order_api.search(query)

关键不在于“每个函数都有 Semaphore”，而在于共享同一外部限额的函数必须共用同一个实例。如果 search_customers 和 search_orders 都调用同一供应商账户，却各自创建 Semaphore(1)，总并发仍然可以达到 2。

本次回归测试覆盖了：

• Semaphore 上限 1、2、4；
• 四个 Tool 的实际最大并发分别为 1、2、4；
• 一个 Tool 主动抛出 RuntimeError；
• 所有调用结束后 active 回到 0；
• Semaphore 没有保持锁定；
• 整套测试 11 passed。

async with semaphore 会在正常返回、异常抛出和取消退出上下文时释放许可。Tool 自身仍应实现合理的超时、幂等和错误处理；Semaphore 不能替代这些保护。

哪些替代方案不够可靠

只在外层给 ainvoke() 加锁

这只能限制整个图调用之间的并发，不能改变一次图调用内部同一 ToolNode 的四个 Tool Call 同时启动。如果你的问题就是单次模型响应生成多个调用，外层锁无法解决。

把 Tool 全改成同步函数

同步路径在本次版本中遵守上限，但把网络 I/O 强制改成同步可能阻塞线程、降低吞吐，并掩盖原本的异步架构问题。它可作为短期诊断对照，不应成为默认生产方案。

依赖工具调用返回顺序

asyncio.gather() 通常按输入顺序返回结果，但“结果列表顺序稳定”不等于“工具按顺序执行”。四个 Tool 可以同时开始，只是最终输出被重新按输入位置排列。对外部副作用来说，开始顺序和重叠区间才重要。

只看平均耗时

网络抖动、缓存和连接复用会让耗时产生噪声。应在 Tool 边界记录当前活跃数、峰值、开始/结束时间、请求 ID 和资源标识。可结合 LangGraph Observability 实战 建立可追踪证据。

正式修复应验证什么

无论上游最终采用 gather_with_concurrency、Semaphore 还是任务队列，回归测试至少要覆盖：

1. max_concurrency=None 保持现有并发行为；
2. 直接多调用 ToolNode 在 1、2、4 下同步与异步一致；
3. Tool 抛异常时其余任务、取消语义和错误消息符合预期；
4. ToolMessage、Command 和多输出合并不被破坏；
5. 同一配置继续正确注入每个 Tool 的 ToolRuntime；
6. 不把 ToolNode 内部限制错误扩展到图级 Send 或其他调度层；
7. 修复发布后能用旧复现仓库确认问题消失。

版本升级时，不要只检查 Issue 是否关闭。应在你的实际 Tool 集合上重新运行峰值并发测试，因为外部 API 配额、连接池和超时配置才是最终约束。

上游状态与执行建议

截至 2026 年 8 月 4 日：

• Issue #8517 为 Open；
• 问题标签包含 bug；
• Issue 提供了完整最小复现；
• 评论区已有开发者提出增加异步一致性测试和受限 gather；
• 尚无维护者确认的发布版本可供本文验证。

因此当前建议是：

1. 先用仓库中的 repro.py 判断你的版本是否受影响；
2. 对共享受限资源的 Tool 加统一 Semaphore；
3. 在测试中断言 max_active，不要只断言结果成功；
4. 为 429、连接池耗尽和重复写操作设置可观测指标；
5. 等正式版本发布后重新跑矩阵，再决定移除临时保护。

如果你的图还涉及多个 Worker 的任务交接，继续看 LangGraph Supervisor/Worker Handoff；如果并发 Tool 需要跨会话恢复状态，则结合 LangGraph Checkpointer：MemorySaver、SQLite、Redis 怎么选 设计恢复边界。更多 LangGraph 主题集中在 LangGraph 专题页。

复现与验证命令

cd experiments/langgraph-toolnode-max-concurrency
python3.11 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python repro.py \
  --json-output results/repro-results.json \
  --text-output results/repro-results.txt
.venv/bin/python -m pytest -q

当前验证结果：

11 passed in 1.11s

当上游修复发布后，测试中“异步限制 1 仍为 4”的断言应改为期望最大并发 1。不要让复现测试永远验证旧 Bug，而要把它转换成防回归门禁。

专题入口 / LangGraph Hub

继续按生产级 LangGraph 路线读，不再重复看泛入门

这一类文章统一沉淀到 LangGraph 专题页，按状态隔离、Checkpointer、HITL、失败恢复、Observability、Supervisor/Worker、Subgraph 和 Memory 顺序阅读。

进入 LangGraph 专题 → 从状态隔离开始 订阅更新