Custom Code Evaluator
Custom code evaluators let you write your own evaluation logic in Python, JavaScript, or TypeScript. Your code has access to the application inputs, outputs, and the full execution trace (spans, latency, token usage, costs).
On self-hosted Agenta, custom evaluator code runs server-side. By default it runs with no sandbox in the services process (trusted, single-tenant use). Operators can change the runner with the AGENTA_SERVICES_CODE_SANDBOX_RUNNER environment variable: restricted runs code in an in-process Python sandbox (no filesystem, network, or host access), daytona runs it in an isolated remote sandbox. Set restricted or daytona to harden a shared/multi-tenant deployment. See environment configuration. Agenta Cloud is unaffected. It isolates evaluator execution.
Function signature
Your code must define an evaluate function with the following signature:
from typing import Dict, Any
def evaluate(
inputs: Dict[str, Any],
outputs: Any,
trace: Dict[str, Any],
) -> float:
Parameters
| Parameter | Type | Description |
|---|---|---|
inputs | Dict[str, Any] | In batch evaluation: the testcase data (all columns). In online evaluation: the application's input from the trace. |
outputs | Any | The application's output (string or dict). |
trace | Dict[str, Any] | The full execution trace with spans, metrics (latency, token counts, costs), and child spans. |
Return value
The function can return one of:
dict— a dictionary of metrics, such as{"score": 0.8, "success": True}or{"relevance": 0.9, "tone": 0.4, "reason": "missed the greeting"}. Each key becomes a separate metric in the evaluation results. Values must be JSON-serializable. Nested dictionaries are flattened into dotted metric names.float(0.0 to 1.0) — a single score where 0.0 is worst and 1.0 is best. Agenta normalizes it to{"score": <value>, "success": <value >= threshold>}.bool— normalized to{"success": <value>}.
Examples
Exact match
from typing import Dict, Any
def evaluate(
inputs: Dict[str, Any],
outputs: Any,
trace: Dict[str, Any],
) -> dict:
success = outputs == inputs.get("correct_answer")
return {
"score": 1.0 if success else 0.0,
"success": success,
}
Multiple metrics
Return a dict to report several metrics from one evaluator. Each key shows up as its own column in the results:
from typing import Dict, Any
def evaluate(
inputs: Dict[str, Any],
outputs: Any,
trace: Dict[str, Any],
) -> dict:
text = outputs if isinstance(outputs, str) else str(outputs)
return {
"length_ok": len(text) <= 280,
"mentions_brand": "Agenta" in text,
"word_count": float(len(text.split())),
}
Latency check
Use trace data to check whether the response was generated within a time budget:
from typing import Dict, Any
def evaluate(
inputs: Dict[str, Any],
outputs: Any,
trace: Dict[str, Any],
) -> float:
if not trace or not trace.get("spans"):
return 0.0
root = list(trace["spans"].values())[0]
ag = root.get("attributes", {}).get("ag", {})
duration = ag.get("metrics", {}).get("unit", {}).get("duration", {}).get("total", 0)
# Fail if response took more than 5 seconds
if duration > 5.0:
return 0.0
return 1.0
Token budget check
Verify the application stayed within a token budget:
from typing import Dict, Any
def evaluate(
inputs: Dict[str, Any],
outputs: Any,
trace: Dict[str, Any],
) -> float:
if not trace or not trace.get("spans"):
return 0.5
root = list(trace["spans"].values())[0]
ag = root.get("attributes", {}).get("ag", {})
tokens = ag.get("metrics", {}).get("unit", {}).get("tokens", {})
total_tokens = tokens.get("total", 0)
max_tokens = 500
if total_tokens <= max_tokens:
return 1.0
elif total_tokens <= max_tokens * 1.5:
return 0.5
return 0.0
Accessing ground truth
In batch evaluation, testcase columns are available directly in inputs. If your testset has a correct_answer column, access it as inputs["correct_answer"] or inputs.get("correct_answer").
You do not need to configure a separate "correct answer key" — just read the column name directly from inputs.
Accessing trace data
The trace parameter contains the full OpenTelemetry trace serialized as a dict. The structure looks like:
{
"spans": {
"<span_id>": {
"name": "my_app",
"start_time": "2025-01-15T10:30:00Z",
"end_time": "2025-01-15T10:30:02.5Z",
"status_code": "OK",
"attributes": {
"ag": {
"data": {
"inputs": {"country": "France"},
"outputs": "The capital is Paris"
},
"metrics": {
"unit": {
"costs": {"total": 0.001},
"tokens": {"prompt": 50, "completion": 20, "total": 70},
"duration": {"total": 2.5}
}
}
}
},
"children": [...]
}
}
}
Useful paths:
| Data | Path |
|---|---|
| Root span | list(trace["spans"].values())[0] |
| App inputs | root["attributes"]["ag"]["data"]["inputs"] |
| App outputs | root["attributes"]["ag"]["data"]["outputs"] |
| Latency (seconds) | root["attributes"]["ag"]["metrics"]["unit"]["duration"]["total"] |
| Token counts | root["attributes"]["ag"]["metrics"]["unit"]["tokens"] |
| Costs | root["attributes"]["ag"]["metrics"]["unit"]["costs"]["total"] |
| Child spans | root["children"] |
JavaScript and TypeScript
The same interface is available in JavaScript and TypeScript:
JavaScript:
function evaluate(inputs, outputs, trace) {
const success = outputs === inputs.correct_answer
return {score: success ? 1.0 : 0.0, success: success}
}
TypeScript:
function evaluate(
inputs: Record<string, any>,
outputs: any,
trace: Record<string, any>
): {score: number; success: boolean} {
const success = outputs === inputs.correct_answer
return {score: success ? 1.0 : 0.0, success: success}
}
Legacy interfaces
Existing evaluators keep working unchanged. There are two older interfaces:
The original 4-parameter interface:
def evaluate(app_params, inputs, output, correct_answer) -> float:
And the float-only 3-parameter interface:
def evaluate(inputs, outputs, trace) -> float:
Evaluators created with either interface continue to return a single score. Dict returns are only supported for evaluators created after this update.
To migrate an old evaluator, create a new code evaluator and copy your logic over. Use the (inputs, outputs, trace) signature, read ground truth directly from inputs, and return a dict of metrics.