Manjusaka的 RSS 预览

Pydantic 不是免费的——聊聊数据校验的边界

2026-03-23 22:30:00

最近给一个后端服务做 perf，跑了一发 py-spy 出来火焰图，盯着看了一会儿，发现 pydantic.main.__init__ 这一条在主热路径上占了一个让人挺难绷的比例。做了几轮调整之后顺手把这件事记一下——核心想说的是：对一个 web 服务来说，数据校验从来不是免费的，pydantic 也一样。它有它的边界，越界使用就要付出对应的代价。

起因：火焰图里看到的东西

某个内部业务服务，写法上挺标准——FastAPI + uvicorn + pydantic + psycopg，业务里所有”长得像数据”的东西，从 DB 行、内部传递的中间态、到 API response，全部都用 BaseModel 表达。看起来一致、规整，review 起来也舒服。

直到我们因为 P95 不太好看，跑了一发 py-spy：

1	py-spy record --duration 60 --rate 100 --output flame.svg --pid <uvicorn-worker-pid>

火焰图大约 8,700 个采样点，里面让我多看了一眼的是：

pydantic/main.py:250 也就是 BaseModel.__init__，单条最厚的火焰柱占了 22.65%（约 1,982 samples）
顺着主链路再往上扒，另一条 __init__ 路径又占了 6.81%
加加减减，光是”实例化 pydantic 模型”这一件事，在这个服务上吃掉了 不止 25% 的 CPU 时间

最骚的是，这些 __init__ 大部分根本不是用户请求的入口校验，而是从我们自己的 PostgreSQL 里 SELECT 出来的行，被 psycopg.rows.class_row(SomeModel) 一行一行喂给 BaseModel.__init__ 的——也就是说，我们在拿”自己写进去的、Schema 完全可控的数据”，反复跑一套带 schema validation 的实例化流程。

那一刻就一个念头：这事真不值得。

这件事意味着什么

Pydantic v2 已经把 validator 内核换成 Rust 写的 pydantic-core 了，比 v1 快了一个数量级。所以很多人下意识会觉得 “v2 的开销可以忽略”。这个印象大致没错——但只要你的 QPS 再高一点、单次请求里要构造的对象再多一点，”忽略不计的开销” × N 之后，它会非常明确地出现在你的 flame graph 上。

更重要的是，pydantic v2 快的是 校验本身。但你每实例化一次 BaseModel，整条链路是这样的：

__init__ 入口
  ├── 查 schema cache
  ├── pydantic-core validator dispatch（rust 侧）
  │     ├── field-by-field 类型校验
  │     ├── model_validator(mode="before")
  │     ├── 默认值填充
  │     └── strict / coerce 分支
  ├── 写回 __dict__ / __pydantic_fields_set__
  └── model_validator(mode="after")

哪怕你的 model 没有任何 validator、没有任何 Field(...)，进 rust 转一圈再回来这件事本身就有不可压缩的成本——函数调用、字典构造、属性写入，每一项都要钱。

而把这套机制用在 从可信源反序列化 的场景上，大部分钱花得没意义——你这一行就是你刚才 INSERT 进去的，schema 对不上的概率约等于”你的 migration 没跑过”，那已经是另一类问题了。

Pydantic 在一个 web 服务里到底干啥

把日常项目里 BaseModel 出现的地方分一下类，差不多是这五种：

API 边界：FastAPI route 的 request body / query / path param —— 输入来自外部，必须严格校验
响应序列化：FastAPI 的 response model —— 决定对外契约的 shape，并且要做 JSON 序列化
应用配置：pydantic-settings 读 YAML / env，启动时一次性校验
业务规则校验：带 field_validator / model_validator 的领域对象，比如 “金额必须是正数、组合字段必须满足某个不变式”
内部数据容器：DB 行映射、内部函数之间传递的中间态、纯粹为了”有类型、有字段名”而存在的小结构体

前四个有明确收益——schema 是契约，validation 就是这份契约的执行机制，它的运行时开销换的是 类型/不变式上的安全保证。

第五个，几乎全是白送的开销。这一类对象是在你自己写的代码里产生、在你自己写的代码里消费的，schema 已经被静态类型检查器（mypy / ty）和 beartype 之类工具覆盖到了；运行时再跑一次 pydantic 的校验，校验的对象要么是你 5 行前刚 query 出来的，要么是你 3 行前刚自己 Item(...) 构造的。换句话说，你是在校验你自己。

怎么替代：dataclass + slots

那第五类该用什么？答案非常无聊——标准库的 @dataclass，加 slots=True：

from dataclasses import dataclass

@dataclass(slots=True)
class RowEntry:
    id: str
    value: float
    position: int
    category: str

slots=True 让实例不再有 __dict__，每次构造省一份哈希表分配，访问字段也直接走 C 层 slot。配合 beartype 的运行时类型检查（如果你像我们一样在 __init__.py 里 beartype_this_package()），整体的”类型安全度”和你用 pydantic 是同一个量级的，但开销低得多。

更顺的一点是：psycopg 的 class_row 原生支持 dataclass，几乎是无缝替换：

from psycopg.rows import class_row

# 之前：BaseModel
class RowEntry(BaseModel):
    id: str
    value: float
    position: int
    category: str

# 之后：dataclass，调用方一行不用改
@dataclass(slots=True)
class RowEntry:
    id: str
    value: float
    position: int
    category: str

async def fetch_entries(conn, n: int) -> list[RowEntry]:
    async with conn.cursor(row_factory=class_row(RowEntry)) as cur:
        await cur.execute(
            "SELECT id, value, position, category FROM entries ORDER BY position LIMIT %s",
            (n,),
        )
        return await cur.fetchall()

调用侧的体感和原来一模一样——你照样有 .id、.value、.position，照样能被 ty / mypy 检查类型，照样能 for item in items: ...。差别只是火焰图里那条粗柱子掉下去了。

一个完整的”边界感”示例

把上面的话翻成代码。假设我们要做一个 /items 接口，从 DB 取数据、做点轻量计算、然后返给调用方：

from dataclasses import dataclass
from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field
from psycopg.rows import class_row

# ---- API 边界：必须 pydantic ----

class ListItemsRequest(BaseModel):
    """Request 来自外部，schema 必须严格校验。"""
    user_id: str = Field(min_length=1, max_length=64)
    limit: int = Field(default=20, ge=1, le=100)
    category: str | None = None

class ItemView(BaseModel):
    """Response 决定对外契约和 JSON 序列化方式。"""
    id: str
    value: float
    position: int

class ListItemsResponse(BaseModel):
    items: list[ItemView]
    total: int

# ---- 内部数据容器：dataclass 就够 ----

@dataclass(slots=True)
class RowEntry:
    """从 DB 读出来的一行，schema 完全可控，不需要再校验一次。"""
    id: str
    value: float
    position: int
    category: str
    raw_value: float

@dataclass(slots=True)
class ProcessedEntry:
    """中间计算态，只在内部函数之间传。"""
    id: str
    final_value: float
    position: int

# ---- 业务实现 ----

async def fetch_entries(conn, category: str | None, n: int) -> list[RowEntry]:
    sql = "SELECT id, value, position, category, raw_value FROM entries"
    params: tuple = ()
    if category is not None:
        sql += " WHERE category = %s"
        params = (category,)
    sql += " ORDER BY position LIMIT %s"
    params = (*params, n)

    async with conn.cursor(row_factory=class_row(RowEntry)) as cur:
        await cur.execute(sql, params)
        return await cur.fetchall()

def process_entries(rows: list[RowEntry], user_id: str) -> list[ProcessedEntry]:
    # 纯内部计算，所有输入输出都来自我们自己
    return [
        ProcessedEntry(
            id=r.id,
            final_value=r.value * _compute_factor(user_id, r.category),
            position=i,
        )
        for i, r in enumerate(rows)
    ]

router = APIRouter()

@router.post("/items")
async def list_items(
    req: ListItemsRequest,                    # pydantic 校验入参
    conn = Depends(get_conn),
) -> ListItemsResponse:                       # pydantic 决定出参 shape
    rows = await fetch_entries(conn, req.category, req.limit)    # dataclass
    processed = process_entries(rows, req.user_id)                # dataclass
    return ListItemsResponse(
        items=[ItemView(id=p.id, value=p.final_value, position=p.position)
               for p in processed],
        total=len(processed),
    )

整段代码里 pydantic 只活在两个位置：进来的请求 和 出去的响应。中间所有”只在我们自己代码里活动”的对象，全是 dataclass。这是一种很朴素的边界感——pydantic 守门，dataclass 干活。

什么时候必须保留 pydantic

不是所有看起来”内部”的 model 都该被改成 dataclass。下面这些情况，老老实实留着 pydantic：

用了 validator（field_validator / model_validator）—— 这些校验本身就是业务规则的载体
用了 model_dump() / model_dump_json() 做序列化 —— pydantic 的 dump 比手写转换更稳
用了 model_validate() / TypeAdapter 做反序列化 —— 输入来源不可信
用了 ConfigDict 配字段别名、from_attributes 之类的特性
继承链上有 BaseModel 且子类依赖父类的 schema 推导
配置类（pydantic-settings）—— 启动时一次性校验，是非常划算的开销
被外部库以 pydantic 协议消费（比如 LangChain 这种把 pydantic schema 当 tool 描述用的库）

辨认起来其实很机械——把这个 model 类的整个引用图扫一遍，看它有没有上面这些特性或这些用法。没有任何一条命中的，那它就是一个伪装成 model 的 NamedTuple，可以下放到 dataclass。

一个可以照抄的判断 checklist

每次新加一个数据结构时，先按下面这串问题走一遍：

要加一个新的 data class
  │
  ├─ 这个数据来自外部（HTTP 请求 / 用户输入 / 第三方 API）？
  │     └─ 是 → pydantic BaseModel
  │
  ├─ 这个数据要序列化到外部（HTTP 响应 / 写消息队列）？
  │     └─ 是 → pydantic BaseModel
  │
  ├─ 这个数据需要业务规则校验（字段间不变式 / 自定义 validator）？
  │     └─ 是 → pydantic BaseModel
  │
  ├─ 这是配置（从 YAML / env 读的）？
  │     └─ 是 → pydantic-settings
  │
  ├─ 这个 model 会被 model_dump / model_validate / TypeAdapter 用？
  │     └─ 是 → pydantic BaseModel
  │
  └─ 否则
        └─ @dataclass(slots=True)

