书架 · Python 学习系列 · 03 · 工程化与异步下一章：04 LLM 应用开发 →

03 · 工程化与异步

本章目标：会用 uv 管项目、ruff 管规范、pytest 管测试、logging 管日志；搞懂 Python 三种并发模型，重点掌握 asyncio——AI 应用（并发调模型、流式输出）全靠它。

前置：学完 01、02。

目录

1. 依赖管理：venv/pip → uv
2. 代码规范：PEP 8 与 ruff
3. 测试：pytest
4. 日志：logging
5. 并发模型总览：线程 / 进程 / 协程
6. asyncio 核心语法
7. 异步 HTTP 客户端 httpx
8. 选型决策表
9. 自测清单
10. 小练习
11. 参考与出处

1. 依赖管理：venv/pip → uv

1.1 为什么需要虚拟环境

Python 的第三方包装进解释器的全局环境，多个项目依赖不同版本会互相打架（Java 有 Maven/Gradle 每个项目独立依赖，Python 的 venv 就是补这个的）。虚拟环境 = 一个项目一个独立的"site-packages 目录 + python 入口"。

1.2 传统三件套（看懂老项目用）

# 创建虚拟环境（在项目目录下）
python -m venv .venv

# 激活（PowerShell）
.venv\Scripts\Activate.ps1
# Git Bash
source .venv/Scripts/activate
# 之后 pip 安装的东西都只进这个环境

pip install requests               # 装包
pip freeze > requirements.txt      # 导出依赖清单（≈ mvn dependency:list）
pip install -r requirements.txt    # 别人拉项目后一键还原
deactivate                         # 退出虚拟环境

1.3 uv：现代方案（新项目用它）

uv（Astral 出品，Rust 编写）把 Python 版本管理、虚拟环境、依赖解析、锁定全合一，2024 年起成为社区主流，类比 Maven 之于 Java：

# 新建项目（生成 pyproject.toml + .venv + hello.py）
uv init my-app
cd my-app

# 加依赖（自动写入 pyproject.toml 并安装，≈ mvn install 加坐标）
uv add requests
uv add openai                     # AI 开发主力包
uv add "fastapi[standard]"        # 带可选依赖组的写法

# 移除
uv remove requests

# 同步环境（按锁文件精确还原，团队协作关键；≈ mvn install 还原锁版本）
uv sync

# 运行（自动确保在项目环境中，不需要手动激活）
uv run main.py
uv run pytest

# 临时跑个工具（不装进项目，≈ mvn exec）
uvx ruff check .

pyproject.toml（Python 版的 pom.xml）长这样：

[project]
name = "my-app"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "requests>=2.32",
    "openai>=2.0",
]

[dependency-groups]
dev = [
    "pytest>=8.0",
    "ruff>=0.6",
]

uv 会生成 uv.lock（≈ Maven 的依赖锁定，精确到哈希）——提交进 git，uv sync 保证队友和你环境完全一致。

☕ 给 Java 开发者的映射：pyproject.toml ≈ pom.xml；uv.lock ≈ 锁定版本；uv add ≈ 加依赖坐标；uv sync ≈ mvn install；uvx ≈ mvn exec。区别：Python 没有"中央仓库强制 group/artifact"，PyPI 上包名唯一即坐标。

📌 uv 是当前官方推荐工具链之一、pip 官方文档也已在入口指引"推荐使用 uv 管理环境"（出处见文末，访问日期 2026-09-05）。

2. 代码规范：PEP 8 与 ruff

PEP 8 是 Python 官方风格规范（官方中文之外另有专门页面，见文末）。核心几条先记住：

• 缩进 4 空格；命名：变量/函数 snake_case，类 PascalCase，常量 UPPER_CASE，内部用 _ 前缀
• 一行不超过 88~100 字符；import 顺序：标准库 → 第三方 → 本地，各自一组、字母排序
• 用 is None / is not None 判空

ruff 现在是事实上的 lint + format 一体化工具（替代 flake8/black/isort 的 Rust 重写）：

uv add --dev ruff          # 装进开发组

uv run ruff check .        # 检查（≈ checkstyle）
uv run ruff check --fix .  # 自动修
uv run ruff format .       # 格式化（≈ google-java-format）

VSCode 装 Ruff 扩展后保存即格式化，体验与 Java 生态无差。

3. 测试：pytest

pytest 是 Python 测试事实标准（ unittest 是标准库自带的 JUnit 风格框架，但社区几乎都用 pytest）：

