AI Agent 全栈
模块 00 · 基础功底 · 第 0.3 课

0.3 Python（够用即可）：读懂 ML 生态的那一半

第 0 部分 · 第 3 课 ｜ 预计 3 小时 ｜ 前置：学完 0.1 Go、0.2 TypeScript，Python 3.12+

目标不是成为 Python 专家，而是读得懂、改得动：模型 SDK、评测脚本、LangGraph、vLLM、训练代码，几乎都是 Python。本课每个语法都对照 Go 和 TS 讲，每段代码后面有逐行解读。

为什么学这个

你的主力是 Go / TS，但 Agent 生态里这些东西基本只有 Python 版或 Python 版最好：

评测：SWE-bench、τ-bench、Inspect、各种 LLM-as-Judge 脚本
框架：LangGraph、OpenAI Agents SDK、Claude Agent SDK 的 Python 版最先更新
模型侧：vLLM、SGLang、transformers、TRL（DPO / GRPO）

本课只学写 Agent 用得到的三件事，和前两课一一对应：

问题 Go（0.1） TS（0.2） Python（本课）
项目与依赖 go mod npm uv
校验外部 JSON json.Unmarshal + 手写校验 Zod Pydantic
并发模型 goroutine（多线程真并行） 事件循环（单线程） asyncio（单线程事件循环，和 JS 一样）
并行工具 errgroup Promise.all + worker asyncio.TaskGroup + Semaphore
超时 / 取消 context.WithTimeout AbortSignal.timeout asyncio.timeout + 任务取消
流式读取 goroutine + channel AsyncGenerator async generator（几乎和 TS 一样）

一句话：Python 的 asyncio 心智模型和 TS 几乎一样，你学 0.2 时建立的直觉可以直接搬过来。

0. Go / TS 程序员的 Python 语法速成

0.1 缩进就是代码块

def add(a: int, b: int) -> int:
    if a > 0:
        return a + b
    return b

关键区别：Python 的类型注解运行时不检查。 add("x", "y") 照样能跑（结果是 "xy"）。类型注解只给编辑器和类型检查工具（mypy / pyright）看，这一点和 TS 一样：类型只在“编译期”有用，运行时要靠 Pydantic 校验。

0.2 变量和基本类型

name = "search"          # 没有 const / let / var，直接赋值
limit: int = 4           # 可以加类型注解
MAX_RETRY = 3            # 全大写只是约定“这是常量”，语言不阻止你改
Python Go TS
str string string
int / float int / float64 number
bool（True / False，首字母大写） bool boolean
None nil null / undefined
list[str] []string string[]
dict[str, int] map[string]int Record<string, number>
tuple[str, int] 多返回值 (string, int) [string, number]
str \| None *string string \| undefined
Any any any 
nums = [1, 2, 3]                  # list
nums.append(4)                    # Go: nums = append(nums, 4)
first = nums[0]                   # 越界会抛 IndexError（类似 Go 的 panic）
last = nums[-1]                   # 负下标从末尾数，Go/TS 都没有
part = nums[1:3]                  # 切片 [2, 3]，和 Go 一样左闭右开

cfg = {"model": "gpt", "temp": 0.7}   # dict，= Go map / TS 对象
cfg["temp"]                       # 0.7；键不存在会抛 KeyError
cfg.get("top_p")                  # 键不存在返回 None，不抛错
cfg.get("top_p", 1.0)             # 带默认值
"model" in cfg                    # True，= Go 的 _, ok := cfg["model"]

0.3 字符串

name = "search"
msg = f"未知工具 {name}"           # f-string，= TS 模板字符串 `未知工具 ${name}`
line = "  data: hi \n".strip()    # 去首尾空白，= strings.TrimSpace / trim()
line.startswith("data:")          # = strings.HasPrefix / startsWith
line[5:]                          # 去掉前 5 个字符，= line[5:] (Go) / slice(5) (TS)

0.4 函数：默认参数、关键字参数、多返回值

def run(calls: list[str], limit: int = 4, timeout_s: float = 20.0) -> list[str]:
    ...