这套规则不是要”消灭 pydantic”，而是把它放回它该在的地方——校验和序列化。其他位置上，标准库 + 类型检查器已经能给你足够的安全感，没必要额外缴一份运行时校验税。

收尾

回到那个火焰图。把”纯数据容器”那一类换成 @dataclass(slots=True) 之后，再跑了一次同等压力的 profile，主热路径上 pydantic.main.__init__ 的占比从 20%+ 掉到了零点几个百分点，相同 QPS 下 P95 也跟着下来一截。代码量倒没什么变化，受影响的也就是十来个类——但收益挺真实。

写到这里其实想说的话已经讲完了：

数据校验这件事，从来不是越多越好。它有明确的边界——边界之内是契约和安全，边界之外是无谓的开销。Pydantic 是非常好的工具，但用它的人也得知道自己什么时候用、为什么用。

下次起新项目，先想清楚这条边界画在哪。等到 flame graph 上 pydantic.main.__init__ 已经长成一根粗柱子才回过头来想，其实就晚了一步——好在还来得及。

差不多就这样。

当我们在维护模型 API 服务时我们在维护什么

2026-03-15 22:00:00

过去这一年我们团队陆陆续续做了几个 AI Service——大多是拿 FastAPI 包一个模型，再对外出 API。做到第三四个的时候我意识到一个尴尬的事实：每个项目长得都差不多，但每次起新项目还是在从零开始复制粘贴，而且不同的人写出来的目录结构、配置方式、metrics 命名都不太一样。一个新人接手第二个服务的时候，几乎要把第一个服务的认知全部重学一遍。

这篇文章想聊的不是某个具体框架，而是把这几个项目踩过的坑、迭代下来的设计、留下来的规范集中讲一遍——也算是给”维护一个模型 API 服务”这件事做一次复盘。

正文

把”维护一个模型 API 服务”拆开看，其实是几个相互独立但又互相影响的小问题：项目长什么样、配置怎么分层、模型版本怎么迭代、可观测性怎么做、日常开发流程怎么走、CI 和镜像怎么发。下面按这几个维度逐个聊。

一个 AI Service 该长什么样

先看目录结构。我们最终收敛到的样子是：

src/<service_name>/
├── __init__.py          # __version__ + beartype_this_package()
├── app.py               # create_app() factory, registries, lifespan, health mount
├── cli.py               # typer CLI: serve with --api-version list
├── config.py            # pydantic-settings BaseSettings + @cache singleton
├── handler.py           # BaseHandler ABC + shared request/response schemas
├── metrics.py           # Prometheus MetricsRegistry (version-aware)
├── v1/
│   ├── handler.py       # V1 Handler + get_handler
│   └── router.py        # register_router(app) with v1 routes
└── v2/                  # added when model evolves
    ├── handler.py       # V2 Handler + get_handler
    └── router.py        # register_router(app) with v2 routes
test/
├── test_app.py          # API integration tests
pyproject.toml           # hatchling build backend, uv managed
Makefile                 # install, format, lint, test, dev, clean
Dockerfile               # multi-stage with uv
entrypoint.sh            # calls CLI serve command

跟我们最早的版本比，最大的变化是引入了版本子目录。顶层的 handler.py 退化成了抽象基类，每个版本（v1、v2……）有自己的 handler 和 router。这是某个项目从 v1 演进到 v2 的时候被迫定型下来的——你不可能在原地修改老接口，又要让两个模型同时在线，唯一可行的就是物理隔离。

下面挑几个真正影响维护体验的点展开聊。

beartype：把类型检查塞进运行时

每个包的 __init__.py 第一行：

from beartype.claw import beartype_this_package
beartype_this_package()

__version__ = "0.1.0"

beartype_this_package() 会给整个包的所有函数挂上运行时类型检查，开销接近零。它能抓住静态检查器搞不定的那一类 bug——你函数签名写的是 str，但运行时传进来一个 None；或者 tensor shape 不对；或者某个第三方库返回值的类型和文档不一致。

在 AI Service 这种场景里，模型推理链路上的类型错误特别常见——预处理、tokenize、forward、后处理，每一步都可能有数据形状或类型上的微妙差异。能在 import 时就把网撒开，越早暴露越好。

配置分三层：Secret / App / CLI

这是看起来朴素但实战非常关键的一个设计——我们把配置严格分成三类：

Secret：HF token、API key、数据库密码——只从环境变量或 Vault 读，永远不进 YAML，不进 git
App config：model id、torch.compile 开关、各种业务阈值——主要从 YAML 读，环境变量可以覆盖
运行时参数：host、port、启用哪些版本——走 CLI

代码层面 Secret 和 App 是两个独立的 pydantic-settings 类：

class SecretSettings(BaseSettings):
    """Secrets only — always from env vars or Vault. Never put these in YAML."""
    model_config = SettingsConfigDict(env_prefix="<APP>_")
    hf_token: str | None = None

class AppSettings(BaseSettings):
    """General service config — from YAML, overridable by env vars."""
    model_config = SettingsConfigDict(
        yaml_file="config.yaml",
        env_prefix="<APP>_",
    )
    model_id: str = "default-model-name"
    torch_compile: bool = True

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        return (
            init_settings,
            env_settings,
            YamlConfigSettingsSource(settings_cls),
            file_secret_settings,
        )

这种分层一开始看起来啰嗦，但维护几个项目下来你会发现：每次出”配置爆炸”问题、每次部署混乱，几乎都来自把这三类东西混在一起处理。强制分层之后，K8s 的 Secret、ConfigMap、Deployment args 各管各的，review 起来一目了然。

Registry-Based App Factory：让 app.py 永远不用动

整个项目最值得说的一个设计决策——用注册表 + 动态导入来管理多版本，让 app.py 在加新版本时一行不用改。

app.py 里有两个 registry dict：

ROUTER_REGISTRY: dict[str, str] = {
    "v1": "<service_name>.v1.router",
    "v2": "<service_name>.v2.router",
}

# (handler module, handler class, app.state attr, settings model_id attr)
HANDLER_REGISTRY: dict[str, tuple[str, str, str, str]] = {
    "v1": ("<service_name>.v1.handler", "Handler", "handler", "v1_model_id"),
    "v2": ("<service_name>.v2.handler", "Handler", "v2_handler", "v2_model_id"),
}

create_app 接受一个版本列表，支持同时跑多个版本：

def create_app(api_versions: list[str] | None = None) -> FastAPI:
    if api_versions is None:
        api_versions = ["v1"]

    app = FastAPI(title="<Service Name>", version=__version__, lifespan=lifespan)
    app.state.api_versions = api_versions
    mount_health_routes(app)

    for version in api_versions:
        module = importlib.import_module(ROUTER_REGISTRY[version])
        module.register_router(app)

    return app

lifespan 遍历版本列表，动态加载每个版本的 Handler，存到不同的 app.state 属性上：

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    for version in app.state.api_versions:
        module_path, class_name, attr_name, model_id_attr = HANDLER_REGISTRY[version]
        model_id = getattr(settings, model_id_attr)
        module = importlib.import_module(module_path)
        handler_cls = getattr(module, class_name)
        handler = await asyncio.to_thread(handler_cls, model_id, settings.hf_token)
        setattr(app.state, attr_name, handler)
    yield

这个设计的好处是：app.py 不需要 import 任何具体版本的模块，加新版本就是加文件 + 加 registry 条目。换句话说，核心入口对版本扩展是封闭的，对版本添加是开放的——OCP 在这种场景下特别好用。

Handler 和 Router 的分版本组织

顶层 handler.py 定义 BaseHandler ABC 和共享的 schema：

class BaseHandler(ABC):
    @abstractmethod
    def _load_tokenizer_and_model(self, model_id: str, hf_token: str) -> tuple: ...

    def predict(self, request: BasePredictRequest) -> list[PredictResponse]:
        # 共享的推理逻辑
        ...

每个版本的 handler.py 实现具体的模型加载，并提供一个 get_handler 读取自己那个 app.state 属性：

# v1/handler.py
def get_handler(request: Request) -> Handler:
    return request.app.state.handler       # v1

# v2/handler.py
def get_handler(request: Request) -> Handler:
    return request.app.state.v2_handler    # v2

每个版本的 router.py 用 register_router(app) 闭包模式定义路由：

# v1/router.py — v1 没有前缀（默认版本）
def register_router(app: FastAPI) -> None:
    router = APIRouter()
    @router.post("/predict")
    async def predict(...) -> list[PredictResponse]:
        ...
    app.include_router(router)

# v2/router.py — v2 带版本前缀
def register_router(app: FastAPI) -> None:
    router = APIRouter(prefix="/v2")
    @router.post("/predict")
    async def predict(...) -> list[PredictResponse]:
        ...
    app.include_router(router)

注意 v1 的路由没有前缀（路径就是 /predict），v2 才开始加 /v2 前缀。这是对历史的一种妥协——v1 是最早的版本，改路径会破坏已有消费者，所以新增版本的代价由新版本自己承担。

所有 CPU 密集的推理操作都用 asyncio.to_thread 包起来。路由函数的返回值类型注解就是响应模型——不用 response_model= 参数，直接写 -> list[PredictResponse]，FastAPI 会自动推断。

模型版本怎么演进：一条铁律

讲完结构，最容易踩坑的部分是模型版本迭代。我们给自己立了一条铁律：

NEVER modify or rename an existing API route when adding a new model version.

听起来是废话，但真做的时候很多人会犯错。以下这些操作我们都见过、踩过、或者在 review 时拦下来过：

把 /api/v1/encode 改名成 /api/v1/encode-legacy，让它”看起来像被废弃了”
修改已有接口的 request/response schema（哪怕只是加一个可选字段）
在已有的 request 里加一个 model_version 字段来复用同一个路由
直接替换已有路由背后的模型（行为变了但版本号没变——这是 bug，不是 feature）