# 文件名必须 test_ 开头或 _test 结尾，函数名必须 test_ 开头 —— 约定优于配置
from mymod import add, split_words

def test_add():
    assert add(1, 2) == 3          # ⭐ 用裸 assert，不用 assertEquals！

def test_split():
    assert split_words("a b") == ["a", "b"]

def test_raises():
    import pytest
    with pytest.raises(ValueError):        # 断言抛异常（≈ assertThrows）
        int("abc")

uv run pytest              # 跑全部
uv run pytest -v           # 详细
uv run pytest tests/test_mymod.py::test_add   # 跑单个

参数化 + fixture（≈ JUnit 的 @ParameterizedTest + @BeforeEach）：

import pytest

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (0, 0, 0),
    (-1, 1, 0),
])
def test_add_many(a, b, expected):
    assert add(a, b) == expected

@pytest.fixture
def sample_messages():            # fixture：测试前的准备数据
    return [{"role": "user", "content": "hi"}]

def test_len(sample_messages):    # 参数名 = fixture 名，自动注入
    assert len(sample_messages) == 1

def test_call_api(monkeypatch):   # monkeypatch：打桩/替换（≈ Mockito）
    monkeypatch.setattr("mymod.call_api", lambda: "fake")
    assert mymod.use_api() == "fake"

4. 日志：logging

import logging

# 模块级标准姿势：logger 名 = 模块名（≈ slf4j 的 LoggerFactory.getLogger）
logger = logging.getLogger(__name__)

def main():
    logging.basicConfig(                      # 简单场景：一次配置全局
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    logger.info("开始处理 %d 条", 42)          # 占位符，不要 f-string（惰性格式化省开销）
    logger.warning("配置缺失，使用默认值")
    logger.exception("处理失败")               # 自动带堆栈（≈ log.error("...", e)）

main()

☕ 对照：logging ≈ slf4j + logback 合体（标准库自带，无桥接层）；logger 层级继承 ≈ logger name 继承；生产上 uvicorn/FastAPI 各自有 logger，可按 "uvicorn"、"uvicorn.access" 名字调级别。四条军规：不用 print 打日志、占位符不用 f-string、logger.exception 记异常、模块级 getLogger(__name__)。

5. 并发模型总览：线程 / 进程 / 协程

⚠️ 先复习第 02 章的 GIL：CPython 同一时刻只有一个线程执行字节码（free-threaded 构建仍是实验性，主流部署带 GIL，见 What's New 3.13）。

线程与进程的朴素示例：

# 线程：ThreadPoolExecutor ≈ Java Executors.newFixedThreadPool
from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch(url: str) -> str:
    return f"{url} 的模拟响应"          # 真实项目里换成 requests.get(url).text

urls = ["https://a.com", "https://b.org", "https://c.net"]

with ThreadPoolExecutor(max_workers=8) as pool:
    futures = [pool.submit(fetch, u) for u in urls]
    for fut in as_completed(futures):
        print(fut.result())

6. asyncio 核心语法

6.1 心智模型

☕ 给 Java 开发者的一句话：asyncio ≈ 单线程版的 CompletableFuture + Netty 式事件循环。async def 定义协程（不调用不执行），await 等待期间把线程让给别的协程——所以单线程能并发跑几千个网络请求。跟 Java 虚拟线程（Loom）解决的是同一类问题，只是显式 await 标记。

import asyncio

async def fetch_data(name: str, delay: float) -> str:   # async def = 协程函数
    await asyncio.sleep(delay)        # await：非阻塞等待（await 期间让出线程）
    return f"{name} 完成"

async def main():
    # 串行：总耗时 = 1 + 2 = 3 秒
    a = await fetch_data("A", 1)
    b = await fetch_data("B", 2)

    # ⭐ 并发：gather 同时启动，总耗时 = max(1, 2) = 2 秒
    a, b = await asyncio.gather(
        fetch_data("A", 1),
        fetch_data("B", 2),
    )

# 协程必须由事件循环驱动（Python 3.7+）：
asyncio.run(main())

6.2 三条铁律（新手 90% 的 async 报错都来自这）

# 1. async def 函数调用返回的是协程对象，不 await 就不会执行
async def task(): ...
task()                # ❌ 什么都没发生（还会有 warning）
await task()          # ✅

# 2. await 只能出现在 async def 内部；普通函数里没法 await
def normal():
    await task()      # ❌ SyntaxError

# 3. 别在异步代码里调用阻塞函数（会卡死整个事件循环！）
async def bad():
    time.sleep(1)                 # ❌ 阻塞：所有协程全停
    requests.get(url)             # ❌ 同上

async def good():
    await asyncio.sleep(1)        # ✅
    # 同步重活实在要用：丢进线程池
    await asyncio.to_thread(time.sleep, 1)      # ✅

用 openai SDK 时同理：AsyncOpenAI 客户端配 await client.chat.completions.create(...)；同步 OpenAI 客户端在 async 函数里直接调用会卡住事件循环（FastAPI 里这么写，所有请求一起卡）。

6.3 常用工具

import asyncio

async def demo() -> None:
    # ① 超时控制（1 秒拿不到就算了）
    try:
        async with asyncio.timeout(1):
            await asyncio.sleep(10)          # 模拟慢任务
    except TimeoutError:
        print("超时，放弃")

    # ② 并发限流（同时最多 2 个——调 LLM API 必备，防限流）
    sem = asyncio.Semaphore(2)
    async def call(i: int) -> str:
        async with sem:
            await asyncio.sleep(0.1)
            return f"任务{i}完成"
    print(await asyncio.gather(*[call(i) for i in range(5)]))

    # ③ 任务组（3.11+，结构化并发，异常自动传播）
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(asyncio.sleep(0.1, "A"))
        t2 = tg.create_task(asyncio.sleep(0.1, "B"))
    print(t1.result(), t2.result())          # 出了 with 块任务都已完成

    # ④ 生产者-消费者队列
    queue: asyncio.Queue[str] = asyncio.Queue()
    await queue.put("消息")
    print(await queue.get())

asyncio.run(demo())

7. 异步 HTTP 客户端 httpx

httpx ≈ "异步版 requests"，API 几乎同形（openai SDK 底层就是它）：

import httpx

# 同步用法（跟 requests 一样）
resp = httpx.get("https://httpbin.org/get", params={"q": "py"})
resp.status_code
resp.json()

# ⭐ 异步用法（AI 服务并发调用主力）
import asyncio

async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10) as client:      # 复用连接池
        tasks = [client.get(u) for u in urls]
        resps = await asyncio.gather(*tasks)
        return [r.json() for r in resps]

