Opik Python SDK Automation Rule Evaluators REST 客户端详解：以编程方式管理自动化评估规则-平芜编程栈

Opik Python SDK Automation Rule Evaluators REST 客户端详解：以编程方式管理自动化评估规则

【免费下载链接】comet-llmDebug, evaluate, and monitor your LLM applications, RAG systems, and agentic workflows with comprehensive tracing, automated evaluations, and production-ready dashboards.项目地址: https://gitcode.com/GitHub_Trending/co/comet-llm

本文基于 Opik 仓库中 Automation Rule Evaluators 文档页与当前 Python SDK 的生成客户端源码展开，讲清楚AutomationRuleEvaluatorsClient提供哪些方法、每个方法对应的 HTTP 端点与参数、请求/响应数据模型（判别联合类型、sampling_rate、trigger_scope等关键字段）以及同步/异步、原始响应三种使用形态，帮助你在脚本或自动化流水线中批量创建、查询、更新、删除 Opik 平台的在线评估规则（Automation Rule Evaluators），并审计其执行日志。

一、背景：Automation Rule Evaluators 是什么

Automation Rule Evaluators 是 Opik 平台的"在线评估规则"：为项目（project）配置一条规则后，平台会对符合条件的 trace / span / thread 数据自动运行评估器（LLM-as-judge 或用户自定义 Python 指标），把分数写回平台，实现无需离线实验的持续质量监控。

从当前 SDK 的类型定义看，一个规则由两部分构成：

公共字段（Base）：name（必填）、project_id（旧版单项目字段）、project_ids（多项目支持的新字段）、sampling_rate、enabled、trigger_scope，见 automation_rule_evaluator_write.py；
判别联合（discriminated union）：按type字段区分的 6 种评估器类型，每类带有自己的filters和code结构。

6 种类型及其语义（源自 automation_rule_evaluator_write.py）：

`type`取值	评估对象	评估器种类
`llm_as_judge`	trace	LLM 裁判（提示词 + 模型打分）
`user_defined_metric_python`	trace	用户自定义 Python 指标代码
`trace_thread_llm_as_judge`	thread（线程/会话）	LLM 裁判
`trace_thread_user_defined_metric_python`	thread	用户自定义 Python 指标代码
`span_llm_as_judge`	span	LLM 裁判
`span_user_defined_metric_python`	span	用户自定义 Python 指标代码

公共字段中几个容易踩坑的语义，直接引自源码 docstring：

sampling_rate：0 到 1 之间的浮点数，表示该规则对生产（SDK 上报）数据的采样比例；trace 规则对 experiment / playground / optimization 产生的 trace忽略该值并全量评分，而 span 与 thread 规则只评估 SDK 上报的数据（automation_rule_evaluator_write.py#L34-L37）。
trigger_scope：取值为"production"、"experiment"、"both"三者的联合类型（automation_rule_evaluator_write_trigger_scope.py），控制规则在生产 trace、实验 trace 还是两者上触发，省略时默认为production。
project_id被标注为 "Primary project ID (legacy field for backwards compatibility)"，多项目场景应使用project_ids。

二、如何拿到客户端

文档页给出的入口是：

import opik client = opik.Opik() # 该属性返回 OpikApi（底层 REST 客户端） evaluators_client = client.rest_client.automation_rule_evaluators

源码可以印证这条链路：

Opik类的rest_client属性返回一个OpikApi实例（见 opik_client.py#L169-L180），Opik类定义于 opik_client.py#L108；
OpikApi在初始化时挂载各资源子客户端，其中第 123 行即为self.automation_rule_evaluators = AutomationRuleEvaluatorsClient(client_wrapper=self._client_wrapper)（见 rest_api/client.py#L123）。

因此client.rest_client.automation_rule_evaluators拿到的是 AutomationRuleEvaluatorsClient。需要说明：该客户端文件头部标注This file was auto-generated by Fern from our API Definition，即由 OpenAPI 定义经 Fern 生成，仓库中生成配置位于 sdks/code_generation/fern。

三、方法总览与对应 HTTP 端点

AutomationRuleEvaluatorsClient共提供 6 个业务方法。端点信息来自 raw_client.py（该文件同时包含同步RawAutomationRuleEvaluatorsClient与异步AsyncRawAutomationRuleEvaluatorsClient两套实现）：