正确的做法只有一种，不管 API shape 变不变，流程都一样：

加 v2/handler.py + v2/router.py
在 app.py 的 ROUTER_REGISTRY 和 HANDLER_REGISTRY 里加两行
完事

新模型来了？
├── 加 v2/ 子目录（handler + router）
├── 在 app.py 的两个 REGISTRY 里加条目
├── 通过 --api-version 控制启用哪些版本
│   ├── 只跑 v2: --api-version v2
│   └── 同时跑: --api-version v1 --api-version v2
└── 老模型要下线？
    ├── 先在老路由里加 deprecation log warning
    ├── 看版本隔离的 metrics 确认零流量
    └── 确认后删除 v1/ 目录和 registry 条目

CLI 的 --api-version 是一个 list 参数，天然支持多版本同时运行：

@cli.command()
def serve(
    host: Annotated[str, typer.Option(help="Host to bind to")] = "0.0.0.0",
    port: Annotated[int, typer.Option(help="Port to bind to")] = 8000,
    api_version: Annotated[
        list[str], typer.Option("--api-version", help="API versions to enable")
    ] = ["v1"],
) -> None:
    app = create_app(api_versions=api_version)
    uvicorn.run(app, host=host, port=port)

如果新模型的 API shape 变了（比如 request 多了字段），v2 的 handler 里直接 subclass BasePredictRequest 加字段就行。v2 的 router 带 /v2 前缀，v1 的路由完全不受影响。

可观测性：版本感知的 metrics

模型迭代过程中真正让人头疼的不是”加版本”，而是”下版本”——你怎么确认 v1 真的没人用了，可以放心删？这就要求 metrics 必须按版本天然分开。

我们的做法是：每个 API 版本拥有自己的 MetricsRegistry 实例，通过 Prometheus 的 subsystem 命名来隔离。

_METRIC_NAMESPACE = "<service_name>"

class MetricsRegistry(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    predict_time: Histogram
    result_count: Counter

_registries: dict[str, MetricsRegistry] = {}

def get_metrics_registry(version: str) -> MetricsRegistry:
    if version in _registries:
        return _registries[version]
    subsystem = "api" if version == "v1" else f"api_{version}"
    registry = MetricsRegistry(
        predict_time=Histogram(
            name="predict_time",
            documentation="Time taken to run a prediction",
            namespace=_METRIC_NAMESPACE,
            subsystem=subsystem,
            buckets=tuple(2**i * 1e-3 for i in range(13)),  # 1ms ~ 4s
        ),
        result_count=Counter(
            name="predict_result_total",
            documentation="Number of results",
            namespace=_METRIC_NAMESPACE,
            subsystem=subsystem,
            labelnames=["label"],
        ),
    )
    _registries[version] = registry
    return registry

v1 的指标叫 <service_name>_api_predict_time，v2 的叫 <service_name>_api_v2_predict_time。在 Grafana 里天然隔离，不需要额外加 label。

为什么不用 label 区分版本？因为 Histogram 的 time() context manager 不支持直接传 label，你得先 .labels(version=...) 再 .time()，每个调用点都要记得传，容易漏。而 subsystem 方案是在注册时就确定了，调用侧完全无感。

在每个版本的 router.py 的 register_router 里通过一个小 wrapper 注入：

# v1/router.py
def _get_v1_metrics() -> MetricsRegistry:
    return get_metrics_registry("v1")

def register_router(app: FastAPI) -> None:
    router = APIRouter()

    @router.post("/predict")
    async def predict(
        request: PredictRequest,
        handler: Handler = Depends(get_handler),
        metrics_registry: MetricsRegistry = Depends(_get_v1_metrics),
    ) -> list[PredictResponse]:
        with metrics_registry.predict_time.time():
            results = await asyncio.to_thread(handler.predict, request)
            if results:
                metrics_registry.result_count.labels(label=results[0].label).inc()
        return results

    app.include_router(router)

除了自定义 metrics，还有一层全局的 HTTP 自动打点——用 prometheus-fastapi-instrumentator，它自带 handler label 可以按路径过滤，不需要按版本拆：

def mount_health_routes(app: FastAPI) -> None:
    Instrumentator(
        should_instrument_requests_inprogress=True,
        inprogress_name="requests_in_progress",
        inprogress_labels=True,
    ).instrument(
        app,
        metric_namespace="<service_name>",
    ).expose(app)
    app.add_api_route("/health", health([]))

最终一个同时跑 v1 和 v2 的服务，/metrics 端点会输出：

指标名	类型	来源
`<service_name>_api_predict_time`	Histogram	v1 自定义
`<service_name>_api_predict_result_total`	Counter	v1 自定义
`<service_name>_api_v2_predict_time`	Histogram	v2 自定义
`<service_name>_api_v2_predict_result_total`	Counter	v2 自定义
`<service_name>_http_requests_total`	Counter	Instrumentator
`<service_name>_http_request_duration_seconds`	Histogram	Instrumentator
`<service_name>_requests_in_progress`	Gauge	Instrumentator

自定义指标按版本隔离，HTTP 指标全局共享。下线 v1 的时候，看一眼 rate(<service_name>_api_predict_time_count[5m]) 是不是零，决策成本几乎为零。

日常开发的规矩

几个项目下来攒出来的几条不成文规定，后来都被强制写进了项目模板。

分支策略

一条规则，没有例外：

NEVER commit directly to main. All changes go through feature branches and PRs.

分支命名：feat/<topic> 或 fix/<topic>，通过 gh pr create 或 GitHub Web UI 创建 PR，CI 过了再合。不管是新功能、bug fix 还是改个配置文件，都走这个流程。这条规则的意义不在 code review 本身，而是它强制了每一次变更都要走 CI——而 CI 是这套体系里所有约束的最终防线。

提交前检查

每次 commit 前跑这三条：

1
2
3

uv run ruff check src/ scripts/ --fix
uv run ruff format src/ scripts/
uv run ty check

ruff 负责 lint 和格式化，ty（Astral 出的新一代 type checker）负责类型检查。这三条同时也会出现在 pre-commit hook 和 CI 里——本地、commit、CI 三道关口跑同一组命令，保证不会出现”本地通过 CI 挂”或者”CI 通过本地挂”的尴尬。

测试实践

改了 app.py、handler.py、config.py、metrics.py 这些核心文件之后，必须跑测试：

1	uv run pytest test/ -v

测试用 scope="session" fixture 避免每个 test case 都重新加载模型（加载一个 transformer 模型可能要几十秒）：

@pytest.fixture(scope="session")
def client():
    with TestClient(app) as c:
        yield c

GPU 相关的测试通过环境变量控制，在 import 之前就设好：

1 2	import os os.environ["<APP>_TORCH_COMPILE"] = "false" # 测试时跳过 torch.compile

每个 endpoint 至少要覆盖：happy path、默认参数、参数变体、422 校验、结果唯一性。

发版流程

版本号维护在 src/<service_name>/__init__.py 里，hatchling 自动读取。整个发版流程被刻意做得很无聊：

更新 __version__
合并到 main
在 GitHub 上创建 Release（不要手动打 tag）
GitHub Release 自动创建 tag，触发 docker-release CI 构建生产镜像

无聊就是稳定。人为干预越少，事故面就越小。

工具链与镜像

工具链

我们的标准工具链：

包管理：uv + pyproject.toml + uv.lock，构建后端用 hatchling
Lint：ruff（check + format 一把梭）
类型检查：ty
测试：pytest
Pre-commit：prek（pre-commit 的高性能替代）

pyproject.toml 里的工具配置：

[dependency-groups]
dev = [
    "pytest>=8.0.0",
    "ruff>=0.4.0",
    "ty>=0.0.19",
]

[tool.ruff]
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "W", "A", "PLC", "PLE", "PLW", "I", "C"]

[tool.pytest.ini_options]
testpaths = ["test"]

每个项目有一个标准 Makefile（注意 Makefile 的缩进必须是 tab）：

install:
    uv sync --frozen --no-install-project

format:
    uv run ruff format .
    uv run ruff check --fix .
    uv run ruff check --select I --fix .

lint:
    uv run ty check

test:
    uv run pytest

dev:
    $(MAKE) install
    uv run fastapi dev src/<service_name>/app.py

make format && make lint && make test，三板斧走完就可以提交。

Dockerfile

GPU 服务和 CPU 服务的 Dockerfile 不太一样，但有几个共同原则：

两阶段依赖安装：先 uv sync --frozen --no-install-project（只装依赖，利用 Docker 层缓存），再 ADD . /app && uv sync --frozen（装项目本身）
挂载 uv 缓存：--mount=type=cache,target=/root/.cache/uv
构建时健全检查：python -c "import <service_name>; print(<service_name>.__version__)"，在 build 阶段就发现 import 错误
entrypoint 调 CLI：不直接调 uvicorn，而是调 entrypoint.sh → CLI serve 命令——这样 host/port/api-version 这些运行时参数就不会散在 Dockerfile / k8s manifest / shell 多个地方

GPU 服务基于 nvidia/cuda 镜像：

FROM nvidia/cuda:<version>-cudnn-runtime-ubuntu24.04 AS base

COPY --from=ghcr.io/astral-sh/uv:<uv-version> /uv /uvx /bin/

WORKDIR /app

# Stage 1: deps only
RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=bind,source=uv.lock,target=uv.lock \
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
    uv sync --frozen --no-install-project

# Stage 2: project
ADD . /app
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --frozen

EXPOSE 8000
CMD ["/app/entrypoint.sh"]

# Sanity check
RUN python -c "import <service_name>; print(<service_name>.__version__)"

CPU 服务用多阶段构建，运行时镜像不带 uv 和构建工具，还跑非 root 用户：

# Build stage
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS builder
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_NO_DEV=1

# ... install deps and project ...

# Runtime stage
FROM python:3.12-slim-bookworm

RUN groupadd --system --gid 999 nonroot \
    && useradd --system --gid 999 --uid 999 --create-home nonroot

COPY --from=builder --chown=nonroot:nonroot /app /app
USER nonroot

CMD ["/app/entrypoint.sh"]

GitHub Actions

三条流水线分工明确：

