Agent测评-平台实操-把用例变成题库（九）

一、为什么需要题库：一次运行与一批运行是两个数据结构

1.1 轨迹给的是证据，批量要的是题目

前文把一次运行的完整轨迹交给了平台，并且学会了读它：该调的工具调没调、参数对不对、过程绕不绕路、回答有没有依据。四个问题问完，一条轨迹确实能给出答案。但把这套动作重复五十次，问题就换了一个性质。

人读一条轨迹时用的是判断，而不是记录：看到 multiply 节点的输入是 12345678 × 8765432，人脑里做的是"这个参数跟用户说的那个数字对不上"。这个动作有几个特点：无法批量、无法复用、无法审计、每次的判断口径还可能漂移。同一道题今天判对、明天判错，不是人不够认真，而是没有把判断依据固化成数据。

要把它变成可重复的批量动作，需要的不是更好的轨迹，而是题目。一句题面、一个参考值、一个标签——三样东西凑成一道题，五十道题凑成一份题库，题库加上一套判分规则加上一轮执行，才是一份成绩单。这三样东西对应平台上的三个名词，而轨迹只在最外层提供一个可选的来源：这道题是从哪次真实运行里长出来的。

这里有一个必须点破的差别：轨迹是"一次运行的过程记录"，题库是"一批判断的输入数据"。它们是两种数据结构，不是一个东西的两种视图。用轨迹去当题库会立刻撞上三个问题：轨迹没有参考值（当时模型答了什么，不等于正确答案是什么）；轨迹没有稳定的组织单位（一次运行是一条树，不是一道题）；轨迹不可复用（同一个输入下次跑出来的过程一定不同，而题目必须每次都能重跑）。反过来，只有题库没有轨迹也不行——批量分数无法解释时，要能落回某一次真实运行。两边是配对的，但数据结构各是各的。

所以本文要建的东西很具体：一批能反复跑、能改名、能溯源、跑第二遍不会翻倍的题目。

1.2 三个名词先对齐

课程材料用三个生活化的词来记这三个概念，本文沿用它，因为它们是理解后面所有内容的骨架：

第三行被本文刻意搁下了。原因不是它不重要，而是在把题建出来之前就谈成绩单，等于还没出题就先讨论评分标准——分数无法解释时，第一个要问的总是"这道题的参考值到底是不是这个"。

功能块之间的关系可以借那张四块地图定位。这个观测平台的能力分四块，对应评估里四个递进动作：

本文的边界要划得比听起来更窄一点：不写评估器、不出分数、不做版本对比。整篇只回答一个问题——题目怎么从散在代码里的用例，变成平台上有名字、可引用、可反复跑的一等公民。

1.3 一个容易被跳过的顺序问题

有人会问：为什么不直接把 Agent 的输出和参考值写在脚本里，跑个循环打印结果就行？这个问法在五十道题、一个 Agent、一份提示词的时候确实成立，前文里那种最小记录器就是这条路。它会在三个时刻失效，而且这三个失效是叠加的：

换提示词的时候。 手写的用例散在代码里，改一版提示词想看看哪些题变好了，只能靠肉眼看打印输出；有题库时，同一份题库跑两次就是两份成绩单，能并排比。

换人的时候。 别人接手，他不知道哪些题是核心题、哪些题是当时随手加的，也不知道有没有重复题。题库把"这批判断的输入"变成一个有名字的对象，别人可以打开列表逐条看。

换判断口径的时候。 今天用精确匹配判，明天想改成"命中任一写法就算对"，如果题目和参考值混在评分代码里，改口径要重跑数据准备；分开放之后，参考值不动，动的只是拿它比较的方式。

这三条是本文全部工程动作的目的，不是平台功能宣传。后面每一节的实验，最终都在给这三件事兜底：能不能安全地跑第二遍（第六节）、写进去的内容是不是原样（第七节）、题库变大之后成本与行为怎么变（第八节）。

二、这三个概念在代码里到底是什么形状

2.1 先看类型定义，不要先猜

"题库是一道题的容器"这种说法在文档里很好读，但写代码时会立刻撞上四个具体问题：Example 有哪些字段、哪些必填、inputs 和 outputs 的内层键名是谁规定的、题库的"类型"能填什么。这四问靠猜的答案往往是错的，而且错了之后不会报错——它会安静地把脏数据写进平台。

所以下面这段脚本不联网、不调接口，只读 SDK 自己的类型定义，把答案打印出来。这是所有平台类 SDK 都适用的第一动作：先 introspect，再决定怎么写代码。

"""probe_concepts.py —— 三个核心概念在 SDK 里到底是什么形状（不联网，只读类型定义）。

Dataset / Example / Experiment 是平台上的三个名词，但写代码时真正打交道的是
langsmith.schemas 里的模型类。这个脚本不猜：直接把字段名、必填项、枚举值打印出来。
"""
import inspect

import langsmith
from langsmith import Client
from langsmith.schemas import Dataset, Example
from langsmith import schemas as ls_schemas

print("langsmith 版本:", langsmith.__version__)

print("\\n== Dataset 的字段 ==")
print(" 必填:", sorted(name for name, field in Dataset.model_fields.items() if field.is_required()))
print(" 全部:", list(Dataset.model_fields))

print("\\n== Example 的字段 ==")
print(" 必填:", sorted(name for name, field in Example.model_fields.items() if field.is_required()))
print(" 全部:", list(Example.model_fields))

print("\\n== 写入题目时能用的字段（ExampleCreate） ==")
print(" 全部:", list(ls_schemas.ExampleCreate.model_fields))
print(" 必填:", sorted(name for name, field in ls_schemas.ExampleCreate.model_fields.items()
if field.is_required()) or "无")

print("\\n== 题库类型 DataType 的可选值 ==")
print(" ", [f"{item.name}={item.value}" for item in ls_schemas.DataType])

print("\\n== create_examples 的返回值结构 ==")
annotations = getattr(ls_schemas.UpsertExamplesResponse, "__annotations__", {})
print(" 键与类型:", {key: str(value) for key, value in annotations.items()})

print("\\n== 题库侧四个接口的签名 ==")
for method_name in ("list_datasets", "create_dataset", "read_dataset", "list_examples", "create_examples"):
signature = inspect.signature(getattr(Client, method_name))
print(f" {method_name}{signature}")

print("\\n== 课程脚本里的 get_or_create_dataset 是不是 SDK 自带的 ==")
print(" SDK 中存在同名接口:", hasattr(Client, "get_or_create_dataset"))

langsmith 版本: 0.14.11

== Dataset 的字段 ==
必填: ['created_at', 'id', 'name']
全部: ['name', 'description', 'data_type', 'id', 'created_at', 'modified_at', 'example_count', 'session_count', 'last_session_start_time', 'inputs_schema', 'outputs_schema', 'transformations', 'metadata']

== Example 的字段 ==
必填: ['id']
全部: ['dataset_id', 'inputs', 'outputs', 'metadata', 'id', 'created_at', 'modified_at', 'source_run_id', 'attachments']

== 写入题目时能用的字段（ExampleCreate） ==
全部: ['id', 'created_at', 'inputs', 'outputs', 'metadata', 'split', 'attachments', 'use_source_run_io', 'use_source_run_attachments', 'source_run_id']
必填: 无

== 题库类型 DataType 的可选值 ==
['kv=kv', 'llm=llm', 'chat=chat']

== create_examples 的返回值结构 ==
键与类型: {'count': "ForwardRef('int', module='langsmith.schemas')", 'example_ids': "ForwardRef('list[str]', module='langsmith.schemas')", 'as_of': "ForwardRef('NotRequired[str | None]', module='langsmith.schemas')"}

== 题库侧四个接口的签名 ==
list_datasets(self, *, dataset_ids: 'Optional[list[ID_TYPE]]' = None, data_type: 'Optional[str]' = None, dataset_name: 'Optional[str]' = None, dataset_name_contains: 'Optional[str]' = None, metadata: 'Optional[dict[str, Any]]' = None, limit: 'Optional[int]' = None) -> 'Iterator[ls_schemas.Dataset]'
create_dataset(self, dataset_name: 'str', *, description: 'Optional[str]' = None, data_type: 'ls_schemas.DataType' = <DataType.kv: 'kv'>, inputs_schema: 'Optional[dict[str, Any]]' = None, outputs_schema: 'Optional[dict[str, Any]]' = None, transformations: 'Optional[list[ls_schemas.DatasetTransformation]]' = None, metadata: 'Optional[dict]' = None, tag_value_ids: 'Optional[list[ID_TYPE]]' = None) -> 'ls_schemas.Dataset'
read_dataset(self, *, dataset_name: 'Optional[str]' = None, dataset_id: 'Optional[ID_TYPE]' = None) -> 'ls_schemas.Dataset'
list_examples(self, dataset_id: 'Optional[ID_TYPE]' = None, dataset_name: 'Optional[str]' = None, example_ids: 'Optional[Sequence[ID_TYPE]]' = None, as_of: 'Optional[Union[datetime.datetime, str]]' = None, splits: 'Optional[Sequence[str]]' = None, inline_s3_urls: 'bool' = True, *, offset: 'int' = 0, limit: 'Optional[int]' = None, metadata: 'Optional[dict]' = None, filter: 'Optional[str]' = None, include_attachments: 'bool' = False, **kwargs: 'Any') -> 'Iterator[ls_schemas.Example]'
create_examples(self, *, dataset_name: 'Optional[str]' = None, dataset_id: 'Optional[ID_TYPE]' = None, examples: 'Optional[Sequence[ls_schemas.ExampleCreate | dict]]' = None, dangerously_allow_filesystem: 'bool' = False, max_concurrency: 'Annotated[int, Field(ge=1, le=3)]' = 1, **kwargs: 'Any') -> 'ls_schemas.UpsertExamplesResponse | dict[str, Any]'

== 课程脚本里的 get_or_create_dataset 是不是 SDK 自带的 ==
SDK 中存在同名接口: False

2.2 从这份输出里能读出的五件事

第一件：Dataset 必填三项是 name、id、created_at，后两项由服务端给。也就是说客户端只提供名字，其余字段是读回来的结果。这决定了第五节的写法：建完题库之后拿到的那个对象，是当次响应的快照，不是一个会自己更新的句柄。

**第二件：Example 的必填字段只有 id。**这一条要停一下，因为它和直觉相反。直觉上"一道题"应该至少有题面，但模型层面它不强制。这个设计不是给使用者开的绿灯——它存在的理由更可能是服务端另有校验、以及 Example 这个模型同时承担了"读回来的记录"这种不需要必填的场景。无论理由是什么，后果是确定的：客户端类型层面不会因为缺题面而拦住你。第九节用五条畸形题目证明了这件事：全都写成功了。

第三件：inputs、outputs、metadata 的内层键名完全自定义。Dataset 与 Example 的字段列表里没有任何一处规定"题面必须叫 question"。这意味着键名不是协议，而是你和评分代码之间的私人约定。这个约定一旦写进平台就变成隐性契约：改名不会让写入失败，只会让判分取不到值。第十一节专门处理这件事。

**第四件：写入侧字段比读取侧多。**对比两份清单就能看出来：ExampleCreate 有 split、use_source_run_io、use_source_run_attachments，而 Example 没有 split。第三件与第四件的差别在第十一节用实测收口。

**第五件：data_type 只有 kv、llm、chat 三个取值，且默认是 kv。**三个值对应三种题目形态：键值对形态（一道题就是若干字段）、单轮大模型形态、多轮对话形态。默认值是 kv，意味着如果什么都不传，题库就被建成"键值对题库"。而计算器靶子的题面是一句话、参考值是一个答案，形态上更接近单轮问答——这一格该怎么填、填错有什么后果，第九节的错误面实验给了答案：传字符串而不是枚举会当场抛 AttributeError，传错枚举值不会报错，且本文无法验证平台侧对类型的实际约束（假服务端只回显）。

2.3 get_or_create_dataset 不是 SDK 自带的

输出最后一行是 SDK 中存在同名接口: False。这一行看起来像凑数，实际上是被反复踩出来的：课程脚本里用了 get_or_create_dataset，很多读者会以为它是客户端自带的方法，于是去翻文档、找不到、开始怀疑自己拼写错了。它不是——它是脚本里自己写的一个辅助函数，作用就是第六节要讲的幂等。

这类"看起来像 API 其实像业务代码"的命名在评估工程里很常见，因为评估工具的胶水层往往比业务代码更薄：三个动作（查、建、返回）凑成一个函数，名字起得像内置接口。判断办法只有两个，成本都很低：hasattr 一下，或者 inspect.signature 打出来（本节脚本用的就是后者，签名列出来就看得出哪些是 SDK 的方法、哪些是自定义函数）。

2.4 题目和轨迹的关系：两个字段就够

题库和轨迹的关系是本文唯一需要跨前文的地方，而 SDK 里只用两个字段表达：source_run_id 与 use_source_run_io。

• source_run_id：这道题是从哪条轨迹长出来的。填一个轨迹节点 id，题目就多了一条来路。
• use_source_run_io：要不要让平台直接用那条轨迹的输入输出当本题的输入输出，而不是自己手写一份。默认 false。

这两个字段为什么关键：真实业务里的题目常常不是设计出来的，而是从一次真实出问题或者真实做对了的运行里抄出来的。抄的时候如果不留 source_run_id，题目就成了一句没有出处的断言——过三个月没人知道那个参考值是谁定的、依据什么定的；留着它，题目可以落回一条完整轨迹，那里有当时的输入原文、工具参数和模型输出。评估做到后期，"参考值从哪来"会变成一个必须能回答的问题。

2.4 节那个请求体实验会看到，客户端确实把这两个字段原样带上行：

"""probe_request_body.py —— 导入脚本发出去的报文到底长什么样。

只看代码是看不出来的：客户端会自己补字段、改字段名、附带运行环境信息。
把端点指向本地假服务端，请求体就是一份可以逐字打印的 JSON。

时间戳字段 created_at 每次运行都不同，为了让输出可逐字复现，
本文把它替换成占位符 <本次运行时间>，其余字段一律原样。
"""
import json
import sys

sys.path.insert(0, ".")
import dataset_server as srv

PORT = 8403
srv.configure(PORT)
srv.start(PORT)
srv.reset()

from langsmith import Client
from langsmith.schemas import DataType

client = Client()
dataset = client.create_dataset("body_probe", description="看请求体", data_type=DataType.chat)

print("== 建题库：POST /datasets 的请求体 ==")
for method, path, query, body in srv.REQUESTS:
if method == "POST" and path == "/datasets":
print(json.dumps(body, ensure_ascii=False, indent=1))
print("返回的 Dataset 对象字段:", sorted(type(dataset).model_fields))
print("data_type 读回来:", dataset.data_type)

mark = len(srv.REQUESTS)
client.create_examples(
dataset_id=dataset.id,
examples=[
{"inputs": {"question": "12 乘以 34"}, "outputs": {"answer": "408|408"},
"metadata": {"case_type": "small"}, "split": "train"},
{"inputs": {"question": "从一条轨迹长成的题"}, "outputs": {"answer": "1|1"},
"source_run_id": "01a15911-9d3a-7da6-860d-6816337268a0"},
],
)
print("\\n== 写题目：POST /examples/bulk 的请求体（整个 body 是一个数组） ==")
for method, path, query, body in srv.REQUESTS[mark:]:
if method == "POST" and path == "/examples/bulk":
for item in body:
if "created_at" in item:
item["created_at"] = "<本次运行时间>" # 只有这一个字段被替换
print(json.dumps(body, ensure_ascii=False, indent=1))
print("这一步新增的请求（按顺序）:", srv.request_log()[mark:])

mark = len(srv.REQUESTS)
list(client.list_datasets(dataset_name="body_probe"))
client.read_dataset(dataset_name="body_probe")
list(client.list_examples(dataset_id=dataset.id))
print("\\n== 三个读接口实际发出的请求（含查询参数） ==")
for method, path, query, body in srv.REQUESTS[mark:]:
print(f" {method} {path} {json.dumps(query, ensure_ascii=False)}")

mark = len(srv.REQUESTS)
pending = client.list_datasets(dataset_name="body_probe") # 只拿到生成器，一次都没遍历
print("\\n== 只赋值不遍历，请求发出去了吗 ==")
print(" 返回类型:", type(pending).__name__, "| 未消费时新增请求数:", len(srv.REQUESTS) – mark)
first = next(iter(pending))
print(" 消费一条之后才出现请求:", srv.request_log()[mark:], "| 拿到:", first.name)

== 建题库：POST /datasets 的请求体 ==
{
"name": "body_probe",
"data_type": "chat",
"transformations": null,
"extra": {
"metadata": {
"runtime": {
"sdk": "langsmith-py",
"sdk_version": "0.14.11",
"library": "langsmith",
"platform": "macOS-26.5.1-arm64-arm-64bit",
"runtime": "python",
"py_implementation": "CPython",
"runtime_version": "3.11.9",
"langchain_version": "1.4.6",
"langchain_core_version": "1.6.12"
}
},
"source": "sdk"
},
"description": "看请求体"
}
返回的 Dataset 对象字段: ['created_at', 'data_type', 'description', 'example_count', 'id', 'inputs_schema', 'last_session_start_time', 'metadata', 'modified_at', 'name', 'outputs_schema', 'session_count', 'transformations']
data_type 读回来: DataType.chat

