企业官网建设流程全解析

TensorZero 数据集与数据点编程指南：用 Python 客户端与 curl 实现 Datasets & Datapoints 的创建、读取与删除

【免费下载链接】tensorzeroTensorZero is an open-source LLMOps platform that unifies an LLM gateway, observability, evaluation, optimization, and experimentation.项目地址: https://gitcode.com/GitHub_Trending/te/tensorzero

本文以 TensorZero 仓库中的官方示例 examples/guides/datasets-datapoints 为主线，系统讲解如何以编程方式管理数据集（Dataset）与数据点（Datapoint）。你将学会两种完全可运行的实操路径：基于tensorzeroPython 客户端的方法调用，以及基于 curl 的纯 HTTP 调用，并深入理解 Gateway 层的数据校验、软删除与分页等底层实现，从而把数据集操作无缝接入自己的评估与优化工作流。

一、数据集与数据点：TensorZero 数据资产的基石

在 TensorZero 中，数据集（Dataset）是数据点的命名集合，而每个数据点（Datapoint）归属于一个具体的函数（function），其字段随函数类型（chat或json）而异。从结构上看，数据点与一次推理（inference）高度镜像：都包含input（输入）、可选的output（输出）以及其他元数据（如tags标签）。

数据集是评估（evaluation）与优化（optimization，如 GEPA、SFT、DICL）等上层工作流的数据来源。你可以通过 TensorZero UI 手动管理，也可以通过 TensorZero Gateway 提供的 HTTP 接口以编程方式管理（详见仓库文档 docs/gateway/api-reference/datasets-datapoints.mdx）。本文聚焦后者，给出可直接复制运行的完整示例。

二、示例项目结构与环境准备

示例位于 examples/guides/datasets-datapoints，目录结构如下：

examples/guides/datasets-datapoints/ ├── config/ │ ├── functions/ │ │ └── extract_recipient/ │ │ └── output_schema.json # json 函数的输出 JSON Schema │ └── tensorzero.toml # Gateway 配置 ├── README.md # 官方指南 ├── docker-compose.yml # 一键启动 Postgres + Gateway + UI ├── main.py # Python 客户端示例 ├── main.sh # curl HTTP 示例 ├── pyproject.toml # Python 依赖（tensorzero>=2025.5.7） └── uv.lock

2.1 前置准备

按官方指南 README.md 的步骤操作：

1. 安装 Docker；
2. 在 OpenAI 平台生成 API Key（OPENAI_API_KEY）；
3. 将OPENAI_API_KEY设为环境变量；
4. 在示例目录下启动 TensorZero 与 Postgres：

docker compose up

2.2 一键启动的 compose 拓扑

docker-compose.yml 编排了 4 个服务，构成一个完整的本地开发环境：

其中 Gateway 服务通过环境变量注入数据库连接与模型凭据：

environment: TENSORZERO_POSTGRES_URL: postgres://postgres:postgres@postgres:5432/tensorzero OPENAI_API_KEY: ${OPENAI_API_KEY:?Environment variable OPENAI_API_KEY must be set.}

OPENAI_API_KEY使用了${VAR:?}语法强制校验，未设置时 compose 会直接报错，避免带着空凭据启动。

2.3 示例函数配置

config/tensorzero.toml 定义了本次示例用到的两个函数：

[functions.extract_recipient] type = "json" output_schema = "functions/extract_recipient/output_schema.json" [functions.extract_recipient.variants.baseline] type = "chat_completion" model = "openai::gpt-4o-mini" json_mode = "strict" [functions.draft_email] type = "chat" [functions.draft_email.variants.baseline] type = "chat_completion" model = "openai::gpt-4o-mini"

• extract_recipient：json类型函数，输出受 output_schema.json 约束（name与email两个必填字符串字段，且additionalProperties: false），variant 开启json_mode = "strict"；
• draft_email：chat类型函数，无结构化输出约束。

这两个函数恰好对应数据点的两种类型：json数据点与chat数据点，示例将围绕它们构造数据。

三、Python 客户端：三种构造方式与四个核心操作

Python 示例 main.py 演示了"插入 → 按 ID 读取 → 删除 → 列表查询 → 清理"的完整生命周期。下面按步骤拆解。