# lint-and-test：push to main 和 PR 时触发
on:
  push:
    branches: [main]
  pull_request:

# docker-build：PR 时触发，验证 Docker 构建是否通过
on:
  pull_request:
    branches: ["main"]

# docker-release：v* tag 时触发，构建并推送生产镜像
on:
  push:
    tags: ["v*"]

PR push 触发构建验证，GitHub Release 创建 tag 触发生产镜像发布。整个链路自动化，人只需要点一下 Release 按钮。

怎么把这些经验落到下一个项目里

写到这其实想讲的核心都讲完了——那一串配置分层、registry 工厂、版本铁律、版本感知 metrics、CI/Docker 模板，每条背后都是某次具体的踩坑或者某次”诶我们上个服务好像也这样写过”的瞬间。

但写到第四第五个项目我们意识到：经验如果只活在某个人的脑子里，或者只在 wiki 上以”请参考某某规范”的形式存在，那它就只是一种”祝愿”——祝愿下一个起项目的同学读过、记得、并且在赶进度的时候还愿意按规矩来。这个祝愿命中率不高。

所以最近我们把这些经验沉淀成了一组 Claude Code Skill，按”项目脚手架 / 模型演进 / 日常维护 / CI 与镜像”四块拆开，写成了 AI 可以直接执行的指令。下次再起一个新的 AI Service，”帮我加一个 v2 的接口”这种话就直接对应到一组按规范操作的步骤——加新 router、加新 handler、加版本感知的 metrics、不碰已有接口。比起在 wiki 上写规范然后祈祷大家都读了，这种方式靠谱得多。

不过 skill 只是这套经验的一种载体——核心其实还是上面那些设计。哪怕你完全不用 Claude Code，把这些规则当成项目模板或者 cookiecutter 来用，也是一样的效果。

差不多就这样，希望对有类似需求的团队有点参考价值。

怎么样 tracing 你的 SQL？

2026-02-22 18:49:00

过年是个好时候，可以猛猛干活或者看论文。不过上了几天磨，看了几天论文后觉得还是需要整点活

所以这篇文章来聊聊一个经典话题，How to trace your SQL？

正文

在 OpenTelementry 这一套开始铺展开来后，我们对于代码整个生命周期的 tracing 有了一个较为成熟的方案。包括各类 auto-instrumentation 的库，能够让我们在不修改代码的情况下就能对整个调用链进行 tracing。

但是始终有一朵乌云盘绕着 Tracing 世界的大厦上，我们怎么样将 SQL 的执行如同业务代码一样从黑盒中拆出来

在现阶段，我们对于整个 Tracing 的引入都是在做加法，我们选择构建一个 context，注入一些 metadata，让代码中不同的环节都可以获取到上下文。但是还是有一个问题，怎么样在 SQL 中做加法呢？

最直观的想法是，我们可以尝试一个类似机制

begin;
set saka.tracing_id = '1234567890';
select * from users where id = 1;
commit;

我们可以在 SQL 中设置一些 tracing_id 之类的东西，这样我们就可以在明确一个事务的上下文了。但是这个方案有一个问题，这会改变我们使用 SQL 的 pattern，我们需要在每个 SQL 语句前面都加上 set saka.tracing_id = ‘1234567890’ 这样的东西，这样就会导致我们在代码中需要修改大量的 SQL 语句，这显然是不可行的。

那么另外一种方式实现的我们可以将 tracing 注入到 SQL 中，something like this:

1	select * from users where id = 1 /* tracing_id: 1234567890 */;

那么怎么做？

Google 提出了一个通用的方案 or 叫一个事实上的的标准吧，叫作 SQLCommenter，它的核心思想就是通过 Hook ORM 等手段，让我们尽可能简单的在 SQL 中注入一些 comment，这些 comment 中包含了 tracing/Custom Tag 的信息，这样我们可以将元数据注入的成本降到最低。

那么我们来根据 SQLCommenter 的方案来看看我们怎么样在 Python 中实现这个功能，

以 pymysql 为例，我们需要实现这个非常简单

from pymysql.cursors import SSCursor, SSDictCursor
from flask import g

from main.utils.context import INJECT_SQL_COMMENT, TRACE_ID
from main.utils.ip_utils import local_ip

try:
    CURRENT_IP = local_ip()
except:
    CURRENT_IP = "127.0.0.1"


def inject_meta_info(query: str) -> str:
    if INJECT_SQL_COMMENT.get():
        if TRACE_ID.get() != "None":
            trace_id = getattr(g, "TRACE_ID")
            sql_comment = f"/*X-Amzn-Trace-Id={trace_id}*/"
            query = f"{sql_comment} {query}"
    query = f"/*source_ip={CURRENT_IP}*/ {query}"
    return query


class CustomSSCursor(SSCursor):
    def execute(self, query: str, args: Any = None) -> int:
        return super(CustomSSCursor, self).execute(inject_meta_info(query), args)


class CustomSSDictCursor(SSDictCursor):
    def execute(self, query: str, args: Any = None) -> int:
        return super(CustomSSDictCursor, self).execute(inject_meta_info(query), args)

然后我们在使用时，注入自定义的 Cursor 就好了

Python 的生态还是幸福，但是很可惜，我现在是被迫在写 Node.js 的代码了，Node.js 的生态就没有那么幸福了。由于历史原因，我们现在用的是极为美味的 Prisma 作为 ORM。那么我们需要在 Prisma 来看一下怎么样注入 SQL Comment。

首先在 Prisma 最新的 v7.x 版本中，Prisma 本身实现了 SQLCommenter 的功能，something like this:

import { queryTags, withQueryTags } from "@prisma/sqlcommenter-query-tags";
import { PrismaClient } from "../prisma/generated/client";
const prisma = new PrismaClient({
  adapter,
  comments: [queryTags()],
});
// Wrap your queries to add tags
const users = await withQueryTags({ route: "/api/users", requestId: "abc-123" }, () =>
  prisma.user.findMany(),
);

最终会生成类似这样的 SQL

1	SELECT ... FROM "User" /requestId='abc-123',route='/api/users'/

OK，很不错，是预期内行为

但是问题在于 Prisma V7 是一个极为屎一样的版本，我们完全无法如品鉴母鸡卡一样品鉴这个功能。因为性能问题（v7 比 v6 慢了 30%-40%），我们完全无法在生产环境中使用 v7 的版本，所以我们只能在 v6 中实现这个功能了。

在 Prisma v6 中，Prisma 数据映射部分和核心的 Query Engine 是完全分开，他们通过走 NAPI-RS 进行通信，而他们自定义了一套 json based 的传输协议，协议样例如下

{
  "modelName": "User",                    // 可选，raw query 不需要
  "action": "findMany",                   // 操作类型
  "query": {                              // 查询详情
    "arguments": {                        // where/orderBy/take 等
      "where": { "email": { "contains": "prisma.io" } }
    },
    "selection": {                        // 字段选择
      "$scalars": true,
      "$composites": true,
      "posts": {                          // 关联查询嵌套
        "arguments": {},
        "selection": { "$scalars": true }
      }
    }
  }
}

而在这一次，我想实现类似官方的语义

return this.prisma.post.findMany({
  take: 100,
  sqlComments: getSqlComments(req),
});

OK，那么我们直接用一个流程图来输出一下对应的实现流程

flowchart TD    subgraph TS["TypeScript (prisma repo)"]        A["用户代码<br/>prisma.user.findMany({sqlComments: {...}, where: {...}})"]        B["getPrismaClient.ts :: _executeRequest()<br/>调用 serializeJsonQuery()"]        C["serializeJsonQuery.ts :: serializeJsonQuery() <b>[改动1]</b><br/>提取 sqlComments 放入 JsonQuery 顶层<br/>输出: {modelName, action, query, sqlComments}"]        D["RequestHandler.ts :: singleLoader / batchLoader<br/>调用 _engine.request(protocolQuery, {traceparent})"]        E["LibraryEngine.ts :: request()<br/>JSON.stringify(query) + JSON.stringify({traceparent})<br/>engine.query(queryStr, headerStr, txId)"]    end    subgraph FFI["C ABI / NAPI FFI 边界"]        F(("FFI"))    end    subgraph Rust["Rust (prisma-engines repo)"]        G["QueryEngine::query()<br/>解析 body_str → RequestBody<br/>从 header 提取 traceparent"]        H["RequestHandler::handle() <b>[改动2]</b><br/>body.into_doc() → (QueryDocument, SqlCommentsVec)<br/>组装 QueryContext {traceparent, sql_comments}"]        I["Single 查询<br/>QueryContext::new(traceparent, sql_comments#0)"]        J["Batch 查询<br/>每个 operation 独立 QueryContext<br/>共享 traceparent，各自 sql_comments"]        K["JsonBody::into_doc() <b>[改动3]</b><br/>JsonSingleQuery 新增 sql_comments 字段<br/>extract_sql_comments() → Vec&lt;(String,String)&gt;"]        L["QueryExecutor::execute() <b>[改动4]</b><br/>traceparent → query_context: QueryContext<br/>管道传递: execute_operation → interpreter → read/write"]        M["interpreter :: read.rs / write.rs <b>[改动5]</b><br/>query_context.sql_trace() → SqlTrace {traceparent, sql_comments}"]        N["ReadOperations / WriteOperations <b>[改动6]</b><br/>traceparent: Option&lt;TraceParent&gt; → trace: SqlTrace"]        O["sql-query-connector :: connection.rs <b>[改动7]</b><br/>Context::new(&connection_info, trace)"]        P["sql-query-builder :: read/write/select <b>[改动8]</b><br/>构建 Quaint AST<br/>调用 .add_trace_id(ctx)"]        Q["sql_trace.rs :: add_trace_id() <b>[改动9]</b><br/>build_trace_comment(sql_comments, traceparent)<br/>1. sql_comments 按 key 字母排序<br/>2. key/value URL 编码, 格式: key='value'<br/>3.traceparent sampled 则追加<br/>4. 逗号拼接, 调用 .comment(result)"]        R["Quaint AST :: .comment(...)<br/>渲染时在 SQL 末尾追加 /* ... */"]    end    S["最终 SQL<br/>SELECT ... FROM &quot;User&quot; WHERE ...<br/>/* controller='UserController',route='%2Fapi%2Fusers' */"]    A --> B --> C --> D --> E    E --> F --> G --> H    H --> I --> L    H --> J --> L    H -.->|内部调用| K    L --> M --> N --> O --> P --> Q --> R --> S

