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 真值判断
以下值在条件中通常视为假:
FalseNone- 数字
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 Nonecontinue:跳到下一次循环;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() == 3Counter是类;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 = providernanobot 同时使用两者:
- 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) == 6Python 默认不会因为标注了 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:
# 无论成功还是失败都会执行。
passtry:可能失败的代码;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 == 311.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 awaited11.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 QueueLock 不代表整个程序单线程
它只约束使用同一把 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 串行
再满足全局请求上限
两项都满足后才处理 Turn17.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 futureTask 是 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、超时和取消放回真实调用链。