LangChain系列之评测与监控
评测与监控
有一回产线做检索策略调整。
上线之前,我们人工抽了 20 条 query 做前后对比。20 条全部变好,有几条好得相当明显,连措辞都更像人话了。看完这 20 条,我们对这次调整很有信心,直接全量放了出去。
第三天,客诉开始抬头。
复盘花了半天,结论只有一句话:那 20 条抽样,来自最好答的那一个类目。
翻车的是一整类长尾改写。它们既不在我们抽的样本里,也不在任何人的印象里。我自己都没想到那类问法真的有人这么问。
那天之后我明白了一件事:没有评测集的时候,人只会验证自己想验证的那部分。
于是有了这篇。前面几篇我们一直在回答「怎么把它做出来」,这一篇换个问题:做完之后,凭什么说它变好了。
先把这一篇里最硬的四个数字摆在开头,它们都在这台机器上跑出来,全程没有配任何云端账号。
- 一次带工具调用的 Agent 请求,本地能抓到 5 条 run,但 5 条的
parent_run_id全是None,调用链是拍平的。 - 想知道这次请求花了多少 token,得去
message.kwargs.usage_metadata和run.extra两处取,run上没有现成属性。 - 官方评测器的入参
data有 3 种写法,会给你 3 种不同的报错,其中 1 种看起来还跑通了。 - 没有云端凭据,整套评测仍然能跑通。代价是拿不到回链 URL。
评测的第一个门槛不是钱,是你有没有一个能重复跑的标准答案集。
一、先把三件事分开
大多数团队说「我们要做评测」,其实混着三件不同的需求。
它们共用一套底座,但判据、成本、失败后果完全不同。混在一起谈,就会变成一场没有结论的会。
第一件,开发期回归。 我改了这一版,有没有把别的弄坏。它要的是快,能在本地跑,结果能和上一版对比。它对绝对分数不敏感,只关心「有没有变差」。
第二件,上线前门禁。 这一版能不能发。它要的是阈值和阻塞能力,必须能进 CI,跑不过就得拦住发布。它不接受「大概差不多」。
第三件,上线后监控。 线上正在发生什么。它要的是低成本、能告警、能回捞样本。它不看单条质量,只看那批数字有没有跳出正常范围。
| 开发期回归 | 上线前门禁 | 上线后监控 | |
|---|---|---|---|
| 触发时机 | 每次改动 | 发版前 | 持续 |
| 数据来源 | 冻结的评测集 | 冻结的评测集 | 线上真实流量 |
| 判据 | 与上一版对比 | 绝对阈值与基线 | 滚动基线区间 |
| 失败后果 | 自己修 | 拦住发布 | 告警与回捞 |
| 样本量 | 几十条 | 几百条 | 全量 |
| 最小实现 | 一个脚本 | 脚本 + 非零退出码 | 定时任务 + 指标看板 |
表里最后一列是我最想强调的:三件事可以共用一个底座,但不能用同一套指标。
开发期你要的是「哪些样本变差了」,是逐条的。监控你要的是「整体有没有异常」,是聚合的。
把逐条明细塞进监控看板,没人会看。把聚合数字拿去做回归定位,你找不到问题在哪。
三件事在数据上是连起来的。线上出了问题回捞样本,样本进评测集,评测集喂给门禁。
后面就按这个顺序往下走,一道关口一章。
二、地基是 trace:本地能抓到什么
平台的卖点是「看得见每一步」。那就先验证一件事:不上云,本地到底能看见多少。
答案是基本都看得见。
collect_runs() 在没有任何凭据的环境里直接可用。它纯走内存,全程 0 次网络请求。
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tracers.context import collect_runs
agent = create_agent(model=fake_model, tools=[get_order], system_prompt="你是订单助手。")
with collect_runs() as cb:
agent.invoke({"messages": [HumanMessage("查订单 A1001")]})
runs = cb.traced_runs
print([(r.run_type, r.name) for r in runs])
我跑了一次带工具调用的请求,抓到 5 条 run:2 条 llm、2 条 tool、1 条 chain。根 run 的名字是 LangGraph。
然后我把这 5 条的关系打出来,看到一个反直觉的结果。
| 检查项 | 实测结果 |
|---|---|
| run 总数 | 5(llm × 2、tool × 2、chain × 1) |
parent_run_id |
全部为 None |
dotted_order |
全部是单段,没有层级路径 |
根 run 的 child_runs |
0 |
每条 run 的 start_time |
完整可用 |
dotted_order 首段 |
就是该 run 的起始时间戳 |
5 条 run 全部平级。没有父子关系,根 run 下面一个孩子都没有。
为了确认这是谁的行为,我做了一组对照。
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
chain = ChatPromptTemplate.from_template("{q}") | fake_model | RunnableLambda(lambda x: x)
with collect_runs() as cb:
chain.invoke({"q": "hello"})
root = [r for r in cb.traced_runs if r.parent_run_id is None]
print(len(root), len(root[0].child_runs) if root else 0)
普通链式 Runnable 的父子关系是完整的,根 run 名下有 2 个直接子 run。
结论就出来了:拍平是 LangGraph 执行器的行为,跟 collect_runs 没关系。 我又用手写的 StateGraph 试了一次,2 条 run 同样全部平级。
这件事的实际影响只有一个:本地想重建调用顺序,不能靠父子关系,只能按 start_time 排序。
ordered = sorted(runs, key=lambda r: r.start_time)
for r in ordered:
print(r.start_time.strftime("%H:%M:%S.%f")[:-3], r.run_type, r.name)
跑出来的顺序是:chain LangGraph → llm → tool → llm,正是这次请求真实的执行次序。
顺带纠一个很常见的 API 误用。
tracing_v2_enabled() 返回的是 LangChainTracer,它是面向云端上报的,没有 traced_runs 属性。你写 with tracing_v2_enabled() as tracer: ... tracer.traced_runs 会直接抛 AttributeError。想拿本地 run,只能用 collect_runs()。
地基到这里就通了。本地能看见每一步,一分钱不花。
三、评测集从哪来
这是产线里最痛的一问。
很多团队卡在这里,卡的不是技术,是「谁来标」。我把它拆成四步,每一步都能立刻动手。
第一步,捞。 从线上 trace 里筛出值得评的样本。筛选条件就写在那几条信号上:报错的、重试过的、人工纠正过的、用户明确表示不满意的。
这四类样本的价值远高于随机抽样。随机抽样里大半是「本来就答得对」的简单样本,评了也没信息量。
第二步,脱。 脱敏必须在落盘之前做。这件事放到后面第九章单独讲,因为官方的默认规则里,国内云厂商的 AccessKey 一个都不认。
第三步,标。 标注这件事,尺度比人数重要。
有标准答案的,用精确匹配就行。没有标准答案的,只标两档:可接受、不可接受。不要硬凑 1 到 5 分。
原因很简单:让标注员给「这条回复打几分」,不同人之间的标准差会大到让分数没有意义。让他判断「这条能不能直接发给用户」,一致性立刻上来了。
第四步,冻。 评测集一旦开始用于门禁,就要冻结版本。改样本等于改尺子,你的历史分数全作废。
冻的方式很简单,评测集文件带上版本号,每次跑评测把版本号记进结果里。
{"id":"E01","tier":"easy","msg":"我要退货,订单 A1001 还没到","intent":"refund","priority":"normal","order_id":"A1001"}
{"id":"E04","tier":"hard","msg":"东西坏了,但我更想直接退钱","intent":"refund","priority":"high","order_id":null}
{"id":"E06","tier":"hard","msg":"我不急,就是想问问保修到什么时候","intent":"warranty","priority":"low","order_id":null}
注意每行都有一个 tier 字段。这不是装饰。
| 步骤 | 做法 | 判据 |
|---|---|---|
| 捞 | 从 trace 里筛错误、重试、人工纠错、用户不满 | 四类信号的数量占比 |
| 脱 | 落盘前脱敏 | 敏感字段命中数归零 |
| 标 | 有答案的精确匹配,没答案的只标两档 | 标注者之间的一致率 |
| 冻 | 文件带版本号,结果里记版本 | 版本变更必须走评审 |
有一条硬规矩:评测集里必须留一批「本来就会做对」的简单样本。
我用 6 条样本做了个演示,3 条 easy、3 条 hard。另做一个退化版本:常见类目照旧答对,长尾和多意图全部塞进常见类目里。
跑完两个版本的分数是这样的。
| 版本 | easy 档意图准确率 | hard 档意图准确率 | 整体 |
|---|---|---|---|
| 基线 | 1.0 | 1.0 | 1.0 |
| 退化版 | 1.0 | 0.333 | 0.667 |
只看总分,你看到的是掉了 33%。看分档,你才看到真相:简单样本一条没坏,困难样本崩了三分之二。
这就是为什么必须有简单样本。没有它,分数掉了你分不清是模型退化,还是这批数据本身变难了。
四、官方评测器实测:三种写法,三种报错
这一章是全篇的实测重心。
langsmith 的 evaluate() 是官方入口,网上教程很多。我把 3 种 data 写法各跑了一遍,结果是 3 种不同的报错。
第一种写法,data 传 list[dict],用默认上传模式。
from langsmith import evaluate
data = [
{"inputs": {"q": "hello"}, "outputs": {"answer": "HELLO"}},
{"inputs": {"q": "world"}, "outputs": {"answer": "WORLD"}},
]
evaluate(my_target, data=data, evaluators=[exact_match])
报错是 AttributeError: 'dict' object has no attribute 'dataset_id'。
这个报错很容易让人往凭据方向排查。我特意配上了一个假的 API key 再跑一次,报的是同一个错。
所以它跟有没有账号无关,是 data 形态的问题:evaluate 的默认模式会把数据当成平台的 dataset 去引用,而 dict 上没有 dataset 的标识字段。
第二种写法,加个开关 upload_results=False。
res = evaluate(my_target, data=data, evaluators=[exact_match], upload_results=False)
print(res) # 看起来成功了
rows = list(res) # 崩在这里
evaluate() 正常返回,返回值打印出来也很正常。但只要你一想取结果,立刻抛 AttributeError: 'dict' object has no attribute 'modified_at'。
这个坑比第一个危险。第一个当场报错,你会去查。第二个表面成功,你可能写完脚本才发现结果一条都读不出来。
第三种写法,把 data 换成真正的 Example 对象。
import uuid
from langsmith import evaluate
from langsmith.schemas import Example
ds = uuid.uuid4()
data = [
Example(
id=uuid.uuid4(),
dataset_id=ds,
inputs={"q": "hello"},
outputs={"answer": "HELLO"},
created_at="2026-09-17T00:00:00Z",
)
]
res = evaluate(my_target, data=data, evaluators=[exact_match], upload_results=False)
rows = list(res)
for row in rows:
ex, run = row["example"], row["run"]
scores = {r.key: r.score for r in row["evaluation_results"]["results"]}
print(ex.inputs, run.outputs, scores)
这一版完全跑通。逐条能拿到输入、输出、每个 evaluator 的分数,能算均值,能重复迭代。全程 0 次 HTTP 请求。
离线模式的可用边界也一并测出来了。
| 属性 / 方法 | 离线模式(upload_results=False) |
|---|---|
experiment_name |
可用 |
get_dataset_id() |
可用 |
experiment_id |
✘ ValueError: Experiment not started yet |
url |
✘ ValueError: Experiment not started yet |
comparison_url |
✘ 同上 |
| 逐条明细迭代 | 可用 |
能跑门禁,回链不到平台页面。这是离线评测的真实边界,够用。
真正跑起来的时候还有一处要注意的地方:默认上传模式下没有凭据,evaluate() 会当场抛异常,不会静默跳过。
我跑那一组时得到的是 LangSmithAuthError: Authentication failed for .../sessions,整个评测直接中断。所以「本地练手用默认模式」这个想法行不通,必须显式写 upload_results=False。
还有个细节值得知道:upload_results=False 时,evaluate 内部会把追踪上下文设成 local 模式(源码在 langsmith/evaluation/_runner.py)。它管得住 langsmith 自己的 @traceable run,但管不住 LangChain 的 run。
如果你的 target 里跑了 LangChain 的 Agent,日志里会刷出一片 401。这些 401 来自后台上报线程,不阻塞评测,但会污染 CI 日志。
想彻底消掉它,在 target 内部显式关一次:
from langsmith import run_helpers as rh
def target(inputs):
with rh.tracing_context(enabled=False):
return my_agent.invoke({"messages": [HumanMessage(inputs["msg"])]})
我实测这一行能把那批 401 清零。顺带说一句,LANGSMITH_TRACING=false 这个环境变量在 evaluate 里是不起作用的,别指望它。
最后提醒一个接口变动:StringEvaluator 自 0.5.0 起已废弃,源码里明确标注,官方指向 openevals。网上大量示例还在用它,抄的时候注意。
如果连 evaluate 都不想用,还有一条更朴素的路:自己取 Example、自己造 Run,直接调 evaluator 函数。全程不需要 Client,也不需要网络。它就是几个函数的组合,你完全可以自己写。
data 写法 |
默认上传 | upload_results=False |
|---|---|---|
list[dict] |
✘ dataset_id 报错 |
✘ 表面成功,迭代时 modified_at 报错 |
list[Example] |
✘ 无凭据时 401 中断 | ✔ 完全可用,0 次 HTTP |
生成器(Example 流) |
✘ 同上 | ✔ 可用 |
五、谁来当裁判:LLM-as-judge 的用法与边界
评分从哪来,前面解决了一半。另一半是「谁来判」。
先把一条原则放在最前面:能用规则判的,绝不交给模型。
有标准答案的,精确匹配。结构化字段的,写断言。要求必须包含某几个词的,查字符串。
这几种判定确定性 100%,成本接近 0,而且永不变卦。它们应该覆盖你评测集里的大多数样本。
只有那些真的没法写规则的,才轮到模型出手。比如「这条回复有没有真的解决用户的问题」。
交给模型的时候,要把裁判本身当成一段代码来管。有四件事必须做。
第一件,判据写成量规,用结构化输出强制返回。
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage, SystemMessage
class Verdict(BaseModel):
score: int = Field(ge=1, le=5, description="1 到 5 分")
reason: str = Field(max_length=40, description="20 字以内理由")
JUDGE_SYS = """你是客服回复质量评审。按以下量规打分:
5 分:直接回答了用户的问题,给出可执行的具体步骤,并说明了后续。
3 分:回答了问题但缺少具体步骤或后续说明。
1 分:没有回答用户的问题,或只给了无法执行的笼统建议。"""
judge = llm.with_structured_output(Verdict)
verdict = judge.invoke([SystemMessage(JUDGE_SYS), HumanMessage(prompt)])
我实测这套在本地的 9B 模型上可用,with_structured_output 能稳稳拿到 score 和 reason,不用自己写正则去抠。
第二件,成对比较优先于绝对打分。
绝对分数有个致命问题:不同批次不可比。今天模型心情好打 4 分,明天打 3 分,你不知道是系统变差了还是裁判变严了。
成对比较只问一个问题:这两个哪个更好。它的可比性天然更强,因为参照物就在同一段 prompt 里。
第三件,位置偏差要主动测。
这是我这篇里最想讲的一个实测。
我准备了 2 个用户问题、每条 3 个质量档次的候选回复,组成 3 组对比,每组跑两个顺序、各跑 2 轮。一共 12 组有效对照。
结果是 翻转 0 组。
12 组里,交换 A/B 位置之后结论全部稳定,都指向了更优的那一条。这个 9B 裁判在 temperature=0 下没有表现出位置偏差。
这个结果和我在很多文章里看到的不一样,所以我得把话说清楚:这是 12 组小样本的结果,不能推广成「位置偏差不存在」。
我做的判断是:位置偏差值得测,但不必先假设它一定会出问题。测它的成本很低,两个顺序各跑一遍就有答案。两组结论不一致的样本单独挑出来人工看,这一步不能省。
第四件,裁判也要有回归。
换了裁判模型,或者同一个模型升了版本,必须在一小批人工标注过的样本上先验证一致性,再用它跑全量。否则你会把裁判的变化,当成系统的变化。
这件事我在这次实测里就撞上了。
有一条对比是「mid 对 bad」。按我的预设,mid 那条更好。但裁判在两个顺序、两轮里都稳定地选了 bad。
我回头重读了两条回复,发现是我的标注有问题。
- mid 写的是「在发票管理里可以改。」
- bad 写的是「改不了,只能重开。」
真实规则是:发票不能直接改,必须先作废再重开。所以那条 mid 的答案其实是错的,裁判选 bad 反而更准。
裁判和人不一致时,先别急着说裁判错。 很可能你的标注标准本身就有争议。这件事只有把裁判放进回归里跑,才会被发现。
最后算一笔裁判的账。
| 裁判任务 | 调用次数 | 输出 token | 其中思考过程 | 单次平均耗时 |
|---|---|---|---|---|
| 绝对打分 | 6 | 1320 | 1170(88.6%) | 3.03 s |
| 成对比较 | 24 | 25138 | 24922(99.1%) | 13.18 s |
绝对打分每次只回一个 25 token 左右的小 JSON。成对比较更夸张,输出里 99.1% 是模型的思考过程。
按「可见回复的长度」估算裁判成本,会低估一个数量级。 估成本要按总输出 token 算,推理型模型的账单几乎全在思考上。
六、回归测试:让评测不烧钱
三道关口里,最容易被钱劝退的是回归测试。
每次改动都跑几百条全量评测,账单很快会变得不好看。省钱有三条路,从省到花。
第一路,能离线就离线。
把模型响应按输入哈希落盘,重跑时直接读盘。评测跑一百遍,也只花第一遍的钱。
import hashlib, json
from pathlib import Path
class Cassette:
def __init__(self, d: Path):
self.dir = d
self.dir.mkdir(exist_ok=True)
def key(self, model_id, system_prompt, question):
raw = f"{model_id}\x00{system_prompt}\x00{question}"
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
def get(self, k):
f = self.dir / f"{k}.json"
if not f.exists():
return None
try:
return json.loads(f.read_text(encoding="utf-8"))["answer"]
except Exception:
return None # 缓存坏了就回退到真实调用
def put(self, k, answer):
(self.dir / f"{k}.json").write_text(
json.dumps({"answer": answer}, ensure_ascii=False), encoding="utf-8")
我跑了 4 轮看它的实际效果。
| 轮次 | 样本 | 真实调用 | 命中 | 耗时 |
|---|---|---|---|---|
| 第 1 轮 冷启动 | 6 | 6 | 0 | 0.49 s |
| 第 2 轮 热重跑 | 6 | 0 | 6 | 0.02 s |
| 第 3 轮 扩了 2 条 | 8 | 2 | 6 | 0.17 s |
| 第 4 轮 改了 system prompt | 6 | 6 | 0 | 0.49 s |
第 2 轮真实调用 0 次,耗时从 0.49 秒降到 0.02 秒。耗时跟机器有关,你的读数会不一样;调用次数和命中数是确定的。
第 3 轮只在新增的 2 条上付钱,老样本全部命中。
第 4 轮是最值得看的一轮。我只在 system prompt 里加了一句「判断意图与优先级」,6 条全部未命中。缓存 key 里必须包含 prompt,少这一项你会拿到过期答案,而且是静默的。
回放缓存也有它管不了的事,动手前要分清。
它测的是「你的提示词、解析逻辑、编排流程改了之后结果变不变」,这部分确定、可比较,非常适合回放。
它测不了「模型本身换了」。缓存命中的时候你根本没调模型,新模型的表现自然看不出来。所以换模型的那一版,必须跑一次全量真实调用。
同一个道理,任何依赖采样随机性的环节,回放也测不出来。
我的做法是分两条线:日常改动用缓存跑,换模型和发版前跑真实调用。两条线的样本是同一套,只有执行方式不同。
第 2 路是官方给的机制。
langsmith.test(等价于 pytest.mark.langsmith)能把 pytest 用例接到追踪上,配 LANGSMITH_TEST_CACHE 目录把上游 API 调用缓存到磁盘。官方建议把缓存文件提交进仓库,这样 CI 里跑回归几乎不花钱。
这里有个本机实测的提醒:这条能力需要额外装 vcrpy,环境里默认没有。 我这边的情况是 langsmith 装好了,vcrpy 是空的,得单独 pip install "langsmith[vcr]"。不装的话,缓存开关是静默不生效的,你看不出来。
第 3 路是分层。
| 评测频率 | 样本量 | 数据来源 | 单次成本 |
|---|---|---|---|
| 每次改动 | 20 到 50 条 | 回放缓存 | 接近 0 |
| 每天一次 | 100 条左右 | 缓存 + 少量真实调用 | 很低 |
| 发版前 | 全部样本 | 真实模型 | 一次性 |
| 大版本前 | 全部样本 + 新采样 | 真实模型 + 人工抽检 | 最高 |
一句话收口:评测集不是越大越好,是越稳定越好。 一个 50 条、半年没改过、每版都跑的集合,比一个 500 条、每月换一批的集合有用得多。
还有一件必须提前知道的事,它属于省钱路上的暗坑。
@traceable 装饰器在没有凭据时会静默失效。函数正常返回,但函数体里 get_current_run_tree() 返回的是 None,全程 0 次 HTTP 请求。
也就是说你以为什么都记下来了,实际上什么都没记。写评测脚本的时候,别把「我在用 @traceable」当成「我在采集数据」。
真正能拿到本地数据的,还是第 2 章那个 collect_runs()。
七、上线监控:指标定几个就够
先说一句劝退的话:监控指标超过七八个,就等于没有监控。
没有人会每天看 20 个数字。指标多到看不过来的时候,团队的做法就是全都不看,等出事再说。
我建议的最小可用集合是 6 个。每一个都对应一种具体的翻车方式。
| 指标 | 取值路径 | 它对应的翻车 |
|---|---|---|
| 输入 / 输出 token | run.outputs["generations"][0][0]["message"]["kwargs"]["usage_metadata"] |
成本失控,账单漂移 |
| 工具调用次数 | run.extra["tool_call_count"] |
模型开始反复调同一个工具 |
| 单步耗时 | (run.end_time - run.start_time).total_seconds() |
上游接口变慢 |
| 是否出错 | run.error(无错为 None) |
工具挂了、解析失败 |
| run 类型分布 | run.run_type(llm / tool / chain) |
该调工具的没调,绕过去了 |
| 空回答率 | 输出内容长度为 0 | 模型退化或提示词冲突 |
取值路径这一列都是实测确认过的,直接抄就行。
特别注意第一行:run 上没有 total_tokens、prompt_tokens 这类现成属性,getattr 取不到。message 在 outputs 里是序列化后的 dict,真值在它下面的 kwargs 里。
def extract_metrics(run):
row = {
"name": run.name,
"type": run.run_type,
"ms": round((run.end_time - run.start_time).total_seconds() * 1000, 2),
"error": bool(run.error),
}
if run.run_type == "llm":
msg = run.outputs["generations"][0][0]["message"]
use = msg.get("kwargs", {}).get("usage_metadata") or {}
row["input_tokens"] = use.get("input_tokens")
row["output_tokens"] = use.get("output_tokens")
row["tool_call_count"] = run.extra.get("tool_call_count")
row["empty"] = not (msg.get("kwargs", {}).get("content") or "").strip()
return row
阈值怎么定,是这一章第二个重点。
不要拍绝对值。用滚动基线。
取近 7 天同小时段的统计值,取中位数,允许上下浮动一个比例。同小时段很重要,因为流量的形态随时段变化,拿凌晨的基线去卡晚高峰必然误报。
def check(key, value, baseline_median, tol=0.3):
lo, hi = baseline_median * (1 - tol), baseline_median * (1 + tol)
if not (lo <= value <= hi):
return f"告警 {key}: 本次 {value:.0f},基线中位数 {baseline_median:.0f}"
return None
用中位数当基线,是因为长尾会话会把平均值拉高。用平均值当基线,你会天天误报。
第三个重点是指标的形状。
我跑了一个 12 轮的会话,两种策略各一遍,看单轮输入 token 怎么走。
| 轮次 | 不治理:单轮输入 token | 治理后:单轮输入 token |
|---|---|---|
| 1 | 24 | 24 |
| 5 | 105 | 54 |
| 8 | 159 | 49 |
| 12 | 249 | 58 |
不治理的那条,单轮输入从 24 涨到 249,第 12 轮是第 1 轮的 10.4 倍。12 轮累计 token 是 1756。
治理后的那条,单轮输入一直在 24 到 62 之间走平。累计 755,是前者的 43%,省了 57%。
注意一件事:会话越长,差距越大。前 5 轮省 24%,到 12 轮就省了 57%。
这就是为什么监控不能只看「本轮平均 token」。单轮看每一轮都没超预算,累计账单却在按会话长度悄悄滑走。
所以我还想加 3 个不那么常规、但更早暴露问题的指标。
会话内单轮输入 token 的斜率。 连续 3 轮持续上升且没有回落,就该去查上下文治理。
单会话累计 token 的 P95。 它比平均值更容易发现那批「聊得特别长」的会话。
消息条数。 它比 token 更早暴露问题,因为条数是线性涨的,token 是加速涨的。
最后一件事:监控发现异常之后干什么。
三步,缺一不可。
回捞样本。 把出问题那批 trace 捞出来,别只看聚合数字。
进评测集。 挑几条有代表性的,标注之后进评测集。
加回归用例。 这样下次改到同一个地方,门禁会替你记住。
走完这三步,一次线上事故就变成了一条永久用例。这是评测体系唯一能持续变强的路径。
八、要上云先算账:计费边界与保留期
前面几章全部在不花钱的前提下完成。如果你想上平台,有三笔账必须先算。
这一章先算前两笔,它们都跟账单的结构有关,跟流量大小无关。
账一:计费边界由你的代码决定。
平台的口径是:一次计费单位是一整条 trace,包含它内部的全部步骤。不按单个 span 分开算。
听起来很慷慨。但关键在于,「trace 从哪里开始」是你的代码决定的。
同一个业务,你按一个会话一条 trace 埋点,还是按一轮对话一条 trace 埋点,账单能差出数倍。而平台的 trace 视图看起来都很正常,两张图长得差不多。
这条建议很朴素:埋点之前先想清楚一条 trace 的边界在哪,再动手。想不清楚就按轮埋,轮比会话可控。
那怎么知道自己实际埋出了几条 trace?
在上平台之前,用本地的 collect_runs() 数一遍就行。根 run 的数量就约等于 trace 的数量。
with collect_runs() as cb:
handle_one_request(payload)
roots = [r for r in cb.traced_runs if r.parent_run_id is None]
print("本次请求产生了", len(roots), "条 trace")
拿真实流量跑一轮,把每个请求产生的 trace 条数分布打出来。再乘上每天的请求量,账单的量级就清楚了。
| 埋点粒度 | 一个 10 轮会话产生 | 账单特征 |
|---|---|---|
| 一个会话一条 trace | 1 条 | 单条大,内部步骤全含 |
| 一轮对话一条 trace | 10 条 | 条数多,单条小 |
| 每个工具调用一条 | 30 条以上 | 条数增长最快 |
三种写法在平台的 trace 视图里都很正常,长得也差不多。它们只体现在账单上。
账二:保留期是两档,而在线评测器会把你升级到贵的那档。
按官方文档口径,trace 有两档保留期。base 档保留期短、单价低,extended 档保留期更长、单价翻倍。
真正需要注意的是一条默认行为:在线评测器和自动化规则的「延长保留」默认是开启的。
也就是说,你在平台上挂一个在线评测器,匹配到的 trace 会自动升到 extended 档计费。你只是在做质量监控,账单结构却变了。
排查动作很简单:上线任何在线评测器之前,先确认保留策略的开关状态。
顺带一个好消息:dataset 的样本不受 trace 保留期影响。这正好支撑前面说的「冻评测集」,你的评测集是安全资产。
九、数据出境前,先把脱敏规则补齐
账三:数据出境得先脱敏,而官方默认规则不认国内云厂商。
这一笔是三个里最实际的。国内团队上云的第一道关,通常不是钱,是数据能不能出去。
create_secret_anonymizer() 是官方给的脱敏工具。我把它对常见密钥形态的识别能力逐个测了一遍。
| 密钥形态 | 默认是否识别 |
|---|---|
lsv2_pt_(平台自身的 key) |
✔ |
sk- / sk-proj-(OpenAI) |
✔ |
sk-ant-(Anthropic) |
✔ |
ghp_(GitHub) |
✔ |
AKIA(AWS) |
✔ |
阿里云 LTAI 开头 |
✘ 不认 |
| RSA 私钥块 | ✘ 不认 |
| 手机号 / 订单号 / 身份证 | ✘ 默认一个都不管 |
官方的默认规则是围着国外的密钥形态写的。国内团队用的时候,业务字段和国内云厂商的凭据都得自己加。
from langsmith.anonymizer import create_anonymizer
anonymizer = create_anonymizer([
{"pattern": r"LTAI[0-9A-Za-z]{12,}", "replace": "[阿里云AK]"},
{"pattern": r"1[3-9]\d{9}", "replace": "[手机号]"},
{"pattern": r"\d{6}(19|20)\d{2}[01]\d[0-3]\d\d{3}[\dXx]", "replace": "[身份证]"},
])
写自定义规则的时候,有 3 个实测行为必须知道。
第一,键名是 replace,不是 replacement。 这一条最阴。写错了不报错,pattern 照样匹配,但你写的掩码文案会被丢掉,输出统一变成 [redacted]。你看到脱敏生效了,不会发现文案是错的。要是下游靠掩码文案统计「各类敏感信息各命中多少」,从这里就开始全错。
第二,自定义脱敏默认是原地改写。 我实测返回的就是同一个对象,res is orig 为真,深层嵌套的字段也一起被改了。脱敏之前先深拷贝,不然你的业务对象会被改掉。
第三,max_depth 设小了会静默漏字段。 它不会提醒你「有东西没处理」,只是安静地跳过深层。默认不限深度,除非你有明确理由,别去动它。
还有一个实操细节:规则之间会互相干扰。我在测试时看到身份证号被手机号规则先命中,110101199003078515 变成了 110101[手机号]5。规则的顺序要按特异性排,窄的规则放前面。
最后说一条替代路线。按公开定价页的口径,自托管与混合部署属于企业档。如果你的数据确实不能出境,这不是一个能用免费额度解决的问题,得按企业档去谈,或者换开源侧的同类方案。
| 三笔账 | 卡在哪 | 动手前先确认 |
|---|---|---|
| 计费边界 | 一条 trace 从哪算起 | 埋点粒度:按会话还是按轮 |
| 保留期 | 两档单价翻倍 | 在线评测器的延长保留开关 |
| 数据出境 | 默认规则不认国内凭据与业务字段 | 自定义脱敏规则 + 深拷贝 |
写自定义规则的时候还有一件事:脱敏要在落盘之前做。 先落盘再脱敏,敏感数据已经写在磁盘上了,清理起来比挡住它麻烦得多。
十、收尾:三道关口,各记一张清单
这一篇没有推荐任何平台,也没给任何金额。我想留下的是一套判断方法。
三道关口,每道一张短清单。
开发期回归
先有评测集,再谈模型。规则能判的别找模型。评测集要冻版本,改样本等于改尺子。
上线前门禁
跑门禁要能阻塞,退出码非 0 才算数。阈值用滚动基线,不拍绝对值。失败要能定位到具体样本,只报一个总分没有用。
上线后监控
指标别超过八个。异常要回捞进评测集,让它变成永久用例。挂在线评测器之前,先看保留策略开关。
回头看这一路:从把 Agent 做出来,到决定拆不拆,到记忆放在哪,再到工具怎么接。这一篇回答的最后一个问题是:怎么证明前面那些决定是对的。
答案其实挺朴素。你有一个能重复跑的标准答案集,有一套每次改动都会跑的门禁,有几个每天都会看几眼的数字。有这三样,你说「它变好了」才有分量。
没有这三样,我们就是那个抽了 20 条、然后被客诉叫醒的团队。
再往下走一步,视角该切到上线那天。前面所有的零件都装好之后,把它推上生产环境,会遇到并发、成本、容错和灰度这四件事。上线第一周会发生什么,我们一件件说。
如果你想把这一篇里的东西直接跑起来,我准备了一份完整的工程包:从 trace 采集到评测集、从门禁脚本到监控指标,还有国内云厂商的脱敏规则,都写好了。在公众号后台回复 eval 就能拿到。
回复关键词
eval领取《Agent 评测与监控工程包》
内含:本地 trace 采集器 / 评测集样例与版本冻结 / 6 个 evaluator / 离线评测执行器 / 门禁脚本(跑不过返回非 0)/ 回放缓存 / 监控指标聚合 / 国内云厂商脱敏规则
干净目录解压即可跑,不需要任何云端账号。