OK，在调整完 FFI ，扩展完 Query Engine 后，我们再调整一下 Prisma 代码生成相关的部分即可

然后我们可以在业务代码中这样使用

function getSqlComments(req: Request): Record<string, string> {
  const comments: Record<string, string> = {};

  // Request info
  comments.route = req.path;
  comments.method = req.method;
  if (req.route?.path) {
    comments["route_pattern"] = req.route.path;
  }

  // OpenTelemetry trace context
  const span = trace.getSpan(context.active());
  if (span) {
    const ctx = span.spanContext();
    comments.traceparent = `00-${ctx.traceId}-${ctx.spanId}-0${ctx.traceFlags}`;
  }

  return comments;
}

@Controller()
export class AppController {
  constructor(private readonly prisma: PrismaService) {}

  @Get()
  getHello(): string {
    return `Hello World!`;
  }

  @Get("posts")
  getPosts(@Req() req: Request) {
    return this.prisma.post.findMany({
      take: 100,
      sqlComments: getSqlComments(req),
    });
  }

  @Get("posts/:id")
  getPostsById(@Param("id") id: string, @Req() req: Request) {
    return this.prisma.post.findUnique({
      where: { id },
      sqlComments: getSqlComments(req),
    });
  }

  @Get("posts-with-comments")
  getPostsWithComments(@Req() req: Request) {
    return this.prisma.post.findMany({
      take: 100,
      include: {
        comments: true,
      },
      sqlComments: getSqlComments(req),
    });
  }

  @Get("posts-with-comments/:id")
  getPostWithCommentsById(@Param("id") id: string, @Req() req: Request) {
    return this.prisma.post.findUnique({
      where: { id },
      include: {
        comments: true,
      },
      sqlComments: getSqlComments(req),
    });
  }

  @Post("posts")
  createPost(@Body() body: { title: string; body: string }, @Req() req: Request) {
    return this.prisma.post.create({
      data: {
        title: body.title,
        body: body.body,
      },
      sqlComments: getSqlComments(req),
    });
  }
}

然后我们可以得到这样的 SQL

2026-02-22 07:56:38.681 GMT [190] LOG:  execute s412381: SELECT "public"."Post"."id", "public"."Post"."title", "public"."Post"."body", "public"."Post"."createdAt", "public"."Post"."modifiedAt" FROM "public"."Post" WHERE ("public"."Post"."id" = $1 AND 1=1) LIMIT $2 OFFSET $3 /* method='GET',route='%2Fposts%2Fff4ccd6d-2c15-4979-a5f3-0c27e4e2f169',route_pattern='%2Fposts%2F:id',traceparent='00-e984180b391935fbdf2b1e1f6f3b2b12-1286ff6754fbbc57-01' */
2026-02-22 07:56:38.681 GMT [190] DETAIL:  parameters: $1 = 'ff4ccd6d-2c15-4979-a5f3-0c27e4e2f169', $2 = '1', $3 = '0'
2026-02-22 07:56:38.681 GMT [190] LOG:  execute s412382: SELECT "public"."Post"."id", "public"."Post"."title", "public"."Post"."body", "public"."Post"."createdAt", "public"."Post"."modifiedAt" FROM "public"."Post" WHERE ("public"."Post"."id" = $1 AND 1=1) LIMIT $2 OFFSET $3 /* method='GET',route='%2Fposts%2Fff4ccd6d-2c15-4979-a5f3-0c27e4e2f169',route_pattern='%2Fposts%2F:id',traceparent='00-fa7f8686b25c2efe8942718207c28079-bbeb59140c47895c-01' */
2026-02-22 07:56:38.681 GMT [190] DETAIL:  parameters

通常来说，这样的语句已经能在我们常见的数据库调试流程中已经起到很大的帮助了，可以查到某一条慢 SQL 的来源，上下文等信息

但是这就够了吗？我们能不能把数据库也接入到 OpenTelemetry 的 tracing 体系中来呢？我们能不能在数据库的层面上看到整个调用链的 tracing 信息呢？

那没有问题的，这里以我现在更熟悉一点的 PostgreSQL 生态举个例子

Datadog 之前有一个工作，他们在 PostgreSQL 上实现了一个扩展叫作 pg_tracing https://github.com/DataDog/pg_tracing

通过使用 PostgreSQL 的一些扩展点

post_parse_analyze_hook
planner_hook
ExecutorStart_hook
ExecutorRun_hook
ExecutorFinish_hook
ExecutorEnd_hook
ProcessUtility_hook
xact_callback

然后配合一些环形缓冲区和共享内存，就可以在数据库层面上实现一个 tracing 的功能了，这样我们就可以将 PostgreSQL 接入到 OTEL 生态中了。当然这个库的实现里面有不少小技巧，改天可以单独写个文章来聊聊

MySQL 虽然内部的实现是一坨，但是我想如果要在 MySQL 上实现类似的功能应该工作量也不会太大，

最终的效果如下所示

差不多就这样

总结

大家都在写各种 AI/Agent 的文章的时候，我还在搞点这种 old school 的东西。恍惚间看到我的前方有一个巨大的风车。But anyway，我喜欢这些东西，这就够了

祝大家看的开心。

笑ってほしくて

2026-01-01 23:00:00

每年都会选一句话作为年终总结的标题，去年是“本当の僕らをありがとう”，今年我选择“笑ってほしくて”。

这句话出自《葬送的芙莉莲》的片尾曲《Anytime anywhere》。

愿你露出笑容

可能是我想对逝去之人，逝去的猫说的，可能也是他们想对我说的

也是我想对所有看到这篇文章的人说的

开篇

实际上一度想放弃写这篇文章，因为总是会害怕。今年在生死边缘搏战了许久。但终究是没有挽留住。每次想起点点滴滴时，总会不免破防

但是就如同歌词所说

こんなに胸が痛いのは/胸口传来的痛楚

あなたといた証かな/是与你同在的证明

那么所以还是要写点什么来纪念这特殊的2025, 也是 saka 的2025

生活

今年生活中最大的事件就是猫咪的过世。小熊是我们在2023年从小区收养的猫咪。一最开始就给上了强度，重度口炎+肾衰。然后后续情况稳定后带回家进行日常的治疗。

而在24年经过一整年的折腾（开腹两次，病危三次），25年年初医生终于说可以两月复查后，我们以为小熊还能陪伴我们很长时间

但是事与愿违，在6月份过完到家纪念日后，小熊的状态就时好时坏。在8月份最后一次住进医院后，虽然期间还是有好转的时候，但是整体情况还是急转直下。最终我们选择放他离开

很难说，小熊的离去对我的打击是怎么样的。我同事说感觉我从中恢复过来还蛮快的。不过中个泪流时分也只有自己知道（其实写这篇文章的时候也是在默默流泪）

今年也是我好友离开我的第五个年头。恍惚间，我已经比他大了

君埋泉下泥销骨，我寄人间雪满头

不得不说古人是真的会写啊。

聊点开心的，今年我的日常生活比去年又丰富了很多。史无前例的，saka 获取了，他的，第一个全成就游戏！所以 2025 saka 严选最佳游戏的是？

是《空之轨迹 1st 重制版》

对不起 P5R，我也很爱你的（

今年也还在继续的拍照，买了 Z135 F1.8S 圆神，Z105 F2.8 百微，Z35 F1.2S 三个头，在摄影器材的路上越走越远，整体的快门数快 10w 了，整体以猫狗为主。某种意义上来说，我很庆幸用相机记录下了小熊这几年最精彩的时光

今年对于我来说另一个在最大的改变就是，我去了出来能去的最远的地方——日本

我精神图腾中的几大支柱（奥特曼，摇曳露营）都发源自日本，而我第一次出国便是去日本（某种意义上算冥冥之中的一种巧合）

在日本出差的这两周里，解锁了很多新奇的体验。去秋叶原爆米，去抚子的故乡爆米，在出租车上和司机一起唱奥特曼主题曲，和好友一起逛街爆米，和同事一起去热海合宿，见了好看的海边，去了沼津，吃了好吃的海鲜饭等等等。

某种意义上今年开启了我过去很多人生中未曾想过的体验。也让我坚定的了一个想法、

人与人的欢乐实际上是可以相通的

感情

感情进入了第七个年头。如果要为今年的感情沉淀一个主题的话，那可能就是四个字，生死相依

小熊走前的一个月，我们在医院陪护。仪器的滴鸣声，时而奔走的医护。时而逝去的生命。那种氛围真的很压抑

这种时候，两个人的相互的守望，可能是枯燥且祈祷着奇迹发生的日子里，已然发生少数的奇迹。

小熊离去后，我们两也久违的带着小狗出远门去散心，去一起给小狗拍了很多好看的照片，一起救助了其余的小猫，一起给医院捐献了设备。

能在人群中遇到彼此守望的伴侣，本身已然是奇迹

工作&技术

今年年初在朋友邀请下，我加入了一家 AIGC 创业公司，负责一些 infra 上的工作。这算是我职业生涯一个全新的转折点。

如果说这里面最大的感受是什么，那就是我可能在之前聊过的，身份转变所带来的更多的责任与事情。

先聊聊务虚的部分，往常我只需要 focus 在具体的事情的落地，可能其余的东西都会有人帮我兜底。而我今年开始成为一个试图去帮助其余同事能够更好落地事情的人。说实话这种身份上的转变实际上会让你有很多想法转变。就如同我和团队成员 1:1 沟通时我一直在问的一个问题一样“你觉得我还能做什么，能帮助你落地现在的事情？”。很多时候我去思考事情的时候，我不能仅考虑这个事情本身，而是我需要去思考这个事情怎么样才能够帮助团队/同事实现更好的价值。

