读懂 Codex（1.4）：Tool Call、Tool Result 与第二次模型采样

模型输出 FunctionCall 时，它并不知道命令真正运行后会发生什么。它只知道自己需要一个 新观察：测试是否通过、文件里有什么、补丁是否被接受、搜索返回了哪些结果。Runtime 真正 执行工具后，要把这个观察用 Tool Result 回灌给模型，然后模型才能继续推理。

因此，工具使用不是一次模型调用，而是一个最小双采样闭环：

采样 1 → Tool Call → Runtime 执行 → Tool Result → 采样 2

官方 Function Calling 协议也明确把这个过程拆成：应用声明工具，模型返回工具调用， 应用执行工具，用原 call_id 提交结果，然后再发请求；模型可以给出最终回答，也可以 继续请求工具 (OpenAI Function Calling)。

真正的 Runtime 比箭头更复杂：Call 什么时进历史？一次响应有三个 Call 怎么办？哪些工具 能并发？取消时是否还要写 Output？工具先后完成会不会让历史顺序随机变化？本节沿 Codex 的实际路径逐层回答。

一个 Tool Call 完成时，先记录再执行

Codex 的 SSE 处理器收到完整 OutputItemDone(FunctionCall) 后，会把 Item 交给 ToolRouter。 路由成功后，Runtime 不是先等工具跑完，而是先把模型刚刚输出的 Call Item 写入会话历史 和 Rollout，再创建工具 Future。

async def on_output_item_done(item, sampling_state):
    call = tool_router.build_tool_call(item)

    if call is None:
        await history.record(item)
        emit_visible_item_if_needed(item)
        return ItemOutcome(needs_follow_up=False)

    await history.record(item)  # 在任何副作用之前记录已接收的 Call。

    tool_future = tool_runtime.handle_tool_call(
        call=call,
        cancellation=sampling_state.turn_cancellation.child_token(),
        step_context=sampling_state.step_context,
    )
    return ItemOutcome(
        needs_follow_up=True,
        tool_future=tool_future,
    )

“先记录”不是纯粹的日志习惯，而是恢复不变式。Provider 已经把 Call 交给 Runtime，这是一个已发生 的协议事实。如果执行过程中 Turn 被取消、子进程失败或客户端崩溃，恢复后仍应看到当时模型 发出了哪个 Call。否则持久化历史会假装这次调用从未存在。

Call 先落库也带来责任：Runtime 要尽量为它补上一个配对 Output，即使结果是失败或取消。

一次模型响应可以启动多个工具 Future

“一次调用、一个结果”是最简示例，不是 Responses 的上限。模型可以在同一个 Response 中输出 多个 Tool Call Item。Codex 在一次 sampling request 期间维护 FuturesOrdered容器；每个完成 Call Item 都会立即生成一个 Future 并入队。

async def consume_one_model_response(stream):
    in_flight = FuturesOrdered()
    needs_follow_up = False

    async for event in stream:
        if isinstance(event, OutputItemDone):
            outcome = await on_output_item_done(event.item)
            needs_follow_up |= outcome.needs_follow_up
            if outcome.tool_future is not None:
                in_flight.push_back(outcome.tool_future)

        elif isinstance(event, ResponseCompleted):
            record_usage(event.usage)
            if event.end_turn is False:
                needs_follow_up = True
            break

    # response.completed 只表示 Provider 完成这次模型响应，
    # 不表示客户端工具都已完成。
    await drain_and_record(in_flight)
    return SamplingResult(needs_follow_up=needs_follow_up)

这里有两个并行的时间轴：

• Provider 时间轴持续输出 Item，最后发 response.completed；
• Runtime 时间轴在收到每个完整 Call 后就可以启动对应工具。

这种重叠能减少等待时间。例如模型依次输出两个搜索 Call，第一个不必等第二个 Item 和 Completed 事件才开始。但它没有破坏下一次采样的完整性：外层 Loop 拿到 SamplingResult 前， Runtime 会收束当前所有在途工具。

