同步、异步、取消与流式执行

学习目标

区分同步 Run、异步 Run 和流式输出的责任，不把它们当成三种不同的 Agent。
理解取消、超时、任务清理和外部副作用之间的边界。
读框架源码时分清最终运行状态、增量事件和用户界面展示。

前置知识

已阅读本章前六篇，理解 Run、Step、StopReason、Timeout 和 Tool 执行边界。
已有 Python 异步基础；本文只讲 Agent 执行所需的 asyncio 概念，不讲通用并发教程。

三种执行观察方式

SyncRun

SyncRun 是调用者等待函数返回最终结果的执行方式。它适合短任务、脚本和需要简单错误传播的调用链；调用期间，当前线程或任务通常会等待 Agent Loop 完成。

同步只描述调用接口，不保证内部没有并发。一个同步入口可以把工作交给后台线程或远程服务，但必须明确等待、超时和错误如何回到调用者。

AsyncRun

AsyncRun 返回一个可等待的任务或协程，调用者可以在等待期间处理其他事件。它适合 Web 请求、多个独立 Run 或需要响应取消的应用。

异步不会自动让模型更快，也不会自动并行工具。并发是否安全取决于 State 所有权、工具副作用和执行器的调度策略。

EventStream

EventStream 是按时间发送运行事件的接口，例如 run_started、step_started、delta、tool_result、run_completed。事件用于进度、日志或 UI，不一定是可恢复的最终状态。

流式事件应携带 run_id、顺序或版本信息，调用者才能检测乱序、重复和断线。断线重连应读取 RunState 或游标，而不是假设丢失的 delta 可以凭文本拼回。

Delta

Delta 是最终文本或结构化结果的增量片段。它适合低延迟展示，但单个 delta 往往不能独立解析或作为完整事实。工具结果和运行状态事件也可以流式发送，但要区分事件类型。

Final Result

Final Result 是 Run 结束后可供调用者消费的完整结果，通常包含状态、StopReason 和最终输出。它是业务层的事实来源；EventStream 中最后一段文本不能替代它。

阅读提示：SyncRun、AsyncRun 和 EventStream 只是调用或观察方式；它们共享同一运行语义，最终状态仍以 Final Result 和 RunState 为准。

取消和超时的边界

Cooperative Cancellation

协作式取消要求 Agent Loop、Model 客户端和 Tool 在安全检查点观察取消信号并尽快返回。取消信号不是强制回滚；已经发生的文件写入、邮件发送或远程发布仍需通过查询、幂等或人工处理。

Task Cancellation

Python 任务取消通常表现为向协程注入 CancelledError。执行器应在清理资源后继续抛出或转换为明确 cancelled 状态；吞掉取消异常并继续循环会让调用者以为 Run 已停止，实际上副作用仍在继续。

Timeout

Timeout 是等待上限，可以作用于一次 Model/Tool 调用或整个 Run。超时时应同时停止开始新动作、清理本地任务，并记录目标层级的 StopReason。超时不代表远程服务收到取消。

Cancellation Checkpoint

Cancellation Checkpoint 是允许安全检查取消信号的位置，例如发起工具前、工具返回后、写入状态前。检查过于稀疏会让取消响应很慢；检查在不可分割的副作用中间也无法真正撤销操作。

流式数据如何与状态共存

事件与状态分离

事件告诉观察者发生了什么，State 告诉恢复逻辑当前是什么。事件可以丢失或重复，State 必须有版本和所有权。UI 只依据 delta 画进度，提交或恢复逻辑要依据最终状态。

事件顺序

每个事件应包含 run_id、事件类型和单调序号或版本。多执行器并发时，还要有 step_id 或父事件标识。没有顺序信息的字符串流无法可靠地重建工具调用与最终结果的关系。

背压与慢消费者

慢 UI 或网络客户端不能无限制地阻塞 Agent Loop。事件层应限制队列、合并可合并的 delta 或丢弃低价值进度；关键状态变化和最终结果不能因为展示端慢而丢失。

同步、异步和流式的本地示例

下面的代码用同一个确定性工作函数提供同步结果和异步事件，展示执行模式不同但 Run 语义保持一致。

python

from __future__ import annotations

import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class RunResult:
    status: str
    output: str


def run_sync(parts: list[str]) -> RunResult:
    output = "".join(parts)
    # 输出契约：同步入口只在所有片段就绪后返回最终结果。
    return RunResult("completed", output)


async def run_async(parts: list[str], cancel: asyncio.Event | None = None):
    for index, part in enumerate(parts, start=1):
        await asyncio.sleep(0)
        if cancel is not None and cancel.is_set():
            # 状态变化：取消只阻止尚未发出的后续片段，不回滚已产生的输出。
            yield {"type": "run_cancelled", "index": index}
            return
        yield {"type": "delta", "index": index, "text": part}
    yield {"type": "run_completed", "result": RunResult("completed", "".join(parts))}


print(run_sync(["A", "B"]))


async def demo() -> None:
    events = []
    async for event in run_async(["A", "B"]):
        events.append(event)
    print(events)


asyncio.run(demo())
# 输出：RunResult(status='completed', output='AB')
# 输出：[{'type': 'delta', 'index': 1, 'text': 'A'}, {'type': 'delta', 'index': 2, 'text': 'B'}, {'type': 'run_completed', 'result': RunResult(status='completed', output='AB')}]

