In this tutorial, you use the AgentLoop SDK (agentloop-sdk) to run an offline experiment that batch-evaluates an LLM agent application. The experiment reads inputs record by record from an existing dataset, calls your agent to obtain results, reports the experiment data to the AgentLoop platform, and can automatically trigger platform-side evaluation after the run completes.
Prerequisites
You have created an agent space.
You have created a dataset and imported data. The dataset must contain at least one record. The field names of the record are used as the keys of the experiment input.
You have created an offline experiment plan through an evaluation task and associated the experiment plan with the dataset above. The experiment plan type must be offline (
experiment_typeisoffline).(Optional) You have configured an evaluator for the experiment plan. You can also skip the platform configuration and specify the evaluator through
EvaluatorConfigin the SDK code, as described in Step 5: Configure evaluators.You have obtained the Alibaba Cloud
AccessKey IDandAccessKey secretfor SDK authentication.
Step 1: Install the SDK
Install agentloop-sdk with pip:
pip install agentloop-sdkStep 2: Configure experiment parameters
Use the AgentLoopConfig class to configure the core parameters of the experiment. The dataset is resolved automatically from the experiment plan, so you only need to identify the agent space, the experiment plan, and the region.
from agentloop_sdk import AgentLoopConfig
config = AgentLoopConfig(
agent_space="my-agent-space",
experiment_plan_id="my_plan_id",
region_id="cn-hangzhou",
)AgentLoopConfig supports the following parameters:
Parameter | Type | Required | Default | Description |
|
| Yes | - | The name of the agent space. |
|
| Yes | - | The experiment plan ID. The plan must be offline ( |
|
| Yes | - | The Alibaba Cloud region ID, for example |
|
| No |
| The experiment name, used to identify this experiment. The experiment record name shown on the platform is automatically generated from the plan name plus a timestamp. |
|
| No | 1000 | The maximum number of records read from the dataset for the experiment. The SDK reads records with automatic pagination up to this limit. If the dataset contains more records than |
|
| No | - | The Alibaba Cloud |
|
| No | - | The Alibaba Cloud |
|
| No | - | Additional configuration for the experiment, such as the agent name. Example: |
|
| No | - | The evaluators to run after the experiment completes. When set, platform-side evaluation is triggered automatically. See Step 5: Configure evaluators. |
Step 3: Configure credentials
The SDK uses the Alibaba Cloud AccessKey to authenticate your identity. The following two methods are supported.
Method 1: Environment variables (recommended)
After you set the following environment variables, the SDK reads them automatically, so you do not need to pass the credentials in code:
export AGENTLOOP_AK="your-access-key-id"
export AGENTLOOP_SK="your-access-key-secret"The standard Alibaba Cloud environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are also supported.
Method 2: Pass parameters in code
If you do not set the environment variables, pass the AccessKey directly when you create an AgentLoopConfig:
from agentloop_sdk import AgentLoopConfig
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",
)For the full list of AgentLoopConfig parameters, see Step 2: Configure experiment parameters.
Step 4: Define the solution function
The solution function defines how to call the agent and obtain results. The SDK provides two methods: HTTP calls and command-line calls. HTTP calls apply to remote agents accessed through an API. Command-line calls apply to locally deployed agents.
The task.input of the solution function is a dict that corresponds to one record in the dataset, where the keys are the dataset column names.
Call an agent over HTTP
Use http_solution_with to call an agent over an HTTP interface. It supports standard JSON responses and SSE streaming responses.
Standard JSON response
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 streaming response
When the agent endpoint returns the text/event-stream format, the SDK automatically detects and parses the SSE, extracts the content of all data: lines, and concatenates them into the final result:
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,
)If each SSE chunk is in JSON format, you can extract a specific field with 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 supports the following parameters:
Parameter | Type | Required | Default | Description |
|
| Yes | - | The request URL of the agent service. |
|
| No |
| The HTTP request method. |
|
| No |
| The request header. |
|
| No | - | A function that dynamically builds the request body from the Task. |
|
| No | - | A static JSON request body. Use either this or |
|
| No | 60 | The request timeout, in seconds. |
|
| No | - | A function that extracts a specific field from each JSON chunk in an SSE streaming response. |
Underlying API: http_solution
If you need full control over the request construction logic, use the underlying 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)Call an agent from the command line
Use command_solution_with to call a locally deployed agent from the command line:
from agentloop_sdk import command_solution_with
# Static command
solution = command_solution_with(["python", "agent.py", "--input", "hello"])
# Build the command dynamically from the Task
solution = command_solution_with(
lambda task: ["python", "agent.py", "--input", task.input.get("input", "")],
default_timeout=120.0,
)command_solution_with supports the following parameters:
Parameter | Type | Required | Default | Description |
|
| Yes | - | A static command list, or a function that builds the command dynamically from the Task. |
|
| No | - | The command execution timeout, in seconds. |
Underlying API: command_solution
If you need more control, use the underlying 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 supports the following parameters:
Parameter | Type | Required | Default | Description |
|
| Yes | - | A function that builds the command list from the Task. |
|
| No | - | The command execution timeout, in seconds. |
|
| No | - | The working directory for command execution. |
Step 5: Configure evaluators (optional)
To have the AgentLoop platform evaluate the experiment automatically, configure an evaluator. You can configure the evaluator on the platform when you create the experiment plan, or specify it in code with EvaluatorConfig. When you set evaluators on AgentLoopConfig, platform-side evaluation is triggered automatically after the experiment completes.
Pass a list of EvaluatorConfig objects to the evaluators parameter of AgentLoopConfig:
from agentloop_sdk import AgentLoopConfig, EvaluatorConfig
config = AgentLoopConfig(
agent_space="my-agent-space",
experiment_plan_id="my_plan_id",
region_id="cn-hangzhou",
evaluators=[
EvaluatorConfig(
evaluator_ref="Builtin.agent_correctness",
result_type="score",
variable_mapping={
"input": "experiment_input",
"output": "experiment_output",
"expected_output": "dataset.expected_output",
},
),
],
)EvaluatorConfig uses the following fields:
Field | Description |
| The evaluator to run. For example, |
| The type of evaluation result, such as |
| Maps the evaluator's input variables to experiment fields (such as |
If you already configured an evaluator for the experiment plan on the platform, you can skip this step.
Step 6: Run the experiment
After the configuration is complete, call the run function provided by the SDK to run the experiment. The SDK supports sequential and parallel execution.
Sequential execution
Run the tasks in the dataset one by one:
import asyncio
from agentloop_sdk import run_experiment
asyncio.run(run_experiment(
config=config,
solution_fn=solution,
result_dir="./results",
n_repeat=1,
))Parallel execution (recommended)
Run tasks concurrently through multiple workers. This is suitable for large datasets:
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,
))The run functions support the following parameters:
Parameter | Type | Required | Default | Description |
|
| Yes | - | The experiment configuration. |
|
| Yes | - | The solution function. |
|
| No |
| The local directory for saving results. |
|
| No | 1 | The number of repeated executions for each record. |
|
| No | 1 for sequential execution and 4 for parallel execution | The number of concurrent workers. |
Step 7: Verify the experiment results
After the experiment finishes, confirm that the run completed and that the results reached the AgentLoop platform.
Check the local results. The SDK saves results to the directory specified by
result_dir(./resultsby default). Confirm that the directory contains the generated result files.Confirm platform reporting. On the AgentLoop platform, confirm that a new experiment record appears for your experiment plan. The record name is generated from the plan name plus a timestamp.
Confirm evaluation. If an evaluator is configured—either on the platform or through
EvaluatorConfig—platform-side evaluation is triggered automatically after the experiment completes. Confirm that the evaluation results appear on the platform.
Clean up
The tutorial writes experiment results to the local result_dir directory and reports experiment records to the AgentLoop platform. To reclaim local disk space, delete the results directory when you no longer need the local copies.
Trace correlation
When http_solution and http_solution_with send HTTP requests, they automatically inject a traceparent header that complies with the W3C Trace Context specification into the request header. They also extract eagleeye-traceId or traceId from the response header to associate the experiment requests with platform-side trace correlation. You do not need to handle the trace headers manually.
Complete example
The following example shows a complete offline experiment workflow: configuring the experiment parameters, defining an evaluator, calling an agent over HTTP, and running with 4 parallel workers.
import asyncio
from agentloop_sdk import (
AgentLoopConfig,
EvaluatorConfig,
http_solution_with,
run_experiment_parallel,
)
async def main():
config = AgentLoopConfig(
agent_space="default-cn-hangzhou",
experiment_plan_id="cc_plan_id",
region_id="cn-hangzhou",
experiment_name="Offline-Experiment-LocalAgent",
experiment_config={"agent_name": "LocalAgent"},
evaluators=[
EvaluatorConfig(
evaluator_ref="Builtin.agent_correctness",
result_type="score",
variable_mapping={
"input": "experiment_input",
"output": "experiment_output",
"expected_output": "dataset.expected_output",
},
),
],
)
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", ""),
)
await run_experiment_parallel(
config=config,
solution_fn=solution,
n_workers=4,
)
if __name__ == "__main__":
asyncio.run(main())