在代码生成流程前冻结事件名catalog,用 AST 扫描 track/emit 调用并强制未知字符串触发构建失败,防止仪表盘分叉。
Invented telemetry names are a silent product bug, and you should reject them before any model writes a comment. Cheap code generation makes extra track() calls feel free, then your dashboards quietly fork into two dialects. This case study walks one small checkout service from catalog freeze through a scanner, a near-miss report, and a review note. You keep the gate deterministic; you ask a model only to explain names that already failed the catalog.
你维护一个小型 Python 结账服务,包含三个真实产品事件和一堆乱糟糟的复制粘贴插桩习惯。一个在退款路径上帮忙的 agent 经常会吐出 order_refunded_v2,因为这个字符串听起来比 order.refunded 更"新"。然后仓库查询就分裂了,告警也漏掉了新名称,直到每周营收复盘才有人发现。失败不是因为缺少 AI。失败是因为接受了未发布的名字,仿佛它们是一个产品决策。
你不需要平台团队或付费可观测性套件来做第一版。你只需要一个冻结的 YAML 文件、一个遍历 track( 和 emit( 的 AST walker,以及一条把未知字符串当作构建失败的 CI 规则。模型之后仍然可以帮忙,但只有在 catalog 已经 say no 之后才行。
你需要一个 teammate 一条命令就能跑、一屏就能看懂的 gate。下面的示例是一个合成服务,不是生产环境声明,所有数字都来自本文的 fixtures。成功的样子是四个你可以笔记本上重放的结果:
你把事件名称当作 API。如果一个字符串不在 events.yaml 里,它就是一个 bug,不是建议。保持值用点号分隔、过去式、平淡无奇,这样生成的代码就无法用一个富有创意的同义词隐藏产品变更。
# events.yaml
version: 1
events:
- name: checkout.started
owner: payments
pii: false
- name: order.placed
owner: payments
pii: false
- name: order.refunded
owner: payments
pii: false
你也冻结你将接受的调用形式,因为 agent 喜欢添加 capture()、log_event() 和 analytics.send() 作为"贴心"封装。对于这个 case,你只允许 track 和 emit,且第一个参数必须是字符串字面量。动态名称故意不在范围内;它们属于后续 RFC,不属于紧急补丁。
扫描器是普通的 Python。你解析每个文件,遍历 ast.Call 节点,收集第一个参数常量。这让 gate 廉价、可审查、独立于任何 vendor prompt。
# scan_events.py
from __future__ import annotations
import argparse
import ast
import pathlib
import sys
from difflib import get_close_matches
import yaml
ALLOWED = {"track", "emit"}
def load_catalog(path: pathlib.Path) -> set[str]:
data = yaml.safe_load(path.read_text())
return {row["name"] for row in data["events"]}
class EventVisitor(ast.NodeVisitor):
def __init__(self, rel: str) -> None:
self.rel = rel
self.hits: list[tuple[int, str, str]] = []
def visit_Call(self, node: ast.Call) -> None:
name = None
if isinstance(node.func, ast.Name):
name = node.func.id
elif isinstance(node.func, ast.Attribute):
name = node.func.attr
if name in ALLOWED and node.args:
arg = node.args[0]
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
self.hits.append((arg.lineno, name, arg.value))
self.generic_visit(node)
def scan(root: pathlib.Path, catalog: set[str]) -> tuple[list[str], list[str]]:
unknown: list[str] = []
near: list[str] = []
for path in root.rglob("*.py"):
if path.name == "scan_events.py":
continue
tree = ast.parse(path.read_text(), filename=str(path))
visitor = EventVisitor(str(path.relative_to(root)))
visitor.visit(tree)
for lineno, func, event in visitor.hits:
loc = f"{visitor.rel}:{lineno} {func}({event!r})"
if event in catalog:
continue
unknown.append(loc)
close = get_close_matches(event, sorted(catalog), n=1, cutoff=0.72)
if close:
near.append(f"{loc} ~ {close[0]}")
return unknown, near
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=pathlib.Path, default=pathlib.Path("."))
parser.add_argument("--catalog", type=pathlib.Path, default=pathlib.Path("events.yaml"))
args = parser.parse_args()
catalog = load_catalog(args.catalog)
unknown, near = scan(args.root, catalog)
if not unknown:
print("ok: all telemetry names are in the catalog")
return 0
print("unknown telemetry names:")
for row in unknown:
print(f" - {row}")
if near:
print("near misses (typo vs new product event):")
for row in near:
print(f" - {row}")
return 1
if __name__ == "__main__":
sys.exit(main())
然后你添加一个 fixture 模块来模拟一个生成的退款补丁。把三个合法事件保留在原位,这样快乐路径在同一文件里仍然可见。
# app/checkout.py
def track(event: str, **payload: object) -> None:
print(event, payload)
def emit(event: str, **payload: object) -> None:
print(event, payload)
def start_checkout(user_id: str) -> None:
track("checkout.started", user_id=user_id)
def place_order(order_id: str) -> None:
emit("order.placed", order_id=order_id)
def refund_order(order_id: str) -> None:
# Invented by a helpful agent during a refund patch.
track("order_refunded_v2", order_id=order_id)
emit("order.refunded", order_id=order_id)
用 CI 运行时相同的方式跑 gate。你应该看到非零退出码和一个针对 order.refunded 的 near-miss。
pip install pyyaml
python scan_events.py --root . --catalog events.yaml
echo $?
失败后的可选解释
只有当扫描器已经失败后,你才把一个紧凑的 bundle 发送给模型。bundle 包含 unknown 行、最接近的 catalog 名称,以及一个硬性指令:不要提议新事件。这个分离很重要,因为模型擅长炮制听起来合理的产品语言,而这恰恰是你要阻止的失败模式。
Disclosure: This article was prepared as part of MonkeyCode's product outreach. If you already have MonkeyCode's free model access, you can send that bundle through it without putting a model in the blocking path. The free server option is useful when you want a scheduled scan of main in addition to pull-request CI, not as a replacement for the AST check.
# explain_near_miss.py (proposal: run only after scan_events.py exits 1)
PROMPT = """You are reviewing a CI failure, not designing analytics.
Unknown event: {unknown}
Closest catalog name: {closest}
Write four sentences for the pull request.
Cite the catalog name. Do not invent a replacement event.
Do not suggest renaming the catalog to match the patch.
"""
把这个代码片段标记为 proposal。你仍然自己决定 HTTP client、secret store 和 comment poster。重要的契约是操作顺序:catalog 第一,相似度第二,文字最后。
Fixtures 结果
用 app/checkout.py 中的三个函数重新运行扫描器,产生一个稳定、平淡的报告。你应该把这些数字当作 fixture 结果,而不是生产基准。
你从这个表格中学到两个运维事实。第一,合法退款事件已经存在,所以发明的别名不是在填补空白。第二,near-miss 行阻止了reviewer争论品味问题;catalog 已经选择了 order.refunded。
你在免费服务器上调度什么
Pull-request CI 捕获新补丁。夜间任务仍然重要,因为有人会在周五提交带 --no-verify 的生成代码。保持任务简单:clone、安装 pyyaml、运行扫描器、只在有未知名称时 page。你不需要 GPU 来做 AST walk,如果树是干净的也不应该等待模型。
# proposal: cron entry on a small always-on box
*/30 * * * * cd /srv/event-gate && git pull --ff-only && python scan_events.py --root ./service --catalog ./events.yaml
如果那台机器是 MonkeyCode 的免费服务器选项,把它用作调度器和最后一个失败报告的 artifact host。不要把 allow/deny 决策移入聊天记录,因为 transcript 不是你可以重放的 diff。
这个设计故意很窄,在你复制到 monorepo 之前你应该读一下这些 gaps。
track(EVENT_REFUND) 和 track(f"order.{action}") 是不可见的analytics.track 按函数名收集,可能对无关的 emit 辅助函数产生假阳性get_close_matches 没有产品 sense。order.returned 可能紧挨 order.refunded,但仍然是一个真正的新事件order.refunded.v2,除非 prompt 禁止新名称Fixture 退款函数同时发出发明名称和合法名称,这比大多数生成补丁更友好。
谁不应该用这个方法
如果你的产品确实需要运行时定义的事件名称,比如客户特定的工作流步骤,你不应该冻结 catalog。你不应该指望这个扫描器理解 protobuf 生成的 stub 和 enum 别名。你也不应该雇一个模型来发明第一个 catalog;那只是把猜测洗成 YAML。如果你没有权力拒绝 pull request,gate 就会变成一个评论机器人,而评论机器人保护不了 dashboards。
廉价生成不会让未发布 telemetry 的存储更便宜。它只是让额外名称更容易输入,这恰恰是数据契约的反面。你用可以 diff 的文件、可单元测试的 walker、不允许扩展词汇的模型来保护契约。有趣的 review 问题不再是"这个名字听起来清楚吗?"而是"我们已经为这个事实命名了吗?"
把人类决策放在 catalog 变更时,而不是补丁时。当有人真的需要 order.refund.failed 时,他们在同一 pull request 里添加一行 YAML 和 owner,只有到那时才重新生成代码。这个序列比让 agent 即兴发挥更慢,而那个慢就是重点。
如果你想要一个小 next step,添加一个单元测试,给 scan_events.py 喂一个包含单个发明字面量的临时目录。之后如果那个工作了,把同一个命令放在免费服务器上,把模型留在失败的 exit code 后面。当你已经信任 catalog 多于 prompt 时,MonkeyCode 的免费模型访问就足够用于可选的 review 段落了。