Metadata-Version: 2.4
Name: agentloop-experiment-sdk
Version: 0.1.0
Summary: Lightweight AgentLoop offline experiment SDK
License: Apache-2.0
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: alibabacloud_agentloop20260520<3.0.0,>=2.3.7
Requires-Dist: aliyun-log-python-sdk>=0.9.0

# 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

## 仓库结构

```text
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

从仓库安装：

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

本地开发可使用 editable 模式：

```bash
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 命令时建议使用：

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

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

## 配置凭证

推荐使用环境变量：

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

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

```bash
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` 指定配置文件路径。

也可以显式传入：

```python
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

```python
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

```python
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](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` 可以根据每条数据动态生成命令：

```python
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](examples/command_offline_experiment.py)。

命令执行结果会保存为：

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

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

## 方式三：自定义 Python Solution

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

```python
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](examples/custom_solution_experiment.py)。

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

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

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

## 串行与并行运行

串行运行：

```python
from agentloop_experiment_sdk import run_experiment

record_id = run_experiment(config, solution)
```

并行运行：

```python
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 日志可能包含：

```text
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：

```bash
export AGENTLOOP_TRACE_ID_RESPONSE_HEADER="X-Trace-Id"
```

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

```python
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：

```json
{
  "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：

```python
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 发布包

```bash
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 示例](examples/http_offline_experiment.py)
- [CLI 示例](examples/command_offline_experiment.py)
- [自定义 Solution 示例](examples/custom_solution_experiment.py)