示例把 delta 事件和最终 run_completed 分开；调用者可以先展示 A、B，再以完整 RunResult 更新状态。实际模型流还会包含工具调用、错误和取消事件，但每种事件都应有明确类型。

使用 asyncio.wait_for 设置局部超时

下面的代码展示如何给单个异步动作设置等待上限，并在超时后清理任务；它不连接真实服务。

python

import asyncio


async def slow_action() -> str:
    await asyncio.sleep(0.02)
    return "done"


async def call_with_timeout() -> str:
    try:
        return await asyncio.wait_for(slow_action(), timeout=0.001)
    except asyncio.TimeoutError:
        # 输出契约：调用者得到可识别的超时，而不是假装动作成功。
        return "timeout"


print(asyncio.run(call_with_timeout()))
# 输出：timeout

wait_for 通常会请求取消被等待的协程，但远程服务是否停止仍取决于客户端和服务协议。对有副作用的 Tool，应优先使用可查询状态和幂等请求。

源码阅读时还原执行模式

找同步入口

搜索 run、invoke 或 execute 等阻塞入口，确认它是否直接驱动 Agent Loop，还是等待一个后台任务。查看异常如何返回、最终 State 在哪里写入。

找异步边界

搜索 async、await、create_task 和 cancellation 相关代码，确认哪些任务属于当前 Run、谁负责等待和清理。若创建任务后没有保存或 await，可能出现后台副作用泄露。

找事件发布点

搜索 yield、stream、event、delta 或 callback，确认事件在模型输出、工具执行和状态更新的哪个时刻发布。发布早于状态写入时，UI 可能先看到“完成”再发现保存失败。

找最终状态

追踪 run_completed、result 或状态快照的生成，确认它不会仅依赖最后一个 delta。断线、重复事件和慢消费者都应以最终状态作为校准。

与主线项目的对应关系

smolagents

smolagents 的同步执行入口适合先理解基础 Loop；若版本提供流式或步骤回调，应把它当作观察接口，不要把回调片段当作最终结果。工具取消和外部超时仍需由调用方处理。

OpenAI Agents SDK

OpenAI Agents SDK 同时提供同步、异步和流式运行方式。源码阅读重点是这些入口是否共享相同的 RunState、StopReason 和工具边界，而不是只比较方法名称。

LangGraph

LangGraph 的执行可以流式发布节点更新、状态更新或事件。需要确认 stream mode 发布的是增量、节点输出还是完整状态；恢复逻辑不能盲目把所有流片段拼成 State。

DeepSeek Harness

DeepSeek Harness 的事件系统和 Hook 可能在 UI、Driver 和插件之间传播增量。阅读时检查取消信号是否能穿过这些层，并确认事件发布后状态是否仍能一致落盘。

易混点

异步不等于并行：只有显式调度且副作用安全时才会并行。
流式不等于最终结果：delta 适合展示，Final Result 才是业务交付契约。
取消不等于回滚：取消本地任务无法自动撤销已经发生的外部操作。
超时不等于远端停止：客户端停止等待后，服务端请求可能仍在运行。
事件不等于状态：事件可丢失、重复或乱序，恢复需要版本化 State。

课后小问

为什么 UI 已经显示 run_completed，数据库仍可能没有新记录？
答案：事件发布和状态持久化可能不在同一个原子边界，或 UI 收到的是过早的增量事件。
解析：关键状态变化应在持久化后发布，或事件携带版本让客户端校准；业务提交不能只依据展示流。
异步任务被取消后，为什么还要等待它完成清理？
答案：任务可能持有连接、文件或执行中的子任务，等待清理能避免资源泄露和后台副作用继续运行。
解析：CancelledError 是控制信号，不是自动清理证明。Runtime 应在 finally 或任务收尾路径释放资源，并把最终状态写成 cancelled 或 timeout。

本节小结

SyncRun、AsyncRun 和 EventStream 是不同的调用/观察方式，可以共享同一个 Agent Loop 和 RunState。
取消、超时需要协作检查、任务清理和明确 StopReason；它们不保证回滚外部副作用。
Delta 用于低延迟展示，Final Result 和 State 才是恢复与业务判断的依据。
源码阅读应分别追踪同步入口、异步任务所有权、事件发布点和最终状态。

快速回顾

能解释同步、异步、流式三者的职责边界。
能判断一次超时后哪些动作可以重试，哪些必须查询或人工确认。
能说明为什么事件流和最终状态必须分离。
下一篇阅读 Agent、Workflow 与普通程序的区别，把前面概念放到更大的程序设计视角中比较。

同步、异步、取消与流式执行 ​

学习目标 ​

前置知识 ​

三种执行观察方式 ​

SyncRun ​

AsyncRun ​

EventStream ​

Delta ​

Final Result ​

取消和超时的边界 ​

Cooperative Cancellation ​

Task Cancellation ​

Timeout ​

Cancellation Checkpoint ​

流式数据如何与状态共存 ​

事件与状态分离 ​

事件顺序 ​

背压与慢消费者 ​

同步、异步和流式的本地示例 ​

使用 asyncio.wait_for 设置局部超时 ​

源码阅读时还原执行模式 ​

找同步入口 ​

找异步边界 ​

找事件发布点 ​

找最终状态 ​

与主线项目的对应关系 ​

smolagents ​

OpenAI Agents SDK ​

LangGraph ​

DeepSeek Harness ​

易混点 ​

课后小问 ​

本节小结 ​

快速回顾 ​