Day02

写给后端工程师的 Python 速成:uv、类型标注、async、pydantic

你已经会写 Java、Go 或 JS,只是不熟 Python。今天用对照的方式补上写 Agent 够用的那部分:uv 管项目、类型标注、dataclass 和 pydantic、async/await 并发、httpx、.env 管密钥、pytest。最后写一个并发抓取 + 数据校验 + 单测的小程序。

读完约 56 分钟

基础篇 · 第 2 天,共 60 天

今天要解决的问题

Agent 岗的 JD 十份里有九份写 Python。原因不复杂:各家模型的 SDK、LangGraph、MCP、评测工具,第一个支持的都是 Python。你用 Java 或 Go 也能做 Agent(Spring AI、LangChain4j 都有),但生态和资料差一截,面试时也难免被问「Python 熟不熟」。

好消息是,对有经验的工程师来说,Python 一天就能上手到「能写 Agent」的程度。今天不讲语法大全,只讲后面 58 天会反复用到的东西。学完你应该能:

  1. 用 uv 建项目、加依赖、跑脚本,知道 pyproject.toml 和 uv.lock 各是干什么的。
  2. 读懂和写出带类型标注的 Python,知道类型标注在运行时不做检查。
  3. 分清 dataclass 和 pydantic 的用途,用 pydantic 校验外部数据。
  4. 用 async/await 和 asyncio.gather 并发发请求,知道它和 Java 线程、Go goroutine 的区别。
  5. 用 pytest 写和跑测试。
  6. 用 Coding Agent 辅助学习和写代码,并能按一张清单审查它生成的代码。

预计用时:4 到 5 小时。如果你已经熟悉 Python,快速扫一遍「常见误区」和练习就行。

如果你是前端出身,今天会比后端同学轻松一些:Python 的 async/await 和 JS 几乎是同一套心智模型,pydantic 的角色和 zod 很像,uv 的体验也接近 pnpm。要多花时间的反而是类型标注和测试的习惯。如果你是 Java 出身,最不习惯的大概是「没有编译期检查」,代码写错了要跑到那一行才知道。这也是为什么我们第二天就把测试和类型标注一起讲。

先建立对照表

把你熟悉的东西映射过来,学得最快:

你熟悉的Python 里对应的一句话说明
Maven / Gradle / npm / go moduv(+ pyproject.toml)管依赖、管虚拟环境、管 Python 版本
package.json / pom.xmlpyproject.toml项目元数据和依赖声明
package-lock.json / go.sumuv.lock锁定精确版本,要提交进 git
node_modules.venv/ 目录每个项目独立的依赖环境,不提交
Java POJO / Go struct@dataclass纯数据容器
Bean Validation / zodpydantic带校验的数据模型
CompletableFuture / goroutine / Promiseasync def + await + asyncio单线程事件循环的并发
OkHttp / net/http / axioshttpxHTTP 客户端,同步异步都支持
JUnit / go test / Jestpytest测试框架
.env + dotenvpython-dotenv从 .env 加载环境变量

uv:一个工具管完 Python 项目

几年前 Python 的依赖管理是出了名的乱:pip、virtualenv、pipenv、poetry、conda 各有一批用户。2024 年之后 uv 迅速成了事实标准,它是用 Rust 写的,装依赖比 pip 快一个数量级,而且把「装 Python」「建虚拟环境」「装依赖」「锁版本」「跑脚本」全包了。

Day 0 你已经 uv init 过一个项目。常用的命令就这几个:

uv init my-project --python 3.12   # 新建项目
uv add httpx pydantic               # 加依赖,自动写进 pyproject.toml 并更新 uv.lock
uv add --dev pytest                 # 加开发依赖(测试、lint 工具)
uv remove httpx                     # 删依赖
uv run python main.py               # 在项目的虚拟环境里跑
uv run pytest                       # 跑测试
uv sync                             # 别人 clone 了你的仓库,一条命令按 uv.lock 装好全部依赖

你几乎不需要手动「激活虚拟环境」。uv run 会自动用项目下的 .venv。这一点比 Java 和 Node 都省心。

pyproject.toml 长这样:

[project]
name = "agent-60-lab"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "openai>=1.0",
    "python-dotenv>=1.0",
]

[dependency-groups]
dev = ["pytest>=8"]

记住两条:pyproject.toml 和 uv.lock 都提交进 git;.venv/ 不提交(uv 生成的 .gitignore 默认已经忽略了)。

类型标注:写给人和工具看的,运行时不管

Python 是动态类型语言,但 3.5 之后支持类型标注,现在写正经项目基本都标。语法:

def total_cost(prompt_tokens: int, completion_tokens: int, price_in: float, price_out: float) -> float:
    return (prompt_tokens * price_in + completion_tokens * price_out) / 1_000_000