== 写题目：POST /examples/bulk 的请求体（整个 body 是一个数组） ==
[
{
"created_at": "<本次运行时间>",
"inputs": {
"question": "12 乘以 34"
},
"outputs": {
"answer": "408|408"
},
"metadata": {
"case_type": "small"
},
"split": "train",
"use_source_run_io": false,
"dataset_id": "6ecf4c80-1346-594f-8827-d73dfb6d4ca8"
},
{
"created_at": "<本次运行时间>",
"inputs": {
"question": "从一条轨迹长成的题"
},
"outputs": {
"answer": "1|1"
},
"use_source_run_io": false,
"source_run_id": "01a15911-9d3a-7da6-860d-6816337268a0",
"dataset_id": "6ecf4c80-1346-594f-8827-d73dfb6d4ca8"
}
]
这一步新增的请求（按顺序）: ['POST /examples/bulk']

== 三个读接口实际发出的请求（含查询参数） ==
GET /datasets {"limit": "100", "name": "body_probe", "offset": "0"}
GET /datasets {"limit": "1", "name": "body_probe"}
GET /examples {"offset": "0", "inline_s3_urls": "True", "limit": "100", "dataset": "6ecf4c80-1346-594f-8827-d73dfb6d4ca8"}

== 只赋值不遍历，请求发出去了吗 ==
返回类型: generator | 未消费时新增请求数: 0
消费一条之后才出现请求: ['GET /datasets'] | 拿到: body_probe

对着这份输出把关系讲清楚，一共四层。

建题库那一层：请求体里 name 和 description 是脚本传的两个参数，其余三个字段全是客户端自己加的——data_type（本文显式传了枚举 DataType.chat，上行成字符串 "chat"）、transformations（没传，但字段照样占位为 null）、extra。extra 里那一大块 metadata.runtime 是环境信息：SDK 名与版本、库名、操作系统平台串、Python 实现与版本、langchain 与 langchain_core 的版本。它是客户端无条件附带的，不需要你写任何代码。

这一层有个合规含义必须点出来：题库侧的上行报文里带着运行环境的机器指纹。platform 那一串在本文这台机器上是 macOS-26.5.1-arm64-arm-64bit，换成公司内网的构建机就是另一串。真实业务里，如果导入脚本跑在某台带敏感主机名的机器上，这个串就把机器特征送了出去。它不像轨迹报文那样含用户输入原文，风险等级低得多，但**"评估工具的元数据上行"这件事本身应该被写进合规清单**——前文只讨论了轨迹里带原文，题库这一侧同样有一条出域通道，只是内容不同。（这一条是从报文事实做的推论，报文内容是实测的，"该进合规清单"属于经验判断。）

写题目那一层：POST /examples/bulk 的整个请求体是一个数组，数组每一项就是 ExampleCreate 转换出来的字典。四个细节值得逐个核对：

读接口那一层：三个读动作的真实 URL 都摊开了，形态值得记下来，因为排查时会用到。

• list_datasets(dataset_name="body_probe") → GET /datasets?limit=100&name=body_probe&offset=0。名字过滤是等值查询，且 limit 被压到 100。
• read_dataset(dataset_name=…) → GET /datasets?limit=1&name=…。所以"读一个题库"在网络上就是"按名字查列表取第一条"，取不到就抛 LangSmithNotFoundError。
• list_examples(dataset_id=…) → GET /examples?limit=100&offset=0&inline_s3_urls=True。每页一百条。

惰性那一层：最后一段是本文唯一一个"不写出来就会有人踩"的行为差别。pending = client.list_datasets(…) 只赋值不遍历，新增请求数是 0；next(iter(pending)) 之后才出现 GET /datasets。返回类型是 generator，它是惰性的。这带来两个后果：一是"查一下有没有同名题库"这个动作，光写 client.list_datasets(…) 是什么都没发生的，必须遍历（用 for、list() 或 next()）才真的发请求；二是异常发生在消费的那一刻而不是调用那一刻，写 try/except 时包围的位置要是遍历，不是函数调用。第六节的幂等代码正是靠 for dataset in client.list_datasets(…) 这个 for 才生效的。

2.5 三个概念与一条轨迹的关系图

把前面散开的东西收进一张图。箭头方向表示"引用"，虚线表示"可选的溯源"：

#mermaid-svg-OsNHL95vP5f215H7{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-OsNHL95vP5f215H7 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-OsNHL95vP5f215H7 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-OsNHL95vP5f215H7 .error-icon{fill:#552222;}#mermaid-svg-OsNHL95vP5f215H7 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-OsNHL95vP5f215H7 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-OsNHL95vP5f215H7 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-OsNHL95vP5f215H7 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-OsNHL95vP5f215H7 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-OsNHL95vP5f215H7 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-OsNHL95vP5f215H7 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-OsNHL95vP5f215H7 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-OsNHL95vP5f215H7 .marker.cross{stroke:#333333;}#mermaid-svg-OsNHL95vP5f215H7 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-OsNHL95vP5f215H7 p{margin:0;}#mermaid-svg-OsNHL95vP5f215H7 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-OsNHL95vP5f215H7 .cluster-label text{fill:#333;}#mermaid-svg-OsNHL95vP5f215H7 .cluster-label span{color:#333;}#mermaid-svg-OsNHL95vP5f215H7 .cluster-label span p{background-color:transparent;}#mermaid-svg-OsNHL95vP5f215H7 .label text,#mermaid-svg-OsNHL95vP5f215H7 span{fill:#333;color:#333;}#mermaid-svg-OsNHL95vP5f215H7 .node rect,#mermaid-svg-OsNHL95vP5f215H7 .node circle,#mermaid-svg-OsNHL95vP5f215H7 .node ellipse,#mermaid-svg-OsNHL95vP5f215H7 .node polygon,#mermaid-svg-OsNHL95vP5f215H7 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-OsNHL95vP5f215H7 .rough-node .label text,#mermaid-svg-OsNHL95vP5f215H7 .node .label text,#mermaid-svg-OsNHL95vP5f215H7 .image-shape .label,#mermaid-svg-OsNHL95vP5f215H7 .icon-shape .label{text-anchor:middle;}#mermaid-svg-OsNHL95vP5f215H7 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-OsNHL95vP5f215H7 .rough-node .label,#mermaid-svg-OsNHL95vP5f215H7 .node .label,#mermaid-svg-OsNHL95vP5f215H7 .image-shape .label,#mermaid-svg-OsNHL95vP5f215H7 .icon-shape .label{text-align:center;}#mermaid-svg-OsNHL95vP5f215H7 .node.clickable{cursor:pointer;}#mermaid-svg-OsNHL95vP5f215H7 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-OsNHL95vP5f215H7 .arrowheadPath{fill:#333333;}#mermaid-svg-OsNHL95vP5f215H7 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-OsNHL95vP5f215H7 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-OsNHL95vP5f215H7 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-OsNHL95vP5f215H7 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-OsNHL95vP5f215H7 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-OsNHL95vP5f215H7 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-OsNHL95vP5f215H7 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-OsNHL95vP5f215H7 .cluster text{fill:#333;}#mermaid-svg-OsNHL95vP5f215H7 .cluster span{color:#333;}#mermaid-svg-OsNHL95vP5f215H7 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-OsNHL95vP5f215H7 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-OsNHL95vP5f215H7 rect.text{fill:none;stroke-width:0;}#mermaid-svg-OsNHL95vP5f215H7 .icon-shape,#mermaid-svg-OsNHL95vP5f215H7 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-OsNHL95vP5f215H7 .icon-shape p,#mermaid-svg-OsNHL95vP5f215H7 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-OsNHL95vP5f215H7 .icon-shape .label rect,#mermaid-svg-OsNHL95vP5f215H7 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-OsNHL95vP5f215H7 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-OsNHL95vP5f215H7 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-OsNHL95vP5f215H7 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

source_run_id 可选溯源

属于某份题库

批量跑一遍并打分

每次结果可落回

一条轨迹Run Tree来源：观测上报

Dataset 题库name + description + data_type

Example 一道题inputs / outputs / metadata

Experiment 成绩单本文不涉及

这张图里唯一被本文实现过、也唯一需要留神标注的边是那条虚线：2.4 的请求体实验证明客户端会把 source_run_id 带进请求体。平台侧到底怎么用这个字段（建不建索引、能不能跳回轨迹、use_source_run_io=true 时是否真的去取那条轨迹的输入输出），本地假服务端不解释语义，因此本文不宣称任何平台侧行为，只声明"字段能写进去、能原样带上行"。

三、题库先长在 CSV 里

3.1 为什么源头是一份纯文本表格

把题库放在 CSV 而不是"直接在平台界面上敲"，理由有四条，每条都对应后面会实测到的性质：

**第一，能 diff。**题库的内容会变：加题、改参考值、删掉一道设计错的题。这些改动必须能被看见。纯文本表格配合版本管理，一次改动的 diff 是逐行的；平台上的题目列表没有"这一版和上一版差哪三行"这种视图。

**第二，非工程同学能改。**出题目的人往往是最懂业务的人，不是最懂 SDK 的人。给一份表格让他们填，比给他们一个 Python 脚本或者平台账号现实得多。

**第三，有唯一的事实源。**如果平台上的题库是唯一的题存放处，那么误删、改错就没有恢复依据。反过来，"CSV 是唯一事实源，平台上的 Dataset 是它的投影"这句话给了一个可执行的纪律：改题只改 CSV，然后重跑导入脚本。这条纪律能不能成立，取决于导入脚本是否幂等——这正是第六节的全部动机。

**第四，它是检查能发生的唯一位置。**题面重复、参考值为空、逗号没加引号这些问题的检测代码，只能对本地文件写。第十节的校验脚本对着 CSV 跑，一个请求都不发就把脏题挡掉；如果题只存在于平台上，同样的检查要先把全部题目拉回来。

代价也在这里说清：CSV 只适合结构化、一行一题的场景。一道题带多轮对话历史、带附件、带图片，表格就会立刻变成负担（history 那种列表要在单元格里塞 JSON 字符串）。本文的计算器靶子恰好是"一句题面一个参考值"，所以这条路顺；真实业务里更常见的是 CSV 和脚本混合——简单的题进表格，复杂的题从轨迹或另一个数据源生成。

3.2 生成脚本：答案由程序算

"""make_cases_csv.py —— 生成题库 CSV：先让程序算出正确答案，再写成带两种写法的 CSV。

为什么不手填答案：两个 8 位数相乘的结果，人工抄写极易错位；而参考答案一旦错，
后面所有分数都建立在错的地基上，且错得很难被发现（Agent 答对了反而被判错）。
所以答案必须由程序算出来，人手只负责题面。

为什么一格里存两种写法：大模型回答大数时经常自动加千分位逗号，
只存纯数字就会把正确答案判成错。用 | 把「纯数字」和「带逗号」两种写法装进同一格，
评分时任一命中就算对。
"""
import csv

CASES = [
("12 乘以 34 等于多少", 12, 34, "small"),
("帮我算一下 12345678 乘以 87654321 等于多少", 12345678, 87654321, "big"),
("987654321 乘以 123456789 是多少", 987654321, 123456789, "big"),
("999999999 乘以 999999999 等于几", 999999999, 999999999, "edge"),
("31415926 乘以 53589793 的结果是什么", 31415926, 53589793, "big"),
]

rows = []
for question, a, b, case_type in CASES:
answer = a * b # 答案由程序计算，不手抄
plain = str(answer) # 写法一：纯数字
grouped = f"{answer:,}" # 写法二：带千分位逗号
rows.append({"question": question, "answer": f"{plain}|{grouped}", "case_type": case_type})
print(f"{case_type:6s} {a} × {b} = {plain} （千分位写法：{grouped}）")

