6.1 自建最小可观测：纯标准库完整脚本

6.1 自建最小可观测：纯标准库完整脚本

本节摘要：这是全书的落地时刻。前五章立了四类零件——tracer（第 2.1 节）、token 三本账（第 3.1 节）、在线评测探针（第 4.1 节）、预算三线（第 3.3 节）——本节把它们接进同一个进程，成为一个约 170 行、纯标准库、含 MOCK 模式的完整脚本 mini_obs.py。MOCK 模式的意思是：模型调用用假实现顶替（返回假答案与假 usage），不接任何供应商也能跑通全链路——span 树、记账、抽样裁判、查线、报告一次到位。跑通之后，本节给出从 MOCK 到生产的四类替换点：模型调用、数据落库、定时任务、告警路由；会话熔断（第 3.3 节的 SessionCap）如何并入，留作练习（附录 C 练习五）。理解了这个脚本，第 6.2 节的工具选型才有判断力：你知道每个平台报表背后，自己要的是什么。

学习目标

• 画出四类零件的总装接线图，说清一次请求里它们的执行顺序
• 完整运行 mini_obs.py，读懂四张输出：span 树、成本日结、质量日结、告警
• 会改脚本：调预算线、调抽样率、构造零命中场景
• 列出从 MOCK 到生产必须替换的四类点位

一、总装接线图

四类零件的接口在前半都已留好，总装只是接线：

一次 ask() 请求的执行顺序
┌───────────────────────────────────────────────────────────┐
│ 1 request span 开树（trace_id/user/session 打头）          │
│ 2 retrieve span → MOCK 检索（hits 写进 span 属性）         │
│ 3 llm.call span → MOCK 模型 → usage 挂 span（2.2 口径）    │
│ 4 TokenLedger.record() —— 三本账 + 分摊键 + trace_id       │
│ 5 Probe.maybe_record() —— 抽样裁判，评分带 trace_id 回写   │
│ 6 budget_check() —— 消耗速率对三线，越线进告警队列         │
└───────────────────────────────────────────────────────────┘
report() 一键输出：span 树 / 成本日结 / 质量日结 / 告警
​

三条接线纪律（都是前半立过的规矩）：usage 必须来自（这里是 MOCK 的）响应字段并挂在 llm.call span 上（第 2.2 节）；记账行必须带分摊键与 trace_id（第 3.2 节未打标率要趋近 0）；裁判抽样率随评分行记录（第 2.3 节）。

二、完整脚本（MOCK 模式，纯标准库）

# mini_obs.py —— 最小可观测套件总装：tracer+三本账+探针+告警线（纯标准库，MOCK 模式）
# 零件来源：tracer（2.1）· usage 挂 span（2.2）· 三本账（3.1）· 三线告警（3.3）· 探针（4.1）
import random
import time
import uuid
from collections import defaultdict
from contextlib import contextmanager
from dataclasses import dataclass, field

@dataclass
class Span:
    name: str
    span_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    parent_id: str | None = None
    start: float = 0.0
    end: float = 0.0
    attrs: dict = field(default_factory=dict)
    children: list = field(default_factory=list)


class Tracer:
    """栈维护父子；进入压栈，退出弹栈挂到父节点；finally 保证异常也闭合。"""

    def __init__(self):
        self.stack: list[Span] = []
        self.finished: list[Span] = []

    @contextmanager
    def span(self, name: str, **attrs):
        parent = self.stack[-1] if self.stack else None
        sp = Span(name, parent_id=parent.span_id if parent else None, attrs=attrs)
        sp.start = time.perf_counter()
        self.stack.append(sp)
        try:
            yield sp                       # 业务可在 span 进行中补属性
        finally:
            sp.end = time.perf_counter()
            self.stack.pop()
            (parent.children if parent else self.finished).append(sp)


def render(sp: Span, depth: int = 0) -> None:
    ms = (sp.end - sp.start) * 1000
    attrs = f"  {sp.attrs}" if sp.attrs else ""
    print("  " * depth + f"- {sp.name}  {ms:.1f}ms{attrs}")
    for ch in sp.children:
        render(ch, depth + 1)


PRICE = {"demo-large": {"input": 3.00, "output": 15.00,           # 美元/百万 token
                        "cache_read": 0.30, "cache_write": 3.75}}  # （示意，换官方定价）
KEYMAP = {"input_tokens": "input", "output_tokens": "output",
          "cache_read_tokens": "cache_read", "cache_write_tokens": "cache_write"}


