Score by Component

View as Markdown

The quickstart scored one thing: whether the final answer contained the right keyword. But an agent can reach the right answer the wrong way — guessing instead of looking something up, or looping on a tool. Agent evaluation lets you score how the agent worked, not just the outcome, and combine several signals into one reported view.

This guide extends the quickstart. You’ll add a trajectory metric that reads the agent’s tool calls, keep the quickstart’s outcome metric, and combine them into a quality view. It stays zero-dependency — one local process, no API keys.

This builds on the quickstart’s KeywordMatchMetric and tasks. The full runnable script is at the end.

1. The outcome metric

Reuse the quickstart’s KeywordMatchMetric — it scores the final answer, reading it from input.candidate.output_text and the grader-only truth from input.row.data:

from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
class KeywordMatchMetric:
@property
def type(self) -> str:
return "keyword_match"
def output_spec(self) -> list[MetricOutputSpec]:
return [MetricOutputSpec.continuous_score("score")]
async def compute_scores(self, input: MetricInput) -> MetricResult:
expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
answer = (input.candidate.output_text or "").lower()
return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])

2. A trajectory metric

A metric can read more than the final output. input.candidate.evidence exposes the trial’s evidence — for the trajectory, an Agent Trajectory Interchange Format (ATIF) trace of the agent’s steps. evidence.trace(...) gives a handle whose tool_calls() returns the tool calls in order; each ToolCall has a function_name and arguments. This metric scores 1.0 when the agent used the tool you expected and 0.0 otherwise:

from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
from nemo_evaluator_sdk.values.evidence import EVIDENCE_TRACE
class UsedExpectedToolMetric:
"""Score whether the agent's trajectory used a given tool."""
def __init__(self, expected_tool: str) -> None:
self._expected = expected_tool
@property
def type(self) -> str:
return "used_expected_tool"
def output_spec(self) -> list[MetricOutputSpec]:
return [MetricOutputSpec.continuous_score("tool_use")]
async def compute_scores(self, input: MetricInput) -> MetricResult:
used = False
evidence = input.candidate.evidence
if evidence is not None and evidence.get(EVIDENCE_TRACE) is not None:
calls = await (await evidence.trace(EVIDENCE_TRACE)).tool_calls()
used = any(call.function_name == self._expected for call in calls)
return MetricResult(outputs=[MetricOutput(name="tool_use", value=1.0 if used else 0.0)])

The metric checks for evidence first — a trial without a trace simply scores 0.0 rather than erroring.

3. Produce a trajectory

Runners that execute the agent capture this evidence for you — a deployed HTTP agent returns a trajectory when you set trajectory_path, and container/harness backends record one as the agent runs. (A Harbor run scores a verifier reward instead.) Here the agent is a local callable, so it returns a TrialDraft — its final output plus, when it used a tool, a small ATIF trace built from the SDK’s trajectory models. One task’s agent uses the search tool; the other skips it and just guesses, returning no trace at all:

from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import TrialDraft
from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask
from nemo_evaluator_sdk.agent_eval.trials import AgentOutput
from nemo_evaluator_sdk.values.atif import Step, ToolCall, Trajectory
from nemo_evaluator_sdk.values.evidence import (
EVIDENCE_FORMAT_ATIF,
EVIDENCE_TRACE,
CandidateEvidence,
EvidenceDescriptor,
)
def _trace(tool_name: str, query: str) -> CandidateEvidence:
trajectory = Trajectory(
schema_version="ATIF-v1.7",
steps=[Step(source="agent", tool_calls=[ToolCall(function_name=tool_name, arguments={"query": query})])],
)
return CandidateEvidence(
descriptors={
EVIDENCE_TRACE: EvidenceDescriptor(
kind="trace", format=EVIDENCE_FORMAT_ATIF, data=trajectory.model_dump(mode="json")
)
}
)
async def my_agent(task: AgentEvalTask) -> TrialDraft:
answers = {
"capital-france": ("The capital of France is Paris.", "search"), # used the expected tool
"capital-japan": ("The capital of Japan is Tokyo.", None), # right answer, but skipped the tool
}
text, tool = answers[task.id]
evidence = _trace(tool, task.inputs["instruction"]) if tool else None
return TrialDraft(output=AgentOutput(output_text=text), evidence=evidence)

4. Combine the signals into a view

Attach both metrics to each task, then define a view — a named roll-up of this task’s metric outputs. SemanticView reduces its signals (each a metric.output) into one score; MEAN averages them. The view is reported per task and aggregated across the run as view.<name>.

from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalTask, SemanticReducer, SemanticView, ViewSignal
def _quality_view() -> SemanticView:
return SemanticView(
reducer=SemanticReducer.MEAN,
signals=[
ViewSignal(metric="keyword_match", output="score"),
ViewSignal(metric="used_expected_tool", output="tool_use"),
],
)
def make_tasks() -> list[AgentEvalTask]:
metrics = [KeywordMatchMetric(), UsedExpectedToolMetric("search")]
return [
AgentEvalTask(
id="capital-france", intent="Answer the geography question.",
inputs={"instruction": "What is the capital of France?"}, reference={"expected": "Paris"},
metrics=metrics, views={"quality": _quality_view()},
),
AgentEvalTask(
id="capital-japan", intent="Answer the geography question.",
inputs={"instruction": "What is the capital of Japan?"}, reference={"expected": "Tokyo"},
metrics=metrics, views={"quality": _quality_view()},
),
]