与此同时，我的职责以及角色会变得更多元化。这一点也是需要走出舒适区。比如今年我在公司做了很多涉及到 DBA 和搜索相关的工作。严格来说，这一部分工作其实离我的好球区实际上很远，但是如果一个事情是重要的，同时当下没有人比我更擅长这件事（换句话说让我做不会是最坏的结果），那么我就需要去承担起这份责任。

接着是务实的部分，如果要说今年很有成就感的事情的话，就是在加入公司后，从头开始做稳定性相关的建设，且收获颇丰。无论是在多次互联网 infra 大范围 crash 后我们能在最短的时间内恢复业务，还是给我们的业务带来了实际上的稳定性与性能提升。我所做的工作都能直接的作用在用户价值上。某种意义上来说，这也是人生的一件幸事

以上都是好的 part，坏的 part 也不是没有，那就是

教练，我想写代码

我现在写代码基本上只能靠工作之外写写代码来保持手感和练手，呜呜呜呜呜，我真的好想转岗去写代码啊（

不过这也不算坏事，我现在非常享受业余时间写代码的生活，所以今年在社区的贡献比去年多了不少，包括把自建图床项目的代码拾掇了很多，给 CPython 的 JIT 贡献了不少代码。也让自己名字第一次在官方文档里留下了痕迹。

如果说明年有什么想法的话，简单的说就是在工作上带好团队，能够继续突破我自己的舒适区。以及希望我能给 CPython 下一阶段的 JIT 贡献更多的代码（其实有一些想法，正好趁着元旦假期梳理一下）

啊，突然想起，我26年还得好好学学 GPU 相关的编程的东西，说实话学习新的东西真的是非常开心的一件事！

关于 AI

算是凑个热点，聊聊我对 AI 的看法。

很多人都会陷入一种焦虑，“AI 会不会替代我自己”

我自己的看法愈发的清晰，AI 时代会让人的价值更大，而不是被消泯

这里的逻辑很简单，AI 让一切都变得更为高效，内容的生产更为高效，开发效率更为高效，公司的形态更为灵活，在这种情况下，传统的很多东西在 AI 时代都面临的新的挑战。比如我举几个例子

对于愈发高效的内容生产效率，UGC 等形态的 AI Startup 会面临越来越大的 anti nsfw 的合规的挑战

而对于算力的愈发的渴求，边缘 GPU 算力的管控与接入也愈发成为一个重要的课题

AI 带来的生产力提升其实会让人的价值在这个时代更为凸显，而不是被消解

总结

差不多就是这样吧。过去一年有过不少很多的泪水，也有过很多的难眠之夜，也有过很多的快乐。

坦白来说能挺过这一年，是因为身边有着很多的陪伴，有我女朋友，有一群好朋友，有一群好同事。

每一个人的陪伴加上生活中辛酸苦辣甜汇聚在一起就成为了2025年的 saka，或者是 saka 的2025。

曾经想对于以后试图列出很多的展望，但是死生经历过一轮后，觉得剩下的都不太重要了

如同标题一样，

笑ってほしくて/愿你露出笑容

愿我们每一个人都能以笑容度过2026年的每一天

一万年太长，我们只笑今朝（

新年快乐！

成为榜样，但不要成为偶像

2025-10-03 04:00:00

最近刚把空轨 1st 打通关。差不多国庆假期也要结束了，要开始准备接下来的学习和工作了。趁着这个机会，写点东西，当个迟到的 PyCon China 2025 的总结吧

正文

其实去年在结束去年的会议后，我就在考虑要不要彻底退出 PyCon China 一段时间。原因其实也很简单，太累了。不过我向来是个很拖延癌的人。恰逢那时候正在职业的变动期，所以想着要不要再看看。

今年来了一家 AIGC 公司做 infra，说实话我干的还蛮开心的。每天都会学习新的东西。状态相比于去年好了不少（除了每天都需要和该死的 Any 做斗争），所以一度准备继续参加今年的 PyCon China

不过到了筹备期后，原本的计划突生变故。陪伴需求的猫咪病危，去世。一直在医院通宵，陪护以及最终要面临的生死别离。对我的精神和体力造成了极大的消耗。是否继续参加今年的 PyCon China 也成为一个问题。

不过最终还是决定参加了。今年突然出版社那边给我加了一个活，我翻译的书要在现场签售。这就又成为我心里一个新的疑问：嘛，真的会有人来买吗？

嘛，其实这也是每年在参加 PyCon 时都会有的自我怀疑，我真的有资格在这里吗？我讲的东西真的会有人听吗？

不过做了最终的决定那么就继续做吧。所以今年还是给了一个白银赞助+一个主题演讲。

时间过得很快，转眼到了会议前一天，临上飞机前我还在被 AWS 的傻叉 OpenSearch 折磨。下飞机后 yihong，piglei，空想家，jay 他们去开 impact 了。好好好开 impact 不带我是吧。我就孤零零的跑去了酒店。有人演讲前12h文件夹都还没建是谁我不说。

在酒店顶了几个小时把 PPT 赶完，工作收个尾。勉强睡了2h，然后就去会场了。

今年比较轻松的是早上的主持不用我了，我只要负责暖场一下。不过临开场前发现 C 会场的 PPT 还没收，然后摇了晚枫帮忙。

暖场还是我每年的保留节目，Saka 三问.jpg:

现在还在写 Python 的举个手
现在还没写 Python 的举个手（拖出去
用 2.7 的举个手

看起来效果还不错。暖场完我就跑出去了。不过卧槽，今年怎么这么多人来单杀我，~~妈耶社恐狂哭了~~。

不过说实话，在经历生死一圈后，和老朋友打打闹闹，认识一下新朋友，还是蛮开心的。特别是很多人过来说我影响了他们，我是他们的偶像，他们从我的博客和分享里学到很多的时候。我一直以来的疑惑也得到了解决。虽然说人的自我认同最理想应该是内源性的，但是大家的认可也真的会让我很开心。

哦对了，还有很多人说他们也很喜欢摇曳露营！

值得一提的是，我司的宝藏同事从日本来一起面基了（是的，我们是 Remote 公司），还给我带了手办

当然有某屑 HR 说要来参会结果早上睡过了，是谁我不点名了

上午的时间其实过去的很快，很快就结束了。然后我就来到了签售地方。出乎意料的，有很多人都来了，有来捧场的老朋友们，有刚刚线下刚对上号的新朋友们。大家一起打打闹闹，我用我的丑字写了很多祝福，也有很多杂话。要说哪一句最真心，那我觉得是“不要用 Next.js”罢。

签售完火急火燎的吃完盒饭，抽烟的时候遇到师父，~~我开始基情的抚摸他的胸肌~~，这可能算是每次我们师徒相聚的仪式感了。问他了一个问题：你现在还需要我帮你收尸么？这个出处源自于我们之前约好他要是自杀我会来帮他处理后事。他想了想说，我们不如想想怎么活到150岁吧。很好，很强大，我很喜欢的回答。那我就当不会了。

后面纯爷也来加入了聊天，我顺便向他倾诉了一下把 PostgreSQL 当 MongoDB 用的痛苦。纯爷也只能用爱莫能助的眼光看看我

下午 C 会场实际上因为我需要抽烟提神迟到了几分钟，yihong 帮我暖场缓解尴尬，不得不给他磕一个，以及 yihong 抱起来手感很好，建议网友有条件的可以去试试

下午到我的时候其实因为控场的原因给我的时间比预期的要短一些，所以我临时调整了一些内容。要上去讲的时候，发现很多人都从其余会场赶了过来，算是非常满足了。在 QA 的时候，我给大家说我现在在一家用着 Node.js 写着 Any，把 PostgreSQL 当 MongoDB 用，以及还用着 GraphQL 的公司工作。大家一片会心一笑。我想我们屑 HR 大概今天在现场招不到人了罢

下午的时候，屑 HR 和我们在上海的另外一位同事也来，带来了我需要~~昏睡红茶~~零度可乐。说实话在一个 Remote 公司大家面基的机会还是很少的。理所当然，我成为公司群内表情包的一部分。该考虑下找屑 HR 要肖像费了

晚上散场后，去和沪爷阿蔡以及几个同事一起组了一个局。不过说实话拉着明天要去霓虹的两位同事去吃日式拉面算不算另一种职场 80？笑死

总结

说实话今年 PyCon 的当天是我这两个月最快乐的一点，也许在很多年后我记不得了今年讲了什么，但是还会记得当天最简单快乐。

嘛，从18年到现在，7年过去了，我也从一个刚出校园的年轻人变成了一个老登。似乎很多东西都在变，但是很多东西又没变。

我还是很菜，但是我好像比之前影响了更多的人，帮到了更多的人？我还有很多东西不会，但是我好像能学的东西也更多了？还是会经历很多痛苦，很多迷茫，很多挣扎。但是生活似乎也还是一如既往的有着无限的希望与美好？

回北京后，一次吃饭时，我给我女朋友说，你知道吗？很多人说我是他们的偶像。我女朋友说：不，你是他们的榜样

是的，成为榜样，但是不要成为偶像。

这篇文章差不多写到这里。要到中秋了，除了祝大家中秋快乐，阖家欢乐以外。也祝大家每个人都能在这个快速迭代的世道里永葆初心。用 Piglei 老师的话说就是“老而不登”

这个世界唯有爱，希望，奥特曼与摇曳露营不可辜负，抚门！

Python 3.14 的进一步性能进化: Tail Call Interpreter

2025-07-02 23:49:00

最近做安全做的我头晕脑胀，来点轻松的换换脑子，让自己放松下

Python 3.14 正式引入了一个新的机制叫作 Tail Call Interpreter（Made by Ken Jin），这无疑又是一个奠定未来基础的重大工作

正文

在聊 Python 3.14 的 Tail Call Interpreter 之前，我们先要来聊 C 语言中最基本的语法 switch-case

switch (x) {
    case 1:
        // do something
        break;
    case 2:
        // do something else
        break;
    default:
        // do default thing
}

对于这样的代码我们最终的汇编会是什么样的呢？可能大家第一反应是先 cmp 然后 je ，不等式秒了，我们编译一个版本来看看

对于这样一段代码