supports_parallel 不是建议，而是入场约束

多个 Future 存在，不代表所有 Handler 都可以同时修改外部世界。并发读取两个无关文件可能很安全， 但同时应用两个重叠补丁就可能破坏工作区。Codex 让每种工具声明 supports_parallel，然后 在所有工具共享的读写锁上实施。

class ToolExecutionGate:
    def __init__(self):
        self._lock = AsyncReadWriteLock()

    async def run(self, handler, invocation):
        if handler.supports_parallel:
            async with self._lock.read():
                return await handler.invoke(invocation)
        else:
            async with self._lock.write():
                return await handler.invoke(invocation)

读锁可以有多个持有者，所以多个并发工具能同时运行。写锁是独占的，所以一个不支持并发 的工具不只是与其他串行工具互斥，也会与所有正在运行的并发工具互斥。

下面的调用序列说明了这个语义：

A(parallel)  B(parallel)  C(serial)  D(parallel)

A 和 B 可以共存。C 必须等 A/B 释放读锁，然后独占执行。D 不能绕过正在等待或执行的 独占区间来破坏串行工具的隔离。这是一个调度正确性合同，不是性能提示。

FuturesOrdered 分离“执行顺序”与“历史顺序”

假设模型按顺序输出 Call A、B、C，三个工具又都允许并发。真实完成顺序可能是 B、C、A。 如果每个工具完成后直接抢锁追加历史，同一输入在不同机器上可能得到不同顺序。对 Prompt cache、 测试复现与调试都很不友好。

Codex 用 FuturesOrdered 解决这个问题。Future 本身仍可以并发运行，但消费者按入队顺序 取结果：

async def drain_and_record(in_flight):
    # Future 可已在后台任意顺序完成。
    async for result in in_flight.in_insertion_order():
        response_input_item = result.unwrap_or_raise_runtime_error()
        await history.record(response_input_item.to_response_item())

最终历史仍是：

Call A, Call B, Call C, Output A, Output B, Output C

而不是由时序偶然决定的 Output B, Output C, Output A。要强调的是，这个有序性不能取代 call_id。顺序是为了确定性，call_id 才是每个 Output 对应哪个 Call 的真正关联键。

ToolOutput 自己决定如何对模型说话

工具 Handler 的内部结果不一定是字符串。Exec 可能有退出码、标准输出、标准错误和时间； MCP 结果可能有结构化 content；某些工具可能返图片或音频。如果 Turn Loop 对所有结果做 str(result)，既会丢类型，也会让工具协议每次扩展都修改中央循环。

Codex 把模型可见转换放在 ToolOutput 抽象上：

class ToolOutput(Protocol):
    def to_response_item(
        self,
        call_id: str,
        original_payload: ToolPayload,
    ) -> ResponseInputItem:
        ...

original_payload 决定回应应该是 FunctionCallOutput、CustomToolCallOutput 还是其他配对 类型，call_id 建立关联，具体 Output 实现决定 body 是文本还是结构化 content items。

def function_output_to_response(result, call_id, payload):
    body = truncate_for_model_context(result.model_visible_body)

    if isinstance(payload, CustomPayload):
        return CustomToolCallOutput(
            call_id=call_id,
            output=body,
        )

    if isinstance(payload, FunctionPayload):
        return FunctionCallOutput(
            call_id=call_id,
            output=body,
            success=result.success,
        )

    raise FatalRuntimeError("output cannot represent this payload")

这种“输出拥有序列化”的设计让工具可以演进自己的模型观察格式，同时保持外层循环只处理 统一 ResponseInputItem。工具的原始输出与进入 Prompt 的内容也不必完全相同；超长文本会按上下文 策略截断，媒体还要受模型输入能力约束。

可恢复工具错误也是一种观察

如果 shell 被策略拒绝，最坏的处理是 Runtime 直接丢出异常并关闭整个 Agent。对模型而言， “这条路不可用”本身就是有价值的新信息：它可以换一个不需要提权的命令，可以用读文件工具， 也可以告诉用户任务受限。

