上篇聊了 AI Agent 怎么知道自己有哪些工具可以用。那问题来了,工具注册完了,Agent 手里攥着几十个能力,接下来一个很直接的事——你让它同时干好几件事的时候,它是一件一件来,还是一起上?

这个问题看着简单,但往里挖一层就复杂了。

举个例子。你跟 AI 说「帮我对比这三个文件」,AI 可能一次性想干三件事,读 a 文件、读 b 文件、读 c 文件。

两个选择摆在这。

要么串行,a 读完了读 b,b 读完了读 c。安全,但你要等三轮。

要么并行,三个一起读。快,但如果里面混了一个写文件的操作,两个写同时往同一个文件里塞东西呢?那叫竞态条件,文件直接乱掉。

如果三个都只是读,那并行显然更好。但如果一个是写文件,另一个也在写同一个文件呢?或者有个「向用户提问」混在里面,你总不能同时弹三个问题让用户回答。

这就是这篇文章要拆的东西,hermes-agent 怎么判断一批操作能不能一起跑,以及这背后的安全分级思想。
AI Agent 工具执行策略

先搞清楚 AI 到底能干什么

在看怎么执行之前,得先知道 AI 手里有哪些牌。hermes-agent 有 46 个内置工具,外加动态加载的 MCP 工具,大部分跟并行串行没关系,我把跟执行策略相关的挑出来。

文件操作

读文件,让 AI 去看你电脑上的某个文件内容。写文件,创建或覆盖一个文件。文件修补,修改文件的某几行而不是整个重写,用过 git 的朋友知道 git diff 长什么样。搜文件,在项目目录里按文件名匹配搜索。

搜索类

网页搜索,上网搜东西。网页抓取,给一个具体网址,把里面的内容扒下来读。搜文件上面说了。翻聊天记录,在跟 AI 的历史对话里全文搜索。

系统操作类

命令行,让 AI 直接在电脑上敲 shell 命令。跑代码,在沙箱里执行一段 Python。

AI 内部管理类

问用户,AI 拿不准的时候弹框让你确认,「这文件删吗?」。待办清单,AI 给自己维护一个任务列表,做完划掉。记事情,把信息存到长期记忆里,关了对话下次还能想起来。识图,分析图片内容。技能列表和看技能,查看自己有哪些能力。派活给子 AI,把大任务拆出去交给另一个 AI 实例并行干。

智能家居类(跟 Home Assistant 系统对接,比较小众)

查设备状态,比如「客厅灯现在开着还是关着」。列设备清单,家里有哪些智能设备。列服务清单,有哪些自动化可用。调用服务,执行开关灯之类的写操作。

另外还有浏览器操作、RL 训练等二十多个工具,跟本文的并行安全分级没有直接关系,先略过。现在你知道 AI 手里有哪些牌了,接下来看它怎么出牌。


不是所有牌都能一起出

最核心的设计是一个四级分类。每个能力被标上一个安全等级,这个等级直接决定它能不能跟别人一起跑。

第一级,随时可以一起跑。

总共 11 个工具。网页搜索、网页抓取、识图、搜文件、翻聊天记录、技能列表、看技能、查设备状态、列设备清单、列服务清单、读文件。共同点是只读,不改变任何东西,没有共享状态。

第二级,看操作的文件是不是同一个。

就 3 个工具。读文件、写文件、文件修补。它们都操作文件,但操作不同文件的时候完全可以一起跑。读 a 文件跟写 b 文件,互不干扰。但如果两个操作都指向同一个文件,就不能一起跑了,得排队。

这里有个细节容易让人困惑。读文件同时在两级里都有名字。什么意思呢?实际判断的时候,先走第二级的路径冲突检测。两个读文件如果读的是不同文件,那就并行,相安无事。但如果读的是同一个文件,检测到路径重叠,就退回串行。

你可能会问,两个读操作读同一个文件,操作系统层面完全安全啊,为什么要串行?