with open("math_agent_eval_cases.csv", "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["question", "answer", "case_type"], quoting=csv.QUOTE_MINIMAL)
writer.writeheader()
writer.writerows(rows)
print("\\n已写入 math_agent_eval_cases.csv，总行数（含表头）:", len(rows) + 1)

small 12 × 34 = 408 （千分位写法：408）
big 12345678 × 87654321 = 1082152022374638 （千分位写法：1,082,152,022,374,638）
big 987654321 × 123456789 = 121932631112635269 （千分位写法：121,932,631,112,635,269）
edge 999999999 × 999999999 = 999999998000000001 （千分位写法：999,999,998,000,000,001）
big 31415926 × 53589793 = 1683572971243318 （千分位写法：1,683,572,971,243,318）

已写入 math_agent_eval_cases.csv，总行数（含表头）: 6

这份脚本里有两个决定是本文全部实验的基础。

**答案由 a * b 算出来，不手抄。**八位数相乘的结果，人在表格里抄一遍出错的概率并不低，而这类错误的后果是整份成绩单不可信：正确答案被判错，错误答案被判对，而且没人会怀疑参考值本身——这是评估工程里最难查的一类 bug，因为系统看起来在正常工作。用程序算，人只负责题面。

**| 分隔两种写法。**第二行题面的答案是 1082152022374638|1,082,152,022,374,638，前一种纯数字、后一种带千分位逗号。这么做的理由很实在：大模型回答大数时经常自动加分隔符。如果参考值只存纯数字，一个语义完全正确的回答会因为多了几个逗号被判错；而把"去掉逗号"的归一化放进评分器又会给错误开一扇门（模型输出 1,082,152,022,374,638 与 108215202237463 只差一个逗号，归一化之后仍然可能蒙对）。把可接受的写法显式列在参考值里，判分依据就是数据而不是规则代码里的假设。

这里有一个反直觉的细节值得注意：small 那一行的两种写法完全相同（408|408）。因为 408 小于 1000，千分位写法就是它自己。这不是 bug，但它暴露了一件事："存两种写法"这个策略只在数值达到四位数时才有意义。第十节的校验脚本因此把这种情况列成"提示"而不是"错误"——不阻断入库，但要让人知道这一行的两种写法没有区分力。

3.3 生成出来的表格长什么样

question,answer,case_type
12 乘以 34 等于多少,408|408,small
帮我算一下 12345678 乘以 87654321 等于多少,"1082152022374638|1,082,152,022,374,638",big
987654321 乘以 123456789 是多少,"121932631112635269|121,932,631,112,635,269",big
999999999 乘以 999999999 等于几,"999999998000000001|999,999,998,000,000,001",edge
31415926 乘以 53589793 的结果是什么,"1683572971243318|1,683,572,971,243,318",big

第五列不用看，前四列值得逐行看：第 3 到第 6 行的 answer 都被英文双引号包住了，第 2 行没有。这个差别就是下面第一个坑的全部原因。

3.4 CSV 的四个坑，以及一个不是坑的坑

用 CSV 存题库，坑不在格式本身多复杂，而在出坑时不报错。这一节把四个坑全部用标准库现场复现，每一个都有打印出来的证据。

"""probe_csv_pitfalls.py —— CSV 的四个坑，全部用标准库 csv 现场复现，不靠记忆。"""
import csv
import io

GOOD = 'question,answer,case_type\\n帮我算 12345678 乘以 87654321,"1082152022374638|1,082,152,022,374,638",big\\n'

print("== 坑一：参考答案含逗号却没加引号 ==")
BAD = 'question,answer,case_type\\n帮我算 12345678 乘以 87654321,1082152022374638|1,082,152,022,374,638,big\\n'
row = next(csv.DictReader(io.StringIO(BAD)))
print("解析出的键:", list(row.keys()))
print("answer 实际拿到:", repr(row["answer"]))
print("被截断的那段去了哪:", repr(row[None]))
print("参考答案是否完整:", row["answer"] == "1082152022374638|1,082,152,022,374,638")

print("\\n== 加了引号的正确写法 ==")
good_row = next(csv.DictReader(io.StringIO(GOOD)))
print("answer 实际拿到:", repr(good_row["answer"]))
print("参考答案是否完整:", good_row["answer"] == "1082152022374638|1,082,152,022,374,638")

print("\\n== 坑二：Excel 导出的 CSV 带 BOM ==")
bom = "\\ufeff" + GOOD
row_bom = next(csv.DictReader(io.StringIO(bom)))
print("用 utf-8 读到的列名:", list(row_bom.keys()))
try:
row_bom["question"]
except KeyError as exc:
print("按 question 取值结果: KeyError", exc)
row_fixed = next(csv.DictReader(io.StringIO(bom.encode("utf-8").decode("utf-8-sig"))))
print("改用 utf-8-sig 后的列名:", list(row_fixed.keys()))

print("\\n== 坑三：读进来全是字符串 ==")
print("answer 的类型:", type(good_row["answer"]).__name__, "| case_type 的类型:", type(good_row["case_type"]).__name__)
print("两种写法拆出来还是字符串:", good_row["answer"].split("|"))

print("\\n== 坑四：空白字符不会被自动清理（这才是真坑） ==")
spaced = 'question,answer,case_type\\n第一题 \\t," 1082152022374638|1,082,152,022,374,638 ",big\\n'
row = next(csv.DictReader(io.StringIO(spaced)))
print("题面 repr:", repr(row["question"]))
print("参考答案 repr:", repr(row["answer"]))
print("与去空格后的参考值是否相等:", row["answer"] == "1082152022374638|1,082,152,022,374,638")

print("\\n== 顺带验证：空行不是坑 ==")
with_trailing = GOOD + "\\n"
rows = list(csv.DictReader(io.StringIO(with_trailing)))
print("带结尾空行时读到的行数:", len(rows))
blank_middle = 'question,answer,case_type\\n第一题,1|1,big\\n\\n第二题,2|2,small\\n'
print("中间夹空行时读到:", len(list(csv.DictReader(io.StringIO(blank_middle)))), "行（空行被跳过）")

print("\\n== 附带事实：csv 写文件的行结束符 ==")
raw = io.StringIO()
csv.DictWriter(raw, fieldnames=["question", "answer", "case_type"], quoting=csv.QUOTE_MINIMAL).writerow(
{"question": "q", "answer": "1|1", "case_type": "small"})
print("写出来的一行 repr:", repr(raw.getvalue()))

== 坑一：参考答案含逗号却没加引号 ==
解析出的键: ['question', 'answer', 'case_type', None]
answer 实际拿到: '1082152022374638|1'
被截断的那段去了哪: ['152', '022', '374', '638', 'big']
参考答案是否完整: False

== 加了引号的正确写法 ==
answer 实际拿到: '1082152022374638|1,082,152,022,374,638'
参考答案是否完整: True

== 坑二：Excel 导出的 CSV 带 BOM ==
用 utf-8 读到的列名: ['\\ufeffquestion', 'answer', 'case_type']
按 question 取值结果: KeyError 'question'
改用 utf-8-sig 后的列名: ['question', 'answer', 'case_type']

== 坑三：读进来全是字符串 ==
answer 的类型: str | case_type 的类型: str
两种写法拆出来还是字符串: ['1082152022374638', '1,082,152,022,374,638']

== 坑四：空白字符不会被自动清理（这才是真坑） ==
题面 repr: '第一题 \\t'
参考答案 repr: ' 1082152022374638|1,082,152,022,374,638 '
与去空格后的参考值是否相等: False

== 顺带验证：空行不是坑 ==
带结尾空行时读到的行数: 1
中间夹空行时读到: 2 行（空行被跳过）

== 附带事实：csv 写文件的行结束符 ==
写出来的一行 repr: 'q,1|1,small\\r\\n'

**坑一：含逗号却不加引号，参考值被截断。**输出第一组是本文最凶的一个坑。没加引号的那一行，answer 实际拿到的是 '1082152022374638|1'——千分位写法里的第一个逗号把单元格切断了。更凶的是被切下去的部分并没有消失，DictReader 把它们塞进了一个键为 None 的余数列：['152', '022', '374', '638', 'big']，连 case_type 的值都被挤掉了。

这件事的凶险在于它不会让任何环节报错。读表格成功、导入成功、平台上有题、评分器按 answer 取值也取得到——只是那个值是残缺的。最后的表现形式是成绩单上一片红，而人第一反应会去查 Agent 的乘法能力，不会去查那半行逗号。

改法两条，按成本从低到高：写的时候永远用 csv 模块写（本文的生成脚本就是这么做的，QUOTE_MINIMAL 会自动把含逗号的字段加引号，不需要人判断），读的时候检查解析出的键集合里有没有 None——键集合多出一个 None，就是这行被切了。

坑二：Excel 导出的 CSV 带 BOM，列名变成 '\\ufeffquestion'。输出第二组：用 utf-8 读，列名第一项前面多了 \\ufeff，row["question"] 直接 KeyError。这个坑反而比第一个温和，因为它当场报错，而且报错信息很怪（明明表头写着 question，程序却说没这个键）。改法是把编码写成 utf-8-sig：它会在开头吞掉 BOM，没有 BOM 时行为与 utf-8 一致。本文所有读 CSV 的地方都用 utf-8-sig，没有例外。

坑三：读进来全是字符串，且平台不会改你的类型。输出第三组显示 answer 和 case_type 的类型都是 str，split("|") 拆出来还是两个字符串。这是 CSV 这个格式的本质：文本文件里没有类型。第七节的回读实验会证明这条链路上没有任何一环会替你转型——写进去是字符串，读回来还是字符串。

对题库来说这条不是缺点，而是要求：如果参考值需要是数字，转换必须显式发生在某个代码里，而且要固定发生在同一处。本文的评分口径是"把参考值当字符串处理、按需拆分"，所以 408|408 保持字符串完全够用。真实业务里如果参考值需要参与数值比较（比如金额、日期），转换要写在读取侧，不要写在表格里期待它自动生效。

**坑四：空白字符不会被自动清理。**输出第四组是那个"反直觉但真实"的结果：'第一题 \\t' 和 ' 1082152022374638|… ' 原样读进来，末尾的空格和制表符都在，与去空格后的值相等判定是 False。

这个坑单独看很轻，配上第六节的幂等就变重了。幂等靠题面做去重键，而 '12 乘以 34 等于多少' 和 '12 乘以 34 等于多少 ' 是两个不同的字符串。于是同一次人工编辑（有人不小心在行尾多敲了个空格）会让幂等判断失效，题库里出现两道看起来一模一样、题面却不完全相等的题。这类"重复但不相同"的题是成绩单最难解释的东西之一：分母里多了一道题，而没有任何地方显示它重复了。

四个坑的共同结构很清楚：它们都不是"读不出来"，而是"读出来的不是你想要的"。这正是第十节要把校验固定在导入之前的原因——这些问题用眼睛看表格是看不出来的（空格、BOM、引号都不是可见字符），只有程序断言能抓。

3.5 空行不是坑，但行结束符是个事实

输出最后两组是刻意加的，因为它们常被误当作坑：

== 顺带验证：空行不是坑 ==
带结尾空行时读到的行数: 1
中间夹空行时读到: 2 行（空行被跳过）

== 附带事实：csv 写文件的行结束符 ==
写出来的一行 repr: 'q,1|1,small\\r\\n'

**空行不是坑。**结尾空行、中间夹空行，DictReader 都跳过，不会读出一条"全空题目"。所以不需要为它写防御代码。这一条值得单独说，是因为太多教程把"跳过空行"当成必要的健壮性处理写进导入脚本，结果是脚本里多了一段永远不会被触发的分支。

**行结束符是 \\r\\n，即使在 macOS 上。**这是 csv 模块的规范行为（RFC 4180），不是 bug。但它有两个实际影响：一是 git diff 会显示整行变化（行尾空格类改动本来就会被 diff 忽略或标成空白变化，配上 \\r\\n 更乱），二是用文本方式手工读 CSV 时，line.split("\\n") 之后每行末尾还挂着一个 \\r。这也是本文所有读 CSV 的代码都老老实实用 csv 模块、并且打开时传 newline="" 的原因——newline="" 是让 csv 自己处理行结束符的标准写法。

3.6 从表格到题库：整条链路

把 CSV、校验、导入、回读串成一张图。图中"本地校验"这一步是第十节加的，本文第一次导入时并没有它：

#mermaid-svg-uaMdHV4ojXfXJqNX{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-uaMdHV4ojXfXJqNX .error-icon{fill:#552222;}#mermaid-svg-uaMdHV4ojXfXJqNX .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-uaMdHV4ojXfXJqNX .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-uaMdHV4ojXfXJqNX .marker{fill:#333333;stroke:#333333;}#mermaid-svg-uaMdHV4ojXfXJqNX .marker.cross{stroke:#333333;}#mermaid-svg-uaMdHV4ojXfXJqNX svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-uaMdHV4ojXfXJqNX p{margin:0;}#mermaid-svg-uaMdHV4ojXfXJqNX .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster-label text{fill:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster-label span{color:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster-label span p{background-color:transparent;}#mermaid-svg-uaMdHV4ojXfXJqNX .label text,#mermaid-svg-uaMdHV4ojXfXJqNX span{fill:#333;color:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX .node rect,#mermaid-svg-uaMdHV4ojXfXJqNX .node circle,#mermaid-svg-uaMdHV4ojXfXJqNX .node ellipse,#mermaid-svg-uaMdHV4ojXfXJqNX .node polygon,#mermaid-svg-uaMdHV4ojXfXJqNX .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-uaMdHV4ojXfXJqNX .rough-node .label text,#mermaid-svg-uaMdHV4ojXfXJqNX .node .label text,#mermaid-svg-uaMdHV4ojXfXJqNX .image-shape .label,#mermaid-svg-uaMdHV4ojXfXJqNX .icon-shape .label{text-anchor:middle;}#mermaid-svg-uaMdHV4ojXfXJqNX .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-uaMdHV4ojXfXJqNX .rough-node .label,#mermaid-svg-uaMdHV4ojXfXJqNX .node .label,#mermaid-svg-uaMdHV4ojXfXJqNX .image-shape .label,#mermaid-svg-uaMdHV4ojXfXJqNX .icon-shape .label{text-align:center;}#mermaid-svg-uaMdHV4ojXfXJqNX .node.clickable{cursor:pointer;}#mermaid-svg-uaMdHV4ojXfXJqNX .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-uaMdHV4ojXfXJqNX .arrowheadPath{fill:#333333;}#mermaid-svg-uaMdHV4ojXfXJqNX .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-uaMdHV4ojXfXJqNX .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-uaMdHV4ojXfXJqNX .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-uaMdHV4ojXfXJqNX .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-uaMdHV4ojXfXJqNX .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-uaMdHV4ojXfXJqNX .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster text{fill:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX .cluster span{color:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-uaMdHV4ojXfXJqNX .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-uaMdHV4ojXfXJqNX rect.text{fill:none;stroke-width:0;}#mermaid-svg-uaMdHV4ojXfXJqNX .icon-shape,#mermaid-svg-uaMdHV4ojXfXJqNX .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-uaMdHV4ojXfXJqNX .icon-shape p,#mermaid-svg-uaMdHV4ojXfXJqNX .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-uaMdHV4ojXfXJqNX .icon-shape .label rect,#mermaid-svg-uaMdHV4ojXfXJqNX .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-uaMdHV4ojXfXJqNX .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-uaMdHV4ojXfXJqNX .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-uaMdHV4ojXfXJqNX :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

不通过

通过

没有

有

人工/程序出题只改 CSV

生成或编辑csv 模块写出

本地校验空表/重复/空白/空参考值

中止一个请求都不发

读 CSV 转成inputs/outputs/metadata

平台上有没有同名题库

create_dataset

复用现有题库

按题面去重只补差集

create_examples

list_examples 回读核对条数与内容

批量评估引用题库名

这张图有两个读法要强调。第一，链路里有两个"闸"，都在网络请求之前：校验闸和判空闸。有闸的好处不是安全，而是反馈速度——本地失败是改一行重跑，写进平台之后要先查有哪些脏题、再决定删还是重建，代价完全不同。第二，"复用现有题库"之后仍然要走去重，题库存在不代表题在里面，这是第六节最核心的一个论断。

四、环境与认证：Client() 是怎么找到配置的

4.1 项目布局

本文所有代码放在同一个文件夹里，文件名与脚本里的常量一致。前文接入观测时的目录继续用，本文只往里面加东西。下面只列课程脚本，本文用来测行为的探针脚本（probe_*.py）不在这个清单里，它们各自在后台线程里起一个假服务端、用完即走：

.env 里本文用到的是四行配置。前三行是接入时就有的，第四行是本文实验才需要改的：

# .env：前三行是接入阶段就有的，第四行把端点指向本机的假服务端
LANGSMITH_TRACING=false
LANGSMITH_API_KEY=lsv2_localdev0000000000000000000000000000
LANGSMITH_PROJECT=calculator-agent
LANGSMITH_ENDPOINT=http://127.0.0.1:8401

第二、四行要解释一下。本文所有写操作都不指向真实平台：LANGSMITH_ENDPOINT 指到本机的假服务端，LANGSMITH_API_KEY 是一眼假的占位串（lsv2_localdev 加一串零）。这么做的目的不是省一个账号，而是让"客户端到底发了什么"这件事可观测、可复现、且不会把任何数据送出本机。换成真实平台时，只有端点和 Key 两行要改，脚本一行不用动——这也是"配置从代码里拿出来"这个原则的一次具体兑现。

另外补一句开关：这里把 LANGSMITH_TRACING 设成 false，是为了让本文的实验只走题库接口、不产生轨迹上报。它不影响题库侧的任何行为：2.4 与第五节的全部写操作都是在这个开关关闭的状态下成功的。

4.2 load_dotenv() 到底从哪里找 .env

create_dataset.py 开头那行 load_dotenv() 没有任何参数，这是最常见的写法，也是踩坑率最高的一行。它的行为经常被误解成"读当前目录下的 .env"，而当前目录这个说法在真实工作流里几乎总会变：从别的目录执行脚本、在 IDE 里点运行、放进 cron 或 CI，工作目录都不是脚本所在目录。

误解释的另一半后果更麻烦：它读不到文件时不报错。前文讲观测接入时专门列过一份静默失败清单，那一侧的失败表现为"没有数据"；题库这一侧的失败表现为"配置没生效"，同样安静。所以这段实验把规则一次性测清楚：脚本放在哪、.env 放在哪、工作目录在哪、返回值是什么、客户端最终看到的端点是什么。

"""probe_env_rule.py —— 把 load_dotenv() 的查找规则变成可打印的事实。

四条规则各配一个证据：
1. 搜索起点是脚本自己所在目录，还是终端的工作目录？
2. 脚本在 .env 的下层子目录里，还能不能往上命中？
3. 找不到时报错，还是静默返回 False？
4. 命中之后会不会覆盖进程里已经存在的环境变量？

实现方式：在系统临时目录下摆一棵独立的演示目录树（跑完删掉），用子进程在不同 cwd 下
执行同一个子脚本。放在临时目录是为了保证这条链的任何祖先目录里都没有 .env 干扰结论。子脚本只打印「返回值 + 客户端最终看到的端点」，路径不外泄，输出可逐字复现。
"""
import os
import shutil
import subprocess
import sys
from pathlib import Path

CHILD = '''
import os
from dotenv import load_dotenv

explicit = os.environ.get("DEMO_DOTENV") # 只有「显式路径」那一例会被设上
found = load_dotenv(explicit) if explicit else load_dotenv()
print(" load_dotenv() 返回:", found, "| 端点:", os.environ.get("LANGSMITH_ENDPOINT", "<未设置>"))
'''

def run_child(script_path, cwd, label, preset=None, dotenv_path=None):
"""在指定 cwd 下执行子脚本，子进程环境里先清空所有 LANGSMITH_* 变量。"""
env = {k: v for k, v in os.environ.items() if not k.startswith("LANGSMITH_")}
if preset:
env.update(preset)
if dotenv_path:
env["DEMO_DOTENV"] = dotenv_path
result = subprocess.run([sys.executable, str(script_path)], cwd=str(cwd),
env=env, capture_output=True, text=True)
print(label)
print(result.stdout.rstrip() or " （子进程无输出）")
if result.stderr.strip():
print(" stderr:", result.stderr.strip().splitlines()[–1])

def main():
import tempfile
root = Path(tempfile.gettempdir()) / "agent-d9-envrule-demo" # 绝对路径，祖先目录里没有 .env
shutil.rmtree(root, ignore_errors=True) # 清掉上次残留，保证每次从零搭建
proj, sub, elsewhere = root / "langsmith-calculator", root / "langsmith-calculator" / "scripts", root / "elsewhere"
for directory in (proj, sub, elsewhere):
directory.mkdir(parents=True)
(proj / ".env").write_text("LANGSMITH_ENDPOINT=http://127.0.0.1:8401\\n")
children = {name: directory / "child.py" for name, directory in
(("proj", proj), ("sub", sub), ("elsewhere", elsewhere))}
for path in children.values():
path.write_text(CHILD)

print("== 规则一：起点是脚本所在目录，与终端 cwd 无关 ==")
run_child(children["proj"], elsewhere, " 脚本与 .env 同目录，cwd 在别处")

print("\\n== 规则二：命中不了会继续往上层找 ==")
run_child(children["sub"], elsewhere, " 脚本在 .env 的下层子目录里，cwd 在别处")

print("\\n== 规则三：整条链上都没有 .env 时，返回 False 且不报错 ==")
run_child(children["elsewhere"], proj, " 脚本挪到别处，cwd 反而是 .env 所在目录")

print("\\n== 规则四：显式传路径时，脚本放哪都能读到 ==")
run_child(children["elsewhere"], elsewhere, " 脚本在别处 + load_dotenv(显式指向 proj/.env)",
dotenv_path=str(proj / ".env"))

print("\\n== 规则五：进程里已有的值优先，.env 默认不覆盖 ==")
run_child(children["proj"], elsewhere, " 预先设端点 9999，再读同一个 .env（里面写的是 8401）",
preset={"LANGSMITH_ENDPOINT": "http://127.0.0.1:9999"})

print("\\n== 顺带核对：两个函数各自接受的参数 ==")
import inspect
from dotenv import find_dotenv, load_dotenv
print(" load_dotenv 参数:", str(inspect.signature(load_dotenv)))
print(" find_dotenv 参数:", str(inspect.signature(find_dotenv)))

shutil.rmtree(root) # 收尾：删掉这棵演示目录树

if __name__ == "__main__":
main()

== 规则一：起点是脚本所在目录，与终端 cwd 无关 ==
脚本与 .env 同目录，cwd 在别处
load_dotenv() 返回: True | 端点: http://127.0.0.1:8401

== 规则二：命中不了会继续往上层找 ==
脚本在 .env 的下层子目录里，cwd 在别处
load_dotenv() 返回: True | 端点: http://127.0.0.1:8401

== 规则三：整条链上都没有 .env 时，返回 False 且不报错 ==
脚本挪到别处，cwd 反而是 .env 所在目录
load_dotenv() 返回: False | 端点: <未设置>

== 规则四：显式传路径时，脚本放哪都能读到 ==
脚本在别处 + load_dotenv(显式指向 proj/.env)
load_dotenv() 返回: True | 端点: http://127.0.0.1:8401

== 规则五：进程里已有的值优先，.env 默认不覆盖 ==
预先设端点 9999，再读同一个 .env（里面写的是 8401）
load_dotenv() 返回: True | 端点: http://127.0.0.1:9999

== 顺带核对：两个函数各自接受的参数 ==
load_dotenv 参数: (dotenv_path: Union[str, ForwardRef('os.PathLike[str]'), NoneType] = None, stream: Optional[IO[str]] = None, verbose: bool = False, override: bool = False, interpolate: bool = True, encoding: Optional[str] = 'utf-8') -> bool
find_dotenv 参数: (filename: str = '.env', raise_error_if_not_found: bool = False, usecwd: bool = False) -> str

五条规则的准确表述，逐条对着输出读：

**规则一：搜索起点是调用 load_dotenv() 的那个脚本文件所在目录，与终端在哪个目录无关。**证据是第一组：脚本与 .env 同目录，而工作目录在别处，返回 True、端点生效。

**规则二：从脚本所在目录开始逐层往上找。**证据是第二组：脚本被放进 .env 的下层子目录（scripts/），仍然命中上层那份。这个"往上找"的行为很多人不知道，它让项目子目录里的脚本能自然继承根目录的配置。

规则三：整条链上都没有 .env 时，返回 False，不抛异常。证据是第三组：脚本被搬到另一个目录树里（哪怕工作目录恰好就是 .env 所在目录），返回值是 False，端点没被设置。这里也顺带说明了为什么"看返回值"比"等报错"更重要：这一行代码是有返回值的，if not load_dotenv(): … 是完全写得出来的，只是很少有人写。

**规则四：显式传路径时，脚本在哪都读得到。**证据是第四组：脚本在别处（规则三里返回 False 的那个位置），把路径显式指过去就成功。这条给的是一条兜底写法：当项目结构复杂、脚本会被复制到各处时，load_dotenv(Path(__file__).with_name(".env")) 比裸调用可预期。

**规则五：进程里已经存在的环境变量优先，.env 默认不覆盖。**证据是第五组：先把端点设成 9999 再读同一份 .env（里面写的是 8401），返回值是 True（文件找到了），但客户端最终看到的端点仍是 9999。这来自 load_dotenv 的默认参数 override=False——输出最后一组打印了函数签名，override: bool = False 就在里面。

第五条在实际工作里最容易咬人：CI 平台、容器编排、shell 启动脚本都可能已经设了同名变量。此时你在 .env 里改了端点或项目名，改的是一份不会被采用的文件。排查这类问题不能靠读 .env，只能打印进程真正读到的值——也就是本文每个实验都在做的那件事：不看配置文件的字面值，看客户端最终拿到的值。

最后一组签名还顺手澄清一个常见错误：load_dotenv() 没有 usecwd 参数，带这个参数的是 find_dotenv()。把两者混起来写出的代码会以 TypeError 收场，而网上这类写法不少。

一句经验判断收尾：本文推荐的做法是脚本与 .env 同目录（课程项目的布局本来就是这样），这不需要任何额外代码就能命中规则一；一旦脚本要被复用到别的目录树，就改成显式路径（规则四），别指望工作目录。

4.3 让全部实验不依赖账号：一个本地假服务端

要复现本文所有结论，不需要真实账号，但需要一个能收请求的地方。这个服务端只做三件事：按客户端真正会打的接口回包、把请求记在一个列表里、把数据存在两个进程内字典中。它不校验 Key、不落盘、不解释任何业务语义——这三点同时也是它的局限，本文凡是涉及"平台侧会怎么处理"的结论，都会显式标注无法验证。

"""dataset_server.py —— 一个只有本机回环地址的假题库服务端。

为什么要有它：本文所有结论都要求「可复现、不依赖账号」。真实平台的请求体、
错误码、分批行为，光看文档只能靠猜；把端点指到 127.0.0.1 上自己的进程，
客户端到底发了什么就变成可以打印、可以逐字核对的文本。

它按客户端真正会打的接口回包，因此 create_dataset / create_examples /
list_datasets / list_examples / read_dataset 能真的走完一轮「写进去再读回来」：

– GET /info 能力探测，返回空对象
– GET /sessions 租户探测，返回空数组
– POST /datasets 建题库；同名回 409
– GET /datasets 按名字查题库；带 name 过滤
– GET /datasets/{id} 读单个题库
– POST /examples/bulk 批量写入题目
– GET /examples 分页读题目

两个刻意的设计：
1. id 用 uuid5 从名字派生，不用随机 uuid4——这样「题库 id」在每次运行时都相同，
脚本输出可以逐字比对；真实平台给的是随机 uuid，这一点差别不影响本文任何结论。
2. 所有请求都记在 REQUESTS 里，probe 脚本靠它把「这一次调用到底发了几个请求」打印出来。
"""
import json
import os
import re
import threading
import uuid
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

DATASETS = {} # dataset_id -> 题库字典
EXAMPLES = [] # 全部题目，每条带 dataset_id
REQUESTS = [] # (方法, 路径, 查询参数, 请求体)

DUPLICATE_POLICY = "reject" # 同名题库直接回 409；改成 "allow" 可以演示重复建库

def deterministic_id(kind, name):
"""从名字派生固定 id，保证输出可逐字复现。"""
return str(uuid.uuid5(uuid.NAMESPACE_DNS, f"{kind}:{name}"))

def now():
"""UTC ISO 时间戳，字段格式与真实接口一致。"""
return datetime.now(timezone.utc).isoformat()

TENANT_ID = deterministic_id("tenant", "local-dev")

class Handler(BaseHTTPRequestHandler):
"""把 REST 接口映射到进程内的两个字典上。"""

def _json(self, code, payload):
raw = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(raw)))
self.end_headers()
self.wfile.write(raw)