run(["a"])                         # limit=4, timeout_s=20.0
run(["a"], timeout_s=0.2)          # 关键字参数：按名字传，跳过 limit

多返回值用 tuple：

def parse(raw: str) -> tuple[dict | None, str | None]:
    if not raw:
        return None, "empty"       # 相当于 Go 的 return nil, errors.New("empty")
    return {"ok": True}, None

data, err = parse("x")             # 解包，= Go 的 data, err := parse("x")
if err is not None:                # 判断 None 用 is / is not，不要用 ==
    ...

0.5 异常：try / except / finally / raise

try:
    n = int("abc")                 # 抛 ValueError
except ValueError as e:            # = TS 的 catch (e)，但可以按类型分别捕获
    print("不是数字：", e)
except (KeyError, IndexError):     # 一次捕获多种
    print("下标或键错误")
finally:
    print("总会执行")               # = defer / finally

raise RuntimeError("出错了")        # = TS 的 throw new Error(...)

和 TS 的区别：except 可以按异常类型分别处理，不需要在 catch 里 if (err instanceof ...)。这一点更像 Go 的 errors.As。

0.6 with：自动清理（Python 版 defer）

with open("a.txt") as f:           # 进入时打开文件
    text = f.read()
# 离开 with 块时自动 f.close()，不管是正常结束还是抛异常

对照 Go：

f, err := os.Open("a.txt")
if err != nil { return err }
defer f.Close()

with 后面的对象叫上下文管理器（context manager，和 Go 的 context 不是一个东西）。HTTP 客户端、锁、超时、文件……凡是“用完要清理”的，Python 都用 with。异步版本是 async with，后面大量出现。

0.7 类和 dataclass

from dataclasses import dataclass

@dataclass
class ToolCall:
    id: str
    name: str
    args: str

c = ToolCall("1", "search", "{}")         # 按顺序传
c = ToolCall(id="1", name="search", args="{}")  # 或按名字传
print(c.name)                              # "search"
print(c)                                   # ToolCall(id='1', name='search', args='{}')

@xxx 这种写法统称装饰器，就是“把下面的函数/类交给 xxx 加工一下”。看到 @app.get("/")、@pytest.fixture 都是同一回事。

0.8 列表推导式

results = [r.is_error for r in res]              # = res.map(r => r.isError)
ok = [r for r in res if not r.is_error]          # = res.filter(r => !r.isError)

Go 里你会写一个 for 循环 append。读 Python 代码时这个写法到处都是，看懂就行。

0.9 模块和包

import asyncio                         # 导入整个模块，用 asyncio.sleep(...)
from pydantic import BaseModel, Field  # 只导入几个名字
from py_agent_core.tools import run_tools  # 导入自己项目里的模块（目录 = 包）

1. uv：Python 的 go mod

Python 的老问题是“环境地狱”：系统 Python、pip、venv、requirements.txt、poetry……uv 是 2024 年后的事实标准，一个工具全包，而且很快。

uv init --lib my-agent        # 新建项目，= go mod init / npm init
cd my-agent
uv add pydantic httpx         # 加依赖，= go get / npm i
uv add --dev pytest pytest-asyncio   # 开发依赖，= npm i -D
uv run pytest                 # 在项目的虚拟环境里运行命令，= npx
uv run python -m my_agent     # 运行你的代码
uv sync                       # 按锁文件装依赖，= go mod download / npm ci
uv python install 3.12        # 连 Python 本身都能装
uv 生成的文件 对应 作用
pyproject.toml go.mod / package.json 项目名、Python 版本、依赖列表
uv.lock go.sum / package-lock.json 锁定每个依赖的精确版本，要提交到 git
.venv/ node_modules/ 虚拟环境，项目独立的 Python + 依赖，不提交
.python-version go 指令那一行 项目用哪个 Python 版本

虚拟环境（venv）是什么：Python 默认把包装到全局，A 项目要 pydantic 1、B 项目要 pydantic 2 就打架。venv 给每个项目一个独立目录放依赖，Go 和 Node 天生就是按项目隔离的，Python 要靠 venv。uv 自动帮你建、自动激活，你只要记住所有命令前面加 uv run。

