Python 与异步编程最低要求

这篇从“能够运行一段 Python”开始，逐步讲到“能够跟踪 nanobot 中一个异步 Turn”。不要求你先学完完整的 Python 或 asyncio 课程，但需要建立三个稳定的判断能力：

看到一段 Python 时，能判断对象保存了什么、谁会修改它；
看到 async def、await 和 Task 时，能判断控制权何时可能切换；
看到 Queue、Lock、Semaphore、超时和取消时，能判断它们分别保护什么。

本章的目标不是背语法

真正的完成标准是：你能沿着调用链说明“现在正在执行哪个 Task、它在等待什么、哪些状态可能被其他 Task 观察或修改”。

当前源码基线在 pyproject.toml 中声明 requires-python = ">=3.11"。本文的类型标注和 asyncio 行为按 Python 3.11 及以上版本理解。

学习目标

读完并完成练习后，你应当能够：

看懂变量、可变对象、列表、字典、分支、循环和函数调用；
看懂类、继承、组合、属性、抽象基类和 dataclass；
看懂 str | None、list[dict[str, Any]]、Callable、Awaitable 等类型标注；
区分普通函数、coroutine、Task 和最终结果；
解释事件循环为什么能在一个线程中推进多个 I/O 任务；
区分顺序等待、并发执行和真正的多线程并行；
解释 Queue、Lock、Semaphore、Event 和 Future 的不同职责；
理解超时为何常常会触发取消，以及清理代码为什么要放在 finally 中；
看懂 async with、async for、异步回调和异步生成器；
跟踪 nanobot 从消息入队、创建 Task、取得 Session lock，到调用 Provider 或 Tool 的异步路径。

如何使用本章

如果普通 Python 也不熟悉，从第一部分开始顺序阅读。
如果能熟练阅读普通函数和类，可以从“第二部分：asyncio 心智模型”开始。
第一次阅读只需理解主线；标为“认识即可”的内容不需要立即会写。
每个代码块都建议亲自运行，并在关键语句前后增加 print() 观察顺序。

flowchart LR
    A["对象、容器与函数"] --> B["类、dataclass 与类型标注"]
    B --> C["异常与资源管理"]
    C --> D["事件循环、coroutine 与 await"]
    D --> E["Task 与并发"]
    E --> F["Queue、Lock 与 Semaphore"]
    F --> G["超时、取消与清理"]
    G --> H["跟踪 nanobot 的异步 Turn"]

本章与 nanobot 源码的关系

本章是前置知识，不要求一开始逐行阅读源码。理解某个概念后，可以打开右侧位置核对真实用法。以下路径都相对于 example/nanobot/code-EN。

Python 或 asyncio 概念	nanobot 中的主要位置
`dataclass`、`field`、`@property`	`nanobot/bus/events.py::InboundMessage`、`OutboundMessage`
`Enum`、Turn 工作对象	`nanobot/agent/loop.py::TurnState`、`TurnContext`
类型标注和异步回调	`nanobot/agent/loop.py::TurnContext`、`nanobot/providers/base.py::LLMProvider`
抽象基类	`nanobot/agent/tools/base.py::Tool`、`nanobot/providers/base.py::LLMProvider`、`nanobot/channels/base.py::BaseChannel`
`asyncio.Queue`	`nanobot/bus/queue.py::MessageBus`
Task、Session lock、Semaphore	`nanobot/agent/loop.py::AgentLoop.run`、`AgentLoop._dispatch`
`asyncio.gather`	`nanobot/agent/runner.py::AgentRunner._execute_tools`
超时和取消	`nanobot/agent/runner.py::AgentRunner._request_model`、`nanobot/providers/base.py::_safe_chat`
`async with`、异步上下文管理器	`nanobot/agent/loop.py::_dispatch`、`nanobot/sdk/runtime.py::SDKRuntimeController.override`
`async for`、异步生成器	`nanobot/sdk/streaming.py::RunStream.stream_events`

第一部分：阅读 nanobot 所需的 Python 基础

1. 名字指向对象，不是把对象复制一份

Python 变量更准确的说法是“名字”。赋值语句把一个名字绑定到一个对象：

message = {"role": "user", "content": "你好"}
same_message = message
 
same_message["content"] = "修改后的内容"
 
assert message["content"] == "修改后的内容"

message 和 same_message 指向同一个字典，所以从任意一个名字修改字典，另一个名字都能看到。

1.1 可变对象与不可变对象

类型	是否通常可变	常见操作
`int`、`float`、`bool`	否	计算会产生新对象
`str`	否	拼接、替换会产生新字符串
`tuple`	否	元素位置不能重新赋值
`list`	是	`append`、`extend`、索引赋值
`dict`	是	新增、删除、修改键值
`set`	是	`add`、`remove`
普通类实例	通常是	修改实例属性

nanobot 中的 messages 通常是 list[dict[str, Any]]：外层列表和内层字典都是可变对象。阅读上下文处理代码时，要特别区分：

原地修改：messages.append(...)、message["content"] = ...；
浅复制：copied = list(messages)、copied = messages.copy()；
深复制：a =[[1, 2], [3, 4]]，b = copy.deepcopy(a) 递归复制内部对象。

messages = [{"role": "user", "content": "A"}]
shallow = list(messages)
 
shallow.append({"role": "assistant", "content": "B"})
assert len(messages) == 1  # 外层列表不同
 
shallow[0]["content"] = "C"
assert messages[0]["content"] == "C"  # 内层字典仍然共享

调试上下文时先问“这是同一个对象吗”

很多“内容为什么被改了”的问题并不是异步问题，而是两个变量引用了同一个可变对象。

1.2 `None` 表示“当前没有值”

None 不是空字符串，也不是数字零：

content: str | None = None
 
if content is None:
    print("模型还没有返回最终文本")

源码中优先使用 is None 和 is not None 判断它。str | None 表示这个位置可以保存字符串，也可以保存 None。

2. 列表、字典、元组和集合

2.1 列表 `list`

列表是有顺序、可修改的序列：

messages = [
    {"role": "system", "content": "遵守系统规则"},
    {"role": "user", "content": "读取文件"},
]
 
first = messages[0]
last = messages[-1]
recent = messages[-2:]
messages.append({"role": "assistant", "content": "处理中"})

nanobot 用列表保存消息序列、Tool Call、运行事件和 Task 集合。顺序通常有语义，尤其是 LLM 消息历史。

2.2 字典 `dict`

字典通过键查找值：

metadata = {"_wants_stream": True, "message_id": "m-1"}
 
wants_stream = metadata.get("_wants_stream", False)
metadata["session_key"] = "cli:default"

常见区别：

