迁移心智模型用“相同工程责任,不同运行语义”学习 Python:类型提示不等于运行时校验,Coroutine 不等于 Task,取消是异常传播,Pydantic Model 不等于领域对象,ASGI Worker 也不拥有持久业务状态。

建立与 TypeScript 的关键映射

从 TypeScript 概念映射到 Python,但不要假设语义完全相同
TypeScriptPython关键差异
interface / structural typingProtocol类型检查器配置和 Runtime Check 各自独立
type / discriminated unionTypeAlias / Literal Union运行时仍需 Pydantic/手工解析
class / objectdataclass / class可变默认值、比较和序列化需显式设计
Promisecoroutine / Task调用 async 函数先得到 Coroutine,调度后才执行
try/finally / disposablecontext manager / async context managerwith/async with 负责确定释放时机
package.json lockpyproject.toml + lock工具不同,但生产必须锁定解析结果

使用 Protocol 描述 Model/Tool/Store 端口,用 dataclass 表达内部值对象,用 Pydantic 解析边界 DTO。模块顶层代码在 import 时执行,Composition Root 应避免导入就建立网络连接或读取难以覆盖的全局配置。

虚拟环境只隔离安装位置,Lockfile 才固定解析依赖;镜像构建还要锁定 Python 版本、平台 Wheel 和系统库。类型检查器(mypy 或 pyright)与配置必须进 CI,否则类型提示只是编辑器注释。

理解 asyncio 的执行模型

调用 async def 返回 Coroutine;只有被 await、包装成 Task 或交给 Event Loop 后才执行。Task 在单线程事件循环中协作切换,阻塞文件/网络 SDK 或 CPU 密集计算会冻结同一 Worker 的所有 Run。阻塞 I/O 可受控放入 asyncio.to_thread,CPU 工作应进入进程池或外部 Worker。

Python 3.11+ 的 TaskGroup提供结构化并发:一个子任务以非取消异常失败时,会取消其余任务并在退出时抛出 ExceptionGroup。它与 gather(return_exceptions=True) 的部分失败语义不同,选择必须来自业务策略。

import asyncio
from collections.abc import Awaitable, Callable

async def run_tools(
    calls: list[Callable[[], Awaitable[object]]],
    timeout_s: float,
) -> list[object]:
    results: list[object] = []
    async with asyncio.timeout(timeout_s):
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(call()) for call in calls]
        results = [task.result() for task in tasks]
    return results

asyncio.CancelledError 直接继承 BaseException。Coroutine 可在 finally 清理资源,但通常应在清理后重新传播取消;吞掉它会破坏 TaskGroup 和 Timeout 的结构化并发语义。shield 只在必须完成的短临界区使用,不能让后台工作无限逃离取消预算。

保持跨语言一致的架构边界

Python 服务同样保持 Domain、Application、Adapters 和 Infrastructure 分层。领域状态与错误不依赖 FastAPI、Pydantic ORM 模型或具体 Agent SDK;Runner/Use Case 依赖 Protocol 端口;Adapter 承担 Provider、数据库和队列映射;应用启动阶段组装依赖。

Request Scope 的身份、租户、Deadline 与 Trace 使用显式 Context 对象传递。contextvars 适合日志/Trace 关联,但不应隐藏业务必需参数;跨进程、Queue 或 A2A/MCP 边界必须序列化所需上下文,不能期待内存 Context 自动传播。

共享的是线协议与兼容政策;两端分别解析并映射到自己的领域模型
TS Adapter生成类型 + 运行时解析 + Domain 映射。
Wire Contract稳定错误、时间、ID、幂等与 Trace。
Python AdapterPydantic 解析 + Domain 映射。

Pydantic/JSON Schema 的应用边界

Pydantic 适合解析 HTTP/A2A/MCP 请求、模型结构化输出、Tool 参数和配置。生产边界应明确 Strict Mode、未知字段策略、判别联合和自定义验证;默认强制转换可能把字符串数字悄悄接受,是否允许要由契约决定。

Pydantic Model 是 DTO,不等于业务实体。解析成功后仍要检查租户归属、状态转移、金额范围、引用存在性等领域不变量。也不要把 ORM/Pydantic 对象直接全部序列化给模型,这会泄露内部字段并让 Context 跟数据库 Schema 耦合。

版本边界Pydantic v1 与 v2 API、校验和序列化行为存在明显差异。文章不假设两者兼容;项目必须锁定 Major,并在升级时用真实 Payload 做双读回归。