--lib 生成的目录结构：

my-agent/
├── pyproject.toml
├── uv.lock
├── src/
│   └── my_agent/          # 包名：目录名里用下划线
│       └── __init__.py    # 有这个文件，目录才是一个包（可以是空的）
└── tests/
    └── test_xxx.py        # pytest 自动找 test_ 开头的文件和函数

2. Pydantic：Python 的 Zod

和 TS 一样，Python 的类型注解运行时不检查，模型返回的 JSON 必须真校验。Pydantic v2 就是 Python 的 Zod，而且几乎是所有 AI SDK 的底座：OpenAI SDK、Anthropic SDK、LangChain、FastAPI 的数据类型全是 Pydantic。

定义工具参数

from pydantic import BaseModel, Field


class SearchArgs(BaseModel):
    query: str = Field(min_length=1, description="搜索关键词")
    top_k: int = Field(default=5, ge=1, le=20)

逐行解读：

和 Zod 对照：

const SearchArgs = z.object({
  query: z.string().min(1).describe("搜索关键词"),
  topK: z.number().int().min(1).max(20).default(5),
});
type SearchArgs = z.infer<typeof SearchArgs>;

Pydantic 更省一步：类本身既是校验器，又是类型，不需要 z.infer。

生成 JSON Schema（发给模型）

schema = SearchArgs.model_json_schema()

得到：

{
  "properties": {
    "query": {"description": "搜索关键词", "minLength": 1, "title": "Query", "type": "string"},
    "top_k": {"default": 5, "maximum": 20, "minimum": 1, "title": "Top K", "type": "integer"}
  },
  "required": ["query"],
  "title": "SearchArgs",
  "type": "object"
}

= z.toJSONSchema(SearchArgs)。OpenAI / Anthropic 的 Python SDK 里，你经常直接把 Pydantic 类传给 response_format 或工具定义，SDK 内部就是调用这个方法。

校验模型返回的参数

from pydantic import ValidationError


def parse_args(raw: str) -> tuple[SearchArgs | None, str | None]:
    try:
        return SearchArgs.model_validate_json(raw), None
    except ValidationError as e:
        return None, str(e)

逐行解读：

args, err = parse_args(tool_call.args)
if err is not None:
    return ToolResult(tool_call.id, err, is_error=True)   # 错误文本回传给模型
await search(args.query, args.top_k)

str(e) 的内容大致是：

1 validation error for SearchArgs
top_k
  Input should be less than or equal to 20 [type=less_than_equal, input_value=50, input_type=int]

模型看到这段很容易自己改对。和前两课同一个原则：参数错误是给模型的观察，不是让程序崩的异常。

另外两个常用方法：

SearchArgs.model_validate({"query": "x"})   # 校验一个 dict（已经 json.loads 过的数据）
args.model_dump()                           # 实例 → dict，= Go 的 struct 转 map
args.model_dump_json()                      # 实例 → JSON 字符串，= json.Marshal

可辨识联合

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter


class TextEvent(BaseModel):
    type: Literal["text"]
    delta: str


class ToolCallEvent(BaseModel):
    type: Literal["tool_call"]
    id: str
    name: str
    args: str


StreamEvent = Annotated[
    TextEvent | ToolCallEvent,
    Field(discriminator="type"),
]

adapter = TypeAdapter(StreamEvent)
ev = adapter.validate_json('{"type":"text","delta":"你"}')   # → TextEvent(type='text', delta='你')

逐行解读：

用的时候用 match（Python 3.10+ 的模式匹配，= switch）：

match ev:
    case TextEvent():
        print(ev.delta, end="")
    case ToolCallEvent():
        print(ev.name, ev.args)

3. asyncio：和 JS 一样的事件循环

心智模型

Python 有两种并发方式：线程（threading）和 asyncio。写 Agent 用 asyncio：调模型、调工具几乎全是网络 I/O，asyncio 正是为此设计的。

asyncio 和 JS 的模型几乎完全一样：

import asyncio