说实话,这个设计偏保守了。作者的思路是统一逻辑,只要是文件操作,不管读写,都走路径检测。代价是读同一个文件的场景被过度串行化了。好处是代码简单,不会漏判。这是一个典型的工程取舍——用少量不必要的串行化换取冲突检测的零误判。

第三级,绝对不能一起跑。

就 1 个,问用户。同时弹三个问题给用户,用户该回答哪个?交互式操作必须排队,一个问完再问下一个。

第四级,没标注的默认不能一起跑。

46 个内置工具里,进了上面三个安全名单的只是少数,剩下 32 个——命令行、跑代码、派活给子 AI、记事情、待办清单、浏览器操作、RL 训练等等——全都没标注。没说自己安全的,就当你不安全。白名单思维,跟网络安全的默认拒绝一个道理。

四级分类,落到代码里是这样:

# 第一级:只读、无共享状态 —— 11 个,随时可以并行
_PARALLEL_SAFE_TOOLS = frozenset({
    "read_file", "search_files",        # 读文件、搜文件
    "web_search", "web_extract",        # 网页搜索、网页抓取
    "vision_analyze",                   # 识图
    "session_search",                   # 翻聊天记录
    "skill_view", "skills_list",        # 看技能、技能列表
    "ha_get_state", "ha_list_entities",
    "ha_list_services",                 # 智能家居:查状态、列设备、列服务
})

# 第二级:操作文件,路径不重叠时可以并行 —— 3 个
_PATH_SCOPED_TOOLS = frozenset({"read_file", "write_file", "patch"})

# 第三级:绝对不能并行 —— 1 个
_NEVER_PARALLEL_TOOLS = frozenset({"clarify"})  # 问用户

# 第四级:没标注的默认不能并行 —— 32 个
# 没有对应常量,不在上面三个名单里的,自动回串行

怎么判断,整批一起判

当 AI 收到一批操作时,决策不是一个个工具单独判断的,而是整批一起定,要么全并行,要么全串行。判断逻辑就一条线:

  • 只有 1 个操作 → 串行,没有并行的意义
  • 混进了问用户 → 全部串行,哪怕另外两个是读文件也不行,不搞混合执行
  • 参数解析失败 → 全部串行。LLM 返回的东西不是 100% 合法 JSON,缺引号、多逗号、括号没配对太常见了。解析不了就不知道参数里有没有路径,也就不知道会不会冲突,安全第一
  • 有文件操作且路径重叠 → 全部串行。这里的重叠判断很激进,不是只判断完全相同的路径,而是路径前缀匹配——/home/project//home/project/src/main.py 共享同一个目录前缀,就算重叠。它不分析你到底在读写什么,只看路径有没有关联。有关联就串行,宁可多判不要漏判
  • 所有操作都在安全名单里 → 整批并行
  • 有没标注分类的操作 → 串行

上面这六条,翻译成代码不到 40 行:

def _should_parallelize_tool_batch(tool_calls) -> bool:
    if len(tool_calls) <= 1:
        return False                                  # 只有 1 个 → 串行

    tool_names = [tc.function.name for tc in tool_calls]

    if any(name in _NEVER_PARALLEL_TOOLS for name in tool_names):
        return False                                  # 混进了问用户 → 全部串行

    reserved_paths = []
    for tool_call in tool_calls:
        tool_name = tool_call.function.name

        try:
            function_args = json.loads(tool_call.function.arguments)
        except Exception:
            return False                              # 参数解析失败 → 串行
        if not isinstance(function_args, dict):
            return False                              # 非标准参数 → 串行

        if tool_name in _PATH_SCOPED_TOOLS:
            path = _extract_parallel_scope_path(tool_name, function_args)
            if path is None:
                return False                          # 路径提取失败 → 串行
            if any(_paths_overlap(path, p) for p in reserved_paths):
                return False                          # 路径重叠 → 串行
            reserved_paths.append(path)
            continue

        if tool_name not in _PARALLEL_SAFE_TOOLS:
            return False                              # 未分类 → 串行

    return True                                       # 全部通过 → 并行