Codex 将工具层结果大致分成：

结果	是否生成 Tool Output	是否可继续采样
成功	是，可带 success=true	是
RespondToModel / 可恢复失败	是，带错误观察与 success=false	是
用户取消	尽量是，aborted Output	通常 Turn 后续进入取消收尾
Fatal / task join 破坏	不强行伪装成普通结果	否，向 Runtime 错误路径传播

async def invoke_and_convert(call, handler):
    try:
        output = await handler.invoke(call.invocation)
        return output.to_response_item(call.call_id, call.payload)

    except RecoverableToolError as error:
        return failure_output_for(
            call_id=call.call_id,
            payload=call.payload,
            message=str(error),
            success=False,
        )

    except FatalToolContractError as error:
        raise RuntimeFatal(str(error))

Codex 的集成测试真正捕获第二次请求，验证超时、沙箱拒绝、未知 Custom Tool 和提权 被拒绝等内容会进入与原 call_id 配对的 Output。其中提权拒绝测试还证明循环可以走三次采样：

request 1 → Call 1
request 2 含 Call 1 失败 Output → Call 2
request 3 含 Call 2 成功 Output → Final Message

所以可恢复错误是 Agent 与环境对话的一部分，而不只是发给程序员的日志。

取消分支要区分“已完成”和“正在清理”

取消不是一个简单的 task.cancel()。取消信号与 Handler 完成可能几乎同时发生；某些工具还需要 自己终止子进程、回收管道或完成远程取消。如果 Runtime 无条件终止 task，可能把已经成功的结果 误报为 aborted，也可能把子进程留在后台。

Codex 的处理逻辑可概括为：

async def await_tool_with_cancellation(task, call, cancellation, handler):
    winner = await select(task.finished(), cancellation.triggered())

    if winner is task or task.is_finished():
        return await task  # 保留已完成结果与 completed lifecycle。

    if handler.waits_for_runtime_cancellation:
        # Handler 拥有自己的终止与清理协议。
        await task.finish_cleanup_after(cancellation)
    else:
        task.abort()
        await ignore_cancelled_join(task)

    emit_tool_aborted(call.call_id)
    return aborted_output_for(
        call_id=call.call_id,
        payload=call.payload,
    )

这里的 aborted Output 依然保留原 payload 种类和 call_id。一个 CustomToolCall 被取消后不应突然生成 FunctionCallOutput；否则历史在类型上就无法配对。

第二次采样看到什么

当 Provider 发出 response.completed 后，Codex 还会收束 in_flight。每个结果先转为 ResponseInputItem，再转成历史 ResponseItem 并记录。只有这一步完成，一次 sampling request 才向外层 Agent Loop 返回。

下一次采样并不只发送最新 Output。ContextManager 会从当前历史生成模型可见快照，其中 至少保留这个工具往返的两端：

[
    ...prior_context,
    FunctionCall(
        name="exec_command",
        arguments='{"cmd":"pytest"}',
        call_id="call_42",
    ),
    FunctionCallOutput(
        call_id="call_42",
        output="1 failed, 42 passed",
        success=False,
    ),
]

模型因此可以同时看到“我当时请求了什么”与“环境真正返回了什么”。只回灌 Output 而丢掉 Call，会让参数语义和结果来源变得含糊；只留 Call 而丢 Output，则会让协议处于未完成状态。

needs_follow_up 把工具观察转成下一次采样

工具 Future 记录完毕并不会自动在原 stream 上唤醒模型。一次 Responses 请求已经完成；Runtime 必须回到外层 Loop，决定是否发新请求。

每个已识别 Tool Call 都会设 model_needs_follow_up = true。Provider 明确返回 end_turn=false 时也会设置。外层 Loop 还把待处理的新用户输入合并进条件：

needs_follow_up = model_needs_follow_up or has_pending_input

