14 KiB
14 KiB
转岗计划 · 第 8 周学习计划（Python 速成：给 JS 开发者）
所属阶段：第 2 个月最后一周，为第 9~11 周的 Python 迷你多 Agent 项目做准备。总纲见同目录
README.md，项目计划见agent-mini-project.md。
目标：一周内把写 Agent 服务需要的 Python 补齐：基础语法、类型标注、asyncio、pydantic、httpx、pytest、FastAPI 和 SSE。不追求全面，只学项目里马上会用到的。
学法：你有 7 年 JS 和一个月 Go 的底子，每个知识点都先想“JS / Go 里对应的是什么”，再看 Python 的写法。每天的练习文件第一段写一张小对照表。
本周只学 Python，不开 Agent 项目（戒律：同一周不同时开新框架和新功能）。
参考资料：
- Python 官方教程（中文）：https://docs.python.org/zh-cn/3/tutorial/
- asyncio 官方文档（中文）：https://docs.python.org/zh-cn/3/library/asyncio.html
- uv：https://docs.astral.sh/uv/
- pydantic：https://docs.pydantic.dev/
- FastAPI（中文）：https://fastapi.tiangolo.com/zh/
- httpx：https://www.python-httpx.org/
最终项目结构
py-learning/
├── pyproject.toml # uv 管理依赖
├── .python-version
├── day1_basics.py
├── day2_typing.py
├── day3_errors_files.py
├── day4_async.py
├── day5_models.py
├── day6_http.py
├── fastapi_demo/
│ ├── __init__.py
│ ├── main.py # FastAPI 应用
│ └── static/
│ └── sse.html # 用 EventSource 订阅 SSE 的测试页
└── tests/
├── test_day2.py
├── test_day5.py
├── test_day6.py
└── test_fastapi_demo.py
第 1 天：环境和基础语法
学习内容
- 用 uv 管理 Python 版本和依赖（相当于
nvm + npm合在一起）：curl -LsSf https://astral.sh/uv/install.sh | sh。 - 缩进代替大括号；
None/True/False；list、dict、tuple、set。 - f-string（相当于模板字符串）、切片
a[1:3]、解包a, *rest = items。 - 列表推导式和字典推导式（相当于
map+filter）。 - 函数：默认参数、关键字参数、
*args、**kwargs；默认参数不要用可变对象（def f(x=[])是经典坑）。 if __name__ == "__main__":的作用。
要建的文件
py-learning/：运行uv init py-learning && cd py-learning && uv python pin 3.12，生成pyproject.toml和.python-version。day1_basics.py：- 文件开头用注释写一张 JS ↔ Python 对照表（至少 10 行：
const/let、===、null/undefined、arr.map、arr.filter、对象展开、解构、模板字符串、for...of、箭头函数）。 def word_count(text: str) -> dict[str, int]：和第 1 周 Go 的WordCount做同一件事，对比代码长短。def unique(items: list[str]) -> list[str]：去重并保持原顺序。- 演示可变默认参数的坑和正确写法（
def f(x=None)）。
- 文件开头用注释写一张 JS ↔ Python 对照表（至少 10 行：
运行方式：uv run python day1_basics.py
完成标准：能说清楚 Python 的 list 和 tuple 区别，以及为什么默认参数不能用 []。
第 2 天：类型标注、dataclass 和 Protocol
学习内容
- 类型标注只是提示，运行时不检查（和 TS 一样），要靠检查工具：
uv add --dev pyright ruff pytest。 list[str]、dict[str, int]、X | None、Literal["a", "b"]、TypedDict。@dataclass：相当于带构造函数的结构体。Protocol：Python 的“隐式接口”，和 Go 的interface思路一样，Agent 项目里 LLM 客户端和存储层都会用它。- 类、
self、@property、@classmethod。
要建的文件
day2_typing.py：TaskStatus = Literal["todo", "doing", "done"]@dataclass class Task：id: int、title: str、status: TaskStatus = "todo"、tags: list[str] = field(default_factory=list)。class TaskStore(Protocol)：add(self, task: Task) -> None、list(self) -> list[Task]。class MemoryTaskStore：不继承TaskStore，但实现同样的方法（体会“隐式实现”）。def count_by_status(tasks: list[Task]) -> dict[TaskStatus, int]
tests/test_day2.py：测试count_by_status；用pytest.mark.parametrize写表格驱动测试（对应 Go 的表格驱动测试）。
运行方式：uv run pytest -v、uv run pyright day2_typing.py、uv run ruff check .
完成标准：故意把 status 写成 "deleted"，pyright 能报错；能说清楚 Protocol 和 Go 接口的相同点。
第 3 天：异常、上下文管理器和文件
学习内容
try / except / else / finally，raise ... from err（相当于 Go 的%w包装错误）。- 自定义异常层级：一个基类，下面分具体错误。
with语句（上下文管理器）：自动关闭文件、连接，相当于 Go 的defer。pathlib.Path、json.loads/json.dumps（注意ensure_ascii=False才能正常输出中文）。logging基本用法。
要建的文件
day3_errors_files.py：class AppError(Exception)，带code: str属性；子类NotFoundError、ValidationError。def load_tasks(path: Path) -> list[dict]：文件不存在返回空列表；JSON 格式错误时raise ValidationError(...) from err。def save_tasks(path: Path, tasks: list[dict]) -> None：先写临时文件再replace，避免写到一半崩溃导致文件损坏（Agent 项目里“写状态要原子”就是这个道理）。- 用
contextlib.contextmanager写一个timer(name)，在with timer("load"):结束时打印耗时。
运行方式：uv run python day3_errors_files.py
完成标准：能说清楚 raise X from err 保留了什么信息，以及“先写临时文件再改名”为什么更安全。
第 4 天：asyncio（本周最重要的一天）
学习内容
async def/await和 JS 很像，但要有事件循环：入口用asyncio.run(main())。- 和 JS 的区别：调用
async函数只得到协程对象，不会自动开始执行，要await或asyncio.create_task。 asyncio.gather（相当于Promise.all）、asyncio.wait_for/asyncio.timeout（超时）、asyncio.Queue（相当于 Go 的 channel）、asyncio.Event。- 取消：
task.cancel()会在任务里抛出CancelledError，要在finally里清理，不要吞掉这个异常。 - 异步生成器
async def ... yield，以及async for：SSE 推送就是靠它。 - 不能在
async函数里调用会阻塞的同步代码（比如time.sleep、同步的数据库驱动），否则整个服务卡住；实在要用就await asyncio.to_thread(...)。
要建的文件
day4_async.py：async def fake_llm(name: str, delay: float) -> str：await asyncio.sleep(delay)后返回结果，模拟一次模型调用。async def run_parallel()：用gather同时调 3 个fake_llm，打印总耗时，和顺序调用对比。async def run_with_timeout()：给一个 3 秒的调用设 1 秒超时，捕获TimeoutError并打印。async def producer_consumer()：一个生产者往Queue放 5 个任务，两个消费者处理，全部处理完后退出（queue.join()）。async def ticker(n: int)：异步生成器，每 0.5 秒yield一个数字；用async for消费，消费到第 3 个时取消，观察finally是否执行。- 文件开头写对照表：
Promise.all↔gather、setTimeout↔asyncio.sleep、Go 的go f()↔create_task、Go 的 channel ↔Queue、Go 的context取消 ↔task.cancel()。
运行方式：uv run python day4_async.py
完成标准：能说清楚“调用 async 函数但没 await”会发生什么；能解释为什么在 async 服务里用同步阻塞调用会拖垮所有请求。
第 5 天：pydantic 数据模型
学习内容
- pydantic 是“带运行时校验的类型”，作用类似前端的 zod：外部进来的数据（请求体、模型输出的 JSON）都要先过一遍校验。
BaseModel、Field(min_length=..., max_length=...)、默认值、model_validate、model_validate_json、model_dump、model_dump_json。Literal字段和“按某个字段区分类型”的联合类型（discriminated union）。ValidationError里能拿到哪个字段、为什么错。pydantic-settings：从环境变量和.env读配置。
要建的文件
uv add pydantic pydantic-settingsday5_models.py：class Decision(BaseModel)：action: Literal["dispatch", "finish", "ask_human"]、target: str | None = None、instruction: str = ""、reason: str。（这就是第 9 周协调者输出的雏形）def parse_decision(raw: str) -> Decision：模型常常会在 JSON 外面包一层```json代码块或多说几句话，先提取出第一个{...}再model_validate_json；失败时抛出自定义的BadModelOutput，错误信息里带上原始文本的前 200 个字。class Settings(BaseSettings)：llm_base_url: str、llm_api_key: str = ""、llm_model: str、llm_timeout_s: float = 30，从.env读取。
tests/test_day5.py：参数化测试parse_decision：正常 JSON、带代码块的 JSON、前后有废话的 JSON、action拼错、缺少reason、完全不是 JSON。
运行方式：uv run pytest tests/test_day5.py -v
完成标准：6 种输入都按预期通过或报错；能说清楚“为什么模型输出必须当成不可信的外部输入来校验”。
第 6 天：httpx 异步请求和测试
学习内容
httpx.AsyncClient：设置base_url、timeout、请求头；用async with管理生命周期，复用连接。- 状态码和异常：
raise_for_status()、httpx.TimeoutException、httpx.ConnectError。 - 测试时不真的发请求：
httpx.MockTransport可以按请求返回任意响应，用来模拟超时、429、500。 - pytest 跑异步测试：
uv add --dev pytest-asyncio，在pyproject.toml里设置asyncio_mode = "auto"。
要建的文件
uv add httpxday6_http.py：class HttpError(Exception)，带code和retryable: bool。async def get_json(client: httpx.AsyncClient, url: str, retries: int = 2) -> dict：超时和 5xx、429 时等待后重试（等待时间逐次翻倍），401 / 400 不重试，最后抛出带错误码的HttpError。main()里请求一个公开接口（例如https://httpbin.org/json）打印结果。
tests/test_day6.py，用MockTransport测试：- 200 直接返回。
- 前两次 503、第三次 200，最终成功，并断言一共请求了 3 次。
- 401 不重试，直接抛
code == "AUTH"。 - 一直超时，最终抛
code == "TIMEOUT"。
运行方式：uv run python day6_http.py、uv run pytest tests/test_day6.py -v
完成标准：四个测试都通过；能说清楚哪些错误值得重试、哪些不值得。
第 7 天：FastAPI 基础和 SSE
学习内容
- FastAPI 路由、路径参数、查询参数，请求体直接用 pydantic 模型（对照第 2 周的 Gin）。
- 依赖注入
Depends。 lifespan：服务启动和关闭时执行的代码（第 10 周用它在启动时恢复未完成的任务）。- 后台任务：用
asyncio.create_task启动，要把任务对象保存起来，否则可能被垃圾回收。 - SSE 格式：
id: 1\nevent: progress\ndata: {...}\n\n；用StreamingResponse(生成器, media_type="text/event-stream")返回；浏览器用EventSource订阅，断线会自动重连并带上Last-Event-ID请求头。 - 检测客户端断开：
await request.is_disconnected()。
要建的文件
uv add fastapi uvicornfastapi_demo/main.py：GET /health。POST /jobs：请求体{"n": 10}，用create_task启动一个后台任务，每秒把进度追加到内存里的jobs[job_id]["events"]列表，返回{"job_id": ...}。GET /jobs/{job_id}：返回当前状态和已产生的事件数。GET /jobs/{job_id}/events：SSE。先补发Last-Event-ID之后的事件，再等待新事件，任务完成后发event: done并结束；每 15 秒没有新事件就发一行: ping心跳。- 关键实验：浏览器关掉后，后台任务要继续跑完（因为任务和 SSE 连接是分开的）。
fastapi_demo/static/sse.html：一个按钮启动任务，用EventSource显示进度。tests/test_fastapi_demo.py：用fastapi.testclient.TestClient测/health和创建任务。
运行方式：uv run uvicorn fastapi_demo.main:app --reload，浏览器打开测试页；或者 curl -N localhost:8000/jobs/<id>/events。
完成标准
- 启动任务后关掉页面，再打开、重新订阅，能看到任务已经继续往前跑，而且能补齐中间漏掉的事件。
- 能说清楚“编排为什么不能只活在浏览器里”——这就是重点 C，也是第 10 周要做的核心功能的缩小版。
uv run ruff check . && uv run pytest通过，提交并推送到zhuangang仓库（路径py-learning/）。- 在总纲
README.md的进度记录里勾掉第 8 周。
本周自查清单
- 会用 uv 建项目、加依赖、跑脚本
- 熟悉列表推导式、解包、f-string，知道可变默认参数的坑
- 会写类型标注、dataclass、Protocol，会用 pyright 检查
- 会用自定义异常和
raise ... from - 理解 asyncio：协程、
create_task、gather、超时、取消、Queue、异步生成器 - 会用 pydantic 校验外部数据（包括模型输出）
- 会用 httpx 发异步请求、按错误类型决定是否重试，会用
MockTransport测试 - 会写 FastAPI 接口和 SSE，理解后台任务和 SSE 连接是两回事
下周预告：第 9 周开始迷你多 Agent 项目，先做 M1（LLM 调用封装）和 M2（多轮编排循环），见 agent-mini-project.md。