def _body(self):
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b""
try:
return json.loads(raw.decode("utf-8")) if raw else None
except Exception:
return {"__raw__": raw.decode("utf-8", "replace")[:200]}

def do_GET(self):
parsed = urlparse(self.path)
query = {k: v[0] for k, v in parse_qs(parsed.query).items()}
REQUESTS.append(("GET", parsed.path, query, None))
if parsed.path == "/info":
return self._json(200, {})
if parsed.path == "/sessions":
# 客户端用它探测租户 id（_get_optional_tenant_id）。回一个可用的会话对象，
# 租户 id 就会被缓存，后续请求不再重复探测——请求计数才解释得清楚。
return self._json(200, [{"name": "local-dev", "id": deterministic_id("session", "local-dev"),
"start_time": now(), "tenant_id": TENANT_ID,
"reference_dataset_id": None}])
if parsed.path == "/datasets":
name = query.get("name")
items = [d for d in DATASETS.values() if not name or d["name"] == name]
return self._json(200, items)
match = re.match(r"^/datasets/([\\w-]+)$", parsed.path)
if match:
dataset = DATASETS.get(match.group(1))
return self._json(200, dataset) if dataset else self._json(404, {"error": "not found"})
if parsed.path == "/examples":
dataset_id = query.get("dataset")
offset, limit = int(query.get("offset") or 0), int(query.get("limit") or 100)
items = [e for e in EXAMPLES if e.get("dataset_id") == dataset_id]
return self._json(200, items[offset:offset + limit])
return self._json(200, [])

def do_POST(self):
parsed = urlparse(self.path)
body = self._body()
REQUESTS.append(("POST", parsed.path, {}, body))
if parsed.path == "/datasets":
name = (body or {}).get("name")
if DUPLICATE_POLICY == "reject" and any(d["name"] == name for d in DATASETS.values()):
return self._json(409, {"error": {"message": f"Dataset {name} already exists"}})
dataset = {
"id": deterministic_id("dataset", name),
"name": name,
"data_type": body.get("data_type", "kv"),
"description": body.get("description"),
"created_at": now(),
"modified_at": now(),
"example_count": 0,
"extra": body.get("extra", {}),
}
for key in ("inputs_schema_definition", "outputs_schema_definition"):
if key in body:
dataset[key] = body[key]
DATASETS[dataset["id"]] = dataset
return self._json(201, dataset)
if parsed.path == "/examples/bulk":
created = []
for item in body or []:
example = {
# 用写入顺序派生 id：同一条题目重复写两次会得到两个不同的 id，
# 与真实平台一致；同时每次重跑本服务端得到的 id 序列固定，输出可逐字比对。
"id": item.get("id") or deterministic_id("example", str(len(EXAMPLES) + 1)),
"dataset_id": item.get("dataset_id"),
"created_at": now(),
"modified_at": now(),
"inputs": item.get("inputs"),
"outputs": item.get("outputs"),
"metadata": item.get("metadata"),
}
if "split" in item:
example["split"] = item["split"]
if "source_run_id" in item:
example["source_run_id"] = item["source_run_id"]
EXAMPLES.append(example)
dataset = DATASETS.get(example["dataset_id"])
if dataset:
dataset["example_count"] = sum(
1 for e in EXAMPLES if e.get("dataset_id") == dataset["id"])
created.append({"id": example["id"]})
return self._json(200, created)
return self._json(200, {})

do_PATCH = do_PUT = do_DELETE = do_POST

def log_message(self, *args):
"""关掉默认访问日志，避免污染实验输出。"""

def start(port):
"""在 127.0.0.1:port 起后台线程跑服务端，主线程继续做实验。"""
server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
return server

def serve_blocking(port):
"""前台模式：python dataset_server.py 8401，给 create_dataset.py 这类脚本当后端。"""
print(f"本地题库服务端已启动：http://127.0.0.1:{port}")
ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()

def configure(port):
"""把客户端指向本机；Key 用一眼假的，不会误打到真实平台。"""
os.environ.update({
"LANGSMITH_TRACING": "false",
"LANGSMITH_API_KEY": "lsv2_localdev0000000000000000000000000000",
"LANGSMITH_PROJECT": "calculator-agent",
"LANGSMITH_ENDPOINT": f"http://127.0.0.1:{port}",
})

def reset():
"""清空状态与请求记录，让每个实验从同一起点开始。"""
DATASETS.clear()
EXAMPLES.clear()
REQUESTS.clear()

def request_log():
"""按发生顺序把请求列出来，路径里的 36 位 id 统一写成 <id>。"""
return [f"{method} {re.sub(r'/[0-9a-f-]{36}$', '/<id>', path)}" for method, path, _q, _b in REQUESTS]

def count(method, path):
"""数某一类请求发了几次。"""
return len([1 for m, p, _q, _b in REQUESTS if m == method and p == path])

def bulk_posts():
"""写入请求发了几次：只数 POST /examples/bulk。"""
return len([1 for method, path, _q, _b in REQUESTS if method == "POST" and path == "/examples/bulk"])

def examples_gets():
"""读题目发了几次 GET /examples，以及每次带的分页参数。"""
calls = [(q.get("limit"), q.get("offset")) for method, path, q, _b in REQUESTS
if method == "GET" and path == "/examples"]
return len(calls), calls

if __name__ == "__main__":
import sys
serve_blocking(int(sys.argv[1]) if len(sys.argv) > 1 else 8401)

读这份代码有三处是刻意为之，值得说明：

**deterministic_id 用 uuid5 从名字派生 id，而不是随机 uuid4。**真实平台发的 id 是随机的，本文换成派生 id 只为了一个目的：让脚本输出在每次重跑时逐字相同，读者能把自己跑出来的结果和文中的输出做严格比对。id 的形状（36 位带连字符的 UUID）与真实值一致，所有涉及 id 的接口行为都不受影响。题目 id 用"写入顺序"派生（str(len(EXAMPLES) + 1)），这样同一道题重复导入会得到两个不同 id——这一条是对真实行为的刻意模仿，第六节会看到它有多重要。

**/sessions 必须回一个可用的会话对象。**这是实现中最花时间的一处。客户端有一个租户探测逻辑（_get_optional_tenant_id）：它发 GET /sessions?limit=1，把返回的第一条解析成会话对象，取里面的租户 id 并缓存。本文最初的回包只给了 name、id、start_time、tenant_id，缺 reference_dataset_id 这个必填字段，于是解析失败、缓存不生效，客户端每构造一个对象就重新探测一次，请求日志里冒出一堆与题库毫无关系的 GET /sessions。这些噪声不影响功能，但它让"这次动作到底发了几个请求"这个问题没法解释。补上那个字段之后，探测只发生一次，请求序列才干净。

这个教训本身值得记：用假服务端做实验时，"能回 200"和"回包能被客户端模型解析"是两件事。后者不满足时，症状常常不是报错，而是重试、回退路径、或者像这里一样悄悄改变请求次数。

serve_blocking 和 start 是两种用法。start 在后台线程里跑，供探针脚本自己起自己用（同一进程，所以能直接读 REQUESTS）；serve_blocking 给独立进程用，python dataset_server.py 8401 起一个常驻服务，让 create_dataset.py 这类"按课程原样写、只读 .env"的脚本能不改一行就跑起来。第五、第七节用的是后者——那才是真实工作流的样子。

4.4 认证错误当场抛：数据面和观测面正好相反

这一节只测一件事：配置错了，错误什么时候露出来。前文的轨迹上报侧是出了名的安静——Key 不对、端点写错，业务照常跑完，异常一个都不抛，只在标准错误里留一行日志。题库这一侧完全相反：所有接口都是同步返回的，错了就抛。这个差别会直接改变两件事：脚本要不要写重试、以及批量跑失败时是"停下来"还是"跑完一份空成绩单"。

"""probe_auth.py —— 题库侧（数据面）的认证错误怎么暴露。

这一点必须单独测：轨迹上报的失败是静默的（业务照跑，只在日志里留一行），
而题库侧的接口是同步返回的。两种暴露方式不同，排查手段也就完全不同。
"""
import os
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

sys.path.insert(0, ".")

HITS = [] # 服务端收到的 (方法, 路径, 请求头里的 key)

class Always401(BaseHTTPRequestHandler):
"""不管来什么都回 401 的服务端，用来模拟「Key 不对或没权限」。"""

def _handle(self):
length = int(self.headers.get("Content-Length") or 0)
if length:
self.rfile.read(length)
HITS.append((self.command, self.path, self.headers.get("x-api-key")))
self.send_response(401)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(b'{"error":{"type":"auth","message":"API key not found"}}')

do_POST = do_PATCH = do_PUT = do_GET = _handle

def log_message(self, *args):
pass

PORT = 8408
server = ThreadingHTTPServer(("127.0.0.1", PORT), Always401)
threading.Thread(target=server.serve_forever, daemon=True).start()

os.environ.update({"LANGSMITH_TRACING": "false", "LANGSMITH_ENDPOINT": f"http://127.0.0.1:{PORT}"})
from langsmith import Client

print("== 情形一：环境变量里完全没有 Key ==")
os.environ.pop("LANGSMITH_API_KEY", None)
client = Client()
print(" Client() 构造结果: 成功，没有任何提示")
try:
client.create_dataset("auth_probe")
except Exception as exc:
print(" 真正调用时才报错:", type(exc).__name__, "|", str(exc).split(". HTTPError")[0])

print("\\n== 情形二：给一把错的 Key ==")
os.environ["LANGSMITH_API_KEY"] = "lsv2_wrongkey0000000000000000000000000000"
client = Client()
for label, call in [("list_datasets 遍历", lambda: list(client.list_datasets())),
("create_dataset", lambda: client.create_dataset("auth_probe2")),
("read_dataset", lambda: client.read_dataset(dataset_name="x")),
("create_examples", lambda: client.create_examples(
dataset_id="00000000-0000-0000-0000-000000000000", examples=[{"inputs": {"q": "1"}}]))]:
try:
call()
print(f" {label}: 没有报错")
except Exception as exc:
print(f" {label}: {type(exc).__name__} | {str(exc).split('. HTTPError')[0]}")

print("\\n== 服务端看到的请求头里的 Key ==")
print(" （并发下到达顺序不稳定，这里按方法加路径排序后再打印）")
for method, path, key in sorted(HITS, key=lambda row: (row[0], row[1])):
print(f" {method} {path} -> x-api-key: {(key or '<没有这个头>')[:12]}")

print("\\n== 情形三：端点根本不通时，客户端自己打出来的错误长什么样 ==")
os.environ["LANGSMITH_ENDPOINT"] = "http://127.0.0.1:8499"
broken = Client()
try:
broken.read_dataset(dataset_name="any")
except Exception as exc:
print(" 异常类型:", type(exc).__name__)
for line in str(exc).splitlines():
print(" 消息:", line)

== 情形一：环境变量里完全没有 Key ==
Client() 构造结果: 成功，没有任何提示
真正调用时才报错: LangSmithAuthError | Authentication failed for /datasets

== 情形二：给一把错的 Key ==
list_datasets 遍历: LangSmithAuthError | Authentication failed for /datasets
create_dataset: LangSmithAuthError | Authentication failed for /datasets
read_dataset: LangSmithAuthError | Authentication failed for /datasets
create_examples: LangSmithAuthError | Authentication failed for /examples/bulk

== 服务端看到的请求头里的 Key ==
（并发下到达顺序不稳定，这里按方法加路径排序后再打印）
GET /datasets?limit=1&name=x -> x-api-key: lsv2_wrongke
GET /datasets?limit=100&offset=0 -> x-api-key: lsv2_wrongke
GET /info -> x-api-key: <没有这个头>
GET /info -> x-api-key: lsv2_wrongke
POST /datasets -> x-api-key: <没有这个头>
POST /datasets -> x-api-key: lsv2_wrongke
POST /examples/bulk -> x-api-key: lsv2_wrongke

== 情形三：端点根本不通时，客户端自己打出来的错误长什么样 ==
异常类型: LangSmithConnectionError
消息: Connection error caused failure to GET /datasets in LangSmith API. Please confirm your LANGCHAIN_ENDPOINT. ConnectionError(MaxRetryError('HTTPConnectionPool(host=\\'127.0.0.1\\', port=8499): Max retries exceeded with url: /datasets?limit=1&name=any (Caused by NewConnectionError("HTTPConnection(host=\\'127.0.0.1\\', port=8499): Failed to establish a new connection: [Errno 61] Connection refused"))'))
消息: Content-Length: None
消息: API Key: lsv2_**********************************00