metadata["missing"]：键不存在时抛出 KeyError；
metadata.get("missing")：键不存在时返回 None；
metadata.get("missing", default)：键不存在时返回给定默认值；
metadata.setdefault(key, default)：键不存在时写入并返回默认值。

2.3 元组 `tuple`

元组有顺序但不能重新赋值，常用于返回多个值：

def split_result() -> tuple[str, int]:
    return "completed", 3
 
 
status, tool_count = split_result()

AgentRunner._execute_tools() 就返回包含多个部分的元组。左侧的 status, tool_count = ... 叫“解包”。

2.4 集合 `set`

集合保存不重复元素，适合成员判断：

retryable_statuses = {408, 409, 429}
 
if 429 in retryable_statuses:
    print("允许按策略重试")

2.5 推导式

推导式用一行表达“遍历、筛选、变换”：

messages = [
    {"role": "user", "content": "A"},
    {"role": "assistant", "content": "B"},
]
 
roles = [message["role"] for message in messages]
user_messages = [
    message
    for message in messages
    if message.get("role") == "user"
]

阅读复杂推导式时，可以先在纸上还原成普通 for 循环，不需要追求一眼看懂。

3. 条件、循环和提前返回

3.1 条件分支

def classify(content: str | None) -> str:
    if content is None:
        return "missing"
    if not content.strip():
        return "empty"
    return "ready"

return 会立刻结束当前函数。源码中连续的提前返回常用于过滤特殊情况，使主流程少一层缩进。

3.2 真值判断

以下值在条件中通常视为假：

False
None
数字 0
空字符串 ""
空列表 []
空字典 {}
空集合 set()

因此：

tool_calls: list[str] = []
 
if not tool_calls:
    print("没有 Tool Call")

但“没有值”和“值为零”有时语义不同。需要精确区分时，使用 is None，不要只写 if not value。

3.3 `for`、`while`、`break` 和 `continue`

def first_error(results: list[dict[str, str]]) -> dict[str, str] | None:
    for result in results:
        if result.get("status") != "error":
            continue
        return result
    return None

continue：跳到下一次循环；
break：结束整个循环；
return：结束整个函数；
while condition：条件成立时继续循环。

AgentLoop 的状态机和 AgentRunner 的模型—工具执行都包含循环。阅读时要找到：

循环条件；
哪些分支会 continue；
哪些分支会 break 或 return；
是否存在最大迭代次数。

4. 函数、参数和返回值

一个带类型标注的函数可以这样拆读：

def build_session_key(
    channel: str,
    chat_id: str,
    *,
    override: str | None = None,
) -> str:
    return override or f"{channel}:{chat_id}"

从上到下理解：

channel: str：参数名是 channel，预期值是字符串；
override: str | None = None：可选参数，默认没有覆盖值；
单独的 *：后面的参数只能按名称传入；
-> str：函数预期返回字符串；
f"{channel}:{chat_id}"：格式化字符串。

调用方式：

key_a = build_session_key("telegram", "42")
key_b = build_session_key("telegram", "42", override="shared")
 
assert key_a == "telegram:42"
assert key_b == "shared"

4.1 `*args` 和 `**kwargs`

*args 收集额外的位置参数，形成元组；
**kwargs 收集额外的关键字参数，形成字典；
调用时，*values 解包序列，**options 解包字典。

def describe(*parts: str, **metadata: str) -> tuple[tuple[str, ...], dict[str, str]]:
    return parts, metadata
 
 
result = describe("tool", "started", session="cli:default")
assert result == (("tool", "started"), {"session": "cli:default"})

Tool 的 execute(self, **kwargs) 使用这种形式接收模型生成并通过校验的参数。

4.2 函数也是对象

函数可以像其他值一样被保存、传递和调用：

from collections.abc import Awaitable, Callable
 
ProgressCallback = Callable[[str], Awaitable[None]]
 
 
async def emit_progress(text: str) -> None:
    print(text)
 
 
async def run_step(callback: ProgressCallback | None) -> None:
    if callback is not None:
        await callback("开始执行")

这里的回调类型可以从右向左读：

可选的函数；它接收一个字符串；调用后返回一个可等待对象；等待完成后没有业务返回值。

4.3 闭包和 `lambda`

内部函数可以读取外层函数的变量，这种组合叫闭包。lambda 是短小的匿名函数：

def make_prefixer(prefix: str):
    return lambda text: f"{prefix}: {text}"
 
 
format_error = make_prefixer("ERROR")
assert format_error("timeout") == "ERROR: timeout"

AgentLoop.run() 的 Task 完成回调中使用了 lambda t, k=effective_key: ...。k=effective_key 会在创建 lambda 时保存当前 key，避免以后执行回调时读到另一个循环值。

5. 模块、导入和源码路径

Python 文件通常就是一个模块。模块路径与源码路径大致对应：

nanobot.agent.loop
        ↓
nanobot/agent/loop.py

常见导入：

import asyncio
from dataclasses import dataclass, field
from nanobot.bus.events import InboundMessage

理解导入时先回答：

名字定义在哪个文件？
当前文件导入的是模块，还是模块中的类/函数？
是否为了避免循环依赖只在类型检查时导入？

源码中的：

from __future__ import annotations

会让类型标注延迟求值，使类可以更方便地引用尚未完成定义的类型。if TYPE_CHECKING: 中的导入主要提供给类型检查器，运行时通常不会执行。

5.1 `asyncio.run()` 是同步世界进入异步世界的入口

import asyncio
 
 
async def main() -> None:
    print("异步入口")
 
 
if __name__ == "__main__":
    asyncio.run(main())

通常一个进程只在最外层调用 asyncio.run()。已经处在事件循环中时，应直接 await coroutine，而不是再次调用 asyncio.run()。

6. 类、实例、继承和组合

6.1 类与实例

class Counter:
    def __init__(self, start: int = 0) -> None:
        self.value = start
 
    def increment(self) -> int:
        self.value += 1
        return self.value
 
 
counter = Counter(start=2)
assert counter.increment() == 3

Counter 是类；
counter 是实例；
self 表示当前实例；
self.value 是实例状态；
counter.increment() 会把实例自动传给 self。

6.2 继承与组合

继承表示“某类符合另一个类的契约”：

class Base:
    def label(self) -> str:
        return "base"
 
 
class Child(Base):
    def label(self) -> str:
        return "child"

组合表示“一个对象持有并使用另一个对象”：

class Service:
    def __init__(self, provider: object) -> None:
        self.provider = provider

nanobot 同时使用两者：

Tool、Provider、Channel 的具体实现继承抽象基类；
AgentRunner 通过组合持有一个 Provider；
AgentLoop 通过组合持有 MessageBus、ContextBuilder、SessionManager 和 ToolRegistry。