class TokenLedger:                       # 三本账：每行 = token 三账 + 分摊键 + trace_id
    def __init__(self):
        self.rows: list[dict] = []

    def record(self, trace_id: str, feature: str, user: str, model: str, **tok) -> None:
        p = PRICE[model]
        cost = sum(n * p[KEYMAP[k]] for k, n in tok.items()) / 1_000_000
        self.rows.append({"trace_id": trace_id, "feature": feature,
                          "user": user, "cost": cost, **tok})

    def summary(self) -> None:
        agg = defaultdict(lambda: [0.0, 0, 0])
        for r in self.rows:
            a = agg[r["feature"]]
            a[0] += r["cost"]; a[1] += r.get("input_tokens", 0) + r.get("cache_read_tokens", 0)
            a[2] += r.get("output_tokens", 0)
        print("== 成本日结 ==")
        for f, (c, i, o) in sorted(agg.items()):
            print(f"  {f:<6s} {c:.4f} 美元 | 输入 {i:,} | 输出 {o:,}")
        print(f"  合计  {sum(v[0] for v in agg.values()):.4f} 美元")


def rule_judge(answer: str, hits: int) -> tuple[int, list[str]]:
    score, notes = 8, []                 # 四维三档 rubric 的规则化简写（4.1）
    if hits == 0:
        score -= 3; notes.append("检索零命中")
    if len(answer) < 20:
        score -= 2; notes.append("输出过短")
    if hits > 0 and "无法回答" in answer:
        score -= 2; notes.append("有据仍拒答")
    return score, notes


class Probe:                             # 抽样裁判：点踩强制入库，抽样率随行记录（4.1/2.3）
    def __init__(self, rate: float = 0.5, seed: int = 7):
        self.rate, self.rng = rate, random.Random(seed)
        self.rows: list[dict] = []

    def maybe_record(self, trace_id: str, feature: str,
                     answer: str, hits: int, negative: bool) -> None:
        if negative or self.rng.random() < self.rate:    # 点踩强制入库（4.1）
            score, notes = rule_judge(answer, hits)
            self.rows.append({"trace_id": trace_id, "feature": feature, "score": score,
                              "notes": notes, "rate": self.rate})

    def summary(self) -> None:
        agg = defaultdict(lambda: [0, 0])
        for r in self.rows:
            agg[r["feature"]][0] += 1; agg[r["feature"]][1] += r["score"] >= 6
        print("== 质量探针日结 ==")
        for f, (n, ok) in sorted(agg.items()):
            print(f"  {f:<6s} 样本 {n} | 合格率 {ok / n:.0%}")


def budget_check(day_limit: float, spent: float,
                 warn: float = 1.5, pause: float = 3.0) -> list[str]:
    rate = spent / day_limit              # 消耗速率法 + 三线（3.3，参数示意）
    if rate >= pause:
        return [f"[暂停] 消耗速率 {rate:.1f} 倍于日预算，执行降级"]
    if rate >= warn:
        return [f"[提醒] 消耗速率 {rate:.1f} 倍于日预算，查 TopN"]
    return []


class MiniObs:
    """ask() = span 树 → MOCK 模型 → 记账 → 抽样裁判 → 查线；report() 一键四表。"""

    def __init__(self, day_budget: float = 0.02):
        self.tracer, self.ledger, self.probe = Tracer(), TokenLedger(), Probe()
        self.day_budget, self.alerts, self.rng = day_budget, [], random.Random(21)

    def mock_llm(self, question: str, turn: int, hits: int) -> tuple[str, dict]:
        """MOCK 模型（示意）：生产替换为供应商 API，usage 取响应字段（2.2 口径）。"""
        if hits == 0:
            answer = "无法回答该问题"
        else:
            answer = f"根据知识库第 {hits} 条：{question[:10]}…的处理路径见设置页。"
        return answer, {"input_tokens": 900 + 700 * turn,   # 多轮历史重发（示意）
                        "output_tokens": 350 + 40 * turn,
                        "cache_read_tokens": 600 * turn,     # 系统提示词缓存命中（示意）
                        "cache_write_tokens": 600 if turn == 0 else 0}

    def ask(self, question: str, user: str, session: str, turn: int = 0) -> str:
        trace_id = uuid.uuid4().hex[:16]
        with self.tracer.span("request /answer", trace_id=trace_id,
                              user=user, session=session):
            with self.tracer.span("pipeline rag", prompt_ver="v3"):
                with self.tracer.span("retrieve") as sp_r:
                    hits = self.rng.choice((0, 3, 3, 5))    # 1/4 概率零命中（示意）
                    sp_r.attrs["hits"] = hits
                with self.tracer.span("llm.call", model="demo-large") as sp:
                    answer, usage = self.mock_llm(question, turn, hits)
                    sp.attrs.update({f"gen_ai.usage.{k}": v for k, v in usage.items()})
                self.ledger.record(trace_id, "客服问答", user, "demo-large", **usage)
                neg = hits == 0 and self.rng.random() < 0.3  # 坏答案偶发点踩（示意）
                self.probe.maybe_record(trace_id, "客服问答", answer, hits, neg)
        self.alerts += budget_check(self.day_budget,
                                    sum(r["cost"] for r in self.ledger.rows))
        return answer

    def report(self) -> None:
        print("== span 树（最近一条 trace）==")
        render(self.tracer.finished[-1])
        self.ledger.summary()
        self.probe.summary()
        print("== 告警 ==")
        print("\n".join(self.alerts) if self.alerts else "  预算内，无线")


