AgentLoop Experiment SDK

AgentLoop 离线实验 SDK，用于从 AgentLoop 实验计划加载数据集，在本地串行或并行执行 Agent，并将实验输入、原始输出、运行状态及可选的 Pipeline 处理结果写入 SLS experiment_detail。

本仓库是纯 Python SDK，适合 HTTP Agent、本地 CLI 和自定义 Python Agent。

核心能力

• 从实验计划自动解析数据集、数据筛选条件、选中数据项和 Pipeline
• 自动创建、启动、更新和完成离线实验记录
• 支持串行与线程池并行执行
• 支持同步或异步自定义 Solution
• 支持 HTTP、JSON、SSE 和本地 CLI 调用
• 自动注入 W3C traceparent 与 AgentLoop baggage
• 自动提取 Agent 响应中的 trace ID
• 将每条任务结果写入 SLS experiment_detail
• 自动展开数据集字段为 dataset.<column>
• 实验计划关联 Pipeline 时，将结果展开为 pipeline.<field>
• 原始输出与 Pipeline 结果写入同一条 SLS 日志
• 本地保存任务结果，支持审计和断点续跑
• 运行时仅依赖 AgentLoop OpenAPI 和阿里云 SLS SDK

仓库结构

agentloop-experiment-sdk/
├── agentloop_experiment_sdk/         # Python SDK 源码
├── examples/                          # HTTP、CLI 和自定义 Solution 示例
├── pyproject.toml                     # PEP 621 打包配置
├── .gitignore
└── README.md

缓存、测试代码、构建产物、wheel、egg-info、虚拟环境和本地实验结果不会纳入发布分支。

Python 环境要求

• Python 3.8+
• 类型为 offline 的 AgentLoop 实验计划
• 实验计划已关联数据集
• AgentSpace 已绑定 SLS Project
• AccessKey 对 AgentLoop、数据集和 SLS Project 具备所需权限

安装 Python SDK

从仓库安装：

pip install .
# 或使用 uv
uv pip install .

本地开发可使用 editable 模式：

pip install -e .
# 或使用 uv
uv pip install -e .

安装时会自动安装：

• alibabacloud_agentloop20260520>=2.3.7,<3.0.0
• aliyun-log-python-sdk>=0.9.0

Python SDK 没有内置 OpenAPI 生成源码，也不依赖 requests、OpenTelemetry、Web 框架或数据库 ORM。

避免生成 Python 缓存

仓库内的 examples 会在导入 SDK 前设置 sys.dont_write_bytecode = True，直接运行时不会生成 __pycache__ 和 .pyc。执行其他 Python 命令时建议使用：

python3 -B your_script.py
# 或在当前终端统一设置
export PYTHONDONTWRITEBYTECODE=1

.gitignore 同时忽略 Python 缓存、构建产物、egg-info 和本地实验结果，避免误提交。

配置凭证

推荐使用环境变量：

export AGENTLOOP_AK="your-access-key-id"
export AGENTLOOP_SK="your-access-key-secret"

同时兼容阿里云通用环境变量：

export ALIBABA_CLOUD_ACCESS_KEY_ID="your-access-key-id"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="your-access-key-secret"

如果以上变量均未设置，SDK 会读取 ~/.aliyun/config.json 中的 default profile。可通过 ALIBABA_CLOUD_PROFILE 或 ALIYUN_PROFILE 选择其他 profile，也可用 ALIBABA_CLOUD_CREDENTIALS_FILE 指定配置文件路径。

也可以显式传入：

from agentloop_experiment_sdk import AgentLoopConfig

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
)

不要将 AccessKey、Token 或其他凭证提交到 Git。

AgentLoopConfig

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
    project=None,
    max_rows=1000,
    ground_truth_field=None,
    experiment_name=None,
    agentloop_endpoint=None,
    sls_endpoint=None,
)

主要参数：

参数	必填	说明
agent_space	是	AgentSpace 名称
experiment_plan_id	是	离线实验计划 ID
region_id	是	地域 ID
project	否	SLS Project；未填写时从 AgentSpace 自动解析
max_rows	否	最多加载的数据量，默认 1000
ground_truth_field	否	数据集中的标准答案字段
experiment_name	否	本地实验分组名称
access_key_id	否	AccessKey ID，优先级高于环境变量
access_key_secret	否	AccessKey Secret，优先级高于环境变量
agentloop_endpoint	否	自定义 AgentLoop Endpoint
sls_endpoint	否	自定义 SLS Endpoint

调用 resolve_experiment_plan() 后，以下字段会从实验计划自动解析：

• dataset
• plan_name
• pipeline_name
• plan_evaluators
• selected_item_ids
• query_sql

方式一：运行 HTTP Agent

import logging
import os

logging.basicConfig(level=logging.INFO)