async def get_weather(city: str) -> str:
    await asyncio.sleep(1)        # 模拟网络请求；= TS 的 await sleep(1000)
    return f"{city}: 晴"


async def main() -> None:
    w = await get_weather("北京")
    print(w)


asyncio.run(main())               # 启动事件循环，跑 main，跑完退出

逐行解读：

和 JS 的关键区别：协程不 await 就不会运行

async def main():
    get_weather("北京")             # ❌ 什么也不会发生！只是创建了一个协程对象
    await get_weather("北京")       # ✅ 运行并等待

想“开始运行但先不等”（= Go 的 go f()），要显式创建任务：

task = asyncio.create_task(get_weather("北京"))   # 交给事件循环，开始跑
# ... 做别的事
w = await task                                    # 需要结果时再等

并发跑多个：gather 与 TaskGroup

# 方式 1：gather，= Promise.all
a, b = await asyncio.gather(get_weather("北京"), get_weather("上海"))   # 总共约 1 秒

# 方式 2：TaskGroup（Python 3.11+），= errgroup
async with asyncio.TaskGroup() as tg:
    t1 = tg.create_task(get_weather("北京"))
    t2 = tg.create_task(get_weather("上海"))
# 离开 async with 时，所有任务都已完成
print(t1.result(), t2.result())

逐行解读 TaskGroup：

新代码优先用 TaskGroup：它保证“块结束时没有任务在后台乱跑”，相当于语言层面帮你防 goroutine 泄漏。gather 在一个任务失败时不会自动取消其他任务。

限流：Semaphore

sem = asyncio.Semaphore(4)        # 最多 4 个同时进入

async def limited(city: str) -> str:
    async with sem:               # 进入时占一个名额，满了就在这里等
        return await get_weather(city)
    # 离开 with 自动释放名额

= Go 的 g.SetLimit(4)，或者用带缓冲 channel 当信号量 sem := make(chan struct{}, 4)。

超时：asyncio.timeout

try:
    async with asyncio.timeout(2):         # 2 秒
        w = await get_weather("北京")
except TimeoutError:
    print("超时了")

取消：Python 的取消是“注入异常”

这是和 Go / TS 最不同的地方，务必理解：

怎么取消 被取消的代码怎么知道
Go cancel() 自己检查 <-ctx.Done()，不检查就不会停
TS ctrl.abort() 自己监听 signal，不监听就不会停
Python task.cancel() / 超时 在它下一个 await 处自动抛出 CancelledError

也就是说，Python 里你不需要把 ctx / signal 一路往下传。任务被取消时，它正卡在哪个 await 上，那里就会抛出 asyncio.CancelledError，异常一路往上冒，沿途的 finally / with 都会执行清理。这让代码简洁很多。

但有两个代价：

  1. 不 await 的代码取消不了。一个死循环纯计算、或者调用了阻塞函数（time.sleep、requests.get），中间没有 await，取消信号就进不去。和 Go 里不读 ctx.Done() 是一个道理。
  2. 不要吞掉 CancelledError。CancelledError 继承自 BaseException 而不是 Exception，所以 except Exception: 不会误抓它，这是故意设计的。但如果你写了 except BaseException: 或裸 except: 而且不重新抛出，取消就失效了。

4. 并行工具调用：TaskGroup + Semaphore + timeout

0.1 runTools（Go）和 0.2 runTools（TS）的 Python 版：

import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass


@dataclass
class ToolCall:
    id: str
    name: str
    args: str


@dataclass
class ToolResult:
    id: str
    output: str
    is_error: bool


Tool = Callable[[str], Awaitable[str]]


async def run_tools(
    calls: list[ToolCall],
    tools: dict[str, Tool],
    limit: int = 4,
    per_tool_s: float = 20.0,
) -> list[ToolResult]:
    results: list[ToolResult | None] = [None] * len(calls)
    sem = asyncio.Semaphore(limit)

    async def run_one(i: int, c: ToolCall) -> None:
        tool = tools.get(c.name)
        if tool is None:
            results[i] = ToolResult(c.id, f"未知工具 {c.name}", True)
            return
        async with sem:
            try:
                async with asyncio.timeout(per_tool_s):
                    output = await tool(c.args)
                results[i] = ToolResult(c.id, output, False)
            except TimeoutError:
                results[i] = ToolResult(c.id, f"超时（{per_tool_s}s）", True)
            except Exception as e:
                results[i] = ToolResult(c.id, f"{type(e).__name__}: {e}", True)

    async with asyncio.TaskGroup() as tg:
        for i, c in enumerate(calls):
            tg.create_task(run_one(i, c))

    return [r for r in results if r is not None]