3.1 三种数据点构造方式

方式一：直接使用字典（dict），结构天然与 HTTP JSON 一致：

extract_recipient_datapoint = { "function_name": "extract_recipient", "input": { "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please send this to Alice at alice@example.com", } ], } ] }, }

方式二：使用JsonDatapointInsert类型，可附加output与name等元数据：

extract_recipient_datapoint_with_output = JsonDatapointInsert( function_name="extract_recipient", input={ "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please send this to Bob at bob@example.com", } ], } ] }, output={ "name": "Bob", "email": "bob@example.com", }, name="bob_recipient_example", )

方式三：使用ChatDatapointInsert类型，可附加tags标签：

draft_email_datapoint = ChatDatapointInsert( function_name="draft_email", input={ "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please draft an email to Bob at bob@example.com", } ], } ] }, tags={ "customer_id": "123", }, )

兼容性说明：本示例基于早期发布版本的 Python 客户端编写，构造参数名与当前仓库中的客户端签名略有差异——当前 tensorzero.pyi 中create_datapoints的参数名为requests、delete_datapoints支持批量删除（详见第 3.3 节）。示例代码本身与仓库锁定的客户端版本（pyproject.toml中tensorzero>=2025.5.7且exclude-newer3 天）保持一致，可直接运行；底层 HTTP 契约不受影响。

3.2 核心操作：插入、读取、删除、列表

在上下文管理器with TensorZeroGateway.build_http(gateway_url="http://localhost:3000") as t0:中，所有调用都指向本地 Gateway（3000端口）：

with TensorZeroGateway.build_http( gateway_url="http://localhost:3000", ) as t0: # 1. 批量插入：一次创建 3 个数据点 create_datapoints_response = t0.create_datapoints( dataset_name="email_application", datapoints=[ extract_recipient_datapoint, extract_recipient_datapoint_with_output, draft_email_datapoint, ], ) print(create_datapoints_response) # 2. 按 ID 读取单个数据点（返回列表中的第一个 ID） get_datapoints_response = t0.get_datapoints( dataset_name="email_application", ids=[create_datapoints_response[0]], ) print(get_datapoints_response) # 3. 删除单个数据点 t0.delete_datapoint( dataset_name="email_application", datapoint_id=create_datapoints_response[0], ) # 4. 列表查询（查看剩余数据点） list_datapoints_response = t0.list_datapoints(dataset_name="email_application") print(list_datapoints_response) # 5. 清理：删除剩余数据点 for datapoint in list_datapoints_response: t0.delete_datapoint( dataset_name="email_application", datapoint_id=datapoint.id, )

执行方式（需已安装 uv）：

uv run main.py

关键语义：

• create_datapoints返回ids（UUIDv7 列表），示例用create_datapoints_response[0]拿到第一个新数据点的 ID；
• get_datapoints返回完整数据点对象（含input、output、tags等字段）；
• delete_datapoint为软删除，返回值为空（示例打印N/A）；
• list_datapoints返回数据点对象列表，可用datapoint.id逐一清理；
• 若目标数据集不存在，插入时会自动按给定名称创建数据集。

3.3 当前仓库中客户端方法的签名

仓库内 tensorzero.pyi 给出了 Python 客户端当前的完整方法签名（第 707～819 行），可作为升级到新版客户端后的对照：

• list_datapoints(*, dataset_name: str, request: ListDatapointsRequest) -> GetDatapointsResponse：支持分页与过滤参数；
• create_datapoints(*, dataset_name: str, requests: Sequence[CreateDatapointRequest]) -> CreateDatapointsResponse：返回新建数据点 ID 列表；
• get_datapoints(*, dataset_name: str | None = ..., ids: Sequence[str]) -> GetDatapointsResponse：按 ID 读取。注释特别说明：传入dataset_name能提升查询性能，因为数据集是排序键（sorting key）的一部分；
• delete_datapoints(*, dataset_name: str, ids: Sequence[str]) -> DeleteDatapointsResponse：批量删除；
• 另有update_datapoints、update_datapoints_metadata、delete_dataset、create_datapoints_from_inferences等进阶方法（见第六节）。