void small_switch(int x) {
    switch(x) {
        case 1: printf("One\n"); break;
        case 2: printf("Two\n"); break;
        case 3: printf("Three\n"); break;
        default: printf("Other\n"); break;
    }
}

最终汇编的产物会是这样的

00000000000011f0 <small_switch>:
    11f0:83 ff 02             cmp    $0x2,%edi
    11f3:74 2b                je     1220 <small_switch+0x30>
    11f5:83 ff 03             cmp    $0x3,%edi
    11f8:74 16                je     1210 <small_switch+0x20>
    11fa:83 ff 01             cmp    $0x1,%edi
    11fd:75 31                jne    1230 <small_switch+0x40>
    11ff:48 8d 3d fe 0d 00 00 lea    0xdfe(%rip),%rdi        # 2004 <_IO_stdin_used+0x4>
    1206:e9 25 fe ff ff       jmp    1030 <puts@plt>
    120b:0f 1f 44 00 00       nopl   0x0(%rax,%rax,1)
    1210:48 8d 3d f5 0d 00 00 lea    0xdf5(%rip),%rdi        # 200c <_IO_stdin_used+0xc>
    1217:e9 14 fe ff ff       jmp    1030 <puts@plt>
    121c:0f 1f 40 00          nopl   0x0(%rax)
    1220:48 8d 3d e1 0d 00 00 lea    0xde1(%rip),%rdi        # 2008 <_IO_stdin_used+0x8>
    1227:e9 04 fe ff ff       jmp    1030 <puts@plt>
    122c:0f 1f 40 00          nopl   0x0(%rax)
    1230:48 8d 3d db 0d 00 00 lea    0xddb(%rip),%rdi        # 2012 <_IO_stdin_used+0x12>
    1237:e9 f4 fd ff ff       jmp    1030 <puts@plt>
    123c:0f 1f 40 00          nopl   0x0(%rax)

我们能看到整体如我们所预期的一样，不断的 cmp 然后不断的 je，然后我们评估一下这里的复杂度呢？哦，时间复杂度 O(n) 秒了。

卧槽，那对于 Python 这样一个超大的 switch case 的结构，岂不是每次都是一个 O(n) ？这不得原地升天？

其实不是，通常来说，编译器会根据数据的类型和规模来用不同的方案处理 switch case 的结构

我们来准备几组例子

void small_switch(int x) {
    switch(x) {
        case 1: printf("One\n"); break;
        case 2: printf("Two\n"); break;
        case 3: printf("Three\n"); break;
        default: printf("Other\n"); break;
    }
}

void dense_switch(int x) {
    switch(x) {
        case 10: printf("Ten\n"); break;
        case 11: printf("Eleven\n"); break;
        case 12: printf("Twelve\n"); break;
        case 13: printf("Thirteen\n"); break;
        case 14: printf("Fourteen\n"); break;
        case 15: printf("Fifteen\n"); break;
        case 16: printf("Sixteen\n"); break;
        case 17: printf("Seventeen\n"); break;
        default: printf("Other\n"); break;
    }
}

void sparse_switch(int x) {
    switch(x) {
        case 1: printf("One\n"); break;
        case 100: printf("Hundred\n"); break;
        case 1000: printf("Thousand\n"); break;
        case 10000: printf("Ten thousand\n"); break;
        default: printf("Other\n"); break;
    }
}

void large_dense_switch(int x) {
    switch(x) {
        case 1: printf("Case 1\n"); break;
        case 2: printf("Case 2\n"); break;
        case 3: printf("Case 3\n"); break;
        case 4: printf("Case 4\n"); break;
        case 5: printf("Case 5\n"); break;
        case 6: printf("Case 6\n"); break;
        case 7: printf("Case 7\n"); break;
        case 8: printf("Case 8\n"); break;
        case 9: printf("Case 9\n"); break;
        case 10: printf("Case 10\n"); break;
        case 11: printf("Case 11\n"); break;
        case 12: printf("Case 12\n"); break;
        case 13: printf("Case 13\n"); break;
        case 14: printf("Case 14\n"); break;
        case 15: printf("Case 15\n"); break;
        case 16: printf("Case 16\n"); break;
        case 17: printf("Case 17\n"); break;
        case 18: printf("Case 18\n"); break;
        case 19: printf("Case 19\n"); break;
        case 20: printf("Case 20\n"); break;
        default: printf("Other\n"); break;
    }
}

void mixed_switch(int x) {
    switch(x) {
        case 1: printf("Case 1\n"); break;
        case 2: printf("Case 2\n"); break;
        case 3: printf("Case 3\n"); break;
        
        case 50: printf("Case 50\n"); break;
        
        case 100: printf("Case 100\n"); break;
        case 101: printf("Case 101\n"); break;
        case 102: printf("Case 102\n"); break;
        
        default: printf("Other\n"); break;
    }
}

void char_switch(char c) {
    switch(c) {
        case 'a': printf("Letter a\n"); break;
        case 'b': printf("Letter b\n"); break;
        case 'c': printf("Letter c\n"); break;
        case 'd': printf("Letter d\n"); break;
        case 'e': printf("Letter e\n"); break;
        default: printf("Other char\n"); break;
    }
}

然后我们反汇编以下，看下结果（这里我只把关键的部分贴出来）

00000000000011f0 <small_switch>:
    11f0:83 ff 02             cmp    $0x2,%edi      # 比较是否为2
    11f3:74 2b                je     1220            # 跳转到case 2
    11f5:83 ff 03             cmp    $0x3,%edi      # 比较是否为3
    11f8:74 16                je     1210            # 跳转到case 3
    11fa:83 ff 01             cmp    $0x1,%edi      # 比较是否为1
    11fd:75 31                jne    1230            # 不是则跳转到default

0000000000001240 <dense_switch>:
    1240:83 ef 0a             sub    $0xa,%edi      # 减去10 (最小case值)
    1243:83 ff 07             cmp    $0x7,%edi      # 比较范围 (17-10=7)
    1246:0f 87 90 00 00 00    ja     12dc           # 超出范围跳转default
    124c:48 8d 15 15 0f 00 00 lea    0xf15(%rip),%rdx # 加载跳转表地址
    1253:48 63 04 ba          movslq (%rdx,%rdi,4),%rax # 获取偏移量
    1257:48 01 d0             add    %rdx,%rax      # 计算目标地址
    125a:ff e0                jmp    *%rax          # 间接跳转

00000000000012f0 <sparse_switch>:
    12f0:81 ff e8 03 00 00    cmp    $0x3e8,%edi    # 比较1000
    12f6:74 40                je     1338           # 等于则跳转
    12f8:7f 16                jg     1310           # 大于1000则继续检查
    12fa:83 ff 01             cmp    $0x1,%edi      # 小于1000，检查1
    12fd:74 49                je     1348           
    12ff:83 ff 64             cmp    $0x64,%edi     # 检查100
    1302:75 24                jne    1328           # 都不是则default
    ...
    1310:81 ff 10 27 00 00    cmp    $0x2710,%edi   # 检查10000

0000000000001360 <large_dense_switch>:
    1360:83 ff 14             cmp    $0x14,%edi     # 检查是否≤20
    1363:0f 87 53 01 00 00    ja     14bc           # 超出范围
    1369:48 8d 15 18 0e 00 00 lea    0xe18(%rip),%rdx # 跳转表地址
    1372:48 63 04 ba          movslq (%rdx,%rdi,4),%rax # 直接索引
    1376:48 01 d0             add    %rdx,%rax
    1379:ff e0                jmp    *%rax

00000000000014d0 <mixed_switch>:
    14d0:83 ff 32             cmp    $0x32,%edi     # 比较50
    14d3:74 7b                je     1550
    14d5:7f 29                jg     1500           # >50的情况
    14d7:83 ff 02             cmp    $0x2,%edi      # ≤50，检查小值
    14da:74 64                je     1540
    14dc:83 ff 03             cmp    $0x3,%edi
    ...
    1500:83 ff 65             cmp    $0x65,%edi     # >50，检查100,101,102
    1503:74 5b                je     1560

0000000000001580 <char_switch>:
    1580:83 ef 61             sub    $0x61,%edi     # 减去'a'的ASCII值
    1583:40 80 ff 04          cmp    $0x4,%dil      # 检查是否≤4 (a-e)
    1587:77 63                ja     15ec
    1589:48 8d 15 4c 0c 00 00 lea    0xc4c(%rip),%rdx
    1590:40 0f b6 ff          movzbl %dil,%edi      # 零扩展到32位
    1594:48 63 04 ba          movslq (%rdx,%rdi,4),%rax

我们这里能看到编译器根据数据的不同类型来处理了 switch case 的结构，这里我用一个表格总结一下

Switch类型	Case数量	分布特点	编译器策略	时间复杂度
small_switch	3个	连续(1,2,3)	线性比较	O(n)
dense_switch	8个	连续(10-17)	偏移跳转表	O(1)
sparse_switch	4个	稀疏(1,100,1000,10000)	二分查找	O(log n)
large_dense_switch	20个	连续(1-20)	标准跳转表	O(1)
mixed_switch	7个	部分连续	嵌套比较	O(log n)
char_switch	5个	连续(‘a’-‘e’)	字符偏移表	O(1)

OK，这里我们发现，Switch-case 最终的实现因为数据类型的不一样，会导致我们最终的代码存在一个不可预测性。那么我们有没有办法来优化这个问题呢？答案是有的。

我们来看下面一段代码

#include <stdio.h>

void basic_computed_goto(int operation) {
    static void* jump_table[] = {
        &&op_add,   
        &&op_sub,   
        &&op_mul,   
        &&op_div,   
        &&op_mod,   
        &&op_default
    };
    
    int a = 10, b = 3;
    int result;
    
    if (operation < 0 || operation > 4) {
        operation = 5;
    }
    
    printf("Operation %d: a=%d, b=%d -> ", operation, a, b);
    
    goto *jump_table[operation];
    
op_add:
    result = a + b;
    printf("ADD: %d\n", result);
    return;
    
op_sub:
    result = a - b;
    printf("SUB: %d\n", result);
    return;
    
op_mul:
    result = a * b;
    printf("MUL: %d\n", result);
    return;
    
op_div:
    if (b != 0) {
        result = a / b;
        printf("DIV: %d\n", result);
    } else {
        printf("DIV: Error (division by zero)\n");
    }
    return;
    
op_mod:
    if (b != 0) {
        result = a % b;
        printf("MOD: %d\n", result);
    } else {
        printf("MOD: Error (division by zero)\n");
    }
    return;
    
op_default:
    printf("Unknown operation\n");
    return;
}