如果需要继续，Loop 还会根据 token 使用和显式请求决定是否先做 context rollover/compaction， 再重新捕获适用的 StepContext 和世界状态，构建新 Prompt 并采样。工具执行时工作区可能已改变， 所以下一步不能无脑复用旧环境快照。

async def continue_after_sampling(result, turn):
    pending = await turn.mailbox.inspect()
    needs_follow_up = result.model_needs_follow_up or pending.has_input

    if not needs_follow_up:
        return await try_stop_hooks_or_complete(turn)

    if turn.context.requires_rollover():
        await rollover_context(turn)

    turn.step_context = await capture_fresh_step_context(turn)
    return CONTINUE_AGENT_LOOP

“第二次采样”因此不是架构中写死的 for range(2)。它是一次 Tool Call/Result 产生的 最小后继。第二次可以给 Final Message，也可以给新 Call；后者再触发第三次采样。

为什么不能边收到结果边启动下一次采样

如果模型一次输出 A、B 两个 Call，A 先完成，Runtime 是否可以马上用 Output A 发起下一次采样， 让 B 继续在后台运行？Codex 当前路径不这样做，理由可从不变式中推出：

1. 模型在同一 Response 中同时请求了 A/B，下一次观察应该完整回答这批请求；
2. 如果模型基于不完整结果再发起 C，C 可能与仍在运行的 B 产生无法预测的副作用冲突；
3. Prompt 中存在未配对 Call，会破坏 Provider 工具协议的完整性；
4. Turn 取消、历史压缩和时序复现会同时变得更难。

所以 Codex 会让工具执行与 Provider 剩余输出重叠，但在“新的模型采样”这个边界上设置 join 屏障。这是在延迟与一致性之间的明确取舍。

历史破损时的 Prompt-only 修复

正常路径尽量为每个 Call 记录 Output，但恢复旧 Rollout、截断历史或异常终止后，原始历史 仍可能存在缺口：

... FunctionCall(call_42) ...   # 没有 FunctionCallOutput(call_42)

ContextManager 在构建模型可见快照时会做配对规范化：给缺 Output 的 client-side Call 紧邻 补入 aborted Output，并删除没有对应 Call 的孤儿 Output。

def normalize_tool_pairs(items):
    calls = index_client_calls_by_call_id(items)
    outputs = index_client_outputs_by_call_id(items)

    for call_id, call_position in calls.items():
        if call_id not in outputs:
            items.insert(
                call_position + 1,
                aborted_output_with_stable_id(call_id),
            )

    items[:] = [
        item for item in items
        if not is_orphan_client_output(item, calls)
    ]

这只是 Prompt-only 协议修复。它不会恢复当时工具的真实 stdout，不会重新运行命令，也不声称 副作用从未发生。它只告诉模型：这次调用没有可用的完整结果，并保护 Call/Output 的输入形状。

server/hosted Tool Search 还有特例：它的执行和结果由 Provider 管理，不能套用所有 client-side 孤儿删除规则。这再次证明，历史规范化必须理解 Item 类型和执行所有权，不能只看 call_id。

一个可复刻的完整工具采样算法

把前面机制收束成 Mini Codex 伪代码：

async def run_one_sampling_step(turn, prompt):
    stream = await model.open_stream(prompt)
    in_flight = FuturesOrdered()
    needs_follow_up = False
    last_message = None

    async for event in stream:
        turn.cancellation.raise_if_cancelled()

        if isinstance(event, OutputItemDone):
            item = assign_item_id_if_missing(event.item)
            call = tool_router.build_tool_call(item)

            if call is None:
                await turn.history.record(item)
                last_message = update_visible_state(item, last_message)
                continue

            # Provider 已交付 Call，先保存协议事实。
            await turn.history.record(item)
            needs_follow_up = True

            future = tool_runtime.run(
                call=call,
                step_context=turn.step_context,
                cancellation=turn.cancellation.child_token(),
            )
            in_flight.push_back(future)

        elif isinstance(event, ResponseCompleted):
            turn.usage.record(event.usage)
            needs_follow_up |= event.end_turn is False
            break

        elif isinstance(event, StreamError):
            raise classify_model_stream_error(event)

    # Join 屏障：不让新采样看到半批结果。
    async for tool_result in in_flight.in_insertion_order():
        output_item = tool_result.to_response_item_preserving_call_id()
        await turn.history.record(output_item)

    turn.cancellation.raise_if_cancelled()
    return SamplingStepResult(
        needs_follow_up=needs_follow_up,
        last_message=last_message,
    )


