DevNull 知识库

Appearance

17｜流式与并发：工具怎么执行 ​

上一篇讲完调用前的预处理。这一篇讲调用后：模型返回工具调用后，怎么执行。

第 7 篇讲过模型可能一次返回多个工具调用。查 nginx 502 时，模型可能同时要查 nginx 日志和 upstream 端口连通性。这些工具调用能不能同时跑？怎么决定？这是循环里工具执行的最后一个设计问题。

先分清两件事 ​

工具并发涉及两个层面的判断，很多人混在一起。

第一层是模型决定"要不要并行发"。Responses API 的 parallel_tool_calls 默认开，模型判断多个查询互不依赖时，会在一轮里同时发起。模型自己会评估依赖——"先查 A 再根据 A 查 B"这种有依赖的，它不会拆成并行。

第二层是代码决定"并行发的这些能不能同时跑"。模型并行发了两个工具调用，不代表这两个工具能并发执行。read_log 和 read_metric 都是只读查询，能同时跑；edit_config 和另一个写操作可能改同一个文件，不能同时跑。

第一层是 parallel_tool_calls，第二层是工具的并发安全声明，两层不能混。模型决定发几个，代码决定这几个能不能同时跑。

工具自身声明并发安全 ​

并发控制的关键不在调度器，在工具自身的声明。每个工具定义时标注是否支持并发：

python
TOOL_REGISTRY = {
    "read_log": {"handler": read_log, "is_concurrency_safe": True, "is_read_only": True},
    "read_metric": {"handler": read_metric, "is_concurrency_safe": True, "is_read_only": True},
    "edit_config": {"handler": edit_config, "is_concurrency_safe": False, "is_read_only": False},
    "restart_service": {"handler": restart_service, "is_concurrency_safe": False, "is_read_only": False},
}

read_log 和 read_metric 都是只读操作，互不干扰，可以并行。edit_config 和 restart_service 都是写操作，可能修改同一文件或同一进程状态，不能并行。读操作和写操作之间也不能并行——读操作可能在写操作改到一半时读到不一致状态。

并发分区算法 ​

根据工具声明，把一次返回的工具调用分成若干"并行分区"。连续的并发安全工具组成一个并行分区，遇到非并发安全工具时切换到串行边界：

AIOps 概念图：并发分区

python
def partition_tool_calls(tool_calls):
    partitions = []
    current_parallel = []

    for tc in tool_calls:
        tool_def = TOOL_REGISTRY.get(tc.name)
        is_safe = tool_def.get("is_concurrency_safe", False) if tool_def else False

        if is_safe:
            current_parallel.append(tc)
        else:
            if current_parallel:
                partitions.append(current_parallel)
                current_parallel = []
            partitions.append([tc])   # 非安全工具独占串行分区

    if current_parallel:
        partitions.append(current_parallel)
    return partitions

输入 [read_log, read_metric, edit_config, read_log, restart_service]，分区结果是 [[read_log, read_metric], [edit_config], [read_log], [restart_service]]。前两个并行执行，edit_config 串行，下一个 read_log 单独串行（前面是写操作，不能确定写操作是否影响了日志状态），restart_service 串行。

默认保守 ​

判断并发安全时出错，默认按不安全处理。这是 fail-closed 设计：

python
is_safe = tool_def.get("is_concurrency_safe", False)
# 默认 False：没声明就是不能并发

某个工具确实能并发但忘记声明，系统串行执行，不会出错，只是慢一点。反过来，工具不能并发但被错误标成安全，会导致数据竞争或状态不一致。保守默认把风险往慢的一边推，不往错的一边推。

读操作不总是并发安全的。两个进程同时读同一个文件通常安全，但如果读操作内部有副作用（更新"最后访问时间"、修改缓存状态、记录审计日志），并发执行就可能互相干扰。is_concurrency_safe 的声明要基于工具的完整实现行为，不能只看"是不是只读"。

并发工具不能改共享上下文 ​

一个容易被忽视的约束：并发工具之间不能互相修改上下文。

python
# 错误：两个并发工具都改全局状态
def read_log_a():
    global_context["last_read_service"] = "nginx"
    return log_lines

def read_log_b():
    global_context["last_read_service"] = "redis"
    return log_lines

read_log_a 和 read_log_b 并行执行时，global_context["last_read_service"] 的最终值不确定。工具执行应该是纯函数，或者通过受控的 modifier 返回上下文变化，由外层调度器统一合并：

python
def read_log(service, context):
    lines = fetch_log(service)
    modifier = {"last_read_service": service}   # 返回修改，不直接改
    return lines, modifier

并发工具之间的 modifier 不能互相覆盖，由调度器统一合并。

流式执行 ​

除了并发，还有一个执行优化：流式。模型可能一次返回多个工具调用，传统做法是等模型输出全部完成再逐个执行。流式执行是收到一个工具调用块就尽早启动，不等后续内容，减少总延迟。

AIOps 概念图：流式事件

并发分区解决的是"能不能同时跑"，流式解决的是"能不能早点开始跑"。两者可以结合：流式收到工具调用块后，按并发分区尽早启动安全的部分。

第三阶段收尾 ​

到这里循环与控制设计六篇完成：两层循环、错误处理、停止条件、继续工作、预处理管线、流式与并发。agent 现在能安全地一步步做——循环能收敛、出错能恢复、截断能继续、上下文能维持健康、工具能并发执行。

但这些工具本身还没设计。循环里调用的工具，怎么定义接口、怎么声明能力边界、怎么管权限？这些是下一阶段工程化与协作设计要解决的。让 agent 能查日志是一回事，让它执行 rm -rf 是另一回事——必须给它套上安全护栏。