理解生态但不让框架主导架构

FastAPI 建立在 ASGI 上,适合异步 HTTP、SSE 和依赖注入,但 Handler 退出后的工作不应依赖进程内 Background Task 保证 Durable。长 Run 进入 Queue/Worker 与外部 Store;流式连接断开后按任务契约决定取消或继续。

Agent SDK、Graph 和 Team 框架可以提供 Loop、Checkpoint 或 Observability Hook,仍需适配到内部端口。同步 Provider/数据库库包装到线程池会消耗有限 Worker,且取消未必终止底层调用;应优先选择原生 async 客户端或隔离阻塞依赖。

Python 镜像常含原生依赖,跨 CPU 架构和 glibc/musl 可能失败。构建产物、SBOM、漏洞扫描、非 root 用户、只读文件系统和冷启动时间都进入上线门禁。Demo 在开发机可安装,不等于生产平台可复现。

Python 工程质量基线

pytest 的 Fixture 应管理资源生命周期,不把全局 Event Loop、环境变量或真实时间泄漏到用例之间。Async 测试明确插件与 Loop Scope;Fake Model 产生确定事件,契约测试覆盖真实 Adapter,故障注入覆盖取消、Timeout、重复、乱序、部分 Artifact 和旧 Schema。

  • 类型:mypy 或 pyright 严格配置,禁止核心边界无界 Any。
  • 静态质量:Ruff/等价 Lint 与 Formatter,依赖与秘密扫描。
  • 可重复:冻结时间、随机种子和环境,隔离网络与文件。
  • 并发:测试取消传播、Task 泄漏、线程池耗尽和优雅关闭。
  • 模型:代码测试与概率 Eval 分开运行、分别报告。

把 Runtime Warning、未等待 Coroutine、遗留 Task 和资源未关闭升级为测试失败。只有最终回答正确而后台泄漏连接的用例,不算通过。

多语言团队以协议而非对象共享

TypeScript 和 Python 不共享类实例,而共享 OpenAPI/JSON Schema/Event Schema。尤其要写清缺失字段与 null、整数范围、Decimal/金额、UTC 时间、枚举未知值、二进制、ID 大小写和 Map 键语义。生成代码只能减少手工错误,不能代替兼容政策。

错误使用稳定机器码、可重试性和不确定结果字段;Trace 使用 W3C Trace Context;副作用携带幂等键;Event 带 Schema Version 和 Event ID。消费者采用读旧写新或 Version Adapter,不允许两个团队各自“合理猜测”。

同一 Golden Scenario 验证双语言实现
场景共同断言语言特有检查
正常 Run相同终态、Artifact 与事件顺序约束TS 穷尽 Reducer;Python 严格 DTO
取消停止派生工作、核对副作用AbortSignal;CancelledError 传播
未知事件不误判成功、保留原始数据并告警never 编译门;Unknown Variant 策略
重复消息Event ID/幂等键去重各自 Store 并发测试
旧 Schema兼容读取或显式拒绝生成类型与 Pydantic 双版本样本

跨团队 ADR 记录 Schema Owner、兼容窗口、发布顺序、Contract CI 和废弃通知。能用同一 Golden Scenario 在两端复现,是“协议一致”的证据;两个仓库都能编译不是。

主动提取:为什么把同步 SDK 放进 asyncio.to_thread 仍不等于原生异步?
它避免阻塞事件循环,但占用线程池;取消 Coroutine 未必终止底层线程调用,Context、连接池和超时语义也可能不同。它是隔离手段,不是语义等价。
协作判断:Python 与 TypeScript 的 DTO 字段相同,为什么仍需 Golden Scenario?
序列化、默认值、null、枚举、时间、数字范围、错误和并发语义可能不同。真实 Payload 与故障序列才能证明 Wire 行为一致。

资料与事实边界

以下以官方文档、标准、论文和原始工程资料为主。厂商 API、价格、限额与模型能力会变化,落地时仍需按目标版本做能力探测和回归。

  1. Python 3.14 asyncio Tasks — TaskGroup、取消、Timeout 与结构化并发的当前语言语义
  2. Python typing.Protocol — 结构子类型与端口接口
  3. Pydantic Strict Mode — 运行时严格解析与强制转换边界
  4. Pydantic JSON Schema — Model 与 JSON Schema 的生成/映射
  5. FastAPI async guidance — ASGI 服务中 async 与阻塞代码的边界
  6. W3C Trace Context — 跨语言 Trace 传播标准