四、curl 纯 HTTP 方式：四个端点的完整调用链

不依赖任何 SDK 时，可直接用 curl 操作。bash 脚本 main.sh 演示了完整流程：先把 3 个数据点写入临时 JSON 文件，再依次调用插入、按 ID 读取、删除、列表、清理共 5 组请求。

4.1 请求体：三种数据点的 JSON 形态

脚本用 heredoc 构造请求体并写入mktemp创建的临时文件，通过"type"字段区分数据点类型：

{ "datapoints": [ { "type": "json", "function_name": "extract_recipient", "input": { "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please send this to Alice at alice@example.com" } ] } ] } }, { "type": "json", "function_name": "extract_recipient", "input": { "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please send this to Bob at bob@example.com" } ] } ] }, "output": { "name": "Bob", "email": "bob@example.com" } }, { "type": "chat", "function_name": "draft_email", "input": { "messages": [ { "role": "user", "content": [ { "type": "text", "text": "Please draft an email to Bob at bob@example.com" } ] } ] }, "tags": { "customer_id": "123" } } ] }

4.2 四个核心端点的调用

① 批量插入（POST /v1/datasets/{dataset_name}/datapoints）：

curl -s -X POST \ "http://localhost:3000/v1/datasets/email_application/datapoints" \ -H "Content-Type: application/json" \ -d @"$TEMP_FILE"

响应为{"ids": ["<UUIDv7>", ...]}。脚本用jq -e '.ids | type == "array"'校验响应格式，并用jq -r '.ids[0]'提取第一个 ID 供后续步骤使用，校验失败则报错退出——这是生产脚本中值得借鉴的健壮性写法。

② 按 ID 读取（POST /v1/datasets/{dataset_name}/get_datapoints）：

curl -s -X POST \ "http://localhost:3000/v1/datasets/email_application/get_datapoints" \ -H "Content-Type: application/json" \ -d "{\"ids\": [\"${FIRST_DATAPOINT_ID}\"]}"

③ 删除（DELETE /v1/datasets/{dataset_name}/datapoints）：

curl -s -X DELETE \ "http://localhost:3000/v1/datasets/email_application/datapoints" \ -H "Content-Type: application/json" \ -d "{\"ids\": [\"${FIRST_DATAPOINT_ID}\"]}"

④ 列表查询（POST /v1/datasets/{dataset_name}/list_datapoints）：

curl -s -X POST \ "http://localhost:3000/v1/datasets/email_application/list_datapoints" \ -H "Content-Type: application/json" \ -d '{}'

⑤ 清理：脚本把列表响应的[.datapoints[].id]收集为 JSON 数组，若不为空则再一次DELETE批量删除全部剩余数据点，最后删除临时文件。运行方式：

./main.sh

4.3 HTTP 端点速查表

以下端点均在 crates/gateway/src/routes/external.rs 中注册：

注：仓库还保留了一个不带数据集名的旧版读取端点POST /v1/datasets/get_datapoints，并在 get_datapoints.rs 中打印弃用警告，建议改用带数据集名的版本以获得更好的查询性能。

五、Gateway 侧实现原理：从 HTTP 到数据库

理解底层实现有助于排查问题与预估性能。以下均可在仓库源码中找到对应实现。

5.1 创建数据点：校验与并行入库

create_datapoints.rs 中的核心逻辑create_datapoints依次执行：

1. validate_dataset_name(dataset_name)：校验数据集名称合法性；
2. 空列表校验：request.datapoints.is_empty()时直接返回InvalidRequest错误（"At least one datapoint must be provided"）；
3. 将CreateDatapointRequest（Chat或Json两种变体）逐个转换为数据库插入结构，并通过futures::future::try_join_all并行完成（因为可能需要存储输入内容），任一失败则整体回滚；
4. 入库后返回CreateDatapointsResponse { ids: Vec<Uuid> }。

请求类型定义见 types.rs：CreateDatapointRequest是带type标签的枚举（chat/json）。其中json数据点支持动态传入output_schema（未提供时使用函数配置的输出 Schema，提供时会被校验），chat数据点则支持allowed_tools、tool_choice、parallel_tool_calls、episode_id等与推理请求一致的字段。

