Python 接入指南

English | 中文

Python 辅助库负责调用 agentprov、读取执行证据和提交评估信号。普通 Python 函数就能定义评估规则，适合接入测试、基准评估或 RL 流水线。

安装

需要 Python 3.9+ 和可运行的 agentprov。在仓库根目录安装辅助库：

python3 -m venv .venv
. .venv/bin/activate
python3 -m pip install -e .
export AGENTPROV_BIN=/absolute/path/to/agentprov

将 AGENTPROV_BIN 改为 CLI 的实际路径。预编译 CLI 的安装方法见快速开始。

agentprov 和 agentprov_eval 导出同一组 Python 接口；下文使用 from agentprov import ...。 JSON 字段、状态和系统生成的错误统一使用英文，已保存的证据保留原语言。

记录、评估并导入结果

下面的脚本记录一次文件写入，用自定义规则统计文件变化，再将结果导入同一次执行。示例使用临时目录，退出后自动清理。

import os
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from agentprov import Client, Registry, Signal, evaluate_context

registry = Registry(name="example-quality")

@registry.rule("changed_files")
def changed_files(ctx):
    return Signal.quality_signal(
        "changed_files", float(len(ctx.file_changes())), "File changes observed"
    )

with TemporaryDirectory(prefix="agentprov-python-") as directory:
    root = Path(directory)
    work = root / "work"
    work.mkdir()
    client = Client(binary=os.environ["AGENTPROV_BIN"], data_dir=root / "data")
    manifest = client.record(
        [sys.executable, "-c", "from pathlib import Path; Path('result.txt').write_text('ok')"],
        run_id="run-python-example", workdir=work,
    )
    context = client.eval_context(manifest["run_id"])
    report = evaluate_context(context, registry=registry, engine=registry.name)
    imported = client.import_signal_reports([report], engine=registry.name)
    print(manifest["run_id"], report["signal_count"], imported["signal_count"])
    print(client.signals(manifest["run_id"])["schema_version"])

Client.record 会实际执行传入的命令。evaluate_context 在当前 Python 进程中运行规则；调用 import_signal_reports 后，结果才写入存储。

批量处理

run_batch_pipeline 完成批量记录、读取上下文、评估、导入信号和导出取证包。每个并发任务使用独立工作目录：

import os
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from agentprov import Client, Registry, Signal

registry = Registry(name="batch-quality")

@registry.rule("changed_files")
def changed_files(ctx):
    return Signal.quality_signal(
        "changed_files", float(len(ctx.file_changes())), "File changes observed"
    )

with TemporaryDirectory(prefix="agentprov-batch-") as directory:
    root = Path(directory)
    client = Client(binary=os.environ["AGENTPROV_BIN"], data_dir=root / "data")
    jobs = []
    for index in range(2):
        work = root / str(index)
        work.mkdir()
        jobs.append({
            "run_id": f"run-python-batch-{index}", "job_id": f"job-{index}",
            "workdir": str(work),
            "command": [sys.executable, "-c", "from pathlib import Path; Path('result.txt').write_text('ok')"],
        })
    result = client.run_batch_pipeline(jobs, registry, concurrency=2)
    print(result.batch_id, len(result.run_ids), result.signal_count)

默认导入信号并导出取证包。仅需内存中的结果时，可设置 import_signals=False 和 include_forensics=False。返回的 BatchPipelineResult 包含 record_manifest、contexts、reports、import_report、forensics、summary，并提供 batch_id、run_ids、signal_count 和 to_dict()。

Client 配置

record、eval_context、signals、audit_associations 在设置 daemon_url 后直接使用 HTTP。其他方法仍需 CLI。HTTP 从 env 或进程环境读取 AGENTPROV_DAEMON_TOKEN，作为 Bearer Token 发送；拒绝重定向，避免令牌被转发到其他地址。

分页查询已保存的审计决定：

from agentprov import Client

client = Client(daemon_url="http://127.0.0.1:8574")
cursor = ""
while True:
    page = client.audit_associations("run-example", limit=100, cursor=cursor)
    for item in page["items"]:
        print(item["event_id"], item["decision"], item["operation"]["status"])
    if not page["has_more"]:
        break
    cursor = page["next_cursor"]

event_id 可筛选单条决定。未设置 daemon_url 时使用 CLI。决定、候选关联与实际结果的区别见审计证据。

iter_eval_contexts 使用流式子进程读取结果，不使用上述 timeout；消费方需负责中止或限时。

Client 方法

批次查询通常使用 batch_id、run_id、job_id、shard_id、latest 和 limit；上下文读取还支持 run_ids。各方法的完整参数见 Client 源码。

编写评估规则

Registry(name=...) 管理一组规则。用 @registry.rule(...) 或 registry.register(fn, ...) 注册函数，可附带 name、description 和 tags。registry.rules() 返回规则列表，registry.evaluate(ctx) 执行规则。

规则接收 EvalContext，返回一个 Signal、字典、信号序列或 None：

ctx.run_id 提供执行 ID，ctx.raw 保留完整输入。

共同字段包括 run_id、attempt_id、tool_call_id 和 evidence。to_dict() 转为字典，validate() 检查名称、类型和分数。字典形式可调用 validate_signal_dict()。四种类型也提供对应的 KIND_* 常量。

TrajectoryScore 保存 run_id、record_manifest、context 和 report，通过 signals 读取信号。reward() 默认累加奖励特征与惩罚信号的分数；可用 include_kinds 改类型，用 weights 按名称加权。指定权重后，未列出的名称权重为 0。奖励规则由使用方定义。

独立评估器与便捷函数

• evaluate() 返回信号字典列表；evaluate_context() 返回一份评估报告。
• evaluate_batch() 接收上下文对象、字典或 JSONL 行，返回报告列表。
• reports_jsonl() 返回 JSONL 字符串；emit_jsonl() 写入指定输出流，默认标准输出。
• 全局 rule()、register() 使用 default_registry；多个独立评估器建议各建一个 Registry。
• main(registry) 从标准输入读取一个上下文，向标准输出写入信号 JSON，便于接入 CLI 外部评估流程。
• 顶层 record、batch_record、record_batch、batch_eval_contexts、batch_forensics 和 run_batch_pipeline 是便捷入口，可直接传入 binary、data_dir 等参数。

完整评估器例子见 python_signal_eval.py，模型评估接入见 LLM Judge。

错误处理

参数或信号格式错误抛出 ValueError。CLI 非零退出、普通调用超时和 HTTP 错误抛出 RuntimeError，保留命令或服务端错误信息。批处理需要保留成功任务时，可用 batch_record(..., continue_on_error=True)。