今天要解决的问题
Agent 岗的 JD 十份里有九份写 Python。原因不复杂:各家模型的 SDK、LangGraph、MCP、评测工具,第一个支持的都是 Python。你用 Java 或 Go 也能做 Agent(Spring AI、LangChain4j 都有),但生态和资料差一截,面试时也难免被问「Python 熟不熟」。
好消息是,对有经验的工程师来说,Python 一天就能上手到「能写 Agent」的程度。今天不讲语法大全,只讲后面 58 天会反复用到的东西。学完你应该能:
- 用 uv 建项目、加依赖、跑脚本,知道
pyproject.toml和uv.lock各是干什么的。 - 读懂和写出带类型标注的 Python,知道类型标注在运行时不做检查。
- 分清 dataclass 和 pydantic 的用途,用 pydantic 校验外部数据。
- 用 async/await 和
asyncio.gather并发发请求,知道它和 Java 线程、Go goroutine 的区别。 - 用 pytest 写和跑测试。
- 用 Coding Agent 辅助学习和写代码,并能按一张清单审查它生成的代码。
预计用时:4 到 5 小时。如果你已经熟悉 Python,快速扫一遍「常见误区」和练习就行。
如果你是前端出身,今天会比后端同学轻松一些:Python 的 async/await 和 JS 几乎是同一套心智模型,pydantic 的角色和 zod 很像,uv 的体验也接近 pnpm。要多花时间的反而是类型标注和测试的习惯。如果你是 Java 出身,最不习惯的大概是「没有编译期检查」,代码写错了要跑到那一行才知道。这也是为什么我们第二天就把测试和类型标注一起讲。
先建立对照表
把你熟悉的东西映射过来,学得最快:
| 你熟悉的 | Python 里对应的 | 一句话说明 |
|---|---|---|
| Maven / Gradle / npm / go mod | uv(+ pyproject.toml) | 管依赖、管虚拟环境、管 Python 版本 |
package.json / pom.xml | pyproject.toml | 项目元数据和依赖声明 |
package-lock.json / go.sum | uv.lock | 锁定精确版本,要提交进 git |
node_modules | .venv/ 目录 | 每个项目独立的依赖环境,不提交 |
| Java POJO / Go struct | @dataclass | 纯数据容器 |
| Bean Validation / zod | pydantic | 带校验的数据模型 |
CompletableFuture / goroutine / Promise | async def + await + asyncio | 单线程事件循环的并发 |
| OkHttp / net/http / axios | httpx | HTTP 客户端,同步异步都支持 |
| JUnit / go test / Jest | pytest | 测试框架 |
.env + dotenv | python-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 TicketpriorityInput 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 密集的程序,这正好够用。
最小例子:
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 CallableCallable是「可以调用的东西」这个类型,函数、带__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 记:
| Python | JS | 例子 |
|---|---|---|
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 s | s.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 只有两个等号,没有三等号:
| 意思 | Python | JS |
|---|---|---|
| 值相等 / 不等 | a == b / a != b | a === 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/ 目录是第一眼就会注意的。
日志:别只用 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 Rustopenai/openai-python ⭐ 31905 Pythonpydantic/pydantic ⭐ 29131 Pythonencode/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 个测试全部通过。
名词解释
| 名词 | 意思 |
|---|---|
| uv | Rust 写的 Python 项目管理工具,管 Python 版本、虚拟环境、依赖、锁文件 |
| pyproject.toml | Python 项目的配置文件,声明依赖和工具配置 |
| uv.lock | 锁定所有依赖精确版本的文件,提交进 git |
| 虚拟环境(.venv) | 项目独立的 Python 和依赖目录,互不干扰 |
| 类型标注(type hints) | 给变量和函数标注类型的语法,运行时不检查 |
| dataclass | 自动生成构造函数等方法的数据类,不做校验 |
| pydantic | 带类型校验和转换的数据模型库,能导出 JSON Schema |
| 协程(coroutine) | async def 定义的函数,调用后要 await 才执行 |
| 事件循环 | 调度协程的单线程循环,遇到 await 就切去执行别的协程 |
| asyncio.gather | 并发运行多个协程并收集结果 |
| httpx | 支持同步和异步的 HTTP 客户端 |
| pytest | Python 最常用的测试框架 |
常见误区
以为类型标注会在运行时报错。 不会。想要运行时检查用 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 保持一致,不用记两套规则。
学习资料
文档与文章
- uv 文档:重点看 Getting started 和 Working on projects 两节。30 分钟。
- pydantic 文档:Models:看 Basic model usage 和 Validating data 两部分。30 分钟。
- Python 官方 asyncio 文档:当参考手册查,不用通读。
- FastAPI 文档:并发和 async/await(中文):FastAPI 作者用汉堡店排队的比喻讲异步,是我见过最好懂的解释,W6 也会用到 FastAPI。必看,20 分钟。
- httpx 文档:看 QuickStart 和 Async Support。
- pytest 文档:Get Started 一页就够。
- Learn X in Y minutes: Python:一页纸过完 Python 语法,适合有其他语言经验的人当速查表。
- Claude Code 文档、OpenAI Codex 文档、Cursor 文档:三个主流 Coding Agent 的官方文档,挑你用的那个看入门部分,重点找「项目说明文件」(CLAUDE.md / AGENTS.md / rules)怎么写。
- AGENTS.md:一个给 Coding Agent 写项目说明的通用约定,多家工具都支持,10 分钟看完。
- Python 官方教程(中文):语法细节查漏补缺。
B 站视频
- 武沛齐《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:今天练习抓的仓库之一。
自测
pyproject.toml、uv.lock、.venv/三个哪些该提交进 git?
答案
pyproject.toml 和 uv.lock 提交;.venv/ 不提交。别人 clone 后运行 uv sync 会按 uv.lock 重新生成 .venv。
- 下面这段代码运行时会报错吗?为什么?
def add(a: int, b: int) -> int:
return a + b
print(add("1", "2"))
答案
不会报错,输出 12。类型标注在运行时不做检查,两个字符串相加就是拼接。mypy 这类静态检查工具会提示类型不匹配。
- 一个 Agent 一步里要调用三个互不依赖的工具,各耗时 2 秒、3 秒、1 秒。串行和用
asyncio.gather并发,总耗时大概各是多少?
答案
串行约 6 秒,并发约 3 秒(最慢的那个)。前提是三个工具都是真正的异步 IO,如果其中一个内部调用了同步阻塞函数,会拖住其他协程。
- 你在 async 函数里写了
requests.get(url),程序能跑,但并发效果完全没出来。为什么?
答案
requests 是同步阻塞的,调用时整个事件循环被卡住,其他协程无法执行,等于又变成了串行。应该换成 httpx.AsyncClient 并 await。
- 什么时候用 dataclass,什么时候用 pydantic?
答案
数据来自外部、不可信(HTTP 请求、模型输出、配置文件、第三方 API)时用 pydantic,它会校验并转换类型、给出清楚的错误。纯内部数据结构用 dataclass 就够,轻量、没有额外依赖。很多项目为了统一也全部用 pydantic。
- 面试题:「Python 有 GIL,为什么 Agent 服务还能处理很多并发请求?」
答案
GIL 限制的是同一时刻只有一个线程执行 Python 字节码,影响的是 CPU 密集任务的并行。Agent 服务的时间绝大部分在等网络(模型 API、数据库、外部工具),属于 IO 密集。用 asyncio 时,一个协程在等 IO 的时候事件循环会去执行其他协程,所以单进程也能同时挂起成百上千个请求。真正需要 CPU 并行的部分(比如本地跑 embedding 模型)可以用多进程或交给外部服务。
- 面试题:「你平时怎么用 Claude Code 或 Cursor 写代码?怎么保证它写的代码质量?」
答案
参考思路:给它小而明确的任务,一次一个函数或模块;在项目说明文件(CLAUDE.md / AGENTS.md)里写清楚约定(依赖管理、密钥、测试要求);让它先写测试或者写完必须跑通测试;合并前按清单审查,重点看新增依赖、过时的 API、写死的密钥、被吞掉的异常、超出任务范围的改动;每一行都要能自己解释。最好举一个真实的例子:它写错了什么,你是怎么发现的。
明天预告
Day 3 · 第一次调用大模型 API:OpenAI 兼容协议,一份代码切换 DeepSeek、通义和 Claude。把一次 chat completions 请求和响应逐字段拆开讲,弄清楚 usage 怎么换算成钱、401 和 429 分别意味着什么、为什么几乎所有国产模型都兼容 OpenAI 的格式。
评论