# 常用类型
names: list[str] = ["deepseek", "qwen"]
prices: dict[str, float] = {"in": 2.0, "out": 8.0}
maybe_id: str | None = None          # 可能为空,相当于 Java 的 @Nullable / TS 的 string | null
pair: tuple[int, int] = (3, 4)
  • def total_cost(prompt_tokens: int 每个参数后面的 : int、: float 是类型标注,-> float 标的是返回值。它们都存在函数身上,实测 total_cost.__annotations__ 是 {'prompt_tokens': <class 'int'>, 'completion_tokens': <class 'int'>, 'price_in': <class 'float'>, 'price_out': <class 'float'>, 'return': <class 'float'>}。Day 9 的工具注册器就是读这个。
  • return (prompt_tokens * price_in + completion_tokens * price_out) / 1_000_000 单价的单位是「元 / 百万 token」,所以最后除以一百万。1_000_000 里的下划线只是方便人看,和 1000000 完全一样。代入 Day 0 那次真实调用 total_cost(8, 29, 2.0, 8.0):(8×2 + 29×8) / 1_000_000 = 248 / 1_000_000 = 0.000248 元。
    标注不拦错:实测 total_cost("3", 2, 2.0, 8.0) 能进到函数里,直到做乘法才炸 TypeError: can't multiply sequence by non-int of type 'float'。
  • names: list[str] = ["deepseek", "qwen"] 变量也能标类型,写法是 名字: 类型 = 值。list[str] 读作「字符串列表」。
  • prices: dict[str, float] 键是字符串、值是浮点数的 dict。
  • maybe_id: str | None = None | 表示「二选一」,这里是字符串或者空。判断它是不是空要写 if maybe_id is None:。

和 Java、Go 最大的不同:这些标注在运行时不做任何检查。你声明了 int,传个字符串进去照样能跑,直到某一行真正用错了才炸。类型标注的价值在于:编辑器能补全和提示、mypy / pyright 这类工具能静态检查、读代码的人一眼看懂。

tuple(元组)和解包:后面每天都会用

上面最后一行的 tuple 后面几课会频繁出现,单独讲一下。tuple 就是用圆括号写的、长度固定、不能改的 list,相当于 TS 的元组类型 [number, number]:

pair: tuple[int, int] = (3, 4)            # TS: const pair: [number, number] = [3, 4]
price: tuple[float, float, float] = (0.2, 2.0, 3.0)   # 三个位置,各自含义固定
pair[0]          # 3,和 list 一样按下标取
pair[0] = 5      # 报错:tuple 不能改

它最常见的两种用法,都和解包有关(相当于 TS 的数组解构 const [a, b] = ...):

# 1. 函数一次返回多个值:return 后面用逗号隔开,其实就是返回一个 tuple
def run() -> tuple[float, list[str]]:
    return 0.9, ["错例1"]          # 等价于 return (0.9, ["错例1"])

acc, wrong = run()                 # 解包:acc=0.9,wrong=["错例1"]

# 2. 一条数据由几个固定字段组成,懒得为它定义 class
TICKETS: list[tuple[str, str]] = [  # 每个元素是 (工单文本, 正确类别)
    ("扣了两次钱", "billing"),
    ("忘记密码", "account"),
]
for ticket, gold in TICKETS:        # 遍历时直接解包成两个变量
    print(ticket, gold)

类型标注里 tuple[str, str] 读作「两个字符串组成的元组」,list[tuple[str, str]] 读作「元组的列表」。字段一多(超过三四个)或者要起名字时,就改用下面讲的 dataclass。

dict(字典):遍历要用 .items()

dict 就是键值对,相当于 TS 的对象字面量或 Map。课程里存「版本名 → 提示词」「价格名 → 单价」都用它:

VERSIONS: dict[str, str] = {"v1": "你是客服…", "v2": "你是客服,规则如下…"}
VERSIONS["v1"]               # 取值,键不存在会抛 KeyError
VERSIONS.get("v9")           # 键不存在返回 None,不报错(像 TS 里 obj.v9 拿到 undefined)
VERSIONS["v3"] = "新版本"     # 加一项
"v1" in VERSIONS             # 判断键在不在,TS: "v1" in obj

遍历是最容易卡住的地方:直接 for x in 字典 只拿到键。要键和值一起拿,用 .items():

for name in VERSIONS:                   # 只有键:v1, v2
    ...
for name, system in VERSIONS.items():   # 键和值,TS: for (const [name, system] of Object.entries(VERSIONS))
    ...

.keys() / .values() / .items() 分别对应 TS 的 Object.keys / Object.values / Object.entries。.items() 每次给出一个 (键, 值) 的 tuple,for 顺手把它解包成两个变量,就是上面讲的解包。

set(集合):只关心「在不在」

set 是不重复、无序的一组值,相当于 TS 的 Set。Day 5 用它装合法的分类标签:

LABELS: set[str] = {"billing", "technical", "account", "other"}
"billing" in LABELS          # True,TS: LABELS.has("billing")
LABELS[0]                    # 报错:set 无序,没有下标

用 set 不用 list,是因为它查「在不在」很快(哈希查找),而且自动去重。

set 还能直接做集合运算,比对「两边差了什么」时特别好用:

model_ids = {"INV-001", "INV-003"}       # 模型抽出来的
real_ids  = {"INV-001", "INV-002"}       # 正则从原文找到的

model_ids - real_ids    # {"INV-003"}  差集:模型有、原文没有(编造的)
real_ids - model_ids    # {"INV-002"}  差集:原文有、模型没有(漏抽的)
model_ids & real_ids    # {"INV-001"}  交集:两边都有
model_ids | real_ids    # 并集:合在一起去重

JS 的 Set 直到最近才有 difference()、intersection() 这些方法,Python 一个运算符就搞定。差集有方向,A - B 和 B - A 结果不一样。list 不能这么减,要先用 set(...) 转一下。Day 6 的 verify_ids 就是靠这两个方向的差集,找出模型编造和漏抽的发票号。

注意 set 和 dict 都用花括号,有冒号是 dict,没冒号是 set;空的 {} 是 dict,空 set 要写 set()。

四种容器放一起对照:

Python字面量类型标注TS 里对应什么时候用
list[1, 2]list[int]number[]一串同类数据,要按顺序、会增删
tuple(1, 2)tuple[int, int][number, number](只读)几个位置固定的字段、函数返回多个值
dict{"k": 1}dict[str, int]对象 / Map按名字查值
set{1, 2}set[int]Set<number>去重、判断在不在

Literal:只能取这几个值

有些字段只能取固定的几个值,比如工单分类。用 Literal 写:

from typing import Literal

category: Literal["billing", "technical", "account", "other"] = "billing"
# TS: category: "billing" | "technical" | "account" | "other"

str 是什么字符串都行,Literal[...] 只认列出来的这几个。光是类型标注,运行时照样不检查(category = "payment" 能跑,只是编辑器画红线);放进 pydantic 模型里就会真校验:

from pydantic import BaseModel

class Ticket(BaseModel):
    category: Literal["billing", "technical", "account", "other"]

Ticket(category="billing")   # 通过
Ticket(category="Billing")   # ValidationError:大小写不对也不行

它还会进 model_json_schema(),变成 {"enum": ["billing", "technical", "account", "other"], "type": "string"},把 schema 写进提示词,模型就能直接看到只有这几个选项。Day 6 的结构化抽取就是靠这个把模型的输出限制住:Day 5 是自己写 if word in LABELS 检查,Day 6 把规则写进类型里,交给 pydantic。

回到类型标注:它和 Agent 有什么关系?关系很大。Day 9 我们会写一个装饰器,读取函数的类型标注,自动生成给模型看的工具 JSON Schema。LangChain、MCP 的 Python SDK、OpenAI Agents SDK 全是这么干的。你写:

def get_invoice(invoice_id: str, include_lines: bool = False) -> dict:
    """按发票号查询发票详情。"""

框架就能自动生成「这个工具叫 get_invoice,有一个必填字符串参数 invoice_id 和一个可选布尔参数 include_lines」的描述。类型标注在这里就是接口契约。

dataclass 和 pydantic:一个装数据,一个守大门

dataclass 就是 Java 的 record 或 Go 的 struct,自动生成构造函数、__repr__、__eq__:

from dataclasses import dataclass, field


@dataclass
class Usage:
    prompt_tokens: int
    completion_tokens: int
    tags: list[str] = field(default_factory=list)   # 可变默认值必须这么写


u = Usage(prompt_tokens=120, completion_tokens=30)
print(u)  # Usage(prompt_tokens=120, completion_tokens=30, tags=[])
  • from dataclasses import dataclass, field 都是标准库,不用装。field 是小写的函数,下面专门有一节讲它和 pydantic 大写 Field 的区别。
  • @dataclass 装饰器读取下面写的字段,自动生成 __init__(构造函数)、__repr__(打印出来长什么样)、__eq__(怎么判断相等)。你只写了三行字段,等于省了二十行样板代码。
  • prompt_tokens: int 没有默认值的字段是构造时必须传的。
  • tags: list[str] = field(default_factory=list) 每建一个实例,就调用一次 list() 得到一个新的空列表。实测两个实例的 u.tags is u2.tags 是 False,各有各的列表。如果直接写 = [],实测在定义类的那一刻就报错:ValueError: mutable default <class 'list'> for field tags is not allowed: use default_factory。
  • u = Usage(prompt_tokens=120, completion_tokens=30) 调的是自动生成的 __init__。tags 没传,走默认,得到一个新的空列表。
  • print(u) 实测输出 Usage(prompt_tokens=120, completion_tokens=30, tags=[]),这是自动生成的 __repr__。自动生成的 __eq__ 按字段逐个比较:实测 u == Usage(120, 30) 是 True,u is Usage(120, 30) 是 False(值一样,但不是同一个对象)。
    正文下面那句「dataclass 不校验」也实测过:Usage(prompt_tokens="abc", completion_tokens=None) 照样建出来,打印成 Usage(prompt_tokens='abc', completion_tokens=None, tags=[])。

dataclass 不校验。Usage(prompt_tokens="abc", completion_tokens=None) 照样能建出来。

pydantic 是带校验的数据模型,会按类型标注检查并转换输入:

from pydantic import BaseModel, Field, ValidationError


class Ticket(BaseModel):
    ticket_id: str
    customer_email: str
    category: str = Field(description="billing / technical / account / other")
    priority: int = Field(ge=1, le=5)


t = Ticket.model_validate({"ticket_id": "T-1001", "customer_email": "a@b.com",
                           "category": "billing", "priority": "3"})
print(t.priority, type(t.priority))   # 3 <class 'int'>,字符串 "3" 被转成了 int

try:
    Ticket.model_validate({"ticket_id": "T-1002", "customer_email": "x", "category": "billing", "priority": 9})
except ValidationError as e:
    print(e)   # 会指出 priority 必须 <= 5
  • from pydantic import BaseModel, Field, ValidationError 三样东西:模型基类、字段配置、校验失败时抛的异常。
  • class Ticket(BaseModel): 继承 BaseModel 就有了校验、转换、导出 schema 这些能力。
  • ticket_id: str 写法和 dataclass 一样,也是必填字段。
  • category: str = Field(description= 注意 Field(...) 没传默认值,所以 category 仍然是必填的。description 不参与校验,只会写进 model_json_schema() 给模型看。
  • priority: int = Field(ge=1, le=5) ge 是 greater or equal(≥),le 是 less or equal(≤),合起来就是 1 到 5。
  • t = Ticket.model_validate( 拿一个 dict 来建模型,边建边校验。这里故意把 priority 写成字符串 "3",模拟模型输出里数字带了引号的情况。
  • print(t.priority, type(t.priority)) 实测 3 <class 'int'>。pydantic 默认是宽松模式:能干净转成整数的字符串就帮你转了。打印整个对象是 Ticket(ticket_id='T-1001', customer_email='a@b.com', category='billing', priority=3)。换成 "高" 就转不了,实测报 Input should be a valid integer, unable to parse string as an integer。
  • Ticket.model_validate({"ticket_id": "T-1002" 这一条有两处可疑:customer_email 是 "x",priority 是 9。只有 9 会报错,"x" 能通过,因为字段类型只是 str。要校验邮箱格式得用 pydantic 的 EmailStr。
  • except ValidationError as e: 只接住校验错误。别的异常不在这里吞掉,该炸照样炸。
  • print(e) # 会指出 priority 必须 <= 5 实测输出:
    1 validation error for Ticket
    priority
    Input should be less than or equal to 5 [type=less_than_equal, input_value=9, input_type=int]
    哪个字段、违反什么规则、你传的是什么,都写得清清楚楚。e.error_count() 是 1。Day 6 会把这段错误原样发回给模型,让它自己改。

我的选择规则很简单:数据从外部来(HTTP 请求、模型输出、配置文件),用 pydantic;纯内部传来传去,用 dataclass 或者也用 pydantic 都行。Agent 里最常见的「外部数据」就是模型的输出。模型说它返回了 JSON,你不能信,必须校验。Day 6 整天都在讲这件事,主角就是 pydantic。

pydantic 还有一个后面会反复用的能力:Ticket.model_json_schema() 能直接导出 JSON Schema。这就是 Structured Outputs 和工具定义的原料。

小写 field 和大写 Field:两个库,别用混

上面两段代码里,一个用小写 field,一个用大写 Field。它们来自不同的库,只是名字碰巧一样:

from dataclasses import field   # 标准库,给 @dataclass 用
from pydantic import Field      # pydantic 的,给 BaseModel 用

用哪个看类是什么:@dataclass 的类用小写 field,继承 BaseModel 的类用大写 Field。两者都写在 = 右边,也就是放默认值的位置:

@dataclass
class FakeClient:
    seen: list[str] = field(default_factory=list)

class TicketInfo(BaseModel):
    customer_name: str | None = Field(None, description="工单中没有就填 null")
    invoice_ids: list[str] = Field(default_factory=list, description="原样抄写发票号")
    urgency: int = Field(ge=1, le=5)
  • Field(None, ...) 的第一个参数是默认值:字段没给就填 None,所以这个字段是可选的。注意它和类型里的 str | None 管的不是一回事,类型说「值可以是空」,默认值说「不给就当空」。
  • Field(ge=1, le=5) 没给默认值,字段就是必填的。
  • 大写 Field 比小写 field 多了 description(会进 model_json_schema(),模型能看到)和 ge/le/max_length 这类校验,因为 pydantic 本来就是做校验和生成 schema 的。小写 field 只管默认值。

大小写的来历:Python 习惯类名大写开头、函数小写。pydantic 的 Field 写成了类的样子,标准库的 field 就是个普通函数。用混了编辑器会画红线,比如在 BaseModel 里写 field(description=...),小写 field 根本没有这个参数。

async/await:等网络的时候别闲着

Agent 程序的时间大头花在等网络:等模型 API 返回(几秒到几十秒)、等数据库、等外部 API。如果有 5 个工具调用可以同时做,一个一个串行等就很浪费。

Python 的异步模型和 JS 几乎一样:单线程事件循环。async def 定义协程函数,await 表示「这里要等 IO,先去干别的」。和 Java 线程池、Go goroutine 不同,它不会让 CPU 计算并行起来,只是让等待重叠。对 Agent 这种 IO 密集的程序,这正好够用。

串行:for 循环里逐个 await 请求 A 1.2s B 0.8s C 1.5s 总耗时 ≈ 3.5s 并发:asyncio.gather(A, B, C) 请求 A B 请求 C 总耗时 ≈ 最慢的那个 1.5s 横轴是时间,1 秒 ≈ 100 像素。数字是示意。
三个互不依赖的网络请求,串行等待的总时间是三者之和,并发等待的总时间约等于最慢的那一个。Agent 里一次要调多个工具时,这个差别非常明显。

最小例子:

import asyncio
import time


async def fake_call(name: str, seconds: float) -> str:
    await asyncio.sleep(seconds)          # 模拟等网络
    return f"{name} done"


async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fake_call("A", 1.2), fake_call("B", 0.8), fake_call("C", 1.5)
    )
    print(results, f"{time.perf_counter() - start:.1f}s")   # 约 1.5s


asyncio.run(main())
  • async def fake_call(name: str, seconds: float) -> str: 带 async 的函数叫协程函数。调用它不会执行函数体,只会得到一个协程对象:实测 type(fake_call("A", 1.2)).__name__ 是 'coroutine'。要 await 它,或者交给 gather,里面的代码才会跑。
  • await asyncio.sleep(seconds) 「我要等 seconds 秒,这段时间事件循环先去跑别的协程」。换成 time.sleep 就是死等,谁也跑不了。
  • return f"{name} done" 比如 "A done"。
  • start = time.perf_counter() 高精度计时器。它返回的数字本身没意义,只用来算两次调用之间的差值。
  • results = await asyncio.gather( 把三个协程同时交给事件循环,等全部结束再一起返回。我在每个协程开始和结束时各打一次时间,实测时间线是:
    0.00 秒:A、B、C 同时开始等
    0.80 秒:B 结束
    1.20 秒:A 结束
    1.50 秒:C 结束,gather 返回
  • print(results, f"{time.perf_counter() - start:.1f}s") 实测打印 ['A done', 'B done', 'C done'] 1.5s(精确值是 1.5005 秒)。注意结果顺序和传进去的顺序一致,不是完成的顺序:B 最先完成,却排在第二。我把 gather 换成 for 循环逐个 await 再跑一次,实测 3.50 秒,正好是 1.2 + 0.8 + 1.5。
  • asyncio.run(main()) 建一个事件循环,跑完 main(),再把循环关掉。整个程序里一般只在最外层调一次。

三个容易踩的坑,我都踩过:

  • 在 async 函数里调用了同步的阻塞函数,比如 time.sleep(1) 或者同步版的 requests.get()。整个事件循环会被卡住,所有协程一起等。异步代码里要用 asyncio.sleep、httpx.AsyncClient、AsyncOpenAI。
  • 忘了 await。fake_call("A", 1) 不加 await 只会得到一个协程对象,什么也没执行,Python 会给一个 RuntimeWarning: coroutine ... was never awaited。
  • 在普通函数里想调 async 函数。顶层用 asyncio.run(main());如果你在 Jupyter 里,事件循环已经在跑了,直接 await main()。

gather 默认是「一个失败全部失败」。如果你希望某个工具失败了其他照常返回,加 return_exceptions=True,然后逐个检查结果是不是异常。Day 12 讲 Agent 失败处理时会用到。

httpx:同步异步一套 API

requests 很经典但只有同步版。httpx 的接口几乎和 requests 一样,同时支持异步,而且 OpenAI 官方 Python SDK 底层用的就是它。

import httpx

# 同步
r = httpx.get("https://api.github.com/repos/astral-sh/uv", timeout=10)
print(r.status_code, r.json()["stargazers_count"])


# 异步
async def stars(repo: str, client: httpx.AsyncClient) -> int:
    r = await client.get(f"https://api.github.com/repos/{repo}", timeout=10)
    r.raise_for_status()
    return r.json()["stargazers_count"]
  • r = httpx.get("https://api.github.com/repos/astral-sh/uv", timeout=10) 同步请求,拿到响应之前这一行一直卡着。实测从国内访问 GitHub 花了 2.52 秒。
  • print(r.status_code, r.json()["stargazers_count"]) 实测打印 200 90999(10 月 22 日 uv 的星数,你跑的时候会更多)。r.json() 把响应体解析成 dict,GitHub 这个接口一共返回 86 个字段,我们只取了一个。响应头里的 x-ratelimit-remaining 当时是 57,也就是这个小时的匿名请求还剩 57 次。
  • async def stars(repo: str, client: httpx.AsyncClient) -> int: 异步版本。client 由调用方传进来,多个请求共用同一个连接池,比每次新建一个快。
  • r = await client.get(f"https://api.github.com/repos/{repo}", timeout=10) 和同步版差不多,区别是前面加了 await,等网络的时候可以让出去。
  • r.raise_for_status() 状态码是 4xx 或 5xx 就抛 httpx.HTTPStatusError。不加这一行的话,仓库不存在时拿到的是 404,响应体是 {"message": "Not Found", ...},下一行取 stargazers_count 就会报 KeyError。那个报错比「HTTP 404」难懂得多。
  • return r.json()["stargazers_count"] 返回 int,比如 90999。

注意 timeout。httpx 默认超时是 5 秒,模型 API 动辄几十秒,直接用 httpx 调模型时要把超时调大。用 OpenAI SDK 的话它自己有更长的默认值。

读框架源码前要认识的几个写法

后面读 LangGraph、MCP SDK 的示例代码时,有几种 Python 写法会反复出现。Java 和 Go 里没有完全对应的东西,第一次见容易卡住,这里提前过一遍。

装饰器。 函数上面那一行 @xxx,本质是「把下面这个函数传给 xxx,用它的返回值替换原函数」。它在 Agent 框架里到处都是:MCP SDK 用 @mcp.tool() 把普通函数注册成工具,LangChain 用 @tool,FastAPI 用 @app.get("/") 注册路由。你可以把它理解成 Java 的注解加上一段在类加载时执行的处理逻辑,只不过 Python 里它就是一个普通函数。

from collections.abc import Callable

TOOLS: dict[str, Callable[..., str]] = {}


def tool(func):
    """最简单的注册型装饰器:把函数登记到 TOOLS 里,原样返回。"""
    TOOLS[func.__name__] = func
    return func


@tool
def get_weather(city: str) -> str:
    """查询城市天气。"""
    return f"{city}:晴,22 度"


print(TOOLS)                      # {'get_weather': <function get_weather ...>}
print(get_weather.__doc__)        # 查询城市天气。
print(get_weather.__annotations__)  # {'city': <class 'str'>, 'return': <class 'str'>}
  • from collections.abc import Callable Callable 是「可以调用的东西」这个类型,函数、带 __call__ 的对象都算。Python 3.9 起推荐从 collections.abc 导入,typing.Callable 也能用,只是旧写法。
  • TOOLS: dict[str, Callable[..., str]] = {} 工具注册表:名字 → 函数。Callable[..., str] 读作「参数随便、返回 str 的可调用对象」:方括号里第一项是参数列表,... 表示不限定,第二项是返回类型。只写 Callable 不带方括号,basedpyright 会报 Expected type arguments。注意别写成小写的 callable:那是一个内置函数(callable(print) 返回 True),不是类型,能跑但类型检查器会报错。
  • def tool(func): 装饰器本身就是一个普通函数:收一个函数,返回一个函数。
  • TOOLS[func.__name__] = func __name__ 是函数的名字字符串。实测这里是 'get_weather',所以执行完后 TOOLS 里多了一项 'get_weather' → 函数本身。
  • return func 原样还回去,所以被装饰后的 get_weather 还是原来那个函数。实测 TOOLS["get_weather"] is get_weather 为 True。
  • @tool 等价于在 def 后面补一句 get_weather = tool(get_weather)。注意它在定义函数的时候就执行了,不是等到调用 get_weather 时。我在 tool 里加了 print 验证过:def 那几行一执行完,注册就完成了。这也是为什么只要 import 了这个模块,工具就已经登记好了。
  • return f"{city}:晴,22 度" 实测 TOOLS["get_weather"]("上海") 返回 '上海:晴,22 度'。按名字查到函数再调用,Agent 收到模型说「调 get_weather」时干的就是这件事。
  • print(TOOLS) 实测 {'get_weather': <function get_weather at 0x10e9a53a0>}。后面那串是内存地址,每次运行都不一样。
  • print(get_weather.__doc__) 函数下面第一行三引号字符串就是 docstring,实测打印 查询城市天气。。它会当成工具描述发给模型。
  • print(get_weather.__annotations__) 实测 {'city': <class 'str'>, 'return': <class 'str'>}。参数名和类型都在这里,Day 9 用它拼出参数的 JSON Schema。

最后两行很关键:函数的文档字符串和类型标注在运行时都能读到。Day 9 的工具注册器就是用这两样东西,自动拼出给模型看的工具描述。

上下文管理器(with)。 with open(...) as f:、async with httpx.AsyncClient() as client: 这种写法,保证代码块结束时资源一定被释放,哪怕中间抛了异常。相当于 Java 的 try-with-resources 或 Go 的 defer。用 httpx 时记得放进 with 里,不然连接池不会被关闭。

推导式和 f-string。 [m["content"] for m in messages if m["role"] == "user"] 是列表推导,一行完成过滤和映射,相当于 Java Stream 的 filter + map。f"花费 {cost:.4f} 元" 是格式化字符串,冒号后面是格式说明。这两个在 Python 代码里出现频率极高,看几次就熟了。

常用字符串方法。 处理模型输出天天要用,对照 JS 记:

PythonJS例子
s.strip()s.trim()" billing\n".strip() → "billing"
s.lstrip() / s.rstrip()trimStart() / trimEnd()只去左边 / 右边
s.lower() / s.upper()toLowerCase() / toUpperCase()"Billing".lower() → "billing"
s.split(",")s.split(",")"a,b".split(",") → ["a", "b"]
",".join(items)items.join(",")注意 Python 是分隔符在前
s.replace(a, b)s.replaceAll(a, b)Python 默认替换全部
s.startswith(x)s.startsWith(x)
x in ss.includes(x)"INV" in text

strip() 去掉两头的空格、换行、制表符,也可以指定字符:"**billing**".strip("*") → "billing"。模型输出前后经常多带一个换行,所以拿它和预期值比较前,惯例先 .strip().lower()(Day 5 的 parse());判断「模型实际上什么都没返回」写 if not r.content.strip():(Day 6),只有空白的字符串去掉后是 "",在 if 里算假。

Python 的字符串不可修改,这些方法都返回新字符串,原字符串不变。要写 s = s.strip(),只写 s.strip() 什么也没发生。

比较和逻辑运算。 Python 只有两个等号,没有三等号:

意思PythonJS
值相等 / 不等a == b / a != ba === b / a !== b
是不是同一个对象a is b / a is not b大致是 Object.is
非 / 且 / 或not x / a and b / a or b!x / a && b / a || b
不在里面x not in items!items.includes(x)

Python 的 == 不做类型转换,1 == "1" 是 False,行为和 JS 的 === 一样。一个例外:True == 1、False == 0 都成立,因为 bool 在 Python 里是 int 的子类。

== 比的是值,is 比的是是不是同一个对象:两个内容相同的列表 == 为真、is 为假。日常只在判断 None 时用 is:写 if x is None: / if x is not None:,不要写 x == None(Ruff 会提示 E711)。其他比较一律用 == 和 !=。

真值判断。 空列表、空字符串、0、None 在 if 里都是假。if messages: 就是「列表非空」。这很方便,但也有坑:一个工具返回了 0(比如余额为零),你写 if result: 会把它当成「没有结果」。需要区分「没有」和「是零」时,写 if result is not None:。

异常。 Python 用异常处理错误,没有 Go 那种多返回值的 err。try / except 具体异常类型 as e,尽量捕获具体类型,不要一个光秃秃的 except: 把所有错误都吞掉,那样调试时你会完全不知道发生了什么。Agent 里有一个特殊情况:工具执行出错时,通常不让程序崩溃,而是把错误信息转成字符串交给模型,让它自己决定换个参数重试。Day 12 会专门讲。

字典就是 JSON。 Python 的 dict 和 list 跟 JSON 一一对应,json.loads 把字符串变成 dict,json.dumps(obj, ensure_ascii=False) 变回字符串。ensure_ascii=False 记得加,不然中文会变成 云簿 这种转义,日志里没法看,发给模型也白白多占 token。

type[X]:传「类本身」。 平时 x: TicketInfo 表示 x 是一个实例;model: type[TicketInfo] 表示传进来的是类本身:

def schema_prompt(model: type[BaseModel]) -> str:
    return json.dumps(model.model_json_schema(), ensure_ascii=False)

schema_prompt(TicketInfo)                 # 传类,没有括号
schema_prompt(TicketInfo(category=...))   # 错:这是实例

为什么要传类:model_json_schema() 读的是类上定义的字段、类型和 description,这时候还没有任何数据,谈不上实例。TS 里接近 typeof TicketInfo 或 new (...args: any) => BaseModel。Day 6 的 extract(client, TicketInfo, ...) 也是传类进去,函数内部拿它生成 schema、校验模型输出,最后返回一个实例。

泛型。 想让函数「传进什么类、就返回什么类的实例」时用。Day 6 的 extract() 就是这样:

from pydantic import BaseModel

def extract[T: BaseModel](model: type[T], text: str) -> T:   # TS: function extract<T extends BaseModel>(...)
    return model.model_validate_json(text)

info = extract(TicketInfo, raw)        # 编辑器知道 info 是 TicketInfo,能补全字段

函数名后面的 [T: BaseModel] 就是 TS 的 <T extends BaseModel>,限定 T 必须是某个 pydantic 模型;type[T] 是「类本身」,-> T 是「它的实例」。返回类型如果只写 -> BaseModel,编辑器就不知道 info 有哪些字段了。

这是 Python 3.12 的新写法,课程项目要求 ≥3.12,统一用它。读库和框架源码时会经常看到老写法,意思完全一样:

from typing import TypeVar

T = TypeVar("T", bound=BaseModel)      # 先单独定义 T,bound 就是 extends
def extract(model: type[T], text: str) -> T: ...

在 3.12 项目里写老写法,Ruff 会给黄色警告 UP047 Generic function should use type parameters,意思就是「建议换成新写法」,不影响运行。

密钥:.env + python-dotenv

Day 0 已经用过了。原则再说一遍:密钥只放在 .env 或者系统环境变量里,代码里用 os.environ["LLM_API_KEY"] 读。用 os.environ[...] 而不是 os.getenv(...),前者没配置时会立刻报 KeyError,比拿到一个 None 跑到一半再报奇怪的 401 好排查得多。

再提交一个 .env.example 进 git,列出需要哪些变量但不写真实值,方便别人(和面试官)跑你的项目。

load_dotenv() 做的事就是读 .env 文件,把每行 KEY=值 设成环境变量,之后 os.environ 就能读到,和 Node 的 require("dotenv").config() 一样。它有一个容易忽略的参数 override:

load_dotenv()                                  # 读当前目录的 .env;已经存在的环境变量不覆盖
load_dotenv(".env.deepseek", override=True)    # 读指定文件,并用文件里的值覆盖已有的

默认不覆盖,意思是「外面已经设过的值优先,.env 只兜底」。这正是你想要的:服务器上直接配的环境变量比代码目录里的文件优先;测试里用 monkeypatch.setenv("LLM_API_KEY", "test") 设的假 key,不会被本地 .env 的真 key 顶掉,测试也就不会真的去调 API 花钱。

而明确指定了某个文件时,说明你就要这份配置,这时要加 override=True。Day 3 的 LLMClient 就是这么写的:

if env_file:
    load_dotenv(env_file, override=True)   # 指定了文件:以它为准
else:
    load_dotenv()                          # 没指定:.env 只兜底

Day 7 聊天机器人的 /env .env.claude 切换厂商靠的就是这一点:启动时已经从 .env 读进了 LLM_MODEL=deepseek-flash,不加 override=True 的话,新文件里的值不会生效,等于没切。

pytest:最省事的测试框架

pytest 不需要写测试类,函数名以 test_ 开头、文件名以 test_ 开头就行,断言直接用 assert。

先把上面类型标注那节的 total_cost 存成项目根目录的 cost.py(测试要 import 它,没这个文件会报 ModuleNotFoundError: No module named 'cost'),再在 pyproject.toml 末尾加两行,让 pytest 从项目根目录找模块:

[tool.pytest.ini_options]
pythonpath = ["."]

然后写测试:

# tests/test_cost.py
from cost import total_cost


def test_total_cost():
    assert total_cost(1_000_000, 0, 2.0, 8.0) == 2.0
    assert total_cost(0, 500_000, 2.0, 8.0) == 4.0
  • from cost import total_cost 从项目根目录的 cost.py 导入。能找到它,靠的是上面 pyproject.toml 里加的 pythonpath = ["."]。
  • def test_total_cost(): 函数名以 test_ 开头,pytest 就会自动收集并运行它,不用注册。
  • assert total_cost(1_000_000, 0, 2.0, 8.0) == 2.0 代入:(1_000_000 × 2.0 + 0 × 8.0) / 1_000_000 = 2.0。一百万输入 token 正好 2 元,和单价对得上。
  • assert total_cost(0, 500_000, 2.0, 8.0) == 4.0 代入:(0 + 500_000 × 8.0) / 1_000_000 = 4.0。这两个结果都是整数,浮点数能精确表示,所以可以直接用 ==。比较 0.1 + 0.2 这种小数时要改用 pytest.approx。

uv run pytest -q 跑全部测试。测异步函数需要装 pytest-asyncio 或者 anyio 插件,今天先不碰,用 asyncio.run 包一下也能测。

为什么第二天就讲测试?因为 W5 的评测体系就是建在 pytest 上的。模型输出不确定,但你的工具函数、解析逻辑、成本计算都是确定的,这些必须有单元测试。面试官看你的 GitHub,有没有 tests/ 目录是第一眼就会注意的。

隔壁的程序员老王 · 从pip到uv:一口气梳理现代Python项目管理全流程! · 16 分钟,把 Python 依赖管理的历史和 uv 的用法讲得很清楚,今天必看。

日志:别只用 print

练习里用 print 没问题,但从 W2 开始 Agent 一次要跑十几步,满屏 print 会让你分不清哪行是模型说的、哪行是工具返回的、哪行是调试信息。标准库的 logging 够用了:

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s | %(message)s")
log = logging.getLogger("agent")

log.info("调用工具 %s 参数=%s", "get_invoice", {"invoice_id": "INV-1"})
log.warning("工具返回过长,已截断到 %d 字符", 2000)
  • logging.basicConfig(level=logging.INFO 全局配一次:INFO 及以上级别才输出;格式里的 %(asctime)s 是时间,%(levelname)s 是级别,%(name)s 是 logger 名,%(message)s 是正文。默认写到 stderr。
  • log = logging.getLogger("agent") 拿一个叫 agent 的 logger,同一个名字在哪里拿都是同一个对象。
  • log.info("调用工具 %s 参数=%s" 两个 %s 依次填 "get_invoice" 和那个 dict。实测输出:2026-10-22 07:04:58,033 INFO agent | 调用工具 get_invoice 参数={'invoice_id': 'INV-1'}
  • log.warning("工具返回过长,已截断到 %d 字符", 2000) %d 是整数占位。实测输出:2026-10-22 07:04:58,034 WARNING agent | 工具返回过长,已截断到 2000 字符。我顺手加了一行 log.debug(...),级别低于 INFO,果然什么也没打出来。

用 %s 占位而不是 f-string,是 logging 的惯例:日志级别不够时字符串根本不会被拼接,省一点开销。调试时把 level 改成 DEBUG,上线时改回 INFO。W5 会把日志升级成 trace,那时你会感谢现在养成的习惯。

上面的 getLogger("agent") 是手动起名。写在公共模块里时,惯用写法是:

log = logging.getLogger(__name__)

__name__ 是 Python 自动给每个文件的变量,值是模块名,比如在 common/structured.py 里就是 "common.structured"。用它当 logger 名,日志里(格式串的 %(name)s)就能看出是哪个文件打的,也能单独调某个模块的级别。分工记住一句:每个模块开头写 getLogger(__name__) 只管打日志,入口脚本里写一次 basicConfig 决定日志往哪输出、什么格式。顺带一提,文件直接运行时 __name__ 是 "__main__",所以脚本末尾的 if __name__ == "__main__": 意思是「只有直接运行这个文件才执行,被别人 import 时不执行」。

课程仓库的目录约定

从今天起,这门课的所有代码都放在一个仓库 agent-60-lab 里,每周一个目录。提前说好约定,后面的文章就不再重复:

agent-60-lab/
├── .env                 # 你的密钥,不提交
├── .env.example         # 变量清单,不含真实值,提交
├── pyproject.toml
├── uv.lock
├── common/              # Day 4 开始放跨周复用的代码,比如 LLMClient
├── w1/ … w6/            # 每周的练习和小项目
└── tests/               # 所有测试

几条习惯,越早养成越省事。每个练习都能用 uv run python w1/xxx.py 从仓库根目录直接跑,不要依赖「先 cd 到某个目录」。复用的代码放进 common/,别在每个文件里复制一遍客户端初始化。每完成一天提交一次 git,提交信息写清楚「Day 3: ask.py 支持多厂商切换」,几十个提交记录本身就是你学习过程的证据,面试时有人真的会点进去看。

写练习代码时可以用 AI 编程工具帮忙,怎么用、怎么审查它写的东西,下一节专门讲。

用 Coding Agent 学 Python、写代码,以及怎么审查它

这一节是根据 JD 调研加的。国内 120 条 Agent 岗位 JD 里有 26% 点名要求熟练使用 Claude Code、Cursor 或 Codex 这类 Coding Agent,有的写得很直白:「熟练使用 Codex / Claude Code / Cursor 等工具,并能把控生成代码的质量」。后半句才是重点。会让它生成代码谁都会,能判断它生成的东西对不对、该不该合进去,才是招聘方要的能力。

先说怎么拿它学 Python。对有经验的工程师,Coding Agent 是非常好的语言老师,因为你可以用自己熟悉的语言去问。几种我常用的问法:「这段 Python 如果用 Java 写是什么样,逐行对照着解释」;「async with 和 Go 的 defer 有什么异同」;「我写了一个函数,帮我指出哪里不够 Pythonic,但先别改,只列问题」。最后这种「只指出不改」的用法特别适合学习,自己动手改一遍印象深得多。

再说怎么用它写课程代码。我的建议是给它小而清楚的任务,一次一个函数或一个文件,而不是「帮我把今天的练习全写了」。任务越大,它越容易自作主张加一堆你没要的东西,你也越难审查。Claude Code 和 Codex 都支持在仓库根目录放一个说明文件(Claude Code 读 CLAUDE.md,Codex 读 AGENTS.md),把这门课的约定写进去,比如「用 uv 管依赖」「密钥只从环境变量读」「每个功能都要有 pytest 测试」「模型名从 LLM_MODEL 读,不要写死」,它每次都会照着做。这个说明文件本身就是一种提示词,Day 5 讲的原则全部适用。

最重要的是审查。它写完之后,我会按这张清单过一遍:

检查项为什么要查
能不能跑、测试过没过让它自己跑一遍 uv run pytest,别信「我已经写好了」
有没有加你没要求的依赖它很喜欢顺手装个库,每个新依赖都是以后的维护负担
模型名、API 写法是不是过时的它的训练数据有截止时间,经常写出 gpt-3.5-turbo 或者已经改名的参数
密钥有没有写死在代码里为了「先跑起来」,它有时会把 key 直接填进去
异常是不是被吞掉了except Exception: pass 是它最常见的坏习惯之一
async 函数里有没有同步阻塞调用比如在 async 函数里用 requests 或 time.sleep
改动范围是不是比任务大用 git diff --stat 看它动了哪些文件,多动的要问清楚为什么
你能不能逐行解释解释不了的那几行,就是面试时会被问倒的地方

最后一条最关键。我的标准是:合进仓库的每一行,你都要能讲清楚它在干什么、为什么这么写。做不到就让它解释,或者删掉自己重写。Agent 岗的面试官自己天天用这些工具,他们最擅长分辨「我理解这段代码」和「这段代码是 AI 写的,我也不太清楚」。

还有一个更深的理由要认真对待 Coding Agent:它本身就是目前最成熟的 Agent 产品。它的工作方式(读文件、调工具、跑命令、看结果、再改)和你 W2 要手写的 Agent 循环是同一个东西。用的时候多留意它在什么情况下会犯错:上下文太长时忘了前面的约定,工具用错了地方,多轮修改后越改越偏。这些失败模式 Day 12 会系统地讲,到时候你会发现自己早就见过了。

动手练习

写一个小程序:并发查询几个 GitHub 仓库的信息,用 pydantic 校验返回数据,统计耗时,再写测试。这正好是 Agent 里「并发调用多个工具、校验结果」的雏形。

第 1 步:加依赖。

cd agent-60-lab
uv add httpx pydantic
uv add --dev pytest
mkdir -p w1 tests

第 2 步:写主程序。

# agent-60-lab/w1/repo_stats.py
import asyncio
import time

import httpx
from pydantic import BaseModel, Field, ValidationError

REPOS = [
    "astral-sh/uv",
    "pydantic/pydantic",
    "encode/httpx",
    "openai/openai-python",
    "this-owner/does-not-exist-404",   # 故意放一个不存在的,练习错误处理
]


class RepoInfo(BaseModel):
    full_name: str
    stars: int = Field(alias="stargazers_count", ge=0)
    language: str | None = None
    open_issues: int = Field(alias="open_issues_count", ge=0)


async def fetch_repo(client: httpx.AsyncClient, repo: str) -> RepoInfo:
    r = await client.get(f"https://api.github.com/repos/{repo}")
    r.raise_for_status()
    return RepoInfo.model_validate(r.json())


async def fetch_all(repos: list[str]) -> list[RepoInfo | BaseException]:
    async with httpx.AsyncClient(timeout=15, headers={"User-Agent": "agent-60-lab"}) as client:
        return await asyncio.gather(*(fetch_repo(client, r) for r in repos), return_exceptions=True)


def summarize(results: list[RepoInfo | BaseException], repos: list[str]) -> tuple[list[RepoInfo], list[str]]:
    ok: list[RepoInfo] = []
    failed: list[str] = []
    for repo, res in zip(repos, results):
        if isinstance(res, RepoInfo):
            ok.append(res)
        elif isinstance(res, httpx.HTTPStatusError):
            failed.append(f"{repo}: HTTP {res.response.status_code}")
        elif isinstance(res, ValidationError):
            failed.append(f"{repo}: 返回数据不合法 {res.error_count()} 处")
        else:
            failed.append(f"{repo}: {type(res).__name__} {res}")
    return ok, failed


def main() -> None:
    start = time.perf_counter()
    results = asyncio.run(fetch_all(REPOS))
    ok, failed = summarize(results, REPOS)
    for info in sorted(ok, key=lambda x: x.stars, reverse=True):
        print(f"{info.full_name:<28} ⭐ {info.stars:>7}  {info.language or '-'}")
    for f in failed:
        print("失败:", f)
    print(f"耗时 {time.perf_counter() - start:.2f}s")


if __name__ == "__main__":
    main()
  • import asyncio 下面几行导入:asyncio 负责并发,time 负责计时,httpx 发请求,pydantic 那三个负责校验。标准库放一组,第三方库空一行另放一组,这是 Python 的习惯排法。
  • REPOS = [ 要查的 5 个仓库,写成 所有者/仓库名。
  • "this-owner/does-not-exist-404", 故意放一个不存在的,看程序遇到失败会不会整个崩掉。
  • class RepoInfo(BaseModel): GitHub 这个接口实测返回 86 个字段,我们只声明关心的 4 个。没声明的字段 pydantic 默认直接忽略,不报错。
  • stars: int = Field(alias="stargazers_count", ge=0) alias 的意思是:输入数据里这个字段叫 stargazers_count,到了模型里改名叫 stars。ge=0 要求不能是负数。
  • language: str | None = None 有的仓库 GitHub 识别不出主语言,返回 null,所以允许为空,默认也是 None。
  • open_issues: int = Field(alias="open_issues_count", ge=0) 同样是改名。注意 GitHub 的 open_issues_count 把还没合并的 PR 也算进去了。实测 uv 解析完是 {'full_name': 'astral-sh/uv', 'stars': 90999, 'language': 'Rust', 'open_issues': 2974}。
  • async def fetch_repo(client: httpx.AsyncClient, repo: str) -> RepoInfo: 查一个仓库:发请求、检查状态码、校验数据,三步。
  • r = await client.get(f"https://api.github.com/repos/{repo}") 代入 repo = "astral-sh/uv",请求的地址就是 https://api.github.com/repos/astral-sh/uv。
  • r.raise_for_status() 不存在的那个仓库会在这里抛异常,实测内容是 Client error '404 Not Found' for url 'https://api.github.com/repos/this-owner/does-not-exist-404'。
  • return RepoInfo.model_validate(r.json()) 把 86 个字段的 dict 交给 pydantic,只留下 4 个,并顺便把名字改掉。
  • async def fetch_all(repos: list[str]) -> list[RepoInfo | BaseException]: 返回值是一个列表,每个元素要么是查到的结果,要么是异常对象。下面正文解释了为什么写 BaseException。
  • async with httpx.AsyncClient(timeout=15, headers={"User-Agent": "agent-60-lab"}) as client: async with 保证结束时把连接池关掉。GitHub 要求请求必须带 User-Agent,所以这里加上。timeout=15 给慢一点的网络留足时间。
  • return await asyncio.gather(*(fetch_repo(client, r) for r in repos), return_exceptions=True) 从里往外拆:
    (fetch_repo(client, r) for r in repos) 是一个生成器,会产出 5 个协程;
    前面的 * 把它们展开成 5 个独立的参数,相当于 gather(协程1, 协程2, ..., 协程5);
    return_exceptions=True 让出错的那个不要把整批拖垮,而是把异常对象放进结果列表。
    实测返回列表里元素的类型依次是 RepoInfo, RepoInfo, RepoInfo, RepoInfo, HTTPStatusError,顺序和 REPOS 一一对应。
  • def summarize( 普通函数,不涉及网络,所以下面的测试可以直接测它。
  • ok: list[RepoInfo] = [] 两个空列表,一个装成功的,一个装失败的描述。
  • for repo, res in zip(repos, results): zip 把两个列表按位置配成对:第一轮是 ("astral-sh/uv", RepoInfo(...)),……,第五轮是 ("this-owner/does-not-exist-404", HTTPStatusError(...))。
  • if isinstance(res, RepoInfo): isinstance 判断对象是不是某个类的实例。前四轮都走这一支,放进 ok。
  • elif isinstance(res, httpx.HTTPStatusError): 第五轮走这里。res.response.status_code 是 404,于是得到 "this-owner/does-not-exist-404: HTTP 404"。如果被 GitHub 限流,这里会是 403。
  • elif isinstance(res, ValidationError): 数据格式不对的时候走这一支,比如字段缺失或者星数是负数。error_count() 是出错的字段数。这次实跑没有走到这里。
  • failed.append(f"{repo}: {type(res).__name__} {res}") 兜底分支:超时(ReadTimeout)、断网(ConnectError)之类都落在这里,type(res).__name__ 就是异常类名。
  • return ok, failed 返回一个二元组。实测 ok 里有 4 个,failed == ['this-owner/does-not-exist-404: HTTP 404']。
  • start = time.perf_counter() 从这里开始计时。
  • results = asyncio.run(fetch_all(REPOS)) main 是普通函数,要靠 asyncio.run 才能进入异步世界,跑完再回来。
  • ok, failed = summarize(results, REPOS) 解包二元组。
  • for info in sorted(ok, key=lambda x: x.stars, reverse=True): sorted 返回排好序的新列表,原列表不动。key=lambda x: x.stars 表示按星数排,reverse=True 表示从大到小。实测顺序:原来是 uv、pydantic、httpx、openai-python,排完变成 uv 90999 → openai-python 31905 → pydantic 29131 → httpx 15663。
  • print(f"{info.full_name:<28} ⭐ {info.stars:>7} {info.language or '-'}") :<28 左对齐并用空格补到 28 个字符宽,:>7 右对齐补到 7 宽:实测 f"{90999:>7}" 是 ' 90999'。info.language or '-' 的意思是:language 为 None 时显示 -。
  • print("失败:", f) 失败的那些单独打印出来,不会让程序崩溃。
  • print(f"耗时 {time.perf_counter() - start:.2f}s") 这次实跑的完整输出:
    astral-sh/uv ⭐ 90999 Rust
    openai/openai-python ⭐ 31905 Python
    pydantic/pydantic ⭐ 29131 Python
    encode/httpx ⭐ 15663 Python
    失败: this-owner/does-not-exist-404: HTTP 404
    耗时 3.64s
    国内访问 GitHub,单个请求就要两三秒。5 个并发后总耗时接近最慢的那一个,而不是 5 个相加。
  • if __name__ == "__main__": 直接运行这个文件才调 main();测试文件 import 它的时候不会去访问网络。
uv run python w1/repo_stats.py

返回值类型写的是 BaseException,不是 Exception,这是故意的。gather(..., return_exceptions=True) 在类型上声明的返回值是 list[结果 | BaseException],因为它连 KeyboardInterrupt、CancelledError 这类不属于 Exception 的异常也会收进列表。如果写成 Exception,Zed、VS Code 会在 return await asyncio.gather(...) 下面画红线,提示类型对不上。代码照样能跑,但标注应该说实话。

输出大致是:四行仓库信息按星数从高到低排列,一行「失败: this-owner/does-not-exist-404: HTTP 404」,最后是总耗时。GitHub 匿名访问每小时有 60 次的限额,频繁跑可能遇到 403,那也会被当作失败打印出来,不会让程序崩溃。

第 3 步:改成串行对比耗时。 把 fetch_all 里的 gather 改成一个 for 循环逐个 await,再跑一次,比较总耗时。差多少倍?记下来。如果差距没有你预期的大,想想是不是某个请求特别慢,把整体时间拖住了,这正是并发的上限所在。

第 4 步:写测试。 不访问网络,只测纯逻辑:

# agent-60-lab/tests/test_repo_stats.py
import httpx
import pytest
from pydantic import ValidationError

from w1.repo_stats import RepoInfo, summarize


def test_repo_info_parses_github_fields():
    info = RepoInfo.model_validate(
        {"full_name": "a/b", "stargazers_count": 10, "language": "Python", "open_issues_count": 2}
    )
    assert info.stars == 10
    assert info.open_issues == 2


def test_repo_info_rejects_negative_stars():
    with pytest.raises(ValidationError):
        RepoInfo.model_validate({"full_name": "a/b", "stargazers_count": -1, "open_issues_count": 0})


def test_summarize_separates_ok_and_failed():
    req = httpx.Request("GET", "https://example.invalid/repos/x/y")
    err = httpx.HTTPStatusError("not found", request=req, response=httpx.Response(404, request=req))
    good = RepoInfo.model_validate({"full_name": "a/b", "stargazers_count": 1, "open_issues_count": 0})
    ok, failed = summarize([good, err], ["a/b", "x/y"])
    assert ok == [good]
    assert failed == ["x/y: HTTP 404"]
  • from w1.repo_stats import RepoInfo, summarize 只导入两样纯逻辑的东西,测试全程不碰网络。能找到 w1 包,靠的是 w1/__init__.py 加上 pythonpath = ["."]。
  • def test_repo_info_parses_github_fields(): 测 alias 改名有没有生效。
  • {"full_name": "a/b", "stargazers_count": 10 输入用的是 GitHub 那边的字段名,所以 info.stars == 10、info.open_issues == 2 能成立,说明改名是对的。
  • def test_repo_info_rejects_negative_stars(): 测 ge=0 这条规则。
  • with pytest.raises(ValidationError): 意思是「这个代码块里必须抛出 ValidationError」:抛了,测试通过;没抛,测试失败。stargazers_count: -1 违反了 ge=0。顺便注意 language 没传也没关系,它有默认值 None。
  • req = httpx.Request("GET", "https://example.invalid/repos/x/y") 手工造一个请求对象,不会真的发出去。.invalid 是专门保留给测试用的域名,永远解析不到。
  • err = httpx.HTTPStatusError("not found", request=req, response=httpx.Response(404, request=req)) 造一个和真实 404 一模一样的异常对象,err.response.status_code 是 404。
  • good = RepoInfo.model_validate( 再造一个成功的结果。
  • ok, failed = summarize([good, err], ["a/b", "x/y"]) 逐轮代入:zip 第一轮 ("a/b", good) 走 isinstance(res, RepoInfo) 进 ok;第二轮 ("x/y", err) 走 HTTPStatusError 分支,得到 "x/y: HTTP 404"。
  • assert ok == [good] pydantic 模型的 == 按字段值比较。
  • assert failed == ["x/y: HTTP 404"] 失败描述的文本要一字不差。实测这个文件 3 passed in 0.18s。

为了让 from w1.repo_stats import ... 能找到模块,在项目根目录建一个空的 w1/__init__.py,并在 pyproject.toml 里加上:

[tool.pytest.ini_options]
pythonpath = ["."]
touch w1/__init__.py
uv run pytest -q

应该看到 3 passed。

第 5 步(建议做):让 Coding Agent 写一版再审查。 把这个练习的需求原样交给 Claude Code、Cursor 或 Codex,让它在另一个分支上写一版。然后用上面那张审查清单逐项过一遍,把发现的问题记下来,再和你自己写的版本对比。你很可能会发现它多装了依赖、或者把异常处理写得很粗糙。这份「AI 代码审查记录」留着,面试被问到「怎么用 AI 写代码」时就是现成的素材。

验收标准: 程序能并发跑完并正确打印失败项;你记录了串行和并发的耗时对比;3 个测试全部通过。

名词解释

名词意思
uvRust 写的 Python 项目管理工具,管 Python 版本、虚拟环境、依赖、锁文件
pyproject.tomlPython 项目的配置文件,声明依赖和工具配置
uv.lock锁定所有依赖精确版本的文件,提交进 git
虚拟环境(.venv)项目独立的 Python 和依赖目录,互不干扰
类型标注(type hints)给变量和函数标注类型的语法,运行时不检查
dataclass自动生成构造函数等方法的数据类,不做校验
pydantic带类型校验和转换的数据模型库,能导出 JSON Schema
协程(coroutine)async def 定义的函数,调用后要 await 才执行
事件循环调度协程的单线程循环,遇到 await 就切去执行别的协程
asyncio.gather并发运行多个协程并收集结果
httpx支持同步和异步的 HTTP 客户端
pytestPython 最常用的测试框架

常见误区

以为类型标注会在运行时报错。 不会。想要运行时检查用 pydantic,想要静态检查用 mypy 或 pyright。

在 async 函数里调用同步阻塞的库。 requests、time.sleep、同步版的 OpenAI 客户端都会卡死整个事件循环。异步代码要配异步库。

把 dataclass 当成校验工具。 dataclass 只是数据容器,外部数据进系统那一刻要用 pydantic 把关。

直接用 pip install 往系统 Python 里装包。 多个项目依赖冲突、换台电脑就跑不起来。全部用 uv 管,uv sync 一条命令复现环境。

用 os.getenv 读密钥。 变量没配置时 os.getenv 返回 None,程序继续往下跑,直到调用 API 时才报一个莫名其妙的 401 或者「api_key 不能为空」。用 os.environ["KEY"],启动时就报 KeyError: 'LLM_API_KEY',一眼知道少配了什么。

捕获异常后什么都不打印。 except Exception: pass 是排查问题的噩梦。Agent 系统本来就有很多不确定性,再把错误吞掉,你会花一下午去猜到底哪一步出了问题。至少打一行日志,写清楚是哪个工具、什么参数、什么错误。

在 Jupyter 里调用 asyncio.run。 Jupyter 自己已经有一个事件循环在跑,再调用会报「asyncio.run() cannot be called from a running event loop」。在 Notebook 里直接写 await main() 就行,在 .py 脚本里才用 asyncio.run(main())。

可变默认参数。 Python 经典坑,写 TS 的人尤其容易踩:

def add_tag(tag: str, tags: list[str] = []) -> list[str]:
    tags.append(tag)
    return tags

add_tag("a")   # ['a']
add_tag("b")   # ['a', 'b']   ← 上一次的 'a' 还在

原因是 Python 的默认值只在定义函数时算一次,那个 [] 从头到尾只有一个,每次调用拿到的都是同一个列表。TS 的 tags: string[] = [] 每次调用都会重新算,所以不会这样。list、dict、set 这种建好后还能改的「可变对象」都有这个问题;数字、字符串、None、tuple 改不了,当默认值没事。

正确写法:

def add_tag(tag: str, tags: list[str] | None = None) -> list[str]:
    if tags is None:
        tags = []              # 每次调用新建一个
    tags.append(tag)
    return tags

类里的字段也一样:

  • dataclass:写 tags: list[str] = [] 运行时直接报 ValueError: mutable default,必须写 field(default_factory=list)。default_factory 是「默认值的工厂函数」,每创建一个实例就调用一次 list(),得到一个新的空列表。
  • pydantic:直接写 = [] 其实没问题,pydantic 会给每个实例复制一份;但习惯上也写 Field(default_factory=list),和 dataclass 保持一致,不用记两套规则。

学习资料

文档与文章

B 站视频

码农高天 · 【python】asyncio的理解与入门,搞不明白协程?看这个视频就够了。 · 15 分钟。作者是 CPython 核心开发者,讲事件循环和 task 的关系讲得很透,看完再回来读上面的 async 一节。
隔壁的程序员老王 · [Python] Pydantic 让数据结构与验证融为一体 · 10 分钟,pydantic 基本用法,看完能直接上手今天的练习。
  • 武沛齐《python async 异步编程》(B 站搜标题,3 小时):想系统学异步的可以看,今天不要求。

其他经典视频

  • 这一天的内容 B 站资源已经足够。如果你更喜欢英文,YouTube 上搜「ArjanCodes pydantic」和「mCoding asyncio」,这两个频道讲 Python 工程实践很扎实。
  • DeepLearning.AI 短课《Pydantic for LLM Workflows》(课程页):Day 6 讲结构化输出时会用,今天可以先看前三节了解 pydantic 在大模型场景的用法。

书籍

  • 《流畅的 Python(第 2 版)》(Luciano Ramalho 著,人民邮电出版社,2023):有经验的程序员学 Python 最好的书。今天读第 5 章「数据类构建器」(讲 dataclass 等),异步那章(第 21 章)留到 W6。
  • 《Python 编程:从入门到实践(第 3 版)》(Eric Matthes 著,人民邮电出版社,2023):如果你完全没碰过 Python,先花两小时翻前 8 章。

开源项目与代码

  • openai/openai-python:OpenAI 官方 SDK 源码,看它怎么用 httpx 和 pydantic 构建的,是很好的 Python 工程范本。不用今天看,知道有这个参照。
  • pydantic/pydantic:今天练习抓的仓库之一。

自测

  1. pyproject.toml、uv.lock、.venv/ 三个哪些该提交进 git?
答案

pyproject.toml 和 uv.lock 提交;.venv/ 不提交。别人 clone 后运行 uv sync 会按 uv.lock 重新生成 .venv。

  1. 下面这段代码运行时会报错吗?为什么?
def add(a: int, b: int) -> int:
    return a + b

print(add("1", "2"))
答案

不会报错,输出 12。类型标注在运行时不做检查,两个字符串相加就是拼接。mypy 这类静态检查工具会提示类型不匹配。

  1. 一个 Agent 一步里要调用三个互不依赖的工具,各耗时 2 秒、3 秒、1 秒。串行和用 asyncio.gather 并发,总耗时大概各是多少?
答案

串行约 6 秒,并发约 3 秒(最慢的那个)。前提是三个工具都是真正的异步 IO,如果其中一个内部调用了同步阻塞函数,会拖住其他协程。

  1. 你在 async 函数里写了 requests.get(url),程序能跑,但并发效果完全没出来。为什么?
答案

requests 是同步阻塞的,调用时整个事件循环被卡住,其他协程无法执行,等于又变成了串行。应该换成 httpx.AsyncClient 并 await。

  1. 什么时候用 dataclass,什么时候用 pydantic?
答案

数据来自外部、不可信(HTTP 请求、模型输出、配置文件、第三方 API)时用 pydantic,它会校验并转换类型、给出清楚的错误。纯内部数据结构用 dataclass 就够,轻量、没有额外依赖。很多项目为了统一也全部用 pydantic。

  1. 面试题:「Python 有 GIL,为什么 Agent 服务还能处理很多并发请求?」
答案

GIL 限制的是同一时刻只有一个线程执行 Python 字节码,影响的是 CPU 密集任务的并行。Agent 服务的时间绝大部分在等网络(模型 API、数据库、外部工具),属于 IO 密集。用 asyncio 时,一个协程在等 IO 的时候事件循环会去执行其他协程,所以单进程也能同时挂起成百上千个请求。真正需要 CPU 并行的部分(比如本地跑 embedding 模型)可以用多进程或交给外部服务。

  1. 面试题:「你平时怎么用 Claude Code 或 Cursor 写代码?怎么保证它写的代码质量?」
答案

参考思路:给它小而明确的任务,一次一个函数或模块;在项目说明文件(CLAUDE.md / AGENTS.md)里写清楚约定(依赖管理、密钥、测试要求);让它先写测试或者写完必须跑通测试;合并前按清单审查,重点看新增依赖、过时的 API、写死的密钥、被吞掉的异常、超出任务范围的改动;每一行都要能自己解释。最好举一个真实的例子:它写错了什么,你是怎么发现的。

明天预告

Day 3 · 第一次调用大模型 API:OpenAI 兼容协议,一份代码切换 DeepSeek、通义和 Claude。把一次 chat completions 请求和响应逐字段拆开讲,弄清楚 usage 怎么换算成钱、401 和 429 分别意味着什么、为什么几乎所有国产模型都兼容 OpenAI 的格式。

评论

  1. 加载中…

只保存昵称和内容。会显示你的大致地区(国家 / 省),不保存 IP 地址。