from agentloop_experiment_sdk import AgentLoopConfig, http_solution_with, run_experiment_parallel

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
    max_rows=100,
)

solution = http_solution_with(
    "https://your-agent-endpoint.example.com/invoke",
    headers={"Authorization": "Bearer {}".format(os.environ["AGENT_TOKEN"])},
    body_builder=lambda task: {
        "input": task.input.get("input", ""),
    },
    timeout=300,
)

record_id = run_experiment_parallel(
    config=config,
    solution=solution,
    result_dir="./results",
    n_repeat=1,
    n_workers=4,
)
logging.getLogger(__name__).info("实验记录 ID：%s", record_id)

完整示例：examples/http_offline_experiment.py。

HTTP 与 SSE

http_solution / http_solution_with 支持：

• GET、POST、PUT、DELETE、PATCH、HEAD、OPTIONS
• JSON 或普通请求体
• Query 参数
• 自定义 Header
• 请求超时
• 幂等方法的传输层重试
• JSON、文本和 SSE 响应
• 自定义 SSE chunk 提取函数

授权、Token、Cookie 等敏感 Header 在本地结果摘要中会被脱敏。

方式二：运行本地 CLI

command_solution_with 可以根据每条数据动态生成命令：

import logging

logging.basicConfig(level=logging.INFO)

from agentloop_experiment_sdk import AgentLoopConfig, command_solution_with, run_experiment_parallel

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
)

solution = command_solution_with(
    lambda task: [
        "your-agent-cli",
        "--prompt",
        str(task.input.get("input", "")),
        "--output-format",
        "text",
    ],
    default_timeout=120,
)

record_id = run_experiment_parallel(
    config=config,
    solution=solution,
    n_workers=4,
)
logging.getLogger(__name__).info("实验记录 ID：%s", record_id)

完整示例：examples/command_offline_experiment.py。

命令执行结果会保存为：

{
    "stdout": "...",
    "stderr": "...",
    "returncode": 0,
}

命令超时或非零退出码会转换为失败的 SolutionOutput，不会导致其他并行任务停止。

方式三：自定义 Python Solution

Solution 支持普通函数和异步函数：

import logging

logging.basicConfig(level=logging.INFO)

from agentloop_experiment_sdk import AgentLoopConfig, SolutionOutput, Task, run_experiment_parallel

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
)


async def custom_solution(task: Task, pre_hook) -> SolutionOutput:
    question = task.input.get("input", "")
    pre_hook()

    # 在这里调用你的 Agent、模型或任意异步逻辑。
    result = "answer to: {}".format(question)
    return SolutionOutput.success(
        result,
        meta={"experiment_input": question},
    )


record_id = run_experiment_parallel(
    config=config,
    solution=custom_solution,
    n_workers=4,
)
logging.getLogger(__name__).info("实验记录 ID：%s", record_id)

完整示例：examples/custom_solution_experiment.py。

run_experiment 和 run_experiment_parallel 是同步入口，不需要 asyncio.run。执行器会自动识别并等待异步 Solution。

Solution 也可以直接返回普通 Python 对象，SDK 会自动包装为成功结果：

def solution(task):
    return {"answer": task.input.get("input", "")}

串行与并行运行

串行运行：

from agentloop_experiment_sdk import run_experiment

record_id = run_experiment(config, solution)

并行运行：

from agentloop_experiment_sdk import run_experiment_parallel

record_id = run_experiment_parallel(
    config,
    solution,
    n_repeat=1,
    n_workers=4,
)

• n_repeat：每条数据重复执行次数，默认 1
• n_workers：并行线程数量，并行入口默认 4
• 返回值：AgentLoop 实验记录 ID，即 recordId

Pipeline 数据处理

是否使用 Pipeline 完全由实验计划决定。SDK 不要求调用方手工传入 Pipeline 名称。

未关联 Pipeline

每条任务只写入原始字段，包括：

• experiment_input
• experiment_output
• dataset.<column>
• 状态、trace ID、延迟等字段

已关联 Pipeline

SDK 会执行以下流程：

1. 从 GetExperimentPlan 响应解析 pipeline_name。
2. 根据 traceId 调用 GetPipeline 获取 Pipeline 定义。
3. 调用 PreviewPipeline 处理本次 Agent 轨迹。
4. 将第一行结果展开为 pipeline.<field>。
5. 把原始输入、原始输出、pipeline_name 和全部 pipeline.<field> 写入同一条 experiment_detail 日志。

例如同一条 SLS 日志可能包含：

experiment_output={"answer":"raw answer"}
pipeline_name=your-pipeline
pipeline.question=normalized question
pipeline.answer=processed answer
pipeline.score=0.95

Pipeline 不会替换 experiment_output，原始响应始终保留。

Pipeline 权限

除普通离线实验权限外，还需要：

• agentloop:GetPipeline
• agentloop:PreviewPipeline