我们能看到这里核心的一个操作是将我们 Switch-cased 的每个 case 都变成了一个标签，然后我们通过一个 jump_table 来直接跳转到对应的标签上, 我们来看一下最关键位置的汇编

1 2	11d3:48 8d 05 c6 2b 00 00 lea 0x2bc6(%rip),%rax # 3da0 <jump_table.0> 11da:ff 24 d8 jmp *(%rax,%rbx,8)

这里我们可以总结出来使用 Computed Goto 相较于传统的 switch-case 有以下几点优势

减少分支预测 fallback 的代价
指令缓存局部性上更优
减少了 cmp 指令的数量和开销

那么具体能有多快呢？可以参见 CPython 引入的 Computed Goto 的一个测试结果，大概是整体提升了15% 左右

那么 Computed Goto 的方式就是完美的吗？其实也不是，目前 CPython 的解释器 ceval.c 也是目前最大的一个 switch case 中存在几个典型问题

Computed Goto 作为 clang 和 gcc 特化的功能，那么其余平台受益的可能性较小
目前 Computed Goto 其实并不成熟，在同一个编译器不同的版本可能会有不同的问题
超大型的 switch case 会导致编译器对于 switch case 的优化不够好
我们无法使用 perf 精确的去对我们整个过程中 per opcode 的开销进行定量分析，这在于让 Python 变得更快的大背景下将会是一个很大的问题

1，3，4 都很好理解，我们来看一下2的一个例子（感谢 Ken Jin 提供的例子）

在 GCC 11 的时候，switch-case 某个部分会正常的代码

738f: movq%r13, %r15
7392: movzbl%ah, %ebx
7395: movzbl%al, %eax
7398: movq(,%rax,8), %rax
73a0: movl%ebx, -0x248(%rbp)
73a6: jmpq*%rax

而在 GCC 13-15Beta 的时候，则会产生这样的代码

747a: movzbl%ah, %ebx
747d: movzbl%al, %eax
7480: movl%ebx, -0x248(%rbp)
7486: movq(,%rax,8), %rax
748e: jmp0x72a0 <_PyEval_EvalFrameDefault+0x970>

72a0: movq%r15, %xmm0
72a5: movq%r13, %xmm3
72aa: movq%r15, %rbx
72ad: punpcklqdq%xmm3, %xmm0
72b1: movhlps%xmm0, %xmm2
72b4: movq%xmm2, %r10
72b9: movq%r10, %r11
72bc: jmpq*%rax

我们能发现，额外的寄存器被引入了。体系结构 101，额外的寄存器意味着额外的开销。寄存器是很贵的！

那么我们有没有办法来迭代上面的超大的 Switch case 呢？估计有同学在想，既然上面的 switch case 超级大，那么我们将其拆分为多个小的函数
这样编译器可以有足够的上下文来优化，同时我们的 perf 也可以精确的去分析每个函数的开销，岂不美哉？

但是又有同学反对了，函数调用会触发 call 的指令，会带来额外的寄存器入栈和出栈的开销，这样会不会又变慢了呢？

那么能不能优化一下呢？答案是可以的，很多同学可能会想到了，tail call

假设我们有这样一段代码

__attribute__((noinline)) void g(int x) {
    printf("Value: %d\n", x);
};
void f(int x) {
    return g(x);
}

我们能看到这样一段汇编

0000000000001140 <g>:
    1140:55                   push   %rbp
    1141:48 89 e5             mov    %rsp,%rbp
    1144:48 83 ec 10          sub    $0x10,%rsp
    1148:89 7d fc             mov    %edi,-0x4(%rbp)
    114b:8b 75 fc             mov    -0x4(%rbp),%esi
    114e:48 8d 3d af 0e 00 00 lea    0xeaf(%rip),%rdi        # 2004 <_IO_stdin_used+0x4>
    1155:b0 00                mov    $0x0,%al
    1157:e8 d4 fe ff ff       call   1030 <printf@plt>
    115c:48 83 c4 10          add    $0x10,%rsp
    1160:5d                   pop    %rbp
    1161:c3                   ret
    1162:66 66 66 66 66 2e 0f data16 data16 data16 data16 cs nopw 0x0(%rax,%rax,1)
    1169:1f 84 00 00 00 00 00 

0000000000001170 <f>:
    1170:55                   push   %rbp
    1171:48 89 e5             mov    %rsp,%rbp
    1174:48 83 ec 10          sub    $0x10,%rsp
    1178:89 7d fc             mov    %edi,-0x4(%rbp)
    117b:8b 7d fc             mov    -0x4(%rbp),%edi
    117e:e8 bd ff ff ff       call   1140 <g>
    1183:48 83 c4 10          add    $0x10,%rsp
    1187:5d                   pop    %rbp
    1188:c3                   ret
    1189:0f 1f 80 00 00 00 00 nopl   0x0(%rax)

其中 call 1140 <g> 指令非常显眼。这也是函数调用本身的开销的一个重要来源

在现在编译器中，存在一种特殊的优化叫作尾递归，即当函数的最后一步是调用另一个函数时，编译器可以优化掉这个调用的开销

我们来测试一下

#include <stdio.h>
__attribute__((preserve_none)) void g(int x);
__attribute__((noinline, preserve_none)) void g(int x){
    printf("Value: %d\n", x);
}

__attribute__((preserve_none)) void f(int x) {
    [[clang::musttail]] return g(x);
}

int main() {
    f(42);
    return 0;
}

我们来看下相关汇编

0000000000001140 <g>:
    1140:55                   push   %rbp
    1141:48 89 e5             mov    %rsp,%rbp
    1144:48 83 ec 10          sub    $0x10,%rsp
    1148:44 89 65 fc          mov    %r12d,-0x4(%rbp)
    114c:8b 75 fc             mov    -0x4(%rbp),%esi
    114f:48 8d 3d ae 0e 00 00 lea    0xeae(%rip),%rdi        # 2004 <_IO_stdin_used+0x4>
    1156:31 c0                xor    %eax,%eax
    1158:e8 d3 fe ff ff       call   1030 <printf@plt>
    115d:48 83 c4 10          add    $0x10,%rsp
    1161:5d                   pop    %rbp
    1162:c3                   ret
    1163:66 66 66 66 2e 0f 1f data16 data16 data16 cs nopw 0x0(%rax,%rax,1)
    116a:84 00 00 00 00 00 

0000000000001170 <f>:
    1170:55                   push   %rbp
    1171:48 89 e5             mov    %rsp,%rbp
    1174:44 89 65 fc          mov    %r12d,-0x4(%rbp)
    1178:44 8b 65 fc          mov    -0x4(%rbp),%r12d
    117c:5d                   pop    %rbp
    117d:e9 be ff ff ff       jmp    1140 <g>
    1182:66 66 66 66 66 2e 0f data16 data16 data16 data16 cs nopw 0x0(%rax,%rax,1)
    1189:1f 84 00 00 00 00 00

我们能看到，f 函数的最后一步是 jmp 1140 <g>，而不是 call 1140 <g>，这就意味着我们在调用 g 函数的时候不会有额外的寄存器分配等传统 call 指令带来的开销。

可能有同学回过味来了，那么这里在做尾递归处理后，感觉完全可以当作一种高性能 Goto 来看嘛。

Bingo，这里其实思路也是差不多这样的，在77年的一篇论文《Debunking the ‘Expensive Procedure Call’ Myth, or, Procedure Call Implementations Considered Harmful, or, Lambda: The Ultimate GOTO》就提到了，高效的过程调用可以和 Goto 性能相近，而在实现上会更简洁。

在 Python 3.14 中，Tail Call Interpreter 的实现就是基于这个思路的。

我们能看到我们对于 opcode 进行 dispatch 的宏进行了尾递归的处理

#   define Py_MUSTTAIL [[clang::musttail]]
#   define Py_PRESERVE_NONE_CC __attribute__((preserve_none))
    Py_PRESERVE_NONE_CC typedef PyObject* (*py_tail_call_funcptr)(TAIL_CALL_PARAMS);

#   define TARGET(op) Py_PRESERVE_NONE_CC PyObject *_TAIL_CALL_##op(TAIL_CALL_PARAMS)
#   define DISPATCH_GOTO() \
        do { \
            Py_MUSTTAIL return (INSTRUCTION_TABLE[opcode])(TAIL_CALL_ARGS); \
        } while (0)
#   define JUMP_TO_LABEL(name) \
        do { \
            Py_MUSTTAIL return (_TAIL_CALL_##name)(TAIL_CALL_ARGS); \
        } while (0)
#   ifdef Py_STATS
#       define JUMP_TO_PREDICTED(name) \
            do { \
                Py_MUSTTAIL return (_TAIL_CALL_##name)(frame, stack_pointer, tstate, this_instr, oparg, lastopcode); \
            } while (0)
#   else
#       define JUMP_TO_PREDICTED(name) \
            do { \
                Py_MUSTTAIL return (_TAIL_CALL_##name)(frame, stack_pointer, tstate, this_instr, oparg); \
            } while (0)
#   endif
#    define LABEL(name) TARGET(name)

那么在保证我们基线性能和 Compute GOTO 甚至更优一点的同时，我们可以得到如下的一些好处

更广泛的平台支持
将 case 拆分后，编译器更不容易犯错，整体的性能的可预测性更强
happy perf
以及我可以用 eBPF 之类的工具做更多的骚操作（

总结

这篇文章差不多就是这样，虽然说是只介绍 Python 3.14 的 Tail Call Interpreter，但是还是完整的介绍了一些整个的一个演进思路

这也带给我一个启发，很多时候，可预测性真的是非常重要的一个特性。

这算是 3.14 中和 remote debug 一起并列为我最喜欢的两个feature，可观测性万岁！

Manjusaka修改