三组输出的读法：

没有 Key 时，Client() 构造成功。这是最容易误判的一步：客户端不会在构造阶段检查凭据是否存在，它连构造都通过。错误发生在第一次真正的调用上，create_dataset 抛 LangSmithAuthError，消息明确指出是哪个接口（Authentication failed for /datasets）。也就是说，"能不能构造对象"和"能不能用"之间没有任何检查点。

Key 错时，四个读写字接口全部当场抛。这份输出里 list_datasets 的遍历、create_dataset、read_dataset、create_examples 都抛 LangSmithAuthError，而且明确到接口路径。特别提一下 list_datasets：它是惰性生成器（2.4 规则），异常出现在遍历的那一刻，所以 try 要包围 list(…) 或 for 循环本身，只包围函数调用是抓不到的。

请求头 x-api-key 在没有 Key 时根本不存在（不是空串，是没有这个头）。有 Key 时是原样带上。这是排查"权限到底有没有传到服务端"时唯一能依赖的事实来源——本文的服务端把收到的头打了出来。

**连接错误消息里，Key 是被脱敏的。**第三组把端点指向一个没人监听的端口，异常消息除了常规的 Connection refused，还多打了两行：Content-Length: None 和 API Key: lsv2_**********************************00。客户端自己做了脱敏：保留前缀和末尾两个字符，中间全压成星号。这条实测的意义在别处——日志与错误消息里出现 Key 是设计的一部分，但它已经处理过了。做合规评审时可以拿这一条当依据，但前提仍然是别把整段异常原文贴进公开工单：那条消息里同时有主机、端口、完整 URL 和 Key 的末两位，对内网拓扑是一种暴露。

最后一组还澄清了一个易混点：Please confirm your LANGCHAIN_ENDPOINT 这句提示写的是旧前缀 LANGCHAIN_。前文已经实测过新旧两套前缀都有效，这条消息只是措辞里保留了旧名，不代表变量必须叫这个名字。

五、导入脚本逐行：一行 CSV 怎么变成一道题

5.1 完整脚本

这是把题送上台面的那一个文件。它只做四件事：读表、找或建题库、算出差集、写入。

"""create_dataset.py —— 把本地 CSV 题库导入成平台上的 Dataset。

分工说清楚：CSV 是人改题的地方（能 diff、能进版本库），
Dataset 是平台上的引用名（批量跑分按名字取题），两边靠这个脚本对齐。
"""
import csv
import os

from dotenv import load_dotenv
from langsmith import Client

CSV_PATH = "math_agent_eval_cases.csv"
DATASET_NAME = "math_agent_eval_cases"
DESCRIPTION = "计算器 Agent 题库：输入一句乘法，参考值是两个大数乘积"

def load_cases(path):
"""读 CSV，返回题目字典列表。

用 utf-8-sig 而不是 utf-8：从 Excel 导出的 CSV 常带 BOM 头，
用 utf-8 读会让第一列的列名变成 '\\\\ufeffquestion'，后面取值直接 KeyError。
"""
with open(path, encoding="utf-8-sig", newline="") as f:
return list(csv.DictReader(f))

def to_example(row):
"""把一行 CSV 转成平台要求的结构：inputs 是题面，outputs 是参考值，metadata 是标签。"""
return {
"inputs": {"question": row["question"]},
"outputs": {"answer": row["answer"]},
"metadata": {"case_type": row["case_type"], "source": "csv"},
}

def get_or_create_dataset(client, dataset_name, description):
"""同名 Dataset 已存在就复用，不存在才创建——SDK 里没有这个函数，得自己写。

为什么必须这样：导入脚本会被反复执行（改一题跑一次），不做幂等就会
每次都建一份同名题库，几天后列表里全是重复名字，谁也说不清哪份是现行版。
"""
for dataset in client.list_datasets(dataset_name=dataset_name):
print(f"已存在同名题库，直接复用：{dataset.name}（{dataset.example_count} 题）")
return dataset
dataset = client.create_dataset(dataset_name, description=description)
print(f"新建题库：{dataset.name} id={dataset.id}")
return dataset

def main():
load_dotenv() # 读 .env：Key、端点、开关都来自这里
client = Client() # 认证由 Client() 自己从环境变量取

rows = load_cases(CSV_PATH)
print(f"CSV 读到 {len(rows)} 道题")

dataset = get_or_create_dataset(client, DATASET_NAME, DESCRIPTION)

# 幂等的第二半：题库存在不代表题在里面。重复跑脚本时若不去重，题会翻倍。
existing = {e.inputs.get("question") for e in client.list_examples(dataset_id=dataset.id)}
todo = [to_example(r) for r in rows if r["question"] not in existing]
print(f"其中 {len(rows) – len(todo)} 道已在平台上，本次只补 {len(todo)} 道")

if not todo:
print("题库已是最新，无需写入")
return
response = client.create_examples(dataset_id=dataset.id, examples=todo)
print(f"写入完成：count={response['count']}，返回 id 数={len(response['example_ids'])}")

if __name__ == "__main__":
main()

CSV 读到 5 道题
新建题库：math_agent_eval_cases id=f2b9be48-7a1e-536b-832f-c353b37e8eef
其中 0 道已在平台上，本次只补 5 道
写入完成：count=5，返回 id 数=5

第一次跑的输出就这四行，逐行有下文可查：读到 5 道题（表格解析没问题）、题库是新建的并带回一个 id、平台上原本 0 道、本次补 5 道、服务端确认接了 5 条并返回 5 个 id。

5.2 三个转换：CSV 的列名怎么变成平台的字段

to_example 是整份脚本里唯一"没有标准答案"的地方，因为平台没有规定题面必须叫什么名字。转换规则完全由这段代码自己定：

第四行值得说明：metadata 里那条 "source": "csv" 是代码硬加的，CSV 里没有这一列。它的用处很实在——题库里迟早会出现不是从 CSV 来的题（比如手工在界面上补的、从某条轨迹转出来的），有一个来源标记，将来查"哪些题不在事实源里"就有依据。这是给未来的对账留钩子，不是装饰。

5.3 这份脚本里三处"为什么这么写"

**第一处：读 CSV 用 utf-8-sig。**3.4 节第二个坑的直接修复，且 open 时传 newline=""（让 csv 自己处理 \\r\\n）。

第二处：get_or_create_dataset 用 for 遍历而不是 if client.list_datasets(…)。2.4 那条惰性规则在这里是功能性的，不是风格问题：list_datasets 返回生成器，未消费时一个请求都不发。如果写成"判断这个返回值是否为真"，生成器对象永远为真，于是同名检查会永远认为题库已存在，脚本再也不会建新库。这是一个不报错的逻辑死法。

**第三处：if not todo: return 在 create_examples 之前。**看起来像多余的防御，实际是必需的：第九节会看到 create_examples(examples=[]) 抛 ValueError。题库已经最新时若无条件调用接口，脚本会以异常退出，退出码非零，而实际上什么都没做错。这在 CI 里就是一次假失败。

5.4 客户端替你补了什么

脚本里没有一行提到 data_type、transformations、created_at、use_source_run_io、dataset_id、extra.metadata.runtime，但 2.4 抓到的请求体里它们全都在。把这张"脚本写的 / 网络上出现的"对照列出来，是理解这套 SDK 的关键一步：

这张表有两个实操价值。**第一，别把"脚本没写"当成"平台上没有"。**界面上看到题库被标成某种类型、看到题目带了创建时间，不要去找是哪行代码写的。第二，幂等的困难来自 created_at 与"不带 id"这两件事同时成立——第六节展开。

六、幂等：同一份脚本要敢跑第二遍

6.1 先看不做幂等会怎样

"改一题跑一次"是题库维护的日常节奏：加两道题、修一个参考值，都靠同一个脚本上平台。如果脚本没有幂等设计，这个日常会稳步把题库搞坏。这个实验刻意跳过所有去重，只反复调用 create_examples：

"""probe_no_idempotency.py —— 不写幂等会发生什么：同一个导入脚本裸跑三遍。

这里刻意跳过 create_dataset.py 里「先查已存在的题、只补差集」那一步，
直接反复 create_examples，看平台上这道题库的题量怎么变。
"""
import csv
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8404)
srv.start(8404)
srv.reset()
from langsmith import Client

client = Client()
dataset = client.create_dataset("bare_import_probe", description="重复导入实验")

with open("math_agent_eval_cases.csv", encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))

def to_example(row):
"""和 create_dataset.py 里一模一样的转换。"""
return {"inputs": {"question": row["question"]},
"outputs": {"answer": row["answer"]},
"metadata": {"case_type": row["case_type"]}}

for run in (1, 2, 3):
response = client.create_examples(dataset_id=dataset.id, examples=[to_example(r) for r in rows])
total = srv.count("POST", "/examples/bulk") * len(rows)
print(f"第{run}次导入：本次返回 count={response['count']}，平台上该题库累计 {total} 题")

back = list(client.list_examples(dataset_id=dataset.id))
print(f"读回来一共 {len(back)} 条记录，其中不同题面只有 {len({e.inputs['question'] for e in back})} 个")
print("写入请求发了几次:", srv.bulk_posts(), "（每次都真的被服务端接收）")

第1次导入：本次返回 count=5，平台上该题库累计 5 题
第2次导入：本次返回 count=5，平台上该题库累计 10 题
第3次导入：本次返回 count=5，平台上该题库累计 15 题
读回来一共 15 条记录，其中不同题面只有 5 个
写入请求发了几次: 3 （每次都真的被服务端接收）

三行输出把后果写得非常直白：每次都返回 count=5，都算成功，平台上从 5 题涨到 15 题，而题面只有 5 个不同的。

这里要盯住 count=5 这个返回值。它是服务端说"我接收了 5 条"，不是"我保证题库里有 5 条"。把成功写回当成成功去重，是最常见的一次误读。

题库变成 15 道里有 10 道重复，后果不是"难看"，而是成绩单直接失真，且失真方向不可预测：

• **分母变了。**通过率的分母从 5 变成 15。如果重复题完全等价，比例碰巧还看得出来；一旦不同轮次的题目内容有细微差别（比如中途改过一次参考值），比例就无法解释了。
• **加权变了。**某个类型的题被重复了几次，那个类型在总分里的权重就被悄悄抬高。若 big 类恰好是重复最多的那类，成绩单会显示"大数乘法表现良好"，而这只是重复计数的产物。
• **成本变了。**批量跑一遍要调用 Agent 15 次而不是 5 次。这个是真金白银的调用费和时间。
• **对比失效。**这一轮 15 题、下一轮修完是 5 题，两份成绩单的并排比较里，最大的差异来源不是 Agent 变好了，而是题量变了。

最后一条最阴：它会让"我们优化之后通过率上升了"这个结论变得无法辩护。

6.2 为什么客户端不做幂等

上一节的输出里那条关键事实是：写入请求发了 3 次，每次都是新造的题。原因在 5.4 那张表里已经埋好了——客户端不带 id，且每条都自动补 created_at。没有 id，服务端就无从判断"这是不是那道已存在的题"；有 created_at，每次写入在语义上都是"新建一条记录"。

所以这不是 SDK 的缺陷，是它选的默认语义：create_examples 是插入，不是按主键 upsert。名字里的 “Upsert”（返回类型叫 UpsertExamplesResponse）容易误导，但幂等键必须由调用方给（在 ExampleCreate 里显式写 id），不给就一律当新记录。
**但显式给 id 时，客户端确实会把它原样上行。**这一点单独测了一遍，因为它是唯一一条能拿到 upsert 语义的入口：

"""probe_explicit_id.py —— 显式给 id 时客户端带不带、重复给同一个 id 平台会怎样。

要区分两件事：客户端是否把 id 上行（可实测），以及服务端是否按 id 去重或更新
（本地假服务端不实现 upsert 语义，因此本文不对平台侧行为下任何结论）。
"""
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8412)
srv.start(8412)
srv.reset()
from langsmith import Client

FIXED_ID = "11111111-2222-3333-4444-555555555555"
client = Client()
dataset = client.create_dataset("explicit_id_probe", description="显式 id")

mark = len(srv.REQUESTS)
first = client.create_examples(dataset_id=dataset.id, examples=[
{"id": FIXED_ID, "inputs": {"question": "带 id 的题"}, "outputs": {"answer": "1|1"}}])
print("第一次写入返回:", {key: value for key, value in first.items() if key != "example_ids"},
"| example_ids:", [i for i in first["example_ids"]])
sent = next(b for method, path, query, b in srv.REQUESTS[mark:] if path == "/examples/bulk")
print("上行请求体里这一条的键:", sorted(sent[0]))
print("上行的 id 是不是脚本里那个:", sent[0]["id"] == FIXED_ID)

mark = len(srv.REQUESTS)
second = client.create_examples(dataset_id=dataset.id, examples=[
{"id": FIXED_ID, "inputs": {"question": "带同一个 id 的题"}, "outputs": {"answer": "2|2"}}])
print("第二次用同一个 id 写入返回:", {key: value for key, value in second.items() if key != "example_ids"})
back = list(client.list_examples(dataset_id=dataset.id))
print("服务端现在有几条记录:", len(back), "| 不同 id 数:", len({example.id for example in back}))
print("（本文的假服务端只做追加，不实现按 id 更新，所以这个 2 不代表平台行为）")

第一次写入返回: {'count': 1, 'as_of': None} | example_ids: ['11111111-2222-3333-4444-555555555555']
上行请求体里这一条的键: ['created_at', 'dataset_id', 'id', 'inputs', 'outputs', 'use_source_run_io']
上行的 id 是不是脚本里那个: True
第二次用同一个 id 写入返回: {'count': 1, 'as_of': None}
服务端现在有几条记录: 2 | 不同 id 数: 1
（本文的假服务端只做追加，不实现按 id 更新，所以这个 2 不代表平台行为）

这份输出能证明的和不能证明的要分清。能证明：id 会进入请求体，值就是脚本里写的那个；不带 id 时客户端不会自己造一个。不能证明：同一个 id 写两次，平台上最终是一道题还是两道题。本文的假服务端只实现追加，所以那个"2 条记录、1 个不同 id"是本文代码的行为，不是平台行为。真实平台怎么处理重复 id，需要在有账号的环境里另做验证——本文把它列进"待验证清单"，而不是猜一个答案写进结论。

6.3 两段式幂等：先查库，再补差集

create_dataset.py 里的幂等由两个动作组成，缺一不可：

第一段：题库级。get_or_create_dataset 保证同名题库只有一份。它靠 list_datasets(dataset_name=…) 的等值查询实现，命中就复用返回的那个对象。

**第二段：题目级。**复用题库之后，还要拿平台已有的题面集合做一次差集，只写不在里面的。

existing = {e.inputs.get("question") for e in client.list_examples(dataset_id=dataset.id)}
todo = [to_example(r) for r in rows if r["question"] not in existing]

第一行是集合推导，遍历生成器（因此真的会发请求，而且是分页发，第八节会看到这个集合变大之后的请求数）。第二行是纯本地的差集。

效果就是同一个脚本连跑三遍的输出：

CSV 读到 5 道题
已存在同名题库，直接复用：math_agent_eval_cases（5 题）
其中 5 道已在平台上，本次只补 0 道
题库已是最新，无需写入

CSV 读到 5 道题
已存在同名题库，直接复用：math_agent_eval_cases（5 题）
其中 5 道已在平台上，本次只补 0 道
题库已是最新，无需写入

第二遍和第三遍的输出完全相同，这正是幂等的定义：重复执行不产生额外效果。第二遍那句"其中 5 道已在平台上，本次只补 0 道"里，0 是差集算出来的；紧接着的"题库已是最新，无需写入"说明 create_examples 根本没被调用。

这两段合起来还有一个容易被忽略的性质：这个脚本是同步器，不是创建器。它对平台的期望状态来自 CSV，实际状态来自平台，动作是把差集补上。这条性质让"改一题跑一次"这个日常变得安全，但它有一个明确的能力边界——它只补不删。CSV 里删掉一道题，重跑脚本不会把那道题从平台上拿掉，平台上会留下一道孤儿题。要处理删除，得显式做双向比对（update_examples 与 delete_examples 是另外两个接口，本文没有展开；delete_examples 默认还是软删，需要 hard_delete=True 才硬删，这一点在第十一节会作为设计约束提一句）。

6.4 幂等判断的完整决策树