asyncio.run(fetch_all(urls))

# POST JSON（调内部服务/大模型网关的姿势）
async with httpx.AsyncClient() as client:
    r = await client.post(
        "https://api.example.com/chat/completions",
        headers={"Authorization": "Bearer <key>"},
        json={"model": "deepseek-v4-pro", "messages": [...]},
    )
    r.raise_for_status()

☕ ≈ Java 的 WebClient/OkHttp：AsyncClient ≈ 连接池复用客户端；timeout/重试/拦截器（event_hooks）都有。在 asyncio 项目里禁用 requests 库（它是纯阻塞的）。

8. 选型决策表

9. 自测清单

• [ ] uv add / uv sync / uv run / uv.lock 分别对应 Maven 的什么？
• [ ] pytest 怎么发现测试文件和函数？断言用什么关键字？
• [ ] logging.basicConfig 在哪调用一次？为什么日志不用 f-string？
• [ ] GIL 为什么不让 threading 加速 CPU 密集任务？
• [ ] async def 函数直接调用会发生什么？
• [ ] asyncio.gather 三个 1 秒的任务总耗时多少？
• [ ] 在 async 函数里用 requests.get 会怎样？正确做法？
• [ ] asyncio.Semaphore 在批量调 LLM 时管什么用？

10. 小练习

练习 1（uv）：新建项目 async-demo，加 httpx，写脚本并发抓取 5 个 URL 的状态码，打印总耗时；改成串行对比耗时。

练习 2（pytest）：给第 02 章练习 1 的 Conversation 类写 5 个测试：新增消息、超长淘汰（maxlen）、total_tokens 累计、空对话行为、add 非法 role 抛异常。

练习 3（asyncio）：写 async def call_llm(prompt) -> str（内部 await asyncio.sleep(1) 模拟），并用 Semaphore(3) 限流并发跑 10 个 prompt，验证同时最多 3 个在飞。

练习 4（综合）：给一个"并发 + 重试 + 超时"的 call_with_retry(coro_factory, times=3, timeout=5) 通用函数写实现（asyncio.timeout + for 循环重试）。

11. 参考与出处

以下均为官方一手来源（访问日期：2026-09-05）：

⬅️ 返回目录 ｜ ➡️ 下一章：04-LLM应用开发