if __name__ == "__main__":
    obs = MiniObs(day_budget=0.02)         # 预算故意调小便于演示触发（示意）
    q = "怎么查看这个月的账单？"
    for t in range(3):
        print(f"第 {t + 1} 轮 →", obs.ask(q, user="u-42", session="s-1001", turn=t))
    print()
    obs.report()
​

三、运行输出与解读

运行（实测，耗时因机器而异；随机数因固定种子而可复现，trace_id 每次运行不同）：

第 1 轮 → 根据知识库第 3 条：怎么查看这个月的账单…的处理路径见设置页。
第 2 轮 → 根据知识库第 5 条：怎么查看这个月的账单…的处理路径见设置页。
第 3 轮 → 根据知识库第 5 条：怎么查看这个月的账单…的处理路径见设置页。

== span 树（最近一条 trace）==
- request /answer  0.1ms  {'trace_id': 'f5220f812bfe4d3e', 'user': 'u-42', 'session': 's-1001'}
  - pipeline rag  0.1ms  {'prompt_ver': 'v3'}
    - retrieve  0.0ms  {'hits': 5}
    - llm.call  0.0ms  {'model': 'demo-large', 'gen_ai.usage.input_tokens': 2300, 'gen_ai.usage.output_tokens': 430, 'gen_ai.usage.cache_read_tokens': 1200, 'gen_ai.usage.cache_write_tokens': 0}
== 成本日结 ==
  客服问答   0.0347 美元 | 输入 6,600 | 输出 1,170
  合计  0.0347 美元
== 质量探针日结 ==
  客服问答   样本 2 | 合格率 100%
== 告警 ==
[提醒] 消耗速率 1.7 倍于日预算，查 TopN
​

逐段读：span 树里 usage 以 gen_ai.usage.* 键挂在 llm.call 上（第 2.2 节口径，也是给未来接 OTel 兼容后端留的语义退路）；成本日结显示三轮共输入 6,600 token（含缓存读）、输出 1,170 token、合计 0.0347 美元——日预算 0.02 美元（示意值）已花到 1.7 倍，提醒线触发；探针抽到 2 个样本全部合格（零命中的坏答案若被抽到，rule_judge 会扣 3 分）。第 3.3 节的三线语义完整保留：把 day_budget 调到 0.01 再跑，输出会变成暂停线。

建议动手实验三处：把 mock_llm 里 hits 恒置 0，看探针合格率跳水与"检索零命中"原因码；把循环改成 10 轮，看输入 token 如何随多轮历史线性上涨（第 3.1 节"输入膨胀"的活教材）；给 ask() 开头并入第 3.3 节的 SessionCap 做会话预检（参考答案见附录 C 练习五）。

四、从 MOCK 到生产：四类替换点

⚠️ 这个脚本的定位是"理解骨架"，不是生产系统：单进程内存态、无采样留存、无脱敏、熔断未并入（练习五）。让它上生产前，先把第 2.3 节的数据策略补上——否则观测系统自己先变成成本与合规问题。

本节要点回顾

• 总装一次请求的顺序：span 树 → 模型调用记账 → 抽样裁判 → 预算查线；report 一键出四张表。
• 三条接线纪律：usage 挂 llm.call span、记账行带分摊键与 trace_id、抽样率随评分行记录。
• MOCK 模式让全链路零依赖可跑；日预算调小即可演示三线逐级触发。
• 从 MOCK 到生产替换四类点位：模型调用、落库（含采样留存脱敏）、定时聚合、告警路由。

这约 170 行跑起来的那一刻，你已经知道 span 怎么长出来、账怎么算出来、告警怎么触发。第 6.2 节拿这个判断力去回答下一个问题：什么时候该停止自建、换用现成工具——以及怎么挑。