逐行解读

类型定义

准备

内部函数 run_one：处理一个工具调用

两种失败，分开处理

启动所有任务

返回

三个版本对照

Go TS Python
并发上限 g.SetLimit(4) 开 4 个 worker Semaphore(4)
启动 g.Go(...) workers.push(worker()) tg.create_task(...)
等待 g.Wait() await Promise.all(...) 离开 async with TaskGroup
单工具超时 context.WithTimeout AbortSignal.timeout asyncio.timeout
工具怎么感知取消 读 ctx.Done() 监听 signal 自动：await 处抛 CancelledError
单个失败 return nil + 写 Err catch 里写结果 except Exception 里写结果
整体取消 return err throw err 不用写，CancelledError 自动冒出

5. 异步生成器：流式读取 SSE

Python 的异步生成器和 TS 的几乎一模一样。HTTP 客户端用 httpx（OpenAI / Anthropic 官方 Python SDK 底层都是它）。

from collections.abc import AsyncIterator

import httpx


async def sse_events(client: httpx.AsyncClient, url: str) -> AsyncIterator[str]:
    async with client.stream("GET", url) as resp:
        resp.raise_for_status()
        async for line in resp.aiter_lines():
            line = line.strip()
            if not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                return
            yield data

逐行解读

用法，以及一个和 JS 不同的坑

from contextlib import aclosing

async with httpx.AsyncClient() as client:
    async with aclosing(sse_events(client, URL)) as events:
        async for data in events:
            print(data)
            if should_stop:
                break

取消整个流式读取，不需要 AbortController：

async with asyncio.timeout(60):       # 60 秒后自动取消
    async with aclosing(sse_events(client, URL)) as events:
        async for data in events:
            ...

超时时，正在 await 下一行数据的地方抛 CancelledError，生成器里的 async with client.stream 退出、关闭连接，最后在外层变成 TimeoutError。

6. 四个高频坑

坑 1：在协程里调用阻塞函数

async def bad_tool(args: str) -> str:
    time.sleep(5)                      # ❌ 阻塞整个线程，事件循环卡死 5 秒，所有任务都停
    return requests.get(URL).text      # ❌ requests 是同步库，同样阻塞

asyncio 只有一个线程，一个协程不 await 就霸占它，其他所有任务（包括超时计时）都动不了。Go 没有这个问题（goroutine 会被调度器抢占），JS 有同样的问题。

修正：用异步版本 await asyncio.sleep(5)、httpx.AsyncClient；实在要调同步库，丢到线程里：await asyncio.to_thread(requests.get, URL)。

坑 2：忘了 await

async def main():
    asyncio.sleep(1)     # ❌ 协程对象被创建又丢掉，什么也没发生

比 TS 更隐蔽：TS 里没 await 至少还会执行，Python 里根本不执行。看到 coroutine ... was never awaited 警告就是它。

坑 3：create_task 后不保存引用

asyncio.create_task(send_metrics())   # ❌ 返回值被丢弃

事件循环只对任务持有弱引用，你不保存它，任务可能执行到一半被垃圾回收。要么保存到变量 / 集合里，要么直接用 TaskGroup（推荐）。

坑 4：异步生成器 break 不关闭

见第 5 节：用 aclosing。

练习（今天做完）

环境准备

mkdir -p projects && cd projects
uv init --lib --name py-agent-core --python 3.12 02-py-agent-core
cd 02-py-agent-core
uv add pydantic httpx
uv add --dev pytest pytest-asyncio

在 pyproject.toml 末尾加上：

[tool.pytest.ini_options]
asyncio_mode = "auto"