#mermaid-svg-9WpOYaYz59pI6icD{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-9WpOYaYz59pI6icD .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-9WpOYaYz59pI6icD .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-9WpOYaYz59pI6icD .error-icon{fill:#552222;}#mermaid-svg-9WpOYaYz59pI6icD .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-9WpOYaYz59pI6icD .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-9WpOYaYz59pI6icD .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-9WpOYaYz59pI6icD .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-9WpOYaYz59pI6icD .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-9WpOYaYz59pI6icD .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-9WpOYaYz59pI6icD .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-9WpOYaYz59pI6icD .marker{fill:#333333;stroke:#333333;}#mermaid-svg-9WpOYaYz59pI6icD .marker.cross{stroke:#333333;}#mermaid-svg-9WpOYaYz59pI6icD svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-9WpOYaYz59pI6icD p{margin:0;}#mermaid-svg-9WpOYaYz59pI6icD .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-9WpOYaYz59pI6icD .cluster-label text{fill:#333;}#mermaid-svg-9WpOYaYz59pI6icD .cluster-label span{color:#333;}#mermaid-svg-9WpOYaYz59pI6icD .cluster-label span p{background-color:transparent;}#mermaid-svg-9WpOYaYz59pI6icD .label text,#mermaid-svg-9WpOYaYz59pI6icD span{fill:#333;color:#333;}#mermaid-svg-9WpOYaYz59pI6icD .node rect,#mermaid-svg-9WpOYaYz59pI6icD .node circle,#mermaid-svg-9WpOYaYz59pI6icD .node ellipse,#mermaid-svg-9WpOYaYz59pI6icD .node polygon,#mermaid-svg-9WpOYaYz59pI6icD .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-9WpOYaYz59pI6icD .rough-node .label text,#mermaid-svg-9WpOYaYz59pI6icD .node .label text,#mermaid-svg-9WpOYaYz59pI6icD .image-shape .label,#mermaid-svg-9WpOYaYz59pI6icD .icon-shape .label{text-anchor:middle;}#mermaid-svg-9WpOYaYz59pI6icD .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-9WpOYaYz59pI6icD .rough-node .label,#mermaid-svg-9WpOYaYz59pI6icD .node .label,#mermaid-svg-9WpOYaYz59pI6icD .image-shape .label,#mermaid-svg-9WpOYaYz59pI6icD .icon-shape .label{text-align:center;}#mermaid-svg-9WpOYaYz59pI6icD .node.clickable{cursor:pointer;}#mermaid-svg-9WpOYaYz59pI6icD .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-9WpOYaYz59pI6icD .arrowheadPath{fill:#333333;}#mermaid-svg-9WpOYaYz59pI6icD .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-9WpOYaYz59pI6icD .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-9WpOYaYz59pI6icD .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-9WpOYaYz59pI6icD .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-9WpOYaYz59pI6icD .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-9WpOYaYz59pI6icD .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-9WpOYaYz59pI6icD .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-9WpOYaYz59pI6icD .cluster text{fill:#333;}#mermaid-svg-9WpOYaYz59pI6icD .cluster span{color:#333;}#mermaid-svg-9WpOYaYz59pI6icD div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-9WpOYaYz59pI6icD .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-9WpOYaYz59pI6icD rect.text{fill:none;stroke-width:0;}#mermaid-svg-9WpOYaYz59pI6icD .icon-shape,#mermaid-svg-9WpOYaYz59pI6icD .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-9WpOYaYz59pI6icD .icon-shape p,#mermaid-svg-9WpOYaYz59pI6icD .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-9WpOYaYz59pI6icD .icon-shape .label rect,#mermaid-svg-9WpOYaYz59pI6icD .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-9WpOYaYz59pI6icD .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-9WpOYaYz59pI6icD .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-9WpOYaYz59pI6icD :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

否

是

命中

未命中

todo 为空

todo 非空

读 CSV 得到 rows

本地校验通过

中止：0 个请求

list_datasets 命中同名题库

复用该题库对象

create_dataset 新建

list_examples 取已有题面集合 existing

existing 为空集

todo = rows 中题面不在 existing 的部分

跳过写入，正常退出

create_examples 只写 todo

list_examples 回读核对条数

这张图里有两个出口是"什么都不做但正常结束"（G 和 X），它们对应两种完全不同的原因：X 是数据有问题，必须让人看到并修；G 是状态已收敛，不该制造噪音。把这两种情况混成同一个退出码，是 CI 里最常见的假失败来源。

6.5 去重键为什么是题面，以及它哪里脆

用题面当幂等键是个直观选择，但它有三个弱点，前两个已经在 3.4 实测过：

弱点一：空白字符让它失效。'12 乘以 34 等于多少 '（尾部多一个空格）和 '12 乘以 34 等于多少' 是两个不同的题面，差集算不出重复，于是题库里出现两道看起来一模一样的题。3.4 的第四个坑证明了 CSV 读取不会清理空白——清理责任在幂等键这里。本文的处理办法不是在读 CSV 时 strip()（那会让写进去的题面和你看到的不一样），而是在校验阶段禁止空白（第十节第 3 条断言），把问题挡在键值形成之前。

弱点二：改题面等于换题。修一道题最常见的动作就是把题面写得更准确，而这会改变去重键。结果不是替换，而是新增：旧题面那道题仍然在平台上，新题面那道题也被写进去了，题库从 5 道变 6 道，而且没有任何一处提示你这两道其实是同一道题的两个版本。这是"只补不删"这条边界最疼的地方。

对这一步的可操作建议是：改题面时同时删旧题，或者干脆给题目引入一个稳定业务编号（比如 CSV 加一列 case_id，去重键用它）。后者更彻底，因为它把"题目的身份"和"题目的内容"分开了——题面可以随便改，case_id 不变，幂等就能从"只补不删"升级成"能补能改"。代价是要有人负责编号不重复、不复用，这是管理成本，不是技术成本。这个做法属于经验判断，本文没有实测。

**弱点三：题面不是一个字段时，键要自己拼。**真实业务的题面往往是 {question, history, account} 三个字段一起才算一道题（第十一节的结构测试里就有这种题）。那时 existing 集合的构造方式必须改成把这几个字段组合成稳定的键（比如按固定顺序拼一个元组字符串）。任何"随手取 inputs 的某一项"当键的写法，在多字段题库上都会漏判。

6.6 撞名：能不能建出两份同名题库

上一节说"同名题库只有一份"是靠先查再建实现的。那如果绕过这个检查、直接第二次 create_dataset 同名呢？答案取决于服务端策略，所以这件事必须标清楚证据来源：

第九节的错误面实验里，本文构造的服务端对同名题库回 409，客户端把它翻成 LangSmithConflictError，此时同名题库仍然是 1 份，"先查再建"被证明可以避免这次冲突。

必须诚实标注的边界是：真实平台的重名策略本文没有验证。不同租户、不同版本对"同名题库"的处理可能不同（有的允许并存、用 id 区分，有的直接拒绝）。这一条属于"在自己的环境里跑一次就知道"的清单——花两分钟建两次同名库，比在文档里找答案可靠。但无论平台是哪种策略，"先查再建"都是对的：允许并存时它防止你建出第二份，拒绝时它让你绕开冲突。

七、写进去不等于写对了：读回来核对

7.1 回读脚本

"""check_dataset.py —— 导入之后把题库读回来核对：值对不对、类型对不对、标签丢没丢。

为什么这一步不能省：create_examples 返回 count=5 只说明「服务端接了 5 条」，
不说明「存进去的还是你那 5 条」。写入返回的是计数，回读拿到的才是内容。
"""
import csv
import json
import os

from dotenv import load_dotenv
from langsmith import Client

CSV_PATH = "math_agent_eval_cases.csv"
DATASET_NAME = "math_agent_eval_cases"

def main():
load_dotenv()
client = Client()

dataset = client.read_dataset(dataset_name=DATASET_NAME) # 按名字读，读的是平台上的现行版
print(f"题库：{dataset.name} 描述：{dataset.description}")
print(f"对象上的 example_count: {dataset.example_count}（这是读接口返回的值）")

with open(CSV_PATH, encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))

examples = list(client.list_examples(dataset_id=dataset.id)) # 生成器，遍历才真取数
print(f"CSV 行数: {len(rows)} | 平台读回题数: {len(examples)}")

first = examples[0]
print("第一题 inputs:", json.dumps(first.inputs, ensure_ascii=False))
print("第一题 outputs:", json.dumps(first.outputs, ensure_ascii=False))
print("第一题 metadata:", json.dumps(first.metadata, ensure_ascii=False))

print("outputs 里答案的类型:", type(first.outputs["answer"]).__name__,
"| CSV 里答案的类型:", type(rows[0]["answer"]).__name__)

aligned = [e.inputs["question"] == r["question"] and e.outputs["answer"] == r["answer"]
for e, r in zip(examples, rows)]
print("逐题与 CSV 是否完全一致:", aligned, "| 全部一致:", all(aligned))

big = next(e for e in examples if "," in e.outputs["answer"])
print("带千分位的参考答案是否原样保留:", big.outputs["answer"])
print("按 | 拆出来的两种写法:", json.dumps(big.outputs["answer"].split("|"), ensure_ascii=False))

tags = sorted({e.metadata["case_type"] for e in examples})
print("metadata 里的标签种类:", tags)

if __name__ == "__main__":
main()

题库：math_agent_eval_cases 描述：计算器 Agent 题库：输入一句乘法，参考值是两个大数乘积
对象上的 example_count: 5（这是读接口返回的值）
CSV 行数: 5 | 平台读回题数: 5
第一题 inputs: {"question": "12 乘以 34 等于多少"}
第一题 outputs: {"answer": "408|408"}
第一题 metadata: {"case_type": "small", "source": "csv"}
outputs 里答案的类型: str | CSV 里答案的类型: str
逐题与 CSV 是否完全一致: [True, True, True, True, True] | 全部一致: True
带千分位的参考答案是否原样保留: 1082152022374638|1,082,152,022,374,638
按 | 拆出来的两种写法: ["1082152022374638", "1,082,152,022,374,638"]
metadata 里的标签种类: ['big', 'edge', 'small']

7.2 count 是接收数，example_count 是快照，读回才是内容

这三样东西经常被混用，而它们回答的是三个不同问题：

下面这段实测把第二条钉死。建库那一刻拿到的 dataset 对象，example_count 是 0——这份题库里随后写了 5 道题，但那个对象不会自己变。它是一次 HTTP 响应的解析结果，不是活动句柄。想看新值必须重新查（read_dataset 或 list_datasets）。本文的输出里有一行专门做这个对比：对象上的 example_count: 5，那是 check_dataset.py 重新读回来的值，与 create_dataset.py 里那个对象上的 0 不是同一回事。

这个坑在导入脚本里的表现形式很具体：写完题目顺手打印 dataset.example_count，会得到 0，然后有人以为写入失败。

"""probe_snapshot.py —— 刚建好的题库对象，example_count 是多少？写入之后它会变吗？

一句话结论：Dataset 对象是一次 HTTP 响应的解析结果，不是活动句柄。
"""
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8413)
srv.start(8413)
srv.reset()
from langsmith import Client

client = Client()
dataset = client.create_dataset("snapshot_probe", description="快照字段")
print("刚建好时对象上的 example_count:", dataset.example_count)

client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"question": f"第{i}题"}, "outputs": {"answer": "1|1"}} for i in range(3)])
print("写入 3 道题之后，同一个对象上的 example_count:", dataset.example_count)

fresh = client.read_dataset(dataset_name="snapshot_probe")
print("重新读一次拿到的 example_count:", fresh.example_count, "（这才是当前值）")

刚建好时对象上的 example_count: 0
写入 3 道题之后，同一个对象上的 example_count: 0
重新读一次拿到的 example_count: 3 （这才是当前值）

这三行输出把「快照」这个词落实了：对象上的数从 0 开始、写完三道题之后仍是 0、只有重新调用 read_dataset 才拿到 3。所以任何"写入之后确认题量"的代码，都必须重新查一次；顺手复用手里那个 dataset 对象，看到的永远是建库那一刻的值。

7.3 类型：全程 str，没人替你转

输出里 outputs 里答案的类型: str | CSV 里的类型: str 是同一件事的两端。这条链路上没有任何一环做类型推断——CSV 是文本，inputs/outputs 的值是字符串，写上去还是字符串。

它的影响面在判分那一刻。如果评分代码写成 agent_answer == reference（一边是 "408" 一边可能是 408），Python 里 408 == "408" 是 False。这类不匹配不会抛异常，只会静默地把正确的回答判成错误。本节 7.5 的结构测试会给出对照：只要你在 outputs 里放的是整数，平台上读回来还是整数——变形不发生，但也不会发生转换。

所以口径要定死一处：要么全程按字符串判分，要么在评分器里做一次显式转换，且转换只写在一处。第十节的校验脚本禁止空白，就是在为"按字符串判分"这个选择兜底。

7.4 千分位原样保留，说明参考值不是"值"而是"文本"

输出最后两行是这套设计成立的关键：带千分位的参考答案是否原样保留: 1082152022374638|1,082,152,022,374,638。平台没有把这一格当成数字解析（那样会丢掉逗号或报错），也没有按 | 拆开（那样会破坏列语义）。它就是一个字符串。

于是"参考答案"这个词要精确理解：存的是可接受写法的集合，不是唯一真值。存下来的这个字符串本身没有意义，有意义的是评分器把它 split("|") 之后得到的两个候选写法（输出最后一行 ["1082152022374638", "1,082,152,022,374,638"]）。这也解释了为什么 3.4 那个"没加引号被截断"的坑那么危险——截断之后这一格里只剩一个残缺写法，两种候选变成一个残缺值，评分器的行为会完全变样，而数据形状看起来仍然正常。

7.5 结构不变形：多字段、嵌套、列表、整数

上面几节都在讲字符串题目。真实业务的题往往不止一个输入字段，参考值也不止一个值。这一节把一道"复杂题"写进去再读回来，逐层比对。

"""probe_shape.py —— 一道题的形状：字段能不能变形、缺了参考值会怎样、标签往哪放。

三个问题依次是：
1. inputs/outputs/metadata 里放嵌套结构、列表、整数，绕一圈回来还是不是原样；
2. 缺题面、缺参考值、参考值是空串，客户端拦不拦；判分时哪些题会取不到参考值；
3. 题库级 schema 与 metadata 用什么字段上行，题目级 split 写了能不能读回来。
"""
import json
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8409)
srv.start(8409)
srv.reset()
from langsmith import Client

client = Client()

print("== 一、结构变形测试：一题多输入、多参考值、嵌套与列表 ==")
dataset = client.create_dataset("shape_probe", description="题目形状")
payload = {
"inputs": {"question": "帮我算一下 12345678 乘以 87654321", "history": [], "account": "main"},
"outputs": {"answer": 1082152022374638, "expected_tool": "multiply",
"expected_args": {"a": 12345678, "b": 87654321}, "forbidden": ["delete_account"]},
"metadata": {"case_type": "multi", "tags": ["核心集", "中文"]},
}
client.create_examples(dataset_id=dataset.id, examples=[payload])
back = list(client.list_examples(dataset_id=dataset.id))[0]
print(" inputs 读回:", json.dumps(back.inputs, ensure_ascii=False))
print(" outputs 读回:", json.dumps(back.outputs, ensure_ascii=False))
print(" metadata读回:", json.dumps(back.metadata, ensure_ascii=False))
print(" 键集合是否原样保留:", set(back.inputs) == set(payload["inputs"]),
set(back.outputs) == set(payload["outputs"]), set(back.metadata) == set(payload["metadata"]))
print(" answer 的类型:", type(back.outputs["answer"]).__name__,
"| 值是否等于原始乘积:", back.outputs["answer"] == 12345678 * 87654321)
print(" 空列表 history:", repr(back.inputs["history"]),
"| 嵌套 dict:", back.outputs["expected_args"], "| list:", back.outputs["forbidden"])

print("\\n== 二、畸形题目：客户端一条都没拦 ==")
bad_dataset = client.create_dataset("bad_shape_probe", description="畸形题目")
client.create_examples(dataset_id=bad_dataset.id, examples=[
{"inputs": {"question": "有题面没参考值"}},
{"outputs": {"answer": "408|408"}},
{"inputs": {"question": "参考值是空串"}, "outputs": {"answer": ""}},
{"inputs": {}, "outputs": {}},
{"inputs": {"question": "参考值放错了键"}, "outputs": {"result": "408"}},
])
bad_back = list(client.list_examples(dataset_id=bad_dataset.id))
print(f" 写 5 条畸形题目，平台上真实存在 {len(bad_back)} 条")
for index, example in enumerate(bad_back, 1):
print(f" 第{index}条 inputs={json.dumps(example.inputs, ensure_ascii=False)}"
f" outputs={json.dumps(example.outputs, ensure_ascii=False)}")
missing = [index for index, example in enumerate(bad_back, 1)
if not (example.outputs or {}).get("answer")]
print(" 判分时按 reference_outputs['answer'] 取值，取不到的是第", missing, "条")

print("\\n== 三、题库级 schema 与 metadata 落在哪些字段上 ==")
mark = len(srv.REQUESTS)
client.create_dataset("schema_probe", description="带 schema",
inputs_schema={"question": {"type": "string"}},
outputs_schema={"answer": {"type": "string"}},
metadata={"owner": "qa", "scene": "calculator"})
body = next(b for method, path, query, b in srv.REQUESTS[mark:] if path == "/datasets")
print(" 请求体字段:", sorted(body.keys()))
print(" inputs_schema_definition:", json.dumps(body.get("inputs_schema_definition"), ensure_ascii=False))
print(" 自定义 metadata 与 runtime 合并后的键:", sorted(body["extra"]["metadata"]))

print("\\n== 四、题目级 split：写得进去，读不读得回来 ==")
mark = len(srv.REQUESTS)
client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"question": "只属于回归集的题"}, "outputs": {"answer": "1|1"}, "split": "regression"}])
sent = next(b for method, path, query, b in srv.REQUESTS[mark:] if path == "/examples/bulk")
print(" 上行请求体里这一条的字段:", sorted(sent[0]))
split_back = list(client.list_examples(dataset_id=dataset.id, splits=["regression"]))
print(" 用 splits=['regression'] 过滤，客户端实际发的查询参数:",
[q for method, path, q, b in srv.REQUESTS[mark:] if path == "/examples"][0])
from langsmith.schemas import Example
print(" 读回来的条数:", len(split_back), "（假服务端不解释过滤条件，这个数只证明请求发出去了）")
print(" Example 模型的字段里有没有 split:", "split" in Example.model_fields,
"| 有 split 的是写入模型 ExampleCreate:", "split" in __import__("langsmith.schemas", fromlist=["x"]).ExampleCreate.model_fields)

