使用离线实验上报实验结果

方式一：环境变量（推荐）

设置以下环境变量后，SDK 自动读取，无需通过代码传入凭证信息：

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

也支持标准阿里云环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID 和 ALIBABA_CLOUD_ACCESS_KEY_SECRET。

方式二：代码传参

创建 AgentLoopConfig 时直接传入 AccessKey：

config = AgentLoopConfig(
    agent_space="my-agent-space",
    experiment_plan_id="my_plan_id",
    region_id="cn-hangzhou",
    access_key_id="your-ak",
    access_key_secret="your-sk",
)

通过 HTTP 调用 Agent

使用 http_solution_with 通过 HTTP 接口调用 Agent，支持普通 JSON 响应和 SSE 流式响应。

普通 JSON 响应

from agentloop_sdk import http_solution_with

solution = http_solution_with(
    url="https://your-agent-endpoint.com/invoke",
    headers={"key": "your-api-key"},
    body_builder=lambda task: {
        "input": task.input.get("input", ""),
        "stream": False,
    },
    timeout=60.0,
)

SSE 流式响应

当 Agent 端点返回 text/event-stream 格式时，SDK 自动检测并解析 SSE，提取所有 data: 行内容拼接为最终结果：

solution = http_solution_with(
    url="https://your-agent-endpoint.com/invoke",
    headers={"key": "your-api-key"},
    body_builder=lambda task: {
        "input": task.input.get("input", ""),
        "stream": True,
    },
    timeout=120.0,
)

如果 SSE 的每个 chunk 是 JSON 格式，可以通过 chunk_extractor 提取指定字段：

solution = http_solution_with(
    url="https://your-agent-endpoint.com/invoke",
    headers={"key": "your-api-key"},
    body_builder=lambda task: {
        "input": task.input.get("input", ""),
        "stream": True,
    },
    timeout=120.0,
    chunk_extractor=lambda chunk: chunk.get("output", ""),
)

http_solution_with 支持以下参数：

底层 API：http_solution

如果需要完全控制请求构造逻辑，可以使用底层 API http_solution：

from agentloop_sdk import http_solution, HttpRequestSpec

def build_request(task):
    return HttpRequestSpec(
        url="https://your-agent-endpoint.com/invoke",
        headers={"key": "your-api-key"},
        json={"input": task.input.get("input", "")},
    )

solution = http_solution(build_request, default_timeout=60.0)

通过命令行调用 Agent

使用 command_solution_with 通过命令行调用本地部署的 Agent：

from agentloop_sdk import command_solution_with

# 静态命令
solution = command_solution_with(["python", "agent.py", "--input", "hello"])

# 根据 Task 动态构造命令
solution = command_solution_with(
    lambda task: ["python", "agent.py", "--input", task.input.get("input", "")],
    default_timeout=120.0,
)

command_solution_with 支持以下参数：

底层 API：command_solution

如果需要更多控制，可以使用底层 API command_solution：

from agentloop_sdk import command_solution

def build_command(task):
    return ["python", "agent.py", "--input", task.input.get("input", "")]

solution = command_solution(build_command, default_timeout=120.0, cwd="/path/to/agent")

command_solution 支持以下参数：

串行执行

逐条执行数据集中的任务：

import asyncio
from agentloop_sdk import run_experiment

asyncio.run(run_experiment(
    config=config,
    solution_fn=solution,
    result_dir="./results",
    n_repeat=1,
))

并行执行（推荐）

通过多个 worker 并发执行任务，适合数据集较大的场景：

import asyncio
from agentloop_sdk import run_experiment_parallel

asyncio.run(run_experiment_parallel(
    config=config,
    solution_fn=solution,
    result_dir="./results",
    n_repeat=1,
    n_workers=4,
))

运行函数支持以下参数：

实验运行完成后，结果自动保存至 result_dir 指定的本地目录。如果配置了评估器，实验完成后将自动触发平台侧评估。