Pipeline 等待策略

Agent 轨迹通常异步上报到 SLS。PreviewPipeline 返回空数据时：

• 每 10 秒重试一次
• 累计等待最多 2 分钟
• 每次重试会刷新查询结束时间
• Pipeline 失败不会阻断原始实验结果写入
• 失败时日志保留 pipeline_name，但可能没有 pipeline.<field>

Trace 传递

HTTP Solution 会自动：

1. 生成 W3C 格式 trace ID 和 span ID。
2. 注入 traceparent Header。
3. 在 baggage 中注入实验 ID：traffic.agentloop.experiment.id。
4. 在 baggage 中注入数据项 ID：traffic.agentloop.dataset.data_item_id。
5. 优先读取响应头 eagleeye-traceId 或 traceId。
6. 响应头没有合法的 32 位十六进制 trace ID 时使用本地生成值。

如果被测服务通过其他响应头返回 trace ID：

export AGENTLOOP_TRACE_ID_RESPONSE_HEADER="X-Trace-Id"

自定义 Solution 可通过 meta 显式返回：

return SolutionOutput.success(
    output={"answer": "ok"},
    meta={
        "experiment_input": task.input,
        "traceId": "0123456789abcdef0123456789abcdef",
    },
)

SLS experiment_detail 字段

每个任务的每次重复执行对应一条日志，主要字段如下：

字段	说明
experiment_id	实验记录 ID
experiment_name	实验分组名称
experiment_start_time	实验启动毫秒时间戳
experiment_type	当前为 agent
experiment_metadata	记录与计划元数据
experiment_config	实验配置 JSON
task_id	SDK 生成的任务 ID
repeat_id	重复执行序号
data_config	数据集及数据项配置
experiment_input	Agent 实际输入
experiment_output	Agent 原始输出或错误摘要
status	success 或 failed
traceId	本次调用的 trace ID
experiment_metrics	当前包含延迟信息
dataset.<column>	原始数据集列
pipeline_name	实验计划关联的 Pipeline
pipeline.<field>	Pipeline 处理结果

experiment_metadata 中的计划字段使用 snake_case：

{
  "run_env": "local_run",
  "record_id": "...",
  "record_name": "...",
  "experiment_group_name": "...",
  "plan_id": "...",
  "plan_name": "..."
}

本地结果

默认结果目录是 ./results。目录包含：

• evaluation_meta.json
• <task_id>/task_meta.json
• <task_id>/<repeat_id>/solution.json
• 可选的 stats.json
• evaluation_result.json

如果 solution.json 已存在，SDK 会跳过对应任务和重复序号，从而支持基础断点续跑。

实验生命周期

SDK 自动维护 AgentLoop 实验记录：

1. CreateExperimentRun 创建离线实验记录。
2. 数据加载完成后更新为 running 并写入总任务数和执行时间。
3. 在 10%、25%、50%、75%、100% 里程碑上报进度。
4. 全部任务完成后更新为 completed。
5. 执行器抛出未处理异常时更新为 failed。

单条任务的业务异常会转换为失败结果并继续执行其他任务。SLS 或 Pipeline 写入失败会记录日志，本地任务结果仍会保留。

自定义 Endpoint

预发、联调或私有环境可覆盖 Endpoint：

config = AgentLoopConfig(
    agent_space="your-agent-space",
    experiment_plan_id="your-offline-plan-id",
    region_id="cn-hangzhou",
    agentloop_endpoint="agentloop.example.com",
    sls_endpoint="cn-hangzhou.log.aliyuncs.com",
)

构建 Python 发布包

python -m pip wheel --no-deps -w dist .

生成文件位于 dist/。该目录已被 .gitignore 忽略。

常见问题

实验计划类型错误

Python SDK 只运行 offline 类型实验计划。请确认 experiment_plan_id 指向离线实验计划。

数据集为空

确认实验计划已关联数据集，并检查计划中的 selected_item_ids、query_sql 和 max_rows。

SLS 没有实验明细

确认：

• AgentSpace 已绑定 SLS Project
• AccessKey 有目标 Project 和 experiment_detail 写权限
• sls_endpoint 与 Region 匹配
• 本地日志中没有 SLS 写入异常

有 pipeline_name 但没有 pipeline.<field>

常见原因：

• 缺少 GetPipeline 或 PreviewPipeline 权限
• Solution 没有返回可用于检索轨迹的 trace ID
• Agent 轨迹尚未写入 Pipeline 数据源
• Pipeline 源配置或节点配置不完整
• 两分钟等待窗口内仍未检索到轨迹

异步 Solution 如何运行

直接将 async def 函数传给 run_experiment 或 run_experiment_parallel。不要对实验入口使用 asyncio.run，SDK 会在任务执行阶段等待协程。

更多文档

• HTTP 示例
• CLI 示例
• 自定义 Solution 示例