== 一、结构变形测试：一题多输入、多参考值、嵌套与列表 ==
inputs 读回: {"question": "帮我算一下 12345678 乘以 87654321", "history": [], "account": "main"}
outputs 读回: {"answer": 1082152022374638, "expected_tool": "multiply", "expected_args": {"a": 12345678, "b": 87654321}, "forbidden": ["delete_account"]}
metadata读回: {"case_type": "multi", "tags": ["核心集", "中文"]}
键集合是否原样保留: True True True
answer 的类型: int | 值是否等于原始乘积: True
空列表 history: [] | 嵌套 dict: {'a': 12345678, 'b': 87654321} | list: ['delete_account']

== 二、畸形题目：客户端一条都没拦 ==
写 5 条畸形题目，平台上真实存在 5 条
第1条 inputs={"question": "有题面没参考值"} outputs=null
第2条 inputs=null outputs={"answer": "408|408"}
第3条 inputs={"question": "参考值是空串"} outputs={"answer": ""}
第4条 inputs={} outputs={}
第5条 inputs={"question": "参考值放错了键"} outputs={"result": "408"}
判分时按 reference_outputs['answer'] 取值，取不到的是第 [1, 3, 4, 5] 条

== 三、题库级 schema 与 metadata 落在哪些字段上 ==
请求体字段: ['data_type', 'description', 'extra', 'inputs_schema_definition', 'name', 'outputs_schema_definition', 'transformations']
inputs_schema_definition: {"question": {"type": "string"}}
自定义 metadata 与 runtime 合并后的键: ['owner', 'runtime', 'scene']

== 四、题目级 split：写得进去，读不读得回来 ==
上行请求体里这一条的字段: ['created_at', 'dataset_id', 'inputs', 'outputs', 'split', 'use_source_run_io']
用 splits=['regression'] 过滤，客户端实际发的查询参数: {'offset': '0', 'splits': 'regression', 'inline_s3_urls': 'True', 'limit': '100', 'dataset': '9e84be05-6c86-504e-9fad-739eee52d70a'}
读回来的条数: 2 （假服务端不解释过滤条件，这个数只证明请求发出去了）
Example 模型的字段里有没有 split: False | 有 split 的是写入模型 ExampleCreate: True

第一组输出是这份实验里唯一一组"好消息"，四条都要记住，因为第十一节的设计建议全部建立在它们之上：

• 键集合原样保留，三个字段各自比对都是 True。平台不会重排、不会合并、不会给你补默认键。
• 整数保持整数：answer 放进去是 1082152022374638（int），读回来 type 是 int，值等于原始乘积。这条与 7.3 不冲突——7.3 说的是"没人替你转换"，这里说的是"类型是你给的，它就存你给的"。CSV 那条链路全是字符串，而 Python 字典可以直接给整数。
• 空列表保留：history: [] 读回来还是 []，不会被当成"没这个字段"丢掉。这点对多轮对话类题目极其重要：如果空历史被丢掉，判分时代码里 inputs.get("history", []) 侥幸能跑，而换成严格取值就会 KeyError。
• 嵌套结构与列表保留：expected_args 是个 dict、forbidden 是个 list，读回来形状不变。

要标注的边界：这组实验证明的是客户端与假服务端这一圈不改结构。平台侧对深层嵌套、对超长值、对附件（attachments 字段）有没有尺寸或深度限制，本文没有验证，也不属于能从 SDK 读出来的信息。

八、题库长大之后：分批、分页、并发

五十道题时上面所有细节都不痛。痛的是五百道、五万道，以及"导入要跑十分钟还是十秒"这种问题。这一节把三个数字测出来。

"""probe_scale.py —— 题库变大之后的三件事：写入分批、读取分页、并发上限。

关心的不是「能不能写进去」，而是「写成几次、读成几次」——
这两个数字直接决定导入耗时和对平台的请求压力。
"""
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8406)
srv.start(8406)
srv.reset()
from langsmith import Client

client = Client()
dataset = client.create_dataset("scale_probe", description="批量与分页")

BIG = "x" * 4_000_000 # 每条题目带 4MB 输入，用来把批次撑爆
mark = len(srv.REQUESTS)
response = client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"question": f"第{i}题", "context": BIG}, "outputs": {"answer": "1|1"}}
for i in range(6)])
print(f"6 条 4MB 的题目（约 24MB）：写入请求 {srv.request_log()[mark:].count('POST /examples/bulk')} 次"
f" | 返回 count={response['count']}")

mark = len(srv.REQUESTS)
client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"question": "超大单题", "context": "y" * 25_000_000}, "outputs": {"answer": "1|1"}}])
print("单条就超过 20MB：写入请求",
srv.request_log()[mark:].count("POST /examples/bulk"), "次（不会被丢掉，单独成批）")

mark = len(srv.REQUESTS)
response = client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"question": f"小题目{i}"}, "outputs": {"answer": "1|1"}} for i in range(250)])
print(f"250 条小题：写入请求 {srv.request_log()[mark:].count('POST /examples/bulk')} 次"
f" | 返回 count={response['count']}（只按体积切，不按条数切）")

mark = len(srv.REQUESTS)
got = list(client.list_examples(dataset_id=dataset.id))
pages = [q.get("limit") for m, p, q, b in srv.REQUESTS[mark:] if p == "/examples"]
print(f"读回全部 {len(got)} 题：读取请求 {len(pages)} 次 | 每次带的 limit = {sorted(set(pages))}")

mark = len(srv.REQUESTS)
limited = list(client.list_examples(dataset_id=dataset.id, limit=1000))
pages = [q.get("limit") for m, p, q, b in srv.REQUESTS[mark:] if p == "/examples"]
print(f"显式要求 limit=1000：实际带的 limit = {sorted(set(pages))} | 拿到 {len(limited)} 题")

for bad in (0, 4):
try:
client.create_examples(dataset_id=dataset.id, examples=[{"inputs": {"q": "1"}}], max_concurrency=bad)
except Exception as exc:
print(f"max_concurrency={bad} -> {type(exc).__name__} {exc}")

6 条 4MB 的题目（约 24MB）：写入请求 2 次 | 返回 count=6
单条就超过 20MB：写入请求 1 次（不会被丢掉，单独成批）
250 条小题：写入请求 1 次 | 返回 count=250（只按体积切，不按条数切）
读回全部 257 题：读取请求 3 次 | 每次带的 limit = ['100']
显式要求 limit=1000：实际带的 limit = ['100'] | 拿到 257 题
max_concurrency=0 -> ValueError max_concurrency must be between 1 and 3
max_concurrency=4 -> ValueError max_concurrency must be between 1 and 3

8.1 三条规则

规则一：写入按体积切批，一大约 20MB。6 条各带 4MB 输入的题目（合计约 24MB）被切成 2 个 POST /examples/bulk；250 条小题只用 1 次；单条 25MB 的那道题没有被丢弃，仍然发了一次请求（超大条目单独成批）。注意切分的依据是体积而不是条数——这是输出里"250 条小题只用 1 次"这一行的直接推论。20MB 这个阈值是本文构造出来的观察值：4MB×6 触发成 2 批，说明阈值在 20MB 附近，但本文没有做二分测试去精确定位。准确说法应当是"实测在 24MB/6 条时切成 2 批，据此判断单批上限约为 20MB 量级"。

规则二：读取固定每页 100 条。读回 257 题发了 3 次 GET /examples，每次带的 limit 都是 100。更值得注意的是"显式要求 limit=1000"那一行：客户端仍然按 100 一页发请求，最后给你 257 条。也就是说生成器会替你翻页，limit 是总量上限而不是页大小。这条对写代码有直接影响：不要为了"少发点请求"而调大 limit，请求数不会因此减少；也不要以为 limit=1000 就只发一个请求。

规则三：并发有硬闸门，只允许 1 到 3。max_concurrency=0 和 =4 都抛 ValueError: max_concurrency must be between 1 and 3。这是 Pydantic 的 Field(ge=1, le=3) 约束（2.2 打印的签名里能看到 Annotated[int, Field(ge=1, le=3)]），属于本地拦截、零请求。默认值是 1。

8.2 这三条合起来意味着什么

把三个数字合到一个成本模型上看。下面这段属于经验判断，本文没有测真实平台的限流与吞吐：

导入 N 道题的写入请求数约等于 max(1, 向上取整(总字节数 / 20MB))，几乎与题量无关——也就是说一万道小题和一百道小题的导入成本基本相同。真正的成本在读取和批量跑：读取要发 向上取整(N/100) 次请求，而批量评估要为每道题跑一次 Agent。所以"导入慢"在真实项目里几乎不是问题，"跑一遍慢"才是。这个判断的实践含义很直接：如果你感觉导入很慢，第一个要查的不是题量，而是是不是哪里在循环里逐条写入（每调用一次 create_examples 就是至少一次请求，把一百次循环写成一次批量调用是纯收益的改动）。

另一个可以推出来但本文没测的点：既然读取按 100 一页翻，每次做幂等差集都会把整库题目拉回来一遍。五百道题的库、每天导入十次，就是一万次记录的读取。量不大，但它解释了为什么幂等实现要从"每次全量比对"演进成"按 as_of 或按 metadata 过滤"——list_examples 的签名里有 as_of（按时间点取历史版本）和 metadata、filter 三个过滤参数，这些参数的服务端语义本文无法验证（假服务端只按 dataset 和分页过滤，其他条件一律忽略），但**"能少拉就少拉"这个方向是确定的**。

8.3 什么时候该把一份题库拆成几份

这是纯经验判断，本文没有对应实验，但它是题库到了一定规模后必然遇到的决定，给一个可操作的分界线：

本文这份计算器题库属于最小的情形：5 道题、一种形状、全部必跑、无外部状态，因此完全不拆。在不需要拆的时候提前拆好同样是个错误——它会把"哪个是现行版本"变成一个新的管理问题。

九、错误面：三层拦截与两种暴露方式

题库这一侧的接口很多，出错方式却只有三种：本地拦住（零请求）、要发请求才知道（有请求、可能已写一半）、根本不拦（写成功了，但数据是脏的）。第三种最贵，因为它把发现时机推到了成绩单出来之后。这一节按这三层把错误面测完。

"""probe_errors.py —— 题库侧的错误面：哪些在本地就被拦住、哪些要发请求才知道。

分层的意义在于反馈速度：本地拦住的，改一行就能重试；要发请求才知道的，
往往已经写了一半，得先想清楚怎么回滚。
"""
import sys

sys.path.insert(0, ".")
import dataset_server as srv

srv.configure(8407)
srv.start(8407)
srv.reset()
from langsmith import Client

client = Client()
dataset = client.create_dataset("error_probe", description="错误面实验")

def show(label, call):
"""跑一次调用，把结果统一成一行：要么没报错，要么异常类型加消息。"""
try:
call()
print(f" {label}: 没有报错")
except Exception as exc:
print(f" {label}: {type(exc).__name__} | {exc}")

print("== 一、本地就拦住：请求一个都不会发出去 ==")
mark = len(srv.REQUESTS)
show("同时传 dataset_id 和 dataset_name", lambda: client.create_examples(
dataset_id=dataset.id, dataset_name="error_probe", examples=[{"inputs": {"q": "1"}}]))
show("两个都不传", lambda: client.create_examples(examples=[{"inputs": {"q": "1"}}]))
show("新写法 examples 与旧写法 inputs 混用", lambda: client.create_examples(
dataset_id=dataset.id, examples=[{"inputs": {"q": "1"}}], inputs=[{"q": "2"}]))
show("参数名拼错（quizz）", lambda: client.create_examples(dataset_id=dataset.id, quizz=1))
show("旧写法里各列表长度不一致", lambda: client.create_examples(
dataset_id=dataset.id, inputs=[{"q": "1"}, {"q": "2"}], outputs=[{"a": "1"}]))
show("data_type 传字符串而不是枚举", lambda: client.create_dataset("bad_type", data_type="chat"))
print(f" 这一节实际发出的请求数: {len(srv.REQUESTS) – mark}（全部为 0 才算纯本地校验）")

print("\\n== 二、空题目列表：既不静默通过，也不发请求 ==")
mark = len(srv.REQUESTS)
show("examples=[]", lambda: client.create_examples(dataset_id=dataset.id, examples=[]))
show("examples=None", lambda: client.create_examples(dataset_id=dataset.id, examples=None))
print(f" 这一节实际发出的请求数: {len(srv.REQUESTS) – mark}")

print("\\n== 三、要发请求才知道的 ==")
show("往不存在的题库名里写", lambda: client.create_examples(
dataset_name="根本不存在的题库", examples=[{"inputs": {"q": "1"}}]))
show("读一个不存在的题库", lambda: client.read_dataset(dataset_name="根本不存在的题库"))
print(" 查一个不存在的题库名:", list(client.list_datasets(dataset_name="根本没有")),
"（不抛异常，返回空）")

print("\\n== 四、撞名：本文构造的服务端对同名题库回 409 ==")
show("第二次 create_dataset 同名", lambda: client.create_dataset("error_probe", description="同名第二次"))
found = list(client.list_datasets(dataset_name="error_probe"))
print(" 同名题库现在有几份:", len(found), "| 先查再建能否避免:", len(found) == 1)

== 一、本地就拦住：请求一个都不会发出去 ==
同时传 dataset_id 和 dataset_name: ValueError | Exactly one argument in each of the following groups must be defined: dataset_id, dataset_name
两个都不传: ValueError | Exactly one argument in each of the following groups must be defined: dataset_id, dataset_name
新写法 examples 与旧写法 inputs 混用: ValueError | Cannot specify 'inputs' when 'examples' is specified.
参数名拼错（quizz）: ValueError | Received unsupported keyword arguments: ('quizz',).
旧写法里各列表长度不一致: ValueError | Length of outputs (1) does not match length of inputs (2)
data_type 传字符串而不是枚举: AttributeError | 'str' object has no attribute 'value'
这一节实际发出的请求数: 0（全部为 0 才算纯本地校验）

== 二、空题目列表：既不静默通过，也不发请求 ==
examples=[]: ValueError | Must specify either 'examples' or 'inputs.'
examples=None: ValueError | Must specify either 'examples' or 'inputs.'
这一节实际发出的请求数: 0

== 三、要发请求才知道的 ==
往不存在的题库名里写: LangSmithNotFoundError | Dataset 根本不存在的题库 not found
读一个不存在的题库: LangSmithNotFoundError | Dataset 根本不存在的题库 not found
查一个不存在的题库名: [] （不抛异常，返回空）

== 四、撞名：本文构造的服务端对同名题库回 409 ==
第二次 create_dataset 同名: LangSmithConflictError | Conflict for /datasets. HTTPError('409 Client Error: Conflict for url: http://127.0.0.1:8407/datasets', '{"error": {"message": "Dataset error_probe already exists"}}')
同名题库现在有几份: 1 | 先查再建能否避免: True

9.1 第一层：本地就拦住，一个请求都不发

六行输出后面跟着一句"这一节实际发出的请求数: 0"，这个 0 是这一节的全部价值——参数形状错误不需要网络就能确认。逐条看：

• **dataset_id 与 dataset_name 二选一，且必须给一个。**两种错法（都给、都不给）报同一句话：Exactly one argument in each of the following groups must be defined: dataset_id, dataset_name。这句提示本身有点绕，它说的是"以下每一组互斥参数里，必须恰好定义一个"。
• 新旧写法混用会被拒：Cannot specify 'inputs' when 'examples' is specified.。这套 SDK 正在从"并排传 inputs/outputs/metadata 三个列表"过渡到"传 examples 一个列表"，两套不能混。看到这句报错就是撞上了旧教程。
• 参数名拼错直接列出来：Received unsupported keyword arguments: ('quizz',).。这一条比想象中友好——它没有把未知参数塞进请求体发出去，而是当场拒绝。拼错参数名的后果因此不是"静默无效"而是"立刻报错"。
• 旧写法三个列表长度不齐：Length of outputs (1) does not match length of inputs (2)。这个校验放在本地是对的，否则会把残缺数据写进平台。
• data_type 必须是枚举而不是字符串：AttributeError: 'str' object has no attribute 'value'。这条报错的形态很丑（它不是"参数校验失败"，而是内部实现漏出来的属性错误），但意思明确：data_type="chat" 不行，要 from langsmith.schemas import DataType 然后 data_type=DataType.chat。2.2 那条"默认 kv"的规则在这里落地：要么不传（用 kv），要么传枚举。

9.2 第二层：空列表既不静默、也不发请求

examples=[]: ValueError | Must specify either 'examples' or 'inputs.'
examples=None: ValueError | Must specify either 'examples' or 'inputs.'

两条都抛，消息还都写着"必须指定 examples 或 inputs"——但明明传了 examples=[]，这句提示读起来很别扭。这是一个实现细节漏到用户界面上的例子：内部判断用的是"是否为真值"，空列表因此和 None 走到同一分支。

**这条实测直接决定了 5.3 第三处那个 if not todo 是不是必需。**如果幂等算出的差集为空（题库已是最新），无条件调用会抛 ValueError、脚本非零退出，而实际什么都没做错。所以那句判空不是防御性冗余，而是幂等脚本能否安全复跑的必要条件。

9.3 第三层：要发请求才知道