asyncio_mode = "auto" 让 pytest 自动用事件循环运行 async def test_... 测试函数。

文件放在 src/py_agent_core/ 下：events.py、tools.py、sse.py；测试放 tests/ 下。

题目

  1. 用 Pydantic 定义 StreamEvent 可辨识联合（text / tool_call / done / error 四种），写 parse_event(data: str)，返回 (event, None) 或 (None, 错误文本)，永远不抛异常
  2. 用 run_tools 跑三个假工具（0.05s / 0.1s / 5s），per_tool_s=0.2，断言前两个成功、第三个 is_error=True、总耗时 < 0.4s
  3. 写一个测试：外层套 asyncio.timeout(0.05) 调用 run_tools，断言抛出 TimeoutError（整体取消能穿透，没有被 except Exception 吞掉）
  4. 用 asyncio.start_server 起一个假 SSE 服务，客户端读 3 条后 break，断言服务端感知到断开
  5. uv run pytest 全绿

自测题

  1. Python 的协程和 JS 的 Promise 有什么关键区别？
  2. 为什么 Python 版 run_tools 不需要像 Go / TS 那样把 ctx / signal 传给每个工具？代价是什么？
  3. except Exception 为什么不会把整体取消吞掉？
  4. TaskGroup 和 gather 有什么区别？为什么新代码优先用 TaskGroup？
  5. 为什么异步生成器要用 aclosing 包起来？TS 里为什么不需要？

参考答案与详解

自测题答案

1. 协程 vs Promise

JS 的 Promise 是“热”的：doWork() 一调用就开始执行，await 只是等结果。Python 的协程是“冷”的：do_work() 只创建一个协程对象，必须 await 它，或者用 create_task / TaskGroup 交给事件循环，才会开始执行。所以 Python 里忘了 await 的后果更严重：代码根本没跑。

2. 为什么不用传 ctx / signal

Python 的取消是“注入异常”：任务被取消（task.cancel()、asyncio.timeout 到期、TaskGroup 里别的任务失败）时，它正卡在哪个 await 上，那里就抛出 CancelledError。工具代码只要是正常的异步代码（await httpx...、await asyncio.sleep），天然就能被取消，不需要显式检查信号。

代价：没有 await 的代码取消不了，比如阻塞调用 time.sleep、同步库 requests、长时间纯计算循环。这和 Go 里 goroutine 不读 ctx.Done()、TS 里工具不监听 signal 是同一类问题，只是形式不同。

3. 为什么 except Exception 不吞取消

从 Python 3.8 起，asyncio.CancelledError 继承自 BaseException，不是 Exception。except Exception 只接普通错误，取消信号会直接穿过去，一路冒到 TaskGroup 和外层。练习 3 就是在验证这一点。反过来，写 except BaseException: 或裸 except: 却不重新 raise，就会把取消吞掉：这个任务收到了取消，却当成普通错误处理后继续往下执行。

4. TaskGroup vs gather

新代码优先用 TaskGroup，它把“每个任务怎么结束”这件事交给语言保证，和 0.1 里“每启动一个 goroutine，都要想清楚它怎么结束”是同一条原则。

5. 为什么要 aclosing

TS 的 for await 里 break，会立刻调用生成器的 return()，finally 马上执行。Python 的 async for 里 break 不会立刻关闭异步生成器，它只是被挂起，等垃圾回收或事件循环关闭时才清理。在这之前，生成器里的 async with client.stream(...) 不会退出，HTTP 连接一直占着。aclosing 在离开 async with 时显式调用 aclose()，保证立刻清理。

练习参考实现

以下代码均已在 Python 3.12、pydantic 2.14、httpx 0.28 下运行通过：uv run pytest → 6 passed。

练习 1：parse_event（src/py_agent_core/events.py）

from typing import Annotated, Literal

from pydantic import BaseModel, Field, TypeAdapter, ValidationError


class TextEvent(BaseModel):
    type: Literal["text"]
    delta: str


class ToolCallEvent(BaseModel):
    type: Literal["tool_call"]
    id: str
    name: str
    args: str