方法	HTTP 方法与路径	返回类型
`find_evaluators`	`GET v1/private/automations/evaluators`	`AutomationRuleEvaluatorPagePublic`
`create_automation_rule_evaluator`	`POST v1/private/automations/evaluators`	`None`
`delete_automation_rule_evaluator_batch`	`POST v1/private/automations/evaluators/delete`	`None`
`get_evaluator_by_id`	`GET v1/private/automations/evaluators/{id}`	`AutomationRuleEvaluatorPublic`
`update_automation_rule_evaluator`	`PATCH v1/private/automations/evaluators/{id}`	`None`
`get_evaluator_logs_by_id`	`GET v1/private/automations/evaluators/{id}/logs`	`LogPage`

所有方法均接受可选的request_options: RequestOptions参数（请求级配置，如超时、重试等），这是 Fern 生成客户端的通用约定。

四、逐方法详解

4.1 列表查询：`find_evaluators`

签名（client.py#L33-L44）：

def find_evaluators( self, *, project_id: typing.Optional[str] = None, id: typing.Optional[str] = None, name: typing.Optional[str] = None, filters: typing.Optional[str] = None, sorting: typing.Optional[str] = None, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> AutomationRuleEvaluatorPagePublic:

除request_options外全部为关键字参数且可选；project_id、id、name为直查条件，filters与sorting为字符串化的过滤/排序表达式，page/size控制分页；
底层以 query 参数方式发送，见 raw_client.py#L66-L79。

响应模型 AutomationRuleEvaluatorPagePublic 包含：page、size、total（总数，便于翻页判断）、content（当前页规则对象列表）、sortable_by（可排序字段提示）。

4.2 创建规则：`create_automation_rule_evaluator`

def create_automation_rule_evaluator( self, *, request: AutomationRuleEvaluatorWrite, request_options: typing.Optional[RequestOptions] = None ) -> None:

注意当前实现中创建方法只接受一个request参数，其类型是 6 种评估器组成的判别联合（automation_rule_evaluator_write.py#L147-L154）。以 trace 级 LLM 裁判为例：

import opik from opik.rest_api import ( AutomationRuleEvaluatorWrite_LlmAsJudge, LlmAsJudgeCodeWrite, ) client = opik.Opik() client.rest_client.automation_rule_evaluators.create_automation_rule_evaluator( request=AutomationRuleEvaluatorWrite_LlmAsJudge( project_id="my-project-id", # 或使用 project_ids=[...] 绑定多个项目 name="production-hallucination-check", sampling_rate=0.5, # 生产数据 50% 采样；0~1 enabled=True, trigger_scope="production", # 或 "experiment" / "both"，缺省为 "production" # filters=[...], # TraceFilterWrite 列表，限定规则命中的 trace 子集 # code=LlmAsJudgeCodeWrite(...), # LLM 裁判的提示词/模型配置 ) )

request会被序列化为 JSON 后POST，序列化走convert_and_respect_annotation_metadata(object_=request, annotation=AutomationRuleEvaluatorWrite, direction="write")（raw_client.py#L112-L123）。

4.3 按 ID 查询：`get_evaluator_by_id`

def get_evaluator_by_id( self, id: str, *, project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None ) -> AutomationRuleEvaluatorPublic:

id是位置必填参数，拼入路径v1/private/automations/evaluators/{id}（raw_client.py#L202-L208）。

响应模型 AutomationRuleEvaluatorPublic 在写入模型公共字段的基础上，额外返回审计与项目绑定信息：

id、name、sampling_rate、enabled、trigger_scope、action；
projects：ProjectReferencePublic列表，注释说明其"唯一、按项目名字母序排序"；project_id/project_name为兼容旧版的单项目字段；
审计字段：created_at、created_by、last_updated_at、last_updated_by。

4.4 更新规则：`update_automation_rule_evaluator`

def update_automation_rule_evaluator( self, id: str, *, request: AutomationRuleEvaluatorUpdate, request_options: typing.Optional[RequestOptions] = None ) -> None:

请求体为AutomationRuleEvaluatorUpdate，结构与 Write 版本对应（6 种联合成员一致，见 automation_rule_evaluator_update.py），公共字段为name、sampling_rate、enabled、trigger_scope、project_id、project_ids、action。底层走PATCH v1/private/automations/evaluators/{id}（raw_client.py#L248-L259）。典型用法是只调整采样率或启停规则：

from opik.rest_api import AutomationRuleEvaluatorUpdate_UserDefinedMetricPython client.rest_client.automation_rule_evaluators.update_automation_rule_evaluator( id="evaluator-id", request=AutomationRuleEvaluatorUpdate_UserDefinedMetricPython( name="my-evaluator", sampling_rate=0.1, enabled=False, ), )

4.5 批量删除：`delete_automation_rule_evaluator_batch`

def delete_automation_rule_evaluator_batch( self, *, ids: typing.Sequence[str], project_id: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> None:

ids为字符串序列，放入请求体{"ids": [...]}；project_id作为 query 参数（raw_client.py#L155-L169）。

4.6 查询执行日志：`get_evaluator_logs_by_id`

def get_evaluator_logs_by_id( self, id: str, *, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None ) -> LogPage:

访问v1/private/automations/evaluators/{id}/logs，用size控制单次拉取的日志条数。返回 LogPage：content（LogItem列表）、page、size、total。这是排查"规则到底跑没跑、跑了什么、失败在哪"的主要入口。

五、文档页原始示例的对照说明

文档页给出的 Usage Example 如下（完整继承自 automation_rule_evaluators.rst 的用法章节）：

import opik client = opik.Opik() # List automation rule evaluators evaluators = client.rest_client.automation_rule_evaluators.find_automation_rule_evaluators( page=0, size=10 ) # Get an evaluator by ID evaluator = client.rest_client.automation_rule_evaluators.get_automation_rule_evaluator_by_id( "evaluator-id" ) # Create a new evaluator client.rest_client.automation_rule_evaluators.create_automation_rule_evaluator( name="my-evaluator", project_id="project-id", code="def evaluate(trace): return {'score': 0.8}" )

对照当前仓库源码，示例体现的"列表 / 按 ID 获取 / 创建"三段式用法与现有实现一一对应，但方法名与入参形态已演进，落地时应以源码为准：

文档示例写法	当前源码对应实现
`find_automation_rule_evaluators(page=0, size=10)`	`find_evaluators(page=0, size=10)`，另支持`project_id`/`id`/`name`/`filters`/`sorting`
`get_automation_rule_evaluator_by_id("evaluator-id")`	`get_evaluator_by_id("evaluator-id", project_id=...)`
`create_automation_rule_evaluator(name=..., project_id=..., code=...)`	`create_automation_rule_evaluator(request=...)`，`request`为带`type`判别字段的 Pydantic 联合对象

也就是说，文档示例中"把code作为裸字符串直接传入"的简化写法，在当前类型系统里被显式的LlmAsJudgeCodeWrite/UserDefinedMetricPythonCodeWrite等code结构体取代，创建时必须先选定 6 种type之一。这属于 API 文档快照与生成代码之间的版本差异，建议以 automation_rule_evaluators/client.py 的实际签名为准。

六、异步与原始响应两种形态

异步客户端：AsyncAutomationRuleEvaluatorsClient 提供与同步版本完全同构的 6 个async def方法，挂载在AsyncOpikApi.automation_rule_evaluators（rest_api/client.py#L278），适合在高并发采集/管理场景中await调用。
原始响应：AutomationRuleEvaluatorsClient上的with_raw_response属性（client.py#L22-L31）返回RawAutomationRuleEvaluatorsClient，其方法直接返回HttpResponse[T]包装（含status_code、headers与解析后的data）。文档页的 autoclass 指令用:exclude-members: with_raw_response在渲染时排除了它，但运行时代码中该属性依然存在，可用于需要读取响应头（如分页元信息、限流提示）的场景。

七、实现细节：错误处理与序列化

阅读 raw_client.py 可以确认生成客户端的统一处理模式：

每个方法先通过self._client_wrapper.httpx_client.request(...)发起 httpx 请求，URL 中的id经jsonable_encoder转义；
2xx响应：用parse_obj_as(type_=..., object_=_response.json())反序列化为对应 Pydantic 模型（如AutomationRuleEvaluatorPagePublic、LogPage）后返回；
非 2xx 或响应体 JSON 解析失败：抛出ApiError(status_code=..., headers=..., body=...)。因此调用侧只需针对ApiError捕获即可得到状态码与错误体；
写入请求体经convert_and_respect_annotation_metadata(..., direction="write")序列化，配合模块级哨兵OMIT处理可省略字段。

八、进一步阅读

客户端实现：同步客户端、原始客户端与异步客户端
数据模型：Write 模型、Public 模型、Update 模型、分页模型、日志分页模型、触发范围
客户端装配：OpikApi 各子客户端挂载、Opik.rest_client 属性
文档源文件：automation_rule_evaluators.rst

适用前提：以上方法签名、字段语义与端点路径均以当前仓库中sdks/python下的生成代码为准；由于该客户端由 OpenAPI 定义自动生成，若后端 API 演进（字段增删、方法重命名），需以最新生成的代码为准。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Opik Python SDK Automation Rule Evaluators REST 客户端详解：以编程方式管理自动化评估规则