执行起来的工程细节

决策做完了,开始执行。

并行用的是线程池,不是异步。原因上一篇讲过了,Agent 核心是同步的,大部分工具的处理函数是同步的,线程池适配这些场景更自然。

但有个更细的问题。一堆任务提交到线程池之后,主线程怎么等结果?

as_completed() 是标准做法,哪个先跑完就先处理哪个。它虽然也支持 timeout 参数做轮询,但用起来很别扭——在一个迭代器循环里既要处理已完成的任务、又要检查中断、又要发心跳,三个职责搅在一起。

所以这里换成了 wait(timeout=5.0)。每 5 秒醒一次,醒来看看还有没有没跑完的任务,检查一下用户有没有发停止信号,该发心跳发心跳,然后继续等。等待结果和中断检查两个职责在循环里交替执行,逻辑清晰。长任务里的响应性,就靠这种可中断轮询来保证。

中断处理分三层。

线程池启动前,如果中断信号已经亮了,直接跳过全部操作。

执行中检查到中断,取消所有还没开始的任务,给已经跑起来的 3 秒宽限期。

收尾阶段,被中断的操作结果塞一条取消提示,追加到对话记录里。

串行执行也一样,每个操作执行前先看一眼中断信号,亮了就跳过剩下的。


有些操作比较特殊

无论是并行还是串行,最终每个操作都要走到一个统一的入口函数。

上篇讲过,工具注册中心是个纯粹的查表机制,handler 只收参数,返回结果,无状态。但有 4 个工具被标记为必须由 Agent 实例亲自处理,加上另外 2 个特殊工具,一共 6 个不能直接走注册中心。

待办清单,需要当前会话的任务列表对象。记事情,需要持久化的记忆仓库,跨会话保留。翻聊天记录,需要数据库连接来搜历史对话。派活给子 AI,需要当前 Agent 实例的引用,子 AI 得知道它爹是谁。问用户需要回调函数弹交互框,外部记忆提供者需要对接向量数据库。这 6 个都依赖 Agent 自身的内部状态,注册中心是个全局单例,压根不知道当前是哪个 Agent 在跑。

所以这个入口函数做的事就是按需注入状态。安全插件拦截最先,然后是有状态工具逐个匹配注入,最后才丢给通用调度。两层各管各的,互不侵入。跟上篇的核心层纯粹、调度层脏,完全一致的思路。好的分层设计是有传染性的。


防手滑,自动存档

上篇文章提过 Agent 有防手滑的机制,这里展开说一下。

这个东西叫 Checkpoint。它自动存的,但回滚是你手动触发的。Agent 只负责在危险操作前拍个快照,不会自己帮你回滚,控制权在你手里。

你发现 AI 改坏了文件,敲 /rollback 列出所有存档点,然后 /rollback 3 恢复到第 3 个,或者只恢复单个文件。底层就是 git 的快照机制,回滚之前还会自动多存一个存档,这样回滚操作本身也可以撤销。

存档触发点有两个。写文件和文件修补操作前,危险终端命令前(rmmvsed -igit reset 这些,通过正则匹配识别)。注意是在执行前存档,不是执行后。坏了再存就晚了。


总结

第一个,安全分级加白名单模型。把工具按安全等级分级,没声明安全的默认不安全。跟网络安全里的默认拒绝一个思路。你在自己的系统里加工具时,先想清楚,这个东西能不能跟别人一起跑?它操作的是共享状态还是独立资源?

第二个,路径作用域。不是简单的读写串行只读并行。文件工具操作不同文件时可以并行,这个粒度用极少量的路径检查换来了大量场景的性能提升。

第三个,主线程不能睡觉。可中断轮询把等待结果和检查心跳/中断两个职责交织在一个循环里,长任务里的响应性就得靠这个。

第四个,状态注入。注册中心不知道 Agent 实例的存在,Agent 内部状态通过入口函数注入给那几个特殊工具。两层各管各的,互不侵入。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