class Usage(BaseModel):
    input: int
    output: int


class DoneEvent(BaseModel):
    type: Literal["done"]
    usage: Usage


class ErrorEvent(BaseModel):
    type: Literal["error"]
    message: str


StreamEvent = Annotated[
    TextEvent | ToolCallEvent | DoneEvent | ErrorEvent,
    Field(discriminator="type"),
]

_adapter = TypeAdapter(StreamEvent)


def parse_event(data: str) -> tuple[StreamEvent | None, str | None]:
    """返回 (event, None) 或 (None, 错误文本)，永远不抛异常。"""
    try:
        return _adapter.validate_json(data), None
    except ValidationError as e:
        return None, str(e)

逐行解读：

测试（tests/test_events.py）：

from py_agent_core.events import TextEvent, parse_event


def test_valid():
    ev, err = parse_event('{"type":"text","delta":"你好"}')
    assert err is None
    assert isinstance(ev, TextEvent)
    assert ev.delta == "你好"


def test_invalid():
    for bad in ['{"type":"text"}', '{"type":"unknown"}', "not json"]:
        ev, err = parse_event(bad)
        assert ev is None
        assert err

练习 2、3：并行工具（src/py_agent_core/tools.py 见第 4 节，测试 tests/test_tools.py）

import asyncio
import time

import pytest

from py_agent_core.tools import ToolCall, run_tools


def fake_tool(seconds: float):
    async def tool(_args: str) -> str:
        await asyncio.sleep(seconds)
        return f"done {seconds}"

    return tool


async def test_slow_tool_times_out():
    start = time.monotonic()
    res = await run_tools(
        [ToolCall("a", "fast", "{}"), ToolCall("b", "mid", "{}"), ToolCall("c", "slow", "{}")],
        {"fast": fake_tool(0.05), "mid": fake_tool(0.1), "slow": fake_tool(5)},
        per_tool_s=0.2,
    )
    assert [r.is_error for r in res] == [False, False, True]
    assert time.monotonic() - start < 0.4


async def test_unknown_tool():
    res = await run_tools([ToolCall("x", "nope", "{}")], {})
    assert res[0].is_error


async def test_outer_cancel_propagates():
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.05):
            await run_tools([ToolCall("a", "slow", "{}")], {"slow": fake_tool(5)})

逐行解读：

练习 4：服务端感知断开（src/py_agent_core/sse.py 见第 5 节，测试 tests/test_sse.py）

import asyncio
from contextlib import aclosing

import httpx

from py_agent_core.sse import sse_events


async def test_client_break_closes_server():
    closed = asyncio.Event()

    async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        await reader.readuntil(b"\r\n\r\n")  # 读掉请求头
        writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n\r\n")
        n = 0
        try:
            while True:
                writer.write(f'data: {{"type":"text","delta":"t{n}"}}\n\n'.encode())
                await writer.drain()  # 客户端断开后这里会抛 ConnectionResetError
                n += 1
                await asyncio.sleep(0.02)
        except (ConnectionResetError, BrokenPipeError):
            closed.set()
        finally:
            writer.close()

    server = await asyncio.start_server(handle, "127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]

    count = 0
    async with httpx.AsyncClient(trust_env=False) as client:
        async with aclosing(sse_events(client, f"http://127.0.0.1:{port}")) as events:
            async for _ in events:
                count += 1
                if count == 3:
                    break

    await asyncio.wait_for(closed.wait(), timeout=2)
    assert count == 3
    server.close()

逐行解读：

服务端

客户端

断言

说明：这个测试里，即使不用 aclosing，离开外层 async with httpx.AsyncClient() 时连接池关闭，连接也会断，所以测试同样能过。但在真实服务里，AsyncClient 通常是全局复用、不会关闭的，那时没有 aclosing 连接就会一直挂着。养成习惯比通过测试更重要。

练习 5

uv run pytest -q
# ......                                                    [100%]
# 6 passed in 0.51s

全部通过，这一课就算完成，可以在 ROADMAP 里勾上 Python。

© 2026 duang777 · 课件源文件：notes/00-foundations/03-python.md