• 往不存在的题库名里写、读不存在的题库，都抛 LangSmithNotFoundError。注意第一句里的顺序：客户端先把名字解析成 id（发一次 GET /datasets?limit=1&name=…，见 2.4），拿不到就抛，写入请求根本没发。这解释了为什么"用不存在的名字写入"不会造成半写状态。
• 但 list_datasets(dataset_name=不存在的名字) 不抛异常，返回空。这两个接口的不对称是本文最容易被写错的一处：查询用列表接口得到空、按名字读用读接口得到异常。写 get_or_create_dataset 时用的是前者（所以未命中就是空迭代、不会炸），而 check_dataset.py 用 read_dataset（题库不存在时会抛）。两种写法都对，但要知道自己在用哪一种语义。

9.4 第四层：客户端根本不拦

这一层没有异常可看，所以要用数据说话。7.5 那份 probe_shape.py 的第二组写了五条畸形题目：

== 二、畸形题目：客户端一条都没拦 ==
写 5 条畸形题目，平台上真实存在 5 条
第1条 inputs={"question": "有题面没参考值"} outputs=null
第2条 inputs=null outputs={"answer": "408|408"}
第3条 inputs={"question": "参考值是空串"} outputs={"answer": ""}
第4条 inputs={} outputs={}
第5条 inputs={"question": "参考值放错了键"} outputs={"result": "408"}
判分时按 reference_outputs['answer'] 取值，取不到的是第 [1, 3, 4, 5] 条

五条全部写入成功。其中四条在判分那一刻取不到参考值——这就是"第三层错误最贵"的具体含义：写入时零阻力，判分时静默降级。第 2 条尤其值得注意，它有参考值但没有题面，写成功了；跑批量评估时这道题会拿一个 None 去调 Agent，行为取决于 target 函数怎么写，可能是异常、可能是每次都同一句默认回答。

第 5 条是这一组里最阴的一条：{"result": "408"}。它不是畸形数据，只是键名和评分约定不一致。数据完整、内容正确、inputs 也齐，但评分器按 answer 取，就取不到。这条不属于"有人故意写脏数据"，而属于"两个人对同一份数据用了不同的字段名"，而 2.2 已经证明键名完全是自定义的——所以它必然会发生。

四层错误面收进一张图：

#mermaid-svg-02JlUTvn6x8lv0cb{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-02JlUTvn6x8lv0cb .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-02JlUTvn6x8lv0cb .error-icon{fill:#552222;}#mermaid-svg-02JlUTvn6x8lv0cb .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-02JlUTvn6x8lv0cb .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-02JlUTvn6x8lv0cb .marker{fill:#333333;stroke:#333333;}#mermaid-svg-02JlUTvn6x8lv0cb .marker.cross{stroke:#333333;}#mermaid-svg-02JlUTvn6x8lv0cb svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-02JlUTvn6x8lv0cb p{margin:0;}#mermaid-svg-02JlUTvn6x8lv0cb .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-02JlUTvn6x8lv0cb .cluster-label text{fill:#333;}#mermaid-svg-02JlUTvn6x8lv0cb .cluster-label span{color:#333;}#mermaid-svg-02JlUTvn6x8lv0cb .cluster-label span p{background-color:transparent;}#mermaid-svg-02JlUTvn6x8lv0cb .label text,#mermaid-svg-02JlUTvn6x8lv0cb span{fill:#333;color:#333;}#mermaid-svg-02JlUTvn6x8lv0cb .node rect,#mermaid-svg-02JlUTvn6x8lv0cb .node circle,#mermaid-svg-02JlUTvn6x8lv0cb .node ellipse,#mermaid-svg-02JlUTvn6x8lv0cb .node polygon,#mermaid-svg-02JlUTvn6x8lv0cb .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-02JlUTvn6x8lv0cb .rough-node .label text,#mermaid-svg-02JlUTvn6x8lv0cb .node .label text,#mermaid-svg-02JlUTvn6x8lv0cb .image-shape .label,#mermaid-svg-02JlUTvn6x8lv0cb .icon-shape .label{text-anchor:middle;}#mermaid-svg-02JlUTvn6x8lv0cb .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-02JlUTvn6x8lv0cb .rough-node .label,#mermaid-svg-02JlUTvn6x8lv0cb .node .label,#mermaid-svg-02JlUTvn6x8lv0cb .image-shape .label,#mermaid-svg-02JlUTvn6x8lv0cb .icon-shape .label{text-align:center;}#mermaid-svg-02JlUTvn6x8lv0cb .node.clickable{cursor:pointer;}#mermaid-svg-02JlUTvn6x8lv0cb .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-02JlUTvn6x8lv0cb .arrowheadPath{fill:#333333;}#mermaid-svg-02JlUTvn6x8lv0cb .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-02JlUTvn6x8lv0cb .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-02JlUTvn6x8lv0cb .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-02JlUTvn6x8lv0cb .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-02JlUTvn6x8lv0cb .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-02JlUTvn6x8lv0cb .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-02JlUTvn6x8lv0cb .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-02JlUTvn6x8lv0cb .cluster text{fill:#333;}#mermaid-svg-02JlUTvn6x8lv0cb .cluster span{color:#333;}#mermaid-svg-02JlUTvn6x8lv0cb div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-02JlUTvn6x8lv0cb .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-02JlUTvn6x8lv0cb rect.text{fill:none;stroke-width:0;}#mermaid-svg-02JlUTvn6x8lv0cb .icon-shape,#mermaid-svg-02JlUTvn6x8lv0cb .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-02JlUTvn6x8lv0cb .icon-shape p,#mermaid-svg-02JlUTvn6x8lv0cb .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-02JlUTvn6x8lv0cb .icon-shape .label rect,#mermaid-svg-02JlUTvn6x8lv0cb .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-02JlUTvn6x8lv0cb .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-02JlUTvn6x8lv0cb .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-02JlUTvn6x8lv0cb :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

否

是

是

否

不认得

认得

不像

像

一次接口调用

参数形状合法吗

第一层：本地 ValueError，0 请求

examples 是空的吗

第二层：ValueError，0 请求

服务端认得这个题库吗

第三层：NotFoundError，写入未发生

内容像一道题吗

第四层：写入成功，判分时静默失效

写入成功

唯一的防线：导入前的本地校验

这张图的结论只有一句话：第一到第三层由 SDK 兜住，第四层没有任何东西兜，只能自己在发出请求之前查。下一节就做这件事。

9.5 顺带一条与观测面的对照

这张表不是重复前文的结论，而是要说明同一套 SDK 里两种失败风格是并存的。一个把观测和题库都用了的项目，排查手册必须分开写两条路径：现象是"没有轨迹"，走日志那条；现象是"建库失败"，走异常那条。把两者混成一套"检查配置"的清单，会让人在错误的位置浪费时间。

十、把导入做成一个工具：校验、中止、回读

前面九节的实验暴露出来的问题，最后都会收敛成一个动作：在导入之前把能查的全查一遍。这一节给出两个可运行的脚本，一个是纯检查，一个是把检查接进导入流程。

10.1 五道入库前断言

这五条断言不是凭空列的，每一条都对应本文实测过的一个失效模式，对应关系写在脚本的 docstring 里。

"""validate_cases.py —— 入库前的五道断言：把 CSV 的坑挡在平台之外。

为什么要有这一步：平台不会替你检查题目质量（第九节实测：畸形题目一条都不拦），
而导入之后的题要评分、要对比、要追溯，脏题面的代价是一整份成绩单不可信。
所以校验放在读 CSV 之后、发请求之前——本地就能失败，一个请求都不该发出去。

五道断言对应前文五个实测事实：
1. 空表不通过（create_examples 对空列表直接抛 ValueError，脚本要先判空）
2. 题面不重复（去重靠题面，重复题面会让幂等误判）
3. 题面与参考值都不能有前后空白（空白不参与 CSV 清理，会让等值判断失效）
4. 参考值不能为空（评分要按固定键取参考值，取不到就整条判不了）
5. 参考值的每一种写法都得能用（用 | 存多写法，拆开检查非空；写法彼此相同只提示不阻断）

用法：python validate_cases.py <csv 路径> [更多 csv 路径…]
"""
import csv
import sys

QUESTION_FIELD = "question" # 题面字段名，与 create_dataset.py 保持一致
ANSWER_FIELD = "answer" # 参考值字段名，与评分时的 reference_outputs 键一致
WRITING_SEPARATOR = "|" # 一个格子里存多种答案写法的分隔符

def validate(path):
"""返回 (必须修的问题, 只提醒的问题)：前者非空就不许入库。"""
problems = []
notices = []
with open(path, encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))

if not rows:
problems.append("题量为 0：这种表不要发往平台，先确认文件是不是空的")
return problems, notices

seen = {}
for index, row in enumerate(rows, start=1):
question = row.get(QUESTION_FIELD)
answer = row.get(ANSWER_FIELD)

if question is None or answer is None:
problems.append(f"第{index}行：缺 {QUESTION_FIELD} 或 {ANSWER_FIELD} 列")
continue

if question != question.strip():
problems.append(f"第{index}行：题面有前后空白 {question!r}")
if answer != answer.strip():
problems.append(f"第{index}行：参考值有前后空白 {answer!r}")

if not answer.strip():
problems.append(f"第{index}行：参考值为空，评分时取不到判分依据")

writings = answer.strip().split(WRITING_SEPARATOR)
if any(not writing.strip() for writing in writings):
problems.append(f"第{index}行：有一种写法是空的 {answer!r}")
if len(writings) > 1 and len(set(writings)) == 1:
# 小数的纯数字写法与千分位写法本来就相同，不阻断入库，只提醒
notices.append(f"第{index}行：几种写法相同，多写法在这行没有意义 {answer!r}")

if question.strip() in seen:
problems.append(f"第{index}行：题面与第{seen[question.strip()]}行重复")
else:
seen[question.strip()] = index

return problems, notices

def main():
if len(sys.argv) < 2:
print("用法：python validate_cases.py <csv 路径> [更多路径…]")
return
for path in sys.argv[1:]:
problems, notices = validate(path)
with open(path, encoding="utf-8-sig", newline="") as handle:
total = len(list(csv.DictReader(handle)))
verdict = "通过，可以入库" if not problems else "不通过，先修完再导入"
print(f"检查 {path}（{total} 行）: {verdict}")
for problem in problems:
print(" 问题 -", problem)
for notice in notices:
print(" 提示 -", notice)

if __name__ == "__main__":
main()

用它检查三份表格：一份是本文的正规题库，一份是故意写坏的，一份是只有表头的空表。

故意写坏的那一份长这样，四个问题一行一个（第 2 行的题面末尾多了一个空格，肉眼看不出来）：

question,answer,case_type
12 乘以 34 等于多少,408|408,small
12 乘以 34 等于多少 ,408|408,small
帮我算一下 987654321 乘以 123456789,121932631112635269|,edge
这题没有参考值,,big

至于只有表头的空表，就是把这个文件的第 2 到第 5 行删掉、只留第一行。

检查 math_agent_eval_cases.csv（5 行）: 通过，可以入库
提示 – 第1行：几种写法相同，多写法在这行没有意义 '408|408'
检查 bad_cases.csv（4 行）: 不通过，先修完再导入
问题 – 第2行：题面有前后空白 '12 乘以 34 等于多少 '
问题 – 第2行：题面与第1行重复
问题 – 第3行：有一种写法是空的 '121932631112635269|'
问题 – 第4行：参考值为空，评分时取不到判分依据
问题 – 第4行：有一种写法是空的 ''
提示 – 第1行：几种写法相同，多写法在这行没有意义 '408|408'
提示 – 第2行：几种写法相同，多写法在这行没有意义 '408|408'
检查 empty_cases.csv（0 行）: 不通过，先修完再导入
问题 – 题量为 0：这种表不要发往平台，先确认文件是不是空的

坏表格那份输出里五条问题各有出处：题面尾部空格来自 3.4 的坑四、题面重复会打爆 6.5 的去重键、…| 这种尾巴是 3.2 那个"两种写法"约定的破损形式、参考值为空对应 9.4 的第 1 条与第 3 条。空表那份输出只有一行，但它保护的正是 9.2 那个 ValueError：空表在本地就被拦下，压根不会走到 create_examples。

正规题库那份"通过"，但带一条提示：第 1 行两种写法相同（408|408）。它不阻断入库，理由在 3.2 讲过——小于 1000 的数本来就只有一种写法。把这类情形列成提示而不是错误，是为了让校验脚本的退出码保持可信：红色的一律是真问题。这是工具设计里一个很小的决定，但它决定了这个脚本会不会在一两周内被团队忽略掉。

10.2 串成一个命令行工具

"""import_cases.py —— 把导入做成一个能反复跑、能挡住脏题的命令行工具。

用法：python import_cases.py <csv 路径> <题库名> [描述]

与 create_dataset.py 的差别只有三处，但每一处都对应一个实测事实：
1. 导入前先跑 validate_cases.validate()，有问题就中止，一个请求都不发；
平台不校验题目质量，畸形题目写进去就存在。
2. 写入前判空：create_examples(examples=[]) 会抛 ValueError，
而「差集为空」是幂等复跑时的正常状态，不该以异常收场。
3. 写完立刻读回来核对条数，count 只代表服务端接了几条，不代表内容对。
"""
import csv
import sys

from dotenv import load_dotenv
from langsmith import Client

from validate_cases import validate

CSV_PATH = "math_agent_eval_cases.csv"
DATASET_NAME = "math_agent_eval_cases"
DESCRIPTION = "计算器 Agent 题库：输入一句乘法，参考值是两个大数乘积"

def load_cases(path):
"""读 CSV。utf-8-sig 用来吃掉 Excel 导出时的 BOM。"""
with open(path, encoding="utf-8-sig", newline="") as handle:
return list(csv.DictReader(handle))

def to_example(row):
"""一行 CSV 转一道题：题面进 inputs，参考值进 outputs，其余进 metadata。"""
return {
"inputs": {"question": row["question"]},
"outputs": {"answer": row["answer"]},
"metadata": {"case_type": row["case_type"], "source": "csv"},
}

def get_or_create_dataset(client, dataset_name, description):
"""同名就复用，不存在才建——SDK 没有这个接口，得自己写。"""
for dataset in client.list_datasets(dataset_name=dataset_name):
print(f"已存在同名题库，直接复用：{dataset.name}")
return dataset
dataset = client.create_dataset(dataset_name, description=description)
print(f"新建题库：{dataset.name} id={dataset.id}")
return dataset

def main(csv_path, dataset_name, description):
load_dotenv()

problems, notices = validate(csv_path) # 第一道闸：本地校验
for notice in notices:
print("提示 -", notice)
if problems:
for problem in problems:
print("问题 -", problem)
raise SystemExit(f"题库有 {len(problems)} 处问题，已中止导入")

client = Client()
rows = load_cases(csv_path)
print(f"校验通过，CSV 读到 {len(rows)} 道题")

dataset = get_or_create_dataset(client, dataset_name, description)

existing = {example.inputs.get("question")
for example in client.list_examples(dataset_id=dataset.id)}
todo = [to_example(row) for row in rows if row["question"] not in existing]
print(f"其中 {len(rows) – len(todo)} 道已在平台上，本次只补 {len(todo)} 道")

if not todo: # 第二道闸：空就不调接口
print("题库已是最新，无需写入")
return

response = client.create_examples(dataset_id=dataset.id, examples=todo)

read_back = list(client.list_examples(dataset_id=dataset.id)) # 第三道闸：回读核对
expected = len(existing) + response["count"]
print(f"写入返回 count={response['count']}，读回共 {len(read_back)} 题，"
f"与预期 {'一致' if len(read_back) == expected else f'不一致（预期 {expected}）'}")

if __name__ == "__main__":
args = sys.argv[1:]
main(args[0] if args else CSV_PATH,
args[1] if len(args) > 1 else DATASET_NAME,
args[2] if len(args) > 2 else DESCRIPTION)

与第五节那份脚本相比，它多了三道闸，位置分别是：校验（所有网络动作之前）、判空（写入之前）、回读（写入之后）。用法上把 CSV 路径和题库名改成命令行参数，因为同一个脚本迟早要导入第二份、第三份题库——参数化的代价极低，收益是这份代码不用再复制。

跑一次证明闸门真的能挡住请求。这里的关键不是"它打印了中止"，而是服务端确认自己没收到任何写入：

"""probe_guarded_import.py —— 证明「校验不过时一个请求都没发出去」。

只看脚本打印的措辞不算证据：把端点指向本地假服务端，再数服务端收到了什么。
子进程与父进程不共享内存，所以这里用真实端口起服务端，子进程写题时
父进程能直接查到自己被打了几个请求。
"""
import os
import subprocess
import sys

sys.path.insert(0, ".")
import dataset_server as srv

PORT = 8410
srv.configure(PORT)
srv.start(PORT)
srv.reset()

for label, csv_path, dataset_name in (("脏题库", "bad_cases.csv", "guarded_bad"),
("干净题库", "math_agent_eval_cases.csv", "guarded_good")):
srv.reset()
env = dict(os.environ, LANGSMITH_ENDPOINT=f"http://127.0.0.1:{PORT}",
LANGSMITH_API_KEY="lsv2_localdev0000000000000000000000000000",
LANGSMITH_TRACING="false")
result = subprocess.run([sys.executable, "import_cases.py", csv_path, dataset_name],
capture_output=True, text=True, env=env)
posts = [path for method, path, _q, _b in srv.REQUESTS if method == "POST"]
print(f"== {label}（{csv_path}）==")
print(result.stdout.rstrip())
print(f"服务端收到的 POST 请求: {len(posts)} 个 {posts}")
print(f"退出码: {result.returncode}")
print()