6.3 常见装饰器

写法	含义
`@property`	让一个方法可以像属性一样访问
`@classmethod`	定义一个绑定到类的方法。第一个参数是类 cls（接收到的是当前类对象，不是类的实例化），常用于工厂或类级配置
`@staticmethod`	定义一个放在类中的普通函数。函数在逻辑上属于这个类，但不需要访问实例或类本身。
`@abstractmethod`	子类必须实现的最低契约

from abc import ABC, abstractmethod
 
 
class ToolLike(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        ...
 
    @abstractmethod
    async def execute(self, **kwargs: object) -> object:
        ...

真实的 nanobot/agent/tools/base.py::Tool 还定义了 description、parameters、read_only、concurrency_safe 等属性。抽象基类只规定“实现必须提供什么”；具体行为要继续进入具体 Tool 类查看。

7. `dataclass`、`field`、`Enum` 与轻量数据对象

7.1 为什么使用 `dataclass`

数据类适合表达“一组有名字的字段”，自动生成初始化和显示等常用方法：

from dataclasses import dataclass, field
 
 
@dataclass
class Message:
    channel: str
    content: str
    metadata: dict[str, object] = field(default_factory=dict)

创建对象：

message = Message(channel="cli", content="你好")
assert message.metadata == {}

7.2 为什么可变默认值要用 `default_factory`

不能让所有实例共享同一个默认字典：

from dataclasses import dataclass, field
 
 
@dataclass
class SafeMessage:
    metadata: dict[str, object] = field(default_factory=dict)
 
 
left = SafeMessage()
right = SafeMessage()
left.metadata["trace"] = True
 
assert right.metadata == {}

default_factory=dict 会为每个实例新建字典。InboundMessage.media、metadata 和 OutboundMessage.buttons 都使用这一模式。

7.3 `@property` 可以提供计算字段

下面是 InboundMessage.session_key 的核心逻辑：

from dataclasses import dataclass
 
 
@dataclass
class InboundExample:
    channel: str
    chat_id: str
    session_key_override: str | None = None
 
    @property
    def session_key(self) -> str:
        return self.session_key_override or f"{self.channel}:{self.chat_id}"

调用者写 message.session_key，看起来像读取字段，实际上执行了方法。

7.4 `slots=True`、`frozen=True` 和 `dataclasses.replace`

@dataclass(slots=True)：限制实例属性布局，减少意外新增属性并节省部分内存；
@dataclass(frozen=True)：创建后禁止普通字段赋值，表达不可变配置；
dataclasses.replace(obj, field=value)：基于旧对象创建修改后的新对象。

nanobot 中：

AgentRunSpec 和 AgentRunResult 使用 slots=True；
GenerationSettings 使用 frozen=True；
AgentLoop 使用 dataclasses.replace() 调整不可直接原地改写的消息字段。

7.5 `Enum` 表达有限状态

from enum import Enum, auto
 
 
class State(Enum):
    RESTORE = auto()
    BUILD = auto()
    RUN = auto()
    DONE = auto()

它比散落的字符串更容易检查拼写和枚举所有合法状态。nanobot/agent/loop.py::TurnState 就用于表达一次 Turn 的状态机。

8. 类型标注：帮助阅读，不等于运行时校验

8.1 常见写法

标注	含义
`str`	字符串
`str	None`
`list[str]`	字符串列表
`dict[str, Any]`	字符串键、任意类型值的字典
`tuple[str, int]`	两个位置分别是字符串和整数的元组
`asyncio.Task[RunResult]`	最终产生 `RunResult` 的 Task
`Callable[[str], bool]`	接收字符串并返回布尔值的函数
`Callable[[str], Awaitable[None]]`	接收字符串并返回可等待对象的异步回调

Any 表示类型检查器暂不限制这个位置。它不是“任何输入都安全”，也不会替你做运行时校验。

8.2 标注通常不会自动拒绝错误值

def double(value: int) -> int:
    return value * 2
 
 
assert double(3) == 6

Python 默认不会因为标注了 int 就自动验证调用参数。Tool 参数的安全性来自 ToolRegistry 的转换与 schema 校验，而不是仅靠类型标注。

8.3 阅读复杂标注的方法

例如：

from collections.abc import Awaitable, Callable
 
Callback = Callable[[str], Awaitable[None]] | None

从最里面向外读：

输入是一个 str；
返回 Awaitable[None]；
整体是一个可调用对象；
也允许为 None。

不需要第一次就掌握所有泛型规则。先把“输入、输出、是否可选、是否需要 await”读出来。

9. 异常与同步上下文管理器

9.1 `try`、`except`、`else`、`finally`

def parse_limit(raw: str) -> int:
    try:
        value = int(raw)
    except ValueError as exc:
        raise ValueError("limit 必须是整数") from exc
    else:
        return value
    finally:
        # 无论成功还是失败都会执行。
        pass

try：可能失败的代码；
except：处理指定异常；
else：只有没有异常时执行；
finally：无论是否异常、是否提前返回都执行；
raise：主动抛出异常；
raise ... from exc：保留原始原因链。

不要用宽泛异常隐藏真实问题

except Exception: pass 会让调用方误以为操作成功。只有明确知道失败可以被忽略时，才应抑制异常，并留下足够日志或状态。

9.2 `with` 管理资源

from pathlib import Path
 
 
def read_text(path: Path) -> str:
    with path.open("r", encoding="utf-8") as handle:
        return handle.read()

with 会调用上下文管理器的进入和退出协议。即使读取期间抛出异常，文件仍会被关闭。

contextlib.suppress(SomeError) 可以明确忽略指定异常；nullcontext() 可以在“不需要真实资源管理器”时提供空操作上下文。它们都应在确实需要时使用，不能替代错误分析。

第二部分：asyncio 心智模型

10. asyncio 解决的是“等待期间还能做别的事”

asyncio

假如同一个 Agent 同时只能执行一个任务，但多个任务可以通过 await 在等待期间交替推进；前提是任务之间没有依赖，并且等待是可异步让出的。

Agent 经常等待：

LLM 网络响应；
Tool 的网络或子进程结果；
Queue 中的下一条消息；
Channel 收发；
MCP transport；
定时器和重试间隔。

如果每次等待都阻塞整个线程，Gateway 就无法及时处理其他 Session、流式片段或 /stop。asyncio 让多个 Task 在同一个事件循环中协作推进。

sequenceDiagram
    participant E as 事件循环
    participant A as Task A
    participant B as Task B

    E->>A: 运行 A
    A-->>E: await 网络
    E->>B: 运行 B
    B-->>E: await Queue
    E->>A: 网络就绪，恢复 A
    A-->>E: 完成
    E->>B: Queue 有消息，恢复 B

并发不等于并行

asyncio 的常见情况是在一个线程中交替推进多个 Task。某个 Task 在两次 await 之间执行 Python 代码时，其他 Task 通常不能插入。多核 CPU 并行是另一个问题。

11. 普通函数、coroutine function、coroutine object 和 Task

这四个概念不要混在一起。

11.1 普通函数立即执行

def add(left: int, right: int) -> int:
    return left + right
 
 
result = add(1, 2)
assert result == 3

11.2 `async def` 定义 coroutine function

async def fetch_name() -> str:
    return "nanobot"

调用它不会立刻得到字符串，而是得到 coroutine object：

async def inspect_coroutine() -> None:
    coroutine = fetch_name()
    result = await coroutine
    assert result == "nanobot"

如果创建 coroutine 后既不 await、也不把它包装成 Task，Python 通常会警告：

RuntimeWarning: coroutine '...' was never awaited

11.3 `await` 的准确含义

result = await get_data()

把 get_data() 交给事件循环执行，当前协程暂停，等它跑完拿到返回值，赋值给 result，然后才继续执行下面的代码。

await operation() 表示：

当前 Task 需要 operation 的最终结果；
operation 尚未完成时，当前 Task 可以暂停；
事件循环可以推进其他已就绪 Task；
operation 完成后，当前 Task 从这一行继续。

await 不会自动创建新线程，也不保证每次都切换任务；如果被等待对象已经完成，它可能立即返回。

12. 事件循环与协作式调度

事件循环反复做三件事：

找出已经就绪的 Task；
运行其中一个 Task，直到它完成、抛错或再次等待；
监听定时器、网络和其他异步事件，然后唤醒相应 Task。

因此 asyncio 是“协作式”的：Task 必须走到合适的 await，其他 Task 才有机会运行。

12.1 错误：在异步函数中阻塞事件循环

import time
 
 
async def bad_wait() -> None:
    time.sleep(1)  # 整个事件循环线程被阻塞。

12.2 正确：异步等待

import asyncio
 
 
async def good_wait() -> None:
    await asyncio.sleep(1)

12.3 同步阻塞库怎么办

短时间无法替换同步 API 时，可以把阻塞函数放到工作线程：

import asyncio
from pathlib import Path
 
 
async def read_bytes(path: Path) -> bytes:
    return await asyncio.to_thread(path.read_bytes)

nanobot 的部分 Channel、Provider 和 WebUI 路径会使用 asyncio.to_thread() 包装同步 SDK 或文件操作。

to_thread() 不是 CPU 并行的通用解法

它主要避免阻塞事件循环。长时间 CPU 密集计算还涉及 GIL、进程池或外部任务系统，本学习路径暂不展开。

13. Task：让 coroutine 独立参与调度

直接 await 表示“当前 Task 等它完成再继续”。asyncio.create_task() 则把 coroutine 注册成可独立推进的 Task：

import asyncio
 
 
async def work(name: str) -> str:
    await asyncio.sleep(0.01)
    return name
 
 
async def main() -> None:
    task = asyncio.create_task(work("A"))
    print("A 已经被调度，但可能尚未完成")
    result = await task
    assert result == "A"
 
 
if __name__ == "__main__":
    asyncio.run(main())

13.1 coroutine 与 Task 的区别

对象	是什么	会不会独立参与调度	如何取得结果
coroutine object	一次异步调用的待执行对象	否	`await coroutine`
Task	被事件循环管理的 coroutine	是	`await task` 或检查 Task 状态
普通结果	coroutine 完成后的业务值	不适用	直接使用

13.2 创建 Task 后要保留所有权

“创建后不管”会带来三个问题：

异常可能只以“Task exception was never retrieved”形式出现；
关闭时不知道应该取消或等待哪些任务；
Task 可能在资源清理前仍在运行。

nanobot 会保存活跃 Task 或后台 Task，并在完成时移除，在关闭时等待或取消相应任务。

AgentLoop.run() 的核心模式可以压缩为：

import asyncio
 
 
class LoopExample:
    def __init__(self) -> None:
        self._tasks: list[asyncio.Task[None]] = []
 
    async def _dispatch(self, message: str) -> None:
        await asyncio.sleep(0)
 
    async def accept(self, message: str) -> None:
        task = asyncio.create_task(self._dispatch(message))
        self._tasks.append(task)
        task.add_done_callback(self._tasks.remove)

真实实现还按有效 Session Key 保存 Task，以便处理同 Session 的注入和停止。

13.3 Task 状态

一个 Task 大致经历：

stateDiagram-v2
    [*] --> Pending
    Pending --> Running: 事件循环调度
    Running --> Waiting: await 未完成对象
    Waiting --> Running: 等待对象就绪
    Running --> Done: 正常返回
    Running --> Failed: 抛出异常
    Pending --> Cancelling: cancel()
    Waiting --> Cancelling: cancel()
    Cancelling --> Cancelled: 取消传播完成

调试时常见方法：

task.done()：是否结束；
task.cancelled()：是否以取消结束；
task.exception()：结束后取得异常；
task.result()：结束后取得结果，失败时会重新抛出异常。

14. 顺序等待与并发等待

14.1 顺序等待

import asyncio
 
 
async def one(name: str) -> str:
    await asyncio.sleep(0.01)
    return name
 
 
async def sequential() -> list[str]:
    first = await one("A")
    second = await one("B")
    return [first, second]

只有 A 完成后才开始等待 B。

14.2 `asyncio.gather()` 并发等待

import asyncio
 
 
async def one(name: str) -> str:
    await asyncio.sleep(0.01)
    return name
 
 
async def concurrent() -> list[str]:
    results = await asyncio.gather(one("A"), one("B"))
    return list(results)

两个 coroutine 都会参与调度；结果列表仍按传入顺序排列，不按完成顺序排列。

当前 AgentRunner._execute_tools() 只有在：

spec.concurrent_tools 为真；
当前批次包含多个 Tool；
Tool 的并发属性允许它们进入同一批次；

才使用 asyncio.gather()。否则按顺序执行。不能因为语法上可以 gather，就忽略 Tool 的副作用和互斥要求。

并发 Tool 不是越多越好

两个 Tool 如果会修改同一文件、共享进程状态或依赖执行顺序，就不应仅为了速度并发执行。

15. Queue：在生产者和消费者之间交接对象

15.1 最小生产者—消费者

import asyncio
 
 
async def producer(queue: asyncio.Queue[str]) -> None:
    await queue.put("message")
 
 
async def consumer(queue: asyncio.Queue[str]) -> str:
    message = await queue.get()
    return message
 
 
async def main() -> None:
    queue: asyncio.Queue[str] = asyncio.Queue()
    produced, consumed = await asyncio.gather(
        producer(queue),
        consumer(queue),
    )
    assert produced is None
    assert consumed == "message"
 
 
if __name__ == "__main__":
    asyncio.run(main())

queue.get() 在没有对象时暂停当前 Task，而不是不断循环占用 CPU。

15.2 nanobot 的 MessageBus

nanobot/bus/queue.py::MessageBus 的核心非常直接：

import asyncio
from typing import Any
 
 
class MessageBusExample:
    def __init__(self) -> None:
        self.inbound: asyncio.Queue[Any] = asyncio.Queue()
        self.outbound: asyncio.Queue[Any] = asyncio.Queue()
 
    async def publish_inbound(self, message: Any) -> None:
        await self.inbound.put(message)
 
    async def consume_inbound(self) -> Any:
        return await self.inbound.get()

inbound：Channel 或其他入口交给 Agent；
outbound：Agent 交给 Channel；
Queue 保存的是等待交接的对象，不是完整 Session 历史。

15.3 无界 Queue、有限 Queue 和背压

asyncio.Queue() 默认没有显式容量上限。asyncio.Queue(maxsize=20) 最多保存 20 项：

await queue.put(item)：满时等待；
queue.put_nowait(item)：满时立刻抛出 asyncio.QueueFull；
queue.get_nowait()：空时立刻抛出 asyncio.QueueEmpty。

nanobot 的主 MessageBus Queue 默认无界；同一活跃 Session 的中途注入 Queue 当前使用 maxsize=20，并显式处理 QueueFull。

qsize() 只是一刻的观察值

另一个 Task 可能立刻 put 或 get，不能用 qsize() 代替正确的同步协议。

16. Lock：保护跨 `await` 的临界区

即使只有一个事件循环线程，也可能发生竞态。原因是一个 Task 读完状态后在 await 处暂停，另一个 Task 随后修改了相同状态。

import asyncio
 
 
counter = 0
 
 
async def unsafe_increment() -> None:
    global counter
    current = counter
    await asyncio.sleep(0)
    counter = current + 1

两个 Task 同时执行时，都可能读到旧值，最后只增加一次。

16.1 使用 Lock

import asyncio
 
 
counter = 0
lock = asyncio.Lock()
 
 
async def safe_increment() -> None:
    global counter
    async with lock:
        current = counter
        await asyncio.sleep(0)
        counter = current + 1

只有取得同一把 Lock 的 Task 才能进入临界区。

16.2 nanobot 为什么按 Session 加锁

同一 Session 的两个 Turn 如果同时：

读取同一份历史；
构造各自上下文；
调用模型或 Tool；
保存新消息；

就可能产生顺序错乱或覆盖。AgentLoop._dispatch() 因此按有效 Session Key 取得 Lock：

import asyncio
 
 
class SessionSerialExample:
    def __init__(self) -> None:
        self._locks: dict[str, asyncio.Lock] = {}
 
    async def dispatch(self, session_key: str) -> None:
        lock = self._locks.setdefault(session_key, asyncio.Lock())
        async with lock:
            await self.process_turn(session_key)
 
    async def process_turn(self, session_key: str) -> None:
        await asyncio.sleep(0)

结果是：

同一个有效 Session Key：串行
不同 Session Key：允许并发
同一 Turn 中的后续消息：可进入该 Session 的 pending Queue

Lock 不代表整个程序单线程

它只约束使用同一把 Lock 的代码。不同 Session、流式消费者和部分 Tool 仍可以继续推进。

16.3 Lock 的边界比“有没有 Lock”更重要

阅读时要确认：

Lock 的 key 是什么；
从哪一行开始持有；
哪一行释放；
临界区中有哪些 await；
是否有代码绕过了这把 Lock。

nanobot 有意让 Session lock 覆盖完整 Turn，因为历史读取、执行和保存需要保持顺序。这与“尽量缩短普通锁的临界区”并不矛盾：这里保护的业务事务本身就是一个 Turn。

17. Semaphore：限制总并发量

Lock 解决“同一份状态能否同时修改”，Semaphore 解决“最多允许多少项工作同时进入”。

import asyncio
 
 
gate = asyncio.Semaphore(3)
 
 
async def limited_request() -> None:
    async with gate:
        await asyncio.sleep(0.01)

最多三个 Task 同时通过这道 gate。

当前 AgentLoop：

读取 NANOBOT_MAX_CONCURRENT_REQUESTS；
默认值是 3；
值小于等于 0 时不启用全局 Semaphore；
_dispatch() 同时进入 Session lock 和全局 gate。

因此实际并发约束可以理解为：

先满足同 Session 串行
再满足全局请求上限
两项都满足后才处理 Turn

17.1 Queue、Lock、Semaphore 对照

原语	核心问题	nanobot 示例
Queue	对象交给谁、没有对象时怎么等待	MessageBus、Session pending Queue
Lock	同一临界区是否只能有一个 Task	每 Session Turn 串行
Semaphore	同类工作最多同时有多少个	全局请求并发门限

不要用 Queue 模拟互斥，也不要用 Lock 充当消息历史。

18. Event、Future 和 Condition：第一次认识即可

这些不是阅读核心 Turn 的第一优先级，但会在外围源码中出现。

18.1 Event：等待“某件事已经发生”

import asyncio
 
 
async def wait_for_shutdown(event: asyncio.Event) -> None:
    await event.wait()

其他 Task 调用 event.set() 后，所有等待者都会被唤醒。CLI 和部分 Channel 使用 Event 表达关闭或完成信号。

18.2 Future：未来会被填入一个结果

import asyncio
 
 
async def future_example() -> str:
    loop = asyncio.get_running_loop()
    future: asyncio.Future[str] = loop.create_future()
    future.set_result("done")
    return await future

Task 是 Future 的一种高级形式，但 Future 可以由外部代码手动填入结果。CronTurnCoordinator 使用 Future 把某次 Cron 请求与稍后产生的 Session 响应关联起来。

18.3 Condition：状态条件与互斥的组合

Condition 适合“持锁检查状态，不满足就等待，状态改变后通知”。nanobot 的 SDK runtime gate 用它协调普通运行和独占模型覆盖。第一次阅读只需知道：

async with condition 取得相关 Lock；
await condition.wait() 释放 Lock 并等待通知；
condition.notify_all() 唤醒等待者重新检查条件。

19. 超时：等待太久时改变控制流

19.1 `asyncio.wait_for`

import asyncio
 
 
async def request() -> str:
    await asyncio.sleep(0.01)
    return "ok"
 
 
async def request_with_timeout() -> str:
    try:
        return await asyncio.wait_for(request(), timeout=1.0)
    except asyncio.TimeoutError:
        return "timeout"

到达期限后，wait_for 通常会向内部 awaitable 发出取消，并由调用方处理 TimeoutError。

19.2 同样是超时，语义可能完全不同

AgentLoop.run() 用一秒超时等待 inbound Queue。这里的超时不是故障，而是让循环定期醒来检查自动压缩等维护工作：

一秒内有消息 → 处理消息
一秒内没消息 → 检查维护工作，然后继续等待

AgentRunner._request_model() 的超时则代表模型请求超过策略允许的等待时间。

19.3 总时长超时与流空闲超时

当前 Runner 区分：

非流式请求：可以使用外层 LLM wall-clock timeout；
流式请求：Provider 已有 stream idle timeout，不再叠加同一个外层总时长超时。

原因是健康的长推理流可能持续输出，虽然总耗时较长，却没有“卡住”。调试超时时必须先确认是哪一层、测量的是总时长还是连续无数据时长。

20. 取消：请求 Task 尽快停止

20.1 `task.cancel()` 不是强制杀线程

import asyncio
 
 
async def worker() -> None:
    try:
        while True:
            await asyncio.sleep(1)
    finally:
        print("释放资源")
 
 
async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(0)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("已确认取消")
 
 
if __name__ == "__main__":
    asyncio.run(main())

cancel() 提交取消请求。Task 通常在下一个可取消的等待点收到 CancelledError，先执行 finally，然后以取消状态结束。

20.2 `CancelledError` 不应被当作普通业务错误吞掉

在 Python 3.11 中，asyncio.CancelledError 继承自 BaseException。正确的资源边界通常是：

import asyncio
 
 
async def call_provider() -> str:
    await asyncio.sleep(0)
    return "ok"
 
 
async def safe_call() -> str:
    try:
        return await call_provider()
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        return f"普通调用错误：{exc}"

nanobot/providers/base.py::_safe_chat 和 _safe_chat_stream 就显式让取消继续传播，同时把普通 Provider 异常归一为错误响应。

20.3 为什么一定要等待已取消 Task

调用 task.cancel() 后再 await task，可以：

等待 finally 完成；
确认 Task 确实结束；
取得或确认取消异常；
避免资源清理仍在后台运行。

20.4 “停止 Agent 工作”不一定等于取消 Task

系统可能有多种停止语义：

取消某个 asyncio Task；
设置循环的 _running = False；
让 Runner 返回特定 stop_reason；
结束 Goal；
关闭流，但保留已经完成的 Session 状态。

阅读 /stop 或关闭逻辑时，不要看到“stop”就假设它一定调用 Task.cancel()。

21. `async with`：异步进入和退出资源

普通 with 对应 __enter__ / __exit__；async with 对应可以等待的 __aenter__ / __aexit__。

import asyncio
 
 
class AsyncResource:
    async def __aenter__(self) -> "AsyncResource":
        await asyncio.sleep(0)
        return self
 
    async def __aexit__(
        self,
        exc_type: object,
        exc: object,
        traceback: object,
    ) -> None:
        await asyncio.sleep(0)
 
 
async def use_resource() -> None:
    async with AsyncResource():
        print("资源可用")

nanobot 使用 async with 管理：

Session lock；
全局并发 gate；
SDK runtime gate；
HTTP 或流式资源；
某些 MCP transport。

21.1 `@asynccontextmanager`

当“进入—使用—退出”更适合写成一个函数时，可以使用：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
 
 
@asynccontextmanager
async def temporary_value() -> AsyncIterator[str]:
    value = "active"
    try:
        yield value
    finally:
        value = "restored"
        assert value == "restored"

yield 之前是进入逻辑，之后是退出逻辑。SDKRuntimeController.override() 使用这一模式，确保临时模型覆盖最终被恢复。

22. `async for` 与异步生成器

普通 for 读取已经可以同步取得的下一项。async for 允许“等待下一项”：

import asyncio
from collections.abc import AsyncIterator
 
 
async def chunks() -> AsyncIterator[str]:
    for chunk in ("A", "B", "C"):
        await asyncio.sleep(0)
        yield chunk
 
 
async def collect() -> str:
    result = ""
    async for chunk in chunks():
        result += chunk
    return result

异步生成器同时包含 async def 和 yield。每次产生一项后暂停，调用方稍后继续请求下一项。

nanobot/sdk/streaming.py::RunStream.stream_events() 的核心模式是：

from collections.abc import AsyncIterator
from typing import Any
 
 
async def stream_from_queue(queue: Any, sentinel: object) -> AsyncIterator[Any]:
    while True:
        item = await queue.get()
        if item is sentinel:
            break
        yield item

这把 Queue 中陆续到达的事件转换为调用者可以 async for 消费的流。真实实现还负责单消费者约束、提前关闭和取消底层 Turn。

23. 异步回调：核心产生事件，外围决定如何处理

同步回调：

def on_value(value: str) -> None:
    print(value)

异步回调：

async def on_delta(delta: str) -> None:
    await send_to_client(delta)
 
 
async def send_to_client(delta: str) -> None:
    _ = delta

调用者必须 await on_delta(delta)，因为回调本身可能需要等待 WebSocket、HTTP 或 Channel。

nanobot 的 AgentRunSpec、TurnContext 和 Provider 方法中有多种回调：

progress callback；
stream callback；
stream end callback；
retry wait callback；
checkpoint callback；
injection callback。

回调的设计目的不是“把所有逻辑塞进函数参数”，而是让核心 Runner 只产生通用事件，由 CLI、SDK、API 或 WebUI 决定如何展示和传输。

第三部分：用这些知识跟踪 nanobot

24. 一条消息中的异步边界

下面只标出调度和同步关系，不展开 Agent 业务细节：

sequenceDiagram
    participant C as Channel/入口
    participant Q as MessageBus
    participant L as AgentLoop.run
    participant T as Turn Task
    participant P as Provider
    participant R as Tool/Runner

    C->>Q: await publish_inbound(msg)
    L->>Q: await consume_inbound()
    Q-->>L: InboundMessage
    L->>T: asyncio.create_task(_dispatch)
    Note over T: 等待 Session lock 与全局 gate
    T->>T: async with lock, gate
    T->>P: await 模型请求
    P-->>T: LLMResponse
    opt 多个并发安全 Tool
        T->>R: await asyncio.gather(...)
        R-->>T: Tool results
    end
    T->>Q: await publish_outbound(msg)

逐步解释：

Channel 把 InboundMessage 放入 inbound Queue；
AgentLoop.run() 等待并取出消息；
Loop 创建 _dispatch(msg) Task，因此能继续消费其他 Session 的消息；
Turn Task 取得当前 Session 的 Lock 和全局 Semaphore；
同一 Session 的其他 Turn 不能竞争这段生命周期；
Task 在 Provider、Tool、Queue 或持久化 I/O 处等待；
Runner 只在策略与 Tool 并发属性允许时并发执行一批 Tool；
最终响应放入 outbound Queue，或通过 direct/streaming 回调返回。

25. 阅读异步函数的固定步骤

每遇到一个 async def，按下面顺序检查：

谁调用它？ 是 await，还是 create_task()？
谁拥有 Task？ 是否保存引用，关闭时是否等待？
第一个 await 在哪里？ 在它之前的代码会连续执行。
等待什么？ Queue、Lock、网络、子进程、另一个 Task，还是定时器？
共享状态是什么？ 列表、字典、Session、文件还是 Provider 状态？
谁保护共享状态？ Lock、Semaphore、Condition，还是业务上保证单调用者？
超时在哪一层？ 总时长、空闲时间还是 Queue 轮询？
取消如何传播？ 是否显式重新抛出 CancelledError？
资源在哪里清理？ finally、async with、aclose() 还是关闭函数？
正常结果和异常结果如何返回？ 抛异常、返回错误对象，还是设置 stop_reason？

26. 调试时值得观察的对象

只做理解学习时，可以在断点或调试控制台中观察，不必修改源码：

位置	建议观察
`AgentLoop.run()`	`msg`、`effective_key`、新建的 `task`
`AgentLoop._dispatch()`	`session_key`、`lock.locked()`、`gate`、`pending`
AgentLoop 实例	`_active_tasks`、`_pending_queues`、`_session_locks`
`MessageBus`	`inbound_size`、`outbound_size`
`AgentRunner.run()`	`spec`、当前迭代、`messages`、`stop_reason`
`_request_model()`	是否流式、`timeout_s`、`outer_timeout_s`、`response`
`_execute_tools()`	`batches`、`spec.concurrent_tools`、批次结果
取消路径	`asyncio.current_task()`、Task 的 `cancelling()` 和 `cancelled()`

不要把调试器暂停误判为程序死锁

在一个 Task 的断点处暂停整个进程时，事件循环也可能停止推进，其他 Task 自然无法完成它们正在等待的工作。

27. 常见报错与定位方向

现象	常见原因	首先检查
`coroutine was never awaited`	调用了 `async def`，但没有 await 或创建 Task	搜索 coroutine 的调用点
`asyncio.run() cannot be called from a running event loop`	在已有事件循环中再次调用 `asyncio.run()`	改为在当前 async 函数中 `await`
整个 Gateway 短暂无响应	异步路径调用了阻塞同步函数	搜索 `time.sleep`、同步网络和大文件操作
`Task exception was never retrieved`	后台 Task 无所有者，没有等待结果或读取异常	找到 `create_task()`，检查保存和清理
同一 Session 历史顺序异常	绕过 Session lock，或使用了不同有效 Session Key	检查 key 计算和 `_dispatch()` 路径
请求在关闭后仍运行	只设置标志，没有取消或等待子 Task	检查 shutdown 和 `finally`
关闭流程卡住	Task 不响应取消，或清理顺序不正确	找到下一处 `await` 和资源拥有者
流式请求被误判超时	混淆总时长超时与流空闲超时	检查 Runner 和 Provider 两层设置
消息内容意外改变	多个变量共享同一可变列表或字典	比较对象身份和复制方式
两个 Tool 互相干扰	对有副作用 Tool 启用了并发	检查 Tool 的 `read_only`、`concurrency_safe`、`exclusive`

28. 三个最小练习

练习一：观察顺序等待与并发等待

import asyncio
from time import perf_counter
 
 
async def work(delay: float) -> float:
    await asyncio.sleep(delay)
    return delay
 
 
async def main() -> None:
    started = perf_counter()
    sequential = [await work(0.05), await work(0.05)]
    sequential_time = perf_counter() - started
 
    started = perf_counter()
    concurrent = await asyncio.gather(work(0.05), work(0.05))
    concurrent_time = perf_counter() - started
 
    assert sequential == [0.05, 0.05]
    assert list(concurrent) == [0.05, 0.05]
    print(f"顺序约 {sequential_time:.3f}s")
    print(f"并发约 {concurrent_time:.3f}s")
 
 
if __name__ == "__main__":
    asyncio.run(main())

你应观察到顺序版本大约需要两段等待时间，并发版本大约需要一段。实际数值会受机器调度影响。

练习二：验证“跨 Session 并发、同 Session 串行”

import asyncio
from collections import defaultdict
 
 
locks: defaultdict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
active_sessions: set[str] = set()
 
 
async def turn(session: str, name: str) -> None:
    async with locks[session]:
        assert session not in active_sessions
        active_sessions.add(session)
        print("start", session, name)
        await asyncio.sleep(0.02)
        print("end  ", session, name)
        active_sessions.remove(session)
 
 
async def main() -> None:
    await asyncio.gather(
        turn("session-a", "A1"),
        turn("session-a", "A2"),
        turn("session-b", "B1"),
    )
 
 
if __name__ == "__main__":
    asyncio.run(main())

观察重点：

A1 和 A2 的临界区不会重叠；
B1 可以与其中一个 A Turn 交错；
输出的具体先后顺序不应写死为测试条件。

练习三：确认取消会执行清理

import asyncio
 
 
async def worker(cleaned: list[bool]) -> None:
    try:
        await asyncio.Event().wait()
    finally:
        cleaned.append(True)
 
 
async def main() -> None:
    cleaned: list[bool] = []
    task = asyncio.create_task(worker(cleaned))
    await asyncio.sleep(0)
 
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
 
    assert cleaned == [True]
    assert task.cancelled()
 
 
if __name__ == "__main__":
    asyncio.run(main())

如果去掉 await task，清理可能还没有在你检查状态时完成。这正是“取消后仍要等待 Task”的原因。

29. 自检题

先自己回答，再展开参考答案。

为什么调用 async def 不会立即得到业务结果？
await 是否会自动创建线程？
await first(); await second() 与 await asyncio.gather(first(), second()) 有什么区别？
为什么一个线程中的 asyncio 程序仍可能发生竞态？
MessageBus 为什么使用 Queue，而 Session 历史不使用 Queue？
为什么 nanobot 既允许跨 Session 并发，又给每个 Session 加 Lock？
Lock 与 Semaphore 的职责有什么不同？
为什么创建后台 Task 后还要保存它？
为什么不能把 CancelledError 一律转换成普通错误响应？
为什么流式请求可能使用 idle timeout，而不使用相同的总时长 timeout？
Callable[[str], Awaitable[None]] | None 表示什么？
field(default_factory=dict) 解决了什么问题？

参考答案

调用 async def 只创建 coroutine object；必须 await、创建 Task 或交给事件循环后才推进。

不会。await 允许当前 Task 暂停并让事件循环推进其他 Task。

前者通常顺序执行；后者把两个 awaitable 一起调度并等待全部完成。

Task 可以在读状态后于 await 暂停，另一个 Task 修改同一状态，恢复后便可能基于旧值写回。

Queue 解决生产者和消费者的异步交接；Session 历史是需要查询、回放和持久化的数据集合。

不同 Session 没有同一份历史顺序约束，可以并发；同一 Session 的读取、执行和保存必须保持顺序。

Lock 保证同一临界区一次只有一个 Task；Semaphore 限制同类工作同时进入的总数量。

为了取得异常、支持停止、在关闭时等待清理，并明确生命周期所有者。

取消通常代表上层要求停止；吞掉它会导致关闭失效或工作在用户要求停止后继续。

健康流可能总耗时很长但持续有数据；idle timeout 检测的是“是否卡住”。

一个可选异步回调：接收字符串，返回等待完成后没有业务值的 awaitable。

为每个 dataclass 实例创建独立字典，避免多个实例共享同一个可变默认对象。

30. 什么时候算掌握到“可以继续”

不需要会手写事件循环，也不需要背 asyncio 全部 API。满足下面条件即可进入 nanobot 核心路线：

能解释 coroutine、Task、结果三者的区别；
能指出一段异步代码中的所有 await 和 Task 创建点；
能说明 Queue、Lock、Semaphore 各自解决的问题；
能解释同 Session 串行、跨 Session 并发；
能识别阻塞同步调用；
能解释超时、取消和 finally 的关系；
能从类型标注读出函数输入、输出和是否需要 await；
能按照“调用者—等待点—共享状态—同步原语—清理位置”跟踪一个异步函数。

如果某一项仍模糊，优先重做对应的最小练习，不必继续扩展到 asyncio 的底层实现。

下一步

阅读 LLM 消息、流式输出与 Tool Calling，建立模型请求与 Tool Call 心智模型。
阅读源码阅读与调试最低要求，学习搜索调用方和设置断点。
进入一次 Turn 学习导航，把本章的 Task、Queue、Lock、超时和取消放回真实调用链。

AI Agent Engineer

Python 与异步编程最低要求

Python 与异步编程最低要求

学习目标

如何使用本章

本章与 nanobot 源码的关系

第一部分：阅读 nanobot 所需的 Python 基础

1. 名字指向对象，不是把对象复制一份

1.1 可变对象与不可变对象

1.2 None 表示“当前没有值”

2. 列表、字典、元组和集合

2.1 列表 list

2.2 字典 dict

2.3 元组 tuple

2.4 集合 set

2.5 推导式

3. 条件、循环和提前返回

3.1 条件分支

3.2 真值判断

3.3 for、while、break 和 continue

4. 函数、参数和返回值

4.1 *args 和 **kwargs

4.2 函数也是对象

4.3 闭包和 lambda

5. 模块、导入和源码路径

5.1 asyncio.run() 是同步世界进入异步世界的入口

6. 类、实例、继承和组合

6.1 类与实例

6.2 继承与组合

6.3 常见装饰器

7. dataclass、field、Enum 与轻量数据对象

7.1 为什么使用 dataclass

7.2 为什么可变默认值要用 default_factory

7.3 @property 可以提供计算字段

7.4 slots=True、frozen=True 和 dataclasses.replace

7.5 Enum 表达有限状态

8. 类型标注：帮助阅读，不等于运行时校验

8.1 常见写法

8.2 标注通常不会自动拒绝错误值

8.3 阅读复杂标注的方法

9. 异常与同步上下文管理器

9.1 try、except、else、finally

9.2 with 管理资源

第二部分：asyncio 心智模型

10. asyncio 解决的是“等待期间还能做别的事”

11. 普通函数、coroutine function、coroutine object 和 Task

11.1 普通函数立即执行

11.2 async def 定义 coroutine function

11.3 await 的准确含义

12. 事件循环与协作式调度

12.1 错误：在异步函数中阻塞事件循环

12.2 正确：异步等待

12.3 同步阻塞库怎么办

13. Task：让 coroutine 独立参与调度

13.1 coroutine 与 Task 的区别

13.2 创建 Task 后要保留所有权

13.3 Task 状态

14. 顺序等待与并发等待

14.1 顺序等待

14.2 asyncio.gather() 并发等待

15. Queue：在生产者和消费者之间交接对象

15.1 最小生产者—消费者

15.2 nanobot 的 MessageBus

15.3 无界 Queue、有限 Queue 和背压

16. Lock：保护跨 await 的临界区

16.1 使用 Lock

16.2 nanobot 为什么按 Session 加锁

16.3 Lock 的边界比“有没有 Lock”更重要

17. Semaphore：限制总并发量

17.1 Queue、Lock、Semaphore 对照

18. Event、Future 和 Condition：第一次认识即可

18.1 Event：等待“某件事已经发生”

18.2 Future：未来会被填入一个结果

18.3 Condition：状态条件与互斥的组合

19. 超时：等待太久时改变控制流

19.1 asyncio.wait_for

19.2 同样是超时，语义可能完全不同

19.3 总时长超时与流空闲超时

20. 取消：请求 Task 尽快停止

20.1 task.cancel() 不是强制杀线程

1.2 `None` 表示“当前没有值”

2.1 列表 `list`

2.2 字典 `dict`

2.3 元组 `tuple`

2.4 集合 `set`

3.3 `for`、`while`、`break` 和 `continue`

4.1 `*args` 和 `**kwargs`

4.3 闭包和 `lambda`

5.1 `asyncio.run()` 是同步世界进入异步世界的入口

7. `dataclass`、`field`、`Enum` 与轻量数据对象

7.1 为什么使用 `dataclass`

7.2 为什么可变默认值要用 `default_factory`

7.3 `@property` 可以提供计算字段

7.4 `slots=True`、`frozen=True` 和 `dataclasses.replace`

7.5 `Enum` 表达有限状态

9.1 `try`、`except`、`else`、`finally`

9.2 `with` 管理资源

11.2 `async def` 定义 coroutine function

11.3 `await` 的准确含义

14.2 `asyncio.gather()` 并发等待

16. Lock：保护跨 `await` 的临界区

19.1 `asyncio.wait_for`

20.1 `task.cancel()` 不是强制杀线程

20.2 `CancelledError` 不应被当作普通业务错误吞掉

21. `async with`：异步进入和退出资源

21.1 `@asynccontextmanager`

22. `async for` 与异步生成器