5. Run it and read the components

from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import CallableAgentTaskRunner
from nemo_evaluator_sdk.agent_eval.tasks import AgentEvalRunConfig
result = await AgentEvaluator().run(
tasks=make_tasks(),
target=CallableAgentTaskRunner(my_agent),
config=AgentEvalRunConfig(parallelism=2),
)
for aggregate in result.summary.scores.scores:
print(f"{aggregate.name}: {aggregate.mean}")

Output:

keyword_match.score: 1.0
used_expected_tool.tool_use: 0.5
view.quality: 0.75

This is the whole point of scoring by component. The outcome looks perfect — every answer was correct (1.0). But the trajectory metric shows only half the agents actually used the expected tool (0.5); the other got the right answer by guessing. The view combines the two into a single quality score (0.75) you can track over time. Scoring only the answer would have hidden the shortcut entirely.

Full script

import asyncio
from nemo_evaluator_sdk.agent_eval.evaluator import AgentEvaluator
from nemo_evaluator_sdk.agent_eval.runtimes.callable_runtime import CallableAgentTaskRunner, TrialDraft
from nemo_evaluator_sdk.agent_eval.tasks import (
AgentEvalRunConfig,
AgentEvalTask,
SemanticReducer,
SemanticView,
ViewSignal,
)
from nemo_evaluator_sdk.agent_eval.trials import AgentOutput
from nemo_evaluator_sdk.metrics.protocol import MetricInput, MetricOutput, MetricOutputSpec, MetricResult
from nemo_evaluator_sdk.values.atif import Step, ToolCall, Trajectory
from nemo_evaluator_sdk.values.evidence import (
EVIDENCE_FORMAT_ATIF,
EVIDENCE_TRACE,
CandidateEvidence,
EvidenceDescriptor,
)
class KeywordMatchMetric:
@property
def type(self) -> str:
return "keyword_match"
def output_spec(self) -> list[MetricOutputSpec]:
return [MetricOutputSpec.continuous_score("score")]
async def compute_scores(self, input: MetricInput) -> MetricResult:
expected = str(input.row.data.get("reference", {}).get("expected", "")).lower()
answer = (input.candidate.output_text or "").lower()
return MetricResult(outputs=[MetricOutput(name="score", value=1.0 if expected and expected in answer else 0.0)])
class UsedExpectedToolMetric:
"""Score whether the agent's trajectory used a given tool."""
def __init__(self, expected_tool: str) -> None:
self._expected = expected_tool
@property
def type(self) -> str:
return "used_expected_tool"
def output_spec(self) -> list[MetricOutputSpec]:
return [MetricOutputSpec.continuous_score("tool_use")]
async def compute_scores(self, input: MetricInput) -> MetricResult:
used = False
evidence = input.candidate.evidence
if evidence is not None and evidence.get(EVIDENCE_TRACE) is not None:
calls = await (await evidence.trace(EVIDENCE_TRACE)).tool_calls()
used = any(call.function_name == self._expected for call in calls)
return MetricResult(outputs=[MetricOutput(name="tool_use", value=1.0 if used else 0.0)])
def _trace(tool_name: str, query: str) -> CandidateEvidence:
trajectory = Trajectory(
schema_version="ATIF-v1.7",
steps=[Step(source="agent", tool_calls=[ToolCall(function_name=tool_name, arguments={"query": query})])],
)
return CandidateEvidence(
descriptors={
EVIDENCE_TRACE: EvidenceDescriptor(
kind="trace", format=EVIDENCE_FORMAT_ATIF, data=trajectory.model_dump(mode="json")
)
}
)
def _quality_view() -> SemanticView:
return SemanticView(
reducer=SemanticReducer.MEAN,
signals=[
ViewSignal(metric="keyword_match", output="score"),
ViewSignal(metric="used_expected_tool", output="tool_use"),
],
)
def make_tasks() -> list[AgentEvalTask]:
metrics = [KeywordMatchMetric(), UsedExpectedToolMetric("search")]
return [
AgentEvalTask(
id="capital-france", intent="Answer the geography question.",
inputs={"instruction": "What is the capital of France?"}, reference={"expected": "Paris"},
metrics=metrics, views={"quality": _quality_view()},
),
AgentEvalTask(
id="capital-japan", intent="Answer the geography question.",
inputs={"instruction": "What is the capital of Japan?"}, reference={"expected": "Tokyo"},
metrics=metrics, views={"quality": _quality_view()},
),
]
async def my_agent(task: AgentEvalTask) -> TrialDraft:
answers = {
"capital-france": ("The capital of France is Paris.", "search"), # used the expected tool
"capital-japan": ("The capital of Japan is Tokyo.", None), # right answer, but skipped the tool
}
text, tool = answers[task.id]
evidence = _trace(tool, task.inputs["instruction"]) if tool else None
return TrialDraft(output=AgentOutput(output_text=text), evidence=evidence)
async def main() -> None:
result = await AgentEvaluator().run(
tasks=make_tasks(),
target=CallableAgentTaskRunner(my_agent),
config=AgentEvalRunConfig(parallelism=2),
)
for aggregate in result.summary.scores.scores:
print(f"{aggregate.name}: {aggregate.mean}")
asyncio.run(main())

Next steps