async def run_tool_follow_up_loop(turn):
    while True:
        turn.step_context = await capture_step_context(turn)
        prompt = await turn.history.build_prompt(turn.step_context)
        result = await run_one_sampling_step(turn, prompt)

        pending = await turn.mailbox.accept_available()
        needs_follow_up = result.needs_follow_up or pending.has_input

        if not needs_follow_up:
            return result.last_message

        if turn.context.needs_rollover():
            await compact_or_rollover(turn)

这份伪代码保留了本节最重要的边界：

• 完整 Item 触发调用；
• Call 早于副作用记录；
• 多个 Tool Future 可并发；
• 每个 Output 保留 call_id 和 payload 类型；
• 结果按入队顺序记录；
• 下一次采样之前必须 join 全部在途工具；
• 是否继续由显式 follow-up 状态决定。

应该如何测试这个闭环

单元测试应保护调度不变式，端到端测试应捕获真实的第二次模型请求。

async def test_second_request_contains_paired_output():
    model.enqueue(FunctionCall(call_id="c1", name="shell", arguments="{}"))
    model.enqueue(FinalMessage("done"))
    tool.return_value = ToolText("ok")

    await agent.run("inspect")

    second = model.requests[1]
    assert second.contains_call("c1")
    assert second.contains_output(call_id="c1", text="ok")


async def test_next_sample_waits_for_all_tools():
    a = ControlledTool()
    b = ControlledTool()
    model.enqueue(tool_calls=[call("a"), call("b")])

    task = spawn(agent.run("inspect both"))
    await a.finish("A")
    assert len(model.requests) == 1

    await b.finish("B")
    await eventually(lambda: len(model.requests) == 2)


async def test_parallel_completion_keeps_insertion_order():
    # B 先完成，但历史仍按 Call A/B 的入队顺序记 Output。
    model.enqueue(tool_calls=[call("a", "c1"), call("b", "c2")])
    b.finish_first("B")
    a.finish_later("A")
    await agent.run("inspect")
    assert history.tool_output_ids() == ["c1", "c2"]


async def test_recoverable_failure_returns_to_model():
    tool.raise_error(RecoverableToolError("permission denied"))
    await agent.run("try action")
    assert model.requests[1].output_for("c1").success is False

Codex 的现有证据包括：未知 Custom Tool 的下一请求捕获，命令提权被拒后重试的三次 采样，沙箱原始错误进入 Tool Output，超时结果和元数据进入第二请求，并发门前取消， 以及历史缺失 Output/孤儿 Output 的规范化测试。

小结：工具让模型从生成器变成闭环控制器

没有 Tool Result 和第二次采样，Tool Call 只是一段结构化愿望。Codex 将它变成闭环的关键是：

1. 先记录 Call，保存 Provider 已交付的协议事实；
2. 在并发门、工具生命周期和取消边界内执行；
3. 将成功、可恢复失败或 aborted 结果转成配对 Output；
4. 按稳定顺序记录所有 Output，等当前批次完全收束；
5. 把 Call + Output 作为新观察放入 Prompt，用 follow-up 状态启动下一次采样。

这个过程可以反复任意多次。但 Agent 怎样区分“正在采样”、“正在等工具”、“需要压缩”、 “可以停止”与“已被取消”，终止钩子又如何把一次拟似完成改写成继续，这需要一个更完整 的 Agent Loop 状态机。1.5 将在这个闭环上继续向外展开。