5.2 列表查询：默认分页与高级过滤

get_datapoints.rs 定义了列表接口的三个默认值：

const DEFAULT_LIMIT: u32 = 20; const DEFAULT_OFFSET: u32 = 0; const DEFAULT_ALLOW_STALE: bool = false;

对应 types.rs 中ListDatapointsRequest的可选参数：

• function_name：仅返回指定函数的数据点；
• limit/page_size：每页数量，默认 20；page_size已标记弃用（自 2025.11.1 起），请用limit；
• offset：跳过数量，默认 0；
• filter：按标签、时间以及 AND/OR/NOT 逻辑组合过滤；
• order_by：排序条件列表（如按timestamp、search_relevance）；
• search_query_experimental：实验性全文搜索——大小写不敏感的精确子串匹配，不分词、不做相关性打分，无其他过滤条件时可能全表扫描，数据量大时可能极慢，官方明确不建议在关键场景依赖。

5.3 软删除语义

删除操作（delete_datapoints与delete_dataset）执行的是软删除：数据点被标记为 stale（过期），之后列表查询、评估运行等都会忽略它们，但原始数据仍保留在数据库中。官方 API 文档（docs/gateway/api-reference/datasets-datapoints.mdx）特别提示：按 ID 直接读取（get_datapoints）时，stale 数据点仍会出现在响应中。这意味着删除后立刻按 ID 读取，数据依然可见，属于预期行为。

六、进阶操作：更新与从推理生成数据点

除了增删查列四个基础操作，同一份 API 参考还提供了几个高频进阶能力，适合在生产工作流中使用：

• 从推理创建数据点（POST /v1/datasets/{dataset_name}/from_inferences，客户端方法create_datapoints_from_inferences）：通过type字段选择inference_ids（按推理 ID 列表，output_source可指定inference/demonstration/none，默认inference）或inference_query（复用推理列表查询参数）两种模式，将历史推理沉淀为可复用数据集——这是构建持续优化闭环的关键路径；
• 更新数据点（PATCH /v1/datasets/{dataset_name}/datapoints，update_datapoints）：创建新版本——原数据点被标记为 stale（软删除），新数据点获得新 ID；省略字段保持不变，显式传null清空可空字段；
• 更新数据点元数据（PATCH /v1/datasets/{dataset_name}/datapoints/metadata，update_datapoints_metadata）：仅原地更新name等元数据，不产生新版本、不改变 ID；
• 删除整个数据集（DELETE /v1/datasets/{dataset_name}，delete_dataset）：软删除数据集下所有数据点，返回num_deleted_datapoints。

七、测试验证：仓库中的自动化佐证

仓库为这套 API 提供了完整的自动化测试，见 crates/tensorzero-python/tests/test_datapoints_v1.py，同步与异步客户端均有覆盖：

• 按 ID 读取（含传入/不传dataset_name两种形态）；
• 列表查询的分页与过滤（limit/offset组合）；
• 批量删除与整数据集删除；
• 更新元数据后按 ID 读取验证结果；
• 从推理创建数据点（create_datapoints_from_inferences）；
• 边界情况：空 ID 列表返回空、不存在的 ID 返回空列表。

在 Rust 侧，Gateway 的端到端测试同样覆盖数据点流程（如 crates/tensorzero-core/tests/e2e/endpoints/datasets/mod.rs 与 MCP 场景下的create_datapoints测试）。若你在修改或调试相关功能，这些测试是现成的行为基准。

结语

从docker compose up到uv run main.py或./main.sh，本文完整复现了 TensorZero 数据集与数据点的编程管理闭环：三种数据点构造方式、四个核心 HTTP 端点、Gateway 侧的校验与并行入库、默认分页与软删除语义，以及从推理生成数据点等进阶能力。无论你是要搭建评估基准、准备微调数据，还是构建"推理 → 沉淀数据集 → 优化"的自动闭环，这套 API 都是直接可用的基础能力。

【免费下载链接】tensorzeroTensorZero is an open-source LLMOps platform that unifies an LLM gateway, observability, evaluation, optimization, and experimentation.项目地址: https://gitcode.com